diff --git a/examples/egocap/LIVEKIT_PREVIEW.md b/examples/egocap/LIVEKIT_PREVIEW.md new file mode 100644 index 0000000000..30952ee889 --- /dev/null +++ b/examples/egocap/LIVEKIT_PREVIEW.md @@ -0,0 +1,88 @@ + + +# LiveKit camera preview and recording + +Camera video can reach the PC through a LiveKit room instead of through CaptureServer. CaptureServer only +says where the video is; the browser (preview) and the backend (recording) each join the room themselves. +The publisher is the Orin's sender, or `tools/livekit_orin_emulator.py` when no Orin is available. + +Recording LiveKit video and running the emulator need the optional `livekit` package: `uv sync --extra livekit` +(included in `--extra dev`). Without it the backend still starts and the browser preview works, but the Record +button is blocked for LiveKit streams with a message naming the extra. + +Run every command from `examples/egocap`. + +## 1. Describe the streams + +Write a YAML file naming the room, the publisher's identity and one entry per published track. The field +reference is in [`capture_server/README.md`](capture_server/README.md#the---livekit-streams-file). + +```yaml +livekit: + url: wss://LIVEKIT_HOST + room: sensing-preview + publisher: orin-sensing + timestamp_basis: capture_unix_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. + +Start CaptureServer with the file (add `--mock` for synthetic tracking): + +```bash +uv run capture-server --mock --livekit-streams livekit.yaml +``` + +## 2. Create the publisher's join token + +The publisher joins with a token for the room and identity from the file, with publish permission. With the +LiveKit CLI (`lk`), using the key pair of the LiveKit server: + +```bash +lk token create --api-key "$LIVEKIT_API_KEY" --api-secret "$LIVEKIT_API_SECRET" \ + --join --room sensing-preview --identity orin-sensing --valid-for 24h > orin-token.txt +``` + +The CLI flags are not exercised by this repository's tests; check `lk token create --help` for your version. +The token file holds only the token. Keep it private. + +## 3. Run the emulator (no Orin) + +`tools/livekit_orin_emulator.py` joins the room and publishes one test-pattern track per `--add-stream`. + +```bash +uv run python tools/livekit_orin_emulator.py \ + --livekit-url wss://LIVEKIT_HOST \ + --livekit-token-file orin-token.txt \ + --add-stream sensor=0,livekit=sensing-left \ + --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 +encoder, so this checks the PC side and the metadata, not the Jetson encoder. + +## 4. Preview and record on the PC + +Give the backend the LiveKit server's key pair, then start it as in the root [`README.md`](README.md): + +```bash +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 + reports the preview as unavailable. +- Recording joins the room with its own subscribe-only 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. diff --git a/examples/egocap/backend/README.md b/examples/egocap/backend/README.md index 74fa5010cd..ec58128719 100644 --- a/examples/egocap/backend/README.md +++ b/examples/egocap/backend/README.md @@ -33,6 +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/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). | @@ -54,12 +55,19 @@ 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). | | `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. | | `EGOCAP_TRUSTED_HOSTNAMES` | unset (dev default) | Comma-separated hostnames trusted for the same-origin fallback on `WS /api/session`. Leaving this unset derives the expected origin from the request's own `Host` header, which is fine for local development but not a safe default for a deployment reachable from an untrusted network. | | `EGOCAP_VISER_ENABLED` and related `EGOCAP_VISER_*` | on | The embedded 3D tracking view; see [`frontend/README.md`](../frontend/README.md). | +## Optional LiveKit support + +Recording LiveKit camera video needs the `livekit` extra (`uv sync --extra livekit`). Without it the backend +starts normally and the browser preview works; only recording a LiveKit stream is refused, with a message that +names the extra. See [`LIVEKIT_PREVIEW.md`](../LIVEKIT_PREVIEW.md). + ## Development ```bash diff --git a/examples/egocap/backend/egocap_web/app.py b/examples/egocap/backend/egocap_web/app.py index a6b1f79c0c..8d0478220f 100644 --- a/examples/egocap/backend/egocap_web/app.py +++ b/examples/egocap/backend/egocap_web/app.py @@ -17,7 +17,7 @@ from contextlib import asynccontextmanager from dataclasses import replace from pathlib import Path -from typing import AsyncIterator +from typing import Any, AsyncIterator from fastapi import FastAPI, Request from fastapi.responses import JSONResponse @@ -35,6 +35,10 @@ from .capture_preview import MAX_CAMERA_JPEG_BYTES from .capture_status import CaptureStatusMonitor from .config import VisualizationConfig, config_from_env, visualization_from_env +from .livekit_preview import API_KEY_ENV, API_SECRET_ENV, subscriber_token +from .livekit_preview import router as livekit_preview_router +from .livekit_receiver import RECEIVER_FRAMES_PER_STREAM +from .livekit_recording import LiveKitRecordingSource, recordable_livekit_streams, unrecordable_reason from .operator_settings import OperatorSettingsStore, operator_settings_path from .operator_settings import router as operator_settings_router from .preview_routes import router as preview_router @@ -71,7 +75,9 @@ # independently, each of these would look individually reasonable while # still summing to far more memory than intended -- see # _recording_config_factory below for how the decoded-queue capacity is -# derived from what's left after the raw side's own reservation. +# derived from what's left after the raw side's own reservation. The +# LiveKit streams' own ingress share and receiver queues are added on top +# per take (they scale with the stream geometry), not carved out of this. _RECORDING_QUEUE_MEMORY_BUDGET_BYTES = 256 * 1024 * 1024 # RecordingIngress's own raw-item queue, per take -- items queued here are @@ -80,14 +86,37 @@ # blob (a typical camera frame is a few hundred KB, so this holds seconds # of 60 Hz tracking plus two or three 30 fps cameras, where an 8-slot count # held ~50 ms). _INGRESS_QUEUE_CAPACITY is only a safety cap on item count. -# The reservation adds one MAX_CAMERA_JPEG_BYTES for the raw item held live -# in the executor during conversion (capture_preview.py's own cap, not a -# locally-assumed "typical" size). Fixed at coordinator-construction time, -# before any camera is necessarily known, so unlike the decoded queue below -# it cannot be sized from real resolution. +# The reservation adds one raw item held live in the executor during +# conversion: MAX_CAMERA_JPEG_BYTES (capture_preview.py's own cap), or a +# LiveKit stream's decoded frame when that is larger. _INGRESS_QUEUE_BYTES = 64 * 1024 * 1024 _INGRESS_QUEUE_CAPACITY = 4096 -_INGRESS_RESERVED_BYTES = _INGRESS_QUEUE_BYTES + MAX_CAMERA_JPEG_BYTES +# LiveKit frames are decoded RGB (megabytes each, ~15 MB at 2560x1984), so +# the JPEG allowance above holds only a few and a short consumer stall would +# fail the take. Each recordable stream therefore adds this many of its own +# frames to the ingress queue, and the receiver's queues (see +# livekit_receiver.RECEIVER_FRAMES_PER_STREAM) are added to the total budget. +_LIVEKIT_INGRESS_FRAMES_PER_STREAM = 6 + + +def _livekit_frame_bytes(started: Any) -> list[int]: + """Decoded RGB size of one frame of each LiveKit stream a take would record.""" + return [s.width * s.height * 3 for s in recordable_livekit_streams(started)] + + +def _livekit_ingress_bytes(started: Any) -> int: + """Extra raw-ingress bytes the recordable LiveKit streams need.""" + return sum(_livekit_frame_bytes(started)) * _LIVEKIT_INGRESS_FRAMES_PER_STREAM + + +def _livekit_receiver_bytes(started: Any) -> int: + """Worst-case bytes the LiveKit receiver's own queues pin for the recordable streams.""" + return sum(_livekit_frame_bytes(started)) * RECEIVER_FRAMES_PER_STREAM + + +def _ingress_queue_bytes(started: Any) -> int: + """The raw-ingress byte bound of a take: the JPEG/tracking allowance plus the LiveKit frames' share.""" + return _INGRESS_QUEUE_BYTES + _livekit_ingress_bytes(started) def _recording_output_root(station_data_dir: Path | None = None) -> Path: @@ -122,6 +151,14 @@ 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.""" + 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") + return subscriber_token(api_key=api_key, signing_key=signing_key, room=room, now=int(time.time())) + + def _recording_config_factory(monitor: CaptureStatusMonitor, station_label: str | None, output_root: Path): """Builds the real ``ServiceConfig`` lazily, read fresh each time the coordinator actually opens/reopens a session (not once at import time) @@ -139,6 +176,10 @@ def factory() -> ServiceConfig: video_sources = tuple( VideoSourceSpec(cam.name, cam.width, cam.height, cam.fps, intrinsics=False) for cam in monitor.camera_descriptors + ) + tuple( + # LiveKit cameras with a capture-time stamp and a configured size are recorded like any camera. + VideoSourceSpec(s.name, s.width, s.height, s.fps, intrinsics=False) + for s in recordable_livekit_streams(monitor.native.started) ) # A supported profile must record actual tracking AND the # selected cameras -- pico_controller and every configured camera @@ -163,9 +204,19 @@ def factory() -> ServiceConfig: # one in the real service's own writer (the sample it has # already dequeued and is actively encoding). Genuinely bounds # the combined worst case at _RECORDING_QUEUE_MEMORY_BUDGET_BYTES. - decoded_queue_budget = ( - _RECORDING_QUEUE_MEMORY_BUDGET_BYTES - _INGRESS_RESERVED_BYTES - 2 * largest_decoded_frame_bytes + started = monitor.native.started + # The live raw item is a JPEG, or a LiveKit stream's decoded frame when that is larger. + live_item_bytes = max([MAX_CAMERA_JPEG_BYTES, *_livekit_frame_bytes(started)]) + # Owned storage, each counted once: the raw ingress queue, the LiveKit receivers' own frame queues, and + # the one live item. The budget is the base plus the LiveKit ingress allowance plus the receiver + # allowance, so the receiver queues are taken out of it here and what remains is the decoded queue. + reserved = _ingress_queue_bytes(started) + _livekit_receiver_bytes(started) + live_item_bytes + total = ( + _RECORDING_QUEUE_MEMORY_BUDGET_BYTES + + _livekit_ingress_bytes(started) + + _livekit_receiver_bytes(started) ) + decoded_queue_budget = total - reserved - 2 * largest_decoded_frame_bytes capacity = decoded_queue_budget // largest_decoded_frame_bytes if capacity < 1: # A silent floor here would claim a cap that doesn't @@ -192,7 +243,11 @@ def factory() -> ServiceConfig: _READINESS_STALE_SECONDS = 2.0 -def _recording_readiness_check(monitor: CaptureStatusMonitor, station_config_error: str | None = None): +def _recording_readiness_check( + monitor: CaptureStatusMonitor, + station_config_error: str | None = None, + capture_server_url: str | None = None, +): """Real backend-local capture facts, injected into the coordinator -- it never imports CaptureStatusMonitor/CaptureClient itself (see recording_session.py's own module docstring). Checked at start_episode @@ -208,6 +263,8 @@ def _recording_readiness_check(monitor: CaptureStatusMonitor, station_config_err received within the staleness window -- not merely "ever arrived once." + ``capture_server_url`` lets LiveKit recording be refused when CaptureServer is not on this machine. + ``station_config_error``, when set, unconditionally blocks new recording admission with that reason -- a configured-but-invalid station YAML must never silently fall back to a development default, but it must @@ -228,8 +285,14 @@ def check() -> tuple[bool, str | None]: tracking = monitor.preview.tracking() if tracking is None or (now - tracking.received_monotonic) > _READINESS_STALE_SECONDS: return False, "tracking source is stale or has no sample yet" + # An offered LiveKit stream that cannot be recorded blocks recording by name, never silently dropped. + blocked = unrecordable_reason(monitor.native.started, capture_server_url) + if blocked is not None: + return False, blocked cameras = monitor.camera_descriptors - if not cameras: + # A recordable LiveKit stream is a camera too; its liveness is checked when the take prepares and + # while it records, not by a preview sample here. + if not cameras and not recordable_livekit_streams(monitor.native.started): return False, "no camera selected" for cam in cameras: slot = monitor.preview.camera(cam.name) @@ -329,12 +392,16 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # Catalogs the operator imports from the page sit beside that file, apart from the shipped ones. app.state.imported_task_file_root = operator_settings_file.parent / "taskfiles" monitor = CaptureStatusMonitor(config, backend_instance_id=str(uuid.uuid4())) + livekit_source = LiveKitRecordingSource( + lambda: recordable_livekit_streams(monitor.native.started), _recorder_livekit_token + ) coordinator = RecordingSessionCoordinator( _recording_config_factory(monitor, config.station_label, recording_output_root), queue_capacity=_INGRESS_QUEUE_CAPACITY, - queue_bytes=_INGRESS_QUEUE_BYTES, - readiness_check=_recording_readiness_check(monitor, station_config_error), + queue_bytes=lambda: _ingress_queue_bytes(monitor.native.started), + readiness_check=_recording_readiness_check(monitor, station_config_error, config.capture_server_url), initial_context=initial_context, + livekit_source=livekit_source, ) monitor.set_recording(coordinator) await monitor.start() @@ -403,6 +470,7 @@ def _optional_visualization( app = FastAPI(lifespan=lifespan) app.include_router(preview_router) +app.include_router(livekit_preview_router) app.include_router(task_file_router) app.include_router(settings_router) app.include_router(operator_settings_router) diff --git a/examples/egocap/backend/egocap_web/livekit_preview.py b/examples/egocap/backend/egocap_web/livekit_preview.py new file mode 100644 index 0000000000..99566e3b50 --- /dev/null +++ b/examples/egocap/backend/egocap_web/livekit_preview.py @@ -0,0 +1,91 @@ +# SPDX-FileCopyrightText: 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# SPDX-License-Identifier: Apache-2.0 + +"""Viewer tokens for the LiveKit video CaptureServer offers. + +CaptureServer says where the video is (URL, room, publisher, track) in the LiveKit camera entries of the +session's ``streams``. This +module only issues the credential a consumer needs to join that room and subscribe, so a missing signing key can +stop video but never hides the stream list. The backend does not decode or relay video. + +``LIVEKIT_API_KEY`` and ``LIVEKIT_API_SECRET`` belong to the LiveKit server and are read from the backend's +environment only; they never appear in a response or a log line. +""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import os +import secrets +import time + +from fastapi import APIRouter, Request +from fastapi.responses import JSONResponse + +from capture_server import protocol as p + +API_KEY_ENV = "LIVEKIT_API_KEY" +API_SECRET_ENV = "LIVEKIT_API_SECRET" +#: A token only has to outlive the page's connect; LiveKit keeps an established connection going afterwards. +TOKEN_TTL_S = 3600 + +router = APIRouter() + + +def _b64url(data: bytes) -> str: + """URL-safe base64 without padding, as JWT segments require.""" + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +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.""" + header = {"alg": "HS256", "typ": "JWT"} + claims = { + "iss": api_key, + # Each consumer gets its own identity: LiveKit disconnects an existing participant that reuses one. + "sub": f"egocap-viewer-{secrets.token_hex(4)}", + "nbf": now, + "exp": now + ttl_s, + "video": { + "room": room, + "roomJoin": True, + "canSubscribe": True, + "canPublish": False, + "canPublishData": False, + }, + } + # Compact JSON: the signature covers exactly these bytes, so both sides must agree on them. + signing_input = ".".join( + _b64url(json.dumps(part, separators=(",", ":")).encode()) for part in (header, claims) + ) + signature = hmac.new(signing_key.encode(), signing_input.encode("ascii"), hashlib.sha256).digest() + return f"{signing_input}.{_b64url(signature)}" + + +@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. + + 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. + """ + headers = {"Cache-Control": "no-store"} + started = request.app.state.monitor.native.started + offered = ( + {s.room for s in started.streams if isinstance(s, p.LiveKitCameraStream)} if started is not None else set() + ) + # A token for an arbitrary room would let a caller probe or join rooms this station never advertised. + if room not in offered: + return JSONResponse({"detail": "room is not offered by CaptureServer"}, status_code=404, headers=headers) + api_key = os.environ.get(API_KEY_ENV, "") + signing_key = os.environ.get(API_SECRET_ENV, "") + # Offered but unable to sign: say what to set, without echoing any value. + if not api_key or not signing_key: + detail = f"{API_KEY_ENV} and {API_SECRET_ENV} must be set for the backend" + return JSONResponse({"detail": detail}, status_code=503, headers=headers) + token = subscriber_token(api_key=api_key, signing_key=signing_key, room=room, now=int(time.time())) + return JSONResponse({"token": token}, headers=headers) diff --git a/examples/egocap/backend/egocap_web/livekit_receiver.py b/examples/egocap/backend/egocap_web/livekit_receiver.py new file mode 100644 index 0000000000..a6375cb6fd --- /dev/null +++ b/examples/egocap/backend/egocap_web/livekit_receiver.py @@ -0,0 +1,339 @@ +# SPDX-FileCopyrightText: 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# SPDX-License-Identifier: Apache-2.0 + +"""Receives the LiveKit camera video a session offers, for recording. + +One room connection serves every offered stream. Each frame arrives with the publisher's own metadata (the +application timestamp, the frame id and the raw user data) and the time this process took it from the SDK, and is +put on a bounded per-stream queue. Nothing here converts those publisher times to this machine's clock or writes +anything: that belongs to the recorder. A full queue never blocks the receiver or hides a loss; the frame is +dropped and counted, and a jump in the publisher's frame id is counted as missing frames. + +This is separate from the browser's own connection: closing the browser does not touch this one. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import logging +import time +from collections.abc import Sequence +from dataclasses import dataclass, field + +import numpy as np + +from capture_server import protocol as p + +try: + from livekit import rtc +except ImportError: # the optional "livekit" extra is not installed + rtc = None + +logger = logging.getLogger("isaaccapture.egocap.livekit_receiver") + +LIVEKIT_MISSING = "receiving LiveKit video needs the livekit package: install the egocap 'livekit' extra" + + +def livekit_available() -> bool: + """Whether the optional ``livekit`` package is installed, so LiveKit video can be received.""" + return rtc is not None + + +#: Frames a stream may hold between the SDK and the consumer before further frames are dropped and counted. +#: Decoded frames are megabytes each, so the queues are short and sized into the recorder's memory budget. +DEFAULT_QUEUE_FRAMES = 2 +#: The SDK keeps only this many undelivered frames per track and drops the oldest silently; its drops show up +#: as a jump in the publisher's frame id, which is counted below. +SDK_QUEUE_FRAMES = 2 +#: Frames one stream can pin at once with default queues: both queues plus the one the reader is converting. +RECEIVER_FRAMES_PER_STREAM = DEFAULT_QUEUE_FRAMES + SDK_QUEUE_FRAMES + 1 + + +@dataclass(frozen=True) +class ReceivedFrame: + """One decoded frame and everything the publisher attached to it.""" + + stream_id: int + #: Height x width x 3, uint8, RGB. Views the SDK frame's own buffer, which the SDK builds per frame and never + #: reuses, so no further copy is made here. + image: np.ndarray + #: ``time.monotonic_ns()`` of this process when the frame was taken from the SDK, not a capture time. + receive_time_ns: int + #: The WebRTC event timestamp, which the SDK documents as different from the publisher's own metadata. + transport_timestamp_us: int + #: The publisher's application timestamp; its clock is named by the stream's ``timestamp_basis``. + user_timestamp_us: int | None + frame_id: int | None + user_data: bytes | None + + +@dataclass +class StreamCounters: + """What one stream received and what was lost on the way to the consumer.""" + + received: int = 0 + #: Frames dropped because the consumer's queue was full. + queue_overflow: int = 0 + #: Frames the publisher numbered but that never arrived (SDK drops, network loss). + missing_frames: int = 0 + #: Times the publisher's frame id went backwards or repeated: a restarted publisher. + frame_id_resets: int = 0 + #: Times the stream's publication was replaced: a discontinuity, since the new one is a different source epoch. + publication_changes: int = 0 + #: The smallest step between consecutive frame ids seen. A publisher that paces its stream below the capture + #: rate numbers frames in steps (2, 4, 6...), which is not loss; only a deviation from this step is. + id_step: int | None = field(default=None, repr=False) + #: Whether the stream's track is currently subscribed. + subscribed: bool = False + last_frame_id: int | None = field(default=None, repr=False) + + +class LiveKitFrameReceiver: + """Joins the room of the offered streams and queues each stream's frames until the consumer takes them.""" + + def __init__( + self, + streams: Sequence[p.LiveKitCameraStream], + token: str, + *, + queue_frames: int = DEFAULT_QUEUE_FRAMES, + ) -> None: + # The package is optional; fail here with the fix rather than later with an AttributeError. + if rtc is None: + raise RuntimeError(LIVEKIT_MISSING) + if not streams: + raise ValueError("at least one LiveKit camera stream is required") + if len({(s.url, s.room, s.publisher) for s in streams}) != 1: + raise ValueError("all streams of one receiver must share a URL, room and publisher") + if queue_frames < 1: + raise ValueError("queue_frames must be at least 1") + self._streams = {s.stream_id: s for s in streams} + self._by_track = {s.track: s for s in streams} + self._url, self._publisher = streams[0].url, streams[0].publisher + self._token = token + 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} + # Set when a stream's track goes away or the room is lost, so a waiting consumer can fail instead of + # waiting forever; cleared when the track is subscribed again. + self._lost = {s.stream_id: asyncio.Event() for s in streams} + self._readers: dict[int, asyncio.Task[None]] = {} + # The id of the track each stream is currently read from, so an event about a replaced track is recognised. + self._active: dict[int, str] = {} + # Set once the room connection is lost for good; the SDK's own reconnects do not set it. + self.failed = asyncio.Event() + + async def start(self) -> None: + """Connect and begin receiving; frames of tracks published before or after joining are both picked up.""" + room = rtc.Room() + self._room = room + room.on("track_published", self._on_track_published) + room.on("track_subscribed", self._on_track_subscribed) + room.on("track_unsubscribed", self._on_track_unsubscribed) + room.on("participant_disconnected", self._on_participant_disconnected) + room.on("disconnected", self._on_disconnected) + # 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)) + # 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(): + self._subscribe_if_offered(participant, publication) + if publication.track is not None: + self._attach(participant, publication.track, publication.name) + + 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 + for task in self._readers.values(): + task.cancel() + for task in list(self._readers.values()): + with contextlib.suppress(asyncio.CancelledError): + await task + self._readers.clear() + if room is not None: + await room.disconnect() + for counters in self.counters.values(): + counters.subscribed = False + + async def next_frame(self, stream_id: int) -> ReceivedFrame: + """The oldest queued frame of ``stream_id``, waiting for one if none is queued.""" + return await self._queues[stream_id].get() + + async def next_frame_or_lost(self, stream_id: int, timeout: float | None = None) -> ReceivedFrame | None: + """The oldest queued frame of ``stream_id``, or None once its track is lost and nothing is queued. + + Frames already queued when the track goes away are still returned first, so a consumer drains what + arrived before it learns of the loss. With ``timeout`` set, raises ``asyncio.TimeoutError`` when neither a + frame nor a loss arrives in that many seconds: a track that stays published but sends nothing is not lost. + """ + queue = self._queues[stream_id] + if not queue.empty(): + return queue.get_nowait() + get = asyncio.ensure_future(queue.get()) + lost = asyncio.ensure_future(self._lost[stream_id].wait()) + try: + done, _pending = await asyncio.wait({get, lost}, timeout=timeout, return_when=asyncio.FIRST_COMPLETED) + if not done: + raise asyncio.TimeoutError + # A frame that arrived together with the loss is still a frame the consumer is owed. + return get.result() if get.done() else None + finally: + for task in (get, lost): + if not task.done(): + task.cancel() + + def clear_queued(self) -> None: + """Drop every queued frame, so a take starts with frames acquired after it began.""" + for queue in self._queues.values(): + while not queue.empty(): + queue.get_nowait() + + def pending(self, stream_id: int) -> int: + """How many frames of ``stream_id`` are queued for the consumer.""" + return self._queues[stream_id].qsize() + + def _subscribe_if_offered( + self, participant: rtc.RemoteParticipant, publication: rtc.RemoteTrackPublication + ) -> None: + """Subscribe to ``publication`` if it is an offered stream's track from the offered publisher.""" + if ( + participant.identity == self._publisher + and publication.name in self._by_track + and publication.kind == rtc.TrackKind.KIND_VIDEO + ): + publication.set_subscribed(True) + + def _on_track_published( + self, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant + ) -> None: + self._subscribe_if_offered(participant, publication) + + def _attach(self, participant: rtc.RemoteParticipant, track: rtc.Track, name: str) -> None: + """Start reading ``track`` if it is one of the offered streams, from the offered publisher.""" + stream = self._by_track.get(name) + if participant.identity != self._publisher or stream is None or track.kind != rtc.TrackKind.KIND_VIDEO: + return + # A republished track replaces the old reader; its frames must not interleave with the new ones, so what + # the old one queued is dropped, and the replacement is counted as a discontinuity the consumer can see. + previous = self._readers.pop(stream.stream_id, None) + if previous is not None: + previous.cancel() + if previous is not None or self._active.get(stream.stream_id, track.sid) != track.sid: + self.counters[stream.stream_id].publication_changes += 1 + queue = self._queues[stream.stream_id] + while not queue.empty(): + queue.get_nowait() + self._active[stream.stream_id] = track.sid + self.counters[stream.stream_id].subscribed = True + self._lost[stream.stream_id].clear() + # A new publication restarts the publisher's numbering, so a jump from the old track is not a loss. + self.counters[stream.stream_id].last_frame_id = None + self._readers[stream.stream_id] = asyncio.get_running_loop().create_task( + self._read(stream.stream_id, track) + ) + + def _detach(self, stream_id: int) -> None: + """Stop reading ``stream_id``: its track went away.""" + task = self._readers.pop(stream_id, None) + if task is not None: + task.cancel() + self._active.pop(stream_id, None) + self._mark_lost(stream_id) + + def _mark_lost(self, stream_id: int) -> None: + """Record that ``stream_id`` can deliver no more frames, without cancelling the calling reader task.""" + self.counters[stream_id].subscribed = False + self._lost[stream_id].set() + + def _on_track_subscribed( + self, track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant + ) -> None: + self._attach(participant, track, publication.name) + + def _on_track_unsubscribed( + self, track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant + ) -> None: + stream = self._by_track.get(publication.name) + if stream is None or participant.identity != self._publisher: + return + # An unsubscribe of a track that has since been replaced says nothing about the current one. + if self._active.get(stream.stream_id) != track.sid: + return + self._detach(stream.stream_id) + + def _on_participant_disconnected(self, participant: rtc.RemoteParticipant) -> None: + if participant.identity == self._publisher: + # Every offered stream, not only those already being read: one that never appeared is lost too. + for stream_id in self._streams: + self._detach(stream_id) + + def _on_disconnected(self, *_: object) -> None: + # Not a reconnect (the SDK reports those separately): the room is gone, and the recorder must find out. + for stream_id in self._streams: + self._detach(stream_id) + self.failed.set() + + async def _read(self, stream_id: int, track: rtc.Track) -> None: + """Take frames of one track from the SDK, convert them, and queue them for the consumer.""" + counters = self.counters[stream_id] + this_task = asyncio.current_task() + video = rtc.VideoStream(track, capacity=SDK_QUEUE_FRAMES, format=rtc.VideoBufferType.RGB24) + try: + async for event in video: + received_at = time.monotonic_ns() + frame = event.frame + image = np.frombuffer(frame.data, dtype=np.uint8).reshape(frame.height, frame.width, 3) + metadata = event.metadata + has_meta = metadata is not None + frame_id = metadata.frame_id if has_meta and metadata.HasField("frame_id") else None + self._account(counters, frame_id) + item = ReceivedFrame( + stream_id=stream_id, + image=image, + receive_time_ns=received_at, + transport_timestamp_us=event.timestamp_us, + user_timestamp_us=metadata.user_timestamp + if has_meta and metadata.HasField("user_timestamp") + else None, + frame_id=frame_id, + user_data=bytes(metadata.user_data) if has_meta and metadata.HasField("user_data") else None, + ) + try: + self._queues[stream_id].put_nowait(item) + except asyncio.QueueFull: + # Never block the reader or skip silently: the consumer is behind, and the loss is counted. + counters.queue_overflow += 1 + # The SDK ended the stream normally: nothing more will arrive, and a consumer must not wait for it. + if self._readers.get(stream_id) is this_task: + self._readers.pop(stream_id) + self._mark_lost(stream_id) + except Exception: + # A dead reader would leave the consumer waiting forever; marking the stream lost lets the take settle. + # Only the reader still registered for the stream may do so: a replaced one is already superseded. + logger.exception("LiveKit reader for stream %d failed", stream_id) + if self._readers.get(stream_id) is this_task: + self._readers.pop(stream_id) + self._mark_lost(stream_id) + finally: + await video.aclose() + + @staticmethod + def _account(counters: StreamCounters, frame_id: int | None) -> None: + """Count a received frame, and any frames the publisher numbered that did not arrive before it.""" + counters.received += 1 + if frame_id is None: + return + last = counters.last_frame_id + if last is not None: + if frame_id <= last: + # The publisher restarted its numbering: a new run, not a loss. + counters.frame_id_resets += 1 + else: + step = frame_id - last + counters.id_step = step if counters.id_step is None else min(counters.id_step, step) + # A publisher that paces its stream steps its ids by a constant; only a bigger jump is loss. + counters.missing_frames += max(0, round(step / counters.id_step) - 1) + counters.last_frame_id = frame_id diff --git a/examples/egocap/backend/egocap_web/livekit_recording.py b/examples/egocap/backend/egocap_web/livekit_recording.py new file mode 100644 index 0000000000..7aae0a79b5 --- /dev/null +++ b/examples/egocap/backend/egocap_web/livekit_recording.py @@ -0,0 +1,412 @@ +# SPDX-FileCopyrightText: 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# SPDX-License-Identifier: Apache-2.0 + +"""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. + +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. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import ipaddress +import logging +import time +from collections.abc import Callable, Mapping, Sequence +from typing import Any +from urllib.parse import urlparse + +from capture_server import protocol as p +from recording_service import PixelFormat, VideoSample + +from .livekit_receiver import LIVEKIT_MISSING, LiveKitFrameReceiver, livekit_available +from .recording_adapter import RecordingIngress + +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() + + +def is_loopback_url(url: str) -> bool: + """Whether ``url`` names this machine: ``localhost`` or a loopback IP literal.""" + host = urlparse(url).hostname or "" + if host.lower() == "localhost": + return True + try: + return ipaddress.ip_address(host).is_loopback + except ValueError: + return False + + +def unrecordable_reason(started: Any, capture_server_url: str | None = None) -> str | None: + """Why a started session's LiveKit streams cannot all be recorded, naming the first that cannot; else None. + + A stream that cannot be recorded is never left out of a take silently: recording is refused until it is + removed from the streams file or made recordable. When ``capture_server_url`` is given and is not a + loopback address, every LiveKit stream is refused because the clocks cannot be assumed to be shared. + """ + if started is None: + return None + for stream in started.streams: + if not isinstance(stream, p.LiveKitCameraStream): + continue + # Without the optional package no LiveKit stream can be received, so say so before anything else. + if not livekit_available(): + return f"LiveKit stream {stream.name!r}: {LIVEKIT_MISSING}" + if capture_server_url is not None and not is_loopback_url(capture_server_url): + return ( + 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: + return ( + f"LiveKit stream {stream.name!r}: its timestamps are not a capture time, 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" + return None + + +class LiveKitRecordingFeed: + """Moves every offered stream's frames from the receiver into the ingress of one take. + + Stops, with the reason in ``failure``, at the first frame that cannot be recorded honestly or when the ingress + stops accepting; the recorder settles the take from that, so a source never goes missing silently. + """ + + def __init__( + self, + receiver: LiveKitFrameReceiver, + streams: Sequence[p.LiveKitCameraStream], + ingress: RecordingIngress, + source_names: Mapping[int, str], + *, + clock_map: UnixClockMap | None = None, + on_failure: Callable[[str], None] | None = None, + ) -> None: + missing = [s.name for s in streams if s.stream_id not in source_names] + if missing: + raise ValueError(f"no recording source for LiveKit stream(s): {missing}") + self._receiver = receiver + 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() + self._on_failure = on_failure + self._tasks: list[asyncio.Task[None]] = [] + self.failure: str | None = None + self._baseline: dict[int, tuple[int, int]] = {} + + def loss(self) -> dict[str, dict[str, int]]: + """What the receiver lost on the way to this take, per stream name (counted since the take began).""" + out: dict[str, dict[str, int]] = {} + for stream_id, stream in self._streams.items(): + counters = self._receiver.counters[stream_id] + overflow, changes = self._baseline.get(stream_id, (0, 0)) + out[stream.name] = { + "received": counters.received, + "queue_overflow": counters.queue_overflow - overflow, + "publication_changes": counters.publication_changes - changes, + "missing_frames": counters.missing_frames, + "frame_id_resets": counters.frame_id_resets, + } + return out + + def start(self) -> None: + """Begin feeding; one task per stream so a slow stream never delays another.""" + for stream_id in self._streams: + counters = self._receiver.counters[stream_id] + self._baseline[stream_id] = (counters.queue_overflow, counters.publication_changes) + loop = asyncio.get_running_loop() + self._tasks = [loop.create_task(self._run(stream_id)) for stream_id in self._streams] + + async def stop(self) -> None: + """Stop feeding. Frames already offered to the ingress stay with it.""" + for task in self._tasks: + task.cancel() + for task in self._tasks: + with contextlib.suppress(asyncio.CancelledError): + _ = await task + self._tasks = [] + + def _fail(self, reason: str) -> None: + """Record the first reason feeding had to stop, once, and tell the recorder.""" + if self.failure is None: + self.failure = reason + logger.error("LiveKit recording feed stopped: %s", reason) + if self._on_failure is not None: + self._on_failure(reason) + + 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: + # 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" + ) + 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. + self._fail( + f"{stream.name}: no frame for {_MAX_SILENCE_S:.0f} s although the track is still published" + ) + return + if frame is None: + # Whatever was queued has been taken; the camera is gone, and a recorded source must not vanish. + self._fail(f"{stream.name}: the LiveKit track was lost") + return + # What the receiver lost since this take began: video that was not recorded must not look like video + # that was, so a gap or a replaced publication ends the take with its reason. + counters = self._receiver.counters[stream_id] + overflow, changes = self._baseline[stream_id] + if counters.queue_overflow > overflow: + self._fail( + f"{stream.name}: the recorder fell behind and dropped " + f"{counters.queue_overflow - overflow} frame(s)" + ) + return + if counters.publication_changes > changes: + 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") + return + capture_time = self._clock_map.monotonic_seconds(frame.user_timestamp_us) + 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. + self._fail( + f"{stream.name}: the publisher's clock disagrees with this host's (frame age {age:+.3f} s)" + ) + return + sample = VideoSample( + source=source, + image=frame.image, + pixel_format=PixelFormat.RGB8, + capture_time=capture_time, + receive_time=receive_time, + ) + if not self._ingress.offer(sample): + # The ingress refuses once it has failed or stopped; it reports its own reason. + self._fail(f"{stream.name}: the recording ingress is no longer accepting frames") + return + + +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 + schema fixes each source's size and rate when the session opens. The rest remain preview-only. + """ + if started is None: + return () + return tuple( + s + for s in started.streams + if isinstance(s, p.LiveKitCameraStream) + and s.timestamp_basis is p.LiveKitTimestampBasis.SENSOR_CAPTURE_UNIX_US + and None not in (s.width, s.height, s.fps) + ) + + +class LiveKitRecordingSource: + """The recorder's LiveKit video for one take: joins before the take, feeds during it, leaves after it. + + The recordable streams are read when the take is prepared, because the session offers them only once it has + started. The first frame of each stream is checked against its configured size. The browser's own connection + is separate and unaffected. + """ + + def __init__( + self, + recordable_streams: Callable[[], Sequence[p.LiveKitCameraStream]], + make_token: Callable[[str], str], + *, + first_frame_timeout_s: float = 10.0, + join_timeout_s: float = _JOIN_TIMEOUT_S, + leave_timeout_s: float = _LEAVE_TIMEOUT_S, + ) -> None: + self._recordable_streams = recordable_streams + self._make_token = make_token + self._first_frame_timeout_s = first_frame_timeout_s + self._join_timeout_s = join_timeout_s + self._leave_timeout_s = leave_timeout_s + # Room joins and leaves that outlived their deadline; held so they are not garbage-collected mid-flight. + self._background: set[asyncio.Future[None]] = set() + self._streams: tuple[p.LiveKitCameraStream, ...] = () + self._receiver: LiveKitFrameReceiver | None = None + self._feed: LiveKitRecordingFeed | None = None + + async def prepare(self) -> None: + """Join the room and wait for a usable first frame from every stream; raises if one never arrives. + + Runs before the take's storage opens, so a missing camera fails the start instead of a half-written take. + Does nothing when the session offers no recordable stream. + """ + self._streams = tuple(self._recordable_streams()) + if not self._streams: + return + receiver = LiveKitFrameReceiver(self._streams, self._make_token(self._streams[0].room)) + self._receiver = receiver + try: + await self._join(receiver) + for stream in self._streams: + frame = await asyncio.wait_for( + receiver.next_frame_or_lost(stream.stream_id), self._first_frame_timeout_s + ) + if frame is None: + raise RuntimeError(f"{stream.name}: the LiveKit track was lost before recording started") + if frame.image.shape[:2] != (stream.height, stream.width): + raise RuntimeError( + f"{stream.name}: frames are {frame.image.shape[1]}x{frame.image.shape[0]}, " + f"expected {stream.width}x{stream.height}" + ) + except asyncio.TimeoutError as exc: + await self.stop() + raise RuntimeError("a LiveKit stream sent no frame before the recording could start") from exc + except BaseException: + await self.stop() + raise + + async def _join(self, receiver: LiveKitFrameReceiver) -> None: + """Run ``receiver.start()`` with a deadline on the wait, never on the join itself. + + Cancelling ``Room.connect`` can strand the SDK's room-event handshake, so a join that is late (or whose + caller is cancelled) keeps running and the receiver is retired once it finishes. + """ + join = asyncio.ensure_future(receiver.start()) + try: + _ = await asyncio.wait({join}, timeout=self._join_timeout_s) + except BaseException: + self._retire_after_join(join, receiver) + raise + if not join.done(): + self._retire_after_join(join, receiver) + raise RuntimeError(f"joining the LiveKit room took longer than {self._join_timeout_s:g} s") + # A join that failed raises here, with the receiver still ours to stop. + join.result() + + def _retire_after_join(self, join: asyncio.Future[None], receiver: LiveKitFrameReceiver) -> None: + """Hand an unfinished join to a background task that leaves the room as soon as the join ends.""" + if join.done(): + return + # This source no longer owns the receiver: stop() must not leave a room that is still being joined. + self._receiver = None + self._background.add(join) + + def _leave(done: asyncio.Future[None]) -> None: + self._background.discard(done) + if not done.cancelled() and done.exception() is not None: + logger.warning("late LiveKit room join failed: %s", done.exception()) + self._track(asyncio.ensure_future(receiver.stop()), "leaving the room after a late join") + + join.add_done_callback(_leave) + + def _track(self, task: asyncio.Future[None], what: str) -> None: + """Keep a background task referenced until it ends, logging a failure instead of losing it.""" + self._background.add(task) + + def _done(finished: asyncio.Future[None]) -> None: + self._background.discard(finished) + if not finished.cancelled() and finished.exception() is not None: + logger.warning("%s failed: %s", what, finished.exception()) + + task.add_done_callback(_done) + + 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: + return + # Frames queued while storage was opening were acquired before the take started. + 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._feed.start() + + 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 + self._feed = self._receiver = None + if feed is not None: + await feed.stop() + if receiver is not None: + # A stalled room leave must not block retiring the take's ingress; it is left to finish on its own. + leave = asyncio.ensure_future(receiver.stop()) + self._track(leave, "leaving the LiveKit room") + _ = await asyncio.wait({leave}, timeout=self._leave_timeout_s) + if not leave.done(): + logger.warning( + "leaving the LiveKit room took longer than %g s; continuing teardown", self._leave_timeout_s + ) diff --git a/examples/egocap/backend/egocap_web/recording_adapter.py b/examples/egocap/backend/egocap_web/recording_adapter.py index deb0ad6173..8e19c9b769 100644 --- a/examples/egocap/backend/egocap_web/recording_adapter.py +++ b/examples/egocap/backend/egocap_web/recording_adapter.py @@ -348,11 +348,15 @@ def convert_capture_frame(frame: CaptureFrame) -> list[NumericSample]: # --- bounded ingress queue/consumer ------------------------------------------- -RawItem = CaptureFrame | CameraImage +RawItem = CaptureFrame | CameraImage | VideoSample def _raw_item_bytes(raw_item: RawItem) -> int: - """The memory one queued raw item pins: the compressed JPEG or the raw tracking blob.""" + """The memory one queued raw item pins: the compressed JPEG, the raw tracking blob or a decoded frame.""" + image = getattr(raw_item, "image", None) + if image is not None: + # A decoded frame (a VideoSample from LiveKit) pins its pixels, not a compressed payload. + return image.nbytes return len(raw_item.jpeg) if isinstance(raw_item, CameraImage) else len(raw_item.payload) @@ -365,6 +369,9 @@ def _convert_raw_item(raw_item: RawItem) -> list[NumericSample | VideoSample]: return list(convert_capture_frame(raw_item)) if isinstance(raw_item, CameraImage): return [convert_camera_image(raw_item, raw_item.name)] + if isinstance(raw_item, VideoSample): + # Already a decoded frame with its capture time (LiveKit video): nothing to convert. + return [raw_item] raise TypeError(f"unsupported raw item type: {type(raw_item).__name__}") diff --git a/examples/egocap/backend/egocap_web/recording_session.py b/examples/egocap/backend/egocap_web/recording_session.py index 91a6fa5cc2..6737097556 100644 --- a/examples/egocap/backend/egocap_web/recording_session.py +++ b/examples/egocap/backend/egocap_web/recording_session.py @@ -56,7 +56,7 @@ from dataclasses import dataclass, replace from enum import Enum from functools import partial -from typing import Any, TypeVar +from typing import TYPE_CHECKING, Any, TypeVar from capture_server.client import CameraImage, CaptureFrame from egocap_web.recording_adapter import RecordingIngress, validate_tracking_descriptor @@ -68,6 +68,9 @@ RecordingPhase, ServiceConfig, ) + +if TYPE_CHECKING: + from egocap_web.livekit_recording import LiveKitRecordingSource from recording_service import RecordingService as _RecordingService from recording_service import RecordingSnapshot as ServiceSnapshot @@ -180,11 +183,15 @@ def __init__( config_factory: Callable[[], ServiceConfig], *, queue_capacity: int, - queue_bytes: int | None = None, + queue_bytes: int | Callable[[], int] | None = None, readiness_check: ReadinessCheck, initial_context: TaskContext | None = None, + livekit_source: LiveKitRecordingSource | None = None, ) -> None: self._config_factory = config_factory + # 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 self._queue_capacity = queue_capacity self._queue_bytes = queue_bytes self._readiness_check = readiness_check @@ -408,6 +415,21 @@ 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") + 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. + try: + await self._livekit_source.prepare() + except Exception as exc: # noqa: BLE001 -- any prepare failure is a refused start, never a half-open take + raise CoordinatorBusyError(f"LiveKit video is not ready: {exc}") from exc + try: + return await self._start_episode_with_storage(frozen_context) + except BaseException: + # A start that fails or is cancelled after prepare must not leave the room joined. + await self._livekit_source.stop() + raise + + async def _start_episode_with_storage(self, frozen_context: TaskContext) -> EpisodeInfo: service = await self._open_storage_if_needed() # Recheck readiness/closing intent right after storage finished # opening -- the same "don't trust state from before the last @@ -438,14 +460,20 @@ async def _start_episode_impl(self, frozen_context: TaskContext) -> EpisodeInfo: await service.discard_episode(episode.episode_id) raise CoordinatorBusyError("start was superseded before activation") episode_id = episode.episode_id + # A callable is resolved per take: LiveKit frame sizes are known only once the session has started. ingress = RecordingIngress( queue_capacity=self._queue_capacity, - queue_bytes=self._queue_bytes, + queue_bytes=self._queue_bytes() if callable(self._queue_bytes) else self._queue_bytes, submit=partial(service.submit_sample, episode_id), on_failure=lambda message: self._schedule_failure_settlement(service, episode_id, message), ) await ingress.start() self._current_ingress = ingress + if self._livekit_source is not None: + # A feed failure settles the take through the same path as an ingress failure. + self._livekit_source.start( + ingress, lambda message: self._schedule_failure_settlement(service, episode_id, message) + ) if self.on_episode_started is not None: self.on_episode_started() return episode @@ -609,6 +637,9 @@ async def _retire_current_ingress(self) -> _IngressRetirement: ingress = self._current_ingress if ingress is None: return _IngressRetirement(settled=True, failure=None) + if self._livekit_source is not None: + # The producer stops first, so nothing is offered to an ingress that is draining. + await self._livekit_source.stop() settled = await ingress.stop() if not settled: return _IngressRetirement( diff --git a/examples/egocap/capture_server/README.md b/examples/egocap/capture_server/README.md index 577bfc3c39..11e4b6b318 100644 --- a/examples/egocap/capture_server/README.md +++ b/examples/egocap/capture_server/README.md @@ -586,7 +586,7 @@ All enums are IntEnum classes. Supply enum members to the Python API; incoming n | `StartupStage` | `INITIALIZING=1`, `WAITING_FOR_HEADSET=2`, `CONNECTING=3`, `STARTING_SOURCES=4` | | `StopReason` | `REQUESTED=1`, `OWNER_DISCONNECTED=4`, `SERVER_SHUTDOWN=6`, `IC_ERROR=7`, `IC_EXITED=8`, `CAMERA_ERROR=9` | | `RejectionReason` | `INVALID_CONFIG=1`, `BUSY=2`, `UNKNOWN_SESSION=3` | -| `StreamKind` | `TRACKING=1`, `IMAGE=2` | +| `StreamKind` | `TRACKING=1`, `IMAGE=2`, `LIVEKIT_IMAGE=3` | | `BinaryKind` | `CAPTURE_FRAME=1`, `CAPTURE_IMAGE=2` | | `TrackingSchema` | `IC_TRACKING_V1=1` | | `WireDType` | `UINT8=1`, `FLOAT32_LE=2`, `FLOAT64_LE=3` | @@ -594,6 +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` | | `SampleTimeSemantics` | `HOST_POLL=1` | | `PoseRepresentation` | `POSITION_QUATERNION=1` | | `QuaternionOrder` | `XYZW=1` | @@ -928,7 +929,7 @@ This is a complete SessionInit for head, controllers and hands: | `clock.domain` | `ClockDomain`: `CAPTURE_HOST_MONOTONIC=1`. | Times use the capture computer's elapsed-time clock; they do not use Unix/calendar time or the client's elapsed-time clock. | | `clock.unit` | `TimeUnit`: `NANOSECONDS=1`. | An interval of 1,000,000,000 timestamp units means one second. | | `clock.sample_time_semantics` | `SampleTimeSemantics`: `HOST_POLL=1`. | The tracking timestamp describes the host's source-sampling step. It does not claim to be a sensor exposure time or an SDK callback timestamp. | -| `streams` | Array of tracking and camera descriptor objects. | Defines the stream IDs, formats and tracking byte layout the client must register from SessionInit before any binary messages. An empty array is valid only when no streams were requested. | +| `streams` | Array of tracking, camera and LiveKit camera descriptor objects. | Defines the stream IDs, formats and tracking byte layout the client must register from SessionInit before any binary messages. An empty array is valid only when no streams were requested. | The client validates and caches this decoding contract. A layout/format change requires a new session. Unknown camera intrinsics can initially be null; later measured intrinsics are metadata, not a byte-layout change. @@ -1081,7 +1082,7 @@ Dotted field names preserve the public IC group/field identity. Dtype codes are | Tracking descriptor field | Python enum and accepted values, or ordinary type | Meaning | |---|---|---| | `stream_id` | Not an enum: integer 0 for tracking. | Associates a binary message's header with this descriptor. | -| `kind` | `StreamKind`: `TRACKING=1` for this descriptor; `IMAGE=2` for camera descriptors. | Selects the kind of decoder to use for the stream. | +| `kind` | `StreamKind`: `TRACKING=1` for this descriptor; `IMAGE=2` for camera descriptors; `LIVEKIT_IMAGE=3` for LiveKit camera descriptors. | Selects the kind of decoder to use for the stream. | | `name` | String, `tracking` for this stream; no enum. | Provides a readable stream label. Routing uses stream\_id, not this label. | | `schema` | `TrackingSchema`: `IC_TRACKING_V1=1`. | Identifies the complete field contract, including presence/validity meanings and the 1995-byte layout. A client rejects an unsupported schema rather than interpreting a similar-looking blob. | | `sources` | `SourceType`: unique array drawn from `HEAD=1`, `CONTROLLERS=2`, `HANDS=3`. | Records the source channels selected for this tracking stream. Unselected channels retain inert slots in the fixed binary layout. | @@ -1215,6 +1216,42 @@ The camera descriptor repeats name and type from the accepted request and adds t | `orientation.mirrored` | Boolean false. | Advertises no server-applied mirror operation. | | `intrinsics` | Null or a finite numeric array of shape \[3,3\]; not an enum. | A camera projection matrix for these image dimensions. Its focal-length terms express how rays project into pixels, and its principal-point terms locate the image's optical center in pixel coordinates. Null means the server supplies no such calibration; it must not be interpreted as an identity camera matrix. | +#### LiveKit camera descriptors + +A `streams` entry with `kind` `LIVEKIT_IMAGE=3` says where a camera's video is published in a LiveKit room. The server never sends these frames; each consumer joins the room itself and learns the actual dimensions and frame rate from the video. A descriptor does not mean frames are arriving. A client that does not know kind 3 rejects the whole SessionInit, so a server started with `--livekit-streams` cannot be used by older clients. + +| LiveKit descriptor field | Python enum and accepted values, or ordinary type | Meaning | +|---|---|---| +| `stream_id` | Not an enum: integer 16–23. | Identifies the entry in the one stream-id space shared with the tracking and JPEG camera streams. | +| `kind` | `StreamKind`: `LIVEKIT_IMAGE=3`. | Selects this descriptor. | +| `name` | Nonempty string of at most 512 characters. | Operator-facing camera name. It must differ from every other LiveKit entry's name and from every JPEG camera name. | +| `url` | Nonempty string of at most 512 characters starting with `ws://` or `wss://`. | The LiveKit server to join. | +| `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. | +| `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. | + +At most 8 LiveKit entries are allowed, and all of them must share one `url`, `room` and `publisher`: consumers join a single room as a single publisher. Both the server (at startup) and the client (when parsing SessionInit) enforce these rules. + +##### The `--livekit-streams` file + +`--livekit-streams FILE` loads the entries from a YAML file. The server exits with code 2 and a message on stderr if the file cannot be read or is invalid. Stream ids are assigned 16, 17, ... in file order. Unknown keys are errors. + +```yaml +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 + 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. + 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. Both binary families use `struct.Struct("<4sBBHQQQ")`: diff --git a/examples/egocap/capture_server/capture_server/client.py b/examples/egocap/capture_server/capture_server/client.py index 5e3eb228e9..237de9589c 100644 --- a/examples/egocap/capture_server/capture_server/client.py +++ b/examples/egocap/capture_server/capture_server/client.py @@ -134,7 +134,7 @@ class CaptureStarted: mode: p.CaptureMode effective_config: p.CaptureConfiguration clock: p.CaptureClock - streams: tuple[FrameDescription | p.CameraStream, ...] + streams: tuple[FrameDescription | p.CameraStream | p.LiveKitCameraStream, ...] def _to_client_started(wire: p.SessionInit) -> CaptureStarted: @@ -394,8 +394,9 @@ def from_session_init(cls, event: p.SessionInit) -> _StreamRegistry: for stream in event.streams: if isinstance(stream, p.TrackingStream): registry.tracking = stream - else: + elif isinstance(stream, p.CameraStream): registry.cameras[stream.stream_id] = stream + # A LiveKit camera is only located here, never decoded from a binary message: nothing to register. return registry diff --git a/examples/egocap/capture_server/capture_server/livekit_streams.py b/examples/egocap/capture_server/capture_server/livekit_streams.py new file mode 100644 index 0000000000..77e182b828 --- /dev/null +++ b/examples/egocap/capture_server/capture_server/livekit_streams.py @@ -0,0 +1,91 @@ +# SPDX-FileCopyrightText: 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# SPDX-License-Identifier: Apache-2.0 + +"""The LiveKit streams a station offers, read from the ``--livekit-streams`` YAML file. + +Only locates the video (server URL, room, publisher identity and track names). The server never joins the room: +each consumer connects to LiveKit itself and learns the live details there. + +.. code-block:: yaml + + livekit: + url: wss://LIVEKIT_HOST + room: sensing-preview + publisher: orin-sensing + timestamp_basis: processing_unix_us # what the publisher puts in the frame timestamp (see protocol.py) + streams: + - {name: Head left, track: sensing-left, width: 2560, height: 1984, fps: 30} +""" + +from __future__ import annotations + +from pathlib import Path + +import yaml + +from . import protocol as p + +_ROOM_KEYS = {"url", "room", "publisher", "timestamp_basis", "streams"} +_BASES = { + "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, +} +_STREAM_KEYS = {"name", "track", "width", "height", "fps"} + + +def _mapping(value: object, where: str, allowed: set[str]) -> dict: + """A YAML mapping whose keys are all in ``allowed``; a typo is an error, not a silently ignored setting.""" + if not isinstance(value, dict): + raise ValueError(f"{where} must be a mapping") + unknown = set(value) - allowed + if unknown: + raise ValueError(f"{where} has unknown key(s): {sorted(unknown)}") + return value + + +def load_livekit_streams(path: Path) -> tuple[p.LiveKitCameraStream, ...]: + """The streams named by the YAML file at ``path``; raises ``ValueError`` for anything unusable.""" + try: + text = path.read_text(encoding="utf-8") + except OSError as exc: + raise ValueError(f"cannot read LiveKit streams file {path}: {exc}") from exc + try: + data = yaml.safe_load(text) + except yaml.YAMLError as exc: + raise ValueError(f"LiveKit streams file {path} is not valid YAML: {exc}") from exc + room = _mapping(_mapping(data, "the file", {"livekit"}).get("livekit"), "livekit", _ROOM_KEYS) + raw_streams = room.get("streams") + if not isinstance(raw_streams, list): + raise ValueError("livekit.streams must be a list") + basis_name = room.get("timestamp_basis", "monotonic_us") + # A YAML list or mapping is unhashable, so check the type before the lookup. + if not isinstance(basis_name, str) or basis_name not in _BASES: + raise ValueError(f"livekit.timestamp_basis must be one of {sorted(_BASES)}") + streams: list[p.LiveKitCameraStream] = [] + try: + for index, raw in enumerate(raw_streams): + entry = _mapping(raw, f"livekit.streams[{index}]", _STREAM_KEYS) + # The wire parser is the one validator: a file that loads is a list every client will accept. + streams.append( + p.LiveKitCameraStream.from_json( + { + "stream_id": p.LIVEKIT_STREAM_ID_BASE + index, + "kind": int(p.StreamKind.LIVEKIT_IMAGE), + "name": entry.get("name"), + "url": room.get("url"), + "room": room.get("room"), + "publisher": room.get("publisher"), + "track": entry.get("track"), + "timestamp_basis": int(_BASES[basis_name]), + "width": entry.get("width"), + "height": entry.get("height"), + "fps": entry.get("fps"), + } + ) + ) + p.check_livekit_streams(streams) + except p.ProtocolError as exc: + raise ValueError(f"LiveKit streams file {path}: {exc}") from exc + return tuple(streams) diff --git a/examples/egocap/capture_server/capture_server/protocol.py b/examples/egocap/capture_server/capture_server/protocol.py index 767d7844c2..3b5c9fe7a0 100644 --- a/examples/egocap/capture_server/capture_server/protocol.py +++ b/examples/egocap/capture_server/capture_server/protocol.py @@ -28,6 +28,7 @@ from __future__ import annotations +import math import re import struct from dataclasses import dataclass @@ -165,6 +166,8 @@ class CalibrationStopReason(IntEnum): class StreamKind(IntEnum): TRACKING = 1 IMAGE = 2 + # A camera offered through a LiveKit room: described here, never sent as frames by this server. + LIVEKIT_IMAGE = 3 class BinaryKind(IntEnum): @@ -198,6 +201,20 @@ class TimeUnit(IntEnum): NANOSECONDS = 1 +class LiveKitTimestampBasis(IntEnum): + """What a LiveKit frame's ``user_timestamp_us`` means. Only the last is a capture time.""" + + # The publisher host's monotonic clock in microseconds, read when the frame is processed. Not on any clock + # this host can convert. + 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. + SENSOR_CAPTURE_UNIX_US = 3 + + class SampleTimeSemantics(IntEnum): HOST_POLL = 1 @@ -896,6 +913,133 @@ def camera_stream_descriptor(camera: CameraStreamConfig) -> CameraStream: ) +# --------------------------------------------------------------------------- +# LiveKit stream descriptors. Separate from the tracking/camera ``streams`` +# above: these are not frames CaptureServer produces. Video goes straight from +# the publisher to each consumer through a LiveKit room; the descriptor only says +# where to find it and what it contains. +# --------------------------------------------------------------------------- + +_MAX_LIVEKIT_STREAMS = 8 +_MAX_LIVEKIT_TEXT_LEN = 512 +#: LiveKit streams take stream_ids above the JPEG cameras' 1-3 so one registry has one id space. +LIVEKIT_STREAM_ID_BASE = 16 + + +def _require_text(value: Any, context: str) -> str: + """A nonempty bounded string field.""" + if not isinstance(value, str) or not value or len(value) > _MAX_LIVEKIT_TEXT_LEN: + raise ProtocolError(f"{context} must be a nonempty string of at most {_MAX_LIVEKIT_TEXT_LEN} characters") + return value + + +def _optional_positive_int(value: Any, context: str) -> int | None: + """A positive integer, or None when the value is not known.""" + if value is None: + return None + if type(value) is not int or value <= 0: + raise ProtocolError(f"{context} must be a positive integer or null") + return value + + +@dataclass(frozen=True) +class LiveKitCameraStream: + """One camera offered through a LiveKit room: a ``streams`` entry that locates video instead of carrying it. + + ``width``/``height``/``fps`` are the configured values, or None when not known; a consumer learns the + actual values from the video itself. Offering a stream does not mean frames are arriving. + """ + + stream_id: int + name: str + url: str + room: str + publisher: str + track: str + timestamp_basis: LiveKitTimestampBasis = LiveKitTimestampBasis.PUBLISHER_MONOTONIC_US + width: int | None = None + height: int | None = None + fps: float | None = None + kind: StreamKind = dataclass_field(default=StreamKind.LIVEKIT_IMAGE, init=False) + + def to_json(self) -> dict[str, Any]: + return { + "stream_id": self.stream_id, + "kind": int(self.kind), + "name": self.name, + "url": self.url, + "room": self.room, + "publisher": self.publisher, + "track": self.track, + "timestamp_basis": int(self.timestamp_basis), + "width": self.width, + "height": self.height, + "fps": self.fps, + } + + @classmethod + def from_json(cls, data: Any) -> LiveKitCameraStream: + keys = { + "stream_id", "kind", "name", "url", "room", "publisher", "track", "timestamp_basis", "width", "height", + "fps", + } # fmt: skip + _require_keys(data, keys, "a LiveKit camera stream descriptor") + if data.get("kind") != int(StreamKind.LIVEKIT_IMAGE): + raise ProtocolError("LiveKit camera stream.kind must be LIVEKIT_IMAGE") + stream_id = data.get("stream_id") + if type(stream_id) is not int or not ( + LIVEKIT_STREAM_ID_BASE <= stream_id < LIVEKIT_STREAM_ID_BASE + _MAX_LIVEKIT_STREAMS + ): + raise ProtocolError( + f"LiveKit camera stream.stream_id must be an integer from {LIVEKIT_STREAM_ID_BASE} to " + f"{LIVEKIT_STREAM_ID_BASE + _MAX_LIVEKIT_STREAMS - 1}" + ) + url = _require_text(data.get("url"), "LiveKit camera stream.url") + if not url.startswith(("ws://", "wss://")): + raise ProtocolError("LiveKit camera stream.url must start with ws:// or wss://") + fps = data.get("fps") + # NaN and infinity pass an ordinary `<= 0` check but are not valid JSON numbers, so require a finite value. + if fps is not None and ( + isinstance(fps, bool) or not isinstance(fps, (int, float)) or not math.isfinite(fps) or fps <= 0 + ): + raise ProtocolError("LiveKit camera stream.fps must be a finite positive number or null") + return cls( + stream_id=stream_id, + name=_require_text(data.get("name"), "LiveKit camera stream.name"), + url=url, + room=_require_text(data.get("room"), "LiveKit camera stream.room"), + publisher=_require_text(data.get("publisher"), "LiveKit camera stream.publisher"), + track=_require_text(data.get("track"), "LiveKit camera stream.track"), + timestamp_basis=_require_int_enum( + data.get("timestamp_basis"), LiveKitTimestampBasis, "LiveKit camera stream.timestamp_basis" + ), + width=_optional_positive_int(data.get("width"), "LiveKit camera stream.width"), + height=_optional_positive_int(data.get("height"), "LiveKit camera stream.height"), + fps=None if fps is None else float(fps), + ) + + +def check_livekit_streams(streams: Any) -> None: + """Reject a registry whose LiveKit entries are too many, ambiguous, or spread over several rooms. + + Consumers join one room as one publisher, so every entry must share the same url, room and publisher. + Names are checked against the JPEG cameras too: one camera name must mean one stream. Stream ids are unique + across the whole registry already (checked by the caller). + """ + livekit = [s for s in streams if isinstance(s, LiveKitCameraStream)] + if len(livekit) > _MAX_LIVEKIT_STREAMS: + raise ProtocolError(f"at most {_MAX_LIVEKIT_STREAMS} LiveKit camera streams may be offered") + if len({(s.url, s.room, s.publisher) for s in livekit}) > 1: + raise ProtocolError("LiveKit camera streams must all use the same url, room and publisher") + other_names = {s.name for s in streams if isinstance(s, CameraStream)} + names = [s.name for s in livekit] + if len(names) != len(set(names)) or other_names & set(names): + raise ProtocolError("LiveKit camera stream names must be unique among camera streams") + locations = [(s.publisher, s.track) for s in livekit] + if len(locations) != len(set(locations)): + raise ProtocolError("two LiveKit camera streams name the same publisher and track") + + # --------------------------------------------------------------------------- # Shared JSON validation helpers, used only by the wire-parsing functions # below (parse_begin/parse_stop) -- the only two families CaptureServer @@ -1669,13 +1813,28 @@ def _parse_optional_status(data: Any) -> StatusSnapshot | None: # --------------------------------------------------------------------------- +def _parse_stream(data: Any) -> TrackingStream | CameraStream | LiveKitCameraStream: + """One entry of SessionInit's ``streams``, dispatched on its ``kind``. + + A non-dict entry (e.g. a string or list, however malformed) still reaches a ``from_json`` and its own + ``_require_keys`` check, which raises a clean ProtocolError, never AttributeError from calling ``.get`` on + something that is not a dict. + """ + kind = data.get("kind") if isinstance(data, dict) else None + if kind == int(StreamKind.TRACKING): + return TrackingStream.from_json(data) + if kind == int(StreamKind.LIVEKIT_IMAGE): + return LiveKitCameraStream.from_json(data) + return CameraStream.from_json(data) + + @dataclass(frozen=True) class SessionInit: session_id: str mode: CaptureMode effective_config: CaptureConfiguration clock: CaptureClock - streams: tuple[TrackingStream | CameraStream, ...] + streams: tuple[TrackingStream | CameraStream | LiveKitCameraStream, ...] type: MessageType = dataclass_field(default=MessageType.SESSION_INIT, init=False) version: int = dataclass_field(default=PROTOCOL_VERSION, init=False) @@ -1704,20 +1863,11 @@ def from_json(cls, data: Any) -> SessionInit: streams_raw = data.get("streams") if not isinstance(streams_raw, list): raise ProtocolError("SessionInit.streams must be a list") - streams = tuple( - TrackingStream.from_json(s) - # isinstance guard first: a non-dict entry (e.g. a string or - # list, however malformed) must still reach *_.from_json's own - # _require_keys check and raise a clean ProtocolError there, - # never AttributeError from calling .get on something that - # isn't a dict. - if isinstance(s, dict) and s.get("kind") == int(StreamKind.TRACKING) - else CameraStream.from_json(s) - for s in streams_raw - ) + streams = tuple(_parse_stream(s) for s in streams_raw) stream_ids = [s.stream_id for s in streams] if len(stream_ids) != len(set(stream_ids)): raise ProtocolError(f"SessionInit.streams has duplicate stream_id(s): {stream_ids}") + check_livekit_streams(streams) return cls( session_id=session_id, mode=mode, @@ -1732,7 +1882,7 @@ def session_init_event( session_id: str, mode: CaptureMode, begin: BeginRequest, - streams: list[TrackingStream | CameraStream], + streams: list[TrackingStream | CameraStream | LiveKitCameraStream], ) -> SessionInit: return SessionInit( session_id=session_id, diff --git a/examples/egocap/capture_server/capture_server/server.py b/examples/egocap/capture_server/capture_server/server.py index a6a4091be5..2c1ede6191 100644 --- a/examples/egocap/capture_server/capture_server/server.py +++ b/examples/egocap/capture_server/capture_server/server.py @@ -40,6 +40,7 @@ import asyncio import collections import contextlib +import dataclasses import json import logging import signal @@ -180,9 +181,12 @@ class ServerConfig: # Manus glove calibration: the worker executable and the managed MCAL directory. None leaves calibration # unavailable (a StartCalibration is rejected) and capture unaffected. calibration: CalibrationConfig | None = None + # LiveKit video offered to clients in SessionInit. Video never passes through this server; the list only says + # where each stream is published. + livekit_streams: tuple[p.LiveKitCameraStream, ...] = () def __post_init__(self) -> None: - """Resolve the MCAL folder once, then reject one physical camera assigned to more than one logical name.""" + """Resolve the MCAL folder once, then reject duplicate physical cameras and unusable LiveKit streams.""" # Resolved here, at construction, so no later request re-resolves it on the event loop. if self.manus_calibration_dir is not None: self.manus_calibration_dir = self.manus_calibration_dir.resolve() @@ -197,6 +201,15 @@ def __post_init__(self) -> None: duplicates = sorted({i for i in ids if ids.count(i) > 1}) if duplicates: raise ValueError(f"physical camera(s) assigned to more than one logical name: {duplicates}") + # Fail at construction, not when a client first parses SessionInit. + try: + p.check_livekit_streams(self.livekit_streams) + except p.ProtocolError as exc: + raise ValueError(f"ServerConfig.livekit_streams: {exc}") from exc + jpeg_names = set(self.cameras or ()) | set(self.known_cameras or ()) + clashes = sorted(jpeg_names & {s.name for s in self.livekit_streams}) + if clashes: + raise ValueError(f"LiveKit stream name(s) equal to a camera name: {clashes}") @dataclass @@ -586,8 +599,10 @@ def __init__( mode: p.CaptureMode, outbox: _Outbox, owner_ws: ServerConnection, + livekit_streams: tuple[p.LiveKitCameraStream, ...] = (), ) -> None: self.session_id = begin.session_id + self._livekit_streams = livekit_streams self.begin = begin # Which owner connection started this session -- a NEW owner # connecting later (because the old one disconnected while this @@ -791,10 +806,12 @@ async def _run_unprotected_body(self) -> None: exception here into a recorded diagnostic.""" tracking_stream = self._tracking_stream() camera_streams = tuple(p.camera_stream_descriptor(c) for c in self._cameras) - streams: list[p.TrackingStream | p.CameraStream] = list(camera_streams) + streams: list[p.TrackingStream | p.CameraStream | p.LiveKitCameraStream] = list(camera_streams) if tracking_stream is not None: streams.insert(0, tracking_stream) + # Offered LiveKit cameras ride in the same registry; this server never produces frames for them. + streams.extend(self._livekit_streams) init = p.session_init_event(session_id=self.session_id, mode=self._mode, begin=self.begin, streams=streams) self.init_event = init # cached for CaptureServer._handle_begin's replay path # The shared connection Outbox/writer is already running (started @@ -1610,6 +1627,7 @@ async def _handle_begin(self, ws: ServerConnection, data: dict[str, Any]) -> Non self._mode, self._outbox, ws, + livekit_streams=self.config.livekit_streams, ) self._active_session = session asyncio.ensure_future(self._run_session(session)) @@ -1993,6 +2011,12 @@ def _build_arg_parser() -> argparse.ArgumentParser: help="live station: read GMSL camera frames from the camera receiver's local frame feed at this Unix " "socket (tracking still comes from IsaacCapture via ISAAC_ROOT); cameras are named with --gmsl-camera", ) + parser.add_argument( + "--livekit-streams", + type=Path, + metavar="FILE", + help="YAML file naming the LiveKit video streams to offer clients in SessionInit (see livekit_streams.py)", + ) parser.add_argument( "--gmsl-camera", action="append", @@ -2223,6 +2247,10 @@ async def _run_cli(args: argparse.Namespace) -> int: source_factory = build_source_factory( config, scenario=args.scenario, seed=args.seed, faults=frozenset(args.fault) ) + if args.livekit_streams is not None: + from .livekit_streams import load_livekit_streams + + config = dataclasses.replace(config, livekit_streams=load_livekit_streams(args.livekit_streams)) except (ValueError, NotImplementedError, OSError, RuntimeError) as exc: # A malformed episode is a ValueError (EpisodeFormatError); a missing directory an OSError; an # unreachable or malformed frame feed a FeedError (RuntimeError). diff --git a/examples/egocap/frontend/package-lock.json b/examples/egocap/frontend/package-lock.json index 9b8a14900f..173182d2c6 100644 --- a/examples/egocap/frontend/package-lock.json +++ b/examples/egocap/frontend/package-lock.json @@ -8,6 +8,7 @@ "name": "egocap-web-frontend", "version": "0.1.0", "dependencies": { + "livekit-client": "2.9.7", "react": "19.3.1", "react-dom": "19.3.1" }, @@ -105,6 +106,12 @@ "specificity": "bin/cli.js" } }, + "node_modules/@bufbuild/protobuf": { + "version": "1.10.1", + "resolved": "https://registry.npmjs.org/@bufbuild/protobuf/-/protobuf-1.10.1.tgz", + "integrity": "sha512-wJ8ReQbHxsAfXhrf9ixl0aYbZorRuOWpBNzm8pL8ftmSxQx/wnJD5Eg861NwJU/czy2VXFIebCeZnZrI9rktIQ==", + "license": "(Apache-2.0 AND BSD-3-Clause)" + }, "node_modules/@csstools/color-helpers": { "version": "6.1.2", "resolved": "https://registry.npmjs.org/@csstools/color-helpers/-/color-helpers-6.1.2.tgz", @@ -291,6 +298,21 @@ "@jridgewell/sourcemap-codec": "^1.4.14" } }, + "node_modules/@livekit/mutex": { + "version": "1.1.1", + "resolved": "https://registry.npmjs.org/@livekit/mutex/-/mutex-1.1.1.tgz", + "integrity": "sha512-EsshAucklmpuUAfkABPxJNhzj9v2sG7JuzFDL4ML1oJQSV14sqrpTYnsaOudMAw9yOaW53NU3QQTlUQoRs4czw==", + "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==", + "license": "Apache-2.0", + "dependencies": { + "@bufbuild/protobuf": "^1.10.0" + } + }, "node_modules/@oxc-project/types": { "version": "0.154.0", "resolved": "https://registry.npmjs.org/@oxc-project/types/-/types-0.154.0.tgz", @@ -1240,6 +1262,15 @@ "@types/estree": "^1.0.0" } }, + "node_modules/events": { + "version": "3.3.0", + "resolved": "https://registry.npmjs.org/events/-/events-3.3.0.tgz", + "integrity": "sha512-mQw+2fkQbALzQ7V0MY0IqdnXNOeTtP4r0lN9z7AAawCXgqea7bDii20AYrIBrFd/Hx0M2Ocz6S111CaFkUcb0Q==", + "license": "MIT", + "engines": { + "node": ">=0.8.x" + } + }, "node_modules/expect-type": { "version": "1.4.0", "resolved": "https://registry.npmjs.org/expect-type/-/expect-type-1.4.0.tgz", @@ -1612,6 +1643,36 @@ "url": "https://opencollective.com/parcel" } }, + "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==", + "license": "Apache-2.0", + "dependencies": { + "@livekit/mutex": "1.1.1", + "@livekit/protocol": "1.34.0", + "events": "^3.3.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" + } + }, + "node_modules/loglevel": { + "version": "1.9.2", + "resolved": "https://registry.npmjs.org/loglevel/-/loglevel-1.9.2.tgz", + "integrity": "sha512-HgMmCqIJSAKqo68l0rS2AanEWfkxaZ5wNiEFb5ggm08lDs9Xl2KxBlX3PTcaD2chBM1gXAYf491/M2Rv8Jwayg==", + "license": "MIT", + "engines": { + "node": ">= 0.6.0" + }, + "funding": { + "type": "tidelift", + "url": "https://tidelift.com/funding/github/npm/loglevel" + } + }, "node_modules/lru-cache": { "version": "11.5.3", "resolved": "https://registry.npmjs.org/lru-cache/-/lru-cache-11.5.3.tgz", @@ -1844,6 +1905,16 @@ "@rolldown/binding-win32-x64-msvc": "1.2.14" } }, + "node_modules/rxjs": { + "version": "7.8.2", + "resolved": "https://registry.npmjs.org/rxjs/-/rxjs-7.8.2.tgz", + "integrity": "sha512-dhKf903U/PQZY6boNNtAGdWbG85WAbjT/1xYoZIC7FAY0yWapOBQVsVrDl58W86//e1VpMNBtRV4MaXfdMySFA==", + "license": "Apache-2.0", + "optional": true, + "dependencies": { + "tslib": "^2.1.0" + } + }, "node_modules/saxes": { "version": "6.0.0", "resolved": "https://registry.npmjs.org/saxes/-/saxes-6.0.0.tgz", @@ -1863,6 +1934,21 @@ "integrity": "sha512-juorfCmIkIw8tT+p5BXSm6PJjQF/ycEYmKyzURCIt/RaZIhL+PulbQ9Yu2z1HdOJDdqDTlxA1+xKBmHXJsczAw==", "license": "MIT" }, + "node_modules/sdp": { + "version": "3.2.2", + "resolved": "https://registry.npmjs.org/sdp/-/sdp-3.2.2.tgz", + "integrity": "sha512-xZocWwfyp4hkbN4hLWxMjmv2Q8aNa9MhmOZ7L9aCZPT+dZsgRr6wZRrSYE3HTdyk/2pZKPSgqI7ns7Een1xMSA==", + "license": "MIT" + }, + "node_modules/sdp-transform": { + "version": "2.15.0", + "resolved": "https://registry.npmjs.org/sdp-transform/-/sdp-transform-2.15.0.tgz", + "integrity": "sha512-KrOH82c/W+GYQ0LHqtr3caRpM3ITglq3ljGUIb8LTki7ByacJZ9z+piSGiwZDsRyhQbYBOBJgr2k6X4BZXi3Kw==", + "license": "MIT", + "bin": { + "sdp-verify": "checker.js" + } + }, "node_modules/source-map-js": { "version": "1.2.2", "resolved": "https://registry.npmjs.org/source-map-js/-/source-map-js-1.2.2.tgz", @@ -1963,6 +2049,27 @@ "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", + "integrity": "sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w==", + "license": "0BSD" + }, + "node_modules/typed-emitter": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/typed-emitter/-/typed-emitter-2.1.0.tgz", + "integrity": "sha512-g/KzbYKbH5C2vPkaXGu8DJlHrGKHLsM25Zg9WuC9pMGfuvT+X25tZQWo5fK1BjBm8+UrVE9LDCvaY0CQk+fXDA==", + "license": "MIT", + "optionalDependencies": { + "rxjs": "*" + } + }, "node_modules/typescript": { "version": "7.0.2", "resolved": "https://registry.npmjs.org/typescript/-/typescript-7.0.2.tgz", @@ -2192,6 +2299,19 @@ "node": ">=20" } }, + "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==", + "license": "BSD-3-Clause", + "dependencies": { + "sdp": "^3.2.0" + }, + "engines": { + "node": ">=6.0.0", + "npm": ">=3.10.0" + } + }, "node_modules/whatwg-mimetype": { "version": "5.0.0", "resolved": "https://registry.npmjs.org/whatwg-mimetype/-/whatwg-mimetype-5.0.0.tgz", diff --git a/examples/egocap/frontend/package.json b/examples/egocap/frontend/package.json index 967ecf6c65..cd5ce66436 100644 --- a/examples/egocap/frontend/package.json +++ b/examples/egocap/frontend/package.json @@ -11,6 +11,7 @@ "test": "vitest run" }, "dependencies": { + "livekit-client": "2.9.7", "react": "19.3.1", "react-dom": "19.3.1" }, diff --git a/examples/egocap/frontend/src/App.operatorLayout.test.tsx b/examples/egocap/frontend/src/App.operatorLayout.test.tsx index 0fa5e8347f..37fe307ff7 100644 --- a/examples/egocap/frontend/src/App.operatorLayout.test.tsx +++ b/examples/egocap/frontend/src/App.operatorLayout.test.tsx @@ -21,6 +21,7 @@ vi.mock("./useCaptureSession", async (importOriginal) => { afterEach(() => { cleanup(); window.localStorage.clear(); + vi.unstubAllGlobals(); }); function snapshot(overrides: Partial = {}): PreviewSnapshot { diff --git a/examples/egocap/frontend/src/App.test.tsx b/examples/egocap/frontend/src/App.test.tsx index 8ae3e89c73..8999d9fcf4 100644 --- a/examples/egocap/frontend/src/App.test.tsx +++ b/examples/egocap/frontend/src/App.test.tsx @@ -27,6 +27,7 @@ vi.mock("./useCaptureSession", async (importOriginal) => { afterEach(() => { cleanup(); vi.useRealTimers(); + vi.unstubAllGlobals(); }); function makeSnapshot(overrides: Partial = {}): PreviewSnapshot { diff --git a/examples/egocap/frontend/src/CameraGrid.livekit.test.tsx b/examples/egocap/frontend/src/CameraGrid.livekit.test.tsx new file mode 100644 index 0000000000..244aaf774e --- /dev/null +++ b/examples/egocap/frontend/src/CameraGrid.livekit.test.tsx @@ -0,0 +1,170 @@ +// SPDX-FileCopyrightText: 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// +// SPDX-License-Identifier: Apache-2.0 + +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { cleanup, render, screen } from "@testing-library/react"; +import { CameraGrid } from "./CameraGrid"; +import type { LiveKitStreamDescriptor, PreviewSnapshot } from "./captureProtocol"; +import { STREAMS, FakeRoom } from "./livekitTestDoubles"; + +vi.mock("livekit-client", async () => { + const doubles = await import("./livekitTestDoubles"); + return { + Room: doubles.FakeRoom, + RoomEvent: { + TrackSubscribed: "TrackSubscribed", + TrackUnsubscribed: "TrackUnsubscribed", + ParticipantDisconnected: "ParticipantDisconnected", + Reconnecting: "Reconnecting", + Reconnected: "Reconnected", + Disconnected: "Disconnected", + }, + Track: { Kind: { Video: "video", Audio: "audio" } }, + }; +}); + +function snapshot(names: string[], offered: LiveKitStreamDescriptor[] = []): PreviewSnapshot { + return { + version: 1, + type: 3, + backend_instance_id: "b1", + revision: 1, + station_label: "Station", + profile_label: "Profile", + connecting: false, + started: { session_id: "s1", running: true, mode: 0, streams: offered }, + status: null, + stopped: null, + error_message: null, + binary_generation: 1, + tracking: null, + cameras: names.map((name, i) => ({ stream_id: i + 1, name, url: `/stream${i + 1}?a=1`, sequence: null, age_ms: null, diagnostic: null })), + visualization: null, + } as PreviewSnapshot; +} + +function stubFetch(status: number, body: unknown): ReturnType { + const fn = vi.fn(async () => ({ status, ok: status === 200, json: async () => body })); + vi.stubGlobal("fetch", fn); + return fn; +} + +async function settled(): Promise { + // Let the hook's fetch and connect promises resolve. + await new Promise((r) => setTimeout(r, 0)); + await new Promise((r) => setTimeout(r, 0)); +} + +beforeEach(() => FakeRoom.reset()); +afterEach(() => { + cleanup(); + vi.unstubAllGlobals(); +}); + +describe("CameraGrid with offered LiveKit streams", () => { + it("renders LiveKit tiles beside MJPEG tiles, with the MJPEG camera keeping the primary slot", async () => { + stubFetch(200, { token: "tok" }); + const { container } = render( + , + ); + await settled(); + expect(screen.getByText("head_camera")).not.toBeNull(); + expect(screen.getByText("Head left", { exact: false })).not.toBeNull(); + expect(screen.getByText("Head right", { exact: false })).not.toBeNull(); + expect(screen.queryByText(/Preview only/)).toBeNull(); + expect(container.querySelector(".camera-preview-badge")).toBeNull(); + expect(container.querySelector(".camera-grid-items")?.className).toContain("camera-grid-items--n3"); + const primary = container.querySelectorAll(".camera-tile--primary"); + expect(primary).toHaveLength(1); + expect(primary[0].textContent).toContain("head_camera"); + }); + + it("shows only LiveKit tiles, the first as primary, when CaptureServer has no cameras", async () => { + stubFetch(200, { token: "tok" }); + const { container } = render( + , + ); + await settled(); + expect(screen.queryByText("No cameras selected.")).toBeNull(); + expect(container.querySelector(".camera-tile--primary")?.textContent).toContain("Head left"); + }); + + it("gives two head tiles equal primary weight with the heads-only layout", async () => { + stubFetch(200, { token: "tok" }); + const { container } = render( + , + ); + await settled(); + const grid = container.querySelector(".camera-grid-items"); + expect(grid?.className).toContain("camera-grid-items--heads"); + expect(grid?.className).toContain("camera-grid-items--heads-only"); + expect(grid?.style.getPropertyValue("--head-count")).toBe("2"); + expect(container.querySelectorAll(".camera-tile--primary")).toHaveLength(2); + }); + + it("keeps a wrist tile secondary beside two primary head tiles", async () => { + stubFetch(200, { token: "tok" }); + const offered = [ + ...STREAMS, + { ...STREAMS[0], stream_id: 18, name: "Wrist left", track: "wrist" }, + ]; + const { container } = render( + , + ); + await settled(); + const grid = container.querySelector(".camera-grid-items"); + expect(grid?.className).toContain("camera-grid-items--heads"); + expect(grid?.className).not.toContain("camera-grid-items--heads-only"); + expect(grid?.style.getPropertyValue("--head-count")).toBe("2"); + expect(grid?.style.getPropertyValue("--side-count")).toBe("1"); + const primary = container.querySelectorAll(".camera-tile--primary"); + expect(primary).toHaveLength(2); + primary.forEach((tile) => expect(tile.textContent).toContain("Head")); + expect(screen.getByText("Wrist left", { exact: false }).closest(".camera-tile--primary")).toBeNull(); + }); + + it("keeps the single-tile layout when only one LiveKit tile is offered", async () => { + stubFetch(200, { token: "tok" }); + const { container } = render( + , + ); + await settled(); + expect(container.querySelector(".camera-grid-items")?.className).toContain("camera-grid-items--n1"); + const primary = container.querySelectorAll(".camera-tile--primary"); + expect(primary).toHaveLength(1); + expect(primary[0].textContent).toContain("Head left"); + }); + + it("does not contact the backend and keeps MJPEG behavior when no streams are offered", async () => { + const fetchFn = stubFetch(200, { token: "tok" }); + const { container, rerender } = render(); + await settled(); + expect(screen.getByText("No cameras selected.")).not.toBeNull(); + expect(screen.queryByText(/LiveKit preview unavailable/)).toBeNull(); + rerender(); + expect(container.querySelector(".camera-grid-items")?.className).toContain("camera-grid-items--n2"); + expect(screen.queryByText(/Preview only/)).toBeNull(); + expect(FakeRoom.instances).toHaveLength(0); + expect(fetchFn).not.toHaveBeenCalled(); + }); + + it("shows the unavailable note with the backend detail when the token cannot be issued", async () => { + stubFetch(503, { detail: "LIVEKIT_API_KEY and LIVEKIT_API_SECRET must be set" }); + render(); + await settled(); + expect(screen.getByText(/LiveKit preview unavailable: LIVEKIT_API_KEY/)).not.toBeNull(); + }); + + it("shows the unavailable note when the backend cannot be reached", async () => { + vi.stubGlobal( + "fetch", + vi.fn(async () => { + throw new TypeError("fetch failed"); + }), + ); + render(); + await settled(); + expect(screen.getByText("LiveKit preview unavailable: Cannot reach the backend")).not.toBeNull(); + }); +}); diff --git a/examples/egocap/frontend/src/CameraGrid.test.tsx b/examples/egocap/frontend/src/CameraGrid.test.tsx index 7f7d4617c0..5a4f8feebd 100644 --- a/examples/egocap/frontend/src/CameraGrid.test.tsx +++ b/examples/egocap/frontend/src/CameraGrid.test.tsx @@ -14,6 +14,7 @@ import type { PreviewSnapshot } from "./captureProtocol"; afterEach(() => { cleanup(); vi.useRealTimers(); + vi.unstubAllGlobals(); }); function makeSnapshot(sequence: string | null, ageMs: number | null, fps = 60): PreviewSnapshot { diff --git a/examples/egocap/frontend/src/CameraGrid.tsx b/examples/egocap/frontend/src/CameraGrid.tsx index 416a453fd3..e79a257e5e 100644 --- a/examples/egocap/frontend/src/CameraGrid.tsx +++ b/examples/egocap/frontend/src/CameraGrid.tsx @@ -9,11 +9,15 @@ */ import { useEffect, useRef, useState } from "react"; +import type { CSSProperties } from "react"; +import { liveKitStreams } from "./captureProtocol"; import type { PreviewSnapshot } from "./captureProtocol"; import { TRACKING_STALE_AFTER_MS } from "./CaptureTree"; import { CameraRateTracker, formatCameraStats } from "./cameraRates"; import { ROLE_LABELS } from "./settingsApi"; import { roleOfCamera } from "./useOperatorSettings"; +import { LiveKitCameraTile } from "./LiveKitCameraTile"; +import { useLiveKitPreview } from "./useLiveKitPreview"; const CAMERA_ERROR_RETRY_LIMIT = 3; @@ -184,8 +188,11 @@ export function CameraGrid({ rateTracker.current.retainOnly(keys); } + // Direct LiveKit tiles sit beside the CaptureServer ones; they have their own connection and readiness. + const livekit = useLiveKitPreview(liveKitStreams(snapshot?.started)); const cameras = snapshot?.cameras ?? []; - if (cameras.length === 0) { + const livekitShown = livekit.tiles.length > 0 || livekit.phase === "error"; + if (cameras.length === 0 && !livekitShown) { return (

Cameras

@@ -198,10 +205,26 @@ export function CameraGrid({ const egoIndex = cameras.findIndex((cam) => cam.name === cameraRoles?.ego); const headIndex = cameras.findIndex((cam) => /head/i.test(cam.name)); const primaryIndex = egoIndex !== -1 ? egoIndex : headIndex === -1 ? 0 : headIndex; + // With only LiveKit video, every head camera is a main view of equal size; any other camera (a wrist camera) + // stacks smaller beside them. With no more than one named head, the first tile is the main view. + const liveKitOnly = cameras.length === 0; + const isHeadTile = (name: string): boolean => /head/i.test(name); + const headTiles = liveKitOnly ? livekit.tiles.filter((tile) => isHeadTile(tile.name)).length : 0; + const equalHeads = headTiles >= 2; + const sideTiles = livekit.tiles.length - headTiles; + const gridClass = equalHeads + ? `camera-grid-items--heads${sideTiles === 0 ? " camera-grid-items--heads-only" : ""}` + : `camera-grid-items--n${Math.min(cameras.length + livekit.tiles.length, 4)}`; + const gridStyle = equalHeads + ? ({ "--head-count": headTiles, "--side-count": Math.max(sideTiles, 1) } as CSSProperties) + : undefined; return (

Cameras

-
+ {livekit.phase === "error" && livekit.message !== null && ( +

LiveKit preview unavailable: {livekit.message}

+ )} +
{cameras.map((cam, index) => { const effectiveAge = cam.age_ms === null ? null : cam.age_ms + elapsedSinceReceipt; // "Never had a sample" (sequence/age_ms both null) is a distinct, @@ -236,6 +259,17 @@ export function CameraGrid({ /> ); })} + {livekit.tiles.map((tile, index) => ( + + ))}
); diff --git a/examples/egocap/frontend/src/LiveKitCameraTile.test.tsx b/examples/egocap/frontend/src/LiveKitCameraTile.test.tsx new file mode 100644 index 0000000000..7eb88dcd62 --- /dev/null +++ b/examples/egocap/frontend/src/LiveKitCameraTile.test.tsx @@ -0,0 +1,134 @@ +// SPDX-FileCopyrightText: 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// +// SPDX-License-Identifier: Apache-2.0 + +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { act, cleanup, render, screen } from "@testing-library/react"; +import { LiveKitCameraTile } from "./LiveKitCameraTile"; +import { fakeTrack } from "./livekitTestDoubles"; + +let frameCallbacks: Array<() => void> = []; + +beforeEach(() => { + vi.useFakeTimers(); + frameCallbacks = []; + // jsdom has no requestVideoFrameCallback; this one lets a test present frames by hand. + (HTMLVideoElement.prototype as any).requestVideoFrameCallback = function (cb: () => void) { + frameCallbacks.push(cb); + return frameCallbacks.length; + }; + (HTMLVideoElement.prototype as any).cancelVideoFrameCallback = vi.fn(); +}); + +afterEach(() => { + cleanup(); + vi.useRealTimers(); + delete (HTMLVideoElement.prototype as any).requestVideoFrameCallback; + delete (HTMLVideoElement.prototype as any).cancelVideoFrameCallback; +}); + +function presentFrame(): void { + const cbs = frameCallbacks; + frameCallbacks = []; + act(() => cbs.forEach((cb) => cb())); +} + +describe("LiveKitCameraTile placeholders", () => { + it("shows Connecting while the room connects", () => { + render(); + expect(screen.getByText("Connecting")).not.toBeNull(); + }); + + it("shows Reconnecting while the SDK reconnects", () => { + render(); + expect(screen.getByText("Reconnecting")).not.toBeNull(); + }); + + it("shows Disconnected in the error phase", () => { + render(); + expect(screen.getByText("Disconnected")).not.toBeNull(); + }); + + it("shows Waiting for camera when connected but the track is not published", () => { + const { container } = render(); + expect(screen.getByText("Waiting for camera")).not.toBeNull(); + expect(container.querySelector("video")).toBeNull(); + }); + + it("shows Waiting for video until the first frame is presented, then no placeholder", () => { + const track = fakeTrack(); + render(); + expect(screen.getByText("Waiting for video")).not.toBeNull(); + presentFrame(); + expect(screen.queryByText("Waiting for video")).toBeNull(); + expect(document.querySelector(".camera-placeholder")).toBeNull(); + }); + + it("shows No video frames and a warning tone when frames stop for more than three seconds", () => { + const track = fakeTrack(); + const { container } = render(); + presentFrame(); + act(() => { + vi.advanceTimersByTime(3000); + }); + expect(screen.queryByText("No video frames")).toBeNull(); + act(() => { + vi.advanceTimersByTime(2000); + }); + expect(screen.getByText("No video frames")).not.toBeNull(); + expect(container.querySelector("figure")?.className).toContain("tone-warning"); + // Frames resuming clears the warning. + presentFrame(); + expect(screen.queryByText("No video frames")).toBeNull(); + }); + + it("falls back to timeupdate when requestVideoFrameCallback is unavailable", () => { + delete (HTMLVideoElement.prototype as any).requestVideoFrameCallback; + const track = fakeTrack(); + const { container } = render(); + act(() => { + container.querySelector("video")!.dispatchEvent(new Event("timeupdate")); + }); + expect(screen.queryByText("Waiting for video")).toBeNull(); + }); +}); + +describe("LiveKitCameraTile track wiring and labels", () => { + it("attaches the track to a muted inline video and detaches it on unmount", () => { + const track = fakeTrack(); + const { container, unmount } = render(); + const video = container.querySelector("video")!; + expect(track.attach).toHaveBeenCalledWith(video); + expect(video.muted).toBe(true); + expect(video.autoplay).toBe(true); + expect(video.hasAttribute("playsinline")).toBe(true); + unmount(); + expect(track.detach).toHaveBeenCalledWith(video); + }); + + it("detaches the old track and attaches the new one when the track changes", () => { + const a = fakeTrack(); + const b = fakeTrack(); + const { container, rerender } = render(); + const video = container.querySelector("video")!; + rerender(); + expect(a.detach).toHaveBeenCalledWith(video); + expect(b.attach).toHaveBeenCalledWith(video); + }); + + it("goes back to Waiting for video when a replacement track has not produced a frame", () => { + const a = fakeTrack(); + const { rerender } = render(); + presentFrame(); + rerender(); + expect(screen.getByText("Waiting for video")).not.toBeNull(); + }); + + it("shows its name and makes no claim about recording", () => { + render(); + expect(screen.getByText("Head left", { exact: false })).not.toBeNull(); + expect(screen.queryByText(/Preview only/)).toBeNull(); + expect(document.querySelector(".camera-preview-badge")).toBeNull(); + expect(document.querySelector(".camera-tile--primary")).not.toBeNull(); + }); +}); diff --git a/examples/egocap/frontend/src/LiveKitCameraTile.tsx b/examples/egocap/frontend/src/LiveKitCameraTile.tsx new file mode 100644 index 0000000000..e6acb096a3 --- /dev/null +++ b/examples/egocap/frontend/src/LiveKitCameraTile.tsx @@ -0,0 +1,101 @@ +// SPDX-FileCopyrightText: 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// +// SPDX-License-Identifier: Apache-2.0 + +/** + * One live camera tile fed directly by a LiveKit video track. + * + * Readiness comes from the track and from decoded frames actually reaching the