feat: intra-host sync anchor metadata

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "syncfield"
-version = "0.3.21"
+version = "0.3.22"
 description = "Multi-modal capture orchestration framework with precision sync for Physical AI data collection"
 readme = "README.md"
 license = "Apache-2.0"

src/syncfield/adapters/_video_encoder.py

@@ -143,10 +143,47 @@ def open_uvc_input(
 
     The returned container yields packets via ``.demux()`` which the
     caller decodes frame-by-frame.
+
+    Low-latency tuning
+    ------------------
+    Live camera capture is not file playback — ffmpeg's default demuxer
+    fills a 5 MB packet probe buffer and ``analyzeduration`` waits up
+    to 5 s to finalise stream info before handing any frame to Python.
+    On a Continuity Camera / H.264-over-USB path that adds hundreds of
+    milliseconds of arrival-vs-real-shutter delay and produces a
+    warm-up "burst" where the first interval is ~110 ms and the next
+    few are <20 ms as the backlog drains. Both manifest as jitter in
+    downstream sync.
+
+    The options below reduce every ffmpeg-controllable source of
+    buffering on the input path:
+
+    * ``fflags=nobuffer+flush_packets`` — skip the packet queue fill
+      on open; flush each packet as soon as the demuxer emits it.
+    * ``flags=low_delay`` — decoder stops waiting for B-frame reorder
+      (harmless for camera streams, which are IPPP with ``bf=0``).
+    * ``analyzeduration=0`` — hand the first frame off as soon as
+      stream info is known; don't average over a preset window.
+    * ``max_delay=0`` — demuxer max demux-interleaving delay cap.
+
+    After opening, we additionally clamp the video decoder's
+    ``thread_type`` to ``NONE`` and set the ``LOW_DELAY`` codec flag.
+    Multi-threaded decode can sit on a frame waiting for neighbours
+    to finish — we'd rather burn a little more CPU on one thread and
+    get each frame out immediately.
+
+    ``probesize`` is intentionally left at the ffmpeg default. Dropping
+    it below ~32 KB breaks H.264 probe on Continuity Camera (SPS/PPS
+    may not land in the first packet), which costs more than the
+    milliseconds its default value adds.
     """
     options = {
         "video_size": f"{int(width)}x{int(height)}",
         "framerate": str(int(round(fps))),
+        "fflags": "nobuffer+flush_packets",
+        "flags": "low_delay",
+        "analyzeduration": "0",
+        "max_delay": "0",
     }
     if pixel_format is not None:
         options["pixel_format"] = pixel_format
@@ -168,7 +205,36 @@ def open_uvc_input(
     else:
         raise RuntimeError(f"Unsupported platform for UVC input: {sys.platform}")
 
-    return av.open(url, format=fmt, options=options)
+    container = av.open(url, format=fmt, options=options)
+
+    # Decoder-side low-latency tuning. The demuxer options above only
+    # affect the input/container layer; the H.264 decoder has its own
+    # reordering buffer and thread-pool frame latency. For a live
+    # camera stream neither is desirable — no B-frames arrive, and the
+    # worker-thread batching delays each frame by a few ms while it
+    # waits for enough work. Ask the decoder to emit every frame as
+    # soon as it finishes, single-threaded, with the LOW_DELAY codec
+    # flag asserted. Wrapped in try/except because older PyAV releases
+    # exposed slightly different enum shapes — if we fail to clamp the
+    # decoder we just keep the defaults, which is strictly no worse
+    # than before this function existed.
+    try:
+        for stream in container.streams.video:
+            cc = stream.codec_context
+            try:
+                cc.thread_type = "NONE"
+            except (ValueError, AttributeError):
+                pass
+            try:
+                # PyAV exposes codec flags as an IntFlag; LOW_DELAY is
+                # 0x0008 in ffmpeg's AV_CODEC_FLAG_LOW_DELAY.
+                cc.flags |= 0x0008
+            except (AttributeError, TypeError):
+                pass
+    except Exception:  # pragma: no cover — best-effort
+        pass
+
+    return container
 
 
 def compute_jitter_percentiles(

src/syncfield/adapters/ble_imu.py

@@ -304,6 +304,7 @@ def start_recording(self, session_clock: SessionClock) -> None:
         """Begin counting incoming samples toward the finalization report."""
         if self._thread is None or not self._thread.is_alive():
             self.connect()
+        self._begin_recording_window(session_clock)
         self._recording = True
 
     def stop_recording(self) -> FinalizationReport:
@@ -318,6 +319,7 @@ def stop_recording(self) -> FinalizationReport:
             last_sample_at_ns=self._last_at,
             health_events=list(self._collected_health),
             error=None,
+            recording_anchor=self._recording_anchor(),
         )
 
     def disconnect(self) -> None:
@@ -492,6 +494,10 @@ def _handle_payload(self, payload: bytes) -> None:
             sample_ns = recv_ns - (n_samples - 1 - i) * self._sample_period_ns
 
             if self._recording:
+                # No device-side clock in the generic decoder —
+                # ``sample_ns`` is derived from ``recv_ns``, not the
+                # sensor's own clock. Pass None for device_ns.
+                self._observe_first_frame(sample_ns, None)
                 if self._first_at is None:
                     self._first_at = sample_ns
                 self._last_at = sample_ns

src/syncfield/adapters/host_audio.py

@@ -174,6 +174,9 @@ def _audio_callback(indata, frames, time_info, status):
 
             # Write to WAV if recording
             if self._recording and self._wav_writer is not None:
+                # PortAudio / sounddevice exposes no device-side clock
+                # for the host mic — pass None for device_ns.
+                self._observe_first_frame(capture_ns, None)
                 pcm16 = (mono * 32767).astype(np.int16)
                 self._wav_writer.writeframes(pcm16.tobytes())
                 if self._first_at is None:
@@ -203,6 +206,7 @@ def _audio_callback(indata, frames, time_info, status):
 
     def start_recording(self, session_clock: SessionClock) -> None:
         """Start writing audio to WAV file."""
+        self._begin_recording_window(session_clock)
         self._frame_count = 0
         self._first_at = None
         self._last_at = None
@@ -242,6 +246,7 @@ def stop_recording(self) -> FinalizationReport:
             last_sample_at_ns=self._last_at,
             health_events=list(self._collected_health),
             error=None,
+            recording_anchor=self._recording_anchor(),
         )
 
     def disconnect(self) -> None:

src/syncfield/adapters/meta_quest.py

@@ -436,6 +436,7 @@ def connect(self) -> None:
         self._start_discovery_responder()
 
     def start_recording(self, session_clock: SessionClock) -> None:
+        self._begin_recording_window(session_clock)
         self._recording = True
         self._frame_count = 0
         self._first_at = None
@@ -452,6 +453,7 @@ def stop_recording(self) -> FinalizationReport:
             last_sample_at_ns=self._last_at,
             health_events=list(self._collected_health),
             error=None,
+            recording_anchor=self._recording_anchor(),
         )
 
     def disconnect(self) -> None:
@@ -744,6 +746,10 @@ def _process_packet(self, data: bytes) -> None:
         frame_number = self._frame_count
         self._frame_count += 1
         if self._recording:
+            # Quest packet carries a device-side ``ts_ms``, but the
+            # current parser doesn't expose it as a dedicated field —
+            # pass None so the anchor reports only host-side arrival.
+            self._observe_first_frame(capture_ns, None)
             if self._first_at is None:
                 self._first_at = capture_ns
             self._last_at = capture_ns

src/syncfield/adapters/meta_quest_camera/stream.py

@@ -207,6 +207,7 @@ def start_recording(self, session_clock: SessionClock) -> None:
         if self._recorder is not None:
             raise RuntimeError("recording already in progress")
 
+        self._begin_recording_window(session_clock)
         self._session_id = (
             f"ep_{session_clock.sync_point.timestamp_ms}"
             f"_{session_clock.sync_point.host_id}"
@@ -254,6 +255,7 @@ def stop_recording(self) -> FinalizationReport:
             last_sample_at_ns=self._last_at,
             health_events=list(self._collected_health),
             error=error,
+            recording_anchor=self._recording_anchor(),
         )
 
     # ------------------------------------------------------------------
@@ -333,6 +335,10 @@ def _sink(frame: MjpegFrame) -> None:
             recorder.write_frame(
                 frame.jpeg_bytes, frame.capture_ns, frame.quest_native_ns,
             )
+            # Quest exposes a per-frame native-clock timestamp
+            # (``quest_native_ns``) projected into the host domain —
+            # use it as the anchor's device-side ns when present.
+            self._observe_first_frame(frame.capture_ns, frame.quest_native_ns)
             if self._first_at is None:
                 self._first_at = frame.capture_ns
             self._last_at = frame.capture_ns

src/syncfield/adapters/oak_camera.py

@@ -77,22 +77,22 @@
 )
 
 
-def _device_shutter_host_ns(msg: Any) -> Optional[int]:
-    """Return the frame's shutter-close time, projected onto the host's
-    monotonic clock, as integer nanoseconds — or ``None`` if unavailable.
-
-    DepthAI periodically cross-correlates its on-chip Myriad-X clock with
-    the host's ``time.monotonic_ns()``, so ``msg.getTimestamp()`` returns
-    a ``datetime.timedelta`` whose value is the frame's **shutter instant
-    already translated into the host clock domain** — upstream of the
-    ISP / encoder / XLink pipeline depth that otherwise biases the
-    arrival-time ``capture_ns`` we stamp in the capture loop.
-
-    Adapters expose this alongside ``capture_ns`` (never in place of it)
-    so a downstream aligner can opportunistically anchor on the true
-    shutter instant for devices that provide one, and fall back to the
-    arrival timestamp for those that don't — preserving SyncField's
-    hardware-agnostic contract.
+def _device_timestamp_ns(msg: Any) -> Optional[int]:
+    """Return the frame's device-clock timestamp as integer nanoseconds.
+
+    ``msg.getTimestamp()`` is a ``datetime.timedelta`` anchored to the
+    Myriad-X board's own clock (power-up relative). This helper returns
+    the raw value — no attempt to project it onto the host monotonic
+    clock, because DepthAI 3.x does not actually synchronise the two
+    (the earlier ``device_shutter_host_ns`` path discovered ~12 day
+    offsets between boards; cross-domain projection is unsafe).
+
+    Downstream we use this value only for **inter-frame interval
+    smoothing**: the deltas between consecutive frames' device clocks
+    are jitter-free sensor cadence, which — combined with host arrival
+    as the session anchor — removes host-side XLink/transport jitter
+    from ``capture_ns`` without caring about absolute clock alignment.
+    See ``SyncSession._refine_video_with_device_timestamps``.
     """
     if msg is None:
         return None
@@ -102,8 +102,7 @@ def _device_shutter_host_ns(msg: Any) -> Optional[int]:
         return None
     if td is None:
         return None
-    # Integer arithmetic — avoid float rounding at ns scale for the
-    # ~10¹⁸ ns magnitudes reached after long uptimes.
+    # Integer arithmetic — avoid float rounding at ns magnitudes.
     return ((td.days * 86_400 + td.seconds) * 1_000_000 + td.microseconds) * 1_000
 
 
@@ -287,13 +286,21 @@ def prepare(self) -> None:
         pass
 
     #: How many times to poll ``dai.Device.getAllAvailableDevices()``
-    #: before giving up. The first call often returns only a subset on
-    #: dual-OAK rigs because XLink enumeration is asynchronous — the
-    #: second board shows up after 0.5–1 s. Three tries with a short
-    #: sleep between comfortably covers that gap without extending the
-    #: happy-path connect time (which still returns on the first call).
-    _ENUMERATE_RETRIES = 3
-    _ENUMERATE_RETRY_DELAY_S = 0.8
+    #: before giving up. XLink enumeration is asynchronous — on multi-
+    #: board rigs the first probe often returns only a subset, and the
+    #: rest appear up to several seconds later while a sibling board
+    #: boots and the Mac USB stack re-quiesces. We've observed:
+    #:
+    #: * dual-OAK (USB-3 + USB-3 on different controllers): ~2 s window
+    #: * dual-OAK involving OAK-D-Lite (USB-2-only board): 10–20 s
+    #:   before the USB-2 board reappears in the enumeration list
+    #: * triple-OAK with USB-2 hub sharing: even longer
+    #:
+    #: 24 s ceiling covers the worst case we've measured without blowing
+    #: up the happy-path connect time — the probe returns on first hit
+    #: when every board is already visible.
+    _ENUMERATE_RETRIES = 16
+    _ENUMERATE_RETRY_DELAY_S = 1.5
 
     def _locate_device(self) -> Any:
         """Find the target OAK, retrying the XLink enumeration if needed.
@@ -411,6 +418,7 @@ def start_recording(self, session_clock: SessionClock) -> None:
         path), the pipeline is started here first so the writer always
         has a feeder.
         """
+        self._begin_recording_window(session_clock)
         if self._thread is None or not self._thread.is_alive():
             self.connect()
         self._output_dir.mkdir(parents=True, exist_ok=True)
@@ -481,6 +489,7 @@ def stop_recording(self) -> FinalizationReport:
             error=None,
             jitter_p95_ns=jitter_p95,
             jitter_p99_ns=jitter_p99,
+            recording_anchor=self._recording_anchor(),
         )
 
     def _finalize_mp4(self) -> bool:
@@ -715,24 +724,25 @@ def _capture_loop(self) -> None:
             if rgb_msg is None:
                 continue
 
-            # DepthAI projects the on-device shutter moment into host
-            # monotonic time — pull it here, *before* any handler touches
-            # the message, so downstream sync consumers can anchor on the
-            # true shutter instant instead of the pipeline-depth-biased
-            # arrival stamp. Surfaced via ``SampleEvent.channels`` so the
-            # orchestrator lands it in the jsonl ``extras``.
-            device_shutter_host_ns = _device_shutter_host_ns(rgb_msg)
+            # Device-clock timestamp (raw Myriad-X ns since board power-up).
+            # Pulled here — *before* any handler touches the message — so the
+            # value travels alongside ``capture_ns`` to the orchestrator.
+            # Downstream device-interval smoothing in ``SyncSession`` uses
+            # the deltas between consecutive frames' device clocks to scrub
+            # host-arrival jitter out of the recorded ``capture_ns``.
+            device_ts_ns = _device_timestamp_ns(rgb_msg)
 
             # Recording-window-only jitter collection (see UVC adapter for rationale).
             if self._recording:
+                self._observe_first_frame(capture_ns, device_ts_ns)
                 if self._prev_capture_ns is not None:
                     self._intervals_ns.append(capture_ns - self._prev_capture_ns)
                 self._prev_capture_ns = capture_ns
 
             if self._encoding == OAK_ENCODING_H264:
-                self._handle_encoded_packet(rgb_msg, capture_ns, device_shutter_host_ns)
+                self._handle_encoded_packet(rgb_msg, capture_ns, device_ts_ns)
             else:
-                self._handle_raw_frame(rgb_msg, capture_ns, device_shutter_host_ns)
+                self._handle_raw_frame(rgb_msg, capture_ns, device_ts_ns)
 
             if self._recording and self._depth_enabled:
                 self._drain_depth_tick()
@@ -741,7 +751,7 @@ def _handle_encoded_packet(
         self,
         msg: Any,
         capture_ns: int,
-        device_shutter_host_ns: Optional[int],
+        device_ts_ns: Optional[int],
     ) -> None:
         """h264 mode — write the on-device encoded packet to the raw
         ``.h264`` file and emit a :class:`SampleEvent`.
@@ -761,9 +771,7 @@ def _handle_encoded_packet(
             # gives us a plain buffer the OS write path prefers.
             self._h264_file.write(bytes(msg.getData()))
         channels = (
-            {"device_shutter_host_ns": device_shutter_host_ns}
-            if device_shutter_host_ns is not None
-            else None
+            {"device_timestamp_ns": device_ts_ns} if device_ts_ns is not None else None
         )
         self._emit_sample(
             SampleEvent(
@@ -778,7 +786,7 @@ def _handle_raw_frame(
         self,
         msg: Any,
         capture_ns: int,
-        device_shutter_host_ns: Optional[int],
+        device_ts_ns: Optional[int],
     ) -> None:
         """raw mode — publish the BGR frame for preview and host-encode
         via PyAV.
@@ -796,9 +804,7 @@ def _handle_raw_frame(
         if self._video_writer is not None:
             self._video_writer.write(frame)
         channels = (
-            {"device_shutter_host_ns": device_shutter_host_ns}
-            if device_shutter_host_ns is not None
-            else None
+            {"device_timestamp_ns": device_ts_ns} if device_ts_ns is not None else None
         )
         self._emit_sample(
             SampleEvent(