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