File size: 3,553 Bytes
5fd5dd9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | """Pre-deploy check for the public Detect-Vehicle path.
Mirrors the parts of the validator that reject a repo before score matters:
an ONNX-only tree under 30 MB, and one 1024 frame through predict_batch
with ONNX Runtime pinned to 2 threads. The checker budget is 100 ms
(plus 10 ms tolerance). This script fails at 100 ms.
"""
from __future__ import annotations
import sys
import time
from pathlib import Path
import cv2
from miner import Miner
ROOT = Path(__file__).resolve().parent
MAX_REPO_BYTES = 30 * 1024 * 1024
LATENCY_MS = 100.0
# compliance.py uses the 80th percentile and names it p95.
LATENCY_PERCENTILE = 0.80
N_KEYPOINTS = 32
ALLOWED_SUFFIXES = {".onnx", ".py", ".json", ".txt", ".md", ".yml", ".yaml", ".labels", ".csv"}
DISALLOWED_SUFFIXES = {".so", ".dylib", ".dll", ".exe", ".bin", ".pt", ".pth", ".ckpt", ".safetensors"}
SAMPLE = (
ROOT.parent
/ "challenge_data"
/ "Detect-Vehicle"
/ "challenges"
/ "block_009194700_task_85697"
/ "frame_0.png"
)
NAMES = {0: "car", 1: "bus", 2: "truck", 3: "motorcycle", 4: "van"}
def _check_repo() -> None:
total = 0
onnx_count = 0
for path in ROOT.rglob("*"):
if not path.is_file():
continue
if "__pycache__" in path.parts or path.suffix.lower() == ".pyc":
continue
suffix = path.suffix.lower()
total += path.stat().st_size
if suffix == ".onnx":
onnx_count += 1
if suffix in DISALLOWED_SUFFIXES:
raise SystemExit(f"disallowed file for the HF repo: {path.name}")
if suffix and suffix not in ALLOWED_SUFFIXES:
raise SystemExit(f"unsupported file for the HF repo: {path.name}")
if onnx_count < 1:
raise SystemExit("no ONNX weights in the miner directory")
if total > MAX_REPO_BYTES:
raise SystemExit(f"repo is {total} bytes, limit is {MAX_REPO_BYTES}")
print(f"repo: {total / 1e6:.1f} MB, onnx files {onnx_count}")
def _percentile(samples: list[float]) -> float:
ordered = sorted(samples)
index = min(len(ordered) - 1, max(0, int(LATENCY_PERCENTILE * (len(ordered) - 1))))
return ordered[index]
def main() -> None:
_check_repo()
frame = cv2.imread(str(SAMPLE))
if frame is None or frame.shape[0] != 1024 or frame.shape[1] != 1024:
raise SystemExit(f"expected a 1024x1024 challenge frame at {SAMPLE}")
miner = Miner(ROOT)
print(miner)
for _ in range(3):
miner.predict_batch([frame], offset=0, n_keypoints=N_KEYPOINTS)
times_ms: list[float] = []
last = None
for _ in range(15):
started = time.perf_counter()
last = miner.predict_batch([frame], offset=0, n_keypoints=N_KEYPOINTS)
times_ms.append((time.perf_counter() - started) * 1000)
if last is None:
raise SystemExit("no prediction")
latency = _percentile(times_ms)
boxes = last[0].boxes or []
print(f"1024 frame, network {miner.input_width}: p80 {latency:.1f} ms, boxes {len(boxes)}")
for box in boxes[:8]:
print(
f" cls={NAMES.get(box.cls_id, box.cls_id)} conf={box.conf:.2f} "
f"[{box.x1},{box.y1},{box.x2},{box.y2}]"
)
if latency > LATENCY_MS:
raise SystemExit(f"latency {latency:.1f} ms exceeds {LATENCY_MS:.0f} ms")
print("OK — repo and latency match the public checker")
if __name__ == "__main__":
try:
main()
except SystemExit:
raise
except Exception as exc:
print(f"FAILED: {exc}", file=sys.stderr)
raise SystemExit(1) from exc
|