#!/usr/bin/env python3 """Validate, test, checksum, and manifest a completed OD06/OD07 COCO run.""" from __future__ import annotations import argparse import hashlib import json import os import platform import shlex import subprocess import tempfile from pathlib import Path from typing import Any 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 atomic_text(path: Path, value: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle: handle.write(value) temporary = Path(handle.name) os.replace(temporary, path) def run_stage(name: str, command: list[str], root: Path, run_dir: Path) -> dict[str, Any]: command_path = run_dir / f"{name}.command.txt" stdout_path = run_dir / f"{name}.stdout.log" stderr_path = run_dir / f"{name}.stderr.log" exit_path = run_dir / f"{name}.exit_code.txt" atomic_text(command_path, shlex.join(command) + "\n") completed = subprocess.run(command, cwd=root, text=True, capture_output=True, check=False) atomic_text(stdout_path, completed.stdout) atomic_text(stderr_path, completed.stderr) atomic_text(exit_path, f"{completed.returncode}\n") return { "name": name, "status": "PASS" if completed.returncode == 0 else "FAIL", "exit_code": completed.returncode, "command": shlex.join(command), "command_log": str(command_path.relative_to(root)), "stdout_log": str(stdout_path.relative_to(root)), "stderr_log": str(stderr_path.relative_to(root)), "exit_code_log": str(exit_path.relative_to(root)), "stdout_sha256": sha256_file(stdout_path), "stderr_sha256": sha256_file(stderr_path), } def collect_manifest_files(root: Path, config: dict[str, Any], result_dir: Path, run_dir: Path) -> list[Path]: excluded_names = {"artifact_manifest.json", "artifacts.sha256"} files = [path for path in result_dir.rglob("*") if path.is_file() and path.name not in excluded_names] files.extend( path for path in run_dir.parent.rglob("*") if path.is_file() and path.name not in { "checksum.stdout.log", "checksum.stderr.log", "checksum.exit_code.txt", "checksum.command.txt", "final_result.json", } ) files.extend( root / path for path in ( "scripts/stages/evaluate_od06_od07_coco.py", "scripts/stages/resume_od06_od07_coco_metrics.py", "scripts/validate_od06_od07_coco.py", "scripts/finalize_od06_od07_coco_evaluation.py", "tests/test_od06_od07_coco_evaluator.py", "tests/test_od06_od07_coco_results.py", "environment/quality/od_coco/requirements.in", "environment/quality/od_coco/requirements.lock", "research/evidence/detection/od06_od07_coco2017/sources/mediapipe_LICENSE_cb0902f", "research/evidence/detection/od06_od07_coco2017/sources/cocoapi_license_8c9bcc3.txt", "research/evidence/detection/od06_od07_coco2017/sources/mediapipe_object_detector_guide_20260812.html", "research/evidence/detection/od06_od07_coco2017/sources/coco_official_site_20260812.html", "research/evidence/detection/od06_od07_coco2017/method_audit.md", ) ) files.append(root / "configs/evaluation/object_detection/OD06_OD07_coco2017_quality_eval.json") files.extend(root / source["path"] for source in config["authoritative_sources"]) files.extend( [ root / config["dataset"]["annotation_path"], root / config["dataset"]["image_archive"]["path"], root / config["dataset"]["annotation_archive"]["path"], root / config["dataset"]["license"]["terms_path"], root / config["models"]["OD06"]["published_context"]["source_path"], ] ) for model in config["models"].values(): files.extend(root / variant["path"] for variant in model["variants"].values()) unique = sorted(set(path.resolve() for path in files), key=lambda path: str(path.relative_to(root))) missing = [str(path) for path in unique if not path.is_file()] if missing: raise FileNotFoundError(f"manifest inputs missing: {missing}") return unique 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("--run-dir", required=True, type=Path) args = parser.parse_args() root = args.repo_root.resolve() config_path = args.config.resolve() result_dir = args.result_dir.resolve() run_dir = args.run_dir.resolve() config = json.loads(config_path.read_text(encoding="utf-8")) summary_path = result_dir / "quality_summary.json" if not summary_path.is_file(): raise FileNotFoundError("quality_summary.json is required before finalization") if (result_dir / "artifact_manifest.json").exists() or (result_dir / "artifacts.sha256").exists(): raise FileExistsError("refusing to overwrite existing artifact manifest") python = root / config["runtime"]["python"] stages: list[dict[str, Any]] = [] stages.append( run_stage( "validate", [ str(python), "scripts/validate_od06_od07_coco.py", "--repo-root", str(root), "--config", str(config_path), "--result-dir", str(result_dir), "--output", str(result_dir / "validation.json"), ], root, run_dir, ) ) stages.append( run_stage( "targeted_tests", [ str(python), "-m", "pytest", "-q", "tests/test_od06_od07_coco_evaluator.py", "tests/test_od06_od07_coco_results.py", ], root, run_dir, ) ) stages.append( run_stage( "py_compile", [ str(python), "-m", "py_compile", "scripts/stages/evaluate_od06_od07_coco.py", "scripts/stages/resume_od06_od07_coco_metrics.py", "scripts/validate_od06_od07_coco.py", "scripts/finalize_od06_od07_coco_evaluation.py", ], root, run_dir, ) ) summary = json.loads(summary_path.read_text(encoding="utf-8")) validation = json.loads((result_dir / "validation.json").read_text(encoding="utf-8")) if (result_dir / "validation.json").is_file() else {} preliminary_pass = ( summary.get("status") == "PASS" and validation.get("status") == "PASS" and all(stage["status"] == "PASS" for stage in stages) ) inference_attempt = run_dir.parent / "attempt_003" inference_exit_path = inference_attempt / "evaluate.exit_code.txt" inference_exit_code = int(inference_exit_path.read_text().strip()) if inference_exit_path.is_file() else None metrics_exit_path = run_dir / "metrics_only_02.exit_code.txt" metrics_exit_code = int(metrics_exit_path.read_text().strip()) if metrics_exit_path.is_file() else None full_inference_complete = all( summary["models"][model_id]["variants"][variant]["worker_summary"]["image_count"] == 5000 for model_id in ("OD06", "OD07") for variant in ("fp32", "public_int8") ) execution_manifest = { "schema_version": "1.0", "evaluation_id": config["evaluation_id"], "status": "PASS" if preliminary_pass and metrics_exit_code == 0 and full_inference_complete else "FAIL", "inference": { "process_exit_code": inference_exit_code, "process_exit_note": "The process failed only after all four inference bundles were atomically saved, during the first COCOeval compatibility attempt.", "command_log": str((inference_attempt / "evaluate.command.txt").relative_to(root)), "stdout_log": str((inference_attempt / "evaluate.stdout.log").relative_to(root)), "stderr_log": str((inference_attempt / "evaluate.stderr.log").relative_to(root)), "code_freeze": str((inference_attempt / "code_freeze.json").relative_to(root)), "full_coco_val2017_complete": full_inference_complete, "prediction_bundle_count": 4, }, "metrics_resume": { "exit_code": metrics_exit_code, "command_log": str((run_dir / "metrics_only_02.command.txt").relative_to(root)), "stdout_log": str((run_dir / "metrics_only_02.stdout.log").relative_to(root)), "stderr_log": str((run_dir / "metrics_only_02.stderr.log").relative_to(root)), "code_freeze": str((run_dir / "metrics_code_freeze.json").relative_to(root)), "model_runtime_executed": False, }, "post_evaluation_stages": stages, "result_status": summary.get("status"), "validation_status": validation.get("status"), "environment": {"python": platform.python_version(), "platform": platform.platform()}, "policy": { "latency_measurement": False, "training_calibration_or_quantization": False, "model_conversion_or_modification": False, "mlir_allocator_or_codegen": False, "published_family_values_used_as_exact_threshold": False, }, } atomic_json(run_dir / "execution_manifest.json", execution_manifest) if execution_manifest["status"] != "PASS": print(json.dumps({"status": "FAIL", "stages": stages}, sort_keys=True)) return 1 manifest_files = collect_manifest_files(root, config, result_dir, run_dir) records = [ { "path": str(path.relative_to(root)), "bytes": path.stat().st_size, "sha256": sha256_file(path), "role": "input" if path.is_relative_to(root / "research/downloads") else "evidence_or_output", } for path in manifest_files ] artifact_manifest = { "schema_version": "1.0", "evaluation_id": config["evaluation_id"], "status": "PASS", "models": ["OD06", "OD07"], "variants": ["fp32", "public_int8"], "dataset": "COCO 2017 val2017, full 5,000 images", "validation_status": validation["status"], "files": records, } manifest_path = result_dir / "artifact_manifest.json" atomic_json(manifest_path, artifact_manifest) checksum_records = records + [ { "path": str(manifest_path.relative_to(root)), "sha256": sha256_file(manifest_path), } ] checksum_path = result_dir / "artifacts.sha256" atomic_text(checksum_path, "".join(f"{row['sha256']} {row['path']}\n" for row in checksum_records)) checksum_stage = run_stage("checksum", ["sha256sum", "-c", str(checksum_path)], root, run_dir) final_status = "PASS" if checksum_stage["status"] == "PASS" else "FAIL" final_result = { "status": final_status, "result_status": summary["status"], "validation_status": validation["status"], "artifact_manifest": str(manifest_path.relative_to(root)), "artifact_manifest_sha256": sha256_file(manifest_path), "artifacts_sha256": str(checksum_path.relative_to(root)), "artifacts_sha256_sha256": sha256_file(checksum_path), "checksum_stage": checksum_stage, } atomic_json(run_dir / "final_result.json", final_result) print(json.dumps({"status": final_status, "validation": validation["status"], "files": len(records)}, sort_keys=True)) return 0 if final_status == "PASS" else 1 if __name__ == "__main__": raise SystemExit(main())