OliviaCandClaude Opus 5.5
9aa63e4e02
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>
47 lines
1.6 KiB
Python
47 lines
1.6 KiB
Python
"""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
|