feat: frame work off the event loop, each frame encoded once

- Endpoints that wait for the camera or encode/compute (capture, rgb, depth,
  depth/color, pointcloud, pointcloud/roi, measure, snapshot, and the SDK
  calls discover/open/close) are plain `def`: FastAPI runs them in its
  thread pool, so one slow request no longer holds the others. Plain `def`
  rather than asyncio.to_thread in each route: it is FastAPI's own mechanism
  and keeps the route bodies unchanged.
- /ws and the NATS snapshot stay async and use asyncio.to_thread for the
  frame wait, the encoding and the recording write.
- EncodedFrame: the JPEG q85, 16-bit PNG, colormap JPEG q90, their base64
  and the depth stats of a frame are computed once, on first use, and
  shared: parallel requests, /ws and NATS on the same frame get the same
  bytes. The last 4 frames' encodings are kept.
- camera.cached_frame(): the whole cached frame (get_rgb/get_depth unchanged).

Contract unchanged: same endpoints, same JSON fields. frame_id still counts
responses (now under a lock); two parallel /api/capture served by one frame
have different frame_id and the same timestamp and images.

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

src/backend/api/routes.py

@@ -2,6 +2,8 @@
 
 import base64
 import logging
+import threading
+from collections import deque
 
 import cv2
 import numpy as np
@@ -21,6 +23,7 @@ router = APIRouter(dependencies=[Depends(verify_api_key)])
 _camera: RealSenseD435 | None = None
 _saver: ImageSaver | None = None
 _frame_count = 0
+_frame_count_lock = threading.Lock()   # frame_id is incremented from worker threads
 
 
 def set_camera(camera: RealSenseD435) -> None:
@@ -67,30 +70,94 @@ def _jet_colormap(depth: np.ndarray) -> np.ndarray:
     return cv2.applyColorMap(vis, cv2.COLORMAP_JET)
 
 
-def _frame_to_dict(frame: CameraFrame, include_images: bool = True) -> dict:
+def _depth_stats(depth: np.ndarray) -> dict:
-    global _frame_count
-    _frame_count += 1
-
     with timer.phase("depth_stats"):
-        result = {
+        valid = depth > 0
-            "frame_id": _frame_count,
+        return {
-            "timestamp": frame.timestamp,
+            "depth_min_mm": int(depth[valid].min()) if valid.any() else 0,
-            "rgb_shape": list(frame.rgb.shape),
+            "depth_max_mm": int(depth.max()),
-            "depth_shape": list(frame.depth.shape),
+            "depth_valid_pixels": int(valid.sum()),
-            "depth_min_mm": int(frame.depth[frame.depth > 0].min()) if (frame.depth > 0).any() else 0,
-            "depth_max_mm": int(frame.depth.max()),
-            "depth_valid_pixels": int((frame.depth > 0).sum()),
         }
 
-    if include_images:
-        rgb_jpeg = _encode_rgb_jpeg(frame.rgb)
-        depth_png = _encode_depth_png(frame.depth)
-        color_jpeg = _encode_rgb_jpeg(_depth_colormap(frame.depth), 90)
-        with timer.phase("base64"):
-            result["rgb_jpeg_b64"] = base64.b64encode(rgb_jpeg).decode()
-            result["depth_png_b64"] = base64.b64encode(depth_png).decode()
-            result["depth_color_b64"] = base64.b64encode(color_jpeg).decode()
 
+def _b64(data: bytes) -> str:
+    with timer.phase("base64"):
+        return base64.b64encode(data).decode()
+
+
+class EncodedFrame:
+    """The encodings of ONE frame, each computed once, on first use, and shared.
+
+    Clients served the same frame (parallel requests, the WebSocket, NATS)
+    get the same bytes instead of encoding it again each.
+    """
+
+    def __init__(self, frame: CameraFrame):
+        self.frame = frame
+        self._lock = threading.RLock()   # reentrant: images_b64() computes the other fields inside
+        self._values: dict[str, object] = {}
+
+    def _once(self, key: str, compute):
+        with self._lock:   # a second caller waits for the first encoding instead of repeating it
+            if key not in self._values:
+                self._values[key] = compute()
+            return self._values[key]
+
+    def rgb_jpeg(self) -> bytes:
+        return self._once("rgb_jpeg", lambda: _encode_rgb_jpeg(self.frame.rgb))
+
+    def depth_png(self) -> bytes:
+        return self._once("depth_png", lambda: _encode_depth_png(self.frame.depth))
+
+    def depth_color_jpeg(self) -> bytes:
+        return self._once("depth_color_jpeg", lambda: _encode_rgb_jpeg(_depth_colormap(self.frame.depth), 90))
+
+    def depth_stats(self) -> dict:
+        return self._once("depth_stats", lambda: _depth_stats(self.frame.depth))
+
+    def images_b64(self) -> dict:
+        return self._once("images_b64", lambda: {
+            "rgb_jpeg_b64": _b64(self.rgb_jpeg()),
+            "depth_png_b64": _b64(self.depth_png()),
+            "depth_color_b64": _b64(self.depth_color_jpeg()),
+        })
+
+
+ENCODED_FRAMES_KEPT = 4   # recent frames whose encodings are kept (REST and WS may be one frame apart)
+_encoded_frames: deque = deque(maxlen=ENCODED_FRAMES_KEPT)
+_encoded_frames_lock = threading.Lock()
+
+
+def encoded(frame: CameraFrame) -> EncodedFrame:
+    """The shared EncodedFrame of this frame (same object for everyone asking)."""
+    with _encoded_frames_lock:
+        for enc in _encoded_frames:
+            if enc.frame is frame:
+                return enc
+        enc = EncodedFrame(frame)
+        _encoded_frames.append(enc)
+        return enc
+
+
+def _next_frame_id() -> int:
+    global _frame_count
+    with _frame_count_lock:
+        _frame_count += 1
+        return _frame_count
+
+
+def _frame_to_dict(frame: CameraFrame, include_images: bool = True) -> dict:
+    """/api/capture and /ws JSON of a frame. frame_id counts responses, as before."""
+    enc = encoded(frame)
+    result = {
+        "frame_id": _next_frame_id(),
+        "timestamp": frame.timestamp,
+        "rgb_shape": list(frame.rgb.shape),
+        "depth_shape": list(frame.depth.shape),
+        **enc.depth_stats(),
+    }
+    if include_images:
+        result.update(enc.images_b64())
     return result
 
 
@@ -162,7 +229,7 @@ async def status():
 
 
 @router.get("/discover")
-async def discover(timeout_ms: int = Query(default=5000, ge=1000, le=30000)):
+def discover(timeout_ms: int = Query(default=5000, ge=1000, le=30000)):
     """Discover available RealSense cameras on USB bus."""
     if not _camera or not _camera.is_initialized():
         return fail("SDK_NOT_INITIALIZED", "SDK not initialized")
@@ -185,7 +252,7 @@ async def discover(timeout_ms: int = Query(default=5000, ge=1000, le=30000)):
 
 
 @router.post("/open")
-async def open_camera(device_index: int = Query(default=0, ge=0)):
+def open_camera(device_index: int = Query(default=0, ge=0)):
     """Open a discovered camera by index and start streaming.
 
     Idempotent: if a camera is already open, returns the current open state
@@ -214,7 +281,7 @@ async def open_camera(device_index: int = Query(default=0, ge=0)):
 
 
 @router.post("/close")
-async def close_camera():
+def close_camera():
     """Close the active camera and stop streaming."""
     if not _camera or not _camera.is_open():
         return fail("CAMERA_NOT_OPEN", "No camera is open")
@@ -224,7 +291,7 @@ async def close_camera():
 
 
 @router.get("/capture")
-async def capture():
+def capture():
     """Capture a single RGB+depth frame. Returns JSON with base64-encoded images."""
     if not _camera or not _camera.is_open():
         return fail("CAMERA_NOT_OPEN", "Camera not open")
@@ -239,47 +306,46 @@ async def capture():
 
 
 @router.get("/rgb")
-async def get_rgb():
+def get_rgb():
     """Get RGB frame as JPEG image."""
     if not _camera or not _camera.is_open():
         raise HTTPException(503, "Camera not open")
 
-    rgb = _camera.get_rgb()
+    frame = _camera.cached_frame()
-    if rgb is None:
+    if frame is None:
         raise HTTPException(504, "No RGB frame")
 
-    return Response(content=_encode_rgb_jpeg(rgb), media_type="image/jpeg")
+    return Response(content=encoded(frame).rgb_jpeg(), media_type="image/jpeg")
 
 
 @router.get("/depth")
-async def get_depth():
+def get_depth():
     """Get depth frame as 16-bit PNG (values in mm)."""
     if not _camera or not _camera.is_open():
         raise HTTPException(503, "Camera not open")
 
-    depth = _camera.get_depth()
+    frame = _camera.cached_frame()
-    if depth is None:
+    if frame is None:
         raise HTTPException(504, "No depth frame")
 
-    return Response(content=_encode_depth_png(depth), media_type="image/png")
+    return Response(content=encoded(frame).depth_png(), media_type="image/png")
 
 
 @router.get("/depth/color")
-async def get_depth_colormap():
+def get_depth_colormap():
     """Get depth frame as JET colormap JPEG."""
     if not _camera or not _camera.is_open():
         raise HTTPException(503, "Camera not open")
 
-    depth = _camera.get_depth()
+    frame = _camera.cached_frame()
-    if depth is None:
+    if frame is None:
         raise HTTPException(504, "No depth frame")
 
-    colormap = _depth_colormap(depth)
+    return Response(content=encoded(frame).depth_color_jpeg(), media_type="image/jpeg")
-    return Response(content=_encode_rgb_jpeg(colormap, 90), media_type="image/jpeg")
 
 
 @router.get("/pointcloud", summary="3D Point cloud", tags=["capture"])
-async def get_pointcloud(
+def get_pointcloud(
     step: int = Query(default=4, ge=1, le=16),
     max_depth_mm: int = Query(default=4000, ge=100, le=10000),
 ):
@@ -329,7 +395,7 @@ async def get_pointcloud(
 
 
 @router.get("/pointcloud/roi", summary="3D Point cloud within ROI", tags=["capture"])
-async def get_pointcloud_roi(
+def get_pointcloud_roi(
     u_min: int = Query(..., ge=0, description="ROI left pixel (inclusive)"),
     v_min: int = Query(..., ge=0, description="ROI top pixel (inclusive)"),
     u_max: int = Query(..., ge=1, description="ROI right pixel (exclusive)"),
@@ -407,7 +473,7 @@ async def get_pointcloud_roi(
 
 
 @router.get("/measure", summary="3D measurement at pixel", tags=["capture"])
-async def measure_point(
+def measure_point(
     u: int = Query(..., ge=0, description="Pixel X in raw image coords (0..W-1)"),
     v: int = Query(..., ge=0, description="Pixel Y in raw image coords (0..H-1)"),
     radius: int = Query(default=6, ge=0, le=50, description="Search radius (px) for nearest valid depth"),
@@ -472,7 +538,7 @@ async def measure_point(
 
 
 @router.post("/snapshot")
-async def snapshot(
+def snapshot(
     flip_h: bool = Query(default=False),
     flip_v: bool = Query(default=False),
 ):

src/backend/api/websocket.py

@@ -56,15 +56,17 @@ async def websocket_stream(ws: WebSocket):
     try:
         while True:
             if streaming and _camera and _camera.is_open():
-                frame = _camera.capture(timeout_ms=500)
+                # Waiting for the frame, encoding and saving run in threads: the
+                # event loop keeps serving REST and the other sockets meanwhile.
+                frame = await asyncio.to_thread(_camera.capture, 500)
                 if frame:
-                    data = _frame_to_dict(frame)
+                    data = await asyncio.to_thread(_frame_to_dict, frame)
                     if predicting and _detector is not None:
                         data["detections"] = await asyncio.to_thread(_detector.detect, frame.rgb)
                     with timer.phase("ws_send"):
                         await ws.send_json(data)
                     if _saver and _saver.is_recording:
-                        _saver.save_rec_frame(frame.rgb)
+                        await asyncio.to_thread(_saver.save_rec_frame, frame.rgb)
                 await asyncio.sleep(0.033)
 
             try:

src/backend/nats_client/handlers.py

@@ -1,8 +1,10 @@
+import asyncio
 import base64
 import json
 import logging
 
 from nats.aio.client import Client
+from src.backend.api.routes import encoded
 from src.backend.nats_client.subjects import Subjects
 
 logger = logging.getLogger(__name__)
@@ -32,20 +34,21 @@ async def register_all(nc: Client):
                 ))
                 return
 
-            import cv2
+            # Off the event loop: waiting for the frame and encoding it block.
-            frame = _camera.capture()
+            frame = await asyncio.to_thread(_camera.capture)
             if frame is None:
                 await msg.respond(_nats_response(
                     False, error={"code": "TIMEOUT", "message": "Frame capture timeout"}
                 ))
                 return
 
-            _, rgb_buf = cv2.imencode(".jpg", frame.rgb, [cv2.IMWRITE_JPEG_QUALITY, 85])
+            # Same encodings (JPEG q85, 16-bit PNG) as the REST API: shared, not redone.
-            _, depth_buf = cv2.imencode(".png", frame.depth)
+            enc = encoded(frame)
+            rgb_jpeg, depth_png = await asyncio.to_thread(lambda: (enc.rgb_jpeg(), enc.depth_png()))
 
             await msg.respond(_nats_response(True, data={
-                "rgb_jpeg_b64": base64.b64encode(rgb_buf.tobytes()).decode(),
+                "rgb_jpeg_b64": base64.b64encode(rgb_jpeg).decode(),
-                "depth_png_b64": base64.b64encode(depth_buf.tobytes()).decode(),
+                "depth_png_b64": base64.b64encode(depth_png).decode(),
                 "timestamp": frame.timestamp,
                 "width": frame.rgb.shape[1],
                 "height": frame.rgb.shape[0],

src/backend/services/camera.py

@@ -466,6 +466,10 @@ class RealSenseD435:
                 return last
         return self.capture()
 
+    def cached_frame(self) -> Optional[CameraFrame]:
+        """Like get_rgb()/get_depth(), but the whole frame (so its encodings can be shared)."""
+        return self._cached_or_capture()
+
     def get_rgb(self) -> Optional[np.ndarray]:
         f = self._cached_or_capture()
         return f.rgb if f else None