diff --git a/.gitignore b/.gitignore index 39b59ea057..8fbc462026 100644 --- a/.gitignore +++ b/.gitignore @@ -95,3 +95,7 @@ tmp/ !/examples/mujoco_xr/python/isaacteleop_examples/mujoco_xr/assets/follower/follower_arm.xml # Wheel SBOM and license evidence written by the collector in scripts/sbom/ /sbom/ +examples/egocap/recordings/ + +# Agent-only plans and status notes; never checked in. +/agentPlans/ diff --git a/examples/egocap/LIVEKIT_PREVIEW.md b/examples/egocap/LIVEKIT_PREVIEW.md index 30952ee889..d4f829df6a 100644 --- a/examples/egocap/LIVEKIT_PREVIEW.md +++ b/examples/egocap/LIVEKIT_PREVIEW.md @@ -26,15 +26,17 @@ livekit: url: wss://LIVEKIT_HOST room: sensing-preview publisher: orin-sensing - timestamp_basis: capture_unix_us + timestamp_basis: monotonic_us streams: - {name: Head left, track: sensing-left, width: 1280, height: 720, fps: 30} ``` - `track` must equal the name the publisher gives the track (the emulator's `livekit=NAME`). - `publisher` must equal the identity in the publisher's join token. -- Preview works with any `timestamp_basis`. Recording needs `capture_unix_us` and `width`, `height` and - `fps`; recording is refused while any offered stream lacks them. +- `timestamp_basis` (default `monotonic_us`) must equal what the publisher stamps; the values are listed in the + [streams file reference](capture_server/README.md#the---livekit-streams-file). Preview works with any of them. + Recording needs a monotonic basis, `monotonic_us` or `capture_monotonic_us`, and `width`, `height` and `fps`; + recording is refused while any offered stream lacks them. Start CaptureServer with the file (add `--mock` for synthetic tracking): @@ -67,8 +69,10 @@ uv run python tools/livekit_orin_emulator.py \ --width 1280 --height 720 --fps 30 ``` -`--add-stream sensor=N,livekit=NAME` is repeatable. `--timestamp-basis` takes `capture_unix_us` (default), -`processing_unix_us` or `monotonic_us`; set the same value in the streams file. Frames go through a software +`--add-stream sensor=N,livekit=NAME` is repeatable. `--timestamp-basis` takes `monotonic_us` (default), +`capture_monotonic_us`, `capture_unix_us` or `processing_unix_us`; set the same value in the streams file. The Orin's +sender (`--timestamp-basis=` on the sensing app) takes `monotonic_us` (default), `capture_monotonic_us` or +`processing_unix_us`. Both publishers answer the PC's clock-sync requests, which recording relies on. Frames go through a software encoder, so this checks the PC side and the metadata, not the Jetson encoder. ## 4. Preview and record on the PC @@ -80,9 +84,20 @@ export LIVEKIT_API_KEY=... LIVEKIT_API_SECRET=... uv run uvicorn egocap_web.app:app --app-dir backend --port 8000 ``` -- The browser asks `GET /api/livekit-token` for a subscribe-only token, only for a room the current session - offers, and shows the stream in the camera grid. Without the key pair the backend answers 503 and the page +- The browser asks `GET /api/livekit-token` for a viewer token, only for a room the current session + offers, and shows the stream in the camera grid. The token can subscribe and send data messages (the clock-sync + exchange needs them) but cannot publish media. The route is unauthenticated, so anyone who can reach the + backend can send data messages into the room, so the publisher must act only on well-formed + `isaaccapture.clock-sync` requests. Both publishers do: a packet is answered only if its topic is + `isaaccapture.clock-sync` and it is exactly a 13-byte request starting with `Q`; everything else is ignored, and a + reply goes only to the requester. The native sender keeps at most 16 pending requests and drops the oldest when + more arrive. Neither has a per-second rate limit, and the emulator has no bound on pending replies, so keep the + LiveKit room and the backend on a trusted network. Without the key pair the backend answers 503 and the page reports the preview as unavailable. -- Recording joins the room with its own subscribe-only token signed with the same key pair. It starts only +- Recording joins the room with its own viewer-capability token signed with the same key pair. It starts only when every offered stream is recordable (see step 1) and stops with a reason if a frame cannot be recorded - honestly. The backend must run on the machine whose clock CaptureServer reports. + honestly. The backend must run on the machine whose clock CaptureServer reports: recording is refused unless + CaptureServer's URL is a loopback address. +- Recording maps the publisher's monotonic timestamps onto the PC's clock by sending the publisher clock-sync + requests (data messages on topic `isaaccapture.clock-sync`) and reading its replies. The publisher must stay + in the room and answer them; without replies the mapping goes stale. diff --git a/examples/egocap/backend/README.md b/examples/egocap/backend/README.md index ec58128719..c2241b630e 100644 --- a/examples/egocap/backend/README.md +++ b/examples/egocap/backend/README.md @@ -33,7 +33,7 @@ or restarts it. | `GET`/`PUT /api/operator-settings` | The operator's station setup: the camera type (`sensing_gmsl` or `oak_d`), camera roles (`ego`, `wrist_left`, `wrist_right`), the task catalog the recording panel's picker offers (`task_file`) and the task it preselects (`preselected_task`). Kept in `.egocap/operator_settings.json` under the recording output (hidden from the recordings browser) (`EGOCAP_OPERATOR_SETTINGS_FILE` overrides the path) so every browser sees the same values. `PUT` names only the fields it changes and uses `X-Control-Token`; legacy JSON `control_token` remains accepted. A wrong token is 403 and bodies above 16 KiB are rejected with 413 before parsing. | | `POST /api/task-files/import`, `DELETE /api/task-files/{name}` | Import a task catalog the operator supplies — by `name` + `content` (text uploaded from the browser) — into the imported-catalog directory beside the operator settings, once it parses; and remove an imported one. `POST` and `DELETE` use the `X-Control-Token` header (legacy JSON tokens remain accepted for `POST`); `DELETE` refuses removal of the selected catalog (409); imports also refuse replacement of that catalog until it is deselected. Imports appear in `GET /api/task-files` (as `imported`) and shadow a shipped catalog of the same name. | | `GET /api/tools` | The operator tools the Settings drawer offers (headset calibration, glove calibration, fisheye rectification for the SENSING cameras), each with whether this build can launch it. None can yet. | -| `GET /api/livekit-token?room=ROOM` | Subscribe-only viewer token for a LiveKit room the session offers (404 otherwise, 503 without signing credentials). Which streams exist comes from CaptureServer's session message. | +| `GET /api/livekit-token?room=ROOM` | Viewer token (subscribe and data messages, no media publishing) for a LiveKit room the session offers (404 otherwise, 503 without signing credentials). Which streams exist comes from CaptureServer's session message. | | `GET /api/task-files`, `GET /api/task-files/{name}` | Read-only task-catalog API; see [`../data/taskfiles/README.md`](../data/taskfiles/README.md). | | `GET /recordings/...` | The optional recordings browser's API and file routes, off unless `EGOCAP_RECORDINGS_ENABLED=1`; see [`../recordings_frontend/README.md`](../recordings_frontend/README.md). | @@ -55,7 +55,7 @@ Everything else is an environment variable: | --- | --- | --- | | `EGOCAP_STATION_CONFIG` | unset (development default) | Path to the deployment station YAML described above. | | `EGOCAP_CAPTURE_SERVER_URL` | `ws://127.0.0.1:8082` | Where to connect for tracking/camera data. | -| `LIVEKIT_API_KEY`, `LIVEKIT_API_SECRET` | unset | The LiveKit server's key pair. The backend signs two subscribe-only tokens with it: viewer tokens for `GET /api/livekit-token`, and the recorder's own token. LiveKit recording will not start without both variables; see [`LIVEKIT_PREVIEW.md`](../LIVEKIT_PREVIEW.md). | +| `LIVEKIT_API_KEY`, `LIVEKIT_API_SECRET` | unset | The LiveKit server's key pair. The backend signs two viewer tokens with it (subscribe and data messages, no media publishing): viewer tokens for `GET /api/livekit-token`, and the recorder's own token. LiveKit recording will not start without both variables; see [`LIVEKIT_PREVIEW.md`](../LIVEKIT_PREVIEW.md). | | `EGOCAP_TASK_FILES_DIR` | the shipped `data/taskfiles/` | Which task-catalog directory the task-file API serves. | | `EGOCAP_RECORDINGS_ENABLED` | off | Turns on the recordings browser's routes. | | `EGOCAP_ALLOWED_ORIGINS` | none (same-origin only) | Comma-separated extra origins allowed to open `WS /api/session` — needed when the frontend is hosted separately or behind a reverse proxy. | diff --git a/examples/egocap/backend/egocap_web/app.py b/examples/egocap/backend/egocap_web/app.py index 8d0478220f..e8072fb891 100644 --- a/examples/egocap/backend/egocap_web/app.py +++ b/examples/egocap/backend/egocap_web/app.py @@ -152,7 +152,7 @@ def _station_calibration_assets() -> tuple[CalibrationAsset, ...]: def _recorder_livekit_token(room: str) -> str: - """A subscribe-only token for the recorder's own LiveKit connection, signed with the backend's key pair.""" + """A viewer-capability token (subscribe and data messages, no media) for the recorder's LiveKit connection.""" api_key, signing_key = os.environ.get(API_KEY_ENV, ""), os.environ.get(API_SECRET_ENV, "") if not api_key or not signing_key: raise RuntimeError(f"{API_KEY_ENV} and {API_SECRET_ENV} must be set to record LiveKit video") diff --git a/examples/egocap/backend/egocap_web/clock_exchange.py b/examples/egocap/backend/egocap_web/clock_exchange.py new file mode 100644 index 0000000000..aa398c84ba --- /dev/null +++ b/examples/egocap/backend/egocap_web/clock_exchange.py @@ -0,0 +1,173 @@ +# SPDX-FileCopyrightText: 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# SPDX-License-Identifier: Apache-2.0 + +"""Maps a remote machine's monotonic clock onto this one's from four-timestamp exchanges. + +The pattern is the one CloudXR uses for its clock sync: this side stamps ``t1`` when it sends a request, the remote +side stamps ``t2`` on receipt and ``t3`` when it replies, and this side stamps ``t4`` on the reply. The round trip +without the remote's processing time is ``(t4 - t1) - (t3 - t2)``, and the remote-minus-local offset is +``((t2 - t1) + (t3 - t4)) / 2``. That point estimate assumes the two directions took equally long; the true offset +can lie anywhere within half the round trip of it, so that half is reported as the mapping's uncertainty and no +smaller bound is ever claimed. All times are integer nanoseconds and the offset is signed. + +This module only turns exchanges into a mapping. Sending the requests, and the transport under them, belong to +the caller, and so does the policy for what to do when no mapping is available: a mapping that is missing, stale +or discontinuous means the remote's timestamps cannot be placed on this clock. +""" + +from __future__ import annotations + +from collections import deque +from dataclasses import dataclass, replace + +#: Exchanges whose round trip exceeds this are not used: a slow or queued exchange says little about the clocks. +DEFAULT_MAX_RTT_NS = 250_000_000 +#: A mapping older than this (by this side's clock) is stale and is withheld. +DEFAULT_STALE_AFTER_NS = 5_000_000_000 +#: The most recent accepted exchanges the best one is chosen from. +DEFAULT_WINDOW = 8 +#: Relative clock rate difference tolerated between two mappings before their offsets are judged discontinuous. +DEFAULT_DRIFT_ALLOWANCE = 200e-6 +#: Adopted mappings kept as evidence; the newest are the ones a take can be recorded under. +_MAX_HISTORY = 256 + + +@dataclass(frozen=True) +class Exchange: + """One completed four-timestamp exchange: ``t1``/``t4`` on the local clock, ``t2``/``t3`` on the remote's.""" + + t1: int + t2: int + t3: int + t4: int + + @property + def rtt_ns(self) -> int: + """Round trip with the remote's own processing time removed.""" + return (self.t4 - self.t1) - (self.t3 - self.t2) + + @property + def offset_ns(self) -> int: + """Remote minus local, assuming symmetric transit.""" + return ((self.t2 - self.t1) + (self.t3 - self.t4)) // 2 + + @property + def well_formed(self) -> bool: + """Time moved forward on both clocks and the round trip is not negative.""" + return self.t4 >= self.t1 and self.t3 >= self.t2 and self.rtt_ns >= 0 + + +@dataclass(frozen=True) +class ClockMapping: + """An immutable mapping of the remote clock onto the local one; frame conversion only reads this.""" + + #: Starts at 1 and changes whenever the estimator starts over or finds a discontinuity. + epoch: int + offset_ns: int + rtt_ns: int + #: Half the round trip, plus the drift allowed over the exchange's age: the error unknown path asymmetry and + #: clock rate difference can cause. + uncertainty_ns: int + #: Local time at which the exchange behind this mapping completed. + accepted_at_ns: int + + def to_local_ns(self, remote_ns: int) -> int: + """The local-clock time of a remote-clock reading.""" + return remote_ns - self.offset_ns + + +class ClockEstimator: + """Keeps the best recent exchange as the current mapping and notices when the clocks jump apart.""" + + def __init__( + self, + *, + max_rtt_ns: int = DEFAULT_MAX_RTT_NS, + stale_after_ns: int = DEFAULT_STALE_AFTER_NS, + window: int = DEFAULT_WINDOW, + drift_allowance: float = DEFAULT_DRIFT_ALLOWANCE, + ) -> None: + self._max_rtt_ns = max_rtt_ns + self._stale_after_ns = stale_after_ns + self._drift_allowance = drift_allowance + self._recent: deque[Exchange] = deque(maxlen=window) + self._epoch = 1 + self._current: ClockMapping | None = None + #: Local time of the newest accepted exchange: staleness is about whether replies are still arriving, not + #: about how old the best exchange in the window is. + self._newest_ns: int | None = None + #: Every mapping adopted, oldest first, so a take can record the epochs and quality it was made under. + self.history: list[ClockMapping] = [] # trimmed to the newest _MAX_HISTORY + #: Set, and left set, when a new best mapping disagrees with the last by more than their uncertainties and + #: the allowed drift explain: mappings before it stay as they were, and a take spanning it is not aligned. + self.discontinuity = False + + def reset(self) -> None: + """Start over, for a new source session or a restarted remote: the old mapping no longer applies.""" + self._recent.clear() + self._current = None + self._newest_ns = None + self._epoch += 1 + + def add(self, exchange: Exchange) -> str | None: + """Take one exchange into account; returns why it was rejected, or None when it was accepted.""" + if not exchange.well_formed: + return "malformed exchange" + if exchange.rtt_ns > self._max_rtt_ns: + return "round trip too long" + previous = self._current + self._newest_ns = exchange.t4 + newest = self._mapping_of(exchange) + if previous is not None and self._disagrees(previous, newest): + # Not a refinement: the clocks moved. Judged on the newest exchange, not on the best of the window, so + # a jump is seen at once even when old exchanges have an equally good round trip. Keep the evidence and + # start a new epoch from this exchange alone. + self.discontinuity = True + self._epoch += 1 + self._recent.clear() + self._recent.append(exchange) + candidate = self._mapping_of(exchange) + else: + self._recent.append(exchange) + # The exchange with the smallest uncertainty now: half its round trip plus the drift its age could have + # added. A low round trip beats a fresher one only while its age stays cheaper; the newest wins ties. + best = min(self._recent, key=lambda e: (self._uncertainty_ns(e, exchange.t4), -e.t4)) + candidate = self._mapping_of(best, exchange.t4) + if candidate != previous: + self._current = candidate + self.history.append(candidate) + del self.history[:-_MAX_HISTORY] + return None + + def mapping(self, now_ns: int) -> ClockMapping | None: + """The current mapping, or None when there is none yet or no exchange arrived within the stale limit. + + The uncertainty is aged to ``now_ns``: while replies are late the offset's drift allowance keeps growing, + so a budget check on a returned mapping never sees a figure older than the call. History is not appended. + """ + current = self._current + if current is None or self._newest_ns is None or now_ns - self._newest_ns > self._stale_after_ns: + return None + aged = current.rtt_ns // 2 + int(self._drift_allowance * max(0, now_ns - current.accepted_at_ns)) + if aged <= current.uncertainty_ns: + return current + return replace(current, uncertainty_ns=aged) + + def _uncertainty_ns(self, exchange: Exchange, now_ns: int) -> int: + """Half the round trip plus the offset drift the exchange's age allows at the allowed relative rate.""" + return exchange.rtt_ns // 2 + int(self._drift_allowance * max(0, now_ns - exchange.t4)) + + def _mapping_of(self, exchange: Exchange, now_ns: int | None = None) -> ClockMapping: + return ClockMapping( + epoch=self._epoch, + offset_ns=exchange.offset_ns, + rtt_ns=exchange.rtt_ns, + uncertainty_ns=self._uncertainty_ns(exchange, exchange.t4 if now_ns is None else now_ns), + accepted_at_ns=exchange.t4, + ) + + def _disagrees(self, previous: ClockMapping, candidate: ClockMapping) -> bool: + elapsed_ns = max(0, candidate.accepted_at_ns - previous.accepted_at_ns) + allowed = previous.uncertainty_ns + candidate.uncertainty_ns + self._drift_allowance * elapsed_ns + return abs(candidate.offset_ns - previous.offset_ns) > allowed diff --git a/examples/egocap/backend/egocap_web/livekit_clock_sync.py b/examples/egocap/backend/egocap_web/livekit_clock_sync.py new file mode 100644 index 0000000000..df286d1564 --- /dev/null +++ b/examples/egocap/backend/egocap_web/livekit_clock_sync.py @@ -0,0 +1,131 @@ +# SPDX-FileCopyrightText: 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# SPDX-License-Identifier: Apache-2.0 + +"""Sends clock-sync requests to a LiveKit publisher and turns its replies into a clock mapping. + +Once a second this side stamps ``t1`` with its monotonic clock and sends a small request as a LiveKit data message +addressed to the publisher only; the publisher replies with its own monotonic ``t2`` (receipt) and ``t3`` (reply +send); this side stamps ``t4`` on the reply and hands the four times to ``ClockEstimator``. Requests ride the +existing room connection and never the video path. Messages are little-endian: a request is ``Q``, sequence, t1; +a reply is ``R``, sequence, t1, t2, t3 (nanoseconds, t1 on this side's clock, t2/t3 on the publisher's). + +A reply that does not match an outstanding request (sequence and t1) is ignored, so a stale or forged message +cannot become a measurement. Missing replies leave the mapping to go stale, which is the caller's signal. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import logging +import struct +import time +from collections.abc import Callable + +from .clock_exchange import ClockEstimator, ClockMapping, Exchange + +try: + from livekit import rtc +except ImportError: # the optional "livekit" extra is not installed; annotations only, nothing runs without it + rtc = None + +logger = logging.getLogger("isaaccapture.egocap.livekit_clock_sync") + +CLOCK_SYNC_TOPIC = "isaaccapture.clock-sync" +_REQUEST = struct.Struct(" bytes: + return _REQUEST.pack(b"Q", sequence & 0xFFFFFFFF, t1) + + +def decode_reply(data: bytes) -> tuple[int, int, int, int] | None: + """(sequence, t1, t2, t3) of a well-formed reply, else None.""" + if len(data) != _REPLY.size: + return None + kind, sequence, t1, t2, t3 = _REPLY.unpack(data) + return (sequence, t1, t2, t3) if kind == b"R" else None + + +class ClockSyncClient: + """Keeps a ``ClockEstimator`` fed by exchanges with one publisher, until stopped.""" + + def __init__( + self, + room: rtc.Room, + publisher_identity: str, + *, + interval_s: float = 1.0, + monotonic_ns: Callable[[], int] = time.monotonic_ns, + estimator: ClockEstimator | None = None, + ) -> None: + self._room = room + self._publisher = publisher_identity + self._interval_s = interval_s + self._monotonic_ns = monotonic_ns + self.estimator = estimator or ClockEstimator() + self._outstanding: dict[int, int] = {} + self._sequence = 0 + self._task: asyncio.Task[None] | None = None + #: Replies rejected by the estimator (malformed, too slow), counted for diagnostics. + self.rejected = 0 + + def start(self) -> None: + self._room.on("data_received", self._on_data) + self._task = asyncio.get_running_loop().create_task(self._run()) + + async def stop(self) -> None: + task, self._task = self._task, None + if task is not None: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + self._room.off("data_received", self._on_data) + + def mapping(self) -> ClockMapping | None: + """The publisher-to-local clock mapping, or None when there is none yet or it is stale.""" + return self.estimator.mapping(self._monotonic_ns()) + + @property + def discontinuity(self) -> bool: + return self.estimator.discontinuity + + def _on_data(self, packet: rtc.DataPacket) -> None: + t4 = self._monotonic_ns() # first, so it is as close to the receipt as software allows + if packet.topic != CLOCK_SYNC_TOPIC or packet.participant is None: + return + if packet.participant.identity != self._publisher: + return + reply = decode_reply(packet.data) + if reply is None: + return + sequence, t1, t2, t3 = reply + if self._outstanding.get(sequence) != t1: + return # not an answer to a request of ours; a stray reply must not discard the real one's entry + del self._outstanding[sequence] + if self.estimator.add(Exchange(t1=t1, t2=t2, t3=t3, t4=t4)) is not None: + self.rejected += 1 + + async def _run(self) -> None: + while True: + self._sequence += 1 + sequence = self._sequence + while len(self._outstanding) >= _MAX_OUTSTANDING: + self._outstanding.pop(next(iter(self._outstanding))) + try: + # t1 is read as late as possible, immediately before the send. + t1 = self._monotonic_ns() + self._outstanding[sequence & 0xFFFFFFFF] = t1 + await self._room.local_participant.publish_data( + encode_request(sequence, t1), + reliable=False, + destination_identities=[self._publisher], + topic=CLOCK_SYNC_TOPIC, + ) + except Exception: + logger.exception("clock-sync request failed") + await asyncio.sleep(self._interval_s) diff --git a/examples/egocap/backend/egocap_web/livekit_preview.py b/examples/egocap/backend/egocap_web/livekit_preview.py index 99566e3b50..2a9470de9a 100644 --- a/examples/egocap/backend/egocap_web/livekit_preview.py +++ b/examples/egocap/backend/egocap_web/livekit_preview.py @@ -42,7 +42,7 @@ def _b64url(data: bytes) -> str: def subscriber_token(*, api_key: str, signing_key: str, room: str, now: int, ttl_s: int = TOKEN_TTL_S) -> str: - """A signed LiveKit access token that may join ``room`` and subscribe, but not publish.""" + """A signed LiveKit token that may join ``room``, subscribe and send data messages, but not publish media.""" header = {"alg": "HS256", "typ": "JWT"} claims = { "iss": api_key, @@ -55,7 +55,9 @@ def subscriber_token(*, api_key: str, signing_key: str, room: str, now: int, ttl "roomJoin": True, "canSubscribe": True, "canPublish": False, - "canPublishData": False, + # Data messages only, for the clock-sync exchange with the publisher. Any holder of this token can send + # data messages into the room, so the publisher must act only on well-formed clock-sync requests. + "canPublishData": True, }, } # Compact JSON: the signature covers exactly these bytes, so both sides must agree on them. @@ -68,10 +70,11 @@ def subscriber_token(*, api_key: str, signing_key: str, room: str, now: int, ttl @router.get("/api/livekit-token") async def get_livekit_token(request: Request, room: str = "") -> JSONResponse: - """A fresh subscribe-only token for ``room``, which must be one the current session's LiveKit cameras offer. + """A fresh viewer token for ``room``, which must be one the current session's LiveKit cameras offer. Like the camera preview routes this is unauthenticated: anyone who can reach the backend can watch the - cameras, which is why the token cannot publish and is only issued for rooms CaptureServer offers. + cameras and send data messages into the room. The token cannot publish media, and it is only issued for + rooms CaptureServer offers. """ headers = {"Cache-Control": "no-store"} started = request.app.state.monitor.native.started diff --git a/examples/egocap/backend/egocap_web/livekit_receiver.py b/examples/egocap/backend/egocap_web/livekit_receiver.py index a6375cb6fd..0b8e1f4424 100644 --- a/examples/egocap/backend/egocap_web/livekit_receiver.py +++ b/examples/egocap/backend/egocap_web/livekit_receiver.py @@ -26,6 +26,8 @@ from capture_server import protocol as p +from .livekit_clock_sync import ClockSyncClient + try: from livekit import rtc except ImportError: # the optional "livekit" extra is not installed @@ -99,6 +101,7 @@ def __init__( token: str, *, queue_frames: int = DEFAULT_QUEUE_FRAMES, + clock_sync: bool = False, ) -> None: # The package is optional; fail here with the fix rather than later with an AttributeError. if rtc is None: @@ -113,6 +116,9 @@ def __init__( self._by_track = {s.track: s for s in streams} self._url, self._publisher = streams[0].url, streams[0].publisher self._token = token + self._want_clock_sync = clock_sync + #: Maps the publisher's monotonic clock onto this host's; set once connected when requested. + self.clock_sync: ClockSyncClient | None = None self._room: rtc.Room | None = None self._queues = {s.stream_id: asyncio.Queue[ReceivedFrame](maxsize=queue_frames) for s in streams} self.counters = {s.stream_id: StreamCounters() for s in streams} @@ -137,6 +143,9 @@ async def start(self) -> None: # Nothing is subscribed automatically: only the offered tracks are, so unrelated tracks in the room cost no # media work here. _ = await room.connect(self._url, self._token, rtc.RoomOptions(auto_subscribe=False)) + if self._want_clock_sync: + self.clock_sync = ClockSyncClient(room, self._publisher) + self.clock_sync.start() # Tracks published before we joined arrive here rather than through track_published. for participant in room.remote_participants.values(): for publication in participant.track_publications.values(): @@ -147,6 +156,9 @@ async def start(self) -> None: async def stop(self) -> None: """Leave the room and stop every reader; frames still queued stay available to the consumer.""" room, self._room = self._room, None + if self.clock_sync is not None: + await self.clock_sync.stop() + self.clock_sync = None for task in self._readers.values(): task.cancel() for task in list(self._readers.values()): diff --git a/examples/egocap/backend/egocap_web/livekit_recording.py b/examples/egocap/backend/egocap_web/livekit_recording.py index 7aae0a79b5..f44b43d8de 100644 --- a/examples/egocap/backend/egocap_web/livekit_recording.py +++ b/examples/egocap/backend/egocap_web/livekit_recording.py @@ -4,10 +4,15 @@ """Feeds received LiveKit camera frames into a take's recording ingress. -A frame is recorded only when its timestamp is a capture time (``SENSOR_CAPTURE_UNIX_US``), which can then be put -on this machine's monotonic clock, the same clock the tracking samples use. A stream stamped with the publisher's -monotonic clock or with its processing time is not a capture time, so the feed stops with that reason instead of -recording a guessed one. Arrival time is never used as capture time. +A frame is recorded only when its timestamp is on the publisher's monotonic clock: the time the frame was ready +to encode (``PUBLISHER_MONOTONIC_US``) or the sensor's capture event (``SENSOR_CAPTURE_MONOTONIC_US``). The +saved time means whichever the stream declares; frame-ready time is later than capture by the publisher's +capture and conversion delay, and neither is proof of physical alignment with tracking. It is placed on this +machine's monotonic clock, the one the tracking samples use, through the mapping the receiver's clock-sync +client builds from four-timestamp exchanges with the same publisher. There is one path: a stream with any other +basis, a missing or stale mapping, or a mapping that jumped during the take stops the feed with that reason +instead of recording a guessed time. Arrival time is never used as capture time. A mapping carries its +uncertainty (half the exchange round trip), not an accuracy claim about the sensor. Tracking times come from CaptureServer's monotonic clock, which agrees with this process's only when both run on one machine, so recording LiveKit video is refused unless CaptureServer's URL is a loopback address. @@ -19,7 +24,7 @@ import contextlib import ipaddress import logging -import time +import os from collections.abc import Callable, Mapping, Sequence from typing import Any from urllib.parse import urlparse @@ -32,63 +37,52 @@ logger = logging.getLogger("isaaccapture.egocap.livekit_recording") -#: How long a measured pair of clock readings is reused. Unix and monotonic time only drift apart slowly; a step of -#: the Unix clock is simply reflected in the offset at the next measurement, not detected. -_PAIR_MAX_AGE_S = 1.0 #: A frame older than this when it arrives, or stamped further than the lead below in the future, means the #: publisher's clock and this host's disagree grossly. This catches gross disagreement only, not millisecond error. _MAX_FRAME_AGE_S = 2.0 _MAX_FRAME_LEAD_S = 0.05 #: How long a published stream may send nothing before the take ends. Normal frame gaps are tens of milliseconds. _MAX_SILENCE_S = 5.0 -#: Readings per pair; the one read in the shortest span is the most trustworthy. -_PAIR_ATTEMPTS = 3 #: How long the room join may take before the take's start is refused. _JOIN_TIMEOUT_S = 15.0 #: How long leaving the room may take before teardown moves on and lets the leave finish in the background. _LEAVE_TIMEOUT_S = 10.0 - - -class UnixClockMap: - """Converts Unix-time microseconds to this process's monotonic seconds, from a recently measured pair. - - The pair is two clock readings taken back to back and refreshed about once a second, from the consumer's - thread and never from an SDK callback. It converts on this machine only: it does not detect a step of the - Unix clock, and says nothing about whether the publisher's clock agrees with this one. - """ - - def __init__( - self, - *, - monotonic_ns: Callable[[], int] = time.monotonic_ns, - unix_ns: Callable[[], int] = time.time_ns, - ) -> None: - self._monotonic_ns = monotonic_ns - self._unix_ns = unix_ns - self._offset_s = 0.0 - self._measured_at_ns: int | None = None - - def monotonic_seconds(self, unix_us: int) -> float: - """The monotonic-clock time, in seconds, of the Unix time ``unix_us`` (microseconds).""" - now = self._monotonic_ns() - if self._measured_at_ns is None or now - self._measured_at_ns > _PAIR_MAX_AGE_S * 1e9: - self._measure() - return unix_us / 1e6 + self._offset_s - - def _measure(self) -> None: - """Take the pair read in the shortest span: the Unix reading is taken between two monotonic ones.""" - best: tuple[int, float] | None = None - for _ in range(_PAIR_ATTEMPTS): - before = self._monotonic_ns() - unix = self._unix_ns() - after = self._monotonic_ns() - span = after - before - if best is None or span < best[0]: - # The Unix reading happened somewhere in [before, after]; its midpoint is the best estimate. - best = (span, ((before + after) / 2 - unix) / 1e9) - assert best is not None - self._offset_s = best[1] - self._measured_at_ns = self._monotonic_ns() +#: Most mappings recorded as a take's timing evidence, the newest ones; the estimator may have adopted more. +_MAX_EVIDENCE_MAPPINGS = 64 + +# The bases recorded, and what each one's time means in the saved take. +_RECORDABLE_BASES: dict[p.LiveKitTimestampBasis, str] = { + p.LiveKitTimestampBasis.PUBLISHER_MONOTONIC_US: "frame_ready", + p.LiveKitTimestampBasis.SENSOR_CAPTURE_MONOTONIC_US: "sensor_capture", +} + +BUDGET_ENV = "EGOCAP_CLOCK_UNCERTAINTY_BUDGET_MS" +# A frame at 30 fps is 33 ms; a mapping uncertain by more than a few ms would blur which camera frame matches a +# tracking sample. This default is a policy choice, not a measurement; set the environment variable to change it. +_DEFAULT_BUDGET_MS = 5.0 + + +# What each tracking source's saved time means today; none is a physical acquisition time. +_TRACKING_SEMANTICS: dict[str, str] = { + "recorded_time": "the host's poll time for every tracking source", + "controllers": "device value is the pose-evaluation time (a predicted XrTime), not an acquisition time", + "manus_gloves": "device value is host monotonic time at push; the SDK publishTime is not used", + "physical_alignment": "unavailable: no tracking source carries a mapped device measurement time", +} + + +def uncertainty_budget_ns() -> int: + """The largest clock-mapping uncertainty a take may be recorded with, from the environment or the default.""" + raw = os.environ.get(BUDGET_ENV) + if raw is None or not raw.strip(): + return int(_DEFAULT_BUDGET_MS * 1e6) + try: + ms = float(raw) + except ValueError: + raise ValueError(f"{BUDGET_ENV} must be a number of milliseconds, got {raw!r}") from None + if not ms > 0: + raise ValueError(f"{BUDGET_ENV} must be positive, got {raw!r}") + return int(ms * 1e6) def is_loopback_url(url: str) -> bool: @@ -122,9 +116,10 @@ def unrecordable_reason(started: Any, capture_server_url: str | None = None) -> f"LiveKit stream {stream.name!r}: recording needs CaptureServer on this machine " f"(its clock must match tracking's), but it is at {capture_server_url!r}" ) - if stream.timestamp_basis is not p.LiveKitTimestampBasis.SENSOR_CAPTURE_UNIX_US: + if stream.timestamp_basis not in _RECORDABLE_BASES: return ( - f"LiveKit stream {stream.name!r}: its timestamps are not a capture time, so it cannot be recorded" + f"LiveKit stream {stream.name!r}: its timestamps are not on the publisher's monotonic clock, " + "so it cannot be recorded" ) if None in (stream.width, stream.height, stream.fps): return f"LiveKit stream {stream.name!r}: width, height and fps must be configured to record it" @@ -145,8 +140,8 @@ def __init__( ingress: RecordingIngress, source_names: Mapping[int, str], *, - clock_map: UnixClockMap | None = None, on_failure: Callable[[str], None] | None = None, + max_uncertainty_ns: int | None = None, ) -> None: missing = [s.name for s in streams if s.stream_id not in source_names] if missing: @@ -155,8 +150,13 @@ def __init__( self._streams = {s.stream_id: s for s in streams} self._ingress = ingress self._source_names = dict(source_names) - self._clock_map = clock_map or UnixClockMap() + # The receiver was started with clock sync, so its client maps the publisher's clock onto this one. + clock_sync = receiver.clock_sync + if clock_sync is None: + raise ValueError("the receiver has no clock-sync client; start it with clock_sync=True") + self._clock = clock_sync self._on_failure = on_failure + self._max_uncertainty_ns = uncertainty_budget_ns() if max_uncertainty_ns is None else max_uncertainty_ns self._tasks: list[asyncio.Task[None]] = [] self.failure: str | None = None self._baseline: dict[int, tuple[int, int]] = {} @@ -201,23 +201,34 @@ def _fail(self, reason: str) -> None: if self._on_failure is not None: self._on_failure(reason) + def _clock_problem(self) -> str: + """What is wrong with the publisher's clock mapping right now, as a suffix for a failure reason (or "").""" + if self._clock.discontinuity: + return "; the publisher's clock mapping jumped" + mapping = self._clock.mapping() + if mapping is None: + return "; no current clock mapping from the publisher (its clock-sync replies stopped)" + if mapping.uncertainty_ns > self._max_uncertainty_ns: + return f"; the clock mapping is uncertain by {mapping.uncertainty_ns / 1e6:.2f} ms, over the budget" + return "" + async def _run(self, stream_id: int) -> None: """Feed one stream until the take fails or the feed is stopped.""" stream = self._streams[stream_id] source = self._source_names[stream_id] - if stream.timestamp_basis is not p.LiveKitTimestampBasis.SENSOR_CAPTURE_UNIX_US: + if stream.timestamp_basis not in _RECORDABLE_BASES: # Checked before any frame arrives: this stream cannot become recordable by waiting. - self._fail( - f"{stream.name}: its timestamps are not a capture time, so it cannot be aligned with tracking" - ) + self._fail(f"{stream.name}: its timestamps are not on the publisher's monotonic clock") return while self.failure is None: try: frame = await self._receiver.next_frame_or_lost(stream_id, timeout=_MAX_SILENCE_S) except asyncio.TimeoutError: - # A track can stay published and send nothing; a take must not wait on it forever. + # A track can stay published and send nothing; a take must not wait on it forever. With no frame + # to carry the check, the clock mapping is judged here too, so a dead publisher says that it is. self._fail( f"{stream.name}: no frame for {_MAX_SILENCE_S:.0f} s although the track is still published" + + self._clock_problem() ) return if frame is None: @@ -238,13 +249,30 @@ async def _run(self, stream_id: int) -> None: self._fail(f"{stream.name}: the camera's publication was replaced during the take") return if frame.user_timestamp_us is None: - self._fail(f"{stream.name}: a frame arrived without the publisher's capture timestamp") + self._fail(f"{stream.name}: a frame arrived without the publisher's frame timestamp") + return + # A mapping that jumped during the take, or went missing or stale, ends it: frames already recorded + # keep the mapping they were made under, and none after is recorded with a guessed one. + if self._clock.discontinuity: + self._fail(f"{stream.name}: the publisher's clock mapping jumped during the take") + return + mapping = self._clock.mapping() + if mapping is None: + self._fail( + f"{stream.name}: no current clock mapping from the publisher (its clock-sync replies stopped)" + ) + return + if mapping.uncertainty_ns > self._max_uncertainty_ns: + self._fail( + f"{stream.name}: the clock mapping is uncertain by {mapping.uncertainty_ns / 1e6:.2f} ms, " + f"over the {self._max_uncertainty_ns / 1e6:.2f} ms budget ({BUDGET_ENV})" + ) return - capture_time = self._clock_map.monotonic_seconds(frame.user_timestamp_us) + capture_time = mapping.to_local_ns(frame.user_timestamp_us * 1000) / 1e9 receive_time = frame.receive_time_ns / 1e9 age = receive_time - capture_time if not -_MAX_FRAME_LEAD_S <= age <= _MAX_FRAME_AGE_S: - # A coarse validity check: a capture time far from this host's clock is not a capture time here. + # A coarse validity check: a frame time far from this host's clock is not a frame time here. self._fail( f"{stream.name}: the publisher's clock disagrees with this host's (frame age {age:+.3f} s)" ) @@ -265,7 +293,8 @@ async def _run(self, stream_id: int) -> None: def recordable_livekit_streams(started: Any) -> tuple[p.LiveKitCameraStream, ...]: """The LiveKit streams of a started session that a recording may include. - A stream is recordable when its timestamps are a capture time and its geometry is configured: the recording + A stream is recordable when its timestamps are on the publisher's monotonic clock (frame-ready or sensor + capture time) and its geometry is configured: the recording schema fixes each source's size and rate when the session opens. The rest remain preview-only. """ if started is None: @@ -274,7 +303,7 @@ def recordable_livekit_streams(started: Any) -> tuple[p.LiveKitCameraStream, ... s for s in started.streams if isinstance(s, p.LiveKitCameraStream) - and s.timestamp_basis is p.LiveKitTimestampBasis.SENSOR_CAPTURE_UNIX_US + and s.timestamp_basis in _RECORDABLE_BASES and None not in (s.width, s.height, s.fps) ) @@ -306,6 +335,8 @@ def __init__( self._streams: tuple[p.LiveKitCameraStream, ...] = () self._receiver: LiveKitFrameReceiver | None = None self._feed: LiveKitRecordingFeed | None = None + # Read from the environment once in prepare(), so the feed and the saved evidence use the same figure. + self._budget_ns = int(_DEFAULT_BUDGET_MS * 1e6) async def prepare(self) -> None: """Join the room and wait for a usable first frame from every stream; raises if one never arrives. @@ -316,7 +347,9 @@ async def prepare(self) -> None: self._streams = tuple(self._recordable_streams()) if not self._streams: return - receiver = LiveKitFrameReceiver(self._streams, self._make_token(self._streams[0].room)) + # A bad budget setting fails the start here, before the room is joined and storage and ingress are live. + self._budget_ns = uncertainty_budget_ns() + receiver = LiveKitFrameReceiver(self._streams, self._make_token(self._streams[0].room), clock_sync=True) self._receiver = receiver try: await self._join(receiver) @@ -331,6 +364,7 @@ async def prepare(self) -> None: f"{stream.name}: frames are {frame.image.shape[1]}x{frame.image.shape[0]}, " f"expected {stream.width}x{stream.height}" ) + await self._wait_for_mapping(receiver) except asyncio.TimeoutError as exc: await self.stop() raise RuntimeError("a LiveKit stream sent no frame before the recording could start") from exc @@ -383,6 +417,34 @@ def _done(finished: asyncio.Future[None]) -> None: task.add_done_callback(_done) + def _mapping_problem(self, receiver: LiveKitFrameReceiver) -> str | None: + """Why the publisher's clock mapping cannot be used to start a take right now, or None when it can.""" + clock_sync = receiver.clock_sync + mapping = None if clock_sync is None else clock_sync.mapping() + if clock_sync is None or mapping is None: + return "the publisher did not answer clock-sync requests, so its frame times cannot be placed" + if clock_sync.discontinuity: + return "the publisher's clock mapping jumped before the take started" + if mapping.uncertainty_ns > self._budget_ns: + return ( + f"the clock mapping is uncertain by {mapping.uncertainty_ns / 1e6:.2f} ms, " + f"over the {self._budget_ns / 1e6:.2f} ms budget ({BUDGET_ENV})" + ) + return None + + async def _wait_for_mapping(self, receiver: LiveKitFrameReceiver) -> None: + """Wait for a usable clock mapping: present, continuous and within the uncertainty budget. + + A mapping the feed would reject on its first frame refuses the start instead, with the reason, rather + than opening a take that ends at once. The mapping may improve while waiting, so the check repeats until + the deadline. + """ + deadline = asyncio.get_running_loop().time() + self._first_frame_timeout_s + while (problem := self._mapping_problem(receiver)) is not None: + if asyncio.get_running_loop().time() >= deadline: + raise RuntimeError(problem) + await asyncio.sleep(0.05) + def start(self, ingress: RecordingIngress, on_failure: Callable[[str], None]) -> None: """Begin feeding the take's ingress with frames acquired from now on.""" if self._receiver is None: @@ -391,10 +453,65 @@ def start(self, ingress: RecordingIngress, on_failure: Callable[[str], None]) -> self._receiver.clear_queued() source_names = {s.stream_id: s.name for s in self._streams} self._feed = LiveKitRecordingFeed( - self._receiver, self._streams, ingress, source_names, on_failure=on_failure + self._receiver, + self._streams, + ingress, + source_names, + on_failure=on_failure, + max_uncertainty_ns=self._budget_ns, ) self._feed.start() + def timing_evidence(self) -> dict[str, Any] | None: + """The clock mappings this take's video was placed on the PC's clock with, for the take's attributes. + + None when there is no clock sync (nothing LiveKit was recorded). Read before ``stop()``, which leaves the + room. A long take keeps only the newest mappings; ``mappings_omitted_oldest`` counts the rest. + """ + receiver = self._receiver + clock_sync = None if receiver is None else receiver.clock_sync + if clock_sync is None: + return None + # The estimator re-reports one exchange with a growing age allowance while no newer one arrives; keep the + # first report of each exchange. + unique = {(m.epoch, m.accepted_at_ns): m for m in reversed(clock_sync.estimator.history)} + ordered = sorted(unique.values(), key=lambda m: (m.accepted_at_ns, m.epoch)) + mappings = ordered[-_MAX_EVIDENCE_MAPPINGS:] + return { + "clock_mapping": { + "method": "four_timestamp_exchange", + "local_clock": "pc_monotonic", + "remote_clock": "publisher_monotonic", + "streams": [ + { + "name": s.name, + "timestamp_basis": s.timestamp_basis.name, + "time_meaning": _RECORDABLE_BASES[s.timestamp_basis], + } + for s in self._streams + ], + "mappings": [ + { + "epoch": m.epoch, + "offset_ns": m.offset_ns, + "rtt_ns": m.rtt_ns, + "uncertainty_ns": m.uncertainty_ns, + "accepted_at_ns": m.accepted_at_ns, + } + for m in mappings + ], + # A long take keeps only the newest mappings; this counts the older ones left out. + "mappings_omitted_oldest": len(ordered) - len(mappings), + "uncertainty_budget_ns": self._budget_ns, + "discontinuity": clock_sync.discontinuity, + "exchanges_rejected": clock_sync.rejected, + "mapping_current_at_stop": clock_sync.mapping() is not None, + }, + # Tracking samples carry the host's poll time for every source; device measurement times are unmapped. + "tracking_time": "host_poll", + "tracking_semantics": _TRACKING_SEMANTICS, + } + async def stop(self) -> None: """Stop feeding and leave the room. Safe to call at any point, including after a failed prepare.""" feed, receiver = self._feed, self._receiver diff --git a/examples/egocap/backend/egocap_web/recording_session.py b/examples/egocap/backend/egocap_web/recording_session.py index 6737097556..2b331e75da 100644 --- a/examples/egocap/backend/egocap_web/recording_session.py +++ b/examples/egocap/backend/egocap_web/recording_session.py @@ -192,6 +192,8 @@ def __init__( # Optional LiveKit video for each take: joined before storage opens, fed after the ingress starts, left # before the ingress is retired. self._livekit_source = livekit_source + #: The clock-mapping evidence of the take in progress, gathered when its video source stops. + self._timing_evidence: dict[str, Any] | None = None self._queue_capacity = queue_capacity self._queue_bytes = queue_bytes self._readiness_check = readiness_check @@ -415,6 +417,10 @@ async def _start_episode_impl(self, frozen_context: TaskContext) -> EpisodeInfo: # construct a new ingress over it; the old consumer thread # would be left with nothing owning it. raise CoordinatorBusyError("a previous take's ingress has not yet settled") + # A take that was discarded or interrupted never consumed its evidence; it must not describe this take. + # Operations run one at a time and an unsettled take's ingress is still current (refused above), so no + # pending stop can still need the evidence cleared here. + self._timing_evidence = None if self._livekit_source is None: return await self._start_episode_with_storage(frozen_context) # LiveKit video must be flowing before any storage is opened, so a missing camera fails the start itself. @@ -594,6 +600,9 @@ async def _stop_episode_impl(self, service: _RecordingService, episode_id: str) result = await service.interrupt_episode(episode_id, self._last_issue) self._emit_result(result) return result + if self._timing_evidence is not None: + service.set_episode_timing(episode_id, self._timing_evidence) + self._timing_evidence = None result = await service.stop_episode(episode_id) self._emit_result(result) return result @@ -638,6 +647,10 @@ async def _retire_current_ingress(self) -> _IngressRetirement: if ingress is None: return _IngressRetirement(settled=True, failure=None) if self._livekit_source is not None: + # Read before the room is left: the clock mappings the take was recorded under. + evidence = self._livekit_source.timing_evidence() + if evidence is not None: + self._timing_evidence = evidence # The producer stops first, so nothing is offered to an ingress that is draining. await self._livekit_source.stop() settled = await ingress.stop() diff --git a/examples/egocap/capture_server/README.md b/examples/egocap/capture_server/README.md index 11e4b6b318..881fe056fa 100644 --- a/examples/egocap/capture_server/README.md +++ b/examples/egocap/capture_server/README.md @@ -594,7 +594,7 @@ All enums are IntEnum classes. Supply enum members to the Python API; incoming n | `ImageCodec` | `JPEG=1` | | `ClockDomain` | `CAPTURE_HOST_MONOTONIC=1` | | `TimeUnit` | `NANOSECONDS=1` | -| `LiveKitTimestampBasis` | `PUBLISHER_MONOTONIC_US=1`, `PUBLISHER_PROCESSING_UNIX_US=2`, `SENSOR_CAPTURE_UNIX_US=3` | +| `LiveKitTimestampBasis` | `PUBLISHER_MONOTONIC_US=1`, `PUBLISHER_PROCESSING_UNIX_US=2`, `SENSOR_CAPTURE_UNIX_US=3`, `SENSOR_CAPTURE_MONOTONIC_US=4` | | `SampleTimeSemantics` | `HOST_POLL=1` | | `PoseRepresentation` | `POSITION_QUATERNION=1` | | `QuaternionOrder` | `XYZW=1` | @@ -1229,7 +1229,7 @@ A `streams` entry with `kind` `LIVEKIT_IMAGE=3` says where a camera's video is p | `room` | Nonempty string of at most 512 characters. | The room to join. | | `publisher` | Nonempty string of at most 512 characters. | The publishing participant's identity. | | `track` | Nonempty string of at most 512 characters. | The published track's name; unique per `publisher`. | -| `timestamp_basis` | `LiveKitTimestampBasis`: `PUBLISHER_MONOTONIC_US=1`, `PUBLISHER_PROCESSING_UNIX_US=2`, `SENSOR_CAPTURE_UNIX_US=3`. | What the frame's `user_timestamp_us` means. Only `SENSOR_CAPTURE_UNIX_US` is a capture time; the other two are not recordable as one. | +| `timestamp_basis` | `LiveKitTimestampBasis`: `PUBLISHER_MONOTONIC_US=1`, `PUBLISHER_PROCESSING_UNIX_US=2`, `SENSOR_CAPTURE_UNIX_US=3`, `SENSOR_CAPTURE_MONOTONIC_US=4`. | What the frame's `user_timestamp_us` means. 1: the publisher's monotonic clock when the frame was ready to encode. 2: Unix processing time on the publisher. 3: Unix time of the sensor's capture event. 4: the sensor's capture event on the publisher's monotonic clock. Only 1 and 4 are recordable, using the publisher's clock-sync replies; 2 and 3 are preview only. | | `width`, `height` | Positive integer, or null (the key may be omitted). | The configured frame size, or null when not known. | | `fps` | Finite positive number, or null (the key may be omitted). | The configured frame rate, or null when not known. | @@ -1244,13 +1244,13 @@ livekit: url: wss://LIVEKIT_HOST # ws:// or wss:// room: sensing-preview publisher: orin-sensing - timestamp_basis: capture_unix_us # monotonic_us (default), processing_unix_us or capture_unix_us + timestamp_basis: monotonic_us # monotonic_us (default), capture_monotonic_us, processing_unix_us or capture_unix_us streams: # may be empty - {name: Head left, track: sensing-left, width: 2560, height: 1984, fps: 30} - {name: Head right, track: sensing-right} # width, height and fps are optional ``` -`timestamp_basis` applies to every stream and maps to the `LiveKitTimestampBasis` values above (`monotonic_us`=1, `processing_unix_us`=2, `capture_unix_us`=3). `url`, `room` and `publisher` are set once under `livekit`; each stream supplies `name`, `track` and optionally `width`, `height` and `fps`. A LiveKit stream name that equals a camera name configured with `--gmsl-camera` is rejected. +`timestamp_basis` applies to every stream and maps to the `LiveKitTimestampBasis` values above (`monotonic_us`=1, `processing_unix_us`=2, `capture_unix_us`=3, `capture_monotonic_us`=4). `url`, `room` and `publisher` are set once under `livekit`; each stream supplies `name`, `track` and optionally `width`, `height` and `fps`. A LiveKit stream name that equals a camera name configured with `--gmsl-camera` is rejected. Each WebSocket binary message consists of a 32-byte header followed immediately by a payload. For CaptureFrame, that payload is the packed tracking blob defined by SessionInit. For CaptureImage, it is the JPEG bytes. The header itself contains no JSON and no field-name strings. diff --git a/examples/egocap/capture_server/capture_server/livekit_streams.py b/examples/egocap/capture_server/capture_server/livekit_streams.py index 77e182b828..ad44cecc05 100644 --- a/examples/egocap/capture_server/capture_server/livekit_streams.py +++ b/examples/egocap/capture_server/capture_server/livekit_streams.py @@ -31,6 +31,7 @@ "monotonic_us": p.LiveKitTimestampBasis.PUBLISHER_MONOTONIC_US, "processing_unix_us": p.LiveKitTimestampBasis.PUBLISHER_PROCESSING_UNIX_US, "capture_unix_us": p.LiveKitTimestampBasis.SENSOR_CAPTURE_UNIX_US, + "capture_monotonic_us": p.LiveKitTimestampBasis.SENSOR_CAPTURE_MONOTONIC_US, } _STREAM_KEYS = {"name", "track", "width", "height", "fps"} diff --git a/examples/egocap/capture_server/capture_server/protocol.py b/examples/egocap/capture_server/capture_server/protocol.py index 3b5c9fe7a0..3736266cef 100644 --- a/examples/egocap/capture_server/capture_server/protocol.py +++ b/examples/egocap/capture_server/capture_server/protocol.py @@ -202,17 +202,25 @@ class TimeUnit(IntEnum): class LiveKitTimestampBasis(IntEnum): - """What a LiveKit frame's ``user_timestamp_us`` means. Only the last is a capture time.""" + """What a LiveKit frame's ``user_timestamp_us`` means. - # The publisher host's monotonic clock in microseconds, read when the frame is processed. Not on any clock - # this host can convert. + Recording accepts the two monotonic bases: the frame-ready time (1) and the sensor capture event (4). + """ + + # The publisher host's monotonic clock in microseconds, read when the frame is ready to encode: after capture + # and conversion, so later than the sensor's capture event. Recordable: the PC maps it onto its own monotonic + # clock with the publisher's four-timestamp clock-sync exchanges. PUBLISHER_MONOTONIC_US = 1 # Unix time in microseconds on the publisher host when the frame finished conversion: processing time, # later than capture by the capture and conversion delay. PUBLISHER_PROCESSING_UNIX_US = 2 # Unix time in microseconds of the sensor's capture event, mapped from the sensor's own counter by a sender - # that has verified the mapping. The only basis a recording may treat as the frame's capture time. + # that has verified the mapping. Not accepted for synchronized recording: no Unix-clock mapping is part of the + # approved synchronization path, so the value is kept for the wire's sake only. SENSOR_CAPTURE_UNIX_US = 3 + # The sensor's capture event in microseconds on the publisher host's monotonic clock, mapped from the sensor's + # own counter by a sender that has verified the mapping. Recordable the same way as PUBLISHER_MONOTONIC_US. + SENSOR_CAPTURE_MONOTONIC_US = 4 class SampleTimeSemantics(IntEnum): diff --git a/examples/egocap/frontend/package-lock.json b/examples/egocap/frontend/package-lock.json index 173182d2c6..30befa02e6 100644 --- a/examples/egocap/frontend/package-lock.json +++ b/examples/egocap/frontend/package-lock.json @@ -8,7 +8,7 @@ "name": "egocap-web-frontend", "version": "0.1.0", "dependencies": { - "livekit-client": "2.9.7", + "livekit-client": "2.19.2", "react": "19.3.1", "react-dom": "19.3.1" }, @@ -305,9 +305,9 @@ "license": "Apache-2.0" }, "node_modules/@livekit/protocol": { - "version": "1.34.0", - "resolved": "https://registry.npmjs.org/@livekit/protocol/-/protocol-1.34.0.tgz", - "integrity": "sha512-bU7pCLAMRVTVZb1KSxA46q55bhOc4iATrY/gccy2/oX1D57tiZEI+8wGRWHeDwBb0UwnABu6JXzC4tTFkdsaOg==", + "version": "1.45.8", + "resolved": "https://registry.npmjs.org/@livekit/protocol/-/protocol-1.45.8.tgz", + "integrity": "sha512-Q+l57E7w/xxOBFVWzdX5rkAZO7ffyF+rlDzNUYq2SU114+5aTyCq+PK4unaEVDNd4952Af7wteKr3sOgasGuaA==", "license": "Apache-2.0", "dependencies": { "@bufbuild/protobuf": "^1.10.0" @@ -660,6 +660,13 @@ "dev": true, "license": "MIT" }, + "node_modules/@types/dom-mediacapture-record": { + "version": "1.0.22", + "resolved": "https://registry.npmjs.org/@types/dom-mediacapture-record/-/dom-mediacapture-record-1.0.22.tgz", + "integrity": "sha512-mUMZLK3NvwRLcAAT9qmcK+9p7tpU2FHdDsntR3YI4+GY88XrgG4XiE7u1Q2LAN2/FZOz/tdMDC3GQCR4T8nFuw==", + "license": "MIT", + "peer": true + }, "node_modules/@types/estree": { "version": "1.0.9", "resolved": "https://registry.npmjs.org/@types/estree/-/estree-1.0.9.tgz", @@ -1334,6 +1341,15 @@ "dev": true, "license": "MIT" }, + "node_modules/jose": { + "version": "6.2.15", + "resolved": "https://registry.npmjs.org/jose/-/jose-6.2.15.tgz", + "integrity": "sha512-qdF6ewcZARtef63fJ9AKKZPgxw+t7USH6HLXYcD4CrdmSvDp79rpJSPFEKc90QBeGS63u1KQtN2MZQHC0yDIzA==", + "license": "MIT", + "funding": { + "url": "https://github.com/sponsors/panva" + } + }, "node_modules/js-tokens": { "version": "4.0.0", "resolved": "https://registry.npmjs.org/js-tokens/-/js-tokens-4.0.0.tgz", @@ -1644,20 +1660,23 @@ } }, "node_modules/livekit-client": { - "version": "2.9.7", - "resolved": "https://registry.npmjs.org/livekit-client/-/livekit-client-2.9.7.tgz", - "integrity": "sha512-a+Y76HE5k7IaFOpDnr14ON+VOAgh7cCjuBq8Loq5p5xHZzw+/cQyX/xPsMLU4lloKO5zGf45YZJYt/Egk1Xg+g==", + "version": "2.19.2", + "resolved": "https://registry.npmjs.org/livekit-client/-/livekit-client-2.19.2.tgz", + "integrity": "sha512-Kvk07QYDWRAbmYNLRll04ZIuxMQobW/oLPYnmR1kCy8GGHpU0gqyHf704Rz+29zfy8IJZRjKqeVbzGSKn9sumw==", "license": "Apache-2.0", "dependencies": { "@livekit/mutex": "1.1.1", - "@livekit/protocol": "1.34.0", + "@livekit/protocol": "1.45.8", "events": "^3.3.0", + "jose": "^6.1.0", "loglevel": "^1.9.2", "sdp-transform": "^2.15.0", - "ts-debounce": "^4.0.0", "tslib": "2.8.1", "typed-emitter": "^2.1.0", - "webrtc-adapter": "^9.0.1" + "webrtc-adapter": "9.0.5" + }, + "peerDependencies": { + "@types/dom-mediacapture-record": "^1" } }, "node_modules/loglevel": { @@ -2049,12 +2068,6 @@ "node": "^22.14.0 || >=24.0.0" } }, - "node_modules/ts-debounce": { - "version": "4.0.0", - "resolved": "https://registry.npmjs.org/ts-debounce/-/ts-debounce-4.0.0.tgz", - "integrity": "sha512-+1iDGY6NmOGidq7i7xZGA4cm8DAa6fqdYcvO5Z6yBevH++Bdo9Qt/mN0TzHUgcCcKv1gmh9+W5dHqz8pMWbCbg==", - "license": "MIT" - }, "node_modules/tslib": { "version": "2.8.1", "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.8.1.tgz", @@ -2300,9 +2313,9 @@ } }, "node_modules/webrtc-adapter": { - "version": "9.0.6", - "resolved": "https://registry.npmjs.org/webrtc-adapter/-/webrtc-adapter-9.0.6.tgz", - "integrity": "sha512-CHbl2ZQbxx164IgWRgzJno4hWtM4tFbRam1QfI3Yxhs3w/DvqluVxVWeXs3oL5/fbGkSNLKo0Ty5MgUWceNhog==", + "version": "9.0.5", + "resolved": "https://registry.npmjs.org/webrtc-adapter/-/webrtc-adapter-9.0.5.tgz", + "integrity": "sha512-U9vjByy/sK2OMXu5mmfuZFKTMIUQe34c0JXRO+oDrxJTsntdYT2iIFwYMOV7HhMTuktcZLGf2W1N/OcSf9ssWg==", "license": "BSD-3-Clause", "dependencies": { "sdp": "^3.2.0" diff --git a/examples/egocap/frontend/package.json b/examples/egocap/frontend/package.json index cd5ce66436..9d436a8258 100644 --- a/examples/egocap/frontend/package.json +++ b/examples/egocap/frontend/package.json @@ -11,7 +11,7 @@ "test": "vitest run" }, "dependencies": { - "livekit-client": "2.9.7", + "livekit-client": "2.19.2", "react": "19.3.1", "react-dom": "19.3.1" }, diff --git a/examples/egocap/frontend/src/CameraGrid.livekit.test.tsx b/examples/egocap/frontend/src/CameraGrid.livekit.test.tsx index 244aaf774e..e940721371 100644 --- a/examples/egocap/frontend/src/CameraGrid.livekit.test.tsx +++ b/examples/egocap/frontend/src/CameraGrid.livekit.test.tsx @@ -19,6 +19,7 @@ vi.mock("livekit-client", async () => { Reconnecting: "Reconnecting", Reconnected: "Reconnected", Disconnected: "Disconnected", + DataReceived: "DataReceived", }, Track: { Kind: { Video: "video", Audio: "audio" } }, }; diff --git a/examples/egocap/frontend/src/CameraGrid.tsx b/examples/egocap/frontend/src/CameraGrid.tsx index e79a257e5e..0d5569adce 100644 --- a/examples/egocap/frontend/src/CameraGrid.tsx +++ b/examples/egocap/frontend/src/CameraGrid.tsx @@ -265,6 +265,8 @@ export function CameraGrid({ name={tile.name} track={tile.track} phase={livekit.phase} + clock={livekit.clock} + basis={tile.basis} // A LiveKit tile is a main view only when no CaptureServer camera claims that place: every head // camera when there are several, otherwise the first tile. primary={liveKitOnly && (equalHeads ? isHeadTile(tile.name) : index === 0)} diff --git a/examples/egocap/frontend/src/LiveKitCameraTile.test.tsx b/examples/egocap/frontend/src/LiveKitCameraTile.test.tsx index 7eb88dcd62..46ebc3cf4a 100644 --- a/examples/egocap/frontend/src/LiveKitCameraTile.test.tsx +++ b/examples/egocap/frontend/src/LiveKitCameraTile.test.tsx @@ -4,16 +4,21 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { act, cleanup, render, screen } from "@testing-library/react"; -import { LiveKitCameraTile } from "./LiveKitCameraTile"; +import { LiveKitCameraTile, displayedFps } from "./LiveKitCameraTile"; import { fakeTrack } from "./livekitTestDoubles"; -let frameCallbacks: Array<() => void> = []; +type FrameMetadata = { presentedFrames: number; [field: string]: number }; +let frameCallbacks: Array<(now: number, metadata: FrameMetadata) => void> = []; +let presentedTotal = 0; +let presentationClockMs = 0; beforeEach(() => { vi.useFakeTimers(); frameCallbacks = []; + presentedTotal = 0; + presentationClockMs = 0; // jsdom has no requestVideoFrameCallback; this one lets a test present frames by hand. - (HTMLVideoElement.prototype as any).requestVideoFrameCallback = function (cb: () => void) { + (HTMLVideoElement.prototype as any).requestVideoFrameCallback = function (cb: (typeof frameCallbacks)[number]) { frameCallbacks.push(cb); return frameCallbacks.length; }; @@ -27,10 +32,18 @@ afterEach(() => { delete (HTMLVideoElement.prototype as any).cancelVideoFrameCallback; }); -function presentFrame(): void { +/** + * Presents `frames` frames at once: the browser's cumulative presentedFrames grows by that much. The frame's + * presentationTime is its own, independent of the fake timers' clock: by default a second after the previous call's. + */ +function presentFrame(frames = 1, times: Record = {}): void { + presentedTotal += frames; + presentationClockMs += 1000; const cbs = frameCallbacks; frameCallbacks = []; - act(() => cbs.forEach((cb) => cb())); + act(() => + cbs.forEach((cb) => cb(0, { presentedFrames: presentedTotal, presentationTime: presentationClockMs, ...times })), + ); } describe("LiveKitCameraTile placeholders", () => { @@ -132,3 +145,405 @@ describe("LiveKitCameraTile track wiring and labels", () => { expect(document.querySelector(".camera-tile--primary")).not.toBeNull(); }); }); + +describe("displayedFps", () => { + it("is frames per second over the window", () => { + expect(displayedFps({ presented: 10, at: 0 }, 40, 1000)).toBe(30); + }); + + it("returns null when the counter went backwards", () => { + expect(displayedFps({ presented: 40, at: 0 }, 10, 1000)).toBeNull(); + }); + + it("returns null for a non-positive interval", () => { + expect(displayedFps({ presented: 0, at: 1000 }, 5, 1000)).toBeNull(); + expect(displayedFps({ presented: 0, at: 1000 }, 5, 500)).toBeNull(); + }); +}); + +const statsTexts = (): string[] => + Array.from(document.querySelectorAll(".camera-stats")).map((e) => e.textContent ?? ""); + +describe("LiveKitCameraTile FPS label", () => { + function renderTile() { + return render(); + } + + it("shows the presented-frame delta per second after two windows", () => { + renderTile(); + presentFrame(0); + act(() => { + vi.advanceTimersByTime(1000); + }); + expect(document.body.textContent).not.toMatch(/fps/); // first window only sets the baseline + presentFrame(30); + act(() => { + vi.advanceTimersByTime(1000); + }); + expect(statsTexts()[0]).toBe("30 fps"); + }); + + it("divides by the callback presentationTime endpoints, not by the timer tick", () => { + renderTile(); + presentFrame(0, { presentationTime: 5000 }); + act(() => { + vi.advanceTimersByTime(1000); + }); + presentFrame(30, { presentationTime: 6000 }); + act(() => { + vi.advanceTimersByTime(1500); // a late tick: the window's length still comes from the frames + }); + expect(statsTexts()[0]).toBe("30 fps"); + }); + + it("gives no figure for a window with a counter reset and is right again in the next", () => { + renderTile(); + presentFrame(30, { presentationTime: 1000 }); + act(() => { + vi.advanceTimersByTime(1000); + }); + presentedTotal = 0; // the browser restarted its count + presentFrame(10, { presentationTime: 2000 }); + act(() => { + vi.advanceTimersByTime(1000); + }); + expect(document.body.textContent).not.toMatch(/fps/); + presentFrame(30, { presentationTime: 3000 }); + act(() => { + vi.advanceTimersByTime(1000); + }); + expect(statsTexts()[0]).toBe("30 fps"); + }); + + it("gives no figure for a window in which no new frame arrived", () => { + renderTile(); + presentFrame(5, { presentationTime: 1000 }); + act(() => { + vi.advanceTimersByTime(1000); + }); + act(() => { + vi.advanceTimersByTime(1000); + }); + expect(document.body.textContent).not.toMatch(/fps/); + }); + + it("shows 0 FPS when no frame arrives for more than three seconds", () => { + renderTile(); + presentFrame(); + act(() => { + vi.advanceTimersByTime(5000); + }); + expect(statsTexts()[0]).toBe("0 fps"); + }); + + it("shows a dash while the document is hidden", () => { + Object.defineProperty(document, "hidden", { configurable: true, value: true }); + try { + renderTile(); + presentFrame(); + act(() => { + vi.advanceTimersByTime(1000); + }); + expect(statsTexts()[0]).toBe("– fps"); + } finally { + delete (document as any).hidden; + } + }); + + describe("visibility changes", () => { + const setHidden = (hidden: boolean): void => { + Object.defineProperty(document, "hidden", { configurable: true, value: hidden }); + act(() => { + document.dispatchEvent(new Event("visibilitychange")); + }); + }; + afterEach(() => { + delete (document as any).hidden; + }); + + it("shows a dash at once when the tab is hidden, without waiting for the timer", () => { + renderTile(); + presentFrame(); + setHidden(true); + expect(statsTexts()[0]).toBe("– fps"); + }); + + it("starts a fresh FPS baseline after the tab returns", () => { + renderTile(); + presentFrame(0); + act(() => { + vi.advanceTimersByTime(1000); + }); + presentFrame(30); + act(() => { + vi.advanceTimersByTime(1000); + }); + expect(statsTexts()[0]).toBe("30 fps"); + setHidden(true); + presentFrame(300); // frames presented while hidden must not inflate the first visible window + setHidden(false); + act(() => { + vi.advanceTimersByTime(1000); + }); + expect(document.body.textContent).not.toMatch(/\d fps/); + presentFrame(30); + act(() => { + vi.advanceTimersByTime(1000); + }); + expect(statsTexts()[0]).toBe("30 fps"); + }); + + it("removes the visibilitychange listener on unmount", () => { + const remove = vi.spyOn(document, "removeEventListener"); + const { unmount } = renderTile(); + unmount(); + expect(remove).toHaveBeenCalledWith("visibilitychange", expect.any(Function)); + remove.mockRestore(); + }); + }); + + it("shows a dash for FPS without requestVideoFrameCallback", () => { + delete (HTMLVideoElement.prototype as any).requestVideoFrameCallback; + renderTile(); + act(() => { + vi.advanceTimersByTime(1000); + }); + expect(statsTexts()[0]).toBe("– fps"); + }); +}); + +describe("LiveKitCameraTile latency", () => { + function renderTile() { + return render(); + } + + it("shows one FPS span plus the fixed latency span, with an explanatory title", () => { + renderTile(); + presentFrame(1); + act(() => { + vi.advanceTimersByTime(1000); + }); + presentFrame(30); + act(() => { + vi.advanceTimersByTime(1000); + }); + const spans = document.querySelectorAll(".camera-stats"); + expect(spans).toHaveLength(2); + expect(spans[1].textContent).toBe("– ms latency"); + expect(spans[1].getAttribute("title")).toBeTruthy(); + }); + + it("does not use the browser's captureTime and expectedDisplayTime", () => { + renderTile(); + for (let i = 0; i < 3; i++) presentFrame(1, { captureTime: 1000 + i, expectedDisplayTime: 1070 + i }); + act(() => { + vi.advanceTimersByTime(1000); + }); + presentFrame(30, { captureTime: 2000, expectedDisplayTime: 2070 }); + act(() => { + vi.advanceTimersByTime(1000); + }); + expect(statsTexts()).toContain("– ms latency"); + expect(statsTexts().some((t) => /\d ms latency/.test(t))).toBe(false); + }); + + it("explains the latency tooltip without listing the browser's frame-callback fields", () => { + renderTile(); + presentFrame(1, { rtpTimestamp: 1, expectedDisplayTime: 2 }); + act(() => { + vi.advanceTimersByTime(1000); + }); + presentFrame(30, { rtpTimestamp: 1, expectedDisplayTime: 2 }); + act(() => { + vi.advanceTimersByTime(1000); + }); + const title = document.querySelectorAll(".camera-stats")[1].getAttribute("title") ?? ""; + expect(title).toContain("per-frame stamp"); + expect(title).not.toContain("reports:"); + }); + + it("keeps the fixed latency span beside 0 FPS for a stale stream", () => { + renderTile(); + presentFrame(1, { captureTime: 1000, expectedDisplayTime: 1070 }); + act(() => { + vi.advanceTimersByTime(5000); + }); + expect(statsTexts()).toEqual(["0 fps", "– ms latency"]); + }); + + it("keeps the fixed latency span beside paused while the document is hidden", () => { + Object.defineProperty(document, "hidden", { configurable: true, value: true }); + try { + renderTile(); + presentFrame(1); + act(() => { + vi.advanceTimersByTime(1000); + }); + expect(statsTexts()).toEqual(["– fps", "– ms latency"]); + } finally { + delete (document as any).hidden; + } + }); + + it("shows no latency text before any FPS label", () => { + renderTile(); + expect(document.body.textContent).not.toMatch(/latency/); + }); +}); + +describe("LiveKitCameraTile measured latency", () => { + const MS = 1_000_000n; + /** Remote clock = local + 3 s, so a frame displayed at local 997_072 ms and stamped 1_000_000 us is 72 ms old. */ + const mappingOf = (epoch = 1) => ({ epoch, offsetNs: 3_000n * MS, rttNs: 4n * MS, uncertaintyNs: 2n * MS, acceptedAtNs: 0n }); + const stampedTrack = (stamp: { userTimestamp: bigint } | null = { userTimestamp: 1_000_000_000n }) => { + const track = fakeTrack(); + track.lookupFrameMetadata = vi.fn(() => stamp ?? undefined); + return track; + }; + const frame = { rtpTimestamp: 42, expectedDisplayTime: 997_072 }; + + function renderWith(track: any, clock: { mapping: () => any } | null, basis = 4) { + return render(); + } + const windows = (n = 2): void => { + for (let i = 0; i < n; i++) { + act(() => { + vi.advanceTimersByTime(1000); + }); + } + }; + + it("shows the mean after the window, with uncertainty and p95 in the title, looking the stamp up by RTP timestamp", () => { + const track = stampedTrack(); + renderWith(track, { mapping: () => mappingOf() }); + presentFrame(1); + windows(1); + presentFrame(1, frame); + windows(1); + expect(statsTexts()[1]).toBe("72 ms latency"); + expect(document.querySelectorAll(".camera-stats")[1].getAttribute("title")).toContain("±2.0 ms, 95th percentile 72 ms"); + expect(track.lookupFrameMetadata).toHaveBeenCalledWith({ rtpTimestamp: 42 }); + }); + + it("is unavailable without a stamp", () => { + renderWith(stampedTrack(null), { mapping: () => mappingOf() }); + presentFrame(1, frame); + windows(1); + presentFrame(1, frame); // a frame in the second window, so it has an FPS figure beside the latency + windows(1); + expect(statsTexts()[1]).toBe("– ms latency"); + }); + + it("is available for the publisher's frame-ready monotonic basis", () => { + const track = stampedTrack(); + renderWith(track, { mapping: () => mappingOf() }, 1); + presentFrame(1); + windows(1); + presentFrame(1, frame); + windows(1); + expect(statsTexts()[1]).toBe("72 ms latency"); + expect(track.lookupFrameMetadata).toHaveBeenCalledWith({ rtpTimestamp: 42 }); + }); + + it("is unavailable when the basis is not a monotonic one, even with stamps", () => { + const track = stampedTrack(); + renderWith(track, { mapping: () => mappingOf() }, 2); + presentFrame(1, frame); + windows(1); + presentFrame(1, frame); // a frame in the second window, so it has an FPS figure beside the latency + windows(1); + expect(statsTexts()[1]).toBe("– ms latency"); + expect(track.lookupFrameMetadata).not.toHaveBeenCalled(); + }); + + it("is unavailable without a mapping", () => { + renderWith(stampedTrack(), { mapping: () => null }); + presentFrame(1, frame); + windows(1); + presentFrame(1, frame); // a frame in the second window, so it has an FPS figure beside the latency + windows(1); + expect(statsTexts()[1]).toBe("– ms latency"); + }); + + it("is unavailable for a window whose mapping epoch changed", () => { + let epoch = 1; + renderWith(stampedTrack(), { mapping: () => mappingOf(epoch) }); + presentFrame(1); + windows(1); + presentFrame(1, frame); // made under epoch 1 + epoch = 2; + windows(1); + expect(statsTexts()[1]).toBe("– ms latency"); + }); + + it("is unavailable when the stream is stale or the document is hidden", () => { + renderWith(stampedTrack(), { mapping: () => mappingOf() }); + presentFrame(1, frame); + act(() => { + vi.advanceTimersByTime(5000); + }); + expect(statsTexts()).toEqual(["0 fps", "– ms latency"]); + cleanup(); + Object.defineProperty(document, "hidden", { configurable: true, value: true }); + try { + renderWith(stampedTrack(), { mapping: () => mappingOf() }); + presentFrame(1, frame); + windows(1); + expect(statsTexts()).toEqual(["– fps", "– ms latency"]); + } finally { + delete (document as any).hidden; + } + }); + + const rerenderWith = (rerender: (ui: any) => void, track: any, clock: { mapping: () => any } | null, basis: number) => + rerender(); + + it("picks up a clock that arrives after the track, on the same track object", () => { + const track = stampedTrack(); + const { rerender } = renderWith(track, null); + presentFrame(1, frame); + windows(1); + presentFrame(1, frame); + windows(1); + expect(statsTexts()[1]).toBe("– ms latency"); + rerenderWith(rerender, track, { mapping: () => mappingOf() }, 4); + presentFrame(1, frame); + windows(1); + expect(statsTexts()[1]).toBe("72 ms latency"); + }); + + it("starts and stops taking samples when the basis changes, without re-attaching the track", () => { + const clock = { mapping: () => mappingOf() }; + const track = stampedTrack(); + const { rerender } = renderWith(track, clock, 2); + presentFrame(1, frame); + expect(track.lookupFrameMetadata).not.toHaveBeenCalled(); + windows(1); + rerenderWith(rerender, track, clock, 4); + presentFrame(1, frame); + expect(track.lookupFrameMetadata).toHaveBeenCalledTimes(1); + windows(1); + expect(statsTexts()[1]).toBe("72 ms latency"); + rerenderWith(rerender, track, clock, 2); + presentFrame(1, frame); + expect(track.lookupFrameMetadata).toHaveBeenCalledTimes(1); + windows(1); + expect(statsTexts()[1]).toBe("– ms latency"); + }); + + it("does not re-run the frame effect when the clock or basis changes", () => { + const track = stampedTrack(); + const { rerender, container } = renderWith(track, null, 2); + const video = container.querySelector("video")!; + expect(frameCallbacks).toHaveLength(1); + rerenderWith(rerender, track, { mapping: () => mappingOf() }, 4); + rerenderWith(rerender, track, { mapping: () => mappingOf(2) }, 1); + expect(frameCallbacks).toHaveLength(1); + expect((HTMLVideoElement.prototype as any).cancelVideoFrameCallback).not.toHaveBeenCalled(); + expect(track.attach).toHaveBeenCalledTimes(1); + expect(track.detach).not.toHaveBeenCalled(); + expect(container.querySelector("video")).toBe(video); + presentFrame(1, frame); + expect(frameCallbacks).toHaveLength(1); + }); +}); diff --git a/examples/egocap/frontend/src/LiveKitCameraTile.tsx b/examples/egocap/frontend/src/LiveKitCameraTile.tsx index e6acb096a3..72c5c8820c 100644 --- a/examples/egocap/frontend/src/LiveKitCameraTile.tsx +++ b/examples/egocap/frontend/src/LiveKitCameraTile.tsx @@ -12,11 +12,31 @@ import { useEffect, useRef, useState } from "react"; import type { RemoteTrack } from "livekit-client"; +import { latencySample, summarizeLatency } from "./displayLatency"; +import type { LatencySample } from "./displayLatency"; +import { MONOTONIC_TIMESTAMP_BASES } from "./captureProtocol"; +import type { LiveKitClockSync } from "./liveKitClockSync"; import type { LiveKitPhase } from "./useLiveKitPreview"; /** No decoded frame for this long means the picture is frozen. */ const FRAME_STALE_AFTER_MS = 3000; +/** Seconds-long windows keep the rate readable and the state updates to once per second. */ +const FPS_WINDOW_MS = 1000; +/** Recent latency samples kept per window. */ +const MAX_LATENCY_SAMPLES = 240; +/** Shown in place of a number that is not available. */ +const NO_LATENCY = "– ms latency"; +const NO_FPS = "– fps"; + +/** The displayed rate for one window from cumulative presented-frame counts; null for an invalid interval. */ +export function displayedFps(previous: { presented: number; at: number }, presented: number, at: number): number | null { + const elapsedMs = at - previous.at; + const frames = presented - previous.presented; + if (elapsedMs <= 0 || frames < 0) return null; // a counter reset or a clock oddity: start a new baseline + return (frames * 1000) / elapsedMs; +} + /** Text shown instead of a picture, or null when a fresh picture is showing. */ function placeholderFor(phase: LiveKitPhase, track: RemoteTrack | null, hasFrame: boolean, stale: boolean): string | null { if (phase === "error") return "Disconnected"; @@ -32,14 +52,35 @@ export function LiveKitCameraTile({ track, phase, primary, + clock = null, + basis = 0, }: { name: string; track: RemoteTrack | null; phase: LiveKitPhase; primary: boolean; + /** Maps the publisher's monotonic clock onto this browser's; latency needs it. */ + clock?: LiveKitClockSync | null; + /** What the stream's frame timestamp means; only a basis on the publisher's monotonic clock (frame-ready or sensor capture) gives a latency. */ + basis?: number; }) { const videoRef = useRef(null); + // The frame effect runs once per track, but the clock arrives (and the basis can change) after the track does, so + // it reads their current values through refs instead of the values it closed over when it started. + const clockRef = useRef(clock ?? null); + clockRef.current = clock ?? null; + const basisRef = useRef(basis); + basisRef.current = basis; const lastFrameAtRef = useRef(null); + // The newest frame callback's cumulative presented-frame count and that frame's own presentation time (ms). + const presentedRef = useRef<{ presented: number; at: number } | null>(null); + // The same fields at the start of the current window, for the displayed rate. + const baselineRef = useRef<{ presented: number; at: number } | null>(null); + // Latency of each presented frame in the current window, and the label shown for the last full window. + const latencySamplesRef = useRef([]); + const [latencyLabel, setLatencyLabel] = useState(NO_LATENCY); + const [latencyDetail, setLatencyDetail] = useState(""); + const [fpsLabel, setFpsLabel] = useState(null); const [hasFrame, setHasFrame] = useState(false); const [stale, setStale] = useState(false); @@ -47,6 +88,12 @@ export function LiveKitCameraTile({ useEffect(() => { const element = videoRef.current; lastFrameAtRef.current = null; + presentedRef.current = null; + baselineRef.current = null; + latencySamplesRef.current = []; + setLatencyLabel(NO_LATENCY); + setLatencyDetail(""); + setFpsLabel(null); setHasFrame(false); setStale(false); if (element === null || track === null) return; @@ -69,7 +116,20 @@ export function LiveKitCameraTile({ }; const watchFrames = (): void => { if (stopped) return; - callbackId = element.requestVideoFrameCallback(() => { + callbackId = element.requestVideoFrameCallback((now, metadata) => { + presentedRef.current = { presented: metadata.presentedFrames, at: metadata.presentationTime ?? now }; + // The publisher's stamp for exactly this frame, by its RTP timestamp, against its expected display time. + const rtp = (metadata as { rtpTimestamp?: number }).rtpTimestamp; + if (MONOTONIC_TIMESTAMP_BASES.includes(basisRef.current) && typeof rtp === "number" && "lookupFrameMetadata" in track) { + const stamp = (track as { lookupFrameMetadata: (a: { rtpTimestamp: number }) => { userTimestamp: bigint } | undefined }) + .lookupFrameMetadata({ rtpTimestamp: rtp }); + const sample = latencySample(metadata.expectedDisplayTime, stamp, clockRef.current?.mapping() ?? null); + if (sample !== null) { + const samples = latencySamplesRef.current; + samples.push(sample); + if (samples.length > MAX_LATENCY_SAMPLES) samples.shift(); + } + } markFrame(); watchFrames(); }); @@ -77,13 +137,58 @@ export function LiveKitCameraTile({ // requestVideoFrameCallback reports presented frames exactly; without it, playback progress is the fallback. if (typeof element.requestVideoFrameCallback === "function") watchFrames(); else element.addEventListener("timeupdate", markFrame); + const hasFrameCallback = typeof element.requestVideoFrameCallback === "function"; const timer = setInterval(() => { + const now = performance.now(); const last = lastFrameAtRef.current; - if (last !== null && performance.now() - last > FRAME_STALE_AFTER_MS) setStale(true); - }, 1000); + const isStale = last !== null && now - last > FRAME_STALE_AFTER_MS; + if (isStale) setStale(true); + // Displayed rate: the change in the browser's own presented-frame count over the window, not a callback tally. + const latest = presentedRef.current; + // The window's latency samples are summarized once and cleared; only those made under the clock mapping's + // current epoch count, so a window spanning a discontinuity has no figure. + const samples = latencySamplesRef.current; + latencySamplesRef.current = []; + const mapping = clockRef.current?.mapping() ?? null; + const summary = mapping === null ? null : summarizeLatency(samples, mapping.epoch); + const latencyText = summary === null ? NO_LATENCY : `${summary.mean.toFixed(0)} ms latency`; + setLatencyDetail(summary === null ? "" : `±${summary.uncertaintyMs.toFixed(1)} ms, 95th percentile ${summary.p95.toFixed(0)} ms`); + if (!hasFrameCallback) { + setFpsLabel(NO_FPS); + setLatencyLabel(NO_LATENCY); + } else if (document.hidden) { + baselineRef.current = null; // a hidden tab's scheduling is not comparable: restart after it returns + setFpsLabel(NO_FPS); + setLatencyLabel(NO_LATENCY); + } else if (isStale) { + baselineRef.current = null; + setFpsLabel("0 fps"); + setLatencyLabel(NO_LATENCY); + } else if (latest !== null) { + setLatencyLabel(latencyText); + // Both ends of the window are frames the browser presented, so a late timer tick cannot skew the rate; a + // counter reset gives no figure and the new count becomes the baseline. + const previous = baselineRef.current; + baselineRef.current = latest; + const fps = previous === null || latest.at <= previous.at ? null : displayedFps(previous, latest.presented, latest.at); + if (fps !== null) setFpsLabel(`${fps.toFixed(0)} fps`); + } + }, FPS_WINDOW_MS); + // The timer can be throttled or suspended while the tab is hidden, so the transition itself resets the window: + // paused on hide, and a fresh baseline (no figure) for the first window after it returns. + const onVisibility = (): void => { + baselineRef.current = null; + latencySamplesRef.current = []; + if (document.hidden) { + setFpsLabel(NO_FPS); + setLatencyLabel(NO_LATENCY); + } + }; + document.addEventListener("visibilitychange", onVisibility); return () => { stopped = true; clearInterval(timer); + document.removeEventListener("visibilitychange", onVisibility); if (callbackId !== null) element.cancelVideoFrameCallback(callbackId); element.removeEventListener("timeupdate", markFrame); }; @@ -95,7 +200,22 @@ export function LiveKitCameraTile({ {/* The element stays mounted while a track exists so attach/detach never races a re-created node. */} {track !== null &&