From 0c689567f19993cbc3a776bedb460245adc0ae82 Mon Sep 17 00:00:00 2001 From: Galabavamsi <51828882+Galabavamsi@users.noreply.github.com> Date: Thu, 3 Sep 2026 02:41:17 +0530 Subject: [PATCH] perf(checks): add cold camera frame stats benchmark --- benchmarks/camera_frame_stats_benchmark.py | 259 +++++++++++++++++++++ docs/BENCHMARKS.md | 76 +++++- tests/test_camera_frame_stats_benchmark.py | 48 ++++ 3 files changed, 378 insertions(+), 5 deletions(-) create mode 100644 benchmarks/camera_frame_stats_benchmark.py create mode 100644 tests/test_camera_frame_stats_benchmark.py diff --git a/benchmarks/camera_frame_stats_benchmark.py b/benchmarks/camera_frame_stats_benchmark.py new file mode 100644 index 0000000..6172407 --- /dev/null +++ b/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() diff --git a/docs/BENCHMARKS.md b/docs/BENCHMARKS.md index 03b22a9..7ec35ad 100644 --- a/docs/BENCHMARKS.md +++ b/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 diff --git a/tests/test_camera_frame_stats_benchmark.py b/tests/test_camera_frame_stats_benchmark.py new file mode 100644 index 0000000..aa341e2 --- /dev/null +++ b/tests/test_camera_frame_stats_benchmark.py @@ -0,0 +1,48 @@ +import importlib.util +from pathlib import Path +from types import ModuleType + +BENCHMARK_PATH = Path(__file__).resolve().parents[1] / "benchmarks/camera_frame_stats_benchmark.py" + + +def _load_benchmark() -> ModuleType: + spec = importlib.util.spec_from_file_location("camera_frame_stats_benchmark", BENCHMARK_PATH) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_camera_frame_stats_benchmark_measures_each_cold_run_from_canonical_input() -> None: + """Reusing one workdir or measuring the pre-transform source breaks this result.""" + benchmark = _load_benchmark() + + result = benchmark.run_synthetic_benchmark( + benchmark.BenchmarkSettings( + duration_s=1.0, + frames_per_second=10.0, + width=160, + height=90, + repetitions=2, + profile_filters=True, + ) + ) + + assert result.transform_seconds > 0 + assert result.camera_topic == "/camera_0/compressed" + assert result.ffmpeg_version + assert result.logical_cpu_count > 0 + assert len(result.check_runs) == 2 + assert [run.decoded_frame_count for run in result.check_runs] == [10, 10] + assert all(run.seconds > 0 for run in result.check_runs) + assert all(run.instrument_cache_bytes > 0 for run in result.check_runs) + assert len({run.instrument_cache_path.parent for run in result.check_runs}) == 2 + assert [profile.name for profile in result.filter_profiles] == [ + "decode only", + "decode + format=yuv420p", + "decode + blackframe", + "decode + freezedetect", + "decode + signalstats=stat=tout+brng", + "complete shipped graph", + ] + assert all(profile.seconds > 0 for profile in result.filter_profiles)