egocap: serve SENSING plugin cameras from CaptureServer

examples/egocap/backend/egocap_web/capture_preview.py

@@ -2,7 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-"""Latest-sample preview cache: one tracking slot and at most three camera
+"""Latest-sample preview cache: one tracking slot and at most four camera
 slots, each holding its immutable callback value and backend-local receipt
 time. Replacing a slot drops the superseded value -- there is no history,
 replay queue or recording buffer here.
@@ -18,7 +18,7 @@
 # capture.
 MAX_TRACKING_PAYLOAD_BYTES = 64 * 1024
 MAX_CAMERA_JPEG_BYTES = 8 * 1024 * 1024
-MAX_CAMERAS = 3
+MAX_CAMERAS = 4
 
 
 @dataclass(frozen=True)

examples/egocap/backend/egocap_web/station_config.py

@@ -34,9 +34,16 @@
 
 STATION_CONFIG_ENV = "EGOCAP_STATION_CONFIG"
 _DEFAULT_FLAT_STATION_ID = "station_01"
-#: Matches capture_server.protocol's own BeginRequest.cameras limit -- checked here too so an
-#: over-limit YAML is a configuration error, not a later capture-startup failure.
-_MAX_CAMERAS = 3
+#: capture_server.protocol's own BeginRequest.cameras limit, reused rather than duplicated so this
+#: loader and the server can never drift apart -- checked here too so an over-limit YAML is a
+#: configuration error, not a later capture-startup failure.
+_MAX_CAMERAS = p._MAX_CAMERAS
+
+#: Maps each supported YAML ``video_sources[].type`` spelling to its wire CameraType.
+_SOURCE_TYPES: dict[str, p.CameraType] = {
+    "oak_camera": p.CameraType.OAK_D_COLOR,
+    "sensing_camera": p.CameraType.SENSING_COLOR,
+}
 
 
 class StationConfigError(ValueError):
@@ -122,11 +129,13 @@ def _resolve_cameras(station_id: str, entry: dict) -> tuple[p.CameraSelection, .
             raise StationConfigError(f"station {station_id!r}: duplicate video source name {name!r}")
         seen_names.add(name)
         source_type = source.get("type")
-        if source_type != "oak_camera":
+        camera_type = _SOURCE_TYPES.get(source_type) if isinstance(source_type, str) else None
+        if camera_type is None:
+            supported = ", ".join(sorted(repr(t) for t in _SOURCE_TYPES))
             raise StationConfigError(
-                f"station {station_id!r}: video_sources[{i}].type {source_type!r} is not 'oak_camera'"
+                f"station {station_id!r}: video_sources[{i}].type {source_type!r} is not one of {supported}"
             )
-        cameras.append(p.CameraSelection(name=name, type=p.CameraType.OAK_D_COLOR))
+        cameras.append(p.CameraSelection(name=name, type=camera_type))
     return tuple(cameras)
 
 

examples/egocap/capture_server/README.md

@@ -82,6 +82,24 @@ the link at USB 2; some host ports or cables cannot boot an OAK-D at USB 3. The
 runs on-device NV12 to MJPEG; each frame's own intrinsics are reported in the camera
 status when the device provides them.
 
+**SENSING cameras.** Cameras served by IsaacCapture's SENSING plugin are configured with one
+YAML file (`sensing_rig:` plus `cameras:` entries of `type: SENSING_COLOR`) and
+`--sensing-config FILE`; the file format is documented on `load_sensing_station` in
+`capture_server/acquisition/sensing.py`. The server launches the plugin's executable itself, reads each
+sensor's H.264 output file, decodes it (every unit in order) and sends the newest decoded frame as a JPEG
+(`av`, i.e. PyAV, is a declared dependency; nothing is loaded unless a SENSING stream is used).
+The decoder is `sensing_rig.decoder`: `auto` (default) uses the Jetson NVDEC hardware decoder through GStreamer's
+`nvv4l2decoder` when `/dev/v4l2-nvdec`, `gst-launch-1.0` and that element exist, and otherwise PyAV on the CPU;
+`hardware` fails the stream if NVDEC is unavailable; `software` forces the CPU. Each stream logs which decoder it
+uses (`isaaccapture.capture_server.sensing_h264`). A 2560x1984 stream at 60 fps is beyond one CPU core of software
+decoding (about 30 fps on an Orin core), which shows as a picture that falls further behind; use hardware decode there.
+`--mock` with a SENSING station requires the plugin's MOCK build, whose encoder must write real H.264;
+an older mock encoder wrote placeholder bytes that are not video, which the decoder rejects, so such a
+build shows a degraded camera instead of pictures. With `--sensing-config`, `cameras` accepts only the
+SENSING names it lists (`--camera` is real-acquisition only), so the OAK mock cameras cannot be mixed in.
+Measure per-frame cost on the target with `tests/capture_server/bench_sensing_decode.py`
+before choosing `delivery_fps`.
+
 **Failure policy.** A camera that produces no frame for max(5 s, five frame periods)
 counts as failed. A required camera failure ends the session; an optional one reports
 a degraded camera. There is no automatic reconnection: retry by starting a new session.

examples/egocap/capture_server/capture_server/acquisition/__init__.py

@@ -12,12 +12,15 @@
 from .depthai import DepthAIAdapter
 from .errors import CameraSourceError, SourceError, SourceExited
 from .isaac_capture import IsaacCaptureAdapter, TrackingSelection
+from .sensing import SensingAdapter, SensingRig
 
 __all__ = [
     "Acquisition",
     "CameraSourceError",
     "DepthAIAdapter",
     "IsaacCaptureAdapter",
+    "SensingAdapter",
+    "SensingRig",
     "SourceError",
     "SourceExited",
     "TrackingSelection",

examples/egocap/capture_server/capture_server/acquisition/acquisition.py

@@ -2,20 +2,17 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-"""``Acquisition``: the one concrete object ``_Session`` holds for its
-acquisition source -- composing one optional :class:`IsaacCaptureAdapter`
-(``None`` for a camera-only request, which must never construct an IC
-session) with zero or more :class:`DepthAIAdapter`, one per configured
-camera stream.
-
-This is not a generic protocol/interface: it is one
-fixed, concretely named composition that always holds exactly these two
-concrete adapter types, mode-agnostic by construction since both adapters
-already are. ``_Session`` calls it directly --
-``enter``/``next_tracking_sample``/``next_camera_jpeg``/``status``/``close``.
-``next_tracking_sample``/``next_camera_jpeg`` take no
-``t``/``tick`` argument: each adapter paces and times itself -- see their
-own docstrings.
+"""``Acquisition``: the one concrete object ``_Session`` holds for its acquisition source.
+
+It composes one optional :class:`IsaacCaptureAdapter` (``None`` for a camera-only request, which must never
+construct an IC session), zero or more camera adapters (:class:`DepthAIAdapter` for OAK-D,
+:class:`SensingAdapter` for SENSING), and the optional :class:`SensingRig` -- the one native producer process
+the SENSING adapters read from. The rig is owned here and only borrowed by its adapters, so it is started
+before them and reaped after they have settled.
+
+This is NOT a generic source interface: it is one fixed, concretely named composition. ``_Session`` calls it
+directly -- ``enter``/``next_tracking_sample``/``next_camera_jpeg``/``status``/``close`` -- and each adapter
+paces and times itself (see their own docstrings).
 """
 
 from __future__ import annotations
@@ -28,14 +25,30 @@
 from .depthai import DepthAIAdapter
 from .errors import CameraSourceError
 from .isaac_capture import IsaacCaptureAdapter
+from .sensing import SensingAdapter, SensingRig
 
 __all__ = ["Acquisition"]
 
 
+def _observe_close(task: "asyncio.Future") -> "asyncio.Future":
+    """Register an observer on a close task as soon as it exists, so a late failure is never left unretrieved.
+
+    Observation only: it does not own, cancel or await the task.
+    """
+    task.add_done_callback(lambda f: f.cancelled() or f.exception())
+    return task
+
+
+# The server's whole source-release budget for one session; adapters and the SENSING producer share it.
+_CLOSE_BUDGET_S = 5.0
+
+
 @dataclass
 class Acquisition:
     tracking: IsaacCaptureAdapter | None
-    cameras: tuple[DepthAIAdapter, ...]
+    cameras: tuple[DepthAIAdapter | SensingAdapter, ...]
+    # The one native SENSING producer shared by the SENSING adapters in ``cameras``; owned here, borrowed by them.
+    sensing_rig: SensingRig | None = None
 
     async def enter(self) -> None:
         """A REQUIRED source's startup failure is fatal and propagates
@@ -51,6 +64,10 @@ async def enter(self) -> None:
         for that camera."""
         if self.tracking is not None:
             await self.tracking.enter()
+        if self.sensing_rig is not None:
+            # Never raises: a start failure is recorded on the rig and each SENSING adapter reports it under
+            # its own required/optional policy below, so optional streams degrade while OAK/tracking go on.
+            await self.sensing_rig.start()
         for camera in self.cameras:
             try:
                 await camera.enter()
@@ -84,28 +101,59 @@ async def status(self, updated_at_ns: int) -> p.StatusSnapshot:
         )
 
     async def close(self) -> bool:
-        """Closes every adapter CONCURRENTLY, each on its own dedicated
-        worker thread (native thread affinity is per-adapter already --
-        see ThreadRunner -- so running them concurrently never puts two
-        threads on the same native resource). Up to four independent
-        adapters (one IC + up to three cameras) each bound to their own
-        release timeout would, awaited sequentially, sum to up to four
-        times that bound -- comfortably exceeding the outer release-wait
-        budget server.py's own finalization assumes for ONE source
-        release. Running them concurrently keeps the total wall-clock
-        cost to whichever single adapter is slowest, not their sum.
-        Every adapter's close is still attempted regardless of whether
-        another raised or timed out -- one release failure must never
-        skip releasing the others, and a failure to confirm release
-        within bound is reported (``ok=False``), never silently assumed
-        successful."""
-        closers = ([self.tracking.close()] if self.tracking is not None else []) + [
-            camera.close() for camera in self.cameras
+        """Releases every adapter and the SENSING producer, concurrently, each on its own bound.
+
+        Every adapter is closed on its own dedicated worker thread (native thread affinity is per-adapter -- see
+        ThreadRunner). The tracking and OAK adapters each keep their own bound. The SENSING stream adapters and
+        the producer they borrow form ONE chain (:meth:`_close_rig_chain`): the adapters settle first, then the
+        producer is terminated and reaped, so a worker still reading never races its teardown. The chain runs
+        concurrently with the other closes and is sized to fit the same ``_CLOSE_BUDGET_S``, so a slow OAK or
+        tracking release is not cut short by the producer's shutdown, nor the reverse.
+        Waiting is bounded by ``_CLOSE_BUDGET_S``; a timeout does not cancel any close job -- it keeps running and
+        is only reported as unconfirmed (False). Every close is attempted; one failure never skips another. A
+        False means a worker MAY still be running: the wait bounds how long we waited, it does not prove native
+        settlement. A cancelled caller does not prove release either: every job keeps running, observed, so the
+        producer is still reaped.
+        """
+        borrowing = [c for c in self.cameras if getattr(c, "borrows_rig", False)]
+        independent = ([self.tracking] if self.tracking is not None else []) + [
+            c for c in self.cameras if not getattr(c, "borrows_rig", False)
         ]
-        if not closers:
-            return True
-        results = await asyncio.gather(*closers, return_exceptions=True)
-        first_exc = next((r for r in results if isinstance(r, BaseException)), None)
-        if first_exc is not None:
+        tasks = [_observe_close(asyncio.ensure_future(a.close())) for a in independent]
+        if borrowing or self.sensing_rig is not None:
+            tasks.append(_observe_close(asyncio.ensure_future(self._close_rig_chain(borrowing))))
+        pending: set[asyncio.Future] = set()
+        if tasks:
+            _done, pending = await asyncio.wait(tasks, timeout=_CLOSE_BUDGET_S)
+        results: list[bool | BaseException] = []
+        for task in tasks:
+            if task in pending or task.cancelled():
+                results.append(False)  # unconfirmed: may still be running
+            elif (exc := task.exception()) is not None:
+                results.append(exc)
+            else:
+                results.append(task.result())
+        if (first_exc := next((r for r in results if isinstance(r, BaseException)), None)) is not None:
             raise first_exc
         return all(results)
+
+    async def _close_rig_chain(self, adapters: list) -> bool:
+        """Close the adapters that borrow the producer, then terminate and reap it.
+
+        The producer is reaped whether or not an adapter failed to close. True only if every adapter and the
+        producer confirmed release; the first failure is raised after both have been attempted.
+        """
+        results = await asyncio.gather(*(a.close() for a in adapters), return_exceptions=True)
+        rig_result: bool | BaseException = True
+        if self.sensing_rig is not None:
+            try:
+                rig_result = await self.sensing_rig.close()
+            except Exception as exc:  # noqa: BLE001 -- reported below, after the adapters' own results
+                rig_result = exc
+        exceptions = [r for r in (*results, rig_result) if isinstance(r, BaseException)]
+        if exceptions:
+            # A cancellation takes priority over any other exception collected here: dropping it in favor of
+            # an unrelated close failure would silently defeat the caller's own cancellation.
+            cancelled = next((e for e in exceptions if isinstance(e, asyncio.CancelledError)), None)
+            raise cancelled if cancelled is not None else exceptions[0]
+        return all(results) and rig_result is True