feat: GET /api/raw/latest, color + depth of one frameset without align

For aligning on the host: on the Jetson rs.align.process costs ~104 ms and
holds the GIL (optimize-streams R04/R05). JSON in the usual envelope, one
response: frameset_seq, camera_timestamp_ms, timestamp_domain,
host_timestamp, color_jpeg_b64 (color grid), depth_png_b64 (16-bit PNG, mm,
depth grid), color_shape, depth_shape, depth_units. after=N / timeout_ms as
/api/color/latest. JSON + base64 like /api/capture: every client already
parses that envelope, one request gets both images, base64 costs ~0.1 ms.
Built once per frameset (EncodedRaw, last 4 kept); the color JPEG is the
color channel's own (same bytes). The depth is converted only on request
(camera.raw_frame, phase raw_depth; PNG in phase png_raw_depth).
camera: _published() shared by color_frame() and raw_frame(); _depth_mm()
extracted unchanged from _aligned_frame().

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

src/backend/api/routes.py

@@ -12,7 +12,7 @@ from fastapi.responses import Response, JSONResponse
 
 from src.backend.api.dependencies import verify_api_key
 from src.backend.models.api import ok, fail
-from src.backend.services.camera import RealSenseD435, CameraFrame, ColorFrame
+from src.backend.services.camera import RealSenseD435, CameraFrame, ColorFrame, RawFrame
 from src.backend.services.image_saver import ImageSaver
 from src.backend.services.phase_timing import timer
 
@@ -187,6 +187,56 @@ def color_headers(seq: int, color: ColorFrame) -> dict:
     }
 
 
+ENCODED_RAWS_KEPT = 4   # recent framesets whose raw depth PNG is kept
+
+
+class EncodedRaw:
+    """/api/raw/latest payload of one frameset, built once for all its clients.
+
+    The color JPEG is the one of the color channel (same frameset, same bytes).
+    """
+
+    def __init__(self, seq: int, raw: RawFrame):
+        self.seq = seq
+        self.raw = raw
+        self._lock = threading.Lock()
+        self._payload: dict | None = None
+
+    def payload(self) -> dict:
+        with self._lock:
+            if self._payload is None:
+                with timer.phase("png_raw_depth"):
+                    _, buf = cv2.imencode(".png", self.raw.depth)
+                color = self.raw.color
+                self._payload = {
+                    "frameset_seq": self.seq,
+                    "camera_timestamp_ms": color.camera_timestamp_ms,
+                    "timestamp_domain": color.timestamp_domain,
+                    "host_timestamp": color.host_timestamp,
+                    "color_shape": list(color.rgb.shape),
+                    "depth_shape": list(self.raw.depth.shape),
+                    "depth_units": "mm",
+                    "color_jpeg_b64": _b64(encoded_color(self.seq, color).jpeg()),
+                    "depth_png_b64": _b64(buf.tobytes()),
+                }
+            return self._payload
+
+
+_encoded_raws: deque = deque(maxlen=ENCODED_RAWS_KEPT)
+_encoded_raws_lock = threading.Lock()
+
+
+def encoded_raw(seq: int, raw: RawFrame) -> EncodedRaw:
+    """The shared EncodedRaw of frameset number `seq` (built from the first RawFrame seen for it)."""
+    with _encoded_raws_lock:
+        for enc in _encoded_raws:
+            if enc.seq == seq and enc.raw.color is raw.color:
+                return enc
+        enc = EncodedRaw(seq, raw)
+        _encoded_raws.append(enc)
+        return enc
+
+
 def _next_frame_id() -> int:
     global _frame_count
     with _frame_count_lock:
@@ -417,6 +467,33 @@ def get_color_latest(
                     headers=color_headers(seq, color))
 
 
+@router.get("/raw/latest", summary="Raw color + depth of one frameset (no align)", tags=["capture"])
+def get_raw_latest(
+    after: int | None = Query(default=None, ge=0, description="Wait for a frameset numbered above this"),
+    timeout_ms: int = Query(default=1000, ge=0, le=COLOR_WAIT_MAX_MS),
+):
+    """Color and depth of the SAME frameset, NOT aligned, for aligning on the host.
+
+    JSON in the usual envelope: frameset_seq (the numbering of /api/color/latest
+    and of frameset_seq in /api/capture), camera_timestamp_ms, timestamp_domain,
+    host_timestamp, color_jpeg_b64 (color sensor grid), depth_png_b64 (16-bit
+    PNG, millimeters, depth sensor grid), color_shape, depth_shape,
+    depth_units. Calibration to align them: /api/intrinsics. With `after=N`,
+    waits up to `timeout_ms` for a frameset numbered above N. 504 when none
+    arrives in time.
+    """
+    if not _camera or not _camera.is_open():
+        return fail("CAMERA_NOT_OPEN", "Camera not open")
+
+    seq, raw = _camera.raw_frame(after, timeout_ms)
+    if raw is None:
+        raise HTTPException(504, "No new frameset")
+
+    payload = encoded_raw(seq, raw).payload()
+    with timer.phase("json_raw"):
+        return JSONResponse(ok(payload))
+
+
 @router.get("/pointcloud", summary="3D Point cloud", tags=["capture"])
 def get_pointcloud(
     step: int = Query(default=4, ge=1, le=16),

src/backend/services/camera.py

@@ -68,6 +68,16 @@ class RawFrameset:
     frameset: object
 
 
+@dataclass
+class RawFrame:
+    """Color and depth of ONE frameset, NOT aligned: each on its own sensor's grid.
+
+    For aligning on the host. Holds no SDK memory.
+    """
+    color: ColorFrame
+    depth: np.ndarray             # (H, W) uint16 millimeters, depth sensor grid, own memory
+
+
 def _domain_name(domain) -> str:
     """'timestamp_domain.hardware_clock' -> 'hardware_clock'."""
     return str(domain).split(".")[-1]
@@ -460,13 +470,34 @@ class RealSenseD435:
         With `after_seq`, waits up to `timeout_ms` for a frame numbered above it;
         without, returns the latest one (waiting only for the very first).
         """
+        seq, published = self._published(after_seq, timeout_ms)
+        return seq, (published.color if published is not None else None)
+
+    def raw_frame(self, after_seq: Optional[int] = None, timeout_ms: int = 1000) -> tuple[int, Optional[RawFrame]]:
+        """(number, color + unaligned depth) of one frameset of the continuous thread; no align.
+
+        Same numbering and waiting as color_frame(). The depth is converted
+        here, on request, never at the camera rate.
+        """
+        seq, published = self._published(after_seq, timeout_ms)
+        if published is None:
+            return seq, None
+        depth_frame = published.frameset.get_depth_frame()
+        if not depth_frame:
+            return seq, None
+        with timer.phase("raw_depth"):
+            depth = _owned(self._depth_mm(np.asanyarray(depth_frame.get_data())))
+        return seq, RawFrame(color=published.color, depth=depth)
+
+    def _published(self, after_seq: Optional[int], timeout_ms: int) -> tuple[int, Optional[RawFrameset]]:
+        """Latest frameset of the continuous thread, or the first one numbered above `after_seq`."""
         raw = self._raw
         if not self._open or raw is None:
             return 0, None
         seq, published = raw.latest()
         if after_seq is not None or published is None:
             seq, published = raw.wait_newer(after_seq or 0, timeout_ms / 1000.0)
-        return seq, (published.color if published is not None else None)
+        return seq, published
 
     def _grab_raw(self, timeout_ms: int) -> Optional[RawFrameset]:
         """wait_for_frames, and the color image as an owned BGR array. Continuous thread only.
@@ -548,13 +579,14 @@ class RealSenseD435:
                 interpolation=cv2.INTER_LINEAR,
             )
 
+        return CameraFrame(rgb=rgb, depth=self._depth_mm(depth_raw), timestamp=time.time())
+
+    def _depth_mm(self, depth_raw: np.ndarray) -> np.ndarray:
+        """Depth in millimeters from raw z16 units (a view when the scale is already 1 mm)."""
         # depth_scale ~ 0.001 m/unit on D4xx → raw uint16 is already millimeters.
         if abs(self._depth_scale - 0.001) < 1e-6:
-            depth_mm = depth_raw
-        else:
-            depth_mm = (depth_raw.astype(np.float32) * self._depth_scale * 1000.0).astype(np.uint16)
-
-        return CameraFrame(rgb=rgb, depth=depth_mm, timestamp=time.time())
+            return depth_raw
+        return (depth_raw.astype(np.float32) * self._depth_scale * 1000.0).astype(np.uint16)
 
     def _cached_or_capture(self) -> Optional[CameraFrame]:
         """The frame the cached readers share if younger than frame_cache_ms, else the next aligned one.