#!/usr/bin/env python3 """Independently validate the full OD06/OD07 COCO accuracy result bundle.""" from __future__ import annotations import argparse import contextlib import csv import gzip import hashlib import io import importlib.metadata import json import math import os import tempfile from collections import Counter from pathlib import Path from typing import Any METRIC_NAMES = ( "bbox_ap", "bbox_ap50", "bbox_ap75", "bbox_ap_small", "bbox_ap_medium", "bbox_ap_large", "bbox_ar_1", "bbox_ar_10", "bbox_ar_100", "bbox_ar_small", "bbox_ar_medium", "bbox_ar_large", ) def sha256_file(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 atomic_json(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle: json.dump(value, handle, ensure_ascii=False, indent=2, sort_keys=True) handle.write("\n") temporary = Path(handle.name) os.replace(temporary, path) def read_predictions(path: Path) -> list[dict[str, Any]]: with gzip.open(path, "rt", encoding="utf-8") as handle: value = json.load(handle) if not isinstance(value, list): raise ValueError(f"prediction root must be a list: {path}") return value def independent_coco_metrics(annotation_path: Path, predictions: list[dict[str, Any]]) -> dict[str, float]: import numpy as np from pycocotools.coco import COCO from pycocotools.cocoeval import COCOeval if "float" not in np.__dict__: np.__dict__["float"] = float with contextlib.redirect_stdout(io.StringIO()): ground_truth = COCO(str(annotation_path)) detections = ground_truth.loadRes(predictions) evaluator = COCOeval(ground_truth, detections, iouType="bbox") evaluator.params.imgIds = sorted(ground_truth.getImgIds()) evaluator.params.catIds = sorted(ground_truth.getCatIds()) evaluator.params.maxDets = [1, 10, 100] evaluator.evaluate() evaluator.accumulate() evaluator.summarize() return {name: float(evaluator.stats[index]) for index, name in enumerate(METRIC_NAMES)} def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", required=True, type=Path) parser.add_argument("--config", required=True, type=Path) parser.add_argument("--result-dir", required=True, type=Path) parser.add_argument("--output", required=True, type=Path) args = parser.parse_args() root = args.repo_root.resolve() config_path = args.config.resolve() result_dir = args.result_dir.resolve() output = args.output.resolve() config = json.loads(config_path.read_text(encoding="utf-8")) summary_path = result_dir / "quality_summary.json" summary = json.loads(summary_path.read_text(encoding="utf-8")) metadata_audit = json.loads((result_dir / "metadata_audit.json").read_text(encoding="utf-8")) input_integrity = json.loads((result_dir / "input_integrity.json").read_text(encoding="utf-8")) annotation_path = root / config["dataset"]["annotation_path"] annotations = json.loads(annotation_path.read_text(encoding="utf-8")) expected_image_ids = {int(row["id"]) for row in annotations["images"]} expected_category_ids = {int(row["id"]) for row in annotations["categories"]} expected_category_names = {str(row["name"]) for row in annotations["categories"]} checks: list[dict[str, Any]] = [] def check(name: str, condition: bool, detail: Any) -> None: checks.append({"name": name, "pass": bool(condition), "detail": detail}) check("summary_status", summary.get("status") == "PASS", summary.get("status")) check( "completion_without_threshold_claim", summary.get("acceptance_status") == "MEASURED_NO_EXACT_ARTIFACT_THRESHOLD", summary.get("acceptance_status"), ) check( "official_full_split", len(expected_image_ids) == config["dataset"]["expected_images"] == summary["dataset"]["image_count"] == 5000, {"annotation_images": len(expected_image_ids), "summary_images": summary["dataset"]["image_count"]}, ) check("official_80_categories", len(expected_category_ids) == 80, len(expected_category_ids)) check( "annotation_identity", sha256_file(annotation_path) == config["dataset"]["annotation_sha256"] == summary["dataset"]["annotation_sha256"], sha256_file(annotation_path), ) check( "archive_identity", all( sha256_file(root / config["dataset"][key]["path"]) == config["dataset"][key]["expected_sha256"] for key in ("image_archive", "annotation_archive") ), input_integrity["archives"], ) check("no_prohibited_operations", not any(summary["prohibited_operations"].values()), summary["prohibited_operations"]) check("latency_not_recorded", summary.get("latency_recorded") is False, summary.get("latency_recorded")) installed_cocoeval = Path(__import__("pycocotools.cocoeval", fromlist=["COCOeval"]).__file__).resolve() pinned_cocoeval = next( root / source["path"] for source in config["authoritative_sources"] if source["name"] == "cocoapi_cocoeval" ) check( "official_cocoapi_runtime_source", importlib.metadata.version("pycocotools") == "2.0" and sha256_file(installed_cocoeval) == sha256_file(pinned_cocoeval), { "version": importlib.metadata.version("pycocotools"), "installed_cocoeval_sha256": sha256_file(installed_cocoeval), "pinned_cocoeval_sha256": sha256_file(pinned_cocoeval), }, ) evaluator_path = root / summary["runtime"]["evaluator_path"] check( "evaluator_checksum_bound_to_workers", sha256_file(evaluator_path) == summary["runtime"]["evaluator_sha256"] and all( summary["models"][model_id]["variants"][variant]["worker_summary"]["signature"]["evaluator_sha256"] == summary["runtime"]["evaluator_sha256"] for model_id in ("OD06", "OD07") for variant in ("fp32", "public_int8") ), summary["runtime"], ) check( "published_values_context_only", all( summary["models"][model_id]["published_context"]["use"] == "CONTEXT_ONLY" and summary["models"][model_id]["published_context"]["exact_artifact_threshold"] is False and summary["models"][model_id]["pair_comparison"]["published_comparison_validity"] == "DESCRIPTIVE_ONLY_DIFFERENT_ARTIFACT" for model_id in ("OD06", "OD07") ), {model_id: summary["models"][model_id]["published_context"] for model_id in ("OD06", "OD07")}, ) recomputed: dict[str, Any] = {} prediction_integrity: dict[str, Any] = {} for model_id in ("OD06", "OD07"): model_config = config["models"][model_id] model_summary = summary["models"][model_id] audit_variants = metadata_audit["models"][model_id] check( f"{model_id}_paired_decoder_metadata", audit_variants["fp32"]["detector_metadata_sha256"] == audit_variants["public_int8"]["detector_metadata_sha256"] == model_config["expected_detector_metadata_sha256"], { "fp32": audit_variants["fp32"]["detector_metadata_sha256"], "public_int8": audit_variants["public_int8"]["detector_metadata_sha256"], }, ) check( f"{model_id}_anchor_and_decoder_count", all( audit_variants[variant]["fixed_anchor_count"] == model_config["expected_anchor_count"] and audit_variants[variant]["decoding"]["num_boxes"] == model_config["expected_anchor_count"] and audit_variants[variant]["decoding"]["num_classes"] == 90 for variant in ("fp32", "public_int8") ), {variant: audit_variants[variant]["decoding"] for variant in ("fp32", "public_int8")}, ) check( f"{model_id}_embedded_sparse_labels_match_coco", all( {label for label in audit_variants[variant]["labels"] if label != "???"} == expected_category_names and audit_variants[variant]["placeholder_label_count"] == 10 for variant in ("fp32", "public_int8") ), { variant: { "label_count": audit_variants[variant]["label_count"], "placeholder_count": audit_variants[variant]["placeholder_label_count"], } for variant in ("fp32", "public_int8") }, ) recomputed[model_id] = {} prediction_integrity[model_id] = {} for variant in ("fp32", "public_int8"): variant_config = model_config["variants"][variant] variant_summary = model_summary["variants"][variant] check( f"{model_id}_{variant}_full_completion_scope", variant_summary["status"] == "PASS" and variant_summary["completion_scope"] == "FULL_COCO_VAL2017_5000" and variant_summary["image_count"] == 5000 and variant_summary["worker_summary"]["image_count"] == 5000, { "status": variant_summary["status"], "scope": variant_summary["completion_scope"], "image_count": variant_summary["image_count"], "worker_image_count": variant_summary["worker_summary"]["image_count"], }, ) model_path = root / variant_config["path"] check( f"{model_id}_{variant}_exact_artifact", sha256_file(model_path) == variant_config["expected_sha256"], sha256_file(model_path), ) predictions_path = root / variant_summary["predictions_path"] predictions = read_predictions(predictions_path) counts = Counter(int(row["image_id"]) for row in predictions) record_shapes_valid = all( set(row) == {"image_id", "category_id", "bbox", "score"} and int(row["image_id"]) in expected_image_ids and int(row["category_id"]) in expected_category_ids and isinstance(row["bbox"], list) and len(row["bbox"]) == 4 and all(isinstance(value, (int, float)) and math.isfinite(value) for value in row["bbox"]) and row["bbox"][2] > 0 and row["bbox"][3] > 0 and isinstance(row["score"], (int, float)) and math.isfinite(row["score"]) and 0.0 <= row["score"] <= 1.0 for row in predictions ) check(f"{model_id}_{variant}_prediction_schema", record_shapes_valid, len(predictions)) check( f"{model_id}_{variant}_max_100_per_image", bool(counts) and max(counts.values()) <= 100, {"max": max(counts.values()) if counts else 0, "images_with_predictions": len(counts)}, ) check( f"{model_id}_{variant}_prediction_bundle_identity", sha256_file(predictions_path) == variant_summary["predictions_sha256"] and len(predictions) == variant_summary["prediction_count"], {"sha256": sha256_file(predictions_path), "count": len(predictions)}, ) independent = independent_coco_metrics(annotation_path, predictions) recomputed[model_id][variant] = independent differences = { name: abs(independent[name] - variant_summary["metrics"][name]) for name in METRIC_NAMES } check( f"{model_id}_{variant}_independent_cocoeval", max(differences.values()) <= 1e-12, {"max_abs_difference": max(differences.values()), "recomputed": independent}, ) prediction_integrity[model_id][variant] = { "prediction_count": len(predictions), "images_with_predictions": len(counts), "max_predictions_per_image": max(counts.values()) if counts else 0, "predictions_sha256": sha256_file(predictions_path), } fp32_ap = recomputed[model_id]["fp32"]["bbox_ap"] quantized_ap = recomputed[model_id]["public_int8"]["bbox_ap"] comparison = model_summary["pair_comparison"] check( f"{model_id}_pair_delta", abs(comparison["metric_deltas"]["bbox_ap"] - (quantized_ap - fp32_ap)) <= 1e-12, comparison, ) expected_retention = quantized_ap / fp32_ap * 100.0 check( f"{model_id}_retention", abs(comparison["bbox_ap_retention_percent"] - round(expected_retention, 6)) <= 1e-12, {"expected": expected_retention, "reported": comparison["bbox_ap_retention_percent"]}, ) with (result_dir / "quality_metrics.csv").open(newline="", encoding="utf-8") as handle: metric_csv = list(csv.DictReader(handle)) check("metric_csv_four_exact_variants", len(metric_csv) == 4, len(metric_csv)) with (result_dir / "pair_comparison.csv").open(newline="", encoding="utf-8") as handle: pair_csv = list(csv.DictReader(handle)) check("pair_csv_two_models", len(pair_csv) == 2, len(pair_csv)) with (result_dir / "per_category_ap.csv").open(newline="", encoding="utf-8") as handle: category_csv = list(csv.DictReader(handle)) check("per_category_2x80", len(category_csv) == 160, len(category_csv)) passed = all(row["pass"] for row in checks) validation = { "schema_version": "1.0", "stage": "OD06_OD07_FULL_COCO2017_INDEPENDENT_VALIDATION", "status": "PASS" if passed else "FAIL", "failure_code": None if passed else "FAIL_ANALYSIS", "result_dir": str(result_dir.relative_to(root)), "quality_summary_sha256": sha256_file(summary_path), "config_path": str(config_path.relative_to(root)), "config_sha256": sha256_file(config_path), "checks": checks, "recomputed_cocoeval": recomputed, "prediction_integrity": prediction_integrity, "policy": { "model_runtime_executed_by_validator": False, "model_conversion_or_modification": False, "latency_measurement": False, "independent_metric_recomputation": True, }, } atomic_json(output, validation) print(json.dumps({"status": validation["status"], "checks": len(checks), "passed": sum(row["pass"] for row in checks)}, sort_keys=True)) return 0 if passed else 1 if __name__ == "__main__": raise SystemExit(main())