From 9aa63e4e024240a770e5d77b0d8e4e8d8b61d142 Mon Sep 17 00:00:00 2001 From: Olivia Casadei Date: Tue, 29 Sep 2026 12:57:25 +0200 Subject: [PATCH] feat: per-phase timings of the frame path in the log One log line per phase every PHASE_LOG_EVERY samples (default 100, 0 = off): mean and max of wait_frames, align, convert, each JPEG/PNG encoding, colormap, depth_stats, base64, json and ws_send. Meant for the measurement on PC2 via `docker logs` (optimize-streams R01): where a capture spends its ~130 ms of CPU is not known yet. capture() moves its numpy conversion into _frame_from() unchanged, only so the block can be timed; _depth_colormap() likewise wraps _jet_colormap(). Co-Authored-By: Claude Opus 5.5 --- src/backend/api/routes.py | 46 ++++++++++++------- src/backend/api/websocket.py | 4 +- src/backend/config.py | 4 ++ src/backend/services/camera.py | 19 ++++++-- src/backend/services/phase_timing.py | 69 ++++++++++++++++++++++++++++ tests/test_phase_timing.py | 46 +++++++++++++++++++ 6 files changed, 166 insertions(+), 22 deletions(-) create mode 100644 src/backend/services/phase_timing.py create mode 100644 tests/test_phase_timing.py diff --git a/src/backend/api/routes.py b/src/backend/api/routes.py index 4aaed83..c93311c 100644 --- a/src/backend/api/routes.py +++ b/src/backend/api/routes.py @@ -12,6 +12,7 @@ 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 from src.backend.services.image_saver import ImageSaver +from src.backend.services.phase_timing import timer logger = logging.getLogger(__name__) @@ -37,16 +38,23 @@ def set_image_saver(saver: ImageSaver) -> None: # --------------------------------------------------------------------------- def _encode_rgb_jpeg(rgb: np.ndarray, quality: int = 85) -> bytes: - _, buf = cv2.imencode(".jpg", rgb, [cv2.IMWRITE_JPEG_QUALITY, quality]) + with timer.phase(f"jpeg_q{quality}"): + _, buf = cv2.imencode(".jpg", rgb, [cv2.IMWRITE_JPEG_QUALITY, quality]) return buf.tobytes() def _encode_depth_png(depth: np.ndarray) -> bytes: - _, buf = cv2.imencode(".png", depth) + with timer.phase("png_depth"): + _, buf = cv2.imencode(".png", depth) return buf.tobytes() def _depth_colormap(depth: np.ndarray) -> np.ndarray: + with timer.phase("colormap"): + return _jet_colormap(depth) + + +def _jet_colormap(depth: np.ndarray) -> np.ndarray: valid = depth[depth > 0] if len(valid) == 0: return np.zeros((*depth.shape, 3), dtype=np.uint8) @@ -63,21 +71,25 @@ def _frame_to_dict(frame: CameraFrame, include_images: bool = True) -> dict: global _frame_count _frame_count += 1 - 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()), - } + 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()), + } if include_images: - result["rgb_jpeg_b64"] = base64.b64encode(_encode_rgb_jpeg(frame.rgb)).decode() - result["depth_png_b64"] = base64.b64encode(_encode_depth_png(frame.depth)).decode() - colormap = _depth_colormap(frame.depth) - result["depth_color_b64"] = base64.b64encode(_encode_rgb_jpeg(colormap, 90)).decode() + 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() return result @@ -221,7 +233,9 @@ async def capture(): if frame is None: return fail("CAPTURE_TIMEOUT", "Frame capture timeout") - return JSONResponse(ok(_frame_to_dict(frame))) + data = _frame_to_dict(frame) + with timer.phase("json"): + return JSONResponse(ok(data)) @router.get("/rgb") diff --git a/src/backend/api/websocket.py b/src/backend/api/websocket.py index 9318924..41683d7 100644 --- a/src/backend/api/websocket.py +++ b/src/backend/api/websocket.py @@ -5,6 +5,7 @@ from fastapi import APIRouter, WebSocket, WebSocketDisconnect from src.backend.api.routes import _frame_to_dict from src.backend.services.image_saver import ImageSaver +from src.backend.services.phase_timing import timer try: from src.backend.services.detector import Detector @@ -60,7 +61,8 @@ async def websocket_stream(ws: WebSocket): data = _frame_to_dict(frame) if predicting and _detector is not None: data["detections"] = await asyncio.to_thread(_detector.detect, frame.rgb) - await ws.send_json(data) + with timer.phase("ws_send"): + await ws.send_json(data) if _saver and _saver.is_recording: _saver.save_rec_frame(frame.rgb) await asyncio.sleep(0.033) diff --git a/src/backend/config.py b/src/backend/config.py index 515f7ed..dfcf1b0 100644 --- a/src/backend/config.py +++ b/src/backend/config.py @@ -26,6 +26,10 @@ class Settings(BaseSettings): camera_fps: int = 30 camera_auto_open: bool = True + # Frame-path timings: one log line per phase every N samples (0 = off). + # Read from `docker logs` to see where a capture spends its time on PC2. + phase_log_every: int = 100 + # YOLO detector (optional — disabled if model file or onnxruntime missing) yolo_onnx_path: str = "/app/models/best.onnx" yolo_classes: str = "" diff --git a/src/backend/services/camera.py b/src/backend/services/camera.py index 5ff5ae4..480cc59 100644 --- a/src/backend/services/camera.py +++ b/src/backend/services/camera.py @@ -30,6 +30,8 @@ import cv2 import numpy as np import pyrealsense2 as rs +from src.backend.services.phase_timing import timer + logger = logging.getLogger(__name__) @@ -376,19 +378,28 @@ class RealSenseD435: if not self._open: return None try: - frames = self._pipeline.wait_for_frames(timeout_ms) + with timer.phase("wait_frames"): + frames = self._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: - frames = self._align.process(frames) + with timer.phase("align"): + frames = self._align.process(frames) color_frame = frames.get_color_frame() depth_frame = frames.get_depth_frame() if not color_frame or not depth_frame: return None + with timer.phase("convert"): + frame = self._frame_from(color_frame, depth_frame) + self._last_frame = frame + return 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).""" if self._color_format == rs.format.yuyv: # SDK exposes YUYV as a 1-channel view of the packed buffer; # rebuild it as (H, W, 2) uint8 so cv2 can decode it to BGR. @@ -417,9 +428,7 @@ class RealSenseD435: else: depth_mm = (depth_raw.astype(np.float32) * self._depth_scale * 1000.0).astype(np.uint16) - frame = CameraFrame(rgb=rgb, depth=depth_mm, timestamp=time.time()) - self._last_frame = frame - return frame + return CameraFrame(rgb=rgb, depth=depth_mm, timestamp=time.time()) def _cached_or_capture(self) -> Optional[CameraFrame]: if ( diff --git a/src/backend/services/phase_timing.py b/src/backend/services/phase_timing.py new file mode 100644 index 0000000..e073a35 --- /dev/null +++ b/src/backend/services/phase_timing.py @@ -0,0 +1,69 @@ +"""Per-phase durations of the frame path, summarised in the log every N samples. + +Cheap enough to stay on in production (one perf_counter pair per phase and a +few floats under a lock). The summary lines are what the measurement on PC2 +reads from `docker logs`: mean and max of each phase (wait for the frame, +align, conversions, each encoding, base64/JSON). +""" + +import logging +import threading +import time +from contextlib import contextmanager +from dataclasses import dataclass + +from src.backend.config import settings + +logger = logging.getLogger(__name__) + + +@dataclass +class PhaseStats: + count: int = 0 + total_s: float = 0.0 + max_s: float = 0.0 + + +def summary_line(name: str, stats: PhaseStats) -> str: + """One log line for a phase: sample count, mean and max in milliseconds.""" + mean_ms = stats.total_s / stats.count * 1000.0 if stats.count else 0.0 + return f"phase {name}: n={stats.count} mean={mean_ms:.1f} ms max={stats.max_s * 1000.0:.1f} ms" + + +class PhaseTimer: + """Accumulates durations per phase name; logs and resets a phase every `log_every` samples. + + Safe to call from several threads. `log_every <= 0` turns it off. + """ + + def __init__(self, log_every: int): + self._log_every = log_every + self._lock = threading.Lock() + self._stats: dict[str, PhaseStats] = {} + + def record(self, name: str, seconds: float) -> None: + if self._log_every <= 0: + return + line = None + with self._lock: + stats = self._stats.setdefault(name, PhaseStats()) + stats.count += 1 + stats.total_s += seconds + stats.max_s = max(stats.max_s, seconds) + if stats.count >= self._log_every: + line = summary_line(name, stats) + self._stats[name] = PhaseStats() + if line: + logger.info(line) + + @contextmanager + def phase(self, name: str): + """Time the enclosed block as `name` (recorded also when the block raises).""" + start = time.perf_counter() + try: + yield + finally: + self.record(name, time.perf_counter() - start) + + +timer = PhaseTimer(settings.phase_log_every) diff --git a/tests/test_phase_timing.py b/tests/test_phase_timing.py new file mode 100644 index 0000000..43cc55b --- /dev/null +++ b/tests/test_phase_timing.py @@ -0,0 +1,46 @@ +"""PhaseTimer: per-phase mean/max, logged and reset every N samples.""" + +import logging + +import pytest + +from src.backend.services.phase_timing import PhaseStats, PhaseTimer, summary_line + + +def test_summary_line_mean_and_max_in_ms(): + line = summary_line("align", PhaseStats(count=4, total_s=0.2, max_s=0.08)) + assert line == "phase align: n=4 mean=50.0 ms max=80.0 ms" + + +def test_logs_and_resets_every_n_samples(caplog): + timer = PhaseTimer(log_every=3) + with caplog.at_level(logging.INFO, logger="src.backend.services.phase_timing"): + for seconds in (0.010, 0.030, 0.020, 0.005): + timer.record("wait_frames", seconds) + lines = [r.getMessage() for r in caplog.records] + assert lines == ["phase wait_frames: n=3 mean=20.0 ms max=30.0 ms"] + + +def test_phases_are_counted_separately(caplog): + timer = PhaseTimer(log_every=2) + with caplog.at_level(logging.INFO, logger="src.backend.services.phase_timing"): + timer.record("a", 0.001) + timer.record("b", 0.001) + timer.record("a", 0.003) + assert [r.getMessage() for r in caplog.records] == ["phase a: n=2 mean=2.0 ms max=3.0 ms"] + + +def test_zero_turns_it_off(caplog): + timer = PhaseTimer(log_every=0) + with caplog.at_level(logging.INFO, logger="src.backend.services.phase_timing"): + for _ in range(5): + timer.record("x", 0.001) + assert caplog.records == [] + + +def test_phase_records_even_when_the_block_raises(): + timer = PhaseTimer(log_every=1000) + with pytest.raises(RuntimeError): + with timer.phase("encode"): + raise RuntimeError("boom") + assert timer._stats["encode"].count == 1