diff --git a/examples/meta_quest/record.py b/examples/meta_quest/record.py index c089bb9..a07ba7d 100644 --- a/examples/meta_quest/record.py +++ b/examples/meta_quest/record.py @@ -46,7 +46,12 @@ import syncfield as sf import syncfield.viewer -from syncfield.adapters import MetaQuestCameraStream, MetaQuestHandStream +from syncfield.adapters import ( + BLEImuGenericStream, + MetaQuestCameraStream, + MetaQuestHandStream, +) +from syncfield.adapters.ble_imu_profiles import WIT_WT901BLE_200HZ # Quest 3 IPv4 — copy from the Quest sender app's HUD ("Host" line) or # check Settings → Wi-Fi → → Details → IP address. Must be on @@ -80,4 +85,32 @@ resolution=(1280, 720), )) +# Wrist IMUs — same two WT901BLE units used in examples/iphone_mac_webcam. +# Both advertise as "WT901BLE68", distinguished only by BLE address. Resolved +# via active-scan on 2026-04-14 in the iphone_mac_webcam example. +session.add(BLEImuGenericStream( + "wrist_left_imu", + profile=WIT_WT901BLE_200HZ, + address="5622CCC4-A621-96DC-A7B5-E7650370E8A3", +)) +session.add(BLEImuGenericStream( + "wrist_right_imu", + profile=WIT_WT901BLE_200HZ, + address="6E22ED0E-72CD-0175-6F29-0BA8D502CBAB", +)) + +# Elbow IMUs — two additional WT901BLE units. Resolved via active-scan on +# 2026-04-14. Left/right assignment is arbitrary; swap if the side labels +# don't match what's actually strapped on. +session.add(BLEImuGenericStream( + "elbow_left_imu", + profile=WIT_WT901BLE_200HZ, + address="1CD2DCDE-CE20-905E-7D66-66E20FB01AB6", +)) +session.add(BLEImuGenericStream( + "elbow_right_imu", + profile=WIT_WT901BLE_200HZ, + address="C7CA16B4-AFF6-CC54-C657-83836E96979A", +)) + syncfield.viewer.launch(session) diff --git a/src/syncfield/adapters/meta_quest_camera/__init__.py b/src/syncfield/adapters/meta_quest_camera/__init__.py index 579dbc1..353a16c 100644 --- a/src/syncfield/adapters/meta_quest_camera/__init__.py +++ b/src/syncfield/adapters/meta_quest_camera/__init__.py @@ -1,9 +1,15 @@ -"""Meta Quest 3 stereo passthrough camera adapter. +"""Meta Quest 3 stereo passthrough camera adapter (streaming-only). -Public entry point is :class:`MetaQuestCameraStream`. Internal collaborators -(``QuestHttpClient``, ``MjpegPreviewConsumer``, ``TimestampTailReader``, -``RecordingFilePuller``) live in sibling modules and are composed by the -stream class. +Public entry point is :class:`MetaQuestCameraStream`. Internal +collaborators live in sibling modules and are composed by the stream +class: + +* :mod:`.preview` — :class:`MjpegPreviewConsumer` that pulls one + ``/preview/{eye}`` MJPEG stream per eye. +* :mod:`.mp4_writer` — :class:`StreamingVideoRecorder` that mux-passes + each JPEG into an MP4 container without re-encoding. +* :mod:`.http_client` — :class:`QuestHttpClient` for the small set of + control-plane calls (``/status``, ``/tracker/target``). """ from syncfield.adapters.meta_quest_camera.stream import MetaQuestCameraStream diff --git a/src/syncfield/adapters/meta_quest_camera/file_puller.py b/src/syncfield/adapters/meta_quest_camera/file_puller.py deleted file mode 100644 index 4883414..0000000 --- a/src/syncfield/adapters/meta_quest_camera/file_puller.py +++ /dev/null @@ -1,59 +0,0 @@ -"""Pulls the four per-session artifacts (2 MP4s + 2 timestamps JSONLs) from -the Quest's HTTP surface into the SyncField session output directory.""" - -from __future__ import annotations - -from dataclasses import dataclass -from pathlib import Path - -from syncfield.adapters.meta_quest_camera.http_client import QuestHttpClient - - -@dataclass(frozen=True) -class RecordingArtifacts: - """Paths written to ``output_dir`` by a successful ``pull_all``.""" - - left_mp4: Path - right_mp4: Path - left_timestamps: Path - right_timestamps: Path - - -class RecordingFilePuller: - """Downloads all per-session artifacts into ``output_dir``. - - File naming mirrors the adapter's public contract: - - - ``{stream_id}_{side}.mp4`` - - ``{stream_id}_{side}.timestamps.jsonl`` - """ - - def __init__( - self, - *, - client: QuestHttpClient, - stream_id: str, - output_dir: Path, - ) -> None: - self._client = client - self._stream_id = stream_id - self._output_dir = Path(output_dir) - self._output_dir.mkdir(parents=True, exist_ok=True) - - def pull_all(self) -> RecordingArtifacts: - prefix = self._stream_id - paths = RecordingArtifacts( - left_mp4=self._output_dir / f"{prefix}_left.mp4", - right_mp4=self._output_dir / f"{prefix}_right.mp4", - left_timestamps=self._output_dir / f"{prefix}_left.timestamps.jsonl", - right_timestamps=self._output_dir / f"{prefix}_right.timestamps.jsonl", - ) - self._client.download_file("/recording/files/left", paths.left_mp4) - self._client.download_file("/recording/files/right", paths.right_mp4) - self._client.download_file( - "/recording/timestamps/left", paths.left_timestamps - ) - self._client.download_file( - "/recording/timestamps/right", paths.right_timestamps - ) - return paths diff --git a/src/syncfield/adapters/meta_quest_camera/mp4_writer.py b/src/syncfield/adapters/meta_quest_camera/mp4_writer.py new file mode 100644 index 0000000..51126ac --- /dev/null +++ b/src/syncfield/adapters/meta_quest_camera/mp4_writer.py @@ -0,0 +1,275 @@ +"""StreamingVideoRecorder — MJPEG-passthrough recorder for Quest streams. + +Takes raw JPEG bytes (as produced by Quest's ``/preview/{eye}`` endpoint) +and writes them directly into an MP4 container *without re-encoding*, +plus a sidecar timestamps JSONL with both the host-projected and the +Quest-native nanosecond timestamps for every frame. + +Designed to be fed by an :class:`~syncfield.adapters.meta_quest_camera.preview.MjpegPreviewConsumer`'s +frame-sink callback — the recorder doesn't own the network connection. +That keeps the same MJPEG channel serving both the live viewer panel +*and* the recording artifact at no extra Quest-side cost. + +Why MJPEG passthrough rather than re-encode to H.264: + +* **Zero CPU on the Mac.** No JPEG decode + H.264 encode round trip. +* **Bit-exact preservation of quality** — recorded frames are + byte-identical to what the Quest sender encoded, so quality is + controlled in one place (``previewJpegQuality`` on the device). +* **Variable framerate is honest.** PTS is derived from the real + capture timestamps, so jittered WiFi delivery shows up in the file + rather than being smoothed away to look like a flawless 30 fps. +""" + +from __future__ import annotations + +import json +import logging +import threading +from dataclasses import dataclass +from fractions import Fraction +from pathlib import Path +from typing import IO, Optional + +try: + import av # type: ignore[import-not-found] +except ImportError as exc: # pragma: no cover - exercised via sys.modules patch + raise ImportError( + "MetaQuestCameraStream requires PyAV. " + "Install with `pip install syncfield[viewer]` (or [oak], [uvc])." + ) from exc + + +logger = logging.getLogger(__name__) + + +# Microsecond time-base for the muxed MP4. Fine enough to represent +# Quest's 30 Hz captures without two adjacent PTS colliding (≥33 333 µs +# apart at 30 fps), and below the WiFi-jitter floor we're aiming for. +_MP4_TIME_BASE_DEN = 1_000_000 + + +@dataclass +class StreamingVideoResult: + """Returned by :meth:`StreamingVideoRecorder.stop`.""" + + output_path: Path + timestamps_path: Path + frame_count: int + first_capture_ns: Optional[int] + last_capture_ns: Optional[int] + # Frames the writer accepted but failed to mux. Surfaces silent + # disk / container errors that would otherwise vanish (each + # individual write swallows the exception so the recorder can + # keep going). + write_errors: int + + +class StreamingVideoRecorder: + """Writes a stream of JPEG frames into an MJPEG-in-MP4 container. + + Lifecycle:: + + recorder = StreamingVideoRecorder(output_dir=..., ...) + recorder.start() + # called repeatedly from the network / sink thread: + recorder.write_frame(jpeg, host_ns, quest_native_ns) + result = recorder.stop() + + ``write_frame`` may be called from any thread; an instance lock + serialises mux / JSONL / state mutations so :meth:`stop` is safe + to invoke while a sink is still pushing frames in. + + Calls to ``write_frame`` against an unstarted or already-stopped + recorder are silent no-ops, which simplifies the sink wiring on + the consumer side (it does not have to mirror the lifecycle). + """ + + def __init__( + self, + *, + output_dir: Path, + stream_id: str, + side: str, + fps: int, + width: int, + height: int, + ) -> None: + self._output_dir = Path(output_dir) + self._stream_id = stream_id + self._side = side + self._fps = max(1, int(fps)) + self._width = int(width) + self._height = int(height) + + self._lock = threading.Lock() + self._container: Optional["av.container.OutputContainer"] = None + self._stream: Optional["av.video.stream.VideoStream"] = None + self._timestamps_file: Optional[IO[str]] = None + + self._frame_count = 0 + self._write_errors = 0 + self._first_capture_ns: Optional[int] = None + self._last_capture_ns: Optional[int] = None + # PTS anchor — set on the first muxed frame so the file's PTS + # starts at 0 instead of an enormous absolute monotonic value. + self._first_pts_us: Optional[int] = None + + @property + def output_path(self) -> Path: + return self._output_dir / f"{self._stream_id}_{self._side}.mp4" + + @property + def timestamps_path(self) -> Path: + return self._output_dir / f"{self._stream_id}_{self._side}.timestamps.jsonl" + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + def start(self) -> None: + """Open the MP4 container and timestamps JSONL. + + No-op if the recorder is already started. After a :meth:`stop`, + a fresh ``start()`` reopens the files (overwriting whatever + was written previously). + """ + with self._lock: + if self._container is not None: + return + self._output_dir.mkdir(parents=True, exist_ok=True) + + container = av.open(str(self.output_path), mode="w") + try: + stream = container.add_stream("mjpeg", rate=self._fps) + stream.width = self._width + stream.height = self._height + # JPEG defaults to yuvj420p (full-range chroma); declaring + # this matches the muxer's expectation for MJPEG packets + # produced by Quest's Unity encoder. + stream.pix_fmt = "yuvj420p" + # Microsecond time base lets PTS reflect real capture + # times — variable inter-frame intervals from WiFi + # jitter are preserved instead of being snapped to a + # synthetic 30 fps grid. + stream.time_base = Fraction(1, _MP4_TIME_BASE_DEN) + except Exception: + container.close() + raise + + timestamps_file = open(self.timestamps_path, "w", encoding="utf-8") + self._container = container + self._stream = stream + self._timestamps_file = timestamps_file + self._frame_count = 0 + self._write_errors = 0 + self._first_capture_ns = None + self._last_capture_ns = None + self._first_pts_us = None + logger.info( + "[%s/%s] streaming recorder open → %s (%dx%d @ %d fps)", + self._stream_id, self._side, self.output_path, + self._width, self._height, self._fps, + ) + + def write_frame( + self, + jpeg_bytes: bytes, + host_ns: int, + quest_native_ns: Optional[int] = None, + ) -> None: + """Mux one JPEG packet and emit one timestamp line.""" + with self._lock: + container = self._container + stream = self._stream + ts_file = self._timestamps_file + if container is None or stream is None or ts_file is None: + # Not started or already stopped — silently drop. + return + + try: + if self._first_pts_us is None: + self._first_pts_us = host_ns // 1000 + pts_us = (host_ns // 1000) - self._first_pts_us + + packet = av.Packet(jpeg_bytes) + packet.stream = stream + packet.pts = pts_us + packet.dts = pts_us + # Nominal duration so MP4 readers that don't compute + # from PTS-deltas (some players) still get sensible + # per-frame timing. + packet.duration = max(1, _MP4_TIME_BASE_DEN // self._fps) + container.mux(packet) + + ts_line = json.dumps({ + "frame_number": self._frame_count, + "capture_ns": int(host_ns), + "quest_native_ns": ( + int(quest_native_ns) if quest_native_ns else None + ), + "clock_domain": "remote_quest3", + "uncertainty_ns": 10_000_000, + }) + ts_file.write(ts_line + "\n") + + if self._first_capture_ns is None: + self._first_capture_ns = int(host_ns) + self._last_capture_ns = int(host_ns) + self._frame_count += 1 + except Exception as exc: # noqa: BLE001 — keep recorder alive + self._write_errors += 1 + # Log first failure loudly, then once every 30 to avoid + # drowning logcat if the container goes permanently bad. + if self._write_errors == 1 or self._write_errors % 30 == 0: + logger.warning( + "[%s/%s] mux error (frame #%d, total errors=%d): %s", + self._stream_id, self._side, self._frame_count, + self._write_errors, exc, + ) + + def stop(self) -> StreamingVideoResult: + """Flush + close everything and return the artifact paths. + + Idempotent. A second ``stop()`` returns the same result and + does no further I/O. Errors during close are logged but do not + propagate — the result still describes whatever we managed to + write before the failure. + """ + with self._lock: + container = self._container + ts_file = self._timestamps_file + self._container = None + self._stream = None + self._timestamps_file = None + + if container is not None: + try: + container.close() + except BaseException as exc: # noqa: BLE001 + logger.warning( + "[%s/%s] container close error: %s", + self._stream_id, self._side, exc, + ) + if ts_file is not None: + try: + ts_file.close() + except Exception as exc: # noqa: BLE001 + logger.warning( + "[%s/%s] timestamps close error: %s", + self._stream_id, self._side, exc, + ) + + result = StreamingVideoResult( + output_path=self.output_path, + timestamps_path=self.timestamps_path, + frame_count=self._frame_count, + first_capture_ns=self._first_capture_ns, + last_capture_ns=self._last_capture_ns, + write_errors=self._write_errors, + ) + logger.info( + "[%s/%s] streaming recorder closed: %d frames, %d write errors", + self._stream_id, self._side, result.frame_count, result.write_errors, + ) + return result diff --git a/src/syncfield/adapters/meta_quest_camera/preview.py b/src/syncfield/adapters/meta_quest_camera/preview.py index 4fd4a7f..351c7c2 100644 --- a/src/syncfield/adapters/meta_quest_camera/preview.py +++ b/src/syncfield/adapters/meta_quest_camera/preview.py @@ -3,15 +3,29 @@ from __future__ import annotations from dataclasses import dataclass -from typing import BinaryIO, Iterator +from typing import BinaryIO, Iterator, Optional @dataclass(frozen=True) class MjpegFrame: - """One JPEG frame pulled from the Quest's MJPEG preview endpoint.""" + """One JPEG frame pulled from the Quest's MJPEG preview endpoint. + + ``capture_ns`` is the host-projected nanosecond timestamp + (``X-Frame-Capture-Ns``) — i.e. already in the Mac monotonic clock + domain, courtesy of the ``deltaNs`` offset the Quest sender computes + at recording start. + + ``quest_native_ns`` is the Quest's raw monotonic clock at acquisition + (``X-Quest-Native-Ns``). Optional for backwards compatibility with + older Quest sender builds — older firmwares simply omit the header + and the field stays ``None``. Post-processing can use the native + timestamp to detect / correct host↔quest clock drift over a long + session (the one-shot ``deltaNs`` offset is otherwise frozen). + """ jpeg_bytes: bytes capture_ns: int + quest_native_ns: Optional[int] = None def _readline(stream: BinaryIO) -> bytes: @@ -59,12 +73,26 @@ def iter_mjpeg_frames( except KeyError as exc: raise ValueError(f"missing required header: {exc.args[0]}") from exc + # Optional — present on Quest sender builds that ship the + # quest-native timestamp alongside the host-projected one. + # Old builds will omit it and post-hoc drift correction + # simply isn't available for those recordings. + quest_native_raw = headers.get("x-quest-native-ns") + try: + quest_native_ns = int(quest_native_raw) if quest_native_raw else None + except ValueError: + quest_native_ns = None + body = stream.read(length) if len(body) != length: raise EOFError("truncated MJPEG part body") # Consume the trailing CRLF. stream.readline() - yield MjpegFrame(jpeg_bytes=body, capture_ns=capture_ns) + yield MjpegFrame( + jpeg_bytes=body, + capture_ns=capture_ns, + quest_native_ns=quest_native_ns, + ) import logging @@ -77,6 +105,9 @@ def iter_mjpeg_frames( logger = logging.getLogger(__name__) +FrameSink = Callable[[MjpegFrame], None] + + class MjpegPreviewConsumer: """Background thread that pulls the Quest's MJPEG preview into ``latest_frame``. @@ -86,6 +117,23 @@ class MjpegPreviewConsumer: ``numpy.ndarray`` (BGR) suitable for the viewer; when ``False`` it is the raw :class:`MjpegFrame` — useful for tests that don't want to pull in OpenCV. + + Frame fan-out + ------------- + Each received frame is delivered to *both* of: + + * ``latest_frame`` slot — decoded BGR array used by the viewer + panel. Always populated when ``decode_jpeg=True``. + * the optional :class:`FrameSink` registered via + :meth:`set_frame_sink` — receives the *raw* :class:`MjpegFrame` + so consumers (e.g. the streaming MP4 recorder) get bit-exact + JPEG bytes plus the original timestamps without paying for + JPEG decode + re-encode. + + The sink may be added or cleared at any time (the recorder turns + on at ``start_recording`` and off at ``stop_recording``); the + network connection itself stays up across recording cycles so the + viewer panel never goes black mid-record. """ def __init__( @@ -107,6 +155,9 @@ def __init__( self._latest: Optional[object] = None self._stop_event = threading.Event() self._thread: Optional[threading.Thread] = None + # Sink is a single-slot mutable hook — replacing or clearing + # the sink is atomic at attribute level; we never iterate it. + self._frame_sink: Optional[FrameSink] = None # ------------------------------------------------------------------ @@ -130,6 +181,17 @@ def stop(self) -> None: self._thread.join(timeout=2.0) self._thread = None + def set_frame_sink(self, sink: Optional[FrameSink]) -> None: + """Register (or clear, with ``None``) a per-frame callback. + + Called from the consumer thread once per received frame, before + the JPEG is decoded for the viewer slot — so the sink sees the + raw bytes and original timestamp/native-ns. Sink exceptions are + caught and logged so a misbehaving recorder cannot kill the + viewer feed. + """ + self._frame_sink = sink + # ------------------------------------------------------------------ # Exponential backoff for repeat failures. Quest's HTTP server @@ -177,6 +239,17 @@ def _consume_once(self) -> None: # streaming responses and MockTransport (content=) work correctly. buffer = _StreamAdapter(response.iter_bytes(8192), self._stop_event) for frame in iter_mjpeg_frames(buffer, boundary=self._boundary): + # Sink first — it gets the raw JPEG bytes regardless + # of decode_jpeg, which lets the streaming recorder + # operate on bit-exact source frames even while the + # viewer slot holds a decoded BGR array. + sink = self._frame_sink + if sink is not None: + try: + sink(frame) + except Exception as exc: # noqa: BLE001 + logger.warning("frame sink raised: %s", exc) + decoded: object if self._decode_jpeg: decoded = _decode_jpeg(frame.jpeg_bytes) diff --git a/src/syncfield/adapters/meta_quest_camera/stream.py b/src/syncfield/adapters/meta_quest_camera/stream.py index 65b60a0..1d4ff55 100644 --- a/src/syncfield/adapters/meta_quest_camera/stream.py +++ b/src/syncfield/adapters/meta_quest_camera/stream.py @@ -1,42 +1,88 @@ -"""MetaQuestCameraStream — SyncField adapter for Quest 3 stereo passthrough cameras.""" +"""MetaQuestCameraStream — streaming-only adapter for Quest 3 stereo cameras. + +The Quest companion app exposes a 720p MJPEG stream per eye over HTTP +(``/preview/{left|right}``). This adapter: + +* keeps one persistent MJPEG consumer per eye while connected (powering + the viewer's video panel via the decoded ``latest_frame``); +* on :meth:`start_recording`, attaches a :class:`StreamingVideoRecorder` + *sink* to each consumer so the same JPEG bytes are muxed straight + into per-eye MP4 files on the Mac plus a sidecar timestamps JSONL — + no Quest-side disk write, no end-of-session "pull" stage. + +Why streaming-only: + +* **No stop-time wait.** The previous design recorded MJPEG-AVI on the + Quest then pulled ~90 MB per eye over HTTP after stop. On a 4G-class + WiFi link that took 30 s+ and looked like a hang in the viewer. +* **Single channel.** Both viewer preview and recording read the same + stream — fewer Quest-side resources, fewer failure modes. +* **Bit-exact quality.** Frames are muxed without re-encoding so what + the viewer sees is exactly what ends up in the file. + +See ``docs/superpowers/specs/2026-04-13-metaquest-stereo-camera-design.md`` +for the broader protocol design. +""" from __future__ import annotations import logging +import time from pathlib import Path from typing import Optional, Tuple -import time - import httpx -from syncfield.adapters.meta_quest_camera.file_puller import RecordingFilePuller from syncfield.adapters.meta_quest_camera.http_client import QuestHttpClient -from syncfield.adapters.meta_quest_camera.preview import MjpegPreviewConsumer -from syncfield.adapters.meta_quest_camera.timestamps import TimestampTailReader +from syncfield.adapters.meta_quest_camera.mp4_writer import ( + StreamingVideoRecorder, + StreamingVideoResult, +) +from syncfield.adapters.meta_quest_camera.preview import ( + MjpegFrame, + MjpegPreviewConsumer, +) from syncfield.clock import SessionClock from syncfield.stream import DeviceKey, StreamBase -from syncfield.types import FinalizationReport, HealthEvent, HealthEventKind, SampleEvent, StreamCapabilities +from syncfield.types import ( + FinalizationReport, + HealthEvent, + HealthEventKind, + SampleEvent, + StreamCapabilities, +) logger = logging.getLogger(__name__) -# Matches the Quest companion Unity app's default HTTP port (spec §2). +# Matches the Quest companion Unity app's default HTTP port. DEFAULT_QUEST_HTTP_PORT = 14045 DEFAULT_FPS = 30 DEFAULT_RESOLUTION: Tuple[int, int] = (1280, 720) +# Multipart boundary string the Quest sender embeds in /preview/{eye} +# Content-Type. Lives here (not in the consumer) because it is part of +# the protocol contract between the two sides. +_PREVIEW_BOUNDARY = b"syncfield" + class MetaQuestCameraStream(StreamBase): - """Captures Meta Quest 3 stereo passthrough cameras (hybrid mode). + """Stream + record Meta Quest 3 stereo passthrough cameras over HTTP. + + The Quest sender (``opengraph-studio/unity/SyncFieldQuest3Sender``) + publishes a 720p MJPEG stream per eye on UDP / TCP port + ``quest_port`` (default 14045). Connecting opens both streams and + keeps them alive for the lifetime of the adapter; recording is then + a cheap toggle that attaches a passthrough MP4 writer to each + stream. - Live: low-res MJPEG preview pulled from the Quest for the viewer. - Recorded: 720p×30 H.264 recorded on the Quest, pulled to - ``output_dir`` after :meth:`stop_recording` completes. + Files written under ``output_dir`` per recorded session: - See ``docs/superpowers/specs/2026-04-13-metaquest-stereo-camera-design.md`` - for the full protocol + architecture notes. + ``{stream_id}_left.mp4`` MJPEG-in-MP4, no re-encode + ``{stream_id}_right.mp4`` MJPEG-in-MP4, no re-encode + ``{stream_id}_left.timestamps.jsonl`` per-frame host + quest-native ns + ``{stream_id}_right.timestamps.jsonl`` per-frame host + quest-native ns """ CLOCK_DOMAIN = "remote_quest3" @@ -69,15 +115,20 @@ def __init__( self._resolution = resolution self._output_dir = Path(output_dir) self._transport = _transport + self._http: Optional[QuestHttpClient] = None self._preview_left: Optional[MjpegPreviewConsumer] = None self._preview_right: Optional[MjpegPreviewConsumer] = None self._connected = False - self._timestamp_tail: Optional[TimestampTailReader] = None + + # Per-recording state — non-None only between start_recording + # and stop_recording. + self._recorder_left: Optional[StreamingVideoRecorder] = None + self._recorder_right: Optional[StreamingVideoRecorder] = None self._session_id: Optional[str] = None self._first_at: Optional[int] = None self._last_at: Optional[int] = None - self._frame_count = 0 + self._frame_count = 0 # frames sample-emitted (left eye is authoritative) @property def device_key(self) -> Optional[DeviceKey]: @@ -87,6 +138,13 @@ def device_key(self) -> Optional[DeviceKey]: def is_connected(self) -> bool: return self._connected + # ------------------------------------------------------------------ + # 4-phase lifecycle + # ------------------------------------------------------------------ + + def prepare(self) -> None: + pass + def connect(self) -> None: if self._connected: return @@ -95,7 +153,9 @@ def connect(self) -> None: port=self._quest_port, transport=self._transport, ) - # Probe reachability up front so failures surface before recording starts. + # Probe reachability up front so a missing Quest fails the + # whole connect() instead of silently flapping inside the + # preview consumer's reconnect loop. self._http.status() self._preview_left = self._make_preview("left") self._preview_right = self._make_preview("right") @@ -108,25 +168,32 @@ def connect(self) -> None: ) def disconnect(self) -> None: - if self._preview_left is not None: - self._preview_left.stop() - self._preview_left = None - if self._preview_right is not None: - self._preview_right.stop() - self._preview_right = None + # If a recording is somehow still active when disconnect() is + # called, finalise it best-effort first so the MP4 trailer + # gets written and the file is playable. + if self._recorder_left is not None or self._recorder_right is not None: + try: + self.stop_recording() + except Exception as exc: # noqa: BLE001 + logger.warning( + "[%s] disconnect-time stop_recording failed: %s", self.id, exc, + ) + + for consumer in (self._preview_left, self._preview_right): + if consumer is not None: + consumer.stop() + self._preview_left = None + self._preview_right = None if self._http is not None: self._http.close() self._http = None self._connected = False - # ------------------------------------------------------------------ - - def prepare(self) -> None: - pass - def start_recording(self, session_clock: SessionClock) -> None: - if self._http is None: + if not self._connected: raise RuntimeError("start_recording() called before connect()") + if self._recorder_left is not None or self._recorder_right is not None: + raise RuntimeError("recording already in progress") self._session_id = ( f"ep_{session_clock.sync_point.timestamp_ms}" @@ -136,121 +203,161 @@ def start_recording(self, session_clock: SessionClock) -> None: self._first_at = None self._last_at = None - self._http.start_recording( - session_id=self._session_id, - host_mono_ns=session_clock.sync_point.monotonic_ns, + # Stand up one writer per eye. Both write under the orchestrator's + # output_dir using the stream-id prefix, so the four artifacts of + # this session sit next to each other in the episode folder. + self._recorder_left = StreamingVideoRecorder( + output_dir=self._output_dir, + stream_id=self.id, + side="left", + fps=self._fps, width=self._resolution[0], height=self._resolution[1], - fps=self._fps, ) - - # Tail the LEFT eye's chunked timestamps endpoint; right eye's exact - # per-frame ts lives in the authoritative JSONL written by the puller. - url = ( - f"http://{self._quest_host}:{self._quest_port}" - f"/recording/timestamps/left" - ) - self._timestamp_tail = TimestampTailReader( - url=url, + self._recorder_right = StreamingVideoRecorder( + output_dir=self._output_dir, stream_id=self.id, - on_sample=self._handle_tail_sample, - transport=self._transport, - clock_domain=self.CLOCK_DOMAIN, - uncertainty_ns=self.UNCERTAINTY_NS, + side="right", + fps=self._fps, + width=self._resolution[0], + height=self._resolution[1], ) - self._timestamp_tail.start() + self._recorder_left.start() + self._recorder_right.start() - def stop_recording(self) -> FinalizationReport: - if self._http is None: - raise RuntimeError("stop_recording() called before connect()") + # Hot-attach sinks. The preview consumers stay running across + # the recording cycle — viewer feed never blanks. + assert self._preview_left is not None and self._preview_right is not None + self._preview_left.set_frame_sink(self._make_sink(self._recorder_left, "left")) + self._preview_right.set_frame_sink(self._make_sink(self._recorder_right, "right")) - try: - stop_response = self._http.stop_recording() - if self._timestamp_tail is not None: - self._timestamp_tail.stop() - self._timestamp_tail = None + logger.info( + "[%s] streaming recording started (session=%s, %dx%d @ %dfps)", + self.id, self._session_id, + self._resolution[0], self._resolution[1], self._fps, + ) - puller = RecordingFilePuller( - client=self._http, stream_id=self.id, output_dir=self._output_dir - ) - artifacts = puller.pull_all() - - # Verify file sizes match the /stop response (spec §4.3). - size_errors = [] - left_actual = artifacts.left_mp4.stat().st_size - if left_actual != stop_response.left.bytes: - size_errors.append( - f"left size mismatch: expected {stop_response.left.bytes} bytes," - f" got {left_actual} bytes on disk" - ) - right_actual = artifacts.right_mp4.stat().st_size - if right_actual != stop_response.right.bytes: - size_errors.append( - f"right size mismatch: expected {stop_response.right.bytes} bytes," - f" got {right_actual} bytes on disk" - ) + def stop_recording(self) -> FinalizationReport: + # Detach sinks first so no more frames land in a half-closed + # writer while we flush. + if self._preview_left is not None: + self._preview_left.set_frame_sink(None) + if self._preview_right is not None: + self._preview_right.set_frame_sink(None) - if size_errors: - status = "partial" - error: Optional[str] = "; ".join(size_errors) - logger.warning( - "[%s] Recording files may be truncated: %s", self.id, error - ) - else: - # All good — tell the Quest to clean up the session files. - self._http.delete_recording() - status = "completed" - error = None - except Exception as exc: - status = "failed" - error = str(exc) - artifacts = None + result_left = self._recorder_left.stop() if self._recorder_left else None + result_right = self._recorder_right.stop() if self._recorder_right else None + self._recorder_left = None + self._recorder_right = None + status, error = self._classify_outcome(result_left, result_right) return FinalizationReport( stream_id=self.id, status=status, frame_count=self._frame_count, - file_path=artifacts.left_mp4 if artifacts is not None else None, + file_path=result_left.output_path if result_left is not None else None, first_sample_at_ns=self._first_at, last_sample_at_ns=self._last_at, health_events=list(self._collected_health), error=error, ) - def _handle_tail_sample(self, event: SampleEvent) -> None: - if self._first_at is None: - self._first_at = event.capture_ns - self._last_at = event.capture_ns - self._frame_count += 1 - self._emit_sample(event) - # ------------------------------------------------------------------ + # Internals + # ------------------------------------------------------------------ + + def _classify_outcome( + self, + left: Optional[StreamingVideoResult], + right: Optional[StreamingVideoResult], + ) -> Tuple[str, Optional[str]]: + """Map per-eye write results to (status, error_message).""" + if left is None or right is None: + return "failed", "recorder was not started" + if left.frame_count == 0 and right.frame_count == 0: + return "failed", "no frames received during recording" + msgs = [] + if left.write_errors > 0 or right.write_errors > 0: + msgs.append( + f"mux errors left={left.write_errors} right={right.write_errors}" + ) + if left.frame_count == 0 or right.frame_count == 0: + msgs.append( + f"single-eye recording: left={left.frame_count} right={right.frame_count}" + ) + if msgs: + return "partial", "; ".join(msgs) + return "completed", None def _make_preview(self, side: str) -> MjpegPreviewConsumer: url = f"http://{self._quest_host}:{self._quest_port}/preview/{side}" + return MjpegPreviewConsumer( + url=url, + boundary=_PREVIEW_BOUNDARY, + transport=self._transport, + decode_jpeg=True, + on_health=self._make_health_callback(side), + ) + + def _make_health_callback(self, side: str): + kind_map = { + "drop": HealthEventKind.DROP, + "reconnect": HealthEventKind.RECONNECT, + "warning": HealthEventKind.WARNING, + } def _on_health(kind: str, detail: str) -> None: - mapping = { - "drop": HealthEventKind.DROP, - "reconnect": HealthEventKind.RECONNECT, - "warning": HealthEventKind.WARNING, - } self._emit_health( HealthEvent( stream_id=self.id, - kind=mapping.get(kind, HealthEventKind.WARNING), + kind=kind_map.get(kind, HealthEventKind.WARNING), at_ns=time.monotonic_ns(), detail=f"[{side}] {detail}", ) ) - return MjpegPreviewConsumer( - url=url, - boundary=b"syncfield", - transport=self._transport, - decode_jpeg=True, - on_health=_on_health, - ) + return _on_health + + def _make_sink(self, recorder: StreamingVideoRecorder, side: str): + """Build the per-frame sink the preview consumer calls. + + The sink does two things on each frame: + + 1. Mux the JPEG into the per-eye MP4 + emit a timestamp line. + 2. Emit a SampleEvent through ``StreamBase`` so the orchestrator + sees the recording progressing (viewer counter, sync stats). + + Only the LEFT-eye sink emits SampleEvents — emitting from both + would double-count frames for what is, conceptually, one + synchronised stereo capture per tick. + """ + is_authoritative = side == "left" + + def _sink(frame: MjpegFrame) -> None: + recorder.write_frame( + frame.jpeg_bytes, frame.capture_ns, frame.quest_native_ns, + ) + if not is_authoritative: + return + if self._first_at is None: + self._first_at = frame.capture_ns + self._last_at = frame.capture_ns + frame_number = self._frame_count + self._frame_count += 1 + self._emit_sample(SampleEvent( + stream_id=self.id, + frame_number=frame_number, + capture_ns=frame.capture_ns, + channels={}, # video stream has no scalar channels + uncertainty_ns=self.UNCERTAINTY_NS, + clock_domain=self.CLOCK_DOMAIN, + )) + + return _sink + + # ------------------------------------------------------------------ + # Viewer-facing properties + # ------------------------------------------------------------------ @property def latest_frame_left(self): @@ -267,16 +374,13 @@ def latest_frame_right(self): @property def latest_frame(self): - """Viewer-compat: return a side-by-side ``[left | right]`` composite - so the single video panel shows both eyes at once. The viewer's - ``StreamSnapshot`` polls ``stream.latest_frame`` for any adapter - declaring ``kind="video"``; syncfield's panel model is 1 panel = - 1 stream_id, so until we split into two adapters we surface the - stereo pair as a horizontally-concatenated frame. - - Falls back to whichever eye is available if the other is still - connecting or has dropped — users should see *something* rather - than a black card whenever at least one preview is alive. + """Side-by-side ``[left | right]`` composite for the viewer panel. + + The viewer's video panel polls a single ``latest_frame`` per + stream, so we surface the stereo pair as a horizontally + concatenated array. Falls back to whichever eye is fresh when + the other is still warming up — better to render half a frame + than a black card. """ left = self.latest_frame_left right = self.latest_frame_right @@ -284,9 +388,10 @@ def latest_frame(self): import numpy as np if left.shape == right.shape: return np.hstack((left, right)) - # Shapes can diverge for a frame or two during startup while - # the two previews race to produce their first decoded image. - # Fall through to the single-eye path instead of raising. + # Shapes can disagree for a frame or two during startup + # while the two previews race to produce their first + # decoded image. Fall through to the single-eye path + # rather than raising. if left is not None: return left return right diff --git a/src/syncfield/adapters/meta_quest_camera/timestamps.py b/src/syncfield/adapters/meta_quest_camera/timestamps.py deleted file mode 100644 index b9be4bd..0000000 --- a/src/syncfield/adapters/meta_quest_camera/timestamps.py +++ /dev/null @@ -1,87 +0,0 @@ -"""Tails the Quest's ``/recording/timestamps/{side}`` chunked JSONL response -and emits one :class:`SampleEvent` per successfully-parsed line.""" - -from __future__ import annotations - -import json -import logging -import threading -from typing import Callable, Optional - -import httpx - -from syncfield.types import SampleEvent - - -logger = logging.getLogger(__name__) - - -class TimestampTailReader: - """Background thread that drives the adapter's ``SampleEvent`` stream.""" - - def __init__( - self, - *, - url: str, - stream_id: str, - on_sample: Callable[[SampleEvent], None], - transport: Optional[httpx.BaseTransport] = None, - clock_domain: str = "remote_quest3", - uncertainty_ns: int = 10_000_000, - ) -> None: - self._url = url - self._stream_id = stream_id - self._on_sample = on_sample - self._transport = transport - self._clock_domain = clock_domain - self._uncertainty_ns = uncertainty_ns - - self._stop_event = threading.Event() - self._thread: Optional[threading.Thread] = None - - def start(self) -> None: - if self._thread is not None and self._thread.is_alive(): - return - self._stop_event.clear() - self._thread = threading.Thread( - target=self._run, name=f"quest-ts-{self._stream_id}", daemon=True - ) - self._thread.start() - - def stop(self) -> None: - self._stop_event.set() - if self._thread is not None: - self._thread.join(timeout=2.0) - self._thread = None - - def _run(self) -> None: - client = httpx.Client(transport=self._transport, timeout=None) - try: - with client.stream("GET", self._url) as response: - response.raise_for_status() - for line in response.iter_lines(): - if self._stop_event.is_set(): - return - if not line: - continue - try: - payload = json.loads(line) - frame_number = int(payload["frame_number"]) - capture_ns = int(payload["capture_ns"]) - except (json.JSONDecodeError, KeyError, TypeError, ValueError): - logger.warning("skipping malformed timestamp line: %r", line) - continue - self._on_sample( - SampleEvent( - stream_id=self._stream_id, - frame_number=frame_number, - capture_ns=capture_ns, - channels=None, - uncertainty_ns=self._uncertainty_ns, - clock_domain=self._clock_domain, - ) - ) - except httpx.HTTPError as exc: # pragma: no cover — real-Quest path - logger.warning("timestamp stream closed: %s", exc) - finally: - client.close() diff --git a/tests/unit/adapters/meta_quest_camera/test_file_puller.py b/tests/unit/adapters/meta_quest_camera/test_file_puller.py deleted file mode 100644 index 227be5d..0000000 --- a/tests/unit/adapters/meta_quest_camera/test_file_puller.py +++ /dev/null @@ -1,54 +0,0 @@ -"""Unit tests for RecordingFilePuller.""" - -from __future__ import annotations - -import httpx -import pytest - -from syncfield.adapters.meta_quest_camera.file_puller import ( - RecordingFilePuller, - RecordingArtifacts, -) -from syncfield.adapters.meta_quest_camera.http_client import QuestHttpClient - - -def _router(): - files = { - "/recording/files/left": b"LEFT_MP4", - "/recording/files/right": b"RIGHT_MP4", - "/recording/timestamps/left": - b'{"frame_number":0,"capture_ns":1}\n{"frame_number":1,"capture_ns":2}\n', - "/recording/timestamps/right": - b'{"frame_number":0,"capture_ns":1}\n{"frame_number":1,"capture_ns":2}\n', - } - - def handler(request: httpx.Request) -> httpx.Response: - body = files.get(request.url.path) - if body is None: - return httpx.Response(404) - return httpx.Response( - 200, headers={"Content-Length": str(len(body))}, content=body - ) - - return httpx.MockTransport(handler) - - -class TestRecordingFilePuller: - def test_pulls_all_four_artifacts(self, tmp_path): - client = QuestHttpClient(host="test", port=14045, transport=_router()) - puller = RecordingFilePuller( - client=client, stream_id="quest_cam", output_dir=tmp_path - ) - artifacts = puller.pull_all() - - assert isinstance(artifacts, RecordingArtifacts) - assert artifacts.left_mp4.read_bytes() == b"LEFT_MP4" - assert artifacts.right_mp4.read_bytes() == b"RIGHT_MP4" - assert artifacts.left_timestamps.exists() - assert artifacts.right_timestamps.exists() - - # File naming matches the adapter's documented output layout. - assert artifacts.left_mp4.name == "quest_cam_left.mp4" - assert artifacts.right_mp4.name == "quest_cam_right.mp4" - assert artifacts.left_timestamps.name == "quest_cam_left.timestamps.jsonl" - assert artifacts.right_timestamps.name == "quest_cam_right.timestamps.jsonl" diff --git a/tests/unit/adapters/meta_quest_camera/test_stream_lifecycle.py b/tests/unit/adapters/meta_quest_camera/test_stream_lifecycle.py index d4bf469..8475fe3 100644 --- a/tests/unit/adapters/meta_quest_camera/test_stream_lifecycle.py +++ b/tests/unit/adapters/meta_quest_camera/test_stream_lifecycle.py @@ -1,41 +1,44 @@ -"""Unit tests for the top-level MetaQuestCameraStream adapter.""" +"""Unit tests for the streaming MetaQuestCameraStream adapter. + +The adapter sits on top of three collaborators that have their own +focused tests (preview, mp4_writer, http_client). These tests cover +the surface the orchestrator actually drives: + +* identity / capabilities +* connect/disconnect lifecycle (status probe, preview start) +* start/stop recording without exercising real network I/O — the + ``StreamingVideoRecorder`` integration is verified by feeding raw + JPEG frames into the consumer's sink directly. +""" from __future__ import annotations +import io +import time from pathlib import Path +import httpx import pytest +from PIL import Image from syncfield.adapters.meta_quest_camera import MetaQuestCameraStream +from syncfield.adapters.meta_quest_camera.preview import MjpegFrame +from syncfield.clock import SessionClock, SyncPoint -class TestIdentity: - def test_stream_identity_and_capabilities(self, tmp_path: Path): - stream = MetaQuestCameraStream( - id="quest_cam", - quest_host="192.0.2.10", - output_dir=tmp_path, - ) - assert stream.id == "quest_cam" - assert stream.kind == "video" - assert stream.capabilities.produces_file is True - assert stream.capabilities.supports_precise_timestamps is True - assert stream.capabilities.is_removable is True - assert stream.capabilities.provides_audio_track is False - - def test_device_key_includes_host(self, tmp_path: Path): - stream = MetaQuestCameraStream( - id="quest_cam", - quest_host="192.0.2.10", - output_dir=tmp_path, - ) - assert stream.device_key == ("meta_quest_camera", "192.0.2.10") +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- -import httpx +def _status_only_transport() -> httpx.MockTransport: + """Transport that satisfies /status and stalls preview pulls. + Used for tests that don't care about frame delivery — preview + consumers stay blocked reading an empty body so latest_frame + stays None and no sink callbacks fire. + """ -def _status_only_transport() -> httpx.MockTransport: def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/status": return httpx.Response( @@ -50,7 +53,6 @@ def handler(request: httpx.Request) -> httpx.Response: }, ) if request.url.path.startswith("/preview/"): - # Return a tiny valid MJPEG body (no frames) so the consumer blocks. return httpx.Response( 200, headers={ @@ -63,13 +65,54 @@ def handler(request: httpx.Request) -> httpx.Response: return httpx.MockTransport(handler) +def _make_jpeg(width: int = 64, height: int = 64, color=(120, 50, 200)) -> bytes: + """Build a minimal valid JPEG so PyAV's MJPEG packetiser is happy.""" + img = Image.new("RGB", (width, height), color=color) + buf = io.BytesIO() + img.save(buf, format="JPEG", quality=80) + return buf.getvalue() + + +# --------------------------------------------------------------------------- +# Identity +# --------------------------------------------------------------------------- + + +class TestIdentity: + def test_stream_identity_and_capabilities(self, tmp_path: Path): + stream = MetaQuestCameraStream( + id="quest_cam", + quest_host="192.0.2.10", + output_dir=tmp_path, + ) + assert stream.id == "quest_cam" + assert stream.kind == "video" + assert stream.capabilities.produces_file is True + assert stream.capabilities.supports_precise_timestamps is True + assert stream.capabilities.is_removable is True + assert stream.capabilities.provides_audio_track is False + + def test_device_key_includes_host(self, tmp_path: Path): + stream = MetaQuestCameraStream( + id="quest_cam", + quest_host="192.0.2.10", + output_dir=tmp_path, + ) + assert stream.device_key == ("meta_quest_camera", "192.0.2.10") + + +# --------------------------------------------------------------------------- +# Connect / disconnect +# --------------------------------------------------------------------------- + + class TestConnectDisconnect: def test_connect_runs_status_probe_and_starts_preview(self, tmp_path): stream = MetaQuestCameraStream( id="quest_cam", quest_host="test", output_dir=tmp_path, - _transport=_status_only_transport(), # test-only injection + _transport=_status_only_transport(), ) stream.connect() assert stream.is_connected is True @@ -90,156 +133,125 @@ def handler(request: httpx.Request) -> httpx.Response: stream.connect() -import json -from syncfield.clock import SessionClock, SyncPoint - - -def _full_quest_transport(left_mp4=b"LEFT_MP4", right_mp4=b"RIGHT_MP4"): - state = {"recording": False} - - def handler(request: httpx.Request) -> httpx.Response: - path = request.url.path - if path == "/status": - return httpx.Response(200, json={ - "recording": state["recording"], "session_id": None, - "last_preview_capture_ns": 0, - "left_camera_ready": True, "right_camera_ready": True, - "storage_free_bytes": 1_000_000_000, - }) - if path.startswith("/preview/"): - return httpx.Response(200, headers={ - "Content-Type": "multipart/x-mixed-replace; boundary=syncfield" - }, content=b"") - if path == "/recording/start": - state["recording"] = True - return httpx.Response(200, json={ - "session_id": "ep_x", "quest_mono_ns_at_start": 0, - "delta_ns": 0, "started": True, - }) - if path == "/recording/stop": - state["recording"] = False - return httpx.Response(200, json={ - "session_id": "ep_x", - "left": {"frame_count": 2, "bytes": len(left_mp4), "last_capture_ns": 2}, - "right": {"frame_count": 2, "bytes": len(right_mp4), "last_capture_ns": 2}, - "duration_s": 0.1, - }) - if path == "/recording/files/left": - return httpx.Response(200, headers={"Content-Length": str(len(left_mp4))}, content=left_mp4) - if path == "/recording/files/right": - return httpx.Response(200, headers={"Content-Length": str(len(right_mp4))}, content=right_mp4) - if path == "/recording/timestamps/left" or path == "/recording/timestamps/right": - body = ( - b'{"frame_number":0,"capture_ns":1}\n' - b'{"frame_number":1,"capture_ns":2}\n' - ) - return httpx.Response(200, headers={"Content-Length": str(len(body))}, content=body) - if path == "/recording/files" and request.method == "DELETE": - return httpx.Response(204) - return httpx.Response(404) +# --------------------------------------------------------------------------- +# Recording — exercises the sink + StreamingVideoRecorder integration +# --------------------------------------------------------------------------- - return httpx.MockTransport(handler) +class TestRecording: + def test_start_recording_requires_connect(self, tmp_path): + stream = MetaQuestCameraStream( + id="quest_cam", quest_host="test", output_dir=tmp_path, + ) + clock = SessionClock(sync_point=SyncPoint.create_now("test_host")) + with pytest.raises(RuntimeError): + stream.start_recording(clock) -class TestRecordingRoundtrip: - def test_full_recording_lifecycle(self, tmp_path): + def test_start_then_stop_with_no_frames_marks_failed(self, tmp_path): + """No frames flowed through the sink → stop returns failed status.""" stream = MetaQuestCameraStream( id="quest_cam", quest_host="test", output_dir=tmp_path, - _transport=_full_quest_transport(), + _transport=_status_only_transport(), ) stream.connect() clock = SessionClock(sync_point=SyncPoint.create_now("test_host")) - stream.start_recording(clock) report = stream.stop_recording() stream.disconnect() + assert report.status == "failed" + assert "no frames" in (report.error or "") - assert report.status == "completed" - assert (tmp_path / "quest_cam_left.mp4").read_bytes() == b"LEFT_MP4" - assert (tmp_path / "quest_cam_right.mp4").read_bytes() == b"RIGHT_MP4" - assert (tmp_path / "quest_cam_left.timestamps.jsonl").exists() - assert (tmp_path / "quest_cam_right.timestamps.jsonl").exists() + def test_recording_writes_mp4_and_timestamps_on_frames(self, tmp_path): + """Push a handful of JPEG frames into both sinks → both eyes + produce valid mp4 + jsonl, status==completed.""" + stream = MetaQuestCameraStream( + id="quest_cam", + quest_host="test", + output_dir=tmp_path, + _transport=_status_only_transport(), + resolution=(64, 64), + ) + stream.connect() + clock = SessionClock(sync_point=SyncPoint.create_now("test_host")) + stream.start_recording(clock) + # Reach into the consumers and fire their registered sinks + # directly — bypasses the network thread so the test stays + # deterministic. The sink is the same callable the consumer + # would invoke per real MJPEG frame. + jpeg = _make_jpeg() + base_ns = time.monotonic_ns() + for i in range(5): + host_ns = base_ns + i * 33_333_333 + quest_ns = host_ns - 1_000_000 # arbitrary delta + stream._preview_left._frame_sink( + MjpegFrame(jpeg_bytes=jpeg, capture_ns=host_ns, quest_native_ns=quest_ns) + ) + stream._preview_right._frame_sink( + MjpegFrame(jpeg_bytes=jpeg, capture_ns=host_ns, quest_native_ns=quest_ns) + ) -class TestSizeMismatch: - def test_partial_status_when_size_mismatch(self, tmp_path): - """When /stop says left.bytes=9999 but actual file is 8 bytes, status=partial.""" - # left_mp4 body is b"LEFT_MP4" (8 bytes), but /stop will claim bytes=9999 + report = stream.stop_recording() + stream.disconnect() + + assert report.status == "completed", report.error + assert report.frame_count == 5 + assert (tmp_path / "quest_cam_left.mp4").stat().st_size > 0 + assert (tmp_path / "quest_cam_right.mp4").stat().st_size > 0 + ts_left = (tmp_path / "quest_cam_left.timestamps.jsonl").read_text() + ts_right = (tmp_path / "quest_cam_right.timestamps.jsonl").read_text() + assert ts_left.count("\n") == 5 + assert ts_right.count("\n") == 5 + # Each line must carry both timestamps. + assert "quest_native_ns" in ts_left + + def test_partial_status_when_only_one_eye_received_frames(self, tmp_path): stream = MetaQuestCameraStream( id="quest_cam", quest_host="test", output_dir=tmp_path, - _transport=_full_quest_transport(left_mp4=b"LEFT_MP4", right_mp4=b"RIGHT_MP4"), + _transport=_status_only_transport(), + resolution=(64, 64), ) + stream.connect() + clock = SessionClock(sync_point=SyncPoint.create_now("test_host")) + stream.start_recording(clock) - # Build a transport that overrides /recording/stop to return wrong byte count - def _mismatched_transport(): - state = {"recording": False} - - def handler(request: httpx.Request) -> httpx.Response: - path = request.url.path - if path == "/status": - return httpx.Response(200, json={ - "recording": state["recording"], "session_id": None, - "last_preview_capture_ns": 0, - "left_camera_ready": True, "right_camera_ready": True, - "storage_free_bytes": 1_000_000_000, - }) - if path.startswith("/preview/"): - return httpx.Response(200, headers={ - "Content-Type": "multipart/x-mixed-replace; boundary=syncfield" - }, content=b"") - if path == "/recording/start": - state["recording"] = True - return httpx.Response(200, json={ - "session_id": "ep_x", "quest_mono_ns_at_start": 0, - "delta_ns": 0, "started": True, - }) - if path == "/recording/stop": - state["recording"] = False - return httpx.Response(200, json={ - "session_id": "ep_x", - # 9999 != 8 bytes of b"LEFT_MP4" - "left": {"frame_count": 2, "bytes": 9999, "last_capture_ns": 2}, - "right": {"frame_count": 2, "bytes": len(b"RIGHT_MP4"), "last_capture_ns": 2}, - "duration_s": 0.1, - }) - if path == "/recording/files/left": - body = b"LEFT_MP4" - return httpx.Response(200, headers={"Content-Length": str(len(body))}, content=body) - if path == "/recording/files/right": - body = b"RIGHT_MP4" - return httpx.Response(200, headers={"Content-Length": str(len(body))}, content=body) - if path == "/recording/timestamps/left" or path == "/recording/timestamps/right": - body = ( - b'{"frame_number":0,"capture_ns":1}\n' - b'{"frame_number":1,"capture_ns":2}\n' - ) - return httpx.Response(200, headers={"Content-Length": str(len(body))}, content=body) - if path == "/recording/files": - # DELETE — best-effort cleanup - return httpx.Response(204) - return httpx.Response(404) - - return httpx.MockTransport(handler) - - stream2 = MetaQuestCameraStream( + jpeg = _make_jpeg() + # Only the left eye gets frames (right stays at zero). + for i in range(3): + stream._preview_left._frame_sink( + MjpegFrame(jpeg_bytes=jpeg, capture_ns=time.monotonic_ns() + i, quest_native_ns=None) + ) + + report = stream.stop_recording() + stream.disconnect() + assert report.status == "partial" + assert "single-eye" in (report.error or "") or "right=0" in (report.error or "") + + def test_double_start_recording_raises(self, tmp_path): + stream = MetaQuestCameraStream( id="quest_cam", quest_host="test", - output_dir=tmp_path / "mismatch", - _transport=_mismatched_transport(), + output_dir=tmp_path, + _transport=_status_only_transport(), ) - stream2.connect() + stream.connect() clock = SessionClock(sync_point=SyncPoint.create_now("test_host")) - stream2.start_recording(clock) - report = stream2.stop_recording() - stream2.disconnect() + stream.start_recording(clock) + try: + with pytest.raises(RuntimeError, match="already in progress"): + stream.start_recording(clock) + finally: + stream.stop_recording() + stream.disconnect() - assert report.status == "partial" - assert report.error is not None - assert "size" in report.error.lower() or "bytes" in report.error.lower() + +# --------------------------------------------------------------------------- +# Viewer-facing properties +# --------------------------------------------------------------------------- class TestLatestFrame: @@ -249,6 +261,7 @@ def test_latest_frame_none_before_connect(self, tmp_path): ) assert stream.latest_frame_left is None assert stream.latest_frame_right is None + assert stream.latest_frame is None def test_latest_frame_reads_from_preview_consumers(self, tmp_path): stream = MetaQuestCameraStream( @@ -256,7 +269,8 @@ def test_latest_frame_reads_from_preview_consumers(self, tmp_path): _transport=_status_only_transport(), ) stream.connect() - # Consumers returned empty body in the fixture, so latest_frame stays None. + # Empty preview body in the fixture → consumers never produce + # a decoded frame, so the slot stays None. assert stream.latest_frame_left is None assert stream.latest_frame_right is None stream.disconnect() diff --git a/tests/unit/adapters/meta_quest_camera/test_timestamps.py b/tests/unit/adapters/meta_quest_camera/test_timestamps.py deleted file mode 100644 index f65ec80..0000000 --- a/tests/unit/adapters/meta_quest_camera/test_timestamps.py +++ /dev/null @@ -1,85 +0,0 @@ -"""Unit tests for TimestampTailReader.""" - -from __future__ import annotations - -import json -import time - -import httpx -import pytest - -from syncfield.adapters.meta_quest_camera.timestamps import TimestampTailReader -from syncfield.types import SampleEvent - - -def _chunked_jsonl_transport(lines: list[dict]) -> httpx.MockTransport: - body = b"".join( - (json.dumps(line) + "\n").encode("ascii") for line in lines - ) - - def handler(request: httpx.Request) -> httpx.Response: - return httpx.Response( - 200, - headers={"Content-Type": "application/x-ndjson"}, - content=body, - ) - - return httpx.MockTransport(handler) - - -class TestTimestampTailReader: - def test_emits_sample_event_per_line(self): - lines = [ - {"frame_number": 0, "capture_ns": 100}, - {"frame_number": 1, "capture_ns": 200}, - {"frame_number": 2, "capture_ns": 300}, - ] - events: list[SampleEvent] = [] - - reader = TimestampTailReader( - url="http://test/recording/timestamps/left", - stream_id="quest_cam", - on_sample=events.append, - transport=_chunked_jsonl_transport(lines), - clock_domain="remote_quest3", - uncertainty_ns=10_000_000, - ) - reader.start() - deadline = time.time() + 1.0 - while time.time() < deadline and len(events) < 3: - time.sleep(0.01) - reader.stop() - - assert len(events) == 3 - assert [e.frame_number for e in events] == [0, 1, 2] - assert [e.capture_ns for e in events] == [100, 200, 300] - assert all(e.clock_domain == "remote_quest3" for e in events) - assert all(e.uncertainty_ns == 10_000_000 for e in events) - assert all(e.stream_id == "quest_cam" for e in events) - assert all(e.channels is None for e in events) - - def test_ignores_malformed_lines(self): - def handler(request: httpx.Request) -> httpx.Response: - return httpx.Response( - 200, - headers={"Content-Type": "application/x-ndjson"}, - content=( - b'{"frame_number": 0, "capture_ns": 1}\n' - b"not-json\n" - b'{"frame_number": 1, "capture_ns": 2}\n' - ), - ) - - events: list[SampleEvent] = [] - reader = TimestampTailReader( - url="http://test/recording/timestamps/left", - stream_id="quest_cam", - on_sample=events.append, - transport=httpx.MockTransport(handler), - ) - reader.start() - deadline = time.time() + 1.0 - while time.time() < deadline and len(events) < 2: - time.sleep(0.01) - reader.stop() - assert [e.frame_number for e in events] == [0, 1]