scorevision: push artifact

benchmark_latency.py

@@ -0,0 +1,76 @@
+"""
+Replicates TurboVision's public-model latency gate locally:
+p95 per-frame inference latency must be <= 100 ms in a 2 vCPU environment.
+
+Run pinned to 2 CPUs to match the compliance environment:
+
+    taskset -c 0,1 python benchmark_latency.py [--frames 100] [--imgsz 640]
+
+Frames are synthesized at 1280x720 (the validator resizes the long side to
+1280 before calling miners), or loaded from --image-dir if given.
+"""
+
+import argparse
+import os
+import time
+from pathlib import Path
+
+os.environ.setdefault("OMP_NUM_THREADS", "2")
+# The compliance environment has no GPU; hide any local one so torch runs on CPU.
+os.environ["CUDA_VISIBLE_DEVICES"] = ""
+
+import numpy as np
+
+
+def load_frames(image_dir: str | None, n: int) -> list[np.ndarray]:
+    if image_dir:
+        import cv2
+
+        paths = sorted(Path(image_dir).glob("*"))[:n]
+        frames = [cv2.imread(str(p)) for p in paths]
+        frames = [f for f in frames if f is not None]
+        if frames:
+            return frames
+    rng = np.random.default_rng(0)
+    return [
+        rng.integers(0, 255, size=(720, 1280, 3), dtype=np.uint8) for _ in range(n)
+    ]
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--frames", type=int, default=100)
+    parser.add_argument("--imgsz", type=int, default=640)
+    parser.add_argument("--image-dir", default=None)
+    parser.add_argument("--warmup", type=int, default=5)
+    args = parser.parse_args()
+
+    import torch
+
+    torch.set_num_threads(2)
+
+    os.environ["SV_IMG_SIZE"] = str(args.imgsz)
+    from miner import Miner
+
+    miner = Miner(path_hf_repo=Path(__file__).parent)
+    frames = load_frames(args.image_dir, args.frames)
+
+    for frame in frames[: args.warmup]:
+        miner.predict_batch([frame], offset=0, n_keypoints=0)
+
+    latencies_ms = []
+    for frame in frames:
+        start = time.perf_counter()
+        miner.predict_batch([frame], offset=0, n_keypoints=0)
+        latencies_ms.append((time.perf_counter() - start) * 1000)
+
+    latencies_ms.sort()
+    p50 = latencies_ms[len(latencies_ms) // 2]
+    p95 = latencies_ms[int(len(latencies_ms) * 0.95)]
+    print(f"model={miner.model_name} imgsz={args.imgsz} frames={len(frames)}")
+    print(f"p50={p50:.1f} ms  p95={p95:.1f} ms  mean={sum(latencies_ms)/len(latencies_ms):.1f} ms")
+    print("PASS (<=100 ms p95)" if p95 <= 100 else "FAIL (>100 ms p95)")
+
+
+if __name__ == "__main__":
+    main()