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 <noreply@anthropic.com>
This commit is contained in:
OliviaCandClaude Opus 5.5 committed 2026-09-29 12:57:25 +02:00
1 parent 441e0bd917
commit 9aa63e4e02
6 files changed
+152 -8

No files matched your search

+19 -5
View File
@@ -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:
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:
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,6 +71,7 @@ def _frame_to_dict(frame: CameraFrame, include_images: bool = True) -> dict:
global _frame_count
_frame_count += 1
with timer.phase("depth_stats"):
result = {
"frame_id": _frame_count,
"timestamp": frame.timestamp,
@@ -74,10 +83,13 @@ def _frame_to_dict(frame: CameraFrame, include_images: bool = True) -> dict:
}
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")
+2
View File
@@ -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,6 +61,7 @@ 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)
with timer.phase("ws_send"):
await ws.send_json(data)
if _saver and _saver.is_recording:
_saver.save_rec_frame(frame.rgb)
+4
View File
@@ -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 = ""
+12 -3
View File
@@ -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,12 +378,14 @@ class RealSenseD435:
if not self._open:
return None
try:
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:
with timer.phase("align"):
frames = self._align.process(frames)
color_frame = frames.get_color_frame()
@@ -389,6 +393,13 @@ class RealSenseD435:
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 (
+69
View File
@@ -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)
+46
View File
@@ -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