feat: one acquisition thread owns the pipeline, requests wait for its frames

FrameAcquisition: a single thread calls wait_for_frames + align (the old
capture() body, now _grab_once) and keeps the latest frame, numbered.
capture() no longer touches the pipeline: it waits for the next frame of
that thread; requests waiting at the same moment get the same frame, so N
callers cost one grab. get_rgb()/get_depth() keep their 100 ms cache,
now on the thread's latest frame (_last_frame is gone).

The thread grabs on demand, only while someone waits for a newer frame:
grabbing continuously at 30 fps would align frames nobody reads, and align
on the Jetson CPU is the main suspect of the capture cost. A grab error is
logged and retried, the thread never dies. close() stops it before the
pipeline and joins it after (pipeline.stop() ends a wait in progress).
open/close and camera_open are unchanged, so the healthcheck + autoheal
recovery of ede70a2 sees the same state as before.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/backend/services/camera.py

@@ -30,10 +30,14 @@ import cv2
 import numpy as np
 import pyrealsense2 as rs
 
+from src.backend.services.frame_acquisition import FrameAcquisition
 from src.backend.services.phase_timing import timer
 
 logger = logging.getLogger(__name__)
 
+GRAB_TIMEOUT_MS = 5000   # wait_for_frames timeout of the acquisition thread (the old capture() default)
+STOP_JOIN_S = GRAB_TIMEOUT_MS / 1000.0 + 1.0   # close() waits this long for the thread to leave a grab
+
 
 @dataclass
 class CameraFrame:
@@ -110,7 +114,8 @@ class RealSenseD435:
         self._initialized = False
         self._open = False
         self._discovered: list[DiscoveredDevice] = []
-        self._last_frame: Optional[CameraFrame] = None
+        # The only caller of wait_for_frames once the camera is open (see capture()).
+        self._acquisition: Optional[FrameAcquisition[CameraFrame]] = None
         self._color_available: bool = True
         self._color_format = rs.format.bgr8
         self._usb_type: str = "unknown"
@@ -301,6 +306,7 @@ class RealSenseD435:
 
             self.device_index = device_index
             self._open = True
+            self._start_acquisition()
             logger.info(
                 f"RealSense D4xx opened: serial={serial} usb={usb_type} "
                 f"color={color_w}x{color_h}@{color_fps}({color_fmt}) "
@@ -315,11 +321,19 @@ class RealSenseD435:
     def close(self) -> None:
         if not self._open:
             return
-        self._teardown_pipeline()
+        acquisition, self._acquisition = self._acquisition, None
+        if acquisition is not None:
+            acquisition.request_stop()
+        self._teardown_pipeline()   # pipeline.stop() also ends a wait_for_frames in progress
+        if acquisition is not None:
+            acquisition.join(STOP_JOIN_S)
         self.device_index = None
-        self._last_frame = None
         logger.info("RealSense camera closed")
 
+    def _start_acquisition(self) -> None:
+        self._acquisition = FrameAcquisition(lambda: self._grab_once(GRAB_TIMEOUT_MS), name="realsense-acquisition")
+        self._acquisition.start()
+
     def _teardown_pipeline(self) -> None:
         if self._pipeline is not None:
             try:
@@ -375,18 +389,33 @@ class RealSenseD435:
     # Frame capture
     # ------------------------------------------------------------------
     def capture(self, timeout_ms: int = 5000) -> Optional[CameraFrame]:
-        if not self._open:
+        """The next frame acquired after this call (None on timeout or when closed).
+
+        Does not touch the pipeline: it waits for the acquisition thread.
+        Requests waiting at the same moment get the same frame.
+        """
+        acquisition = self._acquisition
+        if not self._open or acquisition is None:
+            return None
+        seq, _ = acquisition.latest()
+        _, frame = acquisition.wait_newer(seq, timeout_ms / 1000.0)
+        return frame
+
+    def _grab_once(self, timeout_ms: int) -> Optional[CameraFrame]:
+        """wait_for_frames + align + numpy conversion. Called by the acquisition thread only."""
+        pipeline, align = self._pipeline, self._align
+        if pipeline is None:
             return None
         try:
             with timer.phase("wait_frames"):
-                frames = self._pipeline.wait_for_frames(timeout_ms)
+                frames = pipeline.wait_for_frames(timeout_ms)
         except Exception as e:
             logger.warning(f"wait_for_frames failed: {e}")
             return None
 
-        if self._align is not None:
+        if align is not None:
             with timer.phase("align"):
-                frames = self._align.process(frames)
+                frames = align.process(frames)
 
         color_frame = frames.get_color_frame()
         depth_frame = frames.get_depth_frame()
@@ -394,9 +423,7 @@ class RealSenseD435:
             return None
 
         with timer.phase("convert"):
-            frame = self._frame_from(color_frame, depth_frame)
-        self._last_frame = frame
-        return frame
+            return self._frame_from(color_frame, depth_frame)
 
     def _frame_from(self, color_frame, depth_frame) -> CameraFrame:
         """numpy BGR + depth in mm from the SDK frames (YUYV decode, USB 2 resize, depth scale)."""
@@ -431,11 +458,12 @@ class RealSenseD435:
         return CameraFrame(rgb=rgb, depth=depth_mm, timestamp=time.time())
 
     def _cached_or_capture(self) -> Optional[CameraFrame]:
-        if (
-            self._last_frame is not None
-            and (time.time() - self._last_frame.timestamp) * 1000.0 < self._frame_cache_ms
-        ):
-            return self._last_frame
+        """The latest frame if younger than frame_cache_ms, else the next one."""
+        acquisition = self._acquisition
+        if acquisition is not None:
+            _, last = acquisition.latest()
+            if last is not None and (time.time() - last.timestamp) * 1000.0 < self._frame_cache_ms:
+                return last
         return self.capture()
 
     def get_rgb(self) -> Optional[np.ndarray]: