| """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 |
| |
| 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 |
|
|
