scripts/run_model_accuracy_summary.py

#!/usr/bin/env python3
"""Rebuild and verify the 21-model FP32/quantized accuracy CSV."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import shlex
import subprocess
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def atomic_text(path: Path, value: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, temporary = tempfile.mkstemp(prefix=f"{path.name}.", suffix=".tmp", dir=path.parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            handle.write(value)
        os.replace(temporary, path)
    except BaseException:
        Path(temporary).unlink(missing_ok=True)
        raise


def atomic_json(path: Path, value: dict[str, Any]) -> None:
    atomic_text(path, json.dumps(value, indent=2, sort_keys=True) + "\n")


def file_record(path: Path, root: Path) -> dict[str, Any]:
    return {
        "path": str(path.relative_to(root)),
        "bytes": path.stat().st_size,
        "sha256": sha256_file(path),
    }


def display_path(path: Path, root: Path) -> str:
    try:
        return str(path.relative_to(root))
    except ValueError:
        return str(path)


def run_stage(root: Path, run_dir: Path, name: str, command: list[str]) -> dict[str, Any]:
    atomic_text(run_dir / f"{name}.command.txt", shlex.join(command) + "\n")
    completed = subprocess.run(command, cwd=root, text=True, capture_output=True)
    atomic_text(run_dir / f"{name}.stdout.log", completed.stdout)
    atomic_text(run_dir / f"{name}.stderr.log", completed.stderr)
    atomic_text(run_dir / f"{name}.exit_code.txt", f"{completed.returncode}\n")
    return {
        "name": name,
        "status": "PASS" if completed.returncode == 0 else "FAIL",
        "exit_code": completed.returncode,
        "command": display_path(run_dir / f"{name}.command.txt", root),
        "stdout": display_path(run_dir / f"{name}.stdout.log", root),
        "stderr": display_path(run_dir / f"{name}.stderr.log", root),
        "exit_code_file": display_path(run_dir / f"{name}.exit_code.txt", root),
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repo-root", type=Path, required=True)
    parser.add_argument("--run-dir", type=Path, required=True)
    parser.add_argument("--python", type=Path, default=Path(sys.executable))
    args = parser.parse_args()

    root = args.repo_root.resolve()
    run_dir = args.run_dir if args.run_dir.is_absolute() else root / args.run_dir
    if run_dir.exists() and any(run_dir.iterdir()):
        raise SystemExit(f"refusing to overwrite non-empty run directory: {run_dir}")
    run_dir.mkdir(parents=True, exist_ok=True)

    python = str(args.python.absolute())
    csv_path = root / "reports/accuracy/model_accuracy.csv"
    validation_path = root / "reports/accuracy/model_accuracy.validation.json"
    stages = [
        run_stage(root, run_dir, "build", [python, "scripts/build_model_accuracy_summary.py", "--repo-root", str(root), "--output", str(csv_path)]),
        run_stage(root, run_dir, "validate", [python, "scripts/validate_model_accuracy_summary.py", "--repo-root", str(root), "--csv", str(csv_path), "--output", str(validation_path)]),
        run_stage(root, run_dir, "tests", [python, "-m", "pytest", "-q", "tests/test_model_accuracy_summary.py"]),
        run_stage(root, run_dir, "compile", [python, "-m", "py_compile", "scripts/build_model_accuracy_summary.py", "scripts/validate_model_accuracy_summary.py", "scripts/run_model_accuracy_summary.py", "tests/test_model_accuracy_summary.py"]),
    ]
    precheck_status = "PASS" if all(item["status"] == "PASS" for item in stages) else "FAIL"
    execution_path = run_dir / "execution_manifest.json"
    atomic_json(execution_path, {
        "schema_version": "1.0",
        "stage": "MODEL_ACCURACY_CSV_REBUILD",
        "status": precheck_status,
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "stages": stages,
    })

    package_paths = [
        csv_path,
        validation_path,
        root / "model_registry.csv",
        root / "reports/accuracy/model_accuracy_values.json",
        root / "research/evidence/vision/vc11_compute_graph_equivalence.json",
        root / "scripts/build_model_accuracy_summary.py",
        root / "scripts/validate_model_accuracy_summary.py",
        root / "scripts/run_model_accuracy_summary.py",
        root / "tests/test_model_accuracy_summary.py",
    ]
    missing = [str(path) for path in package_paths if not path.is_file()]
    if missing:
        raise FileNotFoundError(f"required accuracy package files missing: {missing}")

    manifest_path = root / "reports/accuracy/model_accuracy.artifact_manifest.json"
    checksum_path = root / "reports/accuracy/model_accuracy.artifacts.sha256"
    records = [file_record(path, root) for path in sorted(set(package_paths), key=lambda item: str(item.relative_to(root)))]
    atomic_json(manifest_path, {
        "schema_version": "1.0",
        "stage": "MODEL_ACCURACY_CSV_MANIFEST",
        "status": precheck_status,
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "file_count": len(records),
        "files": records,
    })
    checksum_records = records + [file_record(manifest_path, root)]
    atomic_text(checksum_path, "".join(f"{item['sha256']}  {item['path']}\n" for item in checksum_records))
    checksum_stage = run_stage(root, run_dir, "checksum", ["sha256sum", "-c", str(checksum_path.relative_to(root))])
    final_status = "PASS" if precheck_status == checksum_stage["status"] == "PASS" else "FAIL"
    atomic_json(run_dir / "final_result.json", {
        "schema_version": "1.0",
        "status": final_status,
        "csv": file_record(csv_path, root),
        "validation": file_record(validation_path, root),
        "manifest": file_record(manifest_path, root),
        "checksum": file_record(checksum_path, root),
    })
    print(f"{final_status}: {csv_path.relative_to(root)}")
    return 0 if final_status == "PASS" else 1


if __name__ == "__main__":
    raise SystemExit(main())