scripts/run_sg08_sp08_diagnostics.py

#!/usr/bin/env python3
"""Run and independently validate the SG08/SP08 diagnostic evaluations."""

from __future__ import annotations

import argparse
import hashlib
import json
import shlex
import subprocess
import sys
from pathlib import Path
from typing import Any


SPECS = {
    "SG08": {
        "config": "configs/evaluation/semantic_segmentation/SG08_camvid_cross_dataset_diagnostic.json",
        "evaluator": "scripts/stages/evaluate_sg08_camvid_diagnostic.py",
    },
    "SP08": {
        "config": "configs/evaluation/speech_kws/SP08_speech_commands_yes_no_diagnostic.json",
        "evaluator": "scripts/stages/evaluate_sp08_speech_commands_diagnostic.py",
    },
}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repo-root", type=Path, default=Path.cwd())
    parser.add_argument("--model", choices=["SG08", "SP08", "all"], default="all")
    return parser.parse_args()


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)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(value, encoding="utf-8")
    temporary.replace(path)


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


def run_logged(command: list[str], root: Path, log_dir: Path, stem: str) -> int:
    log_dir.mkdir(parents=True, exist_ok=True)
    atomic_text(log_dir / f"{stem}.command.txt", shlex.join(command) + "\n")
    completed = subprocess.run(command, cwd=root, capture_output=True, text=True, check=False)
    atomic_text(log_dir / f"{stem}.stdout.log", completed.stdout)
    atomic_text(log_dir / f"{stem}.stderr.log", completed.stderr)
    atomic_text(log_dir / f"{stem}.exit_code.txt", f"{completed.returncode}\n")
    return completed.returncode


def run_model(root: Path, model_id: str) -> dict[str, Any]:
    spec = SPECS[model_id]
    config_path = root / spec["config"]
    config = json.loads(config_path.read_text(encoding="utf-8"))
    result_dir = root / config["output_dir"]
    log_dir = result_dir.parent / "logs"
    python_path = root / config["runtime"]["python"]
    evaluator_command = [
        str(python_path),
        str(root / spec["evaluator"]),
        "--project-root", str(root),
        "--config", str(config_path),
        "--output-dir", str(result_dir),
    ]
    evaluate_exit = run_logged(evaluator_command, root, log_dir, "evaluate")
    if evaluate_exit != 0:
        return {"model_id": model_id, "status": "FAIL", "stage": "evaluate", "exit_code": evaluate_exit}

    validation_path = result_dir / "validation_report.json"
    validator_command = [
        str(root / ".venv/bin/python"),
        str(root / "scripts/stages/validate_sg08_sp08_diagnostic.py"),
        "--project-root", str(root),
        "--config", str(config_path),
        "--result-dir", str(result_dir),
        "--execution-log-dir", str(log_dir),
        "--output", str(validation_path),
    ]
    validate_exit = run_logged(validator_command, root, log_dir, "validate")
    manifest_path = result_dir / "diagnostic_artifact_manifest.json"
    package_paths = [
        config_path,
        root / spec["evaluator"],
        root / "scripts/stages/validate_sg08_sp08_diagnostic.py",
        *[path for path in sorted(result_dir.glob("*")) if path.is_file() and path != manifest_path],
        *sorted(log_dir.glob("*")),
    ]
    atomic_json(
        manifest_path,
        {
            "schema_version": "1.0",
            "model_id": model_id,
            "evaluation_kind": config["evaluation_kind"],
            "canonical_q1_affected": False,
            "status": "PASS" if validate_exit == 0 else "FAIL",
            "artifacts": [
                {
                    "path": str(path.relative_to(root)),
                    "bytes": path.stat().st_size,
                    "sha256": sha256_file(path),
                }
                for path in package_paths
            ],
        },
    )
    return {
        "model_id": model_id,
        "status": "PASS" if validate_exit == 0 else "FAIL",
        "stage": "complete" if validate_exit == 0 else "validate",
        "exit_code": validate_exit,
        "result_dir": str(result_dir.relative_to(root)),
        "manifest": str(manifest_path.relative_to(root)),
    }


def main() -> int:
    args = parse_args()
    root = args.repo_root.resolve()
    model_ids = list(SPECS) if args.model == "all" else [args.model]
    results = [run_model(root, model_id) for model_id in model_ids]
    print(json.dumps({"results": results}, indent=2, sort_keys=True))
    return 0 if all(result["status"] == "PASS" for result in results) else 1


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