"""Run and freeze the label-separated REVYR verification suite. The input manifest and evaluator contract already exist before this runner is started. Model inference, null controls and registration measurements are persisted first. Only then is the evaluator contract opened and applied. """ from __future__ import annotations import hashlib import json from datetime import datetime, timezone from pathlib import Path import cv2 from benchmark_revyr_browser_onnx import analyze from revyr_candidate_detector import build_vehicle_mask, register_reference ROOT = Path(__file__).resolve().parents[1] SUITE = ROOT / "artifacts" / "verified-suite-v1" MANIFEST = SUITE / "input-manifest.json" EVALUATOR = ROOT / "tests" / "verified-suite-evaluator-v1.json" MODEL = ROOT / "assets" / "models" / "cardd-yolov8s-seg-v2.onnx" RAW = SUITE / "raw-inference.json" REPORT = SUITE / "verification-report.json" SUMMARY = SUITE / "verification-summary.json" INTEGRITY = SUITE / "integrity-manifest.json" def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def read_json(path: Path) -> dict: return json.loads(path.read_text(encoding="utf-8")) def write_json(path: Path, value: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, indent=2, ensure_ascii=False), encoding="utf-8") def absolute(relative: str) -> Path: return ROOT / relative def iou(left: dict, right: dict) -> float: x1, y1 = max(left["x"], right["x"]), max(left["y"], right["y"]) x2 = min(left["x"] + left["width"], right["x"] + right["width"]) y2 = min(left["y"] + left["height"], right["y"] + right["height"]) intersection = max(0, x2 - x1) * max(0, y2 - y1) union = left["width"] * left["height"] + right["width"] * right["height"] - intersection return intersection / max(union, 1) def file_matches(record: dict) -> bool: path = absolute(record["path"]) image = cv2.imread(str(path), cv2.IMREAD_COLOR) return bool( path.is_file() and image is not None and sha256(path) == record["sha256"] and path.stat().st_size == record["bytes"] and image.shape[1] == record["width"] and image.shape[0] == record["height"] ) def run_raw(manifest: dict) -> dict: positive = [] null = [] registration = [] integrity = [] for entry in manifest["positive_pairs"]: reference = absolute(entry["reference"]["path"]) returned = absolute(entry["return"]["path"]) integrity.append({ "run_id": entry["run_id"], "reference_ok": file_matches(entry["reference"]), "return_ok": file_matches(entry["return"]), }) positive.append({ "run_id": entry["run_id"], "variant": entry["variant"], "reference_sha256": sha256(reference), "return_sha256": sha256(returned), "findings": analyze(reference, returned), }) for entry in manifest["null_pairs"]: reference = absolute(entry["reference"]["path"]) returned = absolute(entry["return"]["path"]) integrity.append({ "run_id": entry["run_id"], "reference_ok": file_matches(entry["reference"]), "return_ok": file_matches(entry["return"]), }) null.append({ "run_id": entry["run_id"], "variant": entry["variant"], "reference_sha256": sha256(reference), "return_sha256": sha256(returned), "findings": analyze(reference, returned), }) for entry in manifest["registration_pairs"]: reference_path = absolute(entry["reference"]["path"]) return_path = absolute(entry["return"]["path"]) integrity.append({ "run_id": entry["run_id"], "reference_ok": file_matches(entry["reference"]), "return_ok": file_matches(entry["return"]), }) reference = cv2.imread(str(reference_path), cv2.IMREAD_COLOR) returned = cv2.imread(str(return_path), cv2.IMREAD_COLOR) if reference is None or returned is None: raise RuntimeError(f"Cannot decode registration pair {entry['run_id']}") mask = build_vehicle_mask(reference.shape[:2]) _, metrics = register_reference(reference, returned, mask) registration.append({ "run_id": entry["run_id"], "variant": entry["variant"], "reference_sha256": sha256(reference_path), "return_sha256": sha256(return_path), "metrics": metrics, }) raw = { "suite_id": manifest["suite_id"], "phase": "raw inference before evaluator access", "generated_at": datetime.now(timezone.utc).isoformat(), "engine": "public CarDD YOLOv8s-seg v2 browser ONNX plus SIFT/RANSAC registration", "model_sha256": sha256(MODEL), "answers_supplied": False, "integrity": integrity, "positive": positive, "null": null, "registration": registration, } write_json(RAW, raw) return raw def evaluate(raw: dict, evaluator: dict) -> dict: expected = evaluator["positive_expected"] positive_runs = [] for run in raw["positive"]: matches = [] used = set() for item in expected: candidates = [] for index, finding in enumerate(run["findings"]): finding_type = "deformation" if finding["class"] == "dent" else finding["class"] if finding_type not in item["accepted_types"]: continue candidates.append((iou(finding["box"], item["box"]), index)) candidates.sort(reverse=True) best_iou, best_index = candidates[0] if candidates else (0.0, -1) if best_index >= 0: used.add(best_index) matches.append({ "evaluation_id": item["evaluation_id"], "finding_index": best_index + 1 if best_index >= 0 else None, "iou": round(best_iou, 4), "minimum_iou": item["minimum_iou"], "pass": best_iou >= item["minimum_iou"], }) passed = ( len(run["findings"]) == evaluator["positive_pass_rule"]["exact_finding_count"] and len(used) == len(run["findings"]) and all(item["pass"] for item in matches) ) positive_runs.append({**run, "matches": matches, "pass": passed}) null_runs = [ {**run, "pass": len(run["findings"]) == evaluator["null_pass_rule"]["exact_finding_count"]} for run in raw["null"] ] limits = evaluator["registration_pass_rule"] registration_runs = [] for run in raw["registration"]: metrics = run["metrics"] passed = ( metrics["matches"] >= limits["minimum_ratio_matches"] and metrics["inliers"] >= limits["minimum_ransac_inliers"] and metrics["inlier_ratio"] >= limits["minimum_inlier_ratio"] and metrics["rmse"] <= limits["maximum_reprojection_rmse_px"] ) registration_runs.append({**run, "pass": passed}) integrity_runs = [ {**run, "pass": run["reference_ok"] and run["return_ok"]} for run in raw["integrity"] ] tracks = { "input_integrity": {"passed": sum(item["pass"] for item in integrity_runs), "total": len(integrity_runs)}, "damage_detection": {"passed": sum(item["pass"] for item in positive_runs), "total": len(positive_runs)}, "null_control": {"passed": sum(item["pass"] for item in null_runs), "total": len(null_runs)}, "registration": {"passed": sum(item["pass"] for item in registration_runs), "total": len(registration_runs)}, } all_pass = all(track["passed"] == track["total"] for track in tracks.values()) return { "suite_id": raw["suite_id"], "generated_at": datetime.now(timezone.utc).isoformat(), "evaluation_loaded_after_raw_inference": True, "generalization_claim": False, "all_pass": all_pass, "tracks": tracks, "limits": { "proven": "This fixed suite only: ten damage variants, ten null controls and ten registration variants.", "not_proven": "Fleet-wide accuracy, unseen vehicles, unseen camera systems, hidden damage or autonomous liability decisions.", }, "runs": { "input_integrity": integrity_runs, "damage_detection": positive_runs, "null_control": null_runs, "registration": registration_runs, }, } def main() -> None: manifest = read_json(MANIFEST) raw = run_raw(manifest) raw_hash_before_evaluation = sha256(RAW) # This is the first evaluator access in the process. evaluator = read_json(EVALUATOR) report = evaluate(raw, evaluator) report["raw_inference_sha256"] = raw_hash_before_evaluation report["evaluator_sha256"] = sha256(EVALUATOR) write_json(REPORT, report) summary = { "suite_id": report["suite_id"], "generated_at": report["generated_at"], "all_pass": report["all_pass"], "tracks": report["tracks"], "claim_scope": report["limits"], } write_json(SUMMARY, summary) integrity = { "suite_id": report["suite_id"], "generated_at": datetime.now(timezone.utc).isoformat(), "files": [ {"path": path.relative_to(ROOT).as_posix(), "sha256": sha256(path), "bytes": path.stat().st_size} for path in [MANIFEST, EVALUATOR, MODEL, RAW, REPORT, SUMMARY, Path(__file__)] ], } write_json(INTEGRITY, integrity) print(json.dumps(summary, indent=2, ensure_ascii=False)) if not report["all_pass"]: raise SystemExit(1) if __name__ == "__main__": main()