scripts/finalize_od06_od07_coco_evaluation.py

#!/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())