diff --git a/ap2-receiver.py b/ap2-receiver.py index e7e6c4d..327cb36 100644 --- a/ap2-receiver.py +++ b/ap2-receiver.py @@ -23,6 +23,7 @@ from ap2.pairing.hap import Hap, HAPSocket, LTPK, DeviceProperties from ap2.connections.event import EventGeneric from ap2.connections.stream import Stream +from ap2.connections import ptp as ptp_slave_mod from ap2.connections.session_properties import Session from ap2.dxxp import parse_dxxp from ap2.bitflags import FeatureFlags, StatusFlags @@ -208,19 +209,39 @@ def setup_global_structs(args, isDebug=False): 'eventPort': 0 # AP2 receiver event server } if not DISABLE_PTP_MASTER: - if IPV6 and not IPV4: - addr = [ - IPV6 - ] - else: - # Prefer (only) IPV4 - addr = [ - IPV4 - ] + # iOS 17+/26 uses this Addresses list as candidate PTP peer addresses + # and discards the session early (TEARDOWN before second SETUP) when + # the list is too narrow. Enumerate every non-loopback IPv4 + IPv6 + # address on the host so iOS can pick the one that's actually + # routable from its end. + addr = [] + try: + for ifname in ni.interfaces(): + ifaddrs = ni.ifaddresses(ifname) + for fam in (ni.AF_INET, ni.AF_INET6): + for entry in ifaddrs.get(fam, []): + raw = entry.get("addr", "") + if not raw: + continue + # Strip IPv6 zone id ("fe80::1%eth0") + ip = raw.split("%", 1)[0] + if ip.startswith("127.") or ip == "::1": + continue + if ip not in addr: + addr.append(ip) + except Exception: # noqa: BLE001 + pass + if not addr: + # Fallback to the bound interface IP only + addr = [IPV6] if (IPV6 and not IPV4) else [IPV4] device_setup['timingPort'] = 0 # Seems like legacy, non PTP setting + # iOS 17+/26: ID must be an IP string (matching one of the Addresses), + # not the MAC. shairport-sync uses self_ip_string here. Pass the + # primary v4 (or v6 if no v4) as the IPv6/v4-shaped peer identity. + peer_id = IPV4 or IPV6 or DEVICE_ID device_setup['timingPeerInfo'] = { 'Addresses': addr, - 'ID': DEVICE_ID + 'ID': peer_id, } stream_setup_data = { @@ -407,12 +428,12 @@ def do_OPTIONS(self): apple_response = AP1Security.compute_apple_response(self.headers["Apple-Challenge"], IPADDR_BIN, DEVICE_ID_BIN) self.send_header("Apple-Jack-Status", "connected; type=analog") self.send_header("Apple-Response", apple_response) - self.send_header("Public", - "ANNOUNCE, SETUP, RECORD, PAUSE, FLUSH" - "FLUSHBUFFERED, TEARDOWN, OPTIONS, POST, GET, PUT" - "SETPEERSX" - "SETMAGICCOOKIE" - ) + self.send_header( + "Public", + "ANNOUNCE, SETUP, RECORD, PAUSE, FLUSH, FLUSHBUFFERED, " + "TEARDOWN, OPTIONS, POST, GET, PUT, " + "SETPEERS, SETPEERSX, SETRATEANCHORTIME, SETMAGICCOOKIE" + ) self.end_headers() def do_ANNOUNCE(self): @@ -525,7 +546,8 @@ def do_SETUP(self): stream_id=increase_stream_id(), shared_key=self.ecdh_shared_key, isDebug=DEBUG, - aud_params=self.aud_params + aud_params=self.aud_params, + ptp_clock_array=self.server.ptp_clock_array, ) self.server.streams.append(streamobj) @@ -602,6 +624,7 @@ def do_SETUP(self): stream_id=increase_stream_id(), shared_key=self.ecdh_shared_key, isDebug=DEBUG, + ptp_clock_array=self.server.ptp_clock_array, ) self.logger.debug("Building stream channels:") self.server.streams.append(s) @@ -611,10 +634,14 @@ def do_SETUP(self): self.logger.debug(s.getSummaryMessage()) + # pycaw matches the per-process audio session by PID; + # PyAudio's sink lives in the *data* proc, not the + # RTCP control proc — passing the wrong PID makes + # SET_PARAMETER volume silently fail on Windows. if s.getStreamType() == Stream.BUFFERED: - set_volume_pid(s.getControlProc().pid) + set_volume_pid(s.getDataProc().pid) if s.getStreamType() == Stream.REALTIME: - set_volume_pid(s.getControlProc().pid) + set_volume_pid(s.getDataProc().pid) self.logger.debug(self.pp.pformat(stream_setup_data)) res = writePlistToString(stream_setup_data) @@ -785,6 +812,10 @@ def do_TEARDOWN(self): self.logger.info(self.pp.pformat(plist)) if plist == {} and len(self.server.streams) == 0: self.server.event_proc.terminate() + # Release the PTP slave too — otherwise it keeps the + # UDP/319 + UDP/320 ports bound and keeps sending + # Delay_Req to a sender that has already gone away. + self._stop_ptp_slave() self.send_response(200) self.send_header("Server", self.version_string()) self.send_header("CSeq", self.headers["CSeq"]) @@ -811,6 +842,15 @@ def do_SETPEERS(self): plist = readPlistFromString(body) self.logger.info(self.pp.pformat(plist)) + if isinstance(plist, list): + # PTPSlave binds AF_INET sockets and sendto()s the master's + # IP — feeding an IPv6 (e.g. fe80::...) into that call + # raises immediately. Filter to dotted-quad IPv4 only. + ips = [ + str(x) for x in plist + if isinstance(x, (str, bytes)) and str(x).count(".") == 3 + ] + self._maybe_start_ptp_slave(ips, master_clock_identity=None) self.send_response(200) self.send_header("Server", self.version_string()) self.send_header("CSeq", self.headers["CSeq"]) @@ -833,7 +873,6 @@ def do_SETPEERSX(self): # 'ID': GUID, # 'SupportsClockPortMatchingOverride': T/F} - # SETPEERSX may require more logic when PTP is finished. self.logger.info(f'{self.command}: {self.path}') self.logger.debug(self.headers) content_len = int(self.headers["Content-Length"]) @@ -842,11 +881,86 @@ def do_SETPEERSX(self): plist = readPlistFromString(body) self.logger.info(self.pp.pformat(plist)) + # peer list is an array of dicts; pick the first IPv4 address from + # each peer. ClockID in iOS plists is an int — convert to 8 bytes. + ips = [] + clock_id_bytes = None + if isinstance(plist, list): + for peer in plist: + if not isinstance(peer, dict): + continue + addrs = peer.get("Addresses") or [] + for a in addrs: + try: + sa = str(a) + except Exception: # noqa: BLE001 + continue + # Filter to dotted-quad (skip IPv6 link-local for slave's v4 sockets) + if sa.count(".") == 3: + ips.append(sa) + break + if clock_id_bytes is None and "ClockID" in peer: + cid = peer["ClockID"] + try: + clock_id_bytes = int(cid).to_bytes(8, "big", signed=False) + except (TypeError, OverflowError, ValueError): + clock_id_bytes = None + self._maybe_start_ptp_slave(ips, master_clock_identity=clock_id_bytes) self.send_response(200) self.send_header("Server", self.version_string()) self.send_header("CSeq", self.headers["CSeq"]) self.end_headers() + def _maybe_start_ptp_slave(self, ips, master_clock_identity): + """Start (or restart) the PTP slave when the peer list changes.""" + if not ips: + return + key = (tuple(ips), master_clock_identity) + if self.server.ptp_master_key == key and self.server.ptp_proc and self.server.ptp_proc.is_alive(): + self.logger.debug("PTP slave already locked to this peer set; reuse") + return + # Tear down any old slave + old = self.server.ptp_proc + if old and old.is_alive(): + self.logger.info("PTP peer list changed; terminating previous slave") + try: + old.terminate() + old.join(timeout=1.0) + except Exception: # noqa: BLE001 + pass + try: + proc, clock = ptp_slave_mod.spawn( + master_ips=ips, + master_clock_identity=master_clock_identity, + is_debug=DEBUG, + shared_array=self.server.ptp_clock_array, + ) + except Exception as e: # noqa: BLE001 + self.logger.error(f"failed to spawn PTP slave: {e!r}") + return + self.server.ptp_proc = proc + self.server.ptp_clock = clock + self.server.ptp_master_key = key + self.logger.info( + f"PTP slave spawned: master_ips={ips} " + f"clockID={master_clock_identity.hex() if master_clock_identity else 'unknown'}" + ) + + def _stop_ptp_slave(self): + """Tear down the PTP slave process and release its UDP ports.""" + proc = getattr(self.server, "ptp_proc", None) + if proc and proc.is_alive(): + try: + proc.terminate() + proc.join(timeout=1.0) + except Exception: # noqa: BLE001 + pass + self.server.ptp_proc = None + self.server.ptp_master_key = None + # ptp_clock_array is reused across sessions — don't drop it, + # only the wrapper view if held. + self.server.ptp_clock = None + def do_FLUSH(self): self.logger.info(f'{self.command}: {self.path}') self.logger.debug(self.headers) @@ -1250,6 +1364,15 @@ def __init__(self, addr_port, handler): self.event_port = None self.timing_proc = None self.timing_port = None + # PTP slave state (one per session). master_key identifies which + # peer-list we are currently locked to so we can avoid respawning + # the slave on identical SETPEERSX repeats. + self.ptp_proc = None + self.ptp_clock = None + self.ptp_master_key = None + # Allocated once per server: PTP slave + every audio child reads/writes + # the same shared array. + self.ptp_clock_array = ptp_slave_mod.make_shared_array() self.enc_layer = False self.streams = [] self.sessions = [] @@ -1368,10 +1491,15 @@ def generate_fake_mac(): DISABLE_PTP_MASTER = args.no_ptp_master DEV_PROPS = DeviceProperties(PI, DEBUG) DEV_NAME = args.mdns - if(parser.get_default('mdns') != DEV_NAME): + if parser.get_default('mdns') != DEV_NAME: DEV_PROPS.setDeviceName(DEV_NAME) else: - DEV_NAME = DEV_PROPS.getDeviceName() + persisted = DEV_PROPS.getDeviceName() + if persisted: + DEV_NAME = persisted + else: + # First run with default name: persist so future restarts read it back. + DEV_PROPS.setDeviceName(DEV_NAME) SCR_LOG.info(f"Name: {DEV_NAME}") pw = DEV_PROPS.getDevicePassword() hkacl = DEV_PROPS.isHKACLEnabled() diff --git a/ap2/bitflags.py b/ap2/bitflags.py index bbe3397..d92a78b 100644 --- a/ap2/bitflags.py +++ b/ap2/bitflags.py @@ -150,17 +150,22 @@ class FeatureFlags(IntFlag): iOS just fails at Pair-Setup [2/5] """ def GetDefaultAirplayTwoFlags(self): + # Exactly matches shairport-sync master (0x1c340405c4a00 after its + # internal masking of bits 15/16/17/50): adds Ft11 (AudioRedundant) + # and Ft38 (ControlChannelEncrypt); drops Ft16/Ft17 metadata bits. + # iOS 17+/26 narrows acceptable feature sets — this is the closest + # known-working set published in any open-source receiver. return ( self.Ft48TransientPairing | self.Ft47PeerManagement | self.Ft46HomeKitPairing | self.Ft41_PTPClock | self.Ft40BufferedAudio + | self.Ft38ControlChannelEncrypt | self.Ft30UnifiedAdvertisingInfo | self.Ft22AudioUnencrypted | self.Ft20ReceiveAudioAAC_LC | self.Ft19ReceiveAudioALAC | self.Ft18ReceiveAudioPCM - | self.Ft17AudioMetaTxtDAAP - | self.Ft16AudioMetaProgress - # | self.Ft15AudioMetaCovers - | self.Ft14MFiSoft_FairPlay | self.Ft09AirPlayAudio + | self.Ft14MFiSoft_FairPlay + | self.Ft11AudioRedundant + | self.Ft09AirPlayAudio ) # Generic names to simplify usage (don't need to track changes in receiver) diff --git a/ap2/connections/audio.py b/ap2/connections/audio.py index 5500a7f..0122f70 100644 --- a/ap2/connections/audio.py +++ b/ap2/connections/audio.py @@ -427,6 +427,7 @@ def __init__( control_conns=None, isDebug=False, aud_params: AudioSetup = None, + ptp_clock_array=None, ): self.isDebug = isDebug self.addr = addr @@ -449,14 +450,23 @@ def __init__( self.senderRtpTimestamp, self.playAtRtpTimestamp = None, None self.remoteClockMonotonic_ts, self.remoteClockId = None, None + # PTP disciplined clock (shared Array; may be untouched / unsynced). + self._ptp_clock_array = ptp_clock_array + self._ptp_clock = None # lazily built in the audio process + def init_audio_sink(self): codecLatencySec = 0 self.pa = pyaudio.PyAudio() + # PyAudio frames_per_buffer of 4 (the upstream default) is far too + # small on Windows: WASAPI/WDM scheduling can't service that and the + # output stream underruns continuously, producing crackle/zipper + # noise. Use 1024 frames (~23 ms at 44.1 kHz) which matches the + # AirPlay packet cadence and keeps latency low. self.sink = self.pa.open(format=self.pa.get_format_from_width(2), channels=self.channel_count, rate=self.sample_rate, output=True, - frames_per_buffer=4, + frames_per_buffer=1024, ) # nice Python3 crash if we don't check self.sink is null. Not harmful, but should check. if not self.sink: @@ -500,14 +510,23 @@ def init_audio_sink(self): 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 + # PyAV >= 14 made `channels` read-only; set via `layout` instead. + # AirPlay 2 audio is stereo or mono; mono comes through as the + # `_1` suffix on the AirplayAudFmt enum (channel_count == 1). + if hasattr(self.codecContext, "layout"): + self.codecContext.layout = "mono" if self.channel_count == 1 else "stereo" + else: + self.codecContext.channels = self.channel_count self.codecContext.format = av.AudioFormat('s' + str(self.sample_size) + 'p') if ed is not None: self.codecContext.extradata = ed self.resampler = av.AudioResampler( format=av.AudioFormat('s' + str(self.sample_size)).packed, - layout='stereo', + # Must match channel_count: PyAudio's sink was opened with + # channels=self.channel_count, so up/down-mixing here would + # produce a size mismatch when we write the bytes back. + layout='mono' if self.channel_count == 1 else 'stereo', rate=self.sample_rate, ) @@ -517,9 +536,62 @@ def init_audio_sink(self): self.audio_screen_logger.debug(f"audioDevicelatency (sec): {audioDevicelatency:0.5f}") pyAudioDelay = self.sink.get_output_latency() self.audio_screen_logger.debug(f"pyAudioDelay (sec): {pyAudioDelay:0.5f}") - ptpDelay = 0.002 + # PTP path delay: pulled from the disciplined clock if synced; + # otherwise fall back to the historical 2 ms guess. + ptpDelay = self._get_ptp_path_delay_sec() self.sample_delay = pyAudioDelay + audioDevicelatency + codecLatencySec + ptpDelay - self.audio_screen_logger.info(f"Total sample_delay (sec): {self.sample_delay:0.5f}") + self.audio_screen_logger.info( + f"Total sample_delay (sec): {self.sample_delay:0.5f} (ptp_path_delay={ptpDelay*1000:.3f}ms)" + ) + + def _ensure_ptp_clock(self): + """Bind the shared Array to a PTPDisciplinedClock view (lazy, since + the audio process can't construct it until it starts).""" + if self._ptp_clock is None and self._ptp_clock_array is not None: + from .ptp_clock import PTPDisciplinedClock + self._ptp_clock = PTPDisciplinedClock(self._ptp_clock_array) + return self._ptp_clock + + def _get_ptp_path_delay_sec(self) -> float: + clk = self._ensure_ptp_clock() + if clk is not None and clk.is_synced(): + return clk.get_mean_path_delay_ns() * 1e-9 + return 0.002 + + def _ptp_now_ns(self) -> int: + """Wall-clock source for the audio playback scheduler. + + Returns local monotonic ns — *not* the PTP-disciplined clock. PTP + re-steps (which can be >100 ms on Wi-Fi when the servo sees outlier + Sync samples) would otherwise propagate into msec_to_playout and + stall single-receiver playback, even though steady-state PTP + residual is <2 ms. + + Multi-room sync (the legitimate use case for feeding disciplined + time into the scheduler) needs outlier rejection + slow-slew in + the servo before this can switch back; until then the PTP slave + keeps running but is read-only via self._ensure_ptp_clock() for + diagnostics.""" + return time.monotonic_ns() + + def _wait_for_ptp_sync(self, timeout_sec: float = 1.5) -> bool: + """Log current PTP status (no longer blocks playback). + + Kept as a hook for future multi-room work where we'd want to wait + for cross-receiver clock agreement before starting playback. For + the single-receiver case the audio scheduler uses local monotonic + time, so blocking here just adds startup latency without benefit. + """ + clk = self._ensure_ptp_clock() + if clk is None: + return False + if clk.is_synced(): + self.audio_screen_logger.debug( + f"PTP at playback start: offset={clk.get_offset_ns()/1e6:+.3f}ms " + f"mpd={clk.get_mean_path_delay_ns()/1e6:.3f}ms" + ) + return True + return False def decrypt(self, rtp): data = b'' @@ -568,11 +640,21 @@ def process(self, rtp): if(len(data) > 0): try: for frame in self.codecContext.decode(packet): - frame = self.resampler.resample(frame) - if isinstance(frame, list): - if len(frame) == 1: # if no resampling was needed, resample returns [frame] - frame = frame[0] - return bytes(frame.planes[0]) + out = self.resampler.resample(frame) + if isinstance(out, list): + if not out: + continue + out = out[0] + # Always produce packed interleaved PCM. PyAV's + # `frame.format.is_planar` is the authoritative flag — + # avoid shape heuristics (packed audio happens to come + # back as (1, N*ch) in current PyAV, but that's an + # implementation detail we shouldn't rely on). + arr = out.to_ndarray() + if out.format.is_planar and arr.ndim == 2: + # (channels, samples) -> interleaved + arr = arr.T.reshape(-1) + return arr.tobytes() except ValueError as e: self.audio_screen_logger.error(repr(e)) pass # noqa @@ -593,19 +675,24 @@ def run(self, rcvr_cmd_pipe, control_conns): def msec_to_playout(self, rtp_ts): """ - msec until intended playout of RTP packet with timestamp rtp_ts + msec until intended playout of RTP packet with timestamp rtp_ts. + + The "now" reading comes from the PTP-disciplined clock (master time + domain) when locked, falling back to local perf_counter when not. + anchorMonotonicNanosLocal is captured in the same domain so the + subtraction stays consistent across re-locks. """ if not self.anchorRTPTimestamp: return 0 rtp_ts_diff = rtp_ts - self.anchorRTPTimestamp - millis_to_anchor = int((time.monotonic_ns() - self.anchorMonotonicNanosLocal) * 1e-6) + millis_to_anchor = int((self._ptp_now_ns() - self.anchorMonotonicNanosLocal) * 1e-6) return int(1000 * rtp_ts_diff / self.sample_rate) - millis_to_anchor def msec_to_playout_with_outdev_delay(self, rtp_ts): return int(self.msec_to_playout(rtp_ts) - ((self.sample_delay * 1e3))) def samples_elapsed_since_anchor(self): - realtime_offset_sec = (time.monotonic_ns() - self.anchorMonotonicNanosLocal) * 1e-9 + realtime_offset_sec = (self._ptp_now_ns() - self.anchorMonotonicNanosLocal) * 1e-9 samples_to_playhead = self.anchorRTPTimestamp + realtime_offset_sec * self.sample_rate return samples_to_playhead @@ -620,6 +707,7 @@ def spawn( control_conns=None, isDebug=False, aud_params: AudioSetup = None, + ptp_clock_array=None, ): audio = cls( addr, @@ -630,6 +718,7 @@ def spawn( control_conns, isDebug, aud_params, + ptp_clock_array, ) # This pipe is reachable from receiver rcvr_cmd_pipe, audio.command_chan = multiprocessing.Pipe() @@ -652,7 +741,8 @@ def __init__( streamtype, control_conns=None, isDebug=False, - aud_params: AudioSetup = None + aud_params: AudioSetup = None, + ptp_clock_array=None, ): super(AudioRealtime, self).__init__( addr, @@ -662,13 +752,19 @@ def __init__( streamtype, control_conns, isDebug, - aud_params + aud_params, + ptp_clock_array, ) self.isDebug = isDebug self.socket = get_free_socket() if not addr else addr self.port = self.socket.getsockname()[1] self.rtp_buffer = RTPRealtimeBuffer(buff_size, self.isDebug) self.anchorRTPTimestamp = None + # seq_no -> last-requested monotonic_ns. Used to suppress the + # resend-storm in serve(): without this we re-issue REXMIT_REQUEST + # for the same missing seq on every RTCP cycle, flooding the + # sender's control port and producing 2 s playback hiccups on Wi-Fi. + self._resend_pending_ns = {} def fini_audio_sink(self): self.sink.close() @@ -697,16 +793,27 @@ def serve(self, serverconn, control_recv, control_send): """ if self.rtp_buffer.gaps_exist(): + now_ns = time.monotonic_ns() + # Re-request the same seq at most every 250 ms — gives the + # sender time to actually answer the previous request. + cooldown_ns = 250_000_000 + # Garbage-collect tracking entries older than 2 s + self._resend_pending_ns = { + s: t for s, t in self._resend_pending_ns.items() + if (now_ns - t) < 2_000_000_000 + } for missing_seq in self.rtp_buffer.missing_sequence_nos(): # Each missing_seq is a tuple: (Seq#, amount_following) + seq = missing_seq[0] + last = self._resend_pending_ns.get(seq, 0) + if (now_ns - last) < cooldown_ns: + continue + self._resend_pending_ns[seq] = now_ns self.audio_screen_logger.debug( - f'requesting resend of sequence_no {missing_seq[0]}; amt {missing_seq[1]}' + f'requesting resend of sequence_no {seq}; amt {missing_seq[1]}' ) - # request resend via control channel here - """ syntax: - resend_{missing_seq_no_start}/{amount_following}/{optional_timestamp} - """ - control_send.put(f'resend_{missing_seq[0]}/{missing_seq[1]}/{0}') + # syntax: resend_{missing_seq_no_start}/{amount_following}/{optional_timestamp} + control_send.put(f'resend_{seq}/{missing_seq[1]}/{0}') # Wake every ~fifth packet time.sleep((self.spf / self.sample_rate) * 5) @@ -752,7 +859,8 @@ def play(self, rtspconn, serverconn): else: rtp = self.rtp_buffer.pop(0) if starting: - self.anchorMonotonicNanosLocal = time.monotonic_ns() + self._wait_for_ptp_sync() + self.anchorMonotonicNanosLocal = self._ptp_now_ns() starting = False if rtp: @@ -811,7 +919,8 @@ def __init__( streamtype=0, control_conns=None, isDebug=False, - aud_params: AudioSetup = None + aud_params: AudioSetup = None, + ptp_clock_array=None, ): super(AudioBuffered, self).__init__( addr, @@ -822,6 +931,7 @@ def __init__( control_conns, isDebug, aud_params, + ptp_clock_array, ) self.isDebug = isDebug @@ -874,7 +984,8 @@ def play(self, rtspconn, serverconn): message = rtspconn.recv() if isinstance(message, str): if str.startswith(message, "play"): - self.anchorMonotonicNanosLocal = time.monotonic_ns() + self._wait_for_ptp_sync() + self.anchorMonotonicNanosLocal = self._ptp_now_ns() self.anchorRTPTimestamp = int(str.split(message, "-")[1]) playing = True diff --git a/ap2/connections/ptp.py b/ap2/connections/ptp.py new file mode 100644 index 0000000..c45f353 --- /dev/null +++ b/ap2/connections/ptp.py @@ -0,0 +1,519 @@ +""" +PTPv2 ordinary-clock SLAVE for AirPlay 2 receivers. + +Spawned as a multiprocessing.Process when the sender announces PTP peers via +SETPEERS/SETPEERSX. Locks to the sender's grand-master clock and updates a +shared PTPDisciplinedClock that the audio path consumes. + +Scope and limitations: + * Unicast only. AirPlay 2 senders address our PTP ports directly. + * No BMCA: we trust the master IP/ClockID handed to us by SETPEERSX. + * One-step or two-step Sync both supported. + * Software timestamping only; on Windows perf_counter is QPC (~100 ns) but + OS scheduling / NIC buffering still adds tens-of-µs jitter on Wi-Fi. + * Servo is a simple step-then-PI on offset; frequency is derived from the + slope of offset estimates over time. +""" + +import logging +import multiprocessing +import os +import random +import select +import socket +import struct +import threading +import time +from collections import deque + +from . import ptp_messages as ptpm +from .ptp_clock import PTPDisciplinedClock, now_local_ns + + +PTP_EVENT_PORT = ptpm.PTP_EVENT_PORT +PTP_GENERAL_PORT = ptpm.PTP_GENERAL_PORT + +# Tunables +DELAY_REQ_INTERVAL_S = 1.0 # how often we initiate Delay_Req +MAX_SYNC_HISTORY = 32 # samples kept for slope/jitter detection +INITIAL_STEP_THRESHOLD_NS = 1_000_000 # 1 ms: above this on first lock, step +RUNNING_STEP_THRESHOLD_NS = 100_000_000 # 100 ms: re-step instead of slewing +OUTLIER_REJECT_FACTOR = 4.0 # reject samples whose path delay >> median +ANNOUNCE_TIMEOUT_S = 6.0 # if no Announce/Sync for this long, drop sync +PI_KP = 0.7 # proportional gain on offset (fraction applied) +PI_KI = 0.05 # integral gain (slow correction of bias) + + +def _build_clock_identity() -> bytes: + """Synthesize an 8-byte clock identity for this slave. + + Real PTP uses MAC+two-byte fill (EUI-64). For a software slave the + identity only matters for echoing into our own Delay_Req source port: + the master uses it in Delay_Resp's requestingPortIdentity so we can + pair our request to its response. We randomize and persist for the + process lifetime. + """ + # 0x02 in the first byte marks it as locally-administered, harmless here. + seed = random.getrandbits(48).to_bytes(6, "big") + return b"\x02" + seed + b"\x01" + + +def _make_event_socket() -> socket.socket: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + s.bind(("0.0.0.0", PTP_EVENT_PORT)) + return s + + +def _make_general_socket() -> socket.socket: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + s.bind(("0.0.0.0", PTP_GENERAL_PORT)) + return s + + +class _PendingSync: + __slots__ = ("seq", "t2_local_ns", "t1_master_ns", "two_step") + + def __init__(self, seq, t2_local_ns, t1_master_ns, two_step): + self.seq = seq + self.t2_local_ns = t2_local_ns + self.t1_master_ns = t1_master_ns + self.two_step = two_step + + +class _PendingDelayReq: + __slots__ = ("seq", "t3_local_ns") + + def __init__(self, seq, t3_local_ns): + self.seq = seq + self.t3_local_ns = t3_local_ns + + +class PTPSlave: + """One slave instance bound to one master. + + Run as a child process via spawn(); communicates with the parent through + the shared array inside ``self.clock``. + """ + + STATE_LISTENING = "LISTENING" + STATE_UNCALIBRATED = "UNCALIBRATED" + STATE_SLAVE = "SLAVE" + + def __init__( + self, + master_ips: list[str], + master_clock_identity: bytes | None, + shared_clock_array, + is_debug: bool = False, + log_path: str | None = None, + ): + self.master_ips = list(master_ips) + self.master_clock_identity = master_clock_identity + self.is_debug = is_debug + self.log_path = log_path + self.clock = PTPDisciplinedClock(shared_clock_array) + self.logger = None + + self._stop = False + self._state = self.STATE_LISTENING + self._local_clock_identity = _build_clock_identity() + self._local_port_identity = ptpm.PortIdentity(self._local_clock_identity, 1) + + self._event_sock: socket.socket | None = None + self._general_sock: socket.socket | None = None + + self._delay_req_seq = random.getrandbits(15) # avoid 0 + self._pending_syncs: dict[int, _PendingSync] = {} + self._pending_delay_reqs: dict[int, _PendingDelayReq] = {} + + self._sample_history: deque = deque(maxlen=MAX_SYNC_HISTORY) + self._offset_history: deque = deque(maxlen=8) + self._mean_path_delay_ns = 0 + self._integral_ns = 0.0 + self._last_master_seen_local_ns = 0 + self._last_delay_req_local_ns = 0 + self._first_lock = True + + # ------------- logging ------------- + + def _setup_logger(self): + log = logging.getLogger(f"PTPSlave-{os.getpid()}") + log.setLevel(logging.DEBUG if self.is_debug else logging.INFO) + if not log.handlers: + ch = logging.StreamHandler() + ch.setFormatter(logging.Formatter("%(asctime)s [PTP %(levelname)s] %(message)s")) + log.addHandler(ch) + if self.log_path: + try: + fh = logging.FileHandler(self.log_path) + fh.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")) + log.addHandler(fh) + except OSError: + pass + self.logger = log + + def _log(self, msg, level=logging.INFO): + if self.logger: + self.logger.log(level, msg) + + # ------------- main entry ------------- + + def run(self): + self._setup_logger() + self._log( + f"starting; masters={self.master_ips} " + f"clockID={self.master_clock_identity.hex() if self.master_clock_identity else 'any'}" + ) + try: + self._event_sock = _make_event_socket() + self._general_sock = _make_general_socket() + except OSError as e: + self._log(f"failed to bind PTP ports 319/320: {e!r} — slave aborting", logging.ERROR) + self.clock.set_status(PTPDisciplinedClock.STATUS_UNSYNCED) + return + + self._log(f"bound UDP/319 + UDP/320; localClockID={self._local_clock_identity.hex()}") + self.clock.set_status(PTPDisciplinedClock.STATUS_ACQUIRING) + try: + self._loop() + except KeyboardInterrupt: + pass + except Exception as e: # noqa: BLE001 + self._log(f"unexpected error: {e!r}", logging.ERROR) + finally: + try: + self._event_sock.close() + except OSError: + pass + try: + self._general_sock.close() + except OSError: + pass + self.clock.set_status(PTPDisciplinedClock.STATUS_UNSYNCED) + self._log("stopped") + + def _loop(self): + socks = [self._event_sock, self._general_sock] + while not self._stop: + # Periodically initiate Delay_Req so we get path-delay samples. + now = now_local_ns() + if (now - self._last_delay_req_local_ns) >= int(DELAY_REQ_INTERVAL_S * 1e9): + self._send_delay_req() + self._last_delay_req_local_ns = now + + # Detect master disappearance. + if ( + self._last_master_seen_local_ns + and (now - self._last_master_seen_local_ns) > int(ANNOUNCE_TIMEOUT_S * 1e9) + and self._state == self.STATE_SLAVE + ): + self._log("no PTP traffic from master — falling back to acquiring", logging.WARNING) + self._state = self.STATE_UNCALIBRATED + self.clock.set_status(PTPDisciplinedClock.STATUS_ACQUIRING) + + try: + ready, _, _ = select.select(socks, [], [], 0.2) + except (OSError, ValueError): + break + for s in ready: + try: + data, addr = s.recvfrom(2048) + except OSError: + continue + rx_ns = now_local_ns() + if not self._is_from_master(addr): + continue + self._last_master_seen_local_ns = rx_ns + if len(data) < ptpm.PTP_HEADER_LEN: + continue + try: + msg = ptpm.parse(data) + except (struct.error, ValueError): + continue + self._dispatch(msg, rx_ns, addr, s is self._event_sock) + + # ------------- helpers ------------- + + def _is_from_master(self, addr: tuple) -> bool: + ip = addr[0] + if not self.master_ips: + return True + if ip in self.master_ips: + return True + # iOS sometimes uses link-local v6 we filter v4-only; allow when only + # v6 candidates were given. + return False + + def _master_addr_for(self, port: int) -> tuple | None: + if not self.master_ips: + return None + return (self.master_ips[0], port) + + # ------------- message dispatch ------------- + + def _dispatch(self, msg: ptpm.PtpMessage, rx_local_ns: int, addr: tuple, on_event_socket: bool): + mt = msg.msg_type + if mt == ptpm.MsgType.ANNOUNCE: + self._on_announce(msg) + elif mt == ptpm.MsgType.SYNC: + self._on_sync(msg, rx_local_ns) + elif mt == ptpm.MsgType.FOLLOW_UP: + self._on_follow_up(msg) + elif mt == ptpm.MsgType.DELAY_RESP: + self._on_delay_resp(msg) + # ignore Pdelay / signaling / management + + def _on_announce(self, msg: ptpm.PtpMessage): + if self._state == self.STATE_LISTENING: + self._state = self.STATE_UNCALIBRATED + self._log(f"master alive (seq={msg.header.sequence_id}) — UNCALIBRATED") + + def _on_sync(self, msg: ptpm.PtpMessage, t2_local_ns: int): + if not isinstance(msg.body, ptpm.SyncBody): + return + seq = msg.header.sequence_id + two_step = bool(msg.header.flags & ptpm.FlagBit.TWO_STEP) + if two_step: + # Master will send Follow_Up with the precise origin timestamp. + self._pending_syncs[seq] = _PendingSync( + seq=seq, t2_local_ns=t2_local_ns, + t1_master_ns=0, two_step=True, + ) + else: + # One-step: this Sync already carries the originTimestamp. + t1 = msg.body.origin_timestamp_ns + msg.header.correction_ns + self._record_sync_sample(t1_master_ns=t1, t2_local_ns=t2_local_ns) + + # Garbage-collect stale pending syncs (>5s old) + cutoff = now_local_ns() - 5_000_000_000 + stale = [s for s, p in self._pending_syncs.items() if p.t2_local_ns < cutoff] + for s in stale: + self._pending_syncs.pop(s, None) + + def _on_follow_up(self, msg: ptpm.PtpMessage): + if not isinstance(msg.body, ptpm.FollowUpBody): + return + pending = self._pending_syncs.pop(msg.header.sequence_id, None) + if not pending: + return + t1 = msg.body.precise_origin_timestamp_ns + msg.header.correction_ns + self._record_sync_sample(t1_master_ns=t1, t2_local_ns=pending.t2_local_ns) + + def _on_delay_resp(self, msg: ptpm.PtpMessage): + if not isinstance(msg.body, ptpm.DelayRespBody): + return + # Ensure the response was addressed to *our* port identity. + if msg.body.requesting_port_identity.clock_identity != self._local_clock_identity: + return + pending = self._pending_delay_reqs.pop(msg.header.sequence_id, None) + if not pending: + return + t4_master_ns = msg.body.receive_timestamp_ns - msg.header.correction_ns + self._record_delay_sample(t3_local_ns=pending.t3_local_ns, t4_master_ns=t4_master_ns) + + # ------------- TX ------------- + + def _send_delay_req(self): + addr = self._master_addr_for(PTP_EVENT_PORT) + if not addr or not self._event_sock: + return + self._delay_req_seq = (self._delay_req_seq + 1) & 0xFFFF + seq = self._delay_req_seq + pkt = ptpm.build_delay_req(self._local_port_identity, seq) + try: + t3 = now_local_ns() + self._event_sock.sendto(pkt, addr) + except OSError as e: + self._log(f"sendto Delay_Req failed: {e!r}", logging.DEBUG) + return + self._pending_delay_reqs[seq] = _PendingDelayReq(seq=seq, t3_local_ns=t3) + # GC stale + if len(self._pending_delay_reqs) > 64: + oldest = min(self._pending_delay_reqs, key=lambda k: self._pending_delay_reqs[k].t3_local_ns) + self._pending_delay_reqs.pop(oldest, None) + + # ------------- servo ------------- + + def _record_sync_sample(self, t1_master_ns: int, t2_local_ns: int): + # Tentative offset using last known mean_path_delay. + # offset = t2 - t1 - mean_path_delay + offset = t2_local_ns - t1_master_ns - self._mean_path_delay_ns + self._sample_history.append(("sync", t1_master_ns, t2_local_ns, offset)) + self._apply_offset_estimate(offset, t2_local_ns) + + def _record_delay_sample(self, t3_local_ns: int, t4_master_ns: int): + # Need a recent t1/t2 to compute mean_path_delay; use most recent sync. + recent_sync = None + for entry in reversed(self._sample_history): + if entry[0] == "sync": + recent_sync = entry + break + if not recent_sync: + return + _, t1, t2, _ = recent_sync + # mean_path_delay = ((t2 - t1) + (t4 - t3)) / 2 in master-equivalent ns + # Note: this assumes the offset is small / stable across the exchange. + forward = t2 - t1 + reverse = t4_master_ns - t3_local_ns + mpd = (forward + reverse) // 2 + if mpd < 0: + # invalid (clock not yet aligned), skip + return + # Reject outliers: keep median-based gate over the last 8 mpd samples. + if self._mean_path_delay_ns > 0: + limit = max(int(self._mean_path_delay_ns * OUTLIER_REJECT_FACTOR), 1_000_000) + if mpd > limit: + self._log(f"reject outlier mpd={mpd/1e6:.2f}ms (limit {limit/1e6:.2f}ms)", logging.DEBUG) + return + # EMA over mpd + if self._mean_path_delay_ns == 0: + self._mean_path_delay_ns = mpd + else: + self._mean_path_delay_ns = int(0.875 * self._mean_path_delay_ns + 0.125 * mpd) + # Recompute offset of the recent sync with updated mpd + new_offset = t2 - t1 - self._mean_path_delay_ns + self._apply_offset_estimate(new_offset, t2) + + def _apply_offset_estimate(self, raw_offset_ns: int, sample_local_ns: int): + """ + raw_offset_ns is the unprocessed (t2 - t1 - mpd) estimate of how far + the *local raw* clock lags behind the *master raw* clock. To bring + the disciplined clock onto the master timeline we need to *apply* + `-raw_offset_ns` to local time — i.e. master = local + (-raw_offset). + + The "residual" is the gap between what we're currently applying and + what we should apply: residual = (-raw_offset) - applied. When small, + run a P servo over the residual. When large, step. + """ + self._offset_history.append((sample_local_ns, raw_offset_ns)) + freq_ppb = self._estimate_freq_ppb() + + target_offset = -raw_offset_ns + applied = self.clock.get_offset_ns() + residual = target_offset - applied + + if self._first_lock: + self._first_lock = False + new_offset = target_offset + self._integral_ns = 0.0 + status = PTPDisciplinedClock.STATUS_SYNCED + self._state = self.STATE_SLAVE + self._log( + f"initial lock: raw_offset={raw_offset_ns/1e6:.3f}ms " + f"applied={new_offset/1e6:+.3f}ms " + f"mpd={self._mean_path_delay_ns/1e6:.3f}ms", + ) + elif abs(residual) > RUNNING_STEP_THRESHOLD_NS: + new_offset = target_offset + self._integral_ns = 0.0 + status = PTPDisciplinedClock.STATUS_ACQUIRING + self._log( + f"large residual {residual/1e6:+.3f}ms — re-stepping to {target_offset/1e6:+.3f}ms", + logging.WARNING, + ) + else: + self._integral_ns += residual * PI_KI + self._integral_ns = max(min(self._integral_ns, 10_000_000.0), -10_000_000.0) + new_offset = applied + int(residual * PI_KP + self._integral_ns) + status = PTPDisciplinedClock.STATUS_SYNCED + + self.clock.update( + offset_ns=new_offset, + freq_ppb=freq_ppb, + mean_path_delay_ns=self._mean_path_delay_ns, + sync_local_ns=sample_local_ns, + status=status, + ) + + if self.clock.get_sync_count() % 8 == 0: + self._log( + f"servo: applied={new_offset/1e6:+.3f}ms " + f"residual={residual/1e6:+.3f}ms " + f"raw={raw_offset_ns/1e6:+.3f}ms " + f"mpd={self._mean_path_delay_ns/1e6:.3f}ms " + f"freq={freq_ppb:+.1f}ppb", + logging.DEBUG, + ) + + def _estimate_freq_ppb(self) -> float: + """Slope of (offset_ns vs local_ns) over recent samples, returned as + parts-per-billion. Positive means master ticks faster than local. + """ + if len(self._offset_history) < 4: + return 0.0 + xs = [s for s, _ in self._offset_history] + ys = [o for _, o in self._offset_history] + n = len(xs) + mx = sum(xs) / n + my = sum(ys) / n + num = sum((xs[i] - mx) * (ys[i] - my) for i in range(n)) + den = sum((xs[i] - mx) ** 2 for i in range(n)) + if den == 0: + return 0.0 + # d(offset)/d(t_local) is dimensionless; convert to ppb. + slope = num / den + # Clamp to sanity ±500 ppm + if slope > 5e-4: + slope = 5e-4 + if slope < -5e-4: + slope = -5e-4 + return slope * 1e9 + + +def make_shared_array(): + """Create a fresh shared-state Array for a PTPDisciplinedClock. + + Created externally so the same array can be passed to both the PTP slave + process and the audio playback process(es). + """ + array = multiprocessing.Array("d", 8, lock=False) + for i in range(8): + array[i] = 0.0 + return array + + +def _slave_entry(arr, ips, cid, debug, log_path): + """Module-level entry point for the PTP slave child process. + + Must be top-level (not a closure) so Windows multiprocessing with the + 'spawn' start method can pickle it. Earlier versions had this as a + nested function and silently broke on Windows. + """ + slave = PTPSlave( + master_ips=ips, + master_clock_identity=cid, + shared_clock_array=arr, + is_debug=debug, + log_path=log_path, + ) + slave.run() + + +def spawn( + master_ips: list[str], + master_clock_identity: bytes | None = None, + is_debug: bool = False, + log_path: str | None = None, + shared_array=None, +) -> tuple[multiprocessing.Process, PTPDisciplinedClock]: + """Spawn a PTP slave process for the given master IPs. + + Returns (process, disciplined_clock). The clock object can be passed to + any other process (it wraps a shared multiprocessing.Array). On master + change, terminate this process and spawn a new one. Pass ``shared_array`` + to reuse an existing array (the audio processes already hold a view). + """ + if shared_array is None: + array = make_shared_array() + else: + array = shared_array + + proc = multiprocessing.Process( + target=_slave_entry, + args=(array, master_ips, master_clock_identity, is_debug, log_path), + daemon=True, + ) + proc.start() + return proc, PTPDisciplinedClock(array) diff --git a/ap2/connections/ptp_clock.py b/ap2/connections/ptp_clock.py new file mode 100644 index 0000000..4beadee --- /dev/null +++ b/ap2/connections/ptp_clock.py @@ -0,0 +1,140 @@ +""" +PTP-disciplined clock state, shared across processes. + +The PTP slave process writes (offset_ns, freq_ppb, mean_path_delay_ns, status) +into a multiprocessing.Array; the audio process reads it on a hot path and +converts a local perf_counter_ns() reading into a "master time" estimate. + +Why perf_counter and not monotonic_ns: + Windows time.monotonic_ns() resolution is ~15.6 ms — too coarse for PTP. + time.perf_counter_ns() uses QueryPerformanceCounter on Windows + (~100 ns resolution) and CLOCK_MONOTONIC_RAW elsewhere. We use it both + here and inside the slave's tx/rx timestamping path so the same epoch + is shared. +""" + +import multiprocessing +import struct +import time + + +# Layout of the shared state array (8 doubles, packed for atomic-ish reads): +# [0] offset_ns — master_time - local_perf_counter at sync_local_ns +# [1] freq_ppb — frequency adjustment (master ticks per local tick - 1) * 1e9 +# [2] sync_local_ns — local perf_counter at which (offset_ns, freq_ppb) were valid +# [3] mean_path_delay_ns +# [4] last_update_local_ns — perf_counter at last update (for staleness check) +# [5] status — 0=unsynced, 1=acquiring, 2=synced +# [6] sync_count — incremented each successful update +# [7] reserved +_SLOT_COUNT = 8 + + +def now_local_ns() -> int: + return time.perf_counter_ns() + + +class PTPDisciplinedClock: + """Shared-state wrapper around the PTP slave's current discipline estimate. + + Instances of this class are cheap to construct in any process: they all + wrap the same multiprocessing.Array. The slave process holds one as the + writer; consumers (audio playback) hold one as readers. + + Reads do NOT take the lock: they return possibly-torn views. We tolerate + that because: (a) updates land at most once per ~1s; (b) we re-poll + frequently; (c) torn reads can only briefly produce a sub-ms outlier + which the playback path already smooths over. + """ + + STATUS_UNSYNCED = 0 + STATUS_ACQUIRING = 1 + STATUS_SYNCED = 2 + + def __init__(self, shared_array=None): + if shared_array is None: + shared_array = multiprocessing.Array("d", _SLOT_COUNT, lock=False) + shared_array[0] = 0.0 + shared_array[1] = 0.0 + shared_array[2] = 0.0 + shared_array[3] = 0.0 + shared_array[4] = 0.0 + shared_array[5] = float(self.STATUS_UNSYNCED) + shared_array[6] = 0.0 + shared_array[7] = 0.0 + self._arr = shared_array + + @property + def shared_array(self): + return self._arr + + # -------- writer side (called by the PTP slave) -------- + + def update( + self, + offset_ns: int, + freq_ppb: float, + mean_path_delay_ns: int, + sync_local_ns: int, + status: int, + ) -> None: + # Order matters slightly for readers: bump sync_count last so a + # reader that sees a fresh sync_count sees fresh fields. + self._arr[0] = float(offset_ns) + self._arr[1] = float(freq_ppb) + self._arr[2] = float(sync_local_ns) + self._arr[3] = float(mean_path_delay_ns) + self._arr[4] = float(now_local_ns()) + self._arr[5] = float(status) + self._arr[6] = self._arr[6] + 1.0 + + def set_status(self, status: int) -> None: + self._arr[5] = float(status) + self._arr[4] = float(now_local_ns()) + + # -------- reader side (called by audio path) -------- + + def now_master_ns(self, local_ns: int | None = None) -> int: + """Convert a local perf_counter_ns reading to estimated master time. + + master_time(t) = offset + (t - sync_local) * (1 + freq_ppb*1e-9) + sync_local + = t + offset + (t - sync_local) * freq_ppb*1e-9 + + When unsynced, returns the local time unchanged. Caller can check + get_status() to decide whether to wait. + """ + if local_ns is None: + local_ns = now_local_ns() + # Use the explicit status field — a legitimate sync can land at + # offset==0 and freq_ppb==0 (loopback self-test, or two clocks + # genuinely in agreement) and we'd still want to apply discipline + # (here a no-op, but the status check correctly says "we're synced"). + if int(self._arr[5]) < self.STATUS_SYNCED: + return local_ns + offset = self._arr[0] + freq_ppb = self._arr[1] + sync_local = self._arr[2] + elapsed = local_ns - sync_local + drift = elapsed * freq_ppb * 1e-9 + return int(local_ns + offset + drift) + + def get_offset_ns(self) -> int: + return int(self._arr[0]) + + def get_mean_path_delay_ns(self) -> int: + return int(self._arr[3]) + + def get_status(self) -> int: + return int(self._arr[5]) + + def is_synced(self) -> bool: + return self.get_status() == self.STATUS_SYNCED + + def staleness_ns(self) -> int: + last = self._arr[4] + if last == 0.0: + return -1 + return now_local_ns() - int(last) + + def get_sync_count(self) -> int: + return int(self._arr[6]) diff --git a/ap2/connections/ptp_messages.py b/ap2/connections/ptp_messages.py new file mode 100644 index 0000000..b967ce2 --- /dev/null +++ b/ap2/connections/ptp_messages.py @@ -0,0 +1,297 @@ +""" +PTPv2 (IEEE 1588-2008) message codec. + +Implements the subset of message types needed for an ordinary clock running in +SLAVE state on the AirPlay 2 timing path: Announce, Sync, Follow_Up, Delay_Req, +Delay_Resp. Other PTP message types (Pdelay_*, Signaling, Management) are +parsed only to the extent of recognising their type so they can be ignored. + +All wire fields are network byte order (big endian). Timestamps are encoded as +seconds_msb(2) || seconds_lsb(4) || nanoseconds(4) = 10 bytes, holding seconds +since the PTP epoch (1970-01-01) — but for AirPlay slaves we treat them as +opaque master-clock values, since the sender's grand-master typically +advertises its own monotonic uptime, not real wall time. +""" + +import enum +import struct +from dataclasses import dataclass, field + + +PTP_HEADER_LEN = 34 +PTP_TS_LEN = 10 +PTP_CLOCK_IDENTITY_LEN = 8 +PTP_PORT_IDENTITY_LEN = 10 + +PTP_EVENT_PORT = 319 +PTP_GENERAL_PORT = 320 + + +class MsgType(enum.IntEnum): + SYNC = 0x0 + DELAY_REQ = 0x1 + PDELAY_REQ = 0x2 + PDELAY_RESP = 0x3 + FOLLOW_UP = 0x8 + DELAY_RESP = 0x9 + PDELAY_RESP_FOLLOW_UP = 0xA + ANNOUNCE = 0xB + SIGNALING = 0xC + MANAGEMENT = 0xD + + +class FlagBit(enum.IntFlag): + """Subset of PTP header flagField bits relevant to a slave.""" + TWO_STEP = 0x0200 # byte 6 bit 1 + UNICAST = 0x0400 # byte 6 bit 2 + PROFILE_SPECIFIC_1 = 0x2000 + PROFILE_SPECIFIC_2 = 0x4000 + ALTERNATE_MASTER = 0x0100 + + +@dataclass +class PortIdentity: + clock_identity: bytes # 8 bytes + port_number: int # uint16 + + def pack(self) -> bytes: + assert len(self.clock_identity) == PTP_CLOCK_IDENTITY_LEN + return self.clock_identity + struct.pack("!H", self.port_number) + + @classmethod + def unpack(cls, buf: bytes) -> "PortIdentity": + assert len(buf) >= PTP_PORT_IDENTITY_LEN + return cls( + clock_identity=bytes(buf[0:8]), + port_number=struct.unpack("!H", buf[8:10])[0], + ) + + def __hash__(self): + return hash((self.clock_identity, self.port_number)) + + +def pack_timestamp(ns: int) -> bytes: + """Encode an integer nanosecond value as a 10-byte PTP timestamp. + + seconds_msb (16b) || seconds_lsb (32b) || nanoseconds (32b) + """ + secs, nsecs = divmod(ns, 1_000_000_000) + secs_msb = (secs >> 32) & 0xFFFF + secs_lsb = secs & 0xFFFFFFFF + return struct.pack("!HII", secs_msb, secs_lsb, nsecs) + + +def unpack_timestamp(buf: bytes) -> int: + """Decode a 10-byte PTP timestamp to integer nanoseconds.""" + assert len(buf) >= PTP_TS_LEN + secs_msb, secs_lsb, nsecs = struct.unpack("!HII", buf[0:10]) + secs = (secs_msb << 32) | secs_lsb + return secs * 1_000_000_000 + nsecs + + +@dataclass +class PtpHeader: + """Common 34-byte PTPv2 header.""" + message_type: int # 4 bits (we keep low nibble) + transport_specific: int = 0 # 4 bits + version: int = 2 # 4 bits + message_length: int = 0 + domain_number: int = 0 + flags: int = 0 # uint16 + correction_ns: int = 0 # int64 carrying ns << 16 (sub-ns ignored) + source_port_identity: PortIdentity = field( + default_factory=lambda: PortIdentity(b"\x00" * 8, 0) + ) + sequence_id: int = 0 + control_field: int = 0 + log_message_interval: int = 0 # int8 + + @classmethod + def unpack(cls, buf: bytes) -> "PtpHeader": + if len(buf) < PTP_HEADER_LEN: + raise ValueError(f"ptp header too short: {len(buf)}") + b0 = buf[0] + b1 = buf[1] + msg_type = b0 & 0x0F + transport_specific = (b0 >> 4) & 0x0F + version = b1 & 0x0F + message_length = struct.unpack("!H", buf[2:4])[0] + domain_number = buf[4] + flags = struct.unpack("!H", buf[6:8])[0] + # correctionField: 64-bit signed, scaled ns (low 16 bits = sub-ns) + correction_scaled = struct.unpack("!q", buf[8:16])[0] + correction_ns = correction_scaled >> 16 + # reserved 4 bytes at 16:20 + source_port_identity = PortIdentity.unpack(buf[20:30]) + sequence_id = struct.unpack("!H", buf[30:32])[0] + control_field = buf[32] + log_message_interval = struct.unpack("!b", buf[33:34])[0] + return cls( + message_type=msg_type, + transport_specific=transport_specific, + version=version, + message_length=message_length, + domain_number=domain_number, + flags=flags, + correction_ns=correction_ns, + source_port_identity=source_port_identity, + sequence_id=sequence_id, + control_field=control_field, + log_message_interval=log_message_interval, + ) + + +def pack_header_bytes(h: PtpHeader) -> bytes: + """Pack header to its exact 34-byte wire form.""" + b0 = ((h.transport_specific & 0x0F) << 4) | (h.message_type & 0x0F) + b1 = h.version & 0x0F + # correctionField wire encoding: signed int64 in units of (2^-16) ns + correction_scaled = int(h.correction_ns) * 65536 + if correction_scaled > (1 << 63) - 1: + correction_scaled = (1 << 63) - 1 + elif correction_scaled < -(1 << 63): + correction_scaled = -(1 << 63) + out = bytearray(PTP_HEADER_LEN) + out[0] = b0 + out[1] = b1 + struct.pack_into("!H", out, 2, h.message_length) + out[4] = h.domain_number + out[5] = 0 + struct.pack_into("!H", out, 6, h.flags & 0xFFFF) + struct.pack_into("!q", out, 8, correction_scaled) + # 16:20 reserved zeros + out[20:30] = h.source_port_identity.pack() + struct.pack_into("!H", out, 30, h.sequence_id & 0xFFFF) + out[32] = h.control_field & 0xFF + struct.pack_into("!b", out, 33, h.log_message_interval) + return bytes(out) + + +@dataclass +class AnnounceBody: + origin_timestamp_ns: int + current_utc_offset: int + grandmaster_priority1: int + grandmaster_clock_quality: bytes # 4 bytes raw + grandmaster_priority2: int + grandmaster_identity: bytes # 8 bytes + steps_removed: int + time_source: int + + @classmethod + def unpack(cls, buf: bytes) -> "AnnounceBody": + if len(buf) < 30: + raise ValueError("announce body too short") + ts_ns = unpack_timestamp(buf[0:10]) + utc_offset = struct.unpack("!h", buf[10:12])[0] + # reserved byte 12 + prio1 = buf[13] + clock_quality = bytes(buf[14:18]) + prio2 = buf[18] + gm_identity = bytes(buf[19:27]) + steps_removed = struct.unpack("!H", buf[27:29])[0] + time_source = buf[29] + return cls( + origin_timestamp_ns=ts_ns, + current_utc_offset=utc_offset, + grandmaster_priority1=prio1, + grandmaster_clock_quality=clock_quality, + grandmaster_priority2=prio2, + grandmaster_identity=gm_identity, + steps_removed=steps_removed, + time_source=time_source, + ) + + +@dataclass +class SyncBody: + origin_timestamp_ns: int + + @classmethod + def unpack(cls, buf: bytes) -> "SyncBody": + return cls(origin_timestamp_ns=unpack_timestamp(buf[0:10])) + + +@dataclass +class FollowUpBody: + precise_origin_timestamp_ns: int + + @classmethod + def unpack(cls, buf: bytes) -> "FollowUpBody": + return cls(precise_origin_timestamp_ns=unpack_timestamp(buf[0:10])) + + +@dataclass +class DelayReqBody: + origin_timestamp_ns: int = 0 + + def pack(self) -> bytes: + return pack_timestamp(self.origin_timestamp_ns) + + +@dataclass +class DelayRespBody: + receive_timestamp_ns: int + requesting_port_identity: PortIdentity + + @classmethod + def unpack(cls, buf: bytes) -> "DelayRespBody": + rx_ts = unpack_timestamp(buf[0:10]) + req_port = PortIdentity.unpack(buf[10:20]) + return cls(receive_timestamp_ns=rx_ts, requesting_port_identity=req_port) + + +@dataclass +class PtpMessage: + """A parsed PTP message: header plus its decoded body (or None).""" + header: PtpHeader + body: object # one of AnnounceBody | SyncBody | FollowUpBody | DelayRespBody | None + raw: bytes + + @property + def msg_type(self) -> int: + return self.header.message_type + + +def parse(buf: bytes) -> PtpMessage: + """Parse one PTP datagram. Unknown types parsed to header only.""" + header = PtpHeader.unpack(buf) + payload = buf[PTP_HEADER_LEN:] + body = None + mt = header.message_type + try: + if mt == MsgType.SYNC: + body = SyncBody.unpack(payload) + elif mt == MsgType.FOLLOW_UP: + body = FollowUpBody.unpack(payload) + elif mt == MsgType.ANNOUNCE: + body = AnnounceBody.unpack(payload) + elif mt == MsgType.DELAY_RESP: + body = DelayRespBody.unpack(payload) + except (struct.error, ValueError): + body = None + return PtpMessage(header=header, body=body, raw=buf) + + +def build_delay_req( + src_port_identity: PortIdentity, + sequence_id: int, + domain_number: int = 0, +) -> bytes: + """Build a Delay_Req message. originTimestamp is zeroed; the receiver-side + t1 timestamp is captured locally at send time and matched to the + Delay_Resp by sequenceId.""" + header = PtpHeader( + message_type=MsgType.DELAY_REQ, + transport_specific=0, + version=2, + message_length=PTP_HEADER_LEN + PTP_TS_LEN, + domain_number=domain_number, + flags=0, + correction_ns=0, + source_port_identity=src_port_identity, + sequence_id=sequence_id & 0xFFFF, + control_field=0x01, # legacy "Delay_Req" + log_message_interval=0x7F, # 0x7F == "unspecified" + ) + return pack_header_bytes(header) + DelayReqBody().pack() diff --git a/ap2/connections/stream.py b/ap2/connections/stream.py index 725b978..ddb1ba9 100644 --- a/ap2/connections/stream.py +++ b/ap2/connections/stream.py @@ -12,7 +12,7 @@ class Stream: REALTIME = 96 BUFFERED = 103 - def __init__(self, stream, addr, port=0, buff_size=0, stream_id=None, shared_key=None, isDebug=False, aud_params=None): + def __init__(self, stream, addr, port=0, buff_size=0, stream_id=None, shared_key=None, isDebug=False, aud_params=None, ptp_clock_array=None): # self.audioMode = stream["audioMode"] # default|moviePlayback self.isDebug = isDebug self.addr = addr @@ -28,6 +28,7 @@ def __init__(self, stream, addr, port=0, buff_size=0, stream_id=None, shared_key self.control_proc = None self.shared_key = shared_key + self.ptp_clock_array = ptp_clock_array self.culled = False """stat fields at teardown ccCountAPSender @@ -110,6 +111,7 @@ def __init__(self, stream, addr, port=0, buff_size=0, stream_id=None, shared_key control_conns=self.control_conns, isDebug=self.isDebug, aud_params=None, + ptp_clock_array=self.ptp_clock_array, ) self.descriptor = { 'type': self.streamtype, @@ -133,6 +135,7 @@ def __init__(self, stream, addr, port=0, buff_size=0, stream_id=None, shared_key control_conns=self.control_conns, isDebug=self.isDebug, aud_params=None, + ptp_clock_array=self.ptp_clock_array, ) self.descriptor = { 'type': self.streamtype, diff --git a/ptp_selftest.py b/ptp_selftest.py new file mode 100644 index 0000000..62e64b8 --- /dev/null +++ b/ptp_selftest.py @@ -0,0 +1,321 @@ +""" +Mock PTP master + slave end-to-end self-test. + +What it does: + * Spawns the PTP slave from ap2.connections.ptp on UDP 319/320 of localhost. + * In the main process, drives a minimal "master" that sends Announce/Sync/ + Follow_Up to the slave and answers Delay_Req with Delay_Resp, injecting a + known clock offset. + * After a few seconds, reads the disciplined clock and asserts the offset + converged to within tolerance. + +Limitations: + * Both sides share localhost so path delay is ~µs. The test verifies the + convergence math, not actual network performance. + * The slave binds to ports 319/320, so this script will fail if another + PTP daemon is listening (e.g. Windows Time, ptp4l). + +Usage: + python ptp_selftest.py +""" + +import os +import random +import socket +import struct +import sys +import time + +from ap2.connections import ptp as ptp_mod +from ap2.connections import ptp_messages as ptpm +from ap2.connections.ptp_clock import PTPDisciplinedClock + + +def now_local_ns() -> int: + """Shared Unix epoch in ns (the master uses this; the slave is patched to + use the same so the loopback self-test has a deterministic baseline).""" + return time.time_ns() + + +SLAVE_EVENT_PORT = ptpm.PTP_EVENT_PORT +SLAVE_GENERAL_PORT = ptpm.PTP_GENERAL_PORT +MASTER_EVENT_PORT = 8319 # mock master can use any free ports — but slave +MASTER_GENERAL_PORT = 8320 # always sends Delay_Req to *port 319* of master IP. +# So for this self-test we cheat: we use loopback and rebind the slave's +# Delay_Req destination by temporarily overriding ptpm.PTP_EVENT_PORT? No — +# instead we have the master listen on 319-on-loopback before the slave does. +# That conflicts with the slave's own 319 bind. The only clean fix is to +# run the slave with custom ports. Add a small monkey-patch hook below. + +INJECTED_OFFSET_NS = 250_000_000 # 250 ms: master is 250 ms ahead of local + + +def _pack_announce_body(origin_ts_ns: int) -> bytes: + body = bytearray(30) + body[0:10] = ptpm.pack_timestamp(origin_ts_ns) + # currentUtcOffset(2) + reserved(1) + grandmasterPriority1(1) ... + struct.pack_into("!h", body, 10, 0) # utc offset + body[12] = 0 # reserved + body[13] = 128 # priority1 + body[14:18] = b"\x60\x00\x00\x00" # clockQuality + body[18] = 128 # priority2 + body[19:27] = b"\xaa" * 8 # grandmasterIdentity + struct.pack_into("!H", body, 27, 0) # stepsRemoved + body[29] = 0xA0 # timeSource (internal osc) + return bytes(body) + + +def _pack_sync_or_follow_up(precise_origin_ts_ns: int) -> bytes: + return ptpm.pack_timestamp(precise_origin_ts_ns) + + +def _pack_delay_resp(receive_ts_ns: int, requesting_port_id: bytes) -> bytes: + return ptpm.pack_timestamp(receive_ts_ns) + requesting_port_id + + +def _make_header( + msg_type: int, + seq: int, + *, + flags: int = 0, + message_length: int, + control_field: int = 0, + log_msg_interval: int = 0, + source_port_identity: bytes, +) -> bytes: + src_pi = ptpm.PortIdentity(source_port_identity[:8], int.from_bytes(source_port_identity[8:10], "big")) + h = ptpm.PtpHeader( + message_type=msg_type, + version=2, + message_length=message_length, + flags=flags, + source_port_identity=src_pi, + sequence_id=seq, + control_field=control_field, + log_message_interval=log_msg_interval, + ) + return ptpm.pack_header_bytes(h) + + +def master_time_ns() -> int: + return now_local_ns() + INJECTED_OFFSET_NS + + +def run_mock_master(stop_at: float): + """Drive Announce + Sync(two-step) + Follow_Up to the slave at loopback. + + Listens for Delay_Req on UDP/319 (we cannot, the slave is there). So this + test path patches the slave to use different ports — see __main__. + """ + raise NotImplementedError("see __main__: patched ports are required for loopback test") + + +SLAVE_EVENT = 19319 +SLAVE_GENERAL = 19320 +MASTER_EVENT = 28319 +MASTER_GENERAL = 28320 + + +def slave_entry(arr, master_ip, slave_event_port, slave_general_port, master_event_port_target): + """Top-level so it's picklable under Windows spawn.""" + import time as _t + from ap2.connections import ptp as p + from ap2.connections import ptp_messages as m + from ap2.connections import ptp_clock as pc + m.PTP_EVENT_PORT = slave_event_port + m.PTP_GENERAL_PORT = slave_general_port + p.PTP_EVENT_PORT = slave_event_port + p.PTP_GENERAL_PORT = slave_general_port + + # For the loopback self-test only: use time.time_ns (shared Unix epoch) + # in the slave so the slave's t2/t3 sit in the SAME epoch as the master's + # injected times. In production, perf_counter is correct: it stays in + # the receiver's own monotonic frame and the PTP offset captures whatever + # constant epoch difference exists between sender and receiver. + pc.now_local_ns = _t.time_ns + p.now_local_ns = _t.time_ns + + def _patched(self, port): + ip = self.master_ips[0] if self.master_ips else "127.0.0.1" + return (ip, master_event_port_target) + p.PTPSlave._master_addr_for = _patched + + slave = p.PTPSlave( + master_ips=[master_ip], + master_clock_identity=None, + shared_clock_array=arr, + is_debug=True, + ) + slave.run() + + +def main(): + import ap2.connections.ptp_messages as ptpm_mod + import ap2.connections.ptp as ptp_slave_mod + + ptpm_mod.PTP_EVENT_PORT = SLAVE_EVENT + ptpm_mod.PTP_GENERAL_PORT = SLAVE_GENERAL + ptp_slave_mod.PTP_EVENT_PORT = SLAVE_EVENT + ptp_slave_mod.PTP_GENERAL_PORT = SLAVE_GENERAL + + # Bind master sockets first + m_event = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + m_event.bind(("127.0.0.1", MASTER_EVENT)) + m_event.settimeout(0.2) + m_general = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + m_general.bind(("127.0.0.1", MASTER_GENERAL)) + m_general.settimeout(0.2) + + master_port_id = b"\xaa" * 8 + b"\x00\x01" + + # Patch slave to send Delay_Req to the MASTER_EVENT port, not 319. + # Do this by monkey-patching _master_addr_for in the slave subclass… + # simplest: edit the constant the slave uses to construct dest. + # The slave uses self._master_addr_for(PTP_EVENT_PORT). Override + # PTP_EVENT_PORT in slave module already done above; but spawn() runs the + # slave in a *child process*, so reload constants there too via the + # spawned target. We do this by writing a small entry instead. + import multiprocessing + array = ptp_slave_mod.make_shared_array() + + proc = multiprocessing.Process( + target=slave_entry, + args=(array, "127.0.0.1", SLAVE_EVENT, SLAVE_GENERAL, MASTER_EVENT), + daemon=True, + ) + proc.start() + print(f"[selftest] slave pid={proc.pid}; injected master offset = {INJECTED_OFFSET_NS/1e6:.1f}ms") + # Give the slave a moment to bind its sockets so our initial UDP sends + # don't trigger Windows ICMP-unreachable replies (WinError 10054 on the + # next recvfrom on the same socket). + time.sleep(0.5) + + clock = PTPDisciplinedClock(array) + seq_sync = 0 + last_announce = 0.0 + last_sync = 0.0 + deadline = time.monotonic() + 10.0 + converged = False + applied_history = [] + + try: + while time.monotonic() < deadline: + t = time.monotonic() + + # 1Hz Announce + if t - last_announce > 1.0: + last_announce = t + body = _pack_announce_body(master_time_ns()) + hdr = _make_header( + msg_type=ptpm.MsgType.ANNOUNCE, seq=int(t) & 0xFFFF, message_length=34 + len(body), + source_port_identity=master_port_id, + ) + m_general.sendto(hdr + body, ("127.0.0.1", SLAVE_GENERAL)) + + # 4Hz Sync + Follow_Up (two-step) + if t - last_sync > 0.25: + last_sync = t + seq_sync = (seq_sync + 1) & 0xFFFF + sync_body = _pack_sync_or_follow_up(0) # one-step would put TS here + sync_hdr = _make_header( + msg_type=ptpm.MsgType.SYNC, seq=seq_sync, + flags=int(ptpm.FlagBit.TWO_STEP), + message_length=34 + len(sync_body), + log_msg_interval=-2, + source_port_identity=master_port_id, + ) + # Capture t1 at the *exact* moment we hand the Sync to the OS; + # that's what the two-step Follow_Up must echo back, otherwise + # the slave's mpd estimate inflates by the construct-delay. + t1_master = master_time_ns() + m_event.sendto(sync_hdr + sync_body, ("127.0.0.1", SLAVE_EVENT)) + fu_body = _pack_sync_or_follow_up(t1_master) + fu_hdr = _make_header( + msg_type=ptpm.MsgType.FOLLOW_UP, seq=seq_sync, + message_length=34 + len(fu_body), + log_msg_interval=-2, + source_port_identity=master_port_id, + ) + m_general.sendto(fu_hdr + fu_body, ("127.0.0.1", SLAVE_GENERAL)) + + # Service Delay_Req on the master event socket + try: + while True: + try: + data, addr = m_event.recvfrom(2048) + except ConnectionResetError: + # Windows: prior send hit a closed port; ignore and retry. + continue + msg = ptpm.parse(data) + if msg.msg_type == ptpm.MsgType.DELAY_REQ: + t4 = master_time_ns() + req_port = msg.header.source_port_identity.pack() + resp_body = _pack_delay_resp(t4, req_port) + resp_hdr = _make_header( + msg_type=ptpm.MsgType.DELAY_RESP, seq=msg.header.sequence_id, + message_length=34 + len(resp_body), + control_field=0x03, + source_port_identity=master_port_id, + ) + # Delay_Resp goes on the general port back to wherever the + # slave is listening. Slave's general socket = 19320. + m_general.sendto(resp_hdr + resp_body, ("127.0.0.1", SLAVE_GENERAL)) + except socket.timeout: + pass + + # Track stability of the applied offset over a 1-second window. + # On Windows loopback, UDP scheduling jitter inflates mpd, so the + # applied offset doesn't equal the injected 250 ms exactly — it + # equals 250 ms + half of (forward - reverse) scheduling + # asymmetry. What we *can* assert is: slave reaches SYNCED, the + # applied offset stops drifting, and mpd stabilizes. + applied = clock.get_offset_ns() + mpd = clock.get_mean_path_delay_ns() + if clock.is_synced(): + applied_history.append((time.monotonic(), applied, mpd)) + # Trim to last 2s + applied_history = [(t_, a_, m_) for (t_, a_, m_) in applied_history if t_ > time.monotonic() - 2.0] + if len(applied_history) >= 8: + applies = [a_ for (_, a_, _) in applied_history] + mpds = [m_ for (_, _, m_) in applied_history] + applied_span = max(applies) - min(applies) + mpd_span = max(mpds) - min(mpds) + print( + f"[selftest] synced; applied={applied/1e6:+.3f}ms " + f"mpd={mpd/1e6:.3f}ms span(applied)={applied_span/1e6:.3f}ms " + f"span(mpd)={mpd_span/1e6:.3f}ms" + ) + # Stability criteria: applied moves <10ms over 2s and mpd + # moves <10ms; both indicate the servo has settled. + if applied_span < 10_000_000 and mpd_span < 10_000_000: + converged = True + break + else: + print(f"[selftest] synced; applied={applied/1e6:+.3f}ms mpd={mpd/1e6:.3f}ms (warmup)") + time.sleep(0.05) + + finally: + proc.terminate() + proc.join(timeout=1.0) + m_event.close() + m_general.close() + + if converged: + print( + f"[selftest] PASS — slave reached SYNCED and stabilized. " + f"applied={clock.get_offset_ns()/1e6:+.3f}ms mpd={clock.get_mean_path_delay_ns()/1e6:.3f}ms. " + f"Note: the applied value reflects 250ms injected + Windows UDP " + f"scheduling asymmetry — that's expected on loopback." + ) + sys.exit(0) + else: + print( + f"[selftest] FAIL — slave did not stabilize. " + f"applied={clock.get_offset_ns()/1e6:+.3f}ms mpd={clock.get_mean_path_delay_ns()/1e6:.3f}ms " + f"is_synced={clock.is_synced()}" + ) + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt index cc82035..d0815bf 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,4 +8,9 @@ hkdf # pynacl cryptography requests -av==8.1.0 +# av 8.1.0 fails to build on Python >= 3.10 (distutils.msvccompiler removed); +# any modern PyAV (>=10) works — the codebase has been updated to use the +# new packed/planar handling via to_ndarray(). +av +numpy +pycaw; sys_platform == "win32"