From 1ad422062c1082d1f2fd43057c20ab0ffbd1c3a5 Mon Sep 17 00:00:00 2001 From: Olivia Casadei Date: Tue, 29 Sep 2026 13:03:28 +0200 Subject: [PATCH] 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 --- src/backend/api/routes.py | 148 ++++++++++++++++++++-------- src/backend/api/websocket.py | 8 +- src/backend/nats_client/handlers.py | 15 +-- src/backend/services/camera.py | 4 + 4 files changed, 125 insertions(+), 50 deletions(-) diff --git a/src/backend/api/routes.py b/src/backend/api/routes.py index c93311c..63ac78a 100644 --- a/src/backend/api/routes.py +++ b/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: - global _frame_count - _frame_count += 1 - +def _depth_stats(depth: np.ndarray) -> dict: with timer.phase("depth_stats"): - result = { - "frame_id": _frame_count, - "timestamp": frame.timestamp, - "rgb_shape": list(frame.rgb.shape), - "depth_shape": list(frame.depth.shape), - "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()), + valid = depth > 0 + return { + "depth_min_mm": int(depth[valid].min()) if valid.any() else 0, + "depth_max_mm": int(depth.max()), + "depth_valid_pixels": int(valid.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() - if rgb is None: + frame = _camera.cached_frame() + 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() - if depth is None: + frame = _camera.cached_frame() + 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() - if depth is None: + frame = _camera.cached_frame() + if frame is None: raise HTTPException(504, "No depth frame") - colormap = _depth_colormap(depth) - return Response(content=_encode_rgb_jpeg(colormap, 90), media_type="image/jpeg") + return Response(content=encoded(frame).depth_color_jpeg(), 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), ): diff --git a/src/backend/api/websocket.py b/src/backend/api/websocket.py index 41683d7..ebd3f88 100644 --- a/src/backend/api/websocket.py +++ b/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: diff --git a/src/backend/nats_client/handlers.py b/src/backend/nats_client/handlers.py index 7cbbdb5..b3c490d 100644 --- a/src/backend/nats_client/handlers.py +++ b/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 - frame = _camera.capture() + # Off the event loop: waiting for the frame and encoding it block. + 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]) - _, depth_buf = cv2.imencode(".png", frame.depth) + # Same encodings (JPEG q85, 16-bit PNG) as the REST API: shared, not redone. + 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(), - "depth_png_b64": base64.b64encode(depth_buf.tobytes()).decode(), + "rgb_jpeg_b64": base64.b64encode(rgb_jpeg).decode(), + "depth_png_b64": base64.b64encode(depth_png).decode(), "timestamp": frame.timestamp, "width": frame.rgb.shape[1], "height": frame.rgb.shape[0], diff --git a/src/backend/services/camera.py b/src/backend/services/camera.py index 46f1508..e853be3 100644 --- a/src/backend/services/camera.py +++ b/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