perf(checks): add a reproducible cold camera evidence benchmark

benchmarks/camera_frame_stats_benchmark.py

@@ -0,0 +1,259 @@
+"""Benchmark cold ``camera_frame_stats`` throughput on canonical 1080p30 video.
+
+The default fixture is one deterministic 30-second, one-camera episode. Its
+JPEG-to-H.264 transform is timed separately, then every check repetition gets
+a fresh ``Episode`` workdir so neither the remuxed MP4 nor FFmpeg instrument
+cache can turn a cold measurement into a cache hit.
+
+Run: ``uv run python benchmarks/camera_frame_stats_benchmark.py``
+(``--quick`` for a small development fixture). Timings are wall-clock.
+"""
+
+import argparse
+import os
+import statistics
+import subprocess
+import tempfile
+import time
+from dataclasses import dataclass
+from pathlib import Path
+
+import hflow
+from hflow._video_measurements import FrameStatisticsSettings
+from hflow._video_measurements._frame_statistics import frame_statistics_filter_graph
+from hflow.checks import camera_frame_stats
+from hflow.ffmpeg import ffmpeg_path, ffmpeg_version
+from hflow.testing import SyntheticEpisodeSpec, synthesize_episode
+from hflow.transform import TransformConfig, write_canonical_episode
+
+FULL_DURATION_S = 30.0
+FULL_FRAMES_PER_SECOND = 30.0
+FULL_WIDTH = 1920
+FULL_HEIGHT = 1080
+FULL_REPETITIONS = 3
+
+QUICK_DURATION_S = 3.0
+QUICK_FRAMES_PER_SECOND = 10.0
+QUICK_WIDTH = 320
+QUICK_HEIGHT = 180
+QUICK_REPETITIONS = 2
+
+
+@dataclass(frozen=True)
+class BenchmarkSettings:
+    duration_s: float
+    frames_per_second: float
+    width: int
+    height: int
+    repetitions: int
+    profile_filters: bool = False
+
+
+@dataclass(frozen=True)
+class CheckRun:
+    seconds: float
+    decoded_frame_count: int
+    instrument_cache_path: Path
+    instrument_cache_bytes: int
+
+
+@dataclass(frozen=True)
+class FilterProfile:
+    name: str
+    seconds: float
+
+
+@dataclass(frozen=True)
+class BenchmarkResult:
+    transform_seconds: float
+    camera_topic: str
+    ffmpeg_version: str
+    logical_cpu_count: int
+    check_runs: tuple[CheckRun, ...]
+    filter_profiles: tuple[FilterProfile, ...]
+
+
+def _measure_cold_check(
+    canonical_path: Path,
+    camera_topic: str,
+    frames_per_second: float,
+    workdir: Path,
+) -> CheckRun:
+    started = time.perf_counter()
+    with hflow.Episode(canonical_path, workdir=workdir) as episode:
+        result = camera_frame_stats(
+            episode,
+            cameras=[camera_topic],
+            expected_hz={camera_topic: frames_per_second},
+        )
+    seconds = time.perf_counter() - started
+    instrument_cache_paths = list(workdir.glob("*.instrument.*.txt"))
+    if len(instrument_cache_paths) != 1:
+        raise RuntimeError(
+            f"expected one cold instrument cache in {workdir}, got {instrument_cache_paths}"
+        )
+    decoded_frame_count = result.measurements[f"{camera_topic}/decoded_frame_count"]
+    if not isinstance(decoded_frame_count, int):
+        raise TypeError(f"decoded frame count is not an integer: {decoded_frame_count!r}")
+    return CheckRun(
+        seconds=seconds,
+        decoded_frame_count=decoded_frame_count,
+        instrument_cache_path=instrument_cache_paths[0],
+        instrument_cache_bytes=instrument_cache_paths[0].stat().st_size,
+    )
+
+
+def _profile_filter_paths(video_path: Path) -> tuple[FilterProfile, ...]:
+    settings = FrameStatisticsSettings()
+    filters: tuple[tuple[str, str | None], ...] = (
+        ("decode only", None),
+        ("decode + format=yuv420p", "format=pix_fmts=yuv420p"),
+        (
+            "decode + blackframe",
+            "format=pix_fmts=yuv420p,"
+            f"blackframe=amount=0:threshold={settings.black_pixel_luma_threshold}",
+        ),
+        (
+            "decode + freezedetect",
+            "format=pix_fmts=yuv420p,"
+            "freezedetect="
+            f"n={settings.freeze_noise_tolerance_decibels}dB:"
+            f"d={settings.freeze_minimum_duration_seconds}",
+        ),
+        (
+            "decode + signalstats=stat=tout+brng",
+            "format=pix_fmts=yuv420p,signalstats=stat=tout+brng",
+        ),
+        ("complete shipped graph", frame_statistics_filter_graph(settings)),
+    )
+    profiles: list[FilterProfile] = []
+    for name, filter_graph in filters:
+        command = [
+            str(ffmpeg_path()),
+            "-hide_banner",
+            "-loglevel",
+            "error",
+            "-nostats",
+            "-i",
+            str(video_path),
+        ]
+        if filter_graph is not None:
+            command.extend(("-vf", filter_graph))
+        command.extend(("-f", "null", "-"))
+        started = time.perf_counter()
+        completed = subprocess.run(command, capture_output=True, check=False)
+        seconds = time.perf_counter() - started
+        if completed.returncode != 0:
+            raise RuntimeError(
+                f"ffmpeg filter profile {name!r} failed: "
+                f"{completed.stderr.decode(errors='replace')}"
+            )
+        profiles.append(FilterProfile(name=name, seconds=seconds))
+    return tuple(profiles)
+
+
+def run_synthetic_benchmark(settings: BenchmarkSettings) -> BenchmarkResult:
+    with tempfile.TemporaryDirectory(prefix="camera-frame-stats-benchmark-") as directory_name:
+        working_dir = Path(directory_name)
+        source_path = synthesize_episode(
+            working_dir / "source.mcap",
+            SyntheticEpisodeSpec(
+                duration_s=settings.duration_s,
+                cameras=("camera_0",),
+                image_hz=settings.frames_per_second,
+                image_width=settings.width,
+                image_height=settings.height,
+                black_segment=None,
+                joint_jump_at_s=None,
+                timestamp_offset_segment=None,
+            ),
+        )
+        canonical_path = working_dir / "canonical.mcap"
+        transform_started = time.perf_counter()
+        write_canonical_episode(source_path, canonical_path, TransformConfig())
+        transform_seconds = time.perf_counter() - transform_started
+        camera_topic = "/camera_0/compressed"
+        check_runs = tuple(
+            _measure_cold_check(
+                canonical_path,
+                camera_topic,
+                settings.frames_per_second,
+                working_dir / f"cold-check-{repetition + 1}",
+            )
+            for repetition in range(settings.repetitions)
+        )
+        if settings.profile_filters:
+            with hflow.Episode(canonical_path, workdir=working_dir / "filter-profile") as episode:
+                filter_profiles = _profile_filter_paths(episode.video(camera_topic))
+        else:
+            filter_profiles = ()
+    return BenchmarkResult(
+        transform_seconds=transform_seconds,
+        camera_topic=camera_topic,
+        ffmpeg_version=ffmpeg_version(),
+        logical_cpu_count=os.cpu_count() or 1,
+        check_runs=check_runs,
+        filter_profiles=filter_profiles,
+    )
+
+
+def _settings(quick: bool, profile_filters: bool) -> BenchmarkSettings:
+    if quick:
+        return BenchmarkSettings(
+            duration_s=QUICK_DURATION_S,
+            frames_per_second=QUICK_FRAMES_PER_SECOND,
+            width=QUICK_WIDTH,
+            height=QUICK_HEIGHT,
+            repetitions=QUICK_REPETITIONS,
+            profile_filters=profile_filters,
+        )
+    return BenchmarkSettings(
+        duration_s=FULL_DURATION_S,
+        frames_per_second=FULL_FRAMES_PER_SECOND,
+        width=FULL_WIDTH,
+        height=FULL_HEIGHT,
+        repetitions=FULL_REPETITIONS,
+        profile_filters=profile_filters,
+    )
+
+
+def _print_result(settings: BenchmarkSettings, result: BenchmarkResult) -> None:
+    print(
+        "camera_frame_stats benchmark: "
+        f"{settings.duration_s:g}s synthetic episode, one camera @ "
+        f"{settings.frames_per_second:g} Hz, {settings.width}x{settings.height}\n"
+    )
+    print(f"logical CPUs: {result.logical_cpu_count}")
+    print(f"ffmpeg: {result.ffmpeg_version}\n")
+    print("| phase | wall-clock | decoded frames | instrument cache |")
+    print("|---|---:|---:|---:|")
+    print(f"| transform to canonical | {result.transform_seconds:.3f} s | - | - |")
+    for run_index, run in enumerate(result.check_runs, start=1):
+        print(
+            f"| cold camera_frame_stats #{run_index} | {run.seconds:.3f} s "
+            f"| {run.decoded_frame_count} | {run.instrument_cache_bytes / 1000:.1f} KB |"
+        )
+    median_seconds = statistics.median(run.seconds for run in result.check_runs)
+    print(f"\ncold check median: {median_seconds:.3f} s per camera")
+    if result.filter_profiles:
+        print("\n| filter path | wall-clock |")
+        print("|---|---:|")
+        for profile in result.filter_profiles:
+            print(f"| {profile.name} | {profile.seconds:.3f} s |")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--quick", action="store_true", help="use a small development fixture")
+    parser.add_argument(
+        "--profile-filters",
+        action="store_true",
+        help="time decode and each evidence-filter component separately",
+    )
+    arguments = parser.parse_args()
+    settings = _settings(arguments.quick, arguments.profile_filters)
+    _print_result(settings, run_synthetic_benchmark(settings))
+
+
+if __name__ == "__main__":
+    main()

docs/BENCHMARKS.md

@@ -1,9 +1,9 @@
-# HFlow MCAP storage and read benchmarks
+# HFlow storage, read, and camera-check benchmarks
 
-These reproducible benchmarks measure the two performance decisions in HFlow's
-canonical MCAP writer: in-band video compression and topic-group chunking.
-They report what the current implementation achieves at honest small scale
-alongside the million-hour results Dyna published in
+These reproducible benchmarks measure HFlow's in-band video compression,
+topic-group chunking, and cold camera-evidence throughput. They report what the
+current implementation achieves at honest small scale alongside the
+million-hour results Dyna published in
 [Training Dyna-2 at million-hour scale, repeatably](https://www.dyna.co/research/dyna-2-infrastructure)
 (Figure 3). Every number below comes from a real run of the scripts in
 [`benchmarks/`](../benchmarks); nothing is extrapolated.
@@ -13,6 +13,7 @@ alongside the million-hour results Dyna published in
 | Workload | Measured result | Why it matters |
 | --- | --- | --- |
 | Six-camera real footage | **48-50.5% less video payload** than the source JPEG payloads | Quantifies the storage effect without relying on synthetic test patterns |
+| One-camera 1080p30 evidence pass | **4.90 s median** to inspect every frame of a 30 s episode | Establishes the cold `camera_frame_stats` cost without mixing in transform or cache time |
 | Four-camera synthetic training windows | **2.42x fewer chunk fetches** than per-topic chunking | Shows how grouping topics by read pattern reduces sample assembly work |
 | Six-camera real footage with 8 MB chunks | **2.81x fewer fetches and 3.21x fewer bytes fetched** than per-topic chunking | Demonstrates that chunk size and grouping policy must be tuned together |
 | Selective state scans | Naive schema grouping fetched **230 MB** for a 0.2 MB `/imu` stream | Shows why HFlow exposes per-topic group overrides instead of treating grouping as a fixed schema rule |
@@ -29,6 +30,8 @@ sections for the reference datasets):
 ```bash
 uv run python benchmarks/storage_benchmark.py
 uv run python benchmarks/read_benchmark.py
+uv run python benchmarks/camera_frame_stats_benchmark.py
+uv run python benchmarks/camera_frame_stats_benchmark.py --profile-filters
 uv run python benchmarks/storage_benchmark.py --input nuscenes-mini-sample.mcap
 uv run python benchmarks/read_benchmark.py --input nuscenes-mini-sample.mcap \
     --grouping read-pattern --chunk-size-bytes 8000000
@@ -51,11 +54,74 @@ uv run python benchmarks/read_benchmark.py --input robotis-button-push-107.mcap
 - Scale: 11.6-60 s episodes, not the forty-three million of Dyna's corpus. The
   point is that the mechanisms behave as Dyna's article describes, not that
   the ratios match.
+- Camera-check wall time is machine- and content-dependent. Its benchmark uses
+  a fresh `Episode` workdir for every repetition, so neither the remuxed MP4 nor
+  the persistent FFmpeg instrument cache can turn a cold run into a cache hit.
 - Storage reductions compare **video payload bytes** (the codec effect, the
   number comparable to Dyna's ~68%). File sizes are shown alongside but carry
   every non-camera channel passed through byte-for-byte; on a lidar-heavy
   recording those dwarf the cameras and would swamp a file-level comparison.
 
+## Camera evidence: cold per-frame throughput (issue #365)
+
+`camera_frame_stats` measures blackout, freeze, luma, frame-difference,
+temporal-outlier, and broadcast-range evidence for every decoded frame. The
+implementation already shares one FFmpeg filter graph between those
+measurements and caches its instrument output, so this benchmark measures the
+remaining first-run cost rather than reintroducing the repeated decode removed
+by #175.
+
+The script first transforms one deterministic synthetic episode, reports that
+time separately, then runs the check three times with a new workdir each time:
+
+```bash
+uv run python benchmarks/camera_frame_stats_benchmark.py
+# --quick uses a 3 s, 320x180 development fixture
+```
+
+Measured on a Ryzen 9 8945H (8 cores/16 threads), 32 GB RAM, Linux x86_64, and
+FFmpeg `n8.1.2-50-g1a748fe2cd-20260901` at commit `050d145`:
+
+| phase | wall-clock | decoded frames | instrument cache |
+|---|---:|---:|---:|
+| transform to canonical | 3.510 s | - | - |
+| cold `camera_frame_stats` #1 | 4.897 s | 900 | 840.3 KB |
+| cold `camera_frame_stats` #2 | 4.940 s | 900 | 840.3 KB |
+| cold `camera_frame_stats` #3 | 4.797 s | 900 | 840.3 KB |
+
+The cold median is **4.897 s per camera** for 30 seconds of 1920x1080 video at
+30 FPS. All three runs decoded all 900 frames. This synthetic result matches
+the shape of the separately reported real Egocentric-10K control in #365:
+seconds per camera, stable across cold repetitions, and independent of whether
+the canonical H.264 arrived directly or through a JPEG transform.
+
+### Where the time goes
+
+`--profile-filters` runs one-variable FFmpeg controls against the same
+canonical MP4 to isolate the cost:
+
+| filter path | wall-clock |
+|---|---:|
+| decode only | 0.422 s |
+| decode + `format=yuv420p` | 0.518 s |
+| decode + `blackframe` | 0.537 s |
+| decode + `freezedetect` | 0.508 s |
+| decode + `signalstats=stat=tout+brng` | 4.536 s |
+| complete shipped graph | 4.711 s |
+
+`signalstats` is the bottleneck, not H.264 decoding. The additional controls
+recorded in #365 found no repeatable gain from overriding FFmpeg's automatic
+filter threading, splitting the graph into parallel branches, or using NVDEC
+before the software-only evidence filters. A luma-only input is not equivalent
+either: FFmpeg's `BRNG` statistic deliberately evaluates the Y, U, and V
+planes, so removing chroma would silently weaken the recorded evidence.
+
+**Conclusion: keep the measurement path unchanged.** The tested changes do not
+produce a repeatable win without changing evidence or adding complexity that
+outweighs noise-level movement. The benchmark is the durable outcome: future
+FFmpeg releases or alternative instruments now have a reproducible baseline
+and must beat it while preserving every per-frame result.
+
 ## Storage: per-frame JPEG vs canonical MCAP by GOP preset (issue #26)
 
 HFlow's transform re-encodes per-frame JPEG into in-band H.264 with GOP