diff --git a/.gitignore b/.gitignore index b6e4761..2624f75 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,12 @@ __pycache__/ *.py[cod] *$py.class +.vscode/ +events.bin +data/ +Scripts/ +artwork* + # C extensions *.so diff --git a/README.md b/README.md index d715c19..51a1972 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,20 @@ +This branch is an attempt to get PTP sync working, based on work from others, I try to cherry pick commits to keep credit + +So far it uses PTP to sync but it have a few issues: + +re-Syncs often causing some audio skips, however it does not drift, keeps quite well in sync. + +PTP Principal +TL;DR; + This branch does not announce itself as a timmingPeer so another AirPlay peer has to do it (e.g. a Home Pod or the iDevice) + +TP principal clock (master in PTP terminology), seems to use mac address to select which one will be the principal, i.e. the lowest mac address is selected as principal +This code does not implement PTP principal so it must be connected to another one. + +Also when the principal is changed (e.g. if a new peer is added to the group), then time also changes, this is because the "time" used is actually a device arbitrary time (usually device uptime). When this happens this code does not yet update the time properly and stops working too. + +Follows readme from forked repo + # Experimental Very quick python implementation of AP2 protocol using **minimal @@ -6,7 +23,10 @@ multi-room** features. For now it implements: - FairPlay (v3) authentication - Receiving of both REALTIME and BUFFERED Airplay2 audio streams - Airplay2 Service publication -- Decoding of ALAC/44100/2 or AAC/44100/2 +- Decoding of all Airplay2 supported CODECs: ALAC, AAC, OPUS, PCM. + Ref: [here](https://emanuelecozzi.net/docs/airplay2/audio/) and + [here](https://emanuelecozzi.net/docs/airplay2/rtsp/#setup) +- Output latency compensation for sync with other Airplay receivers For now it does not implement: - MFi Authentication / FairPlay v2 (one of them is required by iTunes/Windows) diff --git a/ap2-receiver.py b/ap2-receiver.py index da839df..ceeeabd 100644 --- a/ap2-receiver.py +++ b/ap2-receiver.py @@ -24,37 +24,40 @@ from ap2.pairing.hap import Hap, HAPSocket from ap2.connections.event import Event from ap2.connections.stream import Stream +from ap2.daap import parse_daap + +from ap2.connections.ptp_time import PTP # No Auth - coreutils, PairSetupMfi # MFi Verify fail error after pair-setup[2/5] -FEATURES = 0x88340405f8a00 +FEATURES = 0x88340405F8A00 # No Auth - HK and coreutils # Stops after pairing (setup [5/5] verify [2/2])with no supported auth error -FEATURES = 0xc340405f8a00 +FEATURES = 0xC340405F8A00 # No Auth = HK, coreutils, PairSetupMFi # MFi Verify fail error after pair-setup[2/5] -FEATURES = 0x8c340405f8a00 +FEATURES = 0x8C340405F8A00 # Mfi Auth - HK and coreutils # All encrypt after pairing (setup [5/5] verify [2/2]) -FEATURES = 0xc340445f8a00 +FEATURES = 0xC340445F8A00 # FairPlay - HK and coreutils # Stops after pairing (setup [5/5] verify [2/2])with no supported auth error -FEATURES = 0xc340405fca00 +FEATURES = 0xC340405FCA00 # FairPlay - HK and coreutils and transient # fp-setup after pair-setup[2/5] -FEATURES = 0x1c340405fca00 +FEATURES = 0x1C340405FCA00 # MFi - HK and coreutils and transient # auth-setup after pair-setup[2/5] -FEATURES = 0x1c340445f8a00 +FEATURES = 0x1C340445F8A00 # No Auth - No enc - PairSetupMFi # Works!! -FEATURES = 0x8030040780a00 +FEATURES = 0x8030040780A00 # No Auth - No enc # No supported authentication types. # FEATURES = 0x30040780a00 # FEATURES = 0x8030040780a00 | (1 << 27) -FEATURES = 0x1c340405fca00 +FEATURES = 0x1C340405FCA00 DEVICE_ID = None IPV4 = None @@ -67,6 +70,7 @@ HTTP_CT_IMAGE = "image/jpeg" HTTP_CT_DMAP = "application/x-dmap-tagged" + def setup_global_structs(args): global sonos_one_info global sonos_one_setup @@ -77,88 +81,96 @@ def setup_global_structs(args): sonos_one_info = { # 'OSInfo': 'Linux 3.10.53', # 'PTPInfo': 'OpenAVNU ArtAndLogic-aPTP-changes a5d7f94-0.0.1', - 'audioLatencies': [ { 'inputLatencyMicros': 0, - 'outputLatencyMicros': 400000, - 'type': 100}, - { 'audioType': 'default', - 'inputLatencyMicros': 0, - 'outputLatencyMicros': 400000, - 'type': 100}, - { 'audioType': 'media', - 'inputLatencyMicros': 0, - 'outputLatencyMicros': 400000, - 'type': 100}, - { 'audioType': 'media', - 'inputLatencyMicros': 0, - 'outputLatencyMicros': 400000, - 'type': 102}], + "audioLatencies": [ + {"inputLatencyMicros": 0, "outputLatencyMicros": 400000, "type": 100}, + { + "audioType": "default", + "inputLatencyMicros": 0, + "outputLatencyMicros": 400000, + "type": 100, + }, + { + "audioType": "media", + "inputLatencyMicros": 0, + "outputLatencyMicros": 400000, + "type": 100, + }, + { + "audioType": "media", + "inputLatencyMicros": 0, + "outputLatencyMicros": 400000, + "type": 102, + }, + ], # 'build': '16.0', - 'deviceID': DEVICE_ID, - 'features': FEATURES, + "deviceID": DEVICE_ID, + "features": FEATURES, # 'features': 496155769145856, # Sonos One # 'firmwareBuildDate': 'Nov 5 2019', # 'firmwareRevision': '53.3-71050', # 'hardwareRevision': '1.21.1.8-2', - 'keepAliveLowPower': True, - 'keepAliveSendStatsAsBody': True, - 'manufacturer': 'Sonos', - 'model': 'One', - 'name': 'Camera da letto', - 'nameIsFactoryDefault': False, - 'pi': 'ba5cb8df-7f14-4249-901a-5e748ce57a93', # UUID generated casually.. - 'protocolVersion': '1.1', - 'sdk': 'AirPlay;2.0.2', - 'sourceVersion': '366.0', - 'statusFlags': 4, + "keepAliveLowPower": True, + "keepAliveSendStatsAsBody": True, + "manufacturer": "Sonos", + "model": "One", + "name": "Camera da letto", + "nameIsFactoryDefault": False, + "pi": "ba5cb8df-7f14-4249-901a-5e748ce57a93", # UUID generated casually.. + "protocolVersion": "1.1", + "sdk": "AirPlay;2.0.2", + "sourceVersion": "366.0", + "statusFlags": 4, # 'statusFlags': 0x404 # Sonos One - } + } if DISABLE_VM: volume = 0 - else: + else: volume = get_volume() second_stage_info = { "initialVolume": volume, - } + } sonos_one_setup = { - 'eventPort': 0, # AP2 receiver event server - 'timingPort': 0, - 'timingPeerInfo': { - 'Addresses': [ - IPV4, IPV6], - 'ID': IPV4} - } + "eventPort": 0, # AP2 receiver event server + # 'timingPort': 0, + # 'timingPeerInfo': { + # 'Addresses': [ + # IPV4, IPV6], + # 'ID': IPV4} + } sonos_one_setup_data = { - 'streams': [ - { - 'type': 96, - 'dataPort': 0, # AP2 receiver data server - 'controlPort': 0 # AP2 receiver control server - } - ] + "streams": [ + { + "type": 96, + "dataPort": 0, # AP2 receiver data server + "controlPort": 0, # AP2 receiver control server } + ] + } mdns_props = { - "srcvers": SERVER_VERSION, - "deviceid": DEVICE_ID, - "features": "%s,%s" % (hex(FEATURES & 0xffffffff), hex(FEATURES >> 32 & 0xffffffff)), - "flags": "0x4", - # "name": "GINO", # random - "model": "Airplay2-Receiver", # random - # "manufacturer": "Pino", # random - # "serialNumber": "01234xX321", # random - "protovers": "1.1", - "acl": "0", - "rsf": "0x0", - "fv": "p20.78000.12", - "pi": "5dccfd20-b166-49cc-a593-6abd5f724ddb", # UUID generated casually - "gid": "5dccfd20-b166-49cc-a593-6abd5f724ddb", # UUID generated casually - "gcgl": "0", - # "vn": "65537", - "pk": "de352b0df39042e201d31564049023af58a106c6d904b74a68aa65012852997f" - } + "srcvers": SERVER_VERSION, + "deviceid": DEVICE_ID, + "features": "%s,%s" + % (hex(FEATURES & 0xFFFFFFFF), hex(FEATURES >> 32 & 0xFFFFFFFF)), + "flags": "0x4", + # "name": "GINO", # random + "model": "Airplay2-Receiver", # random + # "manufacturer": "Pino", # random + # "serialNumber": "01234xX321", # random + "protovers": "1.1", + "acl": "0", + "rsf": "0x0", + "fv": "p20.78000.12", + "pi": "5dccfd20-b166-49cc-a593-6abd5f724ddb", # UUID generated casually + "gid": "5dccfd20-b166-49cc-a593-6abd5f724ddb", # UUID generated casually + "gcgl": "0", + # "vn": "65537", + "pk": "de352b0df39042e201d31564049023af58a106c6d904b74a68aa65012852997f", + } + class AP2Handler(http.server.BaseHTTPRequestHandler): @@ -177,7 +189,7 @@ def send_response(self, code, message=None): if code in self.responses: message = self.responses[code][0] else: - message = b'' + message = b"" response = "%s %d %s\r\n" % (self.protocol_version, code, message) self.wfile.write(response.encode()) @@ -200,7 +212,10 @@ def do_OPTIONS(self): self.send_response(200) self.send_header("Server", self.version_string()) self.send_header("CSeq", self.headers["CSeq"]) - self.send_header("Public", "ANNOUNCE, SETUP, RECORD, PAUSE, FLUSH, FLUSHBUFFERED, TEARDOWN, OPTIONS, POST, GET, PUT") + self.send_header( + "Public", + "ANNOUNCE, SETUP, RECORD, PAUSE, FLUSH, FLUSHBUFFERED, TEARDOWN, OPTIONS, POST, GET, PUT, SETRATEANCHORTIME", + ) self.end_headers() def do_FLUSHBUFFERED(self): @@ -221,7 +236,9 @@ def do_FLUSHBUFFERED(self): flush_from_seq = plist["flushFromSeq"] if "flushUntilSeq" in plist: flush_until_seq = plist["flushUntilSeq"] - self.server.streams[0].audio_connection.send("flush_from_until_seq-%i-%i" % (flush_from_seq, flush_until_seq)) + self.server.streams[0].audio_connection.send( + "flush_from_until_seq-%i-%i" % (flush_from_seq, flush_until_seq) + ) self.pp.pprint(plist) def do_POST(self): @@ -275,9 +292,10 @@ def do_SETUP(self): self.pp.pprint(plist) if "streams" not in plist: print("Sending EVENT:") - event_port, self.event_proc = Event.spawn() - sonos_one_setup["eventPort"] = event_port - print("[+] eventPort=%d" % event_port) + if self.server.event_proc is None: + event_port, self.server.event_proc = Event.spawn() + sonos_one_setup["eventPort"] = event_port + print("[+] eventPort=%d" % sonos_one_setup["eventPort"]) self.pp.pprint(sonos_one_setup) res = writePlistToString(sonos_one_setup) @@ -290,16 +308,23 @@ def do_SETUP(self): self.wfile.write(res) else: print("Sending CONTROL/DATA:") - - stream = Stream(plist["streams"][0]) + buff = ( + 8388608 # determines how many CODEC frame size 1024 we can hold + ) + stream = Stream(plist["streams"][0], buff, self.server.ptp_link) self.server.streams.append(stream) - sonos_one_setup_data["streams"][0]["controlPort"] = stream.control_port + sonos_one_setup_data["streams"][0][ + "controlPort" + ] = stream.control_port sonos_one_setup_data["streams"][0]["dataPort"] = stream.data_port - print("[+] controlPort=%d dataPort=%d" % (stream.control_port, stream.data_port)) + print( + "[+] controlPort=%d dataPort=%d" + % (stream.control_port, stream.data_port) + ) if stream.type == Stream.BUFFERED: sonos_one_setup_data["streams"][0]["type"] = stream.type - sonos_one_setup_data["streams"][0]["audioBufferSize"] = 8388608 + sonos_one_setup_data["streams"][0]["audioBufferSize"] = buff self.pp.pprint(sonos_one_setup_data) res = writePlistToString(sonos_one_setup_data) @@ -311,6 +336,17 @@ def do_SETUP(self): self.send_header("CSeq", self.headers["CSeq"]) self.end_headers() self.wfile.write(res) + + if "timingProtocol" in plist: + if plist["timingProtocol"] == "PTP": + if self.server.ptp_proc is None: + print("PTP Startup") + mac = int( + (ifen[ni.AF_LINK][0]["addr"]).replace(":", ""), 16 + ) + self.server.ptp_proc, self.server.ptp_link = PTP.spawn(mac) + else: + print("PTP reusing") return self.send_error(404) @@ -335,14 +371,17 @@ def do_GET_PARAMETER(self): if DISABLE_VM: res = b"volume: 0" + b"\r\n" else: - res = b"\r\n".join(b"%s: %s" % (k, v) for k, v in params_res.items()) + b"\r\n" + res = ( + b"\r\n".join(b"%s: %s" % (k, v) for k, v in params_res.items()) + + b"\r\n" + ) self.send_response(200) self.send_header("Content-Length", len(res)) self.send_header("Content-Type", HTTP_CT_PARAM) self.send_header("Server", self.version_string()) self.send_header("CSeq", self.headers["CSeq"]) self.end_headers() - hexdump(res); + hexdump(res) self.wfile.write(res) def do_SET_PARAMETER(self): @@ -371,14 +410,15 @@ def do_SET_PARAMETER(self): elif content_type == HTTP_CT_IMAGE: if content_len > 0: fname = None - with tempfile.NamedTemporaryFile(prefix="artwork", dir=".", delete=False) as f: + with tempfile.NamedTemporaryFile( + prefix="artwork", dir=".", delete=False, suffix=".jpg" + ) as f: f.write(self.rfile.read(content_len)) fname = f.name print("Artwork saved to %s" % fname) elif content_type == HTTP_CT_DMAP: if content_len > 0: - self.rfile.read(content_len) - print("Now plaing DAAP info. (need a daap parser here)") + parse_daap(self.rfile.read(content_len)) self.send_response(200) self.send_header("Server", self.version_string()) self.send_header("CSeq", self.headers["CSeq"]) @@ -409,7 +449,17 @@ def do_SETRATEANCHORTIME(self): plist = readPlistFromString(body) if plist["rate"] == 1: - self.server.streams[0].audio_connection.send("play-%i" % plist["rtpTime"]) + networkTime = plist["networkTimeSecs"] * (10 ** 9) + sample_bytes = plist["networkTimeFrac"].to_bytes( + 8, byteorder="big", signed=True + ) + uint64_sample = int.from_bytes(sample_bytes, byteorder="big") + nthFactor = 0.5 ** 64 + nanos = int(uint64_sample * nthFactor * (10 ** 9)) + networkTime += nanos + self.server.streams[0].audio_connection.send( + f'play-{plist["rtpTime"]}-{networkTime}' + ) if plist["rate"] == 0: self.server.streams[0].audio_connection.send("pause") self.pp.pprint(plist) @@ -441,12 +491,15 @@ def do_TEARDOWN(self): self.send_header("Server", self.version_string()) self.send_header("CSeq", self.headers["CSeq"]) self.end_headers() - + # Erase the hap() instance, otherwise reconnects fail self.server.hap = None - # terminate the forked event_proc, otherwise a zombie process consumes 100% cpu - self.event_proc.terminate() + # # terminate the forked event_proc, otherwise a zombie process consumes 100% cpu + # self.event_proc.terminate() + + # PTP tear-down + # self.ptp_proc.terminate() def do_SETPEERS(self): print("SETPEERS %s" % self.path) @@ -483,8 +536,14 @@ def handle_command(self): iplist = readPlistFromString(p) newin.append(iplist) plist["params"]["mrSupportedCommandsFromSender"] = newin - if "params" in plist["params"] and "kMRMediaRemoteNowPlayingInfoArtworkData" in plist["params"]["params"]: - plist["params"]["params"]["kMRMediaRemoteNowPlayingInfoArtworkData"] = "" + if ( + "params" in plist["params"] + and "kMRMediaRemoteNowPlayingInfoArtworkData" + in plist["params"]["params"] + ): + plist["params"]["params"][ + "kMRMediaRemoteNowPlayingInfoArtworkData" + ] = "" self.pp.pprint(plist) self.send_response(200) self.send_header("Server", self.version_string()) @@ -499,7 +558,7 @@ def handle_feedback(self): plist = readPlistFromString(body) # feedback logs are pretty much noise... - #self.pp.pprint(plist) + # self.pp.pprint(plist) self.send_response(200) self.send_header("Server", self.version_string()) self.send_header("CSeq", self.headers["CSeq"]) @@ -619,7 +678,9 @@ def handle_info(self): self.send_error(404) return else: - print("Content-Type: %s | Not implemented" % self.headers["Content-Type"]) + print( + "Content-Type: %s | Not implemented" % self.headers["Content-Type"] + ) self.send_error(404) else: res = writePlistToString(sonos_one_info) @@ -632,41 +693,45 @@ def handle_info(self): self.wfile.write(res) def upgrade_to_encrypted(self, shared_key): - self.request = self.server.upgrade_to_encrypted( - self.client_address, - shared_key) + self.request = self.server.upgrade_to_encrypted(self.client_address, shared_key) self.connection = self.request - self.rfile = self.connection.makefile('rb', self.rbufsize) - self.wfile = self.connection.makefile('wb') + self.rfile = self.connection.makefile("rb", self.rbufsize) + self.wfile = self.connection.makefile("wb") self.is_encrypted = True print("----- ENCRYPTED CHANNEL -----") + def register_mdns(receiver_name): addresses = [] for ifen in ni.interfaces(): ifenaddr = ni.ifaddresses(ifen) if ni.AF_INET in ifenaddr: - addresses.append(socket.inet_pton(ni.AF_INET, - ifenaddr[ni.AF_INET][0]["addr"])) + addresses.append( + socket.inet_pton(ni.AF_INET, ifenaddr[ni.AF_INET][0]["addr"]) + ) if ni.AF_INET6 in ifenaddr: - addresses.append(socket.inet_pton(ni.AF_INET6, - ifenaddr[ni.AF_INET6][0]["addr"].split("%")[0])) + addresses.append( + socket.inet_pton( + ni.AF_INET6, ifenaddr[ni.AF_INET6][0]["addr"].split("%")[0] + ) + ) info = ServiceInfo( - "_airplay._tcp.local.", - "%s._airplay._tcp.local." % receiver_name, - # addresses=[socket.inet_aton("127.0.0.1")], - addresses=addresses, - port=7000, - properties=mdns_props, - server="%s.local." % receiver_name, - ) + "_airplay._tcp.local.", + "%s._airplay._tcp.local." % receiver_name, + # addresses=[socket.inet_aton("127.0.0.1")], + addresses=addresses, + port=7000, + properties=mdns_props, + server="%s.local." % receiver_name, + ) zeroconf = Zeroconf(ip_version=IPVersion.V4Only) zeroconf.register_service(info) print("mDNS service registered") return (zeroconf, info) + def unregister_mdns(zeroconf, info): print("Unregistering...") zeroconf.unregister_service(info) @@ -675,7 +740,7 @@ def unregister_mdns(zeroconf, info): def get_free_port(): free_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - free_socket.bind(('0.0.0.0', 0)) + free_socket.bind(("0.0.0.0", 0)) free_socket.listen(5) port = free_socket.getsockname()[1] free_socket.close() @@ -683,15 +748,16 @@ def get_free_port(): class AP2Server(socketserver.TCPServer): - def __init__(self, addr_port, handler): super().__init__(addr_port, handler) self.connections = {} self.hap = None self.enc_layer = False self.streams = [] + self.event_proc = None + self.ptp_proc = None - #Override + # Override def get_request(self): client_socket, client_addr = super().get_request() print("Got connection with %s:%d" % client_addr) @@ -704,13 +770,22 @@ def upgrade_to_encrypted(self, client_address, shared_key): self.connections[client_address] = hap_socket return hap_socket + if __name__ == "__main__": multiprocessing.set_start_method("spawn") - parser = argparse.ArgumentParser(prog='AirPlay 2 receiver') + parser = argparse.ArgumentParser(prog="AirPlay 2 receiver") parser.add_argument("-m", "--mdns", required=True, help="mDNS name to announce") - parser.add_argument("-n", "--netiface", required=True, help="Network interface to bind to") - parser.add_argument("-nv", "--no-volume-management", required=False, help="Disable volume management", action='store_true') + parser.add_argument( + "-n", "--netiface", required=True, help="Network interface to bind to" + ) + parser.add_argument( + "-nv", + "--no-volume-management", + required=False, + help="Disable volume management", + action="store_true", + ) parser.add_argument("-f", "--features", required=False, help="Features") args = parser.parse_args() @@ -727,9 +802,18 @@ def upgrade_to_encrypted(self, client_address, shared_key): exit(-1) except Exception: print("[!] Network interface not found") + print("Available network interfaces:") + for p in ni.interfaces(): + print(p) + addrs = ni.ifaddresses(p) + for a in addrs: + # print(' ' + str(a)) + for ak in addrs[a]: + for akx in ak: + if str(akx) == "addr": + print(" " + str(akx) + ": " + str(ak[akx])) exit(-1) - DEVICE_ID = ifen[ni.AF_LINK][0]["addr"] IPV4 = ifen[ni.AF_INET][0]["addr"] IPV6 = ifen[ni.AF_INET6][0]["addr"].split("%")[0] diff --git a/ap2/connections/audio.py b/ap2/connections/audio.py index a1f5f78..756c575 100644 --- a/ap2/connections/audio.py +++ b/ap2/connections/audio.py @@ -1,6 +1,5 @@ import socket import string -import struct import multiprocessing import enum import threading @@ -14,6 +13,8 @@ from ..utils import get_logger, get_free_tcp_socket, get_free_udp_socket +from ap2.connections.ptp_time import PTP + class RTP: def __init__(self, data): @@ -22,8 +23,8 @@ def __init__(self, data): self.extension = (data[0] & 0b00010000) >> 4 self.csrc_count = data[0] & 0b00001111 - self.timestamp = struct.unpack(">I", data[4:8])[0] - self.ssrc = struct.unpack(">I", data[8:12])[0] + self.timestamp = int.from_bytes(data[4:8], byteorder="big") + self.ssrc = int.from_bytes(data[8:12], byteorder="big") self.nonce = data[-8:] self.tag = data[-24:-8] @@ -36,7 +37,7 @@ def __init__(self, data): super(RTP_REALTIME, self).__init__(data) self.payload_type = data[1] & 0b01111111 self.marker = (data[1] & 0b10000000) >> 7 - self.sequence_no = struct.unpack(">H", data[2:4])[0] + self.sequence_no = int.from_bytes(data[2:4], byteorder="big") class RTP_BUFFERED(RTP): @@ -44,15 +45,16 @@ def __init__(self, data): super(RTP_BUFFERED, self).__init__(data) self.payload_type = 0 self.marker = 0 - self.sequence_no = struct.unpack('>I', b'\0' + data[1:4])[0] + self.sequence_no = int.from_bytes(b"\0" + data[1:4], byteorder="big") # Very simple circular buffer implementation class RTPBuffer: # TODO : Centralized for both this buffer size and audioBufferSize returned by SETUP - BUFFER_SIZE = 8192 + BUFFER_SIZE = 1 - def __init__(self): + def __init__(self, size): + self.BUFFER_SIZE = size self.buffer_array = numpy.empty(self.BUFFER_SIZE, dtype=RTP_BUFFERED) # Stores indexes only for quick bisect search self.buffer_array_seqs = numpy.empty(self.BUFFER_SIZE, dtype=int) @@ -65,21 +67,23 @@ def __init__(self): self.flush_to_sequence = None def increment_index(self, index): - if index < self.BUFFER_SIZE - 1: - return index + 1 - else: - return 0 + return (index + 1) % self.BUFFER_SIZE def decrement_index(self, index): - if index == 0: - return self.BUFFER_SIZE - 1 - else: - return index - 1 + return (index + self.BUFFER_SIZE - 1) % self.BUFFER_SIZE def add(self, rtp_data): - #print("write - %i %i" % (self.read_index, self.write_index)) + # print("write - %i %i" % (self.read_index, self.write_index)) if self.write_index % 1000 == 0: - print("buffer: writing - full at %s - ri=%i - wi=%i - seq=%i" % ("{:.1%}".format(self.get_fullness()), self.read_index, self.write_index, rtp_data.sequence_no)) + print( + "buffer: writing - full at %s - ri=%i - wi=%i - seq=%i" + % ( + "{:.1%}".format(self.get_fullness()), + self.read_index, + self.write_index, + rtp_data.sequence_no, + ) + ) used_index = self.write_index self.buffer_array[self.write_index] = rtp_data @@ -90,7 +94,7 @@ def add(self, rtp_data): else: if self.increment_index(self.write_index) == self.read_index: # buffer overflow, we "push" the read index - print("buffer: overrrun") + print("buffer: over-run") self.read_index = self.increment_index(self.read_index) self.write_index = self.increment_index(self.write_index) @@ -101,7 +105,7 @@ def get(self): def can_read(self): return self.read_index != -1 - + def next(self): # print("read - %i %i" % (self.read_index, self.write_index)) if self.read_index == -1: @@ -110,7 +114,15 @@ def next(self): else: buffered_object = self.buffer_array[self.read_index] if self.read_index % 1000 == 0: - print("buffer: reading - full at %s - ri=%i - wi=%i - seq=%i" % ("{:.1%}".format(self.get_fullness()), self.read_index, self.write_index,buffered_object.sequence_no)) + print( + "buffer: reading - full at %s - ri=%i - wi=%i - seq=%i" + % ( + "{:.1%}".format(self.get_fullness()), + self.read_index, + self.write_index, + buffered_object.sequence_no, + ) + ) if self.increment_index(self.read_index) == self.write_index: # buffer underrun, nothing we can do @@ -121,35 +133,44 @@ def next(self): return buffered_object + def previous(self): + self.read_index = self.decrement_index(self.read_index) + buffered_object = self.buffer_array[self.read_index] + return buffered_object + def get_fullness(self): - write_index = self.write_index - read_index = self.read_index - if read_index < write_index: - fill = write_index - read_index - else: - fill = self.BUFFER_SIZE - read_index + write_index - return fill / self.BUFFER_SIZE + # get distance between read and write in relation to buff size + return ( + (self.BUFFER_SIZE + self.write_index - self.read_index) % self.BUFFER_SIZE + ) / self.BUFFER_SIZE def get_bounds(self): - if self.read_index<= self.write_index: + if self.read_index <= self.write_index: return self.read_index, self.write_index else: return self.write_index, self.read_index def find_seq(self, seq): - start_index = self.read_index - end_index = self.write_index + # do binary search. Bin = O(log n) vs linear O(n) + # here we iterate max several times + l = self.read_index + r = self.write_index # len(self.buffer_array) - 1 - if start_index == -1: + if l == -1: + return + if l == r: return - while True: - if start_index == end_index: - return - if self.buffer_array[start_index].sequence_no == seq: - return start_index - else: - start_index = self.increment_index(start_index) + while l <= r: + m = (l + r // 2) % self.BUFFER_SIZE + # print('searching l=%d, r=%d, m=%d, srch=%d, now_at=%d' % \ + # (l, r, m, seq, self.buffer_array_seqs[m] )) + if self.buffer_array_seqs[m] == seq: + return m + if self.buffer_array_seqs[m] < seq: + l = self.increment_index(m) + elif self.buffer_array_seqs[m] > seq: + l = self.decrement_index(m) # initialize buffer for reading def init(self): @@ -163,6 +184,7 @@ def flush_write(self, index_from): else: return False + class Audio: class AudioFormat(enum.Enum): PCM_8000_16_1 = 1 << 2 @@ -197,46 +219,167 @@ class AudioFormat(enum.Enum): AAC_ELD_44100_1 = 1 << 31 AAC_ELD_48000_1 = 1 << 32 - def __init__(self, session_key, audio_format): - if audio_format != Audio.AudioFormat.ALAC_44100_16_2.value \ - and audio_format != Audio.AudioFormat.AAC_LC_44100_2.value: - raise Exception("Unsupported format: %s", Audio.AudioFormat(audio_format)).name + @staticmethod + def set_audio_params(self, audio_format): + # defaults + self.sample_rate = 44100 + self.sample_size = 16 + self.channel_count = 2 + self.af = af = str(Audio.AudioFormat(audio_format)) + + if "8000" in af: + self.sample_rate = 8000 + elif "16000" in af: + self.sample_rate = 16000 + elif "24000" in af: + self.sample_rate = 24000 + elif "32000" in af: + self.sample_rate = 32000 + elif "44100" in af: + self.sample_rate = 44100 + elif "48000" in af: + self.sample_rate = 48000 + else: # default + self.sample_rate = 44100 + + if "_16" in af: + self.sample_size = 16 + elif "_24" in af: + self.sample_size = 24 + else: # default + self.sample_size = 16 + + if af.endswith("_1"): + self.channel_count = 1 + else: + self.channel_count = 2 + + print("Negotiated audio format: ", Audio.AudioFormat(audio_format)) + + # + + def __init__(self, session_key, audio_format, buff_size): self.audio_format = audio_format self.session_key = session_key - self.rtp_buffer = RTPBuffer() + self.rtp_buffer = RTPBuffer(buff_size) + self.set_audio_params(self, audio_format) + + @staticmethod + def set_alac_extradata(self, sample_rate, sample_size, channel_count): + extradata = bytes() # a 36-byte QuickTime atom passed through as extradata + extradata += (36).to_bytes(4, byteorder="big") # 32 bits atom size + extradata += ("alac").encode() # 32 bits tag ('alac') + extradata += (0).to_bytes(4, byteorder="big") # 32 bits tag version (0) + extradata += (352).to_bytes(4, byteorder="big") # 32 bits samples per frame + extradata += (0).to_bytes( + 1, byteorder="big" + ) # 8 bits compatible version (0) + extradata += (sample_size).to_bytes(1, byteorder="big") # 8 bits sample size + extradata += (40).to_bytes( + 1, byteorder="big" + ) # 8 bits history mult (40) + extradata += (10).to_bytes( + 1, byteorder="big" + ) # 8 bits initial history (10) + extradata += (14).to_bytes( + 1, byteorder="big" + ) # 8 bits rice param limit (14) + extradata += (channel_count).to_bytes(1, byteorder="big") # 8 bits channels + extradata += (255).to_bytes( + 2, byteorder="big" + ) # 16 bits maxRun (255) + extradata += (0).to_bytes( + 4, byteorder="big" + ) # 32 bits max coded frame size (0 means unknown) + extradata += (0).to_bytes( + 4, byteorder="big" + ) # 32 bits average bitrate (0 means unknown) + extradata += (sample_rate).to_bytes(4, byteorder="big") # 32 bits samplerate + return extradata def init_audio_sink(self): + codecLatencySec = 0 self.pa = pyaudio.PyAudio() - self.sink = self.pa.open(format=self.pa.get_format_from_width(2), - channels=2, - rate=44100, - output=True) - codec = None + self.use_callback = True + + if self.use_callback: + self.sink = self.pa.open( + format=self.pa.get_format_from_width(2), + channels=self.channel_count, + rate=self.sample_rate, + output=True, + stream_callback=self.callback, + start=False, + ) + else: + self.sink = self.pa.open( + format=self.pa.get_format_from_width(2), + channels=self.channel_count, + rate=self.sample_rate, + output=True, + ) + # nice Python3 crash if we don't check self.sink is null. Not harmful, but should check. + if not self.sink: + exit() + # codec = None extradata = None if self.audio_format == Audio.AudioFormat.ALAC_44100_16_2.value: - extradata = bytes([ - # Offset 0x00000000 to 0x00000035 - 0x00, 0x00, 0x00, 0x24, 0x61, 0x6c, 0x61, 0x63, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x01, 0x60, 0x00, 0x10, 0x28, 0x0a, 0x0e, 0x02, 0x00, 0xff, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xac, 0x44 - ]) - codec = av.codec.Codec('alac', 'r') - elif self.audio_format == Audio.AudioFormat.AAC_LC_44100_2.value: - codec = av.codec.Codec('aac', 'r') - - if codec is not None: - self.codecContext = av.codec.CodecContext.create(codec) - self.codecContext.sample_rate = 44100 - self.codecContext.channels = 2 - self.codecContext.format = AudioFormat('s16p') + extradata = self.set_alac_extradata(self, 44100, 16, 2) + elif self.audio_format == Audio.AudioFormat.ALAC_44100_24_2.value: + extradata = self.set_alac_extradata(self, 44100, 24, 2) + elif self.audio_format == Audio.AudioFormat.ALAC_48000_16_2.value: + extradata = self.set_alac_extradata(self, 48000, 16, 2) + elif self.audio_format == Audio.AudioFormat.ALAC_48000_24_2.value: + extradata = self.set_alac_extradata(self, 48000, 24, 2) + + if "ALAC" in self.af: + self.codec = av.codec.Codec("alac", "r") + elif "AAC" in self.af: + self.codec = av.codec.Codec("aac", "r") + elif "OPUS" in self.af: + self.codec = av.codec.Codec("opus", "r") + # PCM - not sure which. Easy to fix. + elif "PCM" and "_16_" in self.af: + self.codec = av.codec.Codec("pcm_s16be_planar", "r") + elif "PCM" and "_24_" in self.af: + self.codec = av.codec.Codec("pcm_s24be", "r") + + """ + #It seems that these are not required. + if 'ELD' in self.af: + codecLatencySec = (2017 / self.sample_rate) + elif'AAC_LC'in self.af: + codecLatencySec = (2624 / self.sample_rate) + codecLatencySec = 0 + print('codecLatencySec:',codecLatencySec) + """ + + if self.codec is not None: + self.codecContext = av.codec.CodecContext.create(self.codec) + self.codecContext.sample_rate = self.sample_rate + self.codecContext.channels = self.channel_count + self.codecContext.format = AudioFormat("s" + str(self.sample_size) + "p") if extradata is not None: self.codecContext.extradata = extradata self.resampler = av.AudioResampler( - format=av.AudioFormat('s16').packed, - layout='stereo', - rate=44100, + format=av.AudioFormat("s" + str(self.sample_size)).packed, + layout="stereo", + rate=self.sample_rate, + ) + + audioDevicelatency = self.pa.get_default_output_device_info()[ + "defaultHighOutputLatency" + ] + # defaultLowOutputLatency is also available + print(f"audioDevicelatency (sec): {audioDevicelatency:0.5f}") + pyAudioDelay = self.sink.get_output_latency() + print(f"pyAudioDelay (sec): {pyAudioDelay:0.5f}") + ptpDelay = 0.002 + self.sample_delay = ( + pyAudioDelay + audioDevicelatency + codecLatencySec + ptpDelay ) + print(f"Total sample_delay (sec): {self.sample_delay:0.5f}") def decrypt(self, rtp): c = ChaCha20_Poly1305.new(key=self.session_key, nonce=rtp.nonce) @@ -245,11 +388,20 @@ def decrypt(self, rtp): return data def handle(self, rtp): - self.logger.debug("v=%d p=%d x=%d cc=%d m=%d pt=%d seq=%d ts=%d ssrc=%d" % (rtp.version, rtp.padding, - rtp.extension, rtp.csrc_count, - rtp.marker, rtp.payload_type, - rtp.sequence_no, rtp.timestamp, - rtp.ssrc)) + self.logger.debug( + "v=%d p=%d x=%d cc=%d m=%d pt=%d seq=%d ts=%d ssrc=%d" + % ( + rtp.version, + rtp.padding, + rtp.extension, + rtp.csrc_count, + rtp.marker, + rtp.payload_type, + rtp.sequence_no, + rtp.timestamp, + rtp.ssrc, + ) + ) def process(self, rtp): data = self.decrypt(rtp) @@ -258,29 +410,34 @@ def process(self, rtp): frame = self.resampler.resample(frame) return frame.planes[0].to_bytes() - def run(self, parent_reader_connection): + def run(self, parent_reader_connection, ptp_link): # This pipe is between player (read data) and server (write data) parent_writer_connection, writer_connection = multiprocessing.Pipe() server_thread = threading.Thread(target=self.serve, args=(writer_connection,)) - player_thread = threading.Thread(target=self.play, args=(parent_reader_connection,parent_writer_connection)) + player_thread = threading.Thread( + target=self.play, + args=(parent_reader_connection, parent_writer_connection, ptp_link), + ) server_thread.start() player_thread.start() @classmethod - def spawn(cls, session_key, audio_format): - audio = cls(session_key, audio_format) + def spawn(cls, session_key, audio_format, buff, ptp_link): + audio = cls(session_key, audio_format, buff) # This pipe is reachable from receiver parent_reader_connection, audio.audio_connection = multiprocessing.Pipe() - mainprocess = multiprocessing.Process(target=audio.run, args=(parent_reader_connection,)) + mainprocess = multiprocessing.Process( + target=audio.run, args=(parent_reader_connection, ptp_link) + ) mainprocess.start() return audio.port, mainprocess, audio.audio_connection -class AudioRealtime(Audio): - def __init__(self, session_key, audio_format): - super(AudioRealtime, self).__init__(session_key, audio_format) +class AudioRealtime(Audio): + def __init__(self, session_key, audio_format, buff): + super(AudioRealtime, self).__init__(session_key, audio_format, buff) self.socket = get_free_udp_socket() self.port = self.socket.getsockname()[1] @@ -289,7 +446,7 @@ def fini_audio_sink(self): self.pa.terminate() def play(self, rtspconn, serverconn): - # for now RT do no use RTPBuffer at all, we don't use this method + # for now RealTime does not use RTPBuffer at all, we don't use this method pass def serve(self, playerconn): @@ -312,31 +469,41 @@ def serve(self, playerconn): class AudioBuffered(Audio): - def __init__(self, session_key, audio_format): - super(AudioBuffered, self).__init__(session_key, audio_format) + def __init__(self, session_key, audio_format, buff): + super(AudioBuffered, self).__init__(session_key, audio_format, buff) self.socket = get_free_tcp_socket() self.port = self.socket.getsockname()[1] self.anchorMonotonicTime = None self.anchorRtpTime = None + # DEVICE_ID = '9c:b6:d0:f2:3d:29' + + # ptp_p, ptp_pipe = PTP.spawn(int(DEVICE_ID.replace(":",""), 16)) + # self.ptp_pipe = ptp_pipe + + self.prev_nt = 0 + self.prev_rt = 0 + def get_time_offset(self, rtp_ts): if not self.anchorRtpTime: return 0 rtptime_offset = rtp_ts - self.anchorRtpTime realtime_offset_ms = (time.monotonic_ns() - self.anchorMonotonicTime) / 10 ** 6 - # TODO: replace 44100 with real framerate - time_offset_ms = (1000 * rtptime_offset / 44100) - realtime_offset_ms + time_offset_ms = (1000 * rtptime_offset / self.sample_rate) - realtime_offset_ms return time_offset_ms def get_min_timestamp(self): realtime_offset_sec = (time.monotonic_ns() - self.anchorMonotonicTime) / 10 ** 9 - print("player: get_min_timestamp - realtime_offset_sec={:06.4f}".format(realtime_offset_sec)) - res = self.anchorRtpTime + realtime_offset_sec * 44100 + print( + "player: get_min_timestamp - realtime_offset_sec={:06.4f}".format( + realtime_offset_sec + ) + ) + res = self.anchorRtpTime + realtime_offset_sec * self.sample_rate print("player: get_min_timestamp return=%i" % res) return res - def forward(self, requested_timestamp): finished = False while not finished: @@ -346,16 +513,76 @@ def forward(self, requested_timestamp): finished = True else: pass - #print("player: still forwarding.. ts=%i" % rtp.timestamp) + # print("player: still forwarding.. ts=%i" % rtp.timestamp) else: print("player: !!! error while forwarding !!!") finished = True - # player moves readindex in buffer - def play(self, rtspconn, serverconn): + def callback(self, in_data, frame_count, time_info, status): + + self.ptp_link.send("get_ptp_master_nanos_timestamped") + if self.ptp_link.poll(1): + network_time_ns, network_time_monotonic_ts = self.ptp_link.recv() + time_monotonic_ns = time.monotonic_ns() + network_time_ns += time_monotonic_ns - network_time_monotonic_ts + else: + return + + rtp = self.rtp_buffer.next() + if not rtp: + print(f"callback {frame_count} no more data") + return (None, pyaudio.paAbort) + + # player_time_offset = time_info.output_buffer_dac_time - self.sink_anchor + + # rtptime_offset = rtp.timestamp - self.anchorRtpTime + # rtpTime_offset_ms = 1000 * rtptime_offset / self.sample_rate + # nt_offset_ms = (network_time_ns - self.anchorNetworkTime) / (10 ** 6) + # time_offset_ms = rtpTime_offset_ms - nt_offset_ms - self.sample_delay * 1000 + + # time_offset_ms = rtpTime_offset_ms - nt_offset_ms - self.sample_delay * 1000 + # 0 = rtpTime_offset_ms - nt_offset_ms + # 0 = (1000 * rtptime_offset / self.sample_rate) - ((network_time_ns - self.anchorNetworkTime) / (10 ** 6)) + # ((network_time_ns - self.anchorNetworkTime) / (10 ** 6)) = (1000 * rtptime_offset / self.sample_rate) + # (network_time_ns - self.anchorNetworkTime) / (10 ** 6) = 1000 * (rtp.timestamp - self.anchorRtpTime) / self.sample_rate + # 1000 * (rtp.timestamp - self.anchorRtpTime) / self.sample_rate = (network_time_ns - self.anchorNetworkTime) / (10 ** 6) + # rtp.timestamp - self.anchorRtpTime = ((network_time_ns - self.anchorNetworkTime) / (10 ** 6)) / 1000 * self.sample_rate + # rtp_timestamp = ((network_time_ns - self.anchorNetworkTime) / (10 ** 6)) / 1000 * self.sample_rate + self.anchorRtpTime + + dac_offset = time_info["output_buffer_dac_time"] - time_info["current_time"] + + rtp_timestamp = ( + (network_time_ns - self.anchorNetworkTime) / (10 ** 9) + dac_offset + ) * self.sample_rate + self.anchorRtpTime + + # print( + # f"callback {frame_count} {rtp.timestamp} {time_info['output_buffer_dac_time']} {time_info['current_time']} ts: {rtp_timestamp} dac offset {dac_offset}" + # ) + skip = 0 + while rtp_timestamp - rtp.timestamp > 1024: + rtp = self.rtp_buffer.next() + if rtp is None: + return + skip += 1 + if skip != 0: + print(f"skipped {skip}") + + back = 0 + while skip == 0 and rtp.timestamp - rtp_timestamp > 1024: + rtp = self.rtp_buffer.previous() + back += 1 + if back > 0: + print(f"went back {back}") + + audio = self.process(rtp) + return (audio, pyaudio.paContinue) + + # player moves readindex in buffer + def play(self, rtspconn, serverconn, ptp_link): playing = False data_ready = False data_ontime = True + self.ptp_link = ptp_link i = 0 while True: if not playing: @@ -367,13 +594,17 @@ def play(self, rtspconn, serverconn): else: server_timeout = 0 - if self.rtp_buffer.can_read(): + if not data_ready and self.rtp_buffer.get_fullness() > 0.2: + print( + f"setting data ready at buffer fullness {self.rtp_buffer.get_fullness()}" + ) data_ready = True if serverconn.poll(server_timeout): message = serverconn.recv() if message == "data_ready": data_ready = True + print(f"setting data ready at from server") elif message == "data_ontime_response": print("player: ontime data response received") ts = self.get_min_timestamp() @@ -386,41 +617,101 @@ def play(self, rtspconn, serverconn): message = rtspconn.recv() if str.startswith(message, "play"): self.anchorMonotonicTime = time.monotonic_ns() - self.anchorRtpTime = int(str.split(message, "-")[1]) + msg_data = str.split(message, "-") + self.anchorRtpTime = int(msg_data[1]) + self.anchorNetworkTime = int(msg_data[2]) playing = True elif message == "pause": playing = False data_ready = False + print("pause event") + if self.use_callback: + print(" stopping stream") + self.sink.stop_stream() elif str.startswith(message, "flush_from_until_seq"): - pending_flush_from_seq, pending_flush_until_seq = str.split(message, "-")[-2:] + pending_flush_from_seq, pending_flush_until_seq = str.split( + message, "-" + )[-2:] pending_flush_from_seq = int(pending_flush_from_seq) pending_flush_until_seq = int(pending_flush_until_seq) - print("player: request flush received from-until %i-%i" % (pending_flush_from_seq, pending_flush_until_seq)) - print("player: relay message to server to flush from-until sequence %i-%i" % (pending_flush_from_seq, pending_flush_until_seq)) + print( + "player: request flush received from-until %i-%i" + % (pending_flush_from_seq, pending_flush_until_seq) + ) + print( + "player: relay message to server to flush from-until sequence %i-%i" + % (pending_flush_from_seq, pending_flush_until_seq) + ) serverconn.send(message) if playing and data_ready: - rtp = self.rtp_buffer.next() - if rtp: - time_offset_ms = self.get_time_offset(rtp.timestamp) - if i % 100 == 0: - print("player: offset is %i ms" % time_offset_ms) - if time_offset_ms >= 100: - print("player: offset %i ms too big - seq = %i - sleeping %s sec" % (time_offset_ms, rtp.sequence_no, "{:05.2f}".format(time_offset_ms /1000))) - time.sleep(time_offset_ms / 1000) - elif time_offset_ms < -100: - print("player: offset %i ms too low - seq = %i - sending ontime data request" % (time_offset_ms, rtp.sequence_no)) - # request on_time data message - serverconn.send("on_time_data_request") - data_ontime = False - - audio = self.process(rtp) - self.sink.write(audio) - i += 1 + if self.use_callback: + if not self.sink.is_active(): + print("starting stream") + self.sink.start_stream() + continue # use callback + else: + ptp_link.send("get_ptp_master_nanos_timestamped") + if ptp_link.poll(10): + network_time_ns, network_time_monotonic_ts = ptp_link.recv() + time_monotonic_ns = time.monotonic_ns() + network_time_ns += time_monotonic_ns - network_time_monotonic_ts + + # nt_delta = (network_time_ns - self.prev_nt ) / (10**6) + # rt_delta = (time_monotonic_ns - self.prev_rt) / (10**6) + # # expected 23.21ms + # # it does happens at this rate but only on average, values are never 23.2 and mostly a combination of 15-16 and 31-32 + # #if nt_delta > 30 or nt_delta < 10: + # #print(f'{i} nt_delta: {nt_delta} rt_delta: {rt_delta}') + # self.prev_nt = network_time_ns + # self.prev_rt = time_monotonic_ns + else: + return + + while True: + rtp = self.rtp_buffer.next() + + if rtp: + # time_offset_ms = self.get_time_offset(rtp.timestamp) + + rtptime_offset = rtp.timestamp - self.anchorRtpTime + rtpTime_offset_ms = 1000 * rtptime_offset / self.sample_rate + nt_offset_ms = ( + network_time_ns - self.anchorNetworkTime + ) / (10 ** 6) + time_offset_ms = ( + rtpTime_offset_ms + - nt_offset_ms + - self.sample_delay * 1000 + ) + + if i % 100 == 0: + print(f"player: offset is {time_offset_ms} ms") + if time_offset_ms >= 50: + sleep_time = time_offset_ms / (10 ** 3) + print( + f"player: offset {time_offset_ms} ms too big - seq = {rtp.sequence_no} - sleeping {sleep_time} sec" + ) + # time.sleep(time_offset_ms / 1000) + time.sleep(sleep_time) + elif time_offset_ms < -50: + # get another rtp? + print(f"player: offset too low, skipping 23.2 ms") + continue + # print("player: offset %i ms too low - seq = %i - sending ontime data request" % (time_offset_ms, rtp.sequence_no)) + + # # request on_time data message + # serverconn.send("on_time_data_request") + # data_ontime = False + + audio = self.process(rtp) + self.sink.write(audio) + i += 1 + break # server moves write index in buffer # the exception to this rule is the buffer initialization (init call) @@ -437,13 +728,18 @@ def serve(self, playerconn): message = playerconn.recv() if str.startswith(message, "flush_from_until_seq"): print("server: flush request received: %s" % message) - pending_flush_from_seq, pending_flush_until_seq = str.split(message, "-")[-2:] + pending_flush_from_seq, pending_flush_until_seq = str.split( + message, "-" + )[-2:] pending_flush_from_seq = int(pending_flush_from_seq) seq_to_overtake = int(pending_flush_until_seq) from_index = self.rtp_buffer.find_seq(pending_flush_from_seq) if from_index: if self.rtp_buffer.flush_write(from_index): - print("server: successfully flushed - write index moved to %i" % from_index) + print( + "server: successfully flushed - write index moved to %i" + % from_index + ) else: print("server: flush did not move write index") elif message == "on_time_data_request": @@ -452,24 +748,30 @@ def serve(self, playerconn): message = conn.recv(2, socket.MSG_WAITALL) if message: - data_len = struct.unpack(">H", message)[0] + data_len = int.from_bytes(message, byteorder="big") data = conn.recv(data_len - 2, socket.MSG_WAITALL) rtp = RTP_BUFFERED(data) self.handle(rtp) time_offset_ms = self.get_time_offset(rtp.timestamp) - #print("server: writing seq %i timeoffset %i" % (rtp.sequence_no, time_offset_ms)) + # print("server: writing seq %i timeoffset %i" % (rtp.sequence_no, time_offset_ms)) if seq_to_overtake is None: self.rtp_buffer.add(rtp) else: - print("server: searching sequence %i - current is %i" % (seq_to_overtake, rtp.sequence_no)) + print( + "server: searching sequence %i - current is %i" + % (seq_to_overtake, rtp.sequence_no) + ) # do not write data if it is expired if rtp.sequence_no >= seq_to_overtake: if pending_flush_from_seq == 0: print("server: buffer initialisation") self.rtp_buffer.init() self.rtp_buffer.add(rtp) - print("server: requested sequence to overtake %i - receiving sequence %i" % (seq_to_overtake, rtp.sequence_no)) + print( + "server: requested sequence to overtake %i - receiving sequence %i" + % (seq_to_overtake, rtp.sequence_no) + ) # as soon as we overtake seq_to_overtake sequence, let's inform the player playerconn.send("data_ready") seq_to_overtake = None @@ -483,4 +785,4 @@ def serve(self, playerconn): pass finally: conn.close() - self.socket.close() \ No newline at end of file + self.socket.close() diff --git a/ap2/connections/ptp_time.py b/ap2/connections/ptp_time.py new file mode 100644 index 0000000..58bfae3 --- /dev/null +++ b/ap2/connections/ptp_time.py @@ -0,0 +1,1248 @@ +""" +# Simple, naïve PTP implementation in Python + +# Basic listening and sync ability. Listens only to UDP unicast on ports 319+20. +# - systemcrash 2021 +# Airplay only cares about *relative* sync, as does this implementation. +# No absolute or NTP references. It currently only slaves to other master clocks +# and follows the PTP election mechanism for grand masters, then syncs to those. +# This implementation also assumes subDomain is 0. +# Apple Airplay uses unicast, not multi. It is specified in e.g.: +# Apple Vendor PTP Profile 2017 +# License: GPLv2 + +Most behaviour in here is derived from PTP within AirPlay. Assume that Apple has its own +PTP Profile. So unless otherwise stated here, the values here apply to Apple's profile. + +""" + +import socket +import select +import threading +import multiprocessing +import enum +from enum import Flag +import random +import time +from collections import deque + +""" +# UDP dest port: 319 for Sync, Delay_Req, Pdelay_Req, Pdelay_Resp; +# UDP dest port: 320 for other messages. +# Sources for this implementation: +# http://www.chronos.co.uk/files/pdfs/cal/TechnicalBrief-IEEE1588v2PTP.pdf +# http://ithitman.blogspot.com/2015/03/precision-time-protocol-ptp-demystified.html +# https://github.com/boundary/wireshark/blob/master/epan/dissectors/packet-ptp.c +# https://github.com/ptpd/ptpd/tree/master/src +# https://www.nist.gov/system/files/documents/el/isd/ieee/tutorial-basic.pdf +# https://www.ieee802.org/1/files/public/docs2008/as-garner-1588v2-summary-0908.pdf +# in 2 step, we see Announce, Del_req, Del_resp, Followup, Sig, Sync + + +# port 319/320 UDP +# first 4 bytes of PTP packets +self.v1_compat # 4 bits +self.msg_type # 4 bits +# self.reserved00 # 1 byte +self.ptp_version # 1 byte +self.msgLength # 2 bytes +self.subdomainNumber # 1 byte +self.reserved01 # 1 byte +self.flags # 2 bytes = 16 bits +self.correctionNanoseconds # 6 bytes = 48 bits +self.correctionSubNanoseconds # 2 bytes = 16 bits +self.reserved02 # 4 bytes +self.ClockIdentity # 8 bytes - typically sender mac, often with fffe in the middle +self.SourcePortID # 2 bytes = 16 bits +self.sequenceID # 2 bytes = 16 bits +self.control # 1 byte +self.logMessagePeriod # 1 byte +# Delay_Req message +self.originTimestampSec # 6 bytes - seconds +self.originTimestampNanoSec # 4 bytes - nanoseconds +# Delay_Resp message +self.rcvTimestampSec # 6 bytes - seconds +self.rcvTimestampNanoSec # 4 bytes - nanoseconds +self.requestingSrcPortIdentity # 8 bytes - mac address +self.requestingSrcPortID # 2 bytes - port number +# Signalling message +self.targetPortIdentity # 8 bytes - mac address +self.targetPortID # 2 bytes - port number +self.tlvType # 2 bytes +self.tlvLen # 2 bytes +self.orgId # 3 bytes (first half of mac) +self.orgSubType # 3 bytes = 01 +""" + +""" +Apple PTP Limits +ppmLimit 10000 +ppmNumerator 10000 +ppmDenominator 1000000 +filter shift 8 + +""" + + +class MsgType(enum.Enum): + def __str__(self): + # when we enumerate, only print the msg name w/o class: + return self.name + # 0x00-0x03 require time stamping + SYNC = 0x00 + # receiver sends del_reqs message to figure out xceive delay + DELAY_REQ = 0x01 + # path_del only for asymmetric routing topo + PATH_DELAY_REQ = 0x02 + PATH_DELAY_RESP = 0x03 + # 0x08-0x0d do not require time stamping + # time increment since last msg - offset + FOLLOWUP = 0x08 + # sender gets del_resp to calculate RTT delay + DELAY_RESP = 0x09 + PATH_DELAY_FOLLOWUP = 0x0A + # Ann declares clock and type + ANNOUNCE = 0x0B + SIGNALLING = 0x0C + MANAGEMENT = 0x0D + + +class GMCAccuracy(enum.Enum): + def __str__(self): + return self.name + # GM = GrandMaster + # 00-1F - reserved + nS25 = 0x20 # 25 nanosec + nS100 = 0x21 + nS250 = 0x22 + µS1 = 0x23 # 1 microsec + µS2_5 = 0x24 + µS10 = 0x25 + µS25 = 0x26 + µS100 = 0x27 + µS250 = 0x28 + mS1 = 0x29 # 1 millisec + mS2_5 = 0x2A + mS10 = 0x2B + mS25 = 0x2C + mS100 = 0x2D + mS250 = 0x2E + S1 = 0x2F # 1 sec + S10 = 0x30 + GTS10 = 0x31 # >10sec + # 32-7F reserved + # 80-FD profiles + UNKNOWN = 0xFE + RESERVED = 0XFF + + +class ClkSource(enum.Enum): + def __str__(self): + return self.name + ATOMIC = 0X10 + GPS = 0x20 + TERRESTRIAL_RADIO = 0x30 + PTP_EXTERNAL = 0x40 + NTP_EXTERNAL = 0x50 + HAND_SET = 0x60 + OTHER = 0x90 + INTERNAL_OSCILLATOR = 0xA0 + # F0-FE - PROFILES + # FF - Reserved + + +class ClkClass(enum.Enum): + def __str__(self): + return self.name + # RESERVED 000-005 + PRIMARY_REF_LOCKED = 6 + PRIMARY_REF_UNLOCKED = 7 + LOCKED_TO_APP_SPECIFIC = 13 + UNLOCKD_FR_APP_SPECIFIC = 14 + PRC_UNLOCKED_DESYNC = 52 + APP_UNLOCKED_DESYNC = 58 + PRC_UNLOCKED_DESYNC_ALT = 187 + APP_UNLOCKED_DESYNC_ALT = 193 + # RESERVED 194-215 + # Profiles 216-232 + # RESERVED 233-247 + DEFAULT = 248 + # RESERVED 249-254 + SLAVE_ONLY = 255 + + +class TLVType(enum.Enum): + def __str__(self): + return self.name + RESERVED = 0x0000 + # standard: + MANAGEMENT = 0x0001 + MANAGEMENT_ERROR_STATUS = 0x0002 + ORGANIZATION_EXTENSION = 0x0003 + # optional: + REQUEST_UNICAST_XMISSION = 0x0004 + GRANT_UNICAST_XMISSION = 0x0005 + CANCEL_UNICAST_XMISSION = 0x0006 + ACK_CANCEL_UNICAST_XMISSION = 0x0007 + # optional trace + PATH_TRACE = 0x0008 + # optional timescale + ALT_TIME_OFFSET_INDICATOR = 0x0009 + # RESERVED for std TLV 000A-1FFF + # From 2008 std: + AUTHENTICATION = 0x2000 + AUTHENTICATION_CHALLENGE = 0x2001 + SECURITY_ASSOCIATION_UPDATE = 0x2002 + CUM_FREQ_SCALE_FACTOR_OFFSE = 0x2003 + # v2.1: + # Experimental 2004-202F + # RESERVED 2030-3FFF + # IEEE 1588 reserved 4002-7EFF + # Experimental 7F00-7FFF + # Interesting 8000-8009 + PAD = 0x8008 + AUTHENTICATIONv2 = 0x8009 + # IEEE 1588 RESERVED 800A-FFEF + # RESERVED FFEF-FFFF + + +class PTPMsg: + class MsgFlags(Flag): + def __str__(self): + return self.name + Twostep = 2 # 1<<1 + Unicast = 4 # 1<<2 + + @staticmethod + def getTLVs(msgLen, data, start): + # TLV = Type, Length, Value Identifier + tlvSeq = [] + while(msgLen - start) > 0: + tlvType = int.from_bytes(data[start: start + 2], byteorder='big') + tlvLen = int.from_bytes(data[start + 2:start + 4], byteorder='big') + # 3 byte OID + 3 byte subOID + # V in TLV are even in length. + """ + 1588-2019: 14.3.2 TLV member specifications + All organization-specific TLV extensions shall have + the format specified in Table 53: + bitfield | Octets | TLV offset + tlvType | 2 | 0 + lengthField | 2 | 2 + organizationId | 3 | 4 + organizationSubType | 3 | 7 + dataField | N | 10 + """ + + """802.1AS-2011 specific TLV in Signalling seems to be: + targetPortIdentity (PortIdentity) (this comes before the TLV) + The value is 0xFF. (Apple seems to use 0x00) + (Message interval request TLV) + bitfield | Octets | TLV offset + tlvType | 2 | 0 <-- 3 + lengthField | 2 | 2 <-- 12 + organizationId | 3 | 4 <-- 00:80:c2 + organizationSubType | 3 | 7 <-- 00:00:02 + linkDelayInterval | 1 | 10 + timeSyncInterval | 1 | 11 + announceInterval | 1 | 12 + flags | 1 | 13 + reserved | 2 | 14 + + uint8 : linkDelayInterval + uint8 : timeSyncInterval + uint8 : announceInterval + uint8 : flags (== 3) + uint16: reserved + + 10.5.4.3.6 linkDelayInterval (Integer8) + = log base 2 of mean time interval, desired by the port that sends this TLV, + between successive Pdelay_Req messages sent by the port at the other end of the link + The format and allowed values of linkDelayInterval are the same as the format and + allowed values of initialLogPdelayReqInterval, see 11.5.2.2. + values 127, 126, and –128 are interpreted as (same for timeSync and announce): + + 127 = stop sending + 126 = set currentX to the value of initialX + –128= not to change the mean time interval between successive X messages. + + 10.5.4.3.7 timeSyncInterval (Integer8) + = log base 2 of mean time interval, desired by the port that sends this TLV, + between successive time-synchronization event messages sent by the port at the other + end of the link. The format and allowed values of timeSyncInterval are the same as + the format and allowed values of initialLogSyncInterval, see 10.6.2.3, 11.5.2.3, + 12.6, and 13.9.2. + + 10.5.4.3.8 announceInterval (Integer8) + = log base 2 of mean time interval, desired by the port that sends this TLV, between + successive Announce messages sent by the port at the other end of the link. The + format and allowed values of announceInterval are the same as the format and + allowed values of initialLogAnnounceInterval, see 10.6.2.2. + + 10.5.4.3.9 flags (Octet) + Bits 1 and 2 of the octet are defined in Table 10-14 and take on values T/F + 1 = computeNeighborRateRatio + 2 = computeNeighborPropDelay + """ + + """802.1AS-2011 specific TLV in Signalling seems to be (CSN TLV): + bitfield | Octets | TLV offset + tlvType | 2 | 0 <-- 3 + lengthField | 2 | 2 <-- 46 + organizationId | 3 | 4 <-- 00:80:c2 + organizationSubType | 3 | 7 <-- 00:00:03 + upstreamTxTime | 12| 10 + neighborRateRatio | 4 | 22 + neighborPropDelay | 12| 26 + delayAsymmetry | 12| 38 + + upstreamTxTime (UScaledNs) + neighborRateRatio (Integer32) + neighborPropDelay (UScaledNs) + delayAsymmetry (UScaledNs) + CSN egress node + + This TLV is not allowed to occur before the Follow_Up information TLV (see 11.4.4.3) + """ + # org specific + if tlvType == 3: + # Usually 00:80:c2:00:00:01 within FOLLOWUP + # https://hwaddress.com/mac-address-range/00-0D-93-00-00-00/00-0D-93-FF-FF-FF/ + # Apple: 00:0d:93 sub: 00:00:0x => meaning: defined by Apple. + # contains clockID(mac)+port + tlvOID = int.from_bytes(data[start + 4:start + 10], byteorder='big') + # Exclude the OID from data, tlvLen includes OID + tlvData = data[start + 10:start + 4 + tlvLen] + + tlvSeq.append([tlvType, tlvLen, tlvOID, tlvData]) + + elif tlvType == 8: # PATH_TRACE + """ + while it may be possible to have Path and other TLV types together, best for now + to keep their handling and return separate. Have not seen such a combination yet. + 1588-2019: 16.2.5 PATH_TRACE TLV specification + The PATH_TRACE TLV format shall be as specified in Table 115. + bitfield | Octets | TLV offset + tlvType | 2 | 0 + lengthField | 2 | 2 + pathSequence | 8N| 4 + + N is equal to stepsRemoved+1 (see 10.5.3.2.6). The size of the pathSequence array + increases by 1 for each time-aware system that the Announce information traverses. + """ + tlvUnitSize = 8 # bytes + tlvRecordAmt = int(tlvLen / tlvUnitSize) + # https://blog.meinbergglobal.com/2019/12/06/tlvs-in-ptp-messages/ + tlvPathSequence = [None] * tlvRecordAmt + for x in range(0, tlvRecordAmt): + tlvPathSequence[x] = int.from_bytes(data[ + start + 4 + (x * tlvUnitSize): + start + 4 + tlvUnitSize + (x * tlvUnitSize) + ], byteorder='big') + # print(tlvPathSequence[x]) + return tlvPathSequence + + # still in the while loop + start += tlvLen + 4 # 4 byte TLV header + return tlvSeq if len(tlvSeq) > 0 else None + + def __init__(self, data): + # self.v1_compat = (data[0] & 0b00010000) >> 4 + self.msg_type = MsgType(data[0] & 0b00001111) + # self.ptp_version= data[1] & 0b00001111 #) >> 0 + # data[2] is 1 Reserved byte + self.msgLength = int.from_bytes(data[2:4], byteorder='big') + if len(data) == self.msgLength: + # domain: 0 = default | 1 = alt 1 | 3 = alt 3 | 4-127, user defined. + self.subdomainNumber = data[4] + msgFlagsA = int.from_bytes(data[6:7], byteorder='big') + # msgFlagsB = int.from_bytes(data[7:8], byteorder='big') + # self.msgFlags = self.getMsgFlags(msgFlagsA, msgFlagsB) + self.msgFlags = PTPMsg.MsgFlags(msgFlagsA) + """ + Semantics dictate that correction is always ZERO for + -Announce + -Signaling + -PTP mgmt + """ + self.correctionNanoseconds = int.from_bytes(data[8:14], byteorder='big') + # unlikely we will ever deal with subNanoSec or ever be accurate in Python + # self.correctionSubNanoseconds = int.from_bytes(data[14:16], byteorder='big') + # data[16:20][0] is 4 Reserved bytes + self.clockIdentity = int.from_bytes( + data[20:28], byteorder='big') + # SrcPortID = ID for the sending address, where each IP may have a diff one, or same. + self.sourcePortID = int.from_bytes( + data[28:30], byteorder='big') + self.sequenceID = int.from_bytes( + data[30:32], byteorder='big') + # unnecessary - from ptpv1: + # self.control = data[32] + # logMessagePeriod / Interval: for Sync, Followup, Del_resp + # multicast = log2(interval between multicast messages) + # y = log2(x) => if lMP = -2, x = 0.25 sec i.e. send 4 Sync every second. + # -3 => 8 per second. + # Sync: -7 -> 1 (i.e. from 128/sec to 1 per 2 sec) + # Ann : -3 -> 3 (i.e. from 8/sec to 1 per 8 sec) + # Delay_Resp: def -4 (16/sec) | -7 -> 6 (i.e. from 128/sec to 1 per 64 sec) + self.logMessagePeriod = data[33] + if((self.msg_type == MsgType.SYNC) + or (self.msg_type == MsgType.ANNOUNCE) + or (self.msg_type == MsgType.DELAY_REQ)): + self.originTimestampSec = int.from_bytes( + data[34:40], byteorder='big') + self.originTimestampNanoSec = int.from_bytes( + data[40:44], byteorder='big') + if(self.msg_type == MsgType.ANNOUNCE): + # self.originCurrentUTCOffset = int.from_bytes(data[44:46], byteorder='big') + # skip 1 reserved byte + # GM determined by (lower = better): + # prio1 < Class < Accuracy < Variance < prio2 < Ident(mac) + self.prio01 = data[47] + # ClockClass = Quality Level (QL) + self.gmClockClass = data[48] + self.gmClockAccuracy = data[49] + # variance: lower = better. Based on Allan Variance / Sync intv + # PTP variance is equal to Allan variance multiplied by (τ^2)/3, + # where τ is the sampling interval + self.gmClockVariance = int.from_bytes( + data[50:52], byteorder='big') + self.prio02 = data[52] + self.gmClockIdentity = int.from_bytes( + data[53:61], byteorder='big') + self.localStepsRemoved = int.from_bytes( + data[61:63], byteorder='big') + self.timeSource = data[63] + tlvStart = 64 + self.hasTLVs = (self.msgLength - tlvStart) > 0 + if self.hasTLVs: + self.tlvPathSequence = self.getTLVs(self.msgLength, data, tlvStart) + + elif(MsgType(self.msg_type) == MsgType.DELAY_RESP): + self.rcvTimestampSec = int.from_bytes( + data[34:40], byteorder='big') + self.rcvTimestampNanoSec = int.from_bytes( + data[40:44], byteorder='big') + self.requestingSrcPortIdentity = int.from_bytes( + data[44:52], byteorder='big') # mac+port + self.requestingSrcPortID = int.from_bytes( + data[52:54], byteorder='big') # ID + + elif(MsgType(self.msg_type) == MsgType.FOLLOWUP): + tlvStart = 44 + self.hasTLVs = (self.msgLength - tlvStart) > 0 + self.preciseOriginTimestampSec = int.from_bytes( + data[34:40], byteorder='big') + self.preciseOriginTimestampNanoSec = int.from_bytes( + data[40:44], byteorder='big') + # in Airplay2 apple products, followups have TLVs (but we don't need them) + if self.hasTLVs: + self.tlvSeq = self.getTLVs(self.msgLength, data, tlvStart) + + elif(MsgType(self.msg_type) == MsgType.SIGNALLING): + tlvStart = 44 + self.hasTLVs = (self.msgLength - tlvStart) > 0 + self.targetPortIdentity = int.from_bytes( + data[34:42], byteorder='big') + self.targetPortID = int.from_bytes( + data[42:44], byteorder='big') + if self.hasTLVs: + self.tlvSeq = self.getTLVs(self.msgLength, data, tlvStart) + + +class PTPMaster: + def __init__(self): + # Defaults are worst case. + self.prio01 = 255 + self.gmClockClass = 255 # slave only + self.gmClockAccuracy = 0xFF + self.gmClockVariance = 0xFFFF + self.prio02 = 255 + self.gmClockIdentity = 0xFFFFFFFFFFFFFFFF + + def __init__(self, data): + self.prio01 = data.prio01 + self.gmClockClass = data.gmClockClass + self.gmClockAccuracy = data.gmClockAccuracy + self.gmClockVariance = data.gmClockVariance + self.prio02 = data.prio02 + self.gmClockIdentity = data.gmClockIdentity + + def __lt__(self, other): + if not isinstance(other, PTPMaster): + return False + if self.prio01 < other.prio01: + return True + if self.gmClockClass < other.gmClockClass: + return True + if self.gmClockAccuracy < other.gmClockAccuracy: + return True + if self.gmClockVariance < other.gmClockVariance: + return True + if self.prio02 < other.prio02: + return True + if self.gmClockIdentity < other.gmClockIdentity: + return True + return False + + def __eq__(self, other): + if not isinstance(other, PTPMaster): + return False + return (self.prio01 == other.prio01 + and self.gmClockClass == other.gmClockClass + and self.gmClockAccuracy == other.gmClockAccuracy + and self.gmClockVariance == other.gmClockVariance + and self.prio02 == other.prio02 + and self.gmClockIdentity == other.gmClockIdentity) + + +class PTPForeignMaster: + def __init__(self): + self.sourcePortID = {} + self.announceAmount = 0 + + def __init__(self, data, arrival): + self.sourcePortID = {data.gmClockIdentity, data.sourcePortID} + self.announceAmount = 0 + # 9.3.2.4.3 : array of fm announces within FOREIGN_MASTER_TIME_WINDOW + self.announceMessages = deque([data] * 4, maxlen=4) + self.announceMessageArrival_ts = deque([arrival] * 10, maxlen=10) + self.announceMessageArrivalDeltas = deque([0] * 10, maxlen=10) + self.mraArrivalNanos = arrival + + def inc(self): + self.announceAmount += 1 + + def setMostRecentAMsg(self, data, arrival): + self.inc() + self.announceMessages.append(data) + self.announceMessageArrival_ts.append(arrival) + self.announceMessageArrivalDeltas.append(arrival - self.announceMessageArrival_ts[len(self.announceMessageArrival_ts) - 2]) + # self.checkMasterQuality() + + def checkMasterQuality(self): + # TODO: Verify the quality of the Master's announce timing. Not needed at receiver. + # IEEE-1588-2019: 9.5.8 + """ + ...the value of the arithmetic mean of the intervals, in seconds, + between message transmissions is within ±30% of the value of 2 ** portDS.logAnnounceInterval + + Also, a PTP Port shall transmit Announce messages such that: + at least 90% of the inter-message intervals are within ±30% of + 2 ** portDS.logAnnounceInterval. + The interval between successive Announce messages should not exceed + twice the value of 2** portDS.logAnnounceInterval, + to prevent causing an announceReceiptTimeout event. + """ + QLength = 10 - self.announceMessageArrivalDeltas.count(0) + ArithMean = sum(self.announceMessageArrivalDeltas) / QLength + AInterval = (2 ** self.getMostRecentAMsg().logMessagePeriod) * 10**9 + AInterval03 = AInterval * 0.3 + isWithin = ((AInterval - AInterval03) < ArithMean and ArithMean < (AInterval + AInterval03)) + # i.e. within ±30% + + def getAnnounceAmt(self): + return self.announceAmount + + def getArrivalNanos(self): + return self.mraArrivalNanos + + def getMostRecentAMsg(self): + return self.announceMessages[len(self.announceMessages) - 1] + + def __lt__(self, other): + return (PTPMaster(self.getMostRecentAMsg()) + < PTPMaster(other.getMostRecentAMsg())) + + def __gt__(self, other): + return (PTPMaster(self.getMostRecentAMsg()) + > PTPMaster(other.getMostRecentAMsg())) + + def __eq__(self, other): + return (PTPMaster(self.getMostRecentAMsg()) + == PTPMaster(other.getMostRecentAMsg())) + # return self.sourcePortID == other.sourcePortID #also works + + +class PTPPortState(enum.Enum): + def __str__(self): + # so when we enumerate, we only print the msg name w/o class: + return self.name + ( + # PRE_MASTER + # MASTER + INITIALIZING, + LISTENING, + PASSIVE, + UNCALIBRATED, + SLAVE + ) = range(5) + # no code yet to run as MASTER + + +class PTP(): + class CFG(Flag): + def __str__(self): + return self.name + # Config(16383) / Config(0x3FFF) toggles everything on. + ShowNothing = 0 + ShowTLVs = 1 # 1<<0 + ShowSYNC = 2 # 1<<1 + ShowDELAY_REQ = 4 # 1<<2 + ShowPATH_DELAY_REQ = 8 # 1<<3 + ShowPATH_DELAY_RESP = 16 # 1<<4 + ShowFOLLOWUP = 32 # 1<<5 + ShowDELAY_RESP = 64 # 1<<6 + ShowPATH_DELAY_FOLLOWUP = 128 # 1<<7 + ShowANNOUNCE = 256 # 1<<8 + ShowSIGNALLING = 512 # 1<<9 + ShowMANAGEMENT = 1024 # 1<<10 + ShowMasterPromotion = 2048 # 1<<11 + ShowPortStateChanges = 4096 # 1<<12 + ShowMeanPathDelay = 8192 # 1<<13 + ShowDebug = 16384 # 1<<14 + SetApplePTPProfile = 32768 # 1<<15 + + def __init__(self, net_interface, config_flags): + self.cfg = self.CFG(config_flags) + # Test individual flags with e.g.: + # self.cfg |= self.CFG.ShowMeanPathDelay + self.portEvent319 = 319 # Sync msgs / Event Port + self.portGeneral320 = 320 # Followup msgs / General port + self.gm = None + self.t1_arr_nanos = 0 + self.t1_ts_s = 0 + self.t1_ts_ns = 0 + self.t1_corr = 0 + self.t2_arr_nanos = 0 + self.t2_ts_s = 0 + self.t2_ts_ns = 0 + self.t3_egress_nanos = 0 + self.t4_arr_at_gm_nanos = 0 + self.ms_propagation_delay = 0 + # Limit Queues to 30 entries + self.QLength = 30 + self.offsetFromMasterNanos = 0 + self.offsetFromMasterNanosMean = 0 + # deque = O(1) perf + self.offsetFromMasterNanosValues = deque([0] * self.QLength, maxlen=self.QLength) + self.meanPathDelayNanos = 0 # bi-directional + self.meanPathDelayNanosMean = 0 # mean of several bi-di results + self.meanPathDelayNanosValues = deque([0] * self.QLength, maxlen=self.QLength) + self.processingOverhead = 0 + self.syncSequenceID = 0 + self.useMasterPromoteAlgo = True + self.DelayReq_PortID = 32768 + self.DelayReq_template = bytearray.fromhex( + '1102002c00000408000000000000000000000000' + '01020304050600018000000100fd00000000000000000000') + if(net_interface is not None): + # DelayReq_template contains dummy MAC '010203040506' + # add 2 empty bytes for 'PTP Port' to end of mac: + self.net_interface = net_interface << 16 + self.net_interface_bytes = (net_interface << 16).to_bytes(8, byteorder='big') + self.DelayReq_template[20:28] = self.net_interface_bytes + else: + self.net_interface = int('010203040506') + self.DelayReq_template[28:30] = self.DelayReq_PortID.to_bytes(2, byteorder='big') + self.portStateChange(PTPPortState.INITIALIZING) + self.PTPcorrection = 0 + self.fML = [] # # 9.3.2.4.6 Size of min 5 + """ + Each entry of the contains two or three members: + - [].foreignMasterPortIdentity, + - [].foreignMasterAnnounceMessages, and optionally + - [].mostRecentAnnounceMessage. + """ + self.fMTW = 4 # FOREIGN_MASTER_TIME_WINDOW = 4 announceInterval + self.fMThr = 2 # FOREIGN_MASTER_THRESHOLD 2 Announce msg within FOREIGN_MASTER_TIME_WINDOW + """ + announceReceiptTimeoutInterval = portDS.announceReceiptTimeout * announceInterval + """ + + # 7.7.3.1 portDS.announceReceiptTimeout: + # "Although 2 is permissible, normally the value should be at least 3." + self.announceReceiptTimeout = 3 + # I.3.2 portDS.logAnnounceInterval: d = 1, 0<->4 + # announceInterval = 2 ** portDS.logAnnounceInterval + self.announceInterval = 0 + + """ + L.4.7 L1SyncReceiptTimeout + This value = # of elapsed L1SyncIntervals that must pass without reception of the + L1_SYNC TLV before the L1_SYNC TLV reception timeout occurs (see L.6.3). + The default init val and allowed values spec'd in the applicable PTP Profile. + """ + self.syncReceiptTimeout = 3 + # 13.3.2.14 logMessageInterval = 0x7F in unicast. + # I.3.2 PTP attribute values + # The default initialization value shall be 0. + # The configurable range shall be −1 to +1. + self.logSyncInterval = 0 + + """ https://github.com/rroussel/OpenAvnu/blob/ArtAndLogic-aPTP-changes/daemons/gptp/gptp_cfg.ini + # Per the Apple Vendor PTP profile + initialLogAnnounceInterval = 0 + initialLogSyncInterval = -3 + # Seconds: + announceReceiptTimeout = 120 + + # Per the Apple Vendor PTP profile (8*announceReceiptTimeout) + syncReceiptTimeout = 960 + """ + if(self.cfg & self.CFG.SetApplePTPProfile): + # prio1 & prio2 = 248 and accuracy = 254 + self.logAnnounceInterval = 0 + self.announceReceiptTimeout = 120 + self.logSyncInterval = -3 + self.syncReceiptTimeout = 8 * self.announceReceiptTimeout + + # count down nanos from last Announce - expires current GM + self.lastAnnounceFromMasterNanos = 0 + + self.network_time_ns = 0 + self.network_time_monotonic_ts = time.monotonic_ns() + self.max_error = 1 * (10 ** 6) # 1 millisecond + + def promoteMaster(self, ptpmsg, reason): + self.gm = PTPMaster(ptpmsg) + if(self.cfg & self.CFG.ShowMasterPromotion): + print("New GM Clock promoted: " + f"{ptpmsg.gmClockIdentity:10x} (Prio{ptpmsg.prio01}/{ptpmsg.prio02})", + f"reason: {reason}" + ) + # reset cumulative mean values to 0 + self.offsetFromMasterValues = deque([0] * self.QLength, maxlen=self.QLength) + self.meanPathDelayNanosValues = deque([0] * self.QLength, maxlen=self.QLength) + self.portStateChange(PTPPortState.SLAVE) + self.announceInterval = 2 ** ptpmsg.logMessagePeriod + + def compareMaster(self, ptpmsg): + # This algo promotes a new master if its properties are better than currently elected GM + # prio1 < Class < Accuracy < Variance < prio2 < Ident(mac) + # Lower values == "better" + if self.gm is None: + self.promoteMaster(ptpmsg, "reset") + else: + incoming = PTPMaster(ptpmsg) + + if (incoming < self.gm): + self.promoteMaster(ptpmsg, "better GM") + self.fML = [] + # else: + # retain current GM + + def sendDelayRequest(self, sequenceID): + self.DelayReq_template[30:32] = sequenceID.to_bytes(2, byteorder='big') + return self.DelayReq_template + + def portStateChange(self, PTPPortState): + self.portState = PTPPortState + if (self.cfg & self.CFG.ShowPortStateChanges): + print(f"PTP State: {self.portState}") + + def getPortState(self): + return self.portState + + def isKnownForeignMaster(self, ptpfm, ptpmsg, arrivalNanos): + """ + Looks at our list of foreignMaster candidates and when we have enough Announce + msgs from one, we kick off the BMCA: compareMaster() + """ + """ + 9.3.2.5 Qualification of Announce messages + + c) Unless otherwise specified by the option of 17.7, if the sender of S is a foreign + master F, and fewer than FOREIGN_MASTER_THRESHOLD distinct Announce messages from F + have been received within the most recent FOREIGN_MASTER_TIME_WINDOW interval, S + shall not be qualified. Distinct Announce messages are those that have different + sequenceIds, subject to the constraints of the rollover of the UInteger16 data type + used for the sequenceId field. + ... + d) If the stepsRemoved field of S is 255 or greater, S shall not be qualified. + ... + e) This specification “e” is optional. ... + ... + f) Otherwise, S shall be qualified. + """ + if ptpfm not in self.fML: + self.fML.append(ptpfm) + # first entry means count == 0, so we skip sorting/comparing + return False + else: + self.fML[self.fML.index(ptpfm)].setMostRecentAMsg(ptpmsg, arrivalNanos) + # check previous Announce arrivalNanos + lMP = 2 ** ptpmsg.logMessagePeriod # e.g. 2^-2 = 0.25 sec + # check interarrival diff of current and stored Announce nanos is + # less than FOREIGN_MASTER_TIME_WINDOW * logMessagePeriod + considerBMCA = ((arrivalNanos - self.fML[ + self.fML.index(ptpfm)] + .getArrivalNanos()) * 10**-9) < (self.fMTW * lMP) # e.g. 4 * 0.25 = 1 sec + self.fML.sort() # keep fML list sorted, and mash [0] into BMCA when time comes + if (self.fML[self.fML.index(ptpfm)].getAnnounceAmt() >= self.fMThr + and considerBMCA): + # run BMCA + self.compareMaster(self.fML[0].getMostRecentAMsg()) + return True + + def handlemsg(self, ptpmsg, address, timestampArrival, processingOverhead): + # print(f"entered handlemsg() with {ptpmsg.sequenceID} and {self.syncSequenceID}") + thinning = 100 # print msg every x msgs + # port 319 + if((ptpmsg.msg_type == MsgType.SYNC) + or (ptpmsg.msg_type == MsgType.DELAY_REQ)): + + if(((self.cfg & self.CFG.ShowSYNC) or (self.cfg & self.CFG.ShowDELAY_REQ)) + and (ptpmsg.sequenceID % thinning == 0)): + print(f"PTP319 {ptpmsg.msg_type: <12}", + f"srcprt-ID: {ptpmsg.sourcePortID:05d}", + f"clockId: {ptpmsg.clockIdentity:016x}", + f"seq-ID: {ptpmsg.sequenceID:08d}", + f"Time: {ptpmsg.originTimestampSec}.{ptpmsg.originTimestampNanoSec:09d}", + ) + # print(f"processingOverhead for {ptpmsg.msg_type}:{processingOverhead:.9f}") + + # were we master, here is when we would respond to DELAY_REQ with DELAY_RESP + # upon receipt of each Sync, we should respond with DELAY_REQ with same seqID + if (MsgType(ptpmsg.msg_type) == MsgType.SYNC + and self.gm is not None + and ptpmsg.clockIdentity == self.gm.gmClockIdentity): + if ptpmsg.msgFlags.Twostep: + # Calculate ms_propagation_delay in FOLLOWUP + self.t2_arr_nanos = timestampArrival + self.t2_ts_s = ptpmsg.originTimestampSec + self.t2_ts_ns = ptpmsg.originTimestampNanoSec + self.syncSequenceID = ptpmsg.sequenceID + # assign t3 to delay_req egress timestamp + self.t3_egress_nanos = time.monotonic_ns() + return self.sendDelayRequest(self.syncSequenceID) + # else: #PTP in airplay does not seem to bother with 1-step + # #iPhone PTP sends ptpmsg.originTimestamp(Nano)Sec = 0... so this won't work + # #1-step: must calculate t2-t1 diff here. + # self.t1_arr_nanos = ptpmsg.originTimestampSec + (ptpmsg.originTimestampNanoSec / 10 ** 9) + # self.ms_propagation_delay = t2_arr - t1_arr + + elif(ptpmsg.msg_type == MsgType.DELAY_RESP + and ptpmsg.requestingSrcPortIdentity == self.net_interface): + """ + IEEE1588-2019 Spec says: + = [(t2 – t1) + (t4 – t3)]/2 = [(t2 – t3) + (t4 – t1)]/2 + + = [(t2 - t3) + (receiveTimestamp of Delay_Resp message – preciseOriginTimestamp of Follow_Up message) – + - correctionField of Follow_Up message – correctionField of Delay_Resp message]/2 + """ + t4 = (ptpmsg.rcvTimestampSec * (10**9) + ptpmsg.rcvTimestampNanoSec) + + self.meanPathDelayNanos = ((self.t2_arr_nanos - self.t3_egress_nanos) + + (t4 - (self.t1_ts_s * (10**9)) - self.t1_ts_ns) + - self.t1_corr - ptpmsg.correctionNanoseconds) / 2 + + self.PTPcorrection = abs(self.meanPathDelayNanos) / (10**9) + # print(f"Current mean path delay (sec): {self.PTPcorrection:.09f}") + + + # store in self.network_time_ns and self.network_time_monotonic_ts the current network time + # and the time the 'fixed' network time is retrieved, so we can calculate network time at any time + + # try to filter out invalid values + network_time_ns = t4 + self.meanPathDelayNanos + network_time_monotonic_ts = time.monotonic_ns() + + # what time would be now if we need to calculate it? + previous_network_time = self.network_time_ns + (network_time_monotonic_ts - self.network_time_monotonic_ts) + # the error is the difference between calculated time from previous ptp sync and the current ptp sync + error = (previous_network_time - network_time_ns) / (10 ** 6) + + if abs(error) < self.max_error or self.network_time_ns == 0: + if (abs(error) > 10): + print(f'updated time error {error} less than {self.max_error} ms') + self.network_time_ns = network_time_ns + self.network_time_monotonic_ts = network_time_monotonic_ts + self.max_error = 2 # millisecond + else: + #print(f'skip update time error {error} bigger than {self.max_error} ms') + # grow error so an update happens + self.max_error = self.max_error * 1.2 + + """ + # This Q builds a sliding avg of all MPDs. + self.meanPathDelayNanosValues.append(mpdNanos) + # must append, otherwise ZeroDivisionError + self.meanPathDelayNanosMean = sum(self.meanPathDelayNanosValues)/ \ + (self.meanPathDelayNanosValues.maxlen-self.meanPathDelayNanosValues.count(0)) + print(f"self.meanPathDelayNanosMean (sec): {abs(self.meanPathDelayNanosMean)/(10**9):.09f}") + """ + + """ + derived from our clock: + t4 = self.t3_egress_nanos + mpd - self.offsetFromMasterNanos + + from master: + t4 = (ptpmsg.rcvTimestampSec*(10**9)) + ptpmsg.rcvTimestampNanoSec) + + diff of the above two: + diff = (self.t3_egress_nanos + mpd - self.offsetFromMasterNanos) - \ + ((ptpmsg.rcvTimestampSec*(10**9)) + ptpmsg.rcvTimestampNanoSec) + + as our clock derived from master: + t4 = (ptpmsg.rcvTimestampSec*(10**9)) + ptpmsg.rcvTimestampNanoSec \ + + self.offsetFromMasterNanos + """ + if ((self.cfg & self.CFG.ShowMeanPathDelay) and (ptpmsg.sequenceID % (thinning / 10) == 0)): + print(f"PTP-correction (sec): {self.PTPcorrection:.09f}") + """ + origin = ptpmsg.rcvTimestampSec + (ptpmsg.rcvTimestampNanoSec/(10**9)) + + self.PTPcorrection + print(f"Timetamp at origin now: {origin:.09f}") + """ + + if ((self.cfg & self.CFG.ShowDELAY_RESP) and (ptpmsg.sequenceID % thinning == 0)): + print(f"PTP320 {ptpmsg.msg_type: <12}", + f"srcprt-ID: {ptpmsg.sourcePortID:05d}", + f"clockId: {ptpmsg.clockIdentity:016x}", + f"seq-ID: {ptpmsg.sequenceID:08d}", + f"correctionNanosec: {ptpmsg.correctionNanoseconds:09d}", + f"receiveTimestamp: {ptpmsg.rcvTimestampSec}.{ptpmsg.rcvTimestampNanoSec:09d}", + ) + elif(ptpmsg.msg_type == MsgType.ANNOUNCE): + ptpfm = PTPForeignMaster(ptpmsg, timestampArrival) + self.isKnownForeignMaster(ptpfm, ptpmsg, timestampArrival) + if not (self.getPortState() == PTPPortState.INITIALIZING + or self.getPortState() == PTPPortState.SLAVE + or self.getPortState() == PTPPortState.PASSIVE + or self.getPortState() == PTPPortState.UNCALIBRATED): + + if(self.gm is None): + """ + Normally, (in AirPlay) PTP masters negotiate amongst themselves who leads, + then only that 1 gm sends announce. + In this half PTP implementation, as a CPU measure, we can let them fight it + out and then just run promoteMaster directly. + """ + if not self.useMasterPromoteAlgo: + self.promoteMaster(ptpmsg, "changeover") + if(self.gm is not None): + # path trace TLV path-seq in Announce (also) has GM + """ + IEEE-1588-2019: + 16.2.3 Receipt of an Announce message + A PTP Port of a Boundary Clock receiving an Announce message from + the current parent PTP Instance shall: + a) Scan the pathSequence member of any PATH_TRACE TLV present for a value of the + clockIdentity field equal to the value of the defaultDS.clockIdentity member of + the receiving PTP Instance, that is, there is a “match.” + b) Discard the message if the TLV is present and a match is found. + c) Copy the pathSequence member of the TLV to the pathTraceDS.list member + (see 16.2.2.2.1) if the TLV is present and no match is found. + """ + if self.gm.gmClockIdentity in ptpmsg.tlvPathSequence: + self.lastAnnounceFromMasterNanos = timestampArrival + pass + else: # if self.gm.gmClockIdentity != ptpmsg.gmClockIdentity: + if not self.useMasterPromoteAlgo: + self.compareMaster(ptpmsg) + + if ((self.cfg & self.CFG.ShowANNOUNCE) and (ptpmsg.sequenceID % thinning == 0)): + # varianceb10 = 2**((ptpmsg.gmClockVariance - 0x8000) / 2**8) + # varianceb2 = ((ptpmsg.gmClockVariance - 0x8000) / 2**8) + # i.e. gmVariance = (log2(variance)*2^8)+32768 + # 0x0000 => 2^-128 | 0xFFFE => 2^127.99219 + print(f"PTP320 {ptpmsg.msg_type: <12}", + f"srcprt-ID: {ptpmsg.sourcePortID:05d}", + f"pri1/2: {ptpmsg.prio01}/{ptpmsg.prio02}", + f"gmClockClass: {ClkClass(ptpmsg.gmClockClass)}", + f"gmClockAccuracy: {GMCAccuracy(ptpmsg.gmClockAccuracy)}", + # f"gmClockVariance(s): {varianceb10:.04g}", + # f"gmClockVariance(s): 2^{varianceb2:.04g}", + f"gmClockId: {ptpmsg.gmClockIdentity:10x}", # x = heX + f"seq-ID: {ptpmsg.sequenceID:08d}", + # f"timeSource: {ptpmsg.timeSource}", + "Time:", ptpmsg.originTimestampSec) + + if(self.cfg & self.CFG.ShowTLVs): + print(f"PTP320 with PathTrace { [f'0x{addr:016x}' for addr in ptpmsg.tlvPathSequence] }") + # print(f"processingOverhead for {ptpmsg.msg_type}:{processingOverhead:.9f}") + + elif(ptpmsg.msg_type == MsgType.FOLLOWUP): + # in Airplay(2) PreciseOriginTimestamp = device uptime. + if(ptpmsg.sequenceID == self.syncSequenceID + and self.gm is not None + and ptpmsg.clockIdentity == self.gm.gmClockIdentity): + + self.t1_arr_nanos = timestampArrival + self.t1_ts_s = ptpmsg.preciseOriginTimestampSec + self.t1_ts_ns = ptpmsg.preciseOriginTimestampNanoSec + self.t1_corr = ptpmsg.correctionNanoseconds + + # when iPhones deep sleep - their uptime (origintimestamp) pauses + self.offsetFromMasterNanos = ( + self.t2_arr_nanos + - ((ptpmsg.preciseOriginTimestampSec * (10**9)) + + ptpmsg.preciseOriginTimestampNanoSec + + ptpmsg.correctionNanoseconds)) + self.offsetFromMasterNanosValues.append(self.offsetFromMasterNanos) + # must append otherwise ZeroDivisionError + self.offsetFromMasterNanosMean = sum( + self.offsetFromMasterNanosValues) / ( + self.offsetFromMasterNanosValues.maxlen + - self.offsetFromMasterNanosValues.count(0)) + # print(f"self.offsetFromMasterMean (sec): {self.offsetFromMasterNanosMean/(10**9):.09f}") + + # in two step PTP - we send a DELAY_REQ, and await its response + # to figure out t3 and t4 + + if ((self.cfg & self.CFG.ShowFOLLOWUP) and (ptpmsg.sequenceID % thinning == 0)): + # print info every nth pkt + print(f"PTP320 {ptpmsg.msg_type: <12}", + f"srcprt-ID: {ptpmsg.sourcePortID:05d}", + f"clockId: {ptpmsg.clockIdentity:10x}", # x = heX + f"seq-ID: {ptpmsg.sequenceID:08d}", + f"correctionNanosec: {ptpmsg.correctionNanoseconds:09d}", + f"PreciseTime: {ptpmsg.preciseOriginTimestampSec}.{ptpmsg.preciseOriginTimestampNanoSec:09d}") + + if((self.cfg & self.CFG.ShowTLVs) and hasattr(ptpmsg, 'hasTLVs') and ptpmsg.hasTLVs): + print(f"PTP320 with TLVs {ptpmsg.tlvSeq}") + self.parseTLVs(ptpmsg.tlvSeq) + + elif(ptpmsg.msg_type == MsgType.SIGNALLING): + if ((self.cfg & self.CFG.ShowSIGNALLING) and (ptpmsg.sequenceID % thinning == 0)): + print("PTP320", ptpmsg.msg_type, + "sequenceID: ", ptpmsg.sequenceID) + if((self.cfg & self.CFG.ShowTLVs) and hasattr(ptpmsg, 'hasTLVs') and ptpmsg.hasTLVs): + print(f"PTP320 with TLVs {ptpmsg.tlvSeq}") + self.parseTLVs(ptpmsg.tlvSeq) + + def parseTLVs(self, tlvSeq): + for x in range(0, len(tlvSeq)): + if(self.cfg & self.CFG.ShowDebug): + print(f"Typ:{tlvSeq[x][0]:04x}", + f"Len:{tlvSeq[x][1]:04x}", + f"OID:{tlvSeq[x][2]:012x}", + f"Val:{tlvSeq[x][3].hex()}", + ) + + OID = tlvSeq[x][2] + + if(tlvSeq[x][0] == 3 and tlvSeq[x][1] == 28 # 0x1c + and OID == 0x0080c2000001): + self.parseFollowUpTLV(tlvSeq[x][3]) + if(tlvSeq[x][0] == 3 and tlvSeq[x][1] == 22 + and OID == 0x000d93000001): + # Master Clock parameters like announceInterval(?) + self.parseApple001TLV(tlvSeq[x][3]) + if(tlvSeq[x][0] == 3 and tlvSeq[x][1] == 16 # 0x10 + and OID == 0x000d93000004): + # Master Clock ID + self.parseApple004TLV(tlvSeq[x][3]) + + def parseSignallingTLV(self, tlvSeq): + """ + uint16_t tlvType; # 2 + uint16_t lengthField; # 12 + uint8_t organizationId[3]; # 0x0080c2 + uint8_t organizationSubType_ms; # 0x00 + uint16_t organizationSubType_ls; # 0x02 + uint8_t linkDelayInterval; + uint8_t timeSyncInterval; + uint8_t announceInterval; + uint8_t flags; + uint16_t reserved; + """ + + def parseApple001TLV(self, value): + """ Apple specific TLV in Signalling seems to be: + bitfield | Octets | TLV offset + tlvType | 2 | 0 <-- 3 + lengthField | 2 | 2 <-- 22 + organizationId | 3 | 4 <-- 00:0d:93 + organizationSubType | 3 | 7 <-- 00:00:01 + dataField | N | 10 <-- where: + + uint8 : linkDelayInterval + uint8 : timeSyncInterval + uint8 : announceInterval + uint8 : flags (== 3) + uint16: reserved + 10 bytes extra ? + + """ + # Master Clock parameters like announceInterval(?) + dataBlock = int.from_bytes(value[0:16], byteorder='big') + if(self.cfg & self.CFG.ShowDebug): + print(f'dataBlock: {dataBlock:032x}') + + def parseApple004TLV(self, value): + """Apple specific TLV in Follow_Up seems to be: + bitfield | Octets | TLV offset + tlvType | 2 | 0 <-- 3 + lengthField | 2 | 2 <-- 10 + organizationId | 3 | 4 <-- 00:0d:93 + organizationSubType | 3 | 7 <-- 00:00:04 + dataField | N | 10 <-- where: + + 8 byte clock ID (including port) + 2 bytes (reserved?) + """ + + # The Master Clock ID - 8 bytes + # 2 bytes reserved + masterClock = int.from_bytes(value[0:8], byteorder='big') + if(self.cfg & self.CFG.ShowDebug): + print(f'masterClock: {masterClock:012x}') + + def parseFollowUpTLV(self, value): + """802.1AS-2011 specific TLV in Follow_Ups: + (Follow_Up information TLV) + bitfield | Octets | TLV offset + tlvType | 2 | 0 <-- 3 + lengthField | 2 | 2 <-- 28 + organizationId | 3 | 4 <-- 00:80:c2 + organizationSubType | 3 | 7 <-- 00:00:01 + cumulativeScaledRateOffset| 4 | 10 + gmTimeBaseIndicator | 2 | 14 + lastGmPhaseChange | 12| 16 + scaledLastGmFreqChange | 4 | 28 + + + int32 : cumulative scaledRateOffset + uint16 : gmTimeBaseIndicator + ScaledNs: scaledLastGmPhaseChange + int32 : scaledLastGmFreqChange: + + ScaledNs = + uint16 Nanos Msb + uint64 Nanos Lsb + uint16 FracNanos + + scaledRateOffset = (rateRatio – 1.0) × (2^41), truncated to the next smaller signed + integer, where rateRatio is the ratio of the frequency of the grandMaster to the + frequency of the LocalClock entity in the time-aware system that sends the message. + + gmTimeBaseIndicator = + timeBaseIndicator of the ClockSource entity for the current grandmaster + + lastGmPhaseChange = + (time of the current GM - time of the prev GM), at the + time that the current GM became GM. + value is copied from the lastGmPhaseChange member of the MDSyncSend structure whose + receipt causes the MD entity to send the Follow_Up message + + scaledLastGmFreqChange = + + fractional frequency offset of the current GM relative to the previous GM, + at the time that the current GM became GM. or relative to itself prior to the last + change in gmTimeBaseIndicator, multiplied by 2^41 and truncated to the next smaller + signed integer. The value is obtained by multiplying the lastGmFreqChange member of + MDSyncSend whose receipt causes the MD entity to send the Follow_Up message + (see 11.2.11) by 2^41 , and truncating to the next smaller signed Integer8 + """ + """ + In Airplay: + int32 cumulativeScaledRateOffset + uint16 gmTimeBaseIndicator + scaledNs scaledLastGmPhaseChange + int32 scaledLastGmFreqChange + + ScaledNs = + uint32 Nanos Msb # 4 + uint64 Nanos Lsb # 8 + """ + cumulativeScaledRateOffset = int.from_bytes(value[0:4], byteorder='big') + gmTimeBaseIndicator = int.from_bytes(value[4:6], byteorder='big') + scaledLastGmPhaseChange = int.from_bytes(value[6:18], byteorder='big') + scaledLastGmFreqChange = int.from_bytes(value[18:22], byteorder='big') + print(f'cumulativeScaledRateOffset: {cumulativeScaledRateOffset}', + f'gmTimeBaseIndicator: {gmTimeBaseIndicator}', + f'scaledLastGmPhaseChange: {scaledLastGmPhaseChange}', + f'scaledLastGmFreqChange: {scaledLastGmFreqChange}', + ) + + def listen(self): + sockets = [] + + for port in range(319, 321): + server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + server_socket.bind(('0.0.0.0', port)) + sockets.append(server_socket) + + empty = [] + self.portStateChange(PTPPortState.LISTENING) + while True: + readable, writable, exceptional = select.select(sockets, empty, empty) + timenow = time.monotonic_ns() + for s in readable: + (data, address) = s.recvfrom(180) + # print(address, data) + # s.sendto(client_data, client_address) + + timestampArrival = time.monotonic_ns() + ptpmsg = PTPMsg(data) + self.processingOverhead = time.monotonic_ns() - timestampArrival + # just bake overhead into timestampArrival + timestampArrival += self.processingOverhead + + delay_req = self.handlemsg(ptpmsg, address, timestampArrival, self.processingOverhead) + if delay_req is not None: + s.sendto(delay_req, address) + """ + 9.2.6.12 ANNOUNCE_RECEIPT_TIMEOUT_EXPIRES + Each protocol engine shall support a timeout mechanism defining the + , with a value of portDS.announceReceiptTimeout + multiplied by the announceInterval (see 7.7.3.1). + The ANNOUNCE_RECEIPT_TIMEOUT_EXPIRES event occurs at the expiration of this timeout + plus a random number uniformly distributed in the range (0,1) announceIntervals. + """ + if (self.gm is not None and ((timenow - self.lastAnnounceFromMasterNanos) * 10**-9) + > (self.announceReceiptTimeout * ( + self.announceInterval + (random.randrange(2) * self.announceInterval)))): + self.gm = None + self.portStateChange(PTPPortState.LISTENING) + # alt self.portStateChange(PTPPortState.MASTER) + + for s in sockets: + s.close() + + def get_ptp_master_correction(self): + # Gets the current MPD applied to master + return self.PTPcorrection + + def get_ptp_master_nanos(self): + # returns locally adjusted (AirPlay) PTP Master Timestamp in nanos + return self.network_time_ns + (time.monotonic_ns() - self.network_time_monotonic_ts) + + def reader(self, conn): + try: + while True: + if conn.poll(): + msg = conn.recv() + if (msg == 'get_ptp_master_correction'): + conn.send(self.get_ptp_master_correction()) + if (msg == 'get_ptp_master_nanos_timestamped'): + conn.send([self.network_time_ns, self.network_time_monotonic_ts]) + + except KeyboardInterrupt: + pass + except BrokenPipeError: + pass + finally: + conn.close() + + def run(self, p_input): + p = threading.Thread(target=self.listen) + # p.daemon = True #triggers nice python crash :D + p.start() + + reader_p = threading.Thread(target=self.reader, args=((p_input),)) + # reader_p.daemon = True #must be True or shutdown hangs here when in pure thread mode + reader_p.start() + + @staticmethod + def spawn(net_interface=None, config_flags=0): + PTPinstance = PTP(net_interface, config_flags) + + p_output, p_input = multiprocessing.Pipe() + + p = multiprocessing.Process(target=PTPinstance.run, args=(p_input,)) + p.start() + + return p, p_output \ No newline at end of file diff --git a/ap2/connections/stream.py b/ap2/connections/stream.py index b80f66f..5ff6763 100644 --- a/ap2/connections/stream.py +++ b/ap2/connections/stream.py @@ -8,21 +8,22 @@ class Stream: REALTIME = 96 BUFFERED = 103 - def __init__(self, stream): + def __init__(self, stream, buff, ptp_link): self.audio_format = stream["audioFormat"] self.compression = stream["ct"] self.session_key = stream["shk"] self.frames_packet = stream["spf"] self.type = stream["type"] + buff = buff // self.frames_packet self.control_port, self.control_proc = Control.spawn() if self.type == Stream.REALTIME: self.server_control = stream["controlPort"] self.latency_min = stream["latencyMin"] self.latency_max = stream["latencyMax"] - self.data_port, self.data_proc, audio_connection = AudioRealtime.spawn(self.session_key, self.audio_format) + self.data_port, self.data_proc, audio_connection = AudioRealtime.spawn(self.session_key, self.audio_format, buff) elif self.type == Stream.BUFFERED: - self.data_port, self.data_proc, self.audio_connection = AudioBuffered.spawn(self.session_key, self.audio_format) + self.data_port, self.data_proc, self.audio_connection = AudioBuffered.spawn(self.session_key, self.audio_format, buff, ptp_link) def teardown(self): self.data_proc.terminate() diff --git a/ap2/daap.py b/ap2/daap.py new file mode 100644 index 0000000..e2c2901 --- /dev/null +++ b/ap2/daap.py @@ -0,0 +1,40 @@ +# DAAP Format +# from http://daap.sourceforge.net/docs/index.html +# +# 0-3 Content Code +# 4-7 Length +# 8+ Data +def parse_daap(frame): + offset = 8 + + #print("DAAP FRAME", frame) + + def read_frame(): + nonlocal offset + + length = int(frame[offset]) * 0xffffff + int(frame[offset + 1]) * 0xffff + int(frame[offset + 2]) * 0xff + int(frame[offset + 3]) + offset += length + 8 + + return frame[(offset - length - 4) : (offset - 4)] + + def read_text_frame(): + return str(read_frame(), 'UTF-8') + + if frame[0:4] == b'mlit': + # skip header + while frame[offset] != 0: + offset += 1 + + # skip unreadable field + read_frame() + + # read text fields + album = read_text_frame() + artist = [read_text_frame(), read_text_frame()] + genres = read_text_frame() + track = read_text_frame() + + print("ALBUM", album) + print("ARTIST", artist) + print("GENRES", genres) + print("TRACK", track) \ No newline at end of file diff --git a/ap2/utils.py b/ap2/utils.py index 4301d04..41aa50f 100644 --- a/ap2/utils.py +++ b/ap2/utils.py @@ -4,33 +4,40 @@ import platform import subprocess + def get_logger(name, level="INFO"): - logging.basicConfig(filename="%s.log" % name, - filemode='a', - format='%(asctime)s,%(msecs)d %(name)s %(levelname)s %(message)s', - datefmt='%H:%M:%S', - level=level) + logging.basicConfig( + filename="%s.log" % name, + filemode="a", + format="%(asctime)s,%(msecs)d %(name)s %(levelname)s %(message)s", + datefmt="%H:%M:%S", + level=level, + ) return logging.getLogger(name) + def get_free_port(): free_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - free_socket.bind(('0.0.0.0', 0)) + free_socket.bind(("0.0.0.0", 0)) free_socket.listen(5) port = free_socket.getsockname()[1] free_socket.close() return port + def get_free_tcp_socket(): free_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - free_socket.bind(('0.0.0.0', 0)) + free_socket.bind(("0.0.0.0", 0)) free_socket.listen(5) return free_socket + def get_free_udp_socket(): free_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - free_socket.bind(('0.0.0.0', 0)) + free_socket.bind(("0.0.0.0", 0)) return free_socket + def interpolate(value, from_min, from_max, to_min, to_max): from_span = from_max - from_min to_span = to_max - to_min @@ -39,10 +46,15 @@ def interpolate(value, from_min, from_max, to_min, to_max): return to_min + (value_scale * to_span) + def get_volume(): subsys = platform.system() if subsys == "Darwin": - pct = int(subprocess.check_output(["osascript", "-e", "output volume of (get volume settings)"]).rstrip()) + pct = int( + subprocess.check_output( + ["osascript", "-e", "output volume of (get volume settings)"] + ).rstrip() + ) vol = interpolate(pct, 0, 100, -30, 0) elif subsys == "Linux": line_pct = subprocess.check_output(["amixer", "get", "PCM"]).splitlines()[-1] @@ -51,11 +63,12 @@ def get_volume(): pct = int(m.group(1)) if pct < 45: pct = 45 - else: pct = 50 + else: + pct = 50 vol = interpolate(pct, 45, 100, -30, 0) elif subsys == "Windows": # Volume get is not managed under windows, let's set to a default volume - vol = 50; + vol = 50 if vol == -30: return -144 return vol @@ -73,3 +86,33 @@ def set_volume(vol): pct = int(interpolate(vol, -30, 0, 45, 100)) subprocess.run(["amixer", "set", "PCM", "%d%%" % pct]) + elif subsys == "Windows": + from ctypes import cast, POINTER + from comtypes import CLSCTX_ALL + from pycaw.pycaw import AudioUtilities, IAudioEndpointVolume + from os import getpid + + # devices = AudioUtilities.GetSpeakers() + # interface = devices.Activate(IAudioEndpointVolume._iid_, CLSCTX_ALL, None) + # volume = cast(interface, POINTER(IAudioEndpointVolume)) + + # # Control volume + # # volume.SetMasterVolumeLevel(-0.0, None) #max + # # volume.SetMasterVolumeLevel(-5.0, None) #72% + # # volume.SetMasterVolumeLevel(-10.0, None) # 51% + + # volume.SetMasterVolumeLevel(int(interpolate(vol, -30, 0, -30, 0)), None) + + pid = getpid() + sessions = AudioUtilities.GetAllSessions() + for session in sessions: + + # TODO: move this code to audio.py: + # process id does not match because mixer process id is the subprocess of the pyaudio subprocess + # if session.Process and session.ProcessId == pid: + + if session.Process and session.Process.name() == "python.exe": + interface = session.SimpleAudioVolume + interface.SetMasterVolume( + float(interpolate(vol, -30, 0, 0.0, 1.0)), None + )