diff --git a/src/syncfield/adapters/meta_quest.py b/src/syncfield/adapters/meta_quest.py index 7bb26ec..8dd1729 100644 --- a/src/syncfield/adapters/meta_quest.py +++ b/src/syncfield/adapters/meta_quest.py @@ -119,6 +119,12 @@ class MetaQuestHandStream(StreamBase): BUFFER_SIZE = 65536 CONNECTION_TIMEOUT_S = 2.0 + # Samples arrive via WiFi UDP from a remote Quest device, so they do + # not share the host monotonic clock domain. Tag them separately so + # downstream sync tooling can account for the wireless jitter. + CLOCK_DOMAIN = "remote_quest3" + UNCERTAINTY_NS = 10_000_000 # 10 ms — typical WiFi jitter budget + _discovery_kind = "sensor" _discovery_adapter_type = "meta_quest" @@ -152,7 +158,11 @@ def __init__( self._first_at: Optional[int] = None self._last_at: Optional[int] = None self._consecutive_errors = 0 - self._last_packet_mono = 0.0 + self._last_packet_mono: float = 0.0 + # "waiting" = socket is up, no packet has ever been received yet + # "connected" = a packet arrived within CONNECTION_TIMEOUT_S + # "lost" = had packets, none for > CONNECTION_TIMEOUT_S (DROP emitted) + self._connection_state: str = "waiting" # ------------------------------------------------------------------ # 4-phase lifecycle @@ -168,6 +178,8 @@ def connect(self) -> None: self._first_at = None self._last_at = None self._consecutive_errors = 0 + self._last_packet_mono = 0.0 + self._connection_state = "waiting" self._create_socket() self._receive_thread = threading.Thread( @@ -248,6 +260,22 @@ def _create_socket(self) -> None: self._socket.bind((self._host, self._port)) self._socket.settimeout(1.0) + @property + def is_connected(self) -> bool: + """True when a packet has arrived within ``CONNECTION_TIMEOUT_S``. + + Mirrors the recorder's ``is_connected`` semantics: the socket + being open is not enough — the Quest must actually be streaming. + Callers can poll this from any thread; it reads a float that + CPython updates atomically. + """ + if self._last_packet_mono == 0.0: + return False + return ( + time.monotonic() - self._last_packet_mono + <= self.CONNECTION_TIMEOUT_S + ) + def _receive_loop(self) -> None: while not self._stop_event.is_set(): if self._socket is None: @@ -256,11 +284,29 @@ def _receive_loop(self) -> None: data, _ = self._socket.recvfrom(self.BUFFER_SIZE) self._process_packet(data) except TimeoutError: + # The 1-second recv timeout doubles as a watchdog tick — + # use it to notice when the Quest has stopped streaming. + self._check_connection_timeout() continue except Exception as exc: if not self._stop_event.is_set(): self._handle_socket_error(exc) + def _check_connection_timeout(self) -> None: + """Emit DROP if we had packets and now haven't seen one for > timeout.""" + if self._connection_state != "connected": + return + if self._last_packet_mono == 0.0: + return + silence_s = time.monotonic() - self._last_packet_mono + if silence_s > self.CONNECTION_TIMEOUT_S: + self._connection_state = "lost" + self._emit_health(HealthEvent( + self.id, HealthEventKind.DROP, + time.monotonic_ns(), + f"No packet for {silence_s:.1f}s (timeout {self.CONNECTION_TIMEOUT_S}s)", + )) + def _handle_socket_error(self, error: Exception) -> None: self._consecutive_errors += 1 if self._consecutive_errors >= self.MAX_CONSECUTIVE_ERRORS: @@ -292,6 +338,7 @@ def _process_packet(self, data: bytes) -> None: self._last_packet_mono = time.monotonic() self._consecutive_errors = 0 + self._update_connection_state_on_packet(capture_ns) # Build channels from packet channels = self._parse_channels(packet) @@ -307,6 +354,25 @@ def _process_packet(self, data: bytes) -> None: frame_number=frame_number, capture_ns=capture_ns, channels=channels, + uncertainty_ns=self.UNCERTAINTY_NS, + clock_domain=self.CLOCK_DOMAIN, + )) + + def _update_connection_state_on_packet(self, at_ns: int) -> None: + """Emit HEARTBEAT on first packet, RECONNECT after a drop.""" + prev = self._connection_state + if prev == "connected": + return + self._connection_state = "connected" + if prev == "waiting": + self._emit_health(HealthEvent( + self.id, HealthEventKind.HEARTBEAT, + at_ns, "First Quest 3 packet received", + )) + elif prev == "lost": + self._emit_health(HealthEvent( + self.id, HealthEventKind.RECONNECT, + at_ns, "Quest 3 packet stream resumed", )) def _parse_channels(self, packet: Dict[str, Any]) -> Dict[str, Any]: diff --git a/src/syncfield/orchestrator.py b/src/syncfield/orchestrator.py index 7ab1457..59564d8 100644 --- a/src/syncfield/orchestrator.py +++ b/src/syncfield/orchestrator.py @@ -2094,6 +2094,7 @@ def _handle(event: SampleEvent) -> None: if not active[0]: return try: + clock_domain = event.clock_domain or host_id if isinstance(writer, SensorWriter): writer.write( SensorSample( @@ -2101,7 +2102,7 @@ def _handle(event: SampleEvent) -> None: capture_ns=event.capture_ns, channels=event.channels or {}, clock_source="host_monotonic", - clock_domain=host_id, + clock_domain=clock_domain, uncertainty_ns=event.uncertainty_ns, ) ) @@ -2111,7 +2112,7 @@ def _handle(event: SampleEvent) -> None: frame_number=event.frame_number, capture_ns=event.capture_ns, clock_source="host_monotonic", - clock_domain=host_id, + clock_domain=clock_domain, uncertainty_ns=event.uncertainty_ns, ) ) diff --git a/src/syncfield/types.py b/src/syncfield/types.py index bc562f1..c062e6f 100644 --- a/src/syncfield/types.py +++ b/src/syncfield/types.py @@ -257,13 +257,21 @@ def to_dict(self) -> dict[str, Any]: @dataclass(frozen=True) class SampleEvent: - """A stream reports a sample (timestamp + optional channels) to the orchestrator.""" + """A stream reports a sample (timestamp + optional channels) to the orchestrator. + + ``clock_domain`` lets an adapter override the default host clock domain + when the timestamp's *origin* is a remote device rather than the local + monotonic clock (e.g. a Meta Quest streaming poses over WiFi). Leaving + it ``None`` — the common case for on-host captures — makes the + orchestrator stamp the host's id so all local streams share one domain. + """ stream_id: str frame_number: int capture_ns: int channels: dict[str, "ChannelValue"] | None = None uncertainty_ns: int = 5_000_000 + clock_domain: str | None = None @dataclass diff --git a/tests/unit/adapters/test_meta_quest.py b/tests/unit/adapters/test_meta_quest.py index bd58cde..d9bc1fa 100644 --- a/tests/unit/adapters/test_meta_quest.py +++ b/tests/unit/adapters/test_meta_quest.py @@ -300,6 +300,116 @@ def test_samples_emitted_during_recording(self): assert len(event.channels["hand_joints"]) == 156 +# --------------------------------------------------------------------------- +# Clock metadata (clock_domain + uncertainty_ns) emitted on SampleEvent +# --------------------------------------------------------------------------- + + +class TestClockMetadata: + def test_sample_event_has_remote_quest_clock_domain(self): + from syncfield.clock import SessionClock, SyncPoint + + port = _find_free_port() + stream = MetaQuestHandStream("test", port=port) + received = [] + stream.on_sample(lambda e: received.append(e)) + + stream.connect() + stream.start_recording(SessionClock(sync_point=SyncPoint.create_now("h"))) + _send_packet(port, _make_quest3_packet()) + time.sleep(0.2) + stream.stop_recording() + stream.disconnect() + + assert received, "expected at least one sample" + assert received[0].clock_domain == "remote_quest3" + assert received[0].uncertainty_ns == 10_000_000 + + +# --------------------------------------------------------------------------- +# Connection health — is_connected property + DROP/HEARTBEAT/RECONNECT +# --------------------------------------------------------------------------- + + +class TestConnectionHealth: + def test_is_connected_false_before_any_packet(self): + stream = MetaQuestHandStream("test", port=0) + assert stream.is_connected is False + + def test_is_connected_true_after_recent_packet(self): + port = _find_free_port() + stream = MetaQuestHandStream("test", port=port) + stream.connect() + try: + _send_packet(port, _make_quest3_packet()) + time.sleep(0.2) + assert stream.is_connected is True + finally: + stream.disconnect() + + def test_heartbeat_emitted_on_first_packet(self): + port = _find_free_port() + stream = MetaQuestHandStream("test", port=port) + events = [] + stream.on_health(lambda e: events.append(e)) + + stream.connect() + try: + _send_packet(port, _make_quest3_packet()) + time.sleep(0.2) + finally: + stream.disconnect() + + kinds = [e.kind.value for e in events] + assert "heartbeat" in kinds + + def test_drop_emitted_after_silence(self): + # Shorten timeout so the watchdog fires quickly in tests. + port = _find_free_port() + stream = MetaQuestHandStream("test", port=port) + stream.CONNECTION_TIMEOUT_S = 0.3 # override at instance level + events = [] + stream.on_health(lambda e: events.append(e)) + + stream.connect() + try: + # Mark the stream connected, then let the receive loop's + # 1-second socket timeout fire once with no further packets. + _send_packet(port, _make_quest3_packet()) + time.sleep(0.2) + assert stream.is_connected is True + # Wait long enough for silence to exceed CONNECTION_TIMEOUT_S + # and for at least one recv timeout tick (socket timeout = 1.0s). + time.sleep(1.5) + finally: + stream.disconnect() + + kinds = [e.kind.value for e in events] + assert "drop" in kinds, f"expected drop in {kinds}" + + def test_reconnect_emitted_after_drop_resumes(self): + port = _find_free_port() + stream = MetaQuestHandStream("test", port=port) + stream.CONNECTION_TIMEOUT_S = 0.3 + events = [] + stream.on_health(lambda e: events.append(e)) + + stream.connect() + try: + _send_packet(port, _make_quest3_packet()) + time.sleep(0.2) + time.sleep(1.5) # allow drop to fire + _send_packet(port, _make_quest3_packet()) + time.sleep(0.2) + finally: + stream.disconnect() + + kinds = [e.kind.value for e in events] + assert "drop" in kinds + assert "reconnect" in kinds + assert kinds.index("reconnect") > kinds.index("drop") + + def _find_free_port() -> int: with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s: s.bind(("127.0.0.1", 0))