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