egocap: show and record Orin cameras streamed over LiveKit

examples/egocap/LIVEKIT_PREVIEW.md

@@ -0,0 +1,88 @@
+<!--
+SPDX-FileCopyrightText: 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+
+SPDX-License-Identifier: Apache-2.0
+-->
+
+# 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.

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

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)