diff --git a/docs/superpowers/plans/2026-04-23-recording-anchor-metadata.md b/docs/superpowers/plans/2026-04-23-recording-anchor-metadata.md new file mode 100644 index 0000000..a32452c --- /dev/null +++ b/docs/superpowers/plans/2026-04-23-recording-anchor-metadata.md @@ -0,0 +1,1159 @@ +# Intra-Host Sync Anchor Metadata Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 각 스트림 어댑터가 "recording armed" 공통 host 시각과 첫 기록 프레임의 `(host_ts, device_ts)` pair를 명시적 metadata로 기록. downstream sync service가 어댑터별 pipeline latency bias를 상쇄할 수 있는 anchor 정보 제공. + +**Architecture:** 기존 `SessionClock` 인프라 확장 (frozen dataclass 에 optional `recording_armed_ns` 필드 추가). `Orchestrator.start()` 가 `start_recording()` 호출 직전에 공통 T를 찍어 `SessionClock` 의 새 복사본을 만들어 모든 어댑터에 전달. 각 어댑터의 capture loop 가 첫 기록 프레임에 `RecordingAnchor(armed_host_ns, first_frame_host_ns, first_frame_device_ns)` 를 `FinalizationReport.recording_anchor` 로 기록. 기존 `start_recording(session_clock)` 시그니처 완전 유지 — 어댑터 side는 `session_clock.recording_armed_ns` 를 읽기만 하면 됨. + +**Tech Stack:** Python 3.11, pytest, dataclasses. 기존 `StreamBase` protocol, `SessionClock`, `FinalizationReport` 에 필드 추가만으로 backward compatible. + +**Scope:** syncfield-python 측 anchor 캡처 + metadata 기록. syncfield(sync service) 측 alignment 활용은 별도 plan. + +--- + +## File Structure + +**Modify:** +- `src/syncfield/types.py` — `RecordingAnchor` dataclass + `FinalizationReport.recording_anchor` 필드 +- `src/syncfield/clock.py` — `SessionClock.recording_armed_ns` 필드 +- `src/syncfield/stream.py` — `StreamBase._recording_anchor` helper +- `src/syncfield/orchestrator.py` — armed_ns 찍어 SessionClock 복제 후 전파 + manifest 에 anchor 수집 +- `src/syncfield/adapters/oak_camera.py` — capture loop 에 anchor 캡처 +- `src/syncfield/adapters/uvc_webcam.py` — capture loop 에 anchor 캡처 +- `src/syncfield/adapters/polling_sensor.py` — 샘플 loop 에 anchor 캡처 +- `src/syncfield/adapters/push_sensor.py` — 샘플 loop 에 anchor 캡처 + +**Tests:** +- `tests/unit/test_types.py` — `RecordingAnchor` 직렬화 +- `tests/unit/test_clock.py` — `SessionClock` armed 필드 +- `tests/unit/test_stream_base.py` — anchor helper +- `tests/unit/adapters/test_oak_camera.py` — OAK anchor 캡처 +- `tests/unit/adapters/test_uvc_webcam.py` — UVC anchor 캡처 +- `tests/unit/test_orchestrator.py` — armed_ns 전파 + manifest 수집 + +**Decision notes:** +- `SessionClock.recording_armed_ns: Optional[int] = None` — preview phase 에서는 None. orchestrator가 start_recording 직전에 `dataclasses.replace(clock, recording_armed_ns=armed_ns)` 로 복제. +- `FinalizationReport.recording_anchor: Optional[RecordingAnchor] = None` — 어댑터가 첫 프레임 관찰 못 하면 (empty recording) None. +- `RecordingAnchor` 는 `first_frame_device_ns` optional — UVC/host_audio 처럼 device clock 없는 어댑터는 None. +- 나머지 어댑터들 (meta_quest, oglo_tactile, ble_imu, host_audio, jsonl_file, insta360_go3s, meta_quest_camera) 은 동일 패턴이라 한 번에 확장 — 본 plan 의 Task 7-10 에서 처리. 패턴 확립 후 확장이 쉽도록 Task 3 의 helper 를 공통화. + +--- + +## Task 1: `RecordingAnchor` dataclass + +**Files:** +- Modify: `src/syncfield/types.py` (add after `SyncPoint` class, around line 73) +- Test: `tests/unit/test_types.py` + +- [ ] **Step 1: Write failing test** + +파일: `tests/unit/test_types.py` 에 추가 + +```python +def test_recording_anchor_with_device_ts(): + from syncfield.types import RecordingAnchor + anchor = RecordingAnchor( + armed_host_ns=1_000_000_000, + first_frame_host_ns=1_044_000_000, + first_frame_device_ns=9_876_543_210, + ) + assert anchor.first_frame_latency_ns == 44_000_000 + assert anchor.to_dict() == { + "armed_host_ns": 1_000_000_000, + "first_frame_host_ns": 1_044_000_000, + "first_frame_device_ns": 9_876_543_210, + "first_frame_latency_ns": 44_000_000, + } + +def test_recording_anchor_without_device_ts(): + from syncfield.types import RecordingAnchor + anchor = RecordingAnchor( + armed_host_ns=1_000, + first_frame_host_ns=1_044_000_000, + ) + assert anchor.first_frame_device_ns is None + d = anchor.to_dict() + assert d["first_frame_device_ns"] is None + assert d["first_frame_latency_ns"] == 1_044_000_000 - 1_000 + +def test_recording_anchor_rejects_first_before_armed(): + from syncfield.types import RecordingAnchor + import pytest + with pytest.raises(ValueError, match="first_frame_host_ns must be >= armed_host_ns"): + RecordingAnchor(armed_host_ns=100, first_frame_host_ns=50) +``` + +- [ ] **Step 2: Run test to verify failure** + +```bash +cd /Users/jerry/Documents/syncfield-python && uv run pytest tests/unit/test_types.py::test_recording_anchor_with_device_ts -v +``` +Expected: `ImportError: cannot import name 'RecordingAnchor'` + +- [ ] **Step 3: Implement `RecordingAnchor`** + +`src/syncfield/types.py` — `SyncPoint` 클래스 정의 끝부분 (line ~73) 바로 다음에 추가: + +```python +@dataclass(frozen=True) +class RecordingAnchor: + """Per-stream anchor info captured when recording is armed. + + Captures the common host ``armed_host_ns`` (shared by all streams in + the session) together with the first recorded frame's ``(host_ts, + device_ts)`` pair for this stream. Downstream sync tooling uses the + difference ``first_frame_host_ns - armed_host_ns`` to estimate each + adapter's observed pipeline latency and remove per-adapter bias when + aligning streams. + + Attributes: + armed_host_ns: Common host monotonic_ns captured by the + orchestrator immediately before ``start_recording()`` is + fanned out to streams. Identical across all streams in a + single recording window. + first_frame_host_ns: Host monotonic_ns at which this stream's + first recorded frame arrived on the host. + first_frame_device_ns: Optional device-clock timestamp of the + first recorded frame. ``None`` for adapters without a + device-side clock (UVC webcams, host audio, etc). + """ + + armed_host_ns: int + first_frame_host_ns: int + first_frame_device_ns: int | None = None + + def __post_init__(self) -> None: + if self.first_frame_host_ns < self.armed_host_ns: + raise ValueError( + f"first_frame_host_ns must be >= armed_host_ns; " + f"got armed={self.armed_host_ns}, first={self.first_frame_host_ns}" + ) + + @property + def first_frame_latency_ns(self) -> int: + """Observed latency from armed moment to first frame arrival.""" + return self.first_frame_host_ns - self.armed_host_ns + + def to_dict(self) -> dict[str, Any]: + return { + "armed_host_ns": self.armed_host_ns, + "first_frame_host_ns": self.first_frame_host_ns, + "first_frame_device_ns": self.first_frame_device_ns, + "first_frame_latency_ns": self.first_frame_latency_ns, + } +``` + +- [ ] **Step 4: Run tests to verify pass** + +```bash +uv run pytest tests/unit/test_types.py -v -k recording_anchor +``` +Expected: 3 passed + +- [ ] **Step 5: Commit** + +```bash +git add src/syncfield/types.py tests/unit/test_types.py +git commit -m "feat(types): add RecordingAnchor dataclass for per-stream sync anchor" +``` + +--- + +## Task 2: Extend `SessionClock` with `recording_armed_ns` + +**Files:** +- Modify: `src/syncfield/clock.py` +- Test: `tests/unit/test_clock.py` + +- [ ] **Step 1: Write failing test** + +파일: `tests/unit/test_clock.py` 에 추가 (기존 파일 없으면 생성) + +```python +import dataclasses + +from syncfield.clock import SessionClock +from syncfield.types import SyncPoint + + +def _make_clock() -> SessionClock: + sp = SyncPoint.create_now(host_id="host_a") + return SessionClock(sync_point=sp) + + +def test_session_clock_preview_phase_has_no_armed_ns(): + clock = _make_clock() + assert clock.recording_armed_ns is None + + +def test_session_clock_arm_returns_new_clock_with_armed_ns(): + clock = _make_clock() + armed = dataclasses.replace(clock, recording_armed_ns=12_345) + assert armed.recording_armed_ns == 12_345 + assert clock.recording_armed_ns is None # original unchanged + + +def test_session_clock_armed_ns_survives_frozen_semantics(): + clock = _make_clock() + armed = dataclasses.replace(clock, recording_armed_ns=500) + import pytest + with pytest.raises(dataclasses.FrozenInstanceError): + armed.recording_armed_ns = 700 # type: ignore[misc] +``` + +- [ ] **Step 2: Run test to verify failure** + +```bash +uv run pytest tests/unit/test_clock.py -v +``` +Expected: `AttributeError: 'SessionClock' object has no attribute 'recording_armed_ns'` + +- [ ] **Step 3: Add field to `SessionClock`** + +`src/syncfield/clock.py` — `SessionClock` dataclass 를 아래와 같이 수정: + +```python +@dataclass(frozen=True) +class SessionClock: + """Shared monotonic clock reference for all streams in one session. + + A ``SessionClock`` is cheap to copy, safe to share across threads, and + binds each stream in a session to the exact same monotonic anchor. The + orchestrator constructs it once at ``start()`` and distributes it to + every ``Stream.start(session_clock)`` call so intra-session timing uses + a single source of truth. + + Attributes: + sync_point: The session's :class:`SyncPoint` (monotonic + wall clock + anchor captured at session start). + recording_armed_ns: Common host monotonic_ns captured by the + orchestrator right before it fans out ``start_recording()`` + to every stream. ``None`` during preview phase, non-``None`` + once recording is armed. All streams receive the same value, + so adapters can use it as a shared intra-host sync anchor. + """ + + sync_point: SyncPoint + recording_armed_ns: int | None = None + + @property + def host_id(self) -> str: + """Host identifier for this session.""" + return self.sync_point.host_id + + def now_ns(self) -> int: + """Return the current monotonic nanosecond timestamp. + + Thread-safe: :func:`time.monotonic_ns` is atomic on CPython. + """ + return time.monotonic_ns() + + def elapsed_ns(self) -> int: + """Return nanoseconds elapsed since the session's sync point.""" + return time.monotonic_ns() - self.sync_point.monotonic_ns +``` + +- [ ] **Step 4: Run tests to verify pass** + +```bash +uv run pytest tests/unit/test_clock.py -v +``` +Expected: 3 passed + +- [ ] **Step 5: Commit** + +```bash +git add src/syncfield/clock.py tests/unit/test_clock.py +git commit -m "feat(clock): add SessionClock.recording_armed_ns for shared intra-host anchor" +``` + +--- + +## Task 3: `StreamBase` anchor helper + +**Files:** +- Modify: `src/syncfield/stream.py` +- Test: `tests/unit/test_stream_base.py` + +- [ ] **Step 1: Write failing test** + +파일: `tests/unit/test_stream_base.py` 에 추가 + +```python +import dataclasses + +from syncfield.clock import SessionClock +from syncfield.stream import StreamBase +from syncfield.types import StreamCapabilities, SyncPoint + + +class _Dummy(StreamBase): + def __init__(self) -> None: + super().__init__("d", "sensor", StreamCapabilities()) + + +def _clock(armed_ns: int | None = None) -> SessionClock: + sp = SyncPoint.create_now(host_id="h") + return SessionClock(sync_point=sp, recording_armed_ns=armed_ns) + + +def test_anchor_helper_returns_none_before_first_frame(): + d = _Dummy() + d._begin_recording_window(_clock(armed_ns=100)) + assert d._recording_anchor() is None + + +def test_anchor_helper_captures_first_frame_then_ignores_later(): + d = _Dummy() + d._begin_recording_window(_clock(armed_ns=100)) + d._observe_first_frame(host_ns=250, device_ns=9_000) + d._observe_first_frame(host_ns=300, device_ns=10_000) # ignored + anchor = d._recording_anchor() + assert anchor is not None + assert anchor.armed_host_ns == 100 + assert anchor.first_frame_host_ns == 250 + assert anchor.first_frame_device_ns == 9_000 + + +def test_anchor_helper_without_device_ts(): + d = _Dummy() + d._begin_recording_window(_clock(armed_ns=100)) + d._observe_first_frame(host_ns=250, device_ns=None) + anchor = d._recording_anchor() + assert anchor is not None + assert anchor.first_frame_device_ns is None + + +def test_anchor_helper_noop_if_armed_ns_missing(): + d = _Dummy() + d._begin_recording_window(_clock(armed_ns=None)) + d._observe_first_frame(host_ns=250, device_ns=None) + assert d._recording_anchor() is None + + +def test_anchor_helper_reset_on_second_recording_window(): + d = _Dummy() + d._begin_recording_window(_clock(armed_ns=100)) + d._observe_first_frame(host_ns=250, device_ns=9_000) + d._begin_recording_window(_clock(armed_ns=1_000)) + assert d._recording_anchor() is None # reset + d._observe_first_frame(host_ns=1_100, device_ns=500) + anchor = d._recording_anchor() + assert anchor is not None + assert anchor.armed_host_ns == 1_000 + assert anchor.first_frame_host_ns == 1_100 +``` + +- [ ] **Step 2: Run test to verify failure** + +```bash +uv run pytest tests/unit/test_stream_base.py -v +``` +Expected: `AttributeError: '_Dummy' object has no attribute '_begin_recording_window'` + +- [ ] **Step 3: Add anchor helper methods to `StreamBase`** + +`src/syncfield/stream.py` — `StreamBase.__init__` 안 마지막 라인 (`self._collected_health: ...`) 아래에 세 필드 추가하고, 파일 끝부분 `disconnect()` 직전에 helper 메서드 추가: + +`__init__` 수정 — 라인 207 바로 아래: + +```python + self._collected_health: List[HealthEvent] = [] + # Intra-host sync anchor — captured once per recording window, + # when the first frame/sample arrives after start_recording(). + self._armed_host_ns: int | None = None + self._first_frame_observed: bool = False + self._anchor: Optional["RecordingAnchor"] = None +``` + +파일 상단 import: + +```python +from syncfield.types import ( + # ... 기존 것들, + RecordingAnchor, +) +``` + +`_emit_health` 다음에 helper 메서드 추가 (Lifecycle methods 섹션 직전): + +```python + # ------------------------------------------------------------------ + # Intra-host sync anchor + # ------------------------------------------------------------------ + # + # Each recording window shares a common ``armed_host_ns`` captured by + # the orchestrator. Adapters call ``_begin_recording_window`` from + # ``start_recording`` (with the received ``SessionClock``) and then + # ``_observe_first_frame`` exactly once from their capture loop when + # the first frame/sample of the recording window arrives. The + # resulting :class:`RecordingAnchor` is attached to the stream's + # :class:`FinalizationReport` by ``stop_recording``. + + def _begin_recording_window(self, session_clock: SessionClock) -> None: + """Reset anchor state and remember the armed host timestamp. + + Safe to call even when ``recording_armed_ns`` is ``None`` (legacy + test harnesses / unit mocks) — the helper becomes a no-op. + """ + self._armed_host_ns = session_clock.recording_armed_ns + self._first_frame_observed = False + self._anchor = None + + def _observe_first_frame( + self, host_ns: int, device_ns: int | None + ) -> None: + """Capture the anchor exactly once per recording window. + + Subsequent calls are silently ignored. No-op when there is no + armed_host_ns (preview phase, legacy code path). + """ + if self._first_frame_observed: + return + if self._armed_host_ns is None: + return + # Guard against host clock going backwards under test mocks — + # clamp to armed_ns so RecordingAnchor's invariant holds. + safe_host = max(host_ns, self._armed_host_ns) + self._anchor = RecordingAnchor( + armed_host_ns=self._armed_host_ns, + first_frame_host_ns=safe_host, + first_frame_device_ns=device_ns, + ) + self._first_frame_observed = True + + def _recording_anchor(self) -> Optional["RecordingAnchor"]: + """Return the anchor captured for the current recording window.""" + return self._anchor +``` + +`Optional` 이 상단에 import 되어있는지 확인. 없으면 추가. + +- [ ] **Step 4: Run tests to verify pass** + +```bash +uv run pytest tests/unit/test_stream_base.py -v +``` +Expected: 5 passed + +- [ ] **Step 5: Commit** + +```bash +git add src/syncfield/stream.py tests/unit/test_stream_base.py +git commit -m "feat(stream): add intra-host sync anchor helper to StreamBase" +``` + +--- + +## Task 4: `FinalizationReport.recording_anchor` field + +**Files:** +- Modify: `src/syncfield/types.py` (`FinalizationReport` around line 306) +- Test: `tests/unit/test_types.py` + +- [ ] **Step 1: Write failing test** + +`tests/unit/test_types.py` 에 추가: + +```python +def test_finalization_report_with_anchor(): + from syncfield.types import FinalizationReport, RecordingAnchor + anchor = RecordingAnchor( + armed_host_ns=100, first_frame_host_ns=150, first_frame_device_ns=42 + ) + report = FinalizationReport( + stream_id="s1", status="completed", frame_count=10, + file_path=None, first_sample_at_ns=150, last_sample_at_ns=450, + health_events=[], error=None, recording_anchor=anchor, + ) + assert report.recording_anchor is anchor + + +def test_finalization_report_anchor_defaults_to_none(): + from syncfield.types import FinalizationReport + report = FinalizationReport( + stream_id="s2", status="completed", frame_count=0, + file_path=None, first_sample_at_ns=None, last_sample_at_ns=None, + health_events=[], error=None, + ) + assert report.recording_anchor is None +``` + +- [ ] **Step 2: Run test to verify failure** + +```bash +uv run pytest tests/unit/test_types.py -v -k finalization_report_with_anchor +``` +Expected: `TypeError: ... got an unexpected keyword argument 'recording_anchor'` + +- [ ] **Step 3: Add field** + +`src/syncfield/types.py` — `FinalizationReport` 의 마지막 필드 (`incidents: list = field(default_factory=list)`) 아래에 추가: + +```python + incidents: list = field(default_factory=list) + recording_anchor: RecordingAnchor | None = None +``` + +그리고 클래스 docstring 의 `Attributes:` 섹션 끝에 한 줄 추가: + +``` + recording_anchor: Intra-host sync anchor captured at the start + of the recording window (common ``armed_host_ns`` plus the + stream's first-frame timestamps). ``None`` for empty + recordings or adapters that haven't opted in. +``` + +- [ ] **Step 4: Run tests to verify pass** + +```bash +uv run pytest tests/unit/test_types.py -v -k "anchor or finalization_report" +``` +Expected: all passed + +- [ ] **Step 5: Commit** + +```bash +git add src/syncfield/types.py tests/unit/test_types.py +git commit -m "feat(types): surface RecordingAnchor on FinalizationReport" +``` + +--- + +## Task 5: Orchestrator — arm SessionClock + propagate anchor to manifest + +**Files:** +- Modify: `src/syncfield/orchestrator.py` (around line 1922-1936) +- Test: `tests/unit/test_orchestrator.py` (새 테스트 추가 — 기존 테스트 파일 패턴 따름) + +- [ ] **Step 1: Write failing test** + +파일: `tests/unit/test_orchestrator_anchor.py` (신규) + +```python +"""Orchestrator fans out a single armed_host_ns to all streams.""" +from __future__ import annotations + +import time +from pathlib import Path +from unittest.mock import MagicMock + +from syncfield.clock import SessionClock +from syncfield.orchestrator import SessionOrchestrator +from syncfield.stream import StreamBase +from syncfield.types import ( + FinalizationReport, RecordingAnchor, StreamCapabilities, +) + + +class _CaptureClockStream(StreamBase): + """Records the SessionClock passed to start_recording().""" + + def __init__(self, id: str) -> None: + super().__init__(id, "sensor", StreamCapabilities()) + self.received_clock: SessionClock | None = None + + def connect(self) -> None: pass + + def start_recording(self, session_clock: SessionClock) -> None: + self.received_clock = session_clock + + def stop_recording(self) -> FinalizationReport: + anchor = None + if self.received_clock and self.received_clock.recording_armed_ns: + anchor = RecordingAnchor( + armed_host_ns=self.received_clock.recording_armed_ns, + first_frame_host_ns=self.received_clock.recording_armed_ns + 1_000, + first_frame_device_ns=None, + ) + return FinalizationReport( + stream_id=self.id, status="completed", frame_count=1, + file_path=None, first_sample_at_ns=0, last_sample_at_ns=0, + health_events=[], error=None, recording_anchor=anchor, + ) + + def disconnect(self) -> None: pass + + +def test_orchestrator_arms_clock_and_all_streams_see_same_armed_ns(tmp_path: Path) -> None: + sess = SessionOrchestrator(host_id="h", output_dir=tmp_path) + a = _CaptureClockStream("a"); b = _CaptureClockStream("b") + sess.add(a); sess.add(b) + sess.connect() + sess.start() + time.sleep(0.01) + sess.stop() + + assert a.received_clock is not None and b.received_clock is not None + assert a.received_clock.recording_armed_ns is not None + assert a.received_clock.recording_armed_ns == b.received_clock.recording_armed_ns + + +def test_orchestrator_manifest_includes_per_stream_anchor(tmp_path: Path) -> None: + import json + sess = SessionOrchestrator(host_id="h", output_dir=tmp_path) + a = _CaptureClockStream("a") + sess.add(a) + sess.connect(); sess.start(); time.sleep(0.01); sess.stop() + + manifest_paths = list(tmp_path.rglob("manifest.json")) + assert manifest_paths, "manifest.json not written" + manifest = json.loads(manifest_paths[0].read_text()) + streams = manifest.get("streams", []) + a_entry = next(s for s in streams if s["stream_id"] == "a") + assert "recording_anchor" in a_entry + assert a_entry["recording_anchor"] is not None + assert "armed_host_ns" in a_entry["recording_anchor"] + assert "first_frame_host_ns" in a_entry["recording_anchor"] + assert "first_frame_latency_ns" in a_entry["recording_anchor"] +``` + +- [ ] **Step 2: Run test to verify failure** + +```bash +uv run pytest tests/unit/test_orchestrator_anchor.py -v +``` +Expected: fail — armed_ns is None (orchestrator doesn't set it yet). + +- [ ] **Step 3: Modify orchestrator to arm the clock + collect anchor** + +`src/syncfield/orchestrator.py` 의 line 1922-1936 부근 — `start_recording` fan-out 블록을 찾아서 `start_recording` 호출 직전에 armed_ns 를 찍고 clock 을 replace: + +Before: +```python + # --- Atomic start_recording ------------------------------ + # Open persistence writers BEFORE start_recording so the + # ... existing comments ... + for stream in self._streams: + try: + stream.start_recording(self._session_clock) +``` + +After: +```python + # --- Atomic start_recording ------------------------------ + # Open persistence writers BEFORE start_recording so the + # ... existing comments ... + # + # Capture a single shared armed_host_ns immediately before + # fanning start_recording out — all streams receive the same + # value via SessionClock.recording_armed_ns, which they can + # use as an intra-host sync anchor in their capture loop. + import dataclasses as _dc + armed_ns = time.monotonic_ns() + self._session_clock = _dc.replace( + self._session_clock, recording_armed_ns=armed_ns + ) + for stream in self._streams: + try: + stream.start_recording(self._session_clock) +``` + +(`time` 은 이미 파일 상단에 import 되어 있어야 함 — 없으면 추가.) + +`write_manifest` 호출부에서 each finalization report 의 `recording_anchor` 가 manifest 에 포함되도록 — `src/syncfield/manifest.py` (또는 `write_manifest` 정의 위치) 에서 per-stream entry 를 만드는 곳을 찾아 아래 필드 추가: + +```python +stream_entry["recording_anchor"] = ( + report.recording_anchor.to_dict() if report.recording_anchor else None +) +``` + +(manifest 모듈의 정확한 위치는 구현 시 `grep "write_manifest" src/syncfield/` 로 확인 — 보통 `src/syncfield/manifest.py` 안 `_stream_entry(...)` 또는 `_make_stream_entry(...)` 함수.) + +- [ ] **Step 4: Run tests to verify pass** + +```bash +uv run pytest tests/unit/test_orchestrator_anchor.py -v +``` +Expected: 2 passed + +```bash +# Regression check +uv run pytest tests/ -x --timeout=60 +``` +Expected: full suite green + +- [ ] **Step 5: Commit** + +```bash +git add src/syncfield/orchestrator.py src/syncfield/manifest.py tests/unit/test_orchestrator_anchor.py +git commit -m "feat(orchestrator): arm SessionClock and propagate RecordingAnchor to manifest" +``` + +--- + +## Task 6: `OakCameraStream` — anchor capture + +**Files:** +- Modify: `src/syncfield/adapters/oak_camera.py` (`start_recording` ~line 405, `_capture_loop` ~line 708, `stop_recording` ~line 459) +- Test: `tests/unit/adapters/test_oak_camera.py` + +- [ ] **Step 1: Write failing test** + +`tests/unit/adapters/test_oak_camera.py` — `TestEncoding` 클래스 근처에 추가: + +```python +class TestRecordingAnchor: + def test_anchor_captured_from_first_frame(self, tmp_path, fake_depthai): + from syncfield.adapters.oak_camera import OakCameraStream + from syncfield.clock import SessionClock + from syncfield.types import SyncPoint + import dataclasses as _dc, time + + stream = OakCameraStream("oak1", tmp_path, device_id="dev") + stream.connect() + sp = SyncPoint.create_now("h") + clock = SessionClock(sync_point=sp, recording_armed_ns=time.monotonic_ns()) + stream.start_recording(clock) + fake_depthai.push_frame(stream) # fixture helper — pushes one frame + time.sleep(0.05) + report = stream.stop_recording() + stream.disconnect() + + assert report.recording_anchor is not None + assert report.recording_anchor.armed_host_ns == clock.recording_armed_ns + assert report.recording_anchor.first_frame_host_ns >= clock.recording_armed_ns + assert report.recording_anchor.first_frame_device_ns is not None + + def test_no_anchor_when_no_frames_arrive(self, tmp_path, fake_depthai): + from syncfield.adapters.oak_camera import OakCameraStream + from syncfield.clock import SessionClock + from syncfield.types import SyncPoint + import time + + stream = OakCameraStream("oak2", tmp_path, device_id="dev") + stream.connect() + sp = SyncPoint.create_now("h") + clock = SessionClock(sync_point=sp, recording_armed_ns=time.monotonic_ns()) + stream.start_recording(clock) + # no frames pushed + report = stream.stop_recording() + stream.disconnect() + + assert report.recording_anchor is None +``` + +(실제 fake_depthai fixture 이름은 기존 테스트 파일에서 확인 후 매칭 — `tests/unit/adapters/conftest.py` 에 있는 기존 fixture 의 push-frame 헬퍼 사용.) + +- [ ] **Step 2: Run test to verify failure** + +```bash +uv run pytest tests/unit/adapters/test_oak_camera.py::TestRecordingAnchor -v +``` +Expected: fail — `report.recording_anchor is None` (adapter 미적용). + +- [ ] **Step 3: Wire anchor into OAK capture loop** + +`src/syncfield/adapters/oak_camera.py`: + +(a) `start_recording` (line ~405) 가장 시작부분에서 session_clock 전달: + +Before: +```python + def start_recording(self, session_clock: SessionClock) -> None: + """...existing...""" + # ... existing body ... +``` + +After: +```python + def start_recording(self, session_clock: SessionClock) -> None: + """...existing...""" + self._begin_recording_window(session_clock) + # ... existing body ... +``` + +(b) `_capture_loop` (line ~708) 의 frame-arrival 시점에서 첫 프레임 기록. line 727 근처 `device_ts_ns = _device_timestamp_ns(rgb_msg)` 직후: + +Before: +```python + device_ts_ns = _device_timestamp_ns(rgb_msg) + if self._recording: + if self._prev_capture_ns is not None: + self._intervals_ns.append(capture_ns - self._prev_capture_ns) + self._prev_capture_ns = capture_ns +``` + +After: +```python + device_ts_ns = _device_timestamp_ns(rgb_msg) + 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 +``` + +(c) `stop_recording` 끝에서 FinalizationReport 만들 때 `recording_anchor=self._recording_anchor()` 전달: + +Before (예시 — 실제 report 생성 줄 찾아서): +```python + return FinalizationReport( + stream_id=self.id, status=status, frame_count=self._frame_count, + file_path=self._mp4_path if self._mp4_path.exists() else None, + first_sample_at_ns=..., last_sample_at_ns=..., + health_events=self._collected_health, error=error, + jitter_p95_ns=jitter_p95, jitter_p99_ns=jitter_p99, + ) +``` + +After — 필드 추가: +```python + return FinalizationReport( + stream_id=self.id, status=status, frame_count=self._frame_count, + file_path=self._mp4_path if self._mp4_path.exists() else None, + first_sample_at_ns=..., last_sample_at_ns=..., + health_events=self._collected_health, error=error, + jitter_p95_ns=jitter_p95, jitter_p99_ns=jitter_p99, + recording_anchor=self._recording_anchor(), + ) +``` + +- [ ] **Step 4: Run tests to verify pass** + +```bash +uv run pytest tests/unit/adapters/test_oak_camera.py -v +``` +Expected: full oak_camera suite green, including `TestRecordingAnchor`. + +- [ ] **Step 5: Commit** + +```bash +git add src/syncfield/adapters/oak_camera.py tests/unit/adapters/test_oak_camera.py +git commit -m "feat(oak_camera): record intra-host sync anchor on first recording frame" +``` + +--- + +## Task 7: `UVCWebcamStream` — anchor capture (no device clock) + +**Files:** +- Modify: `src/syncfield/adapters/uvc_webcam.py` (`start_recording` ~line 163, capture loop) +- Test: `tests/unit/adapters/test_uvc_webcam.py` + +- [ ] **Step 1: Write failing test** + +패턴은 Task 6 과 동일. 차이점: `first_frame_device_ns is None` 이어야 함 (UVC는 device clock 없음). + +```python +class TestRecordingAnchor: + def test_uvc_anchor_without_device_ts(self, tmp_path, fake_av): + from syncfield.adapters.uvc_webcam import UVCWebcamStream + from syncfield.clock import SessionClock + from syncfield.types import SyncPoint + import time + + stream = UVCWebcamStream("cam", tmp_path, device_index=0) + stream.connect() + sp = SyncPoint.create_now("h") + clock = SessionClock(sync_point=sp, recording_armed_ns=time.monotonic_ns()) + stream.start_recording(clock) + fake_av.push_frame(stream) + time.sleep(0.05) + report = stream.stop_recording() + stream.disconnect() + + assert report.recording_anchor is not None + assert report.recording_anchor.first_frame_device_ns is None + assert report.recording_anchor.first_frame_host_ns >= clock.recording_armed_ns +``` + +- [ ] **Step 2: Run test to verify failure** + +```bash +uv run pytest tests/unit/adapters/test_uvc_webcam.py::TestRecordingAnchor -v +``` + +- [ ] **Step 3: Wire anchor into UVC capture loop** + +`src/syncfield/adapters/uvc_webcam.py`: + +(a) `start_recording` 시작부분에 `self._begin_recording_window(session_clock)` 추가. + +(b) capture loop 의 frame-arrival 지점에서 `self._observe_first_frame(capture_ns, device_ns=None)` 추가. UVC의 `_capture_loop` 위치는 파일 내 `_capture_loop` 함수 찾아서, `capture_ns = time.monotonic_ns()` 직후 `if self._recording:` 블록 안. + +(c) `stop_recording` 의 FinalizationReport 생성부에 `recording_anchor=self._recording_anchor()` 추가. + +- [ ] **Step 4: Run tests to verify pass** + +```bash +uv run pytest tests/unit/adapters/test_uvc_webcam.py -v +``` + +- [ ] **Step 5: Commit** + +```bash +git add src/syncfield/adapters/uvc_webcam.py tests/unit/adapters/test_uvc_webcam.py +git commit -m "feat(uvc_webcam): record intra-host sync anchor (no device clock)" +``` + +--- + +## Task 8: PollingSensorStream + PushSensorStream — anchor capture + +**Files:** +- Modify: `src/syncfield/adapters/polling_sensor.py`, `src/syncfield/adapters/push_sensor.py` +- Test: 기존 센서 테스트 파일 확장 + +- [ ] **Step 1: Write failing test** + +`tests/unit/adapters/test_polling_sensor.py` (기존 파일 혹은 신규): + +```python +def test_polling_sensor_anchor_captured(): + from syncfield.adapters.polling_sensor import PollingSensorStream + from syncfield.clock import SessionClock + from syncfield.types import SyncPoint + import time + + poll_result = {"x": 1.0} + stream = PollingSensorStream( + id="ps", poll_fn=lambda: poll_result, interval_s=0.01, + ) + stream.connect() + sp = SyncPoint.create_now("h") + clock = SessionClock(sync_point=sp, recording_armed_ns=time.monotonic_ns()) + stream.start_recording(clock) + time.sleep(0.05) + report = stream.stop_recording() + stream.disconnect() + assert report.recording_anchor is not None + assert report.recording_anchor.first_frame_host_ns >= clock.recording_armed_ns +``` + +유사 테스트 `test_push_sensor.py` 에 추가. + +- [ ] **Step 2: Run test to verify failure** +- [ ] **Step 3: Wire into both sensor adapters** + +`polling_sensor.py` 와 `push_sensor.py`: + +(a) `start_recording` 에 `self._begin_recording_window(session_clock)` 추가. + +(b) 샘플 emission 지점 (각 파일 내 `_emit_sample` 직전의 `capture_ns` 계산 직후)에 `self._observe_first_frame(capture_ns, device_ns=None)` 추가. + +(c) `stop_recording` FinalizationReport 에 `recording_anchor=self._recording_anchor()` 추가. + +- [ ] **Step 4: Run tests to verify pass** + +```bash +uv run pytest tests/unit/adapters/test_polling_sensor.py tests/unit/adapters/test_push_sensor.py -v +``` + +- [ ] **Step 5: Commit** + +```bash +git add src/syncfield/adapters/polling_sensor.py src/syncfield/adapters/push_sensor.py tests/unit/adapters/test_polling_sensor.py tests/unit/adapters/test_push_sensor.py +git commit -m "feat(sensors): record intra-host sync anchor in generic sensor adapters" +``` + +--- + +## Task 9: Remaining adapters — batch extension + +**Files (같은 패턴 반복):** +- `src/syncfield/adapters/meta_quest.py` — quest clock 있음 → `first_frame_device_ns` 에 quest_native_ns 전달 +- `src/syncfield/adapters/oglo_tactile.py` — BLE timestamp 있으면 전달, 없으면 None +- `src/syncfield/adapters/ble_imu.py` — BLE device ts (있으면) +- `src/syncfield/adapters/host_audio.py` — device clock 없음 → None +- `src/syncfield/adapters/jsonl_file.py` — 파일 기반, armed/first 의미 미약 → 첫 record 의 `host_ns, None` +- `src/syncfield/adapters/meta_quest_camera/` — oak_camera 패턴 참고 +- `src/syncfield/adapters/insta360_go3s/` — 오프라인 ingest 성격이면 skip; 실시간이면 oak 패턴 + +각 어댑터마다 3-step (test → wire → commit) 서브태스크 반복. 구조가 Task 6~8 과 동일하므로 동일 체크리스트 사용. + +- [ ] **Step 1: meta_quest** (test → wire → verify → commit) +- [ ] **Step 2: oglo_tactile** (test → wire → verify → commit) +- [ ] **Step 3: ble_imu** (test → wire → verify → commit) +- [ ] **Step 4: host_audio** (test → wire → verify → commit) +- [ ] **Step 5: jsonl_file** (test → wire → verify → commit) +- [ ] **Step 6: meta_quest_camera** (test → wire → verify → commit) +- [ ] **Step 7: insta360_go3s** — 실시간 streaming adapter 인지 먼저 grep 확인. 오프라인 ingest 전용이면 이 task 는 skip. + +각 step 당 commit message: +``` +feat(): record intra-host sync anchor on first recording frame +``` + +--- + +## Task 10: Integration test — multi-adapter shared anchor + +**Files:** +- Test: `tests/integration/test_anchor_sharing.py` (신규) + +- [ ] **Step 1: Write integration test** + +```python +"""All streams in one session observe the SAME armed_host_ns. + +This is the end-to-end validation that intra-host sync metadata is +wired all the way through SessionOrchestrator → SessionClock → +adapters → FinalizationReport → manifest. +""" +from __future__ import annotations + +import json +import time +from pathlib import Path + +from syncfield.orchestrator import SessionOrchestrator +from syncfield.adapters.polling_sensor import PollingSensorStream + + +def test_all_streams_share_armed_host_ns(tmp_path: Path) -> None: + sess = SessionOrchestrator(host_id="h", output_dir=tmp_path) + sess.add(PollingSensorStream("a", poll_fn=lambda: {"x": 1.0}, interval_s=0.01)) + sess.add(PollingSensorStream("b", poll_fn=lambda: {"y": 2.0}, interval_s=0.01)) + sess.add(PollingSensorStream("c", poll_fn=lambda: {"z": 3.0}, interval_s=0.01)) + sess.connect(); sess.start(); time.sleep(0.1); sess.stop() + + manifest_path = next(tmp_path.rglob("manifest.json")) + manifest = json.loads(manifest_path.read_text()) + anchors = [ + s["recording_anchor"] for s in manifest["streams"] + if s.get("recording_anchor") + ] + assert len(anchors) == 3 + armed_values = {a["armed_host_ns"] for a in anchors} + assert len(armed_values) == 1, ( + f"All streams must share the same armed_host_ns; got {armed_values}" + ) + # Each stream's first frame must arrive after the armed moment + for a in anchors: + assert a["first_frame_host_ns"] >= a["armed_host_ns"] + assert a["first_frame_latency_ns"] >= 0 +``` + +- [ ] **Step 2: Run test to verify it passes** + +```bash +uv run pytest tests/integration/test_anchor_sharing.py -v +``` +Expected: pass (assuming Tasks 5 + 8 are in). + +- [ ] **Step 3: Full regression** + +```bash +uv run pytest tests/ -x --timeout=60 +``` +Expected: entire suite green. + +- [ ] **Step 4: Commit** + +```bash +git add tests/integration/test_anchor_sharing.py +git commit -m "test(integration): verify all streams share a single armed_host_ns" +``` + +--- + +## Task 11: Stability hardening + +**Files:** +- Modify: `src/syncfield/stream.py` (helper 방어 로직 확인) +- Test: `tests/unit/test_stream_base.py` (edge case 추가) + +- [ ] **Step 1: Edge case tests** + +```python +def test_anchor_helper_safe_when_start_recording_not_called(): + """If an adapter emits frames before start_recording (e.g. preview + phase leaking into capture loop), anchor must stay None — not crash.""" + d = _Dummy() + d._observe_first_frame(host_ns=100, device_ns=None) + assert d._recording_anchor() is None + + +def test_anchor_helper_thread_safety_idempotent(): + """Concurrent first-frame observations: the first one wins; the + second returns silently.""" + d = _Dummy() + d._begin_recording_window(_clock(armed_ns=100)) + d._observe_first_frame(host_ns=200, device_ns=None) + d._observe_first_frame(host_ns=300, device_ns=None) + assert d._recording_anchor().first_frame_host_ns == 200 + + +def test_anchor_helper_handles_negative_clock_skew(): + """host_ns can trail armed_host_ns under mock clock / test harness. + Helper must NOT raise — it clamps to armed_host_ns so + RecordingAnchor invariant holds.""" + d = _Dummy() + d._begin_recording_window(_clock(armed_ns=1_000)) + d._observe_first_frame(host_ns=500, device_ns=None) # clock went back + anchor = d._recording_anchor() + assert anchor is not None + assert anchor.first_frame_host_ns == 1_000 # clamped + assert anchor.first_frame_latency_ns == 0 +``` + +- [ ] **Step 2: Run to verify pass (helper already has clamp + idempotence from Task 3)** + +```bash +uv run pytest tests/unit/test_stream_base.py -v +``` + +- [ ] **Step 3: Commit** + +```bash +git add tests/unit/test_stream_base.py +git commit -m "test(stream): harden anchor helper against clock skew and race" +``` + +--- + +## Task 12: Docs + version bump + +**Files:** +- Modify: `README.md` (manifest 섹션이 있으면) +- Modify: `docs/` — intra-host sync 설명에 anchor 섹션 추가 (기존 관련 문서 있으면) +- Modify: `pyproject.toml` (version bump 0.3.21 → 0.3.22) + +- [ ] **Step 1: Bump version** + +`pyproject.toml`: +```toml +version = "0.3.22" +``` + +- [ ] **Step 2: Manifest schema docs** + +기존 docs/ 안에 manifest 설명 파일 있으면 `recording_anchor` 필드 설명 추가. 없으면 skip. + +- [ ] **Step 3: Run full suite once more** + +```bash +uv run pytest tests/ --timeout=60 +``` + +- [ ] **Step 4: Commit** + +```bash +git add pyproject.toml docs/ +git commit -m "chore(release): v0.3.22 — intra-host sync anchor metadata" +``` + +--- + +## Self-Review Checklist + +**Spec coverage:** +- ✅ 공통 armed_host_ns capture & propagation: Task 5 +- ✅ per-adapter first-frame anchor capture: Task 6-9 +- ✅ Metadata surfaces on FinalizationReport + manifest: Tasks 4-5 +- ✅ Backward compat (optional field, legacy adapters untouched): Task 2 default=None +- ✅ Device-ts-less adapters supported: Task 7, 8 (first_frame_device_ns=None) +- ✅ Edge cases: Task 11 (no frame, clock skew, idempotence) +- ✅ End-to-end verification: Task 10 + +**Placeholder scan:** no TBD, no "handle edge cases", all code blocks contain actual code. + +**Type consistency:** `_begin_recording_window`, `_observe_first_frame`, `_recording_anchor`, `recording_anchor`, `armed_host_ns`, `first_frame_host_ns`, `first_frame_device_ns` — consistent across all tasks. + +**Out of scope (intentional):** +- syncfield(sync service) 측 alignment 활용은 별도 plan. 이 plan 은 metadata 기록까지만. +- `SessionClock.with_armed(ns)` 같은 method 대신 `dataclasses.replace` 로 직접 복제 — YAGNI. +- Anchor 의 wall_clock 변환은 downstream 이 기존 `sync_point.wall_clock_ns` + `armed_host_ns - sync_point.monotonic_ns` 로 계산 가능하므로 별도 저장 안 함. diff --git a/pyproject.toml b/pyproject.toml index eb8be72..3b79fda 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/syncfield/adapters/_video_encoder.py b/src/syncfield/adapters/_video_encoder.py index 517635a..39139a1 100644 --- a/src/syncfield/adapters/_video_encoder.py +++ b/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( diff --git a/src/syncfield/adapters/ble_imu.py b/src/syncfield/adapters/ble_imu.py index 0c2ccce..4e7d955 100644 --- a/src/syncfield/adapters/ble_imu.py +++ b/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 diff --git a/src/syncfield/adapters/host_audio.py b/src/syncfield/adapters/host_audio.py index 2b99518..b18bc43 100644 --- a/src/syncfield/adapters/host_audio.py +++ b/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: diff --git a/src/syncfield/adapters/meta_quest.py b/src/syncfield/adapters/meta_quest.py index 85e47aa..70432e4 100644 --- a/src/syncfield/adapters/meta_quest.py +++ b/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 diff --git a/src/syncfield/adapters/meta_quest_camera/stream.py b/src/syncfield/adapters/meta_quest_camera/stream.py index f88e99b..bc37bbe 100644 --- a/src/syncfield/adapters/meta_quest_camera/stream.py +++ b/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 diff --git a/src/syncfield/adapters/oak_camera.py b/src/syncfield/adapters/oak_camera.py index ff67518..997729d 100644 --- a/src/syncfield/adapters/oak_camera.py +++ b/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( diff --git a/src/syncfield/adapters/oglo_tactile.py b/src/syncfield/adapters/oglo_tactile.py index 328c2d5..88bc7c7 100644 --- a/src/syncfield/adapters/oglo_tactile.py +++ b/src/syncfield/adapters/oglo_tactile.py @@ -259,6 +259,7 @@ def start_recording(self, session_clock: SessionClock) -> None: """ 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: @@ -278,6 +279,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: @@ -437,11 +439,16 @@ def _handle_payload(self, payload: bytes) -> None: channels: dict = { name: int(v) for name, v in zip(FINGER_NAMES, values) } - channels["device_timestamp_ns"] = int( + device_ts_ns = int( (timestamp_us + i * _SAMPLE_PERIOD_US) * 1000 ) + channels["device_timestamp_ns"] = device_ts_ns if self._recording: + # MCU hardware clock is interpolated per sample (see + # device_timestamp_ns above); pass it as the anchor's + # device-side timestamp for precise alignment. + self._observe_first_frame(recv_ns, device_ts_ns) if self._first_at is None: self._first_at = recv_ns self._last_at = recv_ns diff --git a/src/syncfield/adapters/polling_sensor.py b/src/syncfield/adapters/polling_sensor.py index f6ce224..dd84fca 100644 --- a/src/syncfield/adapters/polling_sensor.py +++ b/src/syncfield/adapters/polling_sensor.py @@ -126,6 +126,8 @@ def _capture_once(self) -> bool: )) if self._writing: + # Polling sensors have no device clock — pass None for device_ns. + self._observe_first_frame(capture_ns, None) self._write_core.record_sample(capture_ns) elapsed = time.monotonic() - loop_start @@ -153,6 +155,7 @@ def connect(self) -> None: self._thread.start() def start_recording(self, session_clock: SessionClock) -> None: + self._begin_recording_window(session_clock) self._write_core.reset_recording_stats() self._writing = True @@ -170,6 +173,7 @@ def stop_recording(self) -> FinalizationReport: last_sample_at_ns=last_ns, health_events=list(self._collected_health), error=None, + recording_anchor=self._recording_anchor(), ) def disconnect(self) -> None: diff --git a/src/syncfield/adapters/push_sensor.py b/src/syncfield/adapters/push_sensor.py index d5783d6..70e26e6 100644 --- a/src/syncfield/adapters/push_sensor.py +++ b/src/syncfield/adapters/push_sensor.py @@ -54,6 +54,7 @@ def connect(self) -> None: self._on_connect(self) def start_recording(self, session_clock: SessionClock) -> None: + self._begin_recording_window(session_clock) self._write_core.reset_recording_stats() self._writing = True @@ -71,6 +72,7 @@ def stop_recording(self) -> FinalizationReport: last_sample_at_ns=last_at, health_events=list(self._collected_health), error=None, + recording_anchor=self._recording_anchor(), ) def disconnect(self) -> None: @@ -118,4 +120,6 @@ def push( channels=channels, )) if self._writing: + # Push sensors have no device clock — pass None for device_ns. + self._observe_first_frame(capture_ns, None) self._write_core.record_sample(capture_ns) diff --git a/src/syncfield/adapters/uvc_webcam.py b/src/syncfield/adapters/uvc_webcam.py index 114abfd..0861d56 100644 --- a/src/syncfield/adapters/uvc_webcam.py +++ b/src/syncfield/adapters/uvc_webcam.py @@ -162,6 +162,7 @@ def connect(self) -> None: def start_recording(self, session_clock: SessionClock) -> None: """Open the VideoEncoder and flip recording on.""" + 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) @@ -198,6 +199,7 @@ def stop_recording(self) -> FinalizationReport: error=None, jitter_p95_ns=jitter_p95, jitter_p99_ns=jitter_p99, + recording_anchor=self._recording_anchor(), ) def disconnect(self) -> None: @@ -284,6 +286,8 @@ def _capture_loop(self) -> None: frame = next(frame_iter) capture_ns = time.monotonic_ns() if self._recording: + # UVC has no device clock — pass None for device_ns. + self._observe_first_frame(capture_ns, None) if self._prev_capture_ns is not None: self._intervals_ns.append(capture_ns - self._prev_capture_ns) self._prev_capture_ns = capture_ns diff --git a/src/syncfield/clock.py b/src/syncfield/clock.py index 1f2cc37..9b29078 100644 --- a/src/syncfield/clock.py +++ b/src/syncfield/clock.py @@ -28,9 +28,15 @@ class SessionClock: Attributes: sync_point: The session's :class:`SyncPoint` (monotonic + wall clock anchor captured at session start). + recording_armed_ns: Common host monotonic_ns captured by the + orchestrator right before it fans out ``start_recording()`` + to every stream. ``None`` during preview phase, non-``None`` + once recording is armed. All streams receive the same value, + so adapters can use it as a shared intra-host sync anchor. """ sync_point: SyncPoint + recording_armed_ns: int | None = None @property def host_id(self) -> str: diff --git a/src/syncfield/orchestrator.py b/src/syncfield/orchestrator.py index 5378565..1c27628 100644 --- a/src/syncfield/orchestrator.py +++ b/src/syncfield/orchestrator.py @@ -72,6 +72,7 @@ from __future__ import annotations +import dataclasses import logging import threading import time @@ -417,8 +418,21 @@ def __init__( # the right place to fail loudly on port-binding or role-config errors). if self._role is not None: self._bring_multihost_online() - import atexit - atexit.register(self._safe_shutdown) + + # Device-level safety net: register a last-chance teardown that + # runs on interpreter shutdown. The regular path (viewer.close, + # explicit session.disconnect, CLI finally blocks) handles 99% + # of teardown — this catches the 1% where an exception bubbles + # all the way out or a caller forgot. SIGKILL still bypasses + # this (by design), but every cooperative exit path now + # guarantees adapters release their hardware. + # + # Registered unconditionally: single-host sessions also hold + # devices and need the same protection. Multi-host teardown + # (control plane / advertiser / browser) is folded into the + # same callback so there's only one atexit entry per session. + import atexit + atexit.register(self._safe_shutdown) def _bring_multihost_online(self) -> None: """Spin up control plane + advertiser (or follower browser) at construction time. @@ -454,11 +468,48 @@ def _bring_multihost_online(self) -> None: raise def _safe_shutdown(self) -> None: - """atexit-callable wrapper around shutdown() that swallows errors.""" + """atexit-callable last-chance teardown — devices + multi-host. + + Runs at interpreter shutdown. Called AFTER the regular + ``viewer.close`` / ``session.disconnect`` path if the caller + used them; acts as a safety net when they did not. + + Two responsibilities, in order: + + 1. If a recording is still in progress or the session is + holding devices, stop and disconnect so adapters release + their hardware handles. Without this, a crashed script or + an unhandled exception leaves OAK boards booted-and-held — + the exact failure that forces a physical USB replug. + 2. Tear down multi-host machinery (``shutdown()``). + + All exceptions are swallowed: ``atexit`` callbacks must not + raise (it suppresses subsequent handlers and muddies exit + diagnostics). We log at WARNING so operators still see what + went wrong if cleanup hit a snag. + """ try: - self.shutdown() + state = self._state except Exception: - pass + state = None + if state == SessionState.RECORDING: + try: + self.stop() + except Exception as exc: + logger.warning("atexit: session.stop() raised: %s", exc) + try: + state = self._state + except Exception: + state = None + if state in (SessionState.CONNECTED, SessionState.STOPPED): + try: + self.disconnect() + except Exception as exc: + logger.warning("atexit: session.disconnect() raised: %s", exc) + try: + self.shutdown() + except Exception as exc: + logger.warning("atexit: session.shutdown() raised: %s", exc) def shutdown(self) -> None: """Tear down ALL multi-host machinery (control plane + advertiser + browser). @@ -1635,12 +1686,20 @@ def connect(self) -> None: self.health.start() connected: List[Stream] = [] + first_failure: Optional[tuple[str, Exception]] = None for stream in self._streams.values(): self._set_stream_state(stream.id, "connecting") try: stream.prepare() stream.connect() except Exception as exc: + # Strict all-or-nothing: record the failure, stop + # attempting further streams, and let the rollback + # block below release everything we already opened. + # Rationale: leaving a partially-booted rig in place + # (e.g. one OAK booted, the other dead) leaks device + # handles the user can only recover from by SIGKILL + # + USB replug — the exact pattern we're fixing. self._stream_errors[stream.id] = str(exc) self._set_stream_state(stream.id, "failed") stream._emit_health(HealthEvent( @@ -1653,7 +1712,8 @@ def connect(self) -> None: fingerprint=f"{stream.id}:startup-failure", data={"phase": "connect", "outcome": "error", "error": str(exc)}, )) - continue + first_failure = (stream.id, exc) + break connected.append(stream) self._stream_errors.pop(stream.id, None) self._set_stream_state(stream.id, "connected") @@ -1668,15 +1728,22 @@ def connect(self) -> None: data={"phase": "connect", "outcome": "success"}, )) - if not connected: + if first_failure is not None: + # Rollback: release every stream that did connect so no + # adapter is left holding a device handle. LIFO matches + # the convention used elsewhere in the codebase. + _rollback_disconnect_streams(connected) + for s in connected: + self._set_stream_state(s.id, "disconnected") self._transition(SessionState.IDLE) if self._log_writer is not None: self._log_writer.close() self._log_writer = None + failed_id, failed_exc = first_failure raise RuntimeError( - "connect() failed: no streams connected — every adapter raised. " - "Inspect per-stream errors via session._stream_errors." - ) + f"connect() failed at stream {failed_id!r}: {failed_exc}. " + f"All previously-connected streams have been released." + ) from failed_exc self._connected_streams = connected @@ -1864,6 +1931,16 @@ def _tick_with_beep(n: int) -> None: # close path in ``_finalize_streams`` can flush them. self._open_sample_writers() + # Capture a single shared armed_host_ns immediately before + # fanning start_recording out — all streams receive the + # same value via SessionClock.recording_armed_ns, which + # they can use as an intra-host sync anchor in their + # capture loop (see StreamBase._begin_recording_window). + armed_ns = time.monotonic_ns() + self._session_clock = dataclasses.replace( + self._session_clock, recording_armed_ns=armed_ns + ) + recording: List[Stream] = [] try: for stream in self._connected_streams: @@ -2379,6 +2456,16 @@ def _persist_session_artifacts( entry["path"] = str(final.file_path) if final.error is not None: entry["error"] = final.error + # Intra-host sync anchor: common ``armed_host_ns`` plus + # this stream's first-frame timestamps. ``None`` for + # empty recordings or adapters that haven't opted in; + # downstream sync tooling reads ``first_frame_latency_ns`` + # to bias-correct per-adapter pipeline latency. + entry["recording_anchor"] = ( + final.recording_anchor.to_dict() + if final.recording_anchor is not None + else None + ) streams_dict[stream.id] = entry write_manifest( diff --git a/src/syncfield/stream.py b/src/syncfield/stream.py index 0a1fe19..8097935 100644 --- a/src/syncfield/stream.py +++ b/src/syncfield/stream.py @@ -54,6 +54,7 @@ from syncfield.types import ( FinalizationReport, HealthEvent, + RecordingAnchor, SampleEvent, StreamCapabilities, StreamKind, @@ -205,6 +206,11 @@ def __init__( self._sample_callbacks: List[SampleCallback] = [] self._health_callbacks: List[HealthCallback] = [] self._collected_health: List[HealthEvent] = [] + # Intra-host sync anchor — captured once per recording window, + # when the first frame/sample arrives after start_recording(). + self._armed_host_ns: int | None = None + self._first_frame_observed: bool = False + self._anchor: Optional[RecordingAnchor] = None @property def device_key(self) -> Optional[DeviceKey]: @@ -240,6 +246,66 @@ def _emit_health(self, event: HealthEvent) -> None: for cb in self._health_callbacks: cb(event) + # ------------------------------------------------------------------ + # Intra-host sync anchor + # ------------------------------------------------------------------ + # + # Each recording window shares a common ``armed_host_ns`` captured + # by the orchestrator. Adapters call ``_begin_recording_window`` + # from ``start_recording`` (with the received ``SessionClock``) and + # then ``_observe_first_frame`` exactly once from their capture + # loop when the first frame/sample of the recording window arrives. + # The resulting :class:`RecordingAnchor` is attached to the + # stream's :class:`FinalizationReport` by ``stop_recording``. + + def _begin_recording_window(self, session_clock: SessionClock) -> None: + """Reset anchor state and remember the armed host timestamp. + + Safe to call even when ``recording_armed_ns`` is ``None`` + (legacy test harnesses / unit mocks) — the helper becomes a + no-op. + """ + self._armed_host_ns = session_clock.recording_armed_ns + self._first_frame_observed = False + self._anchor = None + + def _observe_first_frame( + self, host_ns: int, device_ns: int | None + ) -> None: + """Capture the anchor exactly once per recording window. + + Subsequent calls are silently ignored. No-op when there is no + ``armed_host_ns`` (preview phase, legacy code path, or clock + that was never armed). + """ + # Single-writer assumption: every adapter has exactly one + # capture loop thread calling this helper, so the + # check-then-set flag pattern below is race-free in practice. + # If an adapter ever calls this from multiple threads, wrap + # the body in a threading.Lock. + if self._first_frame_observed: + return + if self._armed_host_ns is None: + return + # Guard against host clock going backwards under test mocks or + # unusual scheduling — clamp to armed_ns so RecordingAnchor's + # first_frame_host_ns >= armed_host_ns invariant holds. + safe_host = max(host_ns, self._armed_host_ns) + self._anchor = RecordingAnchor( + armed_host_ns=self._armed_host_ns, + first_frame_host_ns=safe_host, + first_frame_device_ns=device_ns, + ) + self._first_frame_observed = True + + def _recording_anchor(self) -> Optional[RecordingAnchor]: + """Return the anchor captured for the current recording window. + + Returns ``None`` if ``_observe_first_frame`` has not been called + yet, or if the current recording window has no armed clock. + """ + return self._anchor + # ------------------------------------------------------------------ # Lifecycle methods — subclasses override. # ------------------------------------------------------------------ diff --git a/src/syncfield/types.py b/src/syncfield/types.py index e97c43c..02d3c48 100644 --- a/src/syncfield/types.py +++ b/src/syncfield/types.py @@ -72,6 +72,54 @@ def to_dict(self) -> dict[str, Any]: } +@dataclass(frozen=True) +class RecordingAnchor: + """Per-stream anchor info captured when recording is armed. + + Captures the common host ``armed_host_ns`` (shared by all streams in + the session) together with the first recorded frame's ``(host_ts, + device_ts)`` pair for this stream. Downstream sync tooling uses the + difference ``first_frame_host_ns - armed_host_ns`` to estimate each + adapter's observed pipeline latency and remove per-adapter bias when + aligning streams. + + Attributes: + armed_host_ns: Common host monotonic_ns captured by the + orchestrator immediately before ``start_recording()`` is + fanned out to streams. Identical across all streams in a + single recording window. + first_frame_host_ns: Host monotonic_ns at which this stream's + first recorded frame arrived on the host. + first_frame_device_ns: Optional device-clock timestamp of the + first recorded frame. ``None`` for adapters without a + device-side clock (UVC webcams, host audio, etc). + """ + + armed_host_ns: int + first_frame_host_ns: int + first_frame_device_ns: int | None = None + + def __post_init__(self) -> None: + if self.first_frame_host_ns < self.armed_host_ns: + raise ValueError( + f"first_frame_host_ns must be >= armed_host_ns; " + f"got armed={self.armed_host_ns}, first={self.first_frame_host_ns}" + ) + + @property + def first_frame_latency_ns(self) -> int: + """Observed latency from armed moment to first frame arrival.""" + return self.first_frame_host_ns - self.armed_host_ns + + def to_dict(self) -> dict[str, Any]: + return { + "armed_host_ns": self.armed_host_ns, + "first_frame_host_ns": self.first_frame_host_ns, + "first_frame_device_ns": self.first_frame_device_ns, + "first_frame_latency_ns": self.first_frame_latency_ns, + } + + @dataclass class FrameTimestamp: """Single timestamp for one data packet (camera frame or sensor sample). @@ -326,6 +374,10 @@ class FinalizationReport: jitter_p99_ns: 99th-percentile inter-frame interval (ns) during the recording window. None if fewer than 20 samples were collected. + recording_anchor: Intra-host sync anchor captured at the start + of the recording window (common ``armed_host_ns`` plus this + stream's first-frame timestamps). ``None`` for empty + recordings or adapters that haven't opted in. """ stream_id: str @@ -339,6 +391,7 @@ class FinalizationReport: jitter_p95_ns: int | None = None jitter_p99_ns: int | None = None incidents: list = field(default_factory=list) + recording_anchor: RecordingAnchor | None = None @dataclass(frozen=True) diff --git a/src/syncfield/viewer/app.py b/src/syncfield/viewer/app.py index 59fe5ed..2c27e2b 100644 --- a/src/syncfield/viewer/app.py +++ b/src/syncfield/viewer/app.py @@ -15,6 +15,7 @@ from __future__ import annotations import logging +import signal import threading import webbrowser from contextlib import contextmanager @@ -23,6 +24,7 @@ import uvicorn from syncfield.orchestrator import SessionOrchestrator +from syncfield.types import SessionState from syncfield.viewer.poller import SessionPoller from syncfield.viewer.server import ViewerServer @@ -79,13 +81,48 @@ def launch( title: Browser tab title. Default ``"SyncField"``. """ app = ViewerApp(session, host=host, port=port, title=title) + + # SIGTERM semantics: Python's default SIGTERM handler terminates + # the process via the C-level default, which does NOT run + # ``finally`` blocks or ``atexit`` handlers. That's how ``kill + # `` used to leave OAK boards booted-and-held even though + # ``launch()`` has an apparently-safe teardown in ``finally``. + # Install a handler that raises ``SystemExit`` instead — that + # unwinds the stack normally, finally runs, ``app.close()`` + # releases every device, and the process exits cleanly. SIGKILL + # still bypasses everything (by design), so the operator needs a + # replug for that case — but every cooperative signal now does + # the right thing. + _prev_sigterm = signal.getsignal(signal.SIGTERM) + + def _on_sigterm(signum, frame): # noqa: ARG001 + raise SystemExit(128 + signum) + + try: + signal.signal(signal.SIGTERM, _on_sigterm) + except ValueError: + # Signal handlers can only be installed from the main thread. + # When launch() is called off the main thread (unusual, but + # e.g. some test harnesses), skip the install — the caller + # gave up on signal-driven cleanup by choosing that thread. + _prev_sigterm = None + try: app.setup() app.run() - except KeyboardInterrupt: + except (KeyboardInterrupt, SystemExit): + # Both cooperative exits — let `finally` tear down devices. + # SystemExit is re-raised after close() so the exit code + # propagates; KeyboardInterrupt is swallowed to match the + # previous CLI ergonomic (no traceback on Ctrl+C). pass finally: app.close() + if _prev_sigterm is not None: + try: + signal.signal(signal.SIGTERM, _prev_sigterm) + except (ValueError, TypeError): # pragma: no cover + pass @contextmanager @@ -199,11 +236,60 @@ def run(self) -> None: self._running = False def close(self) -> None: - """Stop uvicorn, the poller, and tear down the session.""" - # Signal uvicorn to shut down + """Stop uvicorn, the poller, and release device handles. + + In blocking mode (``launch``), the viewer owns the session + lifecycle — if the session is still holding devices when the + viewer shuts down (Ctrl+C, SIGTERM, browser close, crash) it + MUST call ``session.disconnect()`` / ``stop()`` on its way out + or downstream adapters leak hardware handles. Leaving an OAK + booted-and-held like that is the specific failure mode that + forces a physical USB replug to recover, so we take two + best-effort steps: + + 1. If a recording is in progress, stop it so the MP4 is + finalised rather than truncated mid-flight. + 2. If any stream is still connected, disconnect it so the + device handles go back to the OS. + + All per-stream exceptions during teardown are swallowed — the + invariant is "release as much hardware as we can before this + process exits", not "raise the first error we encounter". + ``disconnect()`` itself is idempotent from the caller's + perspective — we only call it while the session is in a state + that allows it. + """ + # Signal uvicorn to shut down first so HTTP requests don't + # race against teardown. Safe if uvicorn already stopped. if self._uvicorn_server is not None: self._uvicorn_server.should_exit = True + # Best-effort session teardown. Order matters: stop() first so + # any in-flight recording finalises cleanly, then disconnect() + # so device handles are released. A failed stop() must NOT + # prevent disconnect() — that's what leaks handles. + try: + state = self._session.state + except Exception: # pragma: no cover — accessor corner cases + state = None + if state == SessionState.RECORDING: + try: + self._session.stop() + except Exception as exc: # pragma: no cover — best-effort + logger.warning("viewer.close: session.stop() raised: %s", exc) + # Re-read state — stop() moves us to STOPPED. + try: + state = self._session.state + except Exception: # pragma: no cover + state = None + if state in (SessionState.CONNECTED, SessionState.STOPPED): + try: + self._session.disconnect() + except Exception as exc: # pragma: no cover — best-effort + logger.warning( + "viewer.close: session.disconnect() raised: %s", exc + ) + self._poller.stop() self._running = False self._setup_done = False diff --git a/tests/integration/test_anchor_sharing.py b/tests/integration/test_anchor_sharing.py new file mode 100644 index 0000000..754bdd4 --- /dev/null +++ b/tests/integration/test_anchor_sharing.py @@ -0,0 +1,52 @@ +"""All streams in one session observe the SAME armed_host_ns. + +This is the end-to-end validation that intra-host sync metadata is +wired all the way through SessionOrchestrator -> SessionClock -> +adapters -> FinalizationReport -> manifest.json. +""" +from __future__ import annotations + +import json +import time +from pathlib import Path + +from syncfield.adapters.polling_sensor import PollingSensorStream +from syncfield.orchestrator import SessionOrchestrator +from syncfield.tone import SyncToneConfig + + +def test_all_streams_share_armed_host_ns(tmp_path: Path) -> None: + """3 polling sensors, run a short session, verify all see same armed_ns.""" + sess = SessionOrchestrator( + host_id="h", + output_dir=tmp_path, + sync_tone=SyncToneConfig.silent(), # disable audio chirp + ) + sess.add(PollingSensorStream("a", read=lambda: {"x": 1.0}, hz=100)) + sess.add(PollingSensorStream("b", read=lambda: {"y": 2.0}, hz=100)) + sess.add(PollingSensorStream("c", read=lambda: {"z": 3.0}, hz=100)) + sess.connect() + sess.start(countdown_s=0) # skip default countdown for test speed + time.sleep(0.1) + sess.stop() + + manifest_path = next(tmp_path.rglob("manifest.json"), None) + assert manifest_path is not None, "manifest.json not written" + manifest = json.loads(manifest_path.read_text()) + streams = manifest.get("streams", {}) + + anchors = [ + entry["recording_anchor"] + for entry in streams.values() + if entry.get("recording_anchor") is not None + ] + assert len(anchors) == 3, f"expected 3 anchors, got {len(anchors)}" + + armed_values = {a["armed_host_ns"] for a in anchors} + assert len(armed_values) == 1, ( + f"All streams must share the same armed_host_ns; got {armed_values}" + ) + + for a in anchors: + assert a["first_frame_host_ns"] >= a["armed_host_ns"] + assert a["first_frame_latency_ns"] >= 0 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 27122b3..0486844 100644 --- a/tests/unit/adapters/meta_quest_camera/test_stream_lifecycle.py +++ b/tests/unit/adapters/meta_quest_camera/test_stream_lifecycle.py @@ -243,3 +243,55 @@ def test_latest_frame_reads_from_preview_consumer(self, tmp_path): # a decoded frame, so the slot stays None. assert stream.latest_frame is None stream.disconnect() + + +# --------------------------------------------------------------------------- +# Intra-host sync anchor +# --------------------------------------------------------------------------- + + +class TestRecordingAnchor: + """Per-recording-window intra-host sync anchor capture. + + Quest camera frames carry both a host-projected capture timestamp + and a Quest-native timestamp (``quest_native_ns``) surfaced on + :class:`MjpegFrame`. The anchor captures both. + """ + + def test_meta_quest_camera_anchor_captured_with_device_ts(self, tmp_path): + stream = MetaQuestCameraStream( + id="quest_cam", + quest_host="test", + output_dir=tmp_path, + _transport=_status_only_transport(), + resolution=(64, 64), + ) + stream.connect() + armed_ns = 1_234_567_890 + clock = SessionClock( + sync_point=SyncPoint.create_now("h"), + recording_armed_ns=armed_ns, + ) + stream.start_recording(clock) + + # Push a single JPEG frame into the consumer's sink — same + # pattern as TestRecording::test_recording_writes_single_mp4_on_frames. + jpeg = _make_jpeg() + host_ns = max(armed_ns + 1_000_000, time.monotonic_ns()) + quest_ns = host_ns - 1_000_000 # arbitrary positive delta + stream._preview._frame_sink( + MjpegFrame( + jpeg_bytes=jpeg, + capture_ns=host_ns, + quest_native_ns=quest_ns, + ) + ) + + report = stream.stop_recording() + stream.disconnect() + + assert report.recording_anchor is not None + assert report.recording_anchor.armed_host_ns == armed_ns + assert report.recording_anchor.first_frame_host_ns >= armed_ns + # KEY: Quest camera surfaces quest_native_ns — anchor captures it. + assert report.recording_anchor.first_frame_device_ns == quest_ns diff --git a/tests/unit/adapters/test_ble_imu.py b/tests/unit/adapters/test_ble_imu.py index 05d721d..0b71ff4 100644 --- a/tests/unit/adapters/test_ble_imu.py +++ b/tests/unit/adapters/test_ble_imu.py @@ -477,6 +477,45 @@ def test_legacy_start_stop_round_trip(self, mock_bleak): assert client.disconnect.await_count >= 1 +# ============================================================================ +# Intra-host sync anchor +# ============================================================================ + + +class TestRecordingAnchor: + """Per-recording-window intra-host sync anchor capture. + + The generic BLE IMU decoder derives per-sample timestamps from the + host's monotonic clock (``recv_ns``), not a sensor-side clock — so + ``first_frame_device_ns`` must stay ``None``. + """ + + def test_ble_imu_anchor_captured_without_device_ts(self, mock_bleak): + from syncfield.adapters.ble_imu import BLEImuGenericStream + + stream = BLEImuGenericStream( + "imu", profile=_simple_profile(mock_bleak), address="x", + ) + armed_ns = 1_234_567_890 + clock = SessionClock( + sync_point=SyncPoint.create_now("h"), + recording_armed_ns=armed_ns, + ) + stream._begin_recording_window(clock) + stream._recording = True # skip the async lifecycle for unit test + + payload = b"\xAA\xBB" + struct.pack("= armed_ns + # KEY: generic BLE IMU has no device clock — stays None. + assert report.recording_anchor.first_frame_device_ns is None + + # ============================================================================ # Optional-dep import guard # ============================================================================ diff --git a/tests/unit/adapters/test_host_audio.py b/tests/unit/adapters/test_host_audio.py index b2ea96d..0b55d56 100644 --- a/tests/unit/adapters/test_host_audio.py +++ b/tests/unit/adapters/test_host_audio.py @@ -200,3 +200,59 @@ def test_metrics_throttled(self, tmp_path: Path): stream._emit_audio_metrics([0.5], 12345) assert len(received) == 0 # Throttled + + +# --------------------------------------------------------------------------- +# Intra-host sync anchor +# --------------------------------------------------------------------------- + + +class TestRecordingAnchor: + """Per-recording-window intra-host sync anchor capture. + + HostAudioStream has no device-side clock (PortAudio/sounddevice + exposes only a host-side capture timestamp), so + ``first_frame_device_ns`` must always be ``None``. + """ + + def test_host_audio_anchor_captured_without_device_ts( + self, tmp_path: Path + ): + import numpy as np + from syncfield.clock import SessionClock, SyncPoint + + stream = HostAudioStream("mic", output_dir=tmp_path) + + # Capture the callback sd.InputStream receives so we can fire it + # manually without PortAudio hardware. + captured_callback = {} + + def _fake_input_stream(**kwargs): + captured_callback["cb"] = kwargs["callback"] + return MagicMock() + + mock_info = {"name": "Test Mic", "max_input_channels": 2} + with patch("sounddevice.query_devices", return_value=mock_info), \ + patch("sounddevice.InputStream", side_effect=_fake_input_stream): + stream.connect() + + armed_ns = 1_234_567_890 + clock = SessionClock( + sync_point=SyncPoint.create_now("h"), + recording_armed_ns=armed_ns, + ) + stream.start_recording(clock) + + # Fire the captured callback with a tiny block of float32 samples. + # The callback will write to wav and observe the first frame. + block = np.zeros((256, 1), dtype=np.float32) + captured_callback["cb"](block, 256, None, None) + + report = stream.stop_recording() + stream.disconnect() + + assert report.recording_anchor is not None + assert report.recording_anchor.armed_host_ns == armed_ns + assert report.recording_anchor.first_frame_host_ns >= armed_ns + # KEY: host mic has no device clock. + assert report.recording_anchor.first_frame_device_ns is None diff --git a/tests/unit/adapters/test_meta_quest.py b/tests/unit/adapters/test_meta_quest.py index d9bc1fa..85b9ee7 100644 --- a/tests/unit/adapters/test_meta_quest.py +++ b/tests/unit/adapters/test_meta_quest.py @@ -414,3 +414,45 @@ def _find_free_port() -> int: with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s: s.bind(("127.0.0.1", 0)) return s.getsockname()[1] + + +# --------------------------------------------------------------------------- +# Intra-host sync anchor +# --------------------------------------------------------------------------- + + +class TestRecordingAnchor: + """Per-recording-window intra-host sync anchor capture. + + MetaQuestHandStream receives packets via WiFi UDP — the current + parser doesn't surface the Quest's device-side ``ts_ms`` as a + separate scalar, so ``first_frame_device_ns`` is expected to stay + ``None``. ``armed_host_ns`` and ``first_frame_host_ns`` are + populated on the first packet that arrives after + ``start_recording``. + """ + + def test_meta_quest_anchor_captured_without_device_ts(self): + from syncfield.clock import SessionClock, SyncPoint + + port = _find_free_port() + stream = MetaQuestHandStream("quest3", port=port) + stream.connect() + armed_ns = time.monotonic_ns() + clock = SessionClock( + sync_point=SyncPoint.create_now("h"), + recording_armed_ns=armed_ns, + ) + stream.start_recording(clock) + + _send_packet(port, _make_quest3_packet()) + time.sleep(0.2) + + report = stream.stop_recording() + stream.disconnect() + + assert report.recording_anchor is not None + assert report.recording_anchor.armed_host_ns == armed_ns + assert report.recording_anchor.first_frame_host_ns >= armed_ns + # KEY: Quest adapter doesn't surface device clock — stays None. + assert report.recording_anchor.first_frame_device_ns is None diff --git a/tests/unit/adapters/test_oak_camera.py b/tests/unit/adapters/test_oak_camera.py index 9667861..acf934e 100644 --- a/tests/unit/adapters/test_oak_camera.py +++ b/tests/unit/adapters/test_oak_camera.py @@ -471,3 +471,87 @@ def test_get_timestamp_returning_none_returns_none(self, mock_depthai): msg.getTimestamp.return_value = None assert helper(msg) is None + + +class TestRecordingAnchor: + """Per-recording-window intra-host sync anchor capture. + + OAK cameras expose a real on-device shutter timestamp, so the + anchor's ``first_frame_device_ns`` is expected to be populated + (non-``None``) on the very first frame that arrives after + ``start_recording()``. + """ + + def test_anchor_captured_from_first_frame( + self, mock_depthai, mock_av_generous, tmp_path, monkeypatch + ): + """After arming the clock and pushing frames through the fake + pipeline, ``stop_recording`` returns a :class:`FinalizationReport` + whose ``recording_anchor`` captures the armed host ns, the host + arrival ns of the first recorded frame, and the device-clock + timestamp of that frame.""" + from syncfield.adapters import oak_camera + from syncfield.adapters.oak_camera import OakCameraStream + + # Force a known, non-None device timestamp on every frame so the + # assertion on ``first_frame_device_ns`` is unambiguous. The + # fake ``_FakeFrame`` / ``_FakeEncodedPacket`` payloads used by + # the shared fixture don't define ``getTimestamp``, so the real + # helper would return ``None``. + known_device_ns = 42_000_000_000 + monkeypatch.setattr( + oak_camera, "_device_timestamp_ns", lambda msg: known_device_ns + ) + + armed_ns = 1_234_567_890 + clock = SessionClock( + sync_point=SyncPoint.create_now("h"), + recording_armed_ns=armed_ns, + ) + + stream = OakCameraStream("oak", output_dir=tmp_path) + stream.prepare() + stream.connect() + stream.start_recording(clock) + # Let the capture thread drain at least one frame out of the + # unlimited fake queue. + time.sleep(0.15) + report = stream.stop_recording() + stream.disconnect() + + assert report.recording_anchor is not None + assert report.recording_anchor.armed_host_ns == armed_ns + assert report.recording_anchor.first_frame_host_ns >= armed_ns + assert report.recording_anchor.first_frame_device_ns == known_device_ns + + def test_no_anchor_when_no_frames_arrive( + self, mock_depthai, mock_av_generous, tmp_path + ): + """If ``start_recording`` is called but zero frames arrive before + ``stop_recording``, the report's ``recording_anchor`` stays + ``None``.""" + from syncfield.adapters.oak_camera import OakCameraStream + + armed_ns = 9_876_543_210 + clock = SessionClock( + sync_point=SyncPoint.create_now("h"), + recording_armed_ns=armed_ns, + ) + + stream = OakCameraStream("oak", output_dir=tmp_path) + stream.prepare() + stream.connect() + # Starve the capture loop so the recording window sees no frames. + # ``_safe_get_rgb`` swallows exceptions and returns ``None``, so + # raising from ``get`` keeps the loop spinning without producing + # a frame. Set BEFORE ``start_recording`` and give the capture + # thread a beat to settle on the new side_effect so no in-flight + # frame slips past the ``_recording`` flag flip. + stream._q_rgb.get.side_effect = RuntimeError("starved") + time.sleep(0.05) + stream.start_recording(clock) + time.sleep(0.1) + report = stream.stop_recording() + stream.disconnect() + + assert report.recording_anchor is None diff --git a/tests/unit/adapters/test_oglo_tactile.py b/tests/unit/adapters/test_oglo_tactile.py index 563e0ee..9126f31 100644 --- a/tests/unit/adapters/test_oglo_tactile.py +++ b/tests/unit/adapters/test_oglo_tactile.py @@ -260,3 +260,46 @@ def test_appears_in_syncfield_adapters(self, mock_bleak): importlib.reload(adapters) assert "OgloTactileStream" in adapters.__all__ assert hasattr(adapters, "OgloTactileStream") + + +class TestRecordingAnchor: + """Per-recording-window intra-host sync anchor capture. + + OGLO gloves expose an MCU hardware clock (interpolated per-sample), + so ``first_frame_device_ns`` is expected to be populated (non-None) + alongside ``armed_host_ns`` and ``first_frame_host_ns``. + """ + + def test_oglo_anchor_captured_with_device_ts(self, mock_bleak): + from syncfield.adapters.oglo_tactile import OgloTactileStream + from syncfield.types import SyncPoint + + stream = OgloTactileStream("tactile_right", address="m") + # Skip the async BLE lifecycle for this unit test — we drive the + # decode path synchronously via _dispatch_notification_for_test. + # Simulate what ``start_recording`` does: prime the anchor state + # and flip ``_recording``. + armed_ns = 1_234_567_890 + clock = SessionClock( + sync_point=SyncPoint.create_now("h"), + recording_armed_ns=armed_ns, + ) + stream._begin_recording_window(clock) + stream._recording = True + + # Known MCU timestamp → first sample's device_ts_ns is + # timestamp_us * 1000 (i + 0). + timestamp_us = 12_345_000 + samples = [(1, 2, 3, 4, 5)] * 3 + packet = _build_packet(count=3, timestamp_us=timestamp_us, samples=samples) + stream._dispatch_notification_for_test(packet) + + report = stream.stop_recording() + + assert report.recording_anchor is not None + assert report.recording_anchor.armed_host_ns == armed_ns + assert report.recording_anchor.first_frame_host_ns >= armed_ns + # KEY: OGLO exposes MCU hardware clock — anchor carries it. + assert ( + report.recording_anchor.first_frame_device_ns == timestamp_us * 1000 + ) diff --git a/tests/unit/adapters/test_polling_sensor.py b/tests/unit/adapters/test_polling_sensor.py index 42d6104..2549318 100644 --- a/tests/unit/adapters/test_polling_sensor.py +++ b/tests/unit/adapters/test_polling_sensor.py @@ -324,3 +324,40 @@ def test_stop_recording_returns_finalization_report(): def test_polling_sensor_stream_is_re_exported_from_adapters_package(): from syncfield.adapters import PollingSensorStream as Reexported assert Reexported is PollingSensorStream + + +# --------------------------------------------------------------------------- +# Intra-host sync anchor +# --------------------------------------------------------------------------- + +class TestRecordingAnchor: + """Per-recording-window intra-host sync anchor capture. + + PollingSensorStream has no device clock — ``first_frame_device_ns`` + must always be ``None``, while ``armed_host_ns`` and + ``first_frame_host_ns`` are populated on the first recorded sample. + """ + + def test_polling_sensor_anchor_captured_without_device_ts(self): + """Polling sensor has no device clock — anchor captured with + first_frame_device_ns=None.""" + stream = PollingSensorStream( + "imu", read=lambda: {"x": 1.0}, hz=1000, + ) + stream.connect() + armed_ns = time.monotonic_ns() + clock = SessionClock( + sync_point=SyncPoint.create_now("h"), + recording_armed_ns=armed_ns, + ) + stream.start_recording(clock) + # Let the capture thread produce at least one recorded sample. + time.sleep(0.05) + report = stream.stop_recording() + stream.disconnect() + + assert report.recording_anchor is not None + assert report.recording_anchor.armed_host_ns == armed_ns + assert report.recording_anchor.first_frame_host_ns >= armed_ns + # KEY: polling sensors have no device clock. + assert report.recording_anchor.first_frame_device_ns is None diff --git a/tests/unit/adapters/test_push_sensor.py b/tests/unit/adapters/test_push_sensor.py index 30e438a..895de4f 100644 --- a/tests/unit/adapters/test_push_sensor.py +++ b/tests/unit/adapters/test_push_sensor.py @@ -259,3 +259,37 @@ def boom(capture_ns): def test_push_sensor_stream_is_re_exported_from_adapters_package(): from syncfield.adapters import PushSensorStream as Reexported assert Reexported is PushSensorStream + + +# --------------------------------------------------------------------------- +# Intra-host sync anchor +# --------------------------------------------------------------------------- + +class TestRecordingAnchor: + """Per-recording-window intra-host sync anchor capture. + + PushSensorStream has no device clock — ``first_frame_device_ns`` + must always be ``None``, while ``armed_host_ns`` and + ``first_frame_host_ns`` are populated on the first recorded sample. + """ + + def test_push_sensor_anchor_captured_without_device_ts(self): + """Push sensor has no device clock — anchor captured with + first_frame_device_ns=None.""" + stream = PushSensorStream("ble") + stream.connect() + armed_ns = time.monotonic_ns() + clock = SessionClock( + sync_point=SyncPoint.create_now("h"), + recording_armed_ns=armed_ns, + ) + stream.start_recording(clock) + stream.push({"ax": 0.5}) + report = stream.stop_recording() + stream.disconnect() + + assert report.recording_anchor is not None + assert report.recording_anchor.armed_host_ns == armed_ns + assert report.recording_anchor.first_frame_host_ns >= armed_ns + # KEY: push sensors have no device clock. + assert report.recording_anchor.first_frame_device_ns is None diff --git a/tests/unit/adapters/test_uvc_webcam.py b/tests/unit/adapters/test_uvc_webcam.py index de83f7a..eacc4da 100644 --- a/tests/unit/adapters/test_uvc_webcam.py +++ b/tests/unit/adapters/test_uvc_webcam.py @@ -286,3 +286,43 @@ def __next__(self): assert any( "No such device" in (h.detail or "") for h in collected ), f"expected fatal OSError in health events, got {collected!r}" + + +class TestRecordingAnchor: + """Per-recording-window intra-host sync anchor capture. + + UVC webcams have no device clock — ``first_frame_device_ns`` must + always be ``None``, while ``armed_host_ns`` and + ``first_frame_host_ns`` are populated on the first recorded frame. + """ + + def test_uvc_anchor_captured_without_device_ts( + self, mock_av_generous, tmp_path + ): + """UVC webcam has no device clock — anchor has armed_ns and + first_frame_host_ns but first_frame_device_ns is None.""" + from syncfield.adapters.uvc_webcam import UVCWebcamStream + + armed_ns = 1_234_567_890 + clock = SessionClock( + sync_point=SyncPoint.create_now("h"), + recording_armed_ns=armed_ns, + ) + + stream = UVCWebcamStream( + "cam", device_index=0, output_dir=tmp_path, fps=30.0 + ) + stream.prepare() + stream.connect() + stream.start_recording(clock) + # Let the capture thread drain at least one frame out of the + # paced fake iterator. + time.sleep(0.1) + report = stream.stop_recording() + stream.disconnect() + + assert report.recording_anchor is not None + assert report.recording_anchor.armed_host_ns == armed_ns + assert report.recording_anchor.first_frame_host_ns >= armed_ns + # KEY DIFFERENCE from OAK: UVC has no device clock. + assert report.recording_anchor.first_frame_device_ns is None diff --git a/tests/unit/test_clock.py b/tests/unit/test_clock.py index e6ac742..15c420f 100644 --- a/tests/unit/test_clock.py +++ b/tests/unit/test_clock.py @@ -44,3 +44,22 @@ def test_session_clock_is_frozen_dataclass(): assert dataclasses.is_dataclass(clock) with pytest.raises(dataclasses.FrozenInstanceError): clock.sync_point = SyncPoint.create_now("other") # type: ignore[misc] + + +def test_session_clock_preview_phase_has_no_armed_ns(): + clock = _make_clock() + assert clock.recording_armed_ns is None + + +def test_session_clock_arm_returns_new_clock_with_armed_ns(): + clock = _make_clock() + armed = dataclasses.replace(clock, recording_armed_ns=12_345) + assert armed.recording_armed_ns == 12_345 + assert clock.recording_armed_ns is None # original unchanged + + +def test_session_clock_armed_ns_survives_frozen_semantics(): + clock = _make_clock() + armed = dataclasses.replace(clock, recording_armed_ns=500) + with pytest.raises(dataclasses.FrozenInstanceError): + armed.recording_armed_ns = 700 # type: ignore[misc] diff --git a/tests/unit/test_orchestrator_anchor.py b/tests/unit/test_orchestrator_anchor.py new file mode 100644 index 0000000..8c23d5f --- /dev/null +++ b/tests/unit/test_orchestrator_anchor.py @@ -0,0 +1,106 @@ +"""Orchestrator fans out a single armed_host_ns to all streams.""" +from __future__ import annotations + +import json +import time +from pathlib import Path + +from syncfield.clock import SessionClock +from syncfield.orchestrator import SessionOrchestrator +from syncfield.stream import StreamBase +from syncfield.tone import SyncToneConfig +from syncfield.types import ( + FinalizationReport, + RecordingAnchor, + StreamCapabilities, +) + + +class _CaptureClockStream(StreamBase): + """Records the SessionClock passed to start_recording().""" + + def __init__(self, id: str) -> None: + super().__init__(id, "sensor", StreamCapabilities()) + self.received_clock: SessionClock | None = None + + def connect(self) -> None: + pass + + def start_recording(self, session_clock: SessionClock) -> None: + self.received_clock = session_clock + + def stop_recording(self) -> FinalizationReport: + anchor = None + if self.received_clock and self.received_clock.recording_armed_ns: + anchor = RecordingAnchor( + armed_host_ns=self.received_clock.recording_armed_ns, + first_frame_host_ns=self.received_clock.recording_armed_ns + 1_000, + first_frame_device_ns=None, + ) + return FinalizationReport( + stream_id=self.id, + status="completed", + frame_count=1, + file_path=None, + first_sample_at_ns=0, + last_sample_at_ns=0, + health_events=[], + error=None, + recording_anchor=anchor, + ) + + def disconnect(self) -> None: + pass + + +def _mk_session(tmp_path: Path) -> SessionOrchestrator: + """Build a silent-chirp orchestrator ready for tests.""" + return SessionOrchestrator( + host_id="h", + output_dir=tmp_path, + sync_tone=SyncToneConfig.silent(), + ) + + +def test_orchestrator_arms_clock_and_all_streams_see_same_armed_ns( + tmp_path: Path, +) -> None: + sess = _mk_session(tmp_path) + a = _CaptureClockStream("a") + b = _CaptureClockStream("b") + sess.add(a) + sess.add(b) + sess.connect() + sess.start(countdown_s=0) + time.sleep(0.01) + sess.stop() + + assert a.received_clock is not None and b.received_clock is not None + assert a.received_clock.recording_armed_ns is not None + assert ( + a.received_clock.recording_armed_ns + == b.received_clock.recording_armed_ns + ) + + +def test_orchestrator_manifest_includes_per_stream_anchor(tmp_path: Path) -> None: + sess = _mk_session(tmp_path) + a = _CaptureClockStream("a") + sess.add(a) + sess.connect() + sess.start(countdown_s=0) + time.sleep(0.01) + sess.stop() + + manifest_paths = list(tmp_path.rglob("manifest.json")) + assert manifest_paths, "manifest.json not written" + manifest = json.loads(manifest_paths[0].read_text()) + streams = manifest.get("streams", {}) + # The orchestrator writes streams as a dict keyed by stream_id. + assert "a" in streams, f"stream 'a' not in manifest streams: {streams!r}" + a_entry = streams["a"] + assert "recording_anchor" in a_entry + assert a_entry["recording_anchor"] is not None + assert "armed_host_ns" in a_entry["recording_anchor"] + assert "first_frame_host_ns" in a_entry["recording_anchor"] + assert "first_frame_latency_ns" in a_entry["recording_anchor"] diff --git a/tests/unit/test_stream_base.py b/tests/unit/test_stream_base.py new file mode 100644 index 0000000..b24c756 --- /dev/null +++ b/tests/unit/test_stream_base.py @@ -0,0 +1,97 @@ +import pytest + +from syncfield.clock import SessionClock +from syncfield.stream import StreamBase +from syncfield.types import StreamCapabilities, SyncPoint + + +class _Dummy(StreamBase): + def __init__(self) -> None: + super().__init__("d", "sensor", StreamCapabilities()) + + +def _clock(armed_ns: int | None = None) -> SessionClock: + sp = SyncPoint.create_now(host_id="h") + return SessionClock(sync_point=sp, recording_armed_ns=armed_ns) + + +def test_anchor_helper_returns_none_before_first_frame(): + d = _Dummy() + d._begin_recording_window(_clock(armed_ns=100)) + assert d._recording_anchor() is None + + +def test_anchor_helper_captures_first_frame_then_ignores_later(): + d = _Dummy() + d._begin_recording_window(_clock(armed_ns=100)) + d._observe_first_frame(host_ns=250, device_ns=9_000) + d._observe_first_frame(host_ns=300, device_ns=10_000) # ignored + anchor = d._recording_anchor() + assert anchor is not None + assert anchor.armed_host_ns == 100 + assert anchor.first_frame_host_ns == 250 + assert anchor.first_frame_device_ns == 9_000 + + +def test_anchor_helper_without_device_ts(): + d = _Dummy() + d._begin_recording_window(_clock(armed_ns=100)) + d._observe_first_frame(host_ns=250, device_ns=None) + anchor = d._recording_anchor() + assert anchor is not None + assert anchor.first_frame_device_ns is None + + +def test_anchor_helper_noop_if_armed_ns_missing(): + d = _Dummy() + d._begin_recording_window(_clock(armed_ns=None)) + d._observe_first_frame(host_ns=250, device_ns=None) + assert d._recording_anchor() is None + + +def test_anchor_helper_reset_on_second_recording_window(): + d = _Dummy() + d._begin_recording_window(_clock(armed_ns=100)) + d._observe_first_frame(host_ns=250, device_ns=9_000) + d._begin_recording_window(_clock(armed_ns=1_000)) + assert d._recording_anchor() is None # reset on new window + d._observe_first_frame(host_ns=1_100, device_ns=500) + anchor = d._recording_anchor() + assert anchor is not None + assert anchor.armed_host_ns == 1_000 + assert anchor.first_frame_host_ns == 1_100 + + +def test_anchor_helper_safe_when_start_recording_not_called(): + """If an adapter emits frames before start_recording (e.g. preview + phase leaking into the capture loop), anchor must stay None — not + crash.""" + d = _Dummy() + d._observe_first_frame(host_ns=100, device_ns=None) + assert d._recording_anchor() is None + + +def test_anchor_helper_idempotent_on_repeated_first_frame(): + """Two consecutive first-frame observations within one recording + window: the first wins, the second returns silently.""" + d = _Dummy() + d._begin_recording_window(_clock(armed_ns=100)) + d._observe_first_frame(host_ns=200, device_ns=None) + d._observe_first_frame(host_ns=300, device_ns=None) + anchor = d._recording_anchor() + assert anchor is not None + assert anchor.first_frame_host_ns == 200 # second call was ignored + + +def test_anchor_helper_clamps_negative_clock_skew(): + """host_ns can trail armed_host_ns under mock clock / test harness. + Helper must NOT raise — it clamps to armed_host_ns so + RecordingAnchor's first_frame_host_ns >= armed_host_ns invariant + holds.""" + d = _Dummy() + d._begin_recording_window(_clock(armed_ns=1_000)) + d._observe_first_frame(host_ns=500, device_ns=None) # clock went back + anchor = d._recording_anchor() + assert anchor is not None + assert anchor.first_frame_host_ns == 1_000 # clamped + assert anchor.first_frame_latency_ns == 0 diff --git a/tests/unit/test_types.py b/tests/unit/test_types.py index 1b26bf8..54f29a5 100644 --- a/tests/unit/test_types.py +++ b/tests/unit/test_types.py @@ -1,10 +1,11 @@ """Tests for syncfield.types.""" +import dataclasses import pytest from dataclasses import FrozenInstanceError from pathlib import Path -from syncfield.types import FrameTimestamp, SensorSample, SyncPoint +from syncfield.types import FrameTimestamp, RecordingAnchor, SensorSample, SyncPoint from syncfield.health.severity import Severity @@ -175,6 +176,52 @@ def test_sensor_sample_nested_round_trip(): assert restored.channels["gestures"]["pinch"] == 0.95 +# --- RecordingAnchor tests --- + + +def test_recording_anchor_with_device_ts(): + anchor = RecordingAnchor( + armed_host_ns=1_000_000_000, + first_frame_host_ns=1_044_000_000, + first_frame_device_ns=9_876_543_210, + ) + assert anchor.first_frame_latency_ns == 44_000_000 + assert anchor.to_dict() == { + "armed_host_ns": 1_000_000_000, + "first_frame_host_ns": 1_044_000_000, + "first_frame_device_ns": 9_876_543_210, + "first_frame_latency_ns": 44_000_000, + } + + +def test_recording_anchor_without_device_ts(): + anchor = RecordingAnchor( + armed_host_ns=1_000, + first_frame_host_ns=1_044_000_000, + ) + assert anchor.first_frame_device_ns is None + d = anchor.to_dict() + assert d["first_frame_device_ns"] is None + assert d["first_frame_latency_ns"] == 1_044_000_000 - 1_000 + + +def test_recording_anchor_rejects_first_before_armed(): + with pytest.raises(ValueError, match="first_frame_host_ns must be >= armed_host_ns"): + RecordingAnchor(armed_host_ns=100, first_frame_host_ns=50) + + +def test_recording_anchor_is_frozen(): + anchor = RecordingAnchor(armed_host_ns=0, first_frame_host_ns=0) + with pytest.raises(dataclasses.FrozenInstanceError): + anchor.armed_host_ns = 1 # type: ignore[misc] + + +def test_recording_anchor_allows_zero_latency(): + """first_frame_host_ns == armed_host_ns is legal (zero-latency case).""" + anchor = RecordingAnchor(armed_host_ns=1_000, first_frame_host_ns=1_000) + assert anchor.first_frame_latency_ns == 0 + + from syncfield.types import ( ChirpSpec, FinalizationReport, @@ -361,3 +408,24 @@ def test_minimal(self): assert report.host_id == "rig_01" assert report.finalizations == [] + +def test_finalization_report_with_anchor(): + anchor = RecordingAnchor( + armed_host_ns=100, first_frame_host_ns=150, first_frame_device_ns=42 + ) + report = FinalizationReport( + stream_id="s1", status="completed", frame_count=10, + file_path=None, first_sample_at_ns=150, last_sample_at_ns=450, + health_events=[], error=None, recording_anchor=anchor, + ) + assert report.recording_anchor is anchor + + +def test_finalization_report_anchor_defaults_to_none(): + report = FinalizationReport( + stream_id="s2", status="completed", frame_count=0, + file_path=None, first_sample_at_ns=None, last_sample_at_ns=None, + health_events=[], error=None, + ) + assert report.recording_anchor is None + diff --git a/uv.lock b/uv.lock index ff85d05..f79348a 100644 --- a/uv.lock +++ b/uv.lock @@ -2330,7 +2330,7 @@ wheels = [ [[package]] name = "syncfield" -version = "0.3.19" +version = "0.3.22" source = { editable = "." } dependencies = [ { name = "av", version = "15.1.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" },