scripts/run_netron_checks.py

#!/usr/bin/env python3
"""Rebuild, resume-verify, independently validate, and test Netron exports."""

from __future__ import annotations

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


REPO_ROOT = Path(__file__).resolve().parents[1]


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")


def sha256(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 relative(path: Path, root: Path) -> str:
    resolved = path.resolve()
    try:
        return str(resolved.relative_to(root.resolve()))
    except ValueError:
        return str(resolved)


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, indent=2, sort_keys=True, ensure_ascii=False)
        handle.write("\n")
        temporary = Path(handle.name)
    os.replace(temporary, path)


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


def run_stage(
    name: str,
    command: list[str],
    root: Path,
    run_dir: Path,
    failure_code: str,
    extra_env: dict[str, str] | None = None,
) -> dict[str, Any]:
    stdout_path = run_dir / f"{name}.stdout.log"
    stderr_path = run_dir / f"{name}.stderr.log"
    environment = os.environ.copy()
    if extra_env:
        environment.update(extra_env)
    started_at = utc_now()
    with stdout_path.open("wb") as stdout_handle, stderr_path.open("wb") as stderr_handle:
        completed = subprocess.run(
            command,
            cwd=root,
            env=environment,
            stdout=stdout_handle,
            stderr=stderr_handle,
            check=False,
        )
    display_command = shlex.join(command)
    if extra_env:
        display_command = " ".join(f"{key}={shlex.quote(value)}" for key, value in extra_env.items()) + " " + display_command
    return {
        "stage": name,
        "status": "PASS" if completed.returncode == 0 else "FAIL",
        "failure_code": None if completed.returncode == 0 else failure_code,
        "started_at": started_at,
        "finished_at": utc_now(),
        "command_argv": command,
        "command": display_command,
        "environment_overrides": extra_env or {},
        "working_directory": str(root),
        "exit_code": completed.returncode,
        "stdout_log": file_record(stdout_path, root),
        "stderr_log": file_record(stderr_path, root),
        "patch": "T80 visualization/report automation only; existing model artifacts are immutable inputs",
        "converter_run": False,
        "model_weight_architecture_modified": False,
        "allocator_work_performed": False,
        "prohibited_operations_performed": [],
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repo-root", type=Path, default=REPO_ROOT)
    parser.add_argument("--run-dir", type=Path, required=True)
    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
    run_dir = run_dir.resolve()
    run_dir.mkdir(parents=True, exist_ok=True)
    project_python = str(root / ".venv/bin/python")
    netron_python = str(root / "environment/visualization/netron/.venv/bin/python")
    browser_path = str(root / "environment/visualization/netron/browsers")
    for executable in (project_python, netron_python):
        if not Path(executable).is_file():
            raise SystemExit(f"missing interpreter: {executable}")
    browser_env = {"PLAYWRIGHT_BROWSERS_PATH": browser_path}
    commands = [
        ("record_toolchain", [netron_python, "scripts/record_netron_toolchain.py"], "FAIL_ENVIRONMENT", browser_env),
        ("build_input_inventory", [project_python, "scripts/build_netron_input_inventory.py"], "FAIL_ANALYSIS", None),
        (
            "capture_resume",
            [
                netron_python,
                "scripts/capture_netron_graphs.py",
                "--run-dir",
                str(run_dir / "capture_resume"),
                "--scope",
                "full",
            ],
            "FAIL_ANALYSIS",
            browser_env,
        ),
        ("build_report", [project_python, "scripts/build_netron_report.py"], "FAIL_ANALYSIS", None),
        ("validate", [project_python, "scripts/validate_netron_exports.py"], "FAIL_ANALYSIS", None),
        ("build_manifest", [project_python, "scripts/build_netron_artifact_manifest.py"], "FAIL_ANALYSIS", None),
        ("targeted_tests", [project_python, "-m", "pytest", "-q", "tests/test_netron_exports.py"], "FAIL_ANALYSIS", None),
        ("full_tests", [project_python, "-m", "pytest", "-q"], "FAIL_ANALYSIS", None),
        (
            "compileall",
            [
                project_python,
                "-m",
                "py_compile",
                "scripts/netron_capture_common.py",
                "scripts/build_netron_input_inventory.py",
                "scripts/capture_netron_graphs.py",
                "scripts/build_netron_report.py",
                "scripts/validate_netron_exports.py",
                "scripts/record_netron_toolchain.py",
                "scripts/build_netron_artifact_manifest.py",
                "scripts/run_netron_checks.py",
                "tests/test_netron_exports.py",
            ],
            "FAIL_ANALYSIS",
            None,
        ),
    ]
    stages = [
        run_stage(name, command, root, run_dir, failure_code, environment)
        for name, command, failure_code, environment in commands
    ]
    status = "PASS" if all(stage["status"] == "PASS" for stage in stages) else "FAIL"
    report_dir = root / "reports/graphs/netron"
    output_paths = [
        report_dir / "README.md",
        report_dir / "netron_input_inventory.csv",
        report_dir / "netron_input_summary.json",
        report_dir / "netron_capture_inventory.csv",
        report_dir / "netron_capture_summary.json",
        report_dir / "netron_model_matrix.csv",
        report_dir / "netron_capture_report.md",
        report_dir / "netron_gallery.html",
        report_dir / "validation.json",
        report_dir / "artifact_manifest.json",
        report_dir / "artifacts.sha256",
    ]
    input_paths = [
        root / "model_registry.csv",
        root / "reports/conversion/pipeline_status.csv",
        root / "environment/visualization/netron/requirements.in",
        root / "environment/visualization/netron/requirements.lock",
        *[root / f"scripts/{name}" for name in (
            "netron_capture_common.py",
            "build_netron_input_inventory.py",
            "capture_netron_graphs.py",
            "build_netron_report.py",
            "validate_netron_exports.py",
            "record_netron_toolchain.py",
            "build_netron_artifact_manifest.py",
            "run_netron_checks.py",
        )],
        root / "tests/test_netron_exports.py",
    ]
    capture_summary = json.loads((report_dir / "netron_capture_summary.json").read_text(encoding="utf-8"))
    validation = json.loads((report_dir / "validation.json").read_text(encoding="utf-8"))
    tool_versions = json.loads((root / "environment/visualization/netron/tool_versions.json").read_text(encoding="utf-8"))
    manifest = {
        "schema_version": "1.0",
        "stage": "T80_NETRON_EXPORT_FINAL",
        "status": status,
        "failure_code": None if status == "PASS" else next(
            stage["failure_code"] for stage in stages if stage["status"] == "FAIL"
        ),
        "command_argv": [sys.executable, *sys.argv],
        "command": shlex.join([sys.executable, *sys.argv]),
        "working_directory": str(root),
        "started_at": stages[0]["started_at"],
        "finished_at": utc_now(),
        "tool_versions": {"runner_python": platform.python_version(), **tool_versions},
        "inputs": [file_record(path, root) for path in input_paths if path.is_file()],
        "stages": stages,
        "outputs": [file_record(path, root) for path in output_paths if path.is_file()],
        "result": {
            "models": capture_summary["counts"]["models"],
            "theoretical_slots": capture_summary["counts"]["theoretical_slots"],
            "netron_exports_pass": capture_summary["counts"]["netron_exports_pass"],
            "onnx_exports_pass": capture_summary["counts"]["onnx_exports_pass"],
            "not_available": capture_summary["counts"]["not_available"],
            "canonical_pair_exports_pass": capture_summary["counts"]["canonical_pair_exports_pass"],
            "validation_checks_passed": validation["checks_passed"],
            "validation_checks_failed": validation["checks_failed"],
            "converter_run": False,
            "model_weight_architecture_modified": False,
            "allocator_work_performed": False,
            "prohibited_operations_performed": [],
        },
    }
    atomic_json(run_dir / "execution_manifest.json", manifest)
    print(
        json.dumps(
            {
                "status": status,
                "stages_pass": sum(stage["status"] == "PASS" for stage in stages),
                "stages_fail": sum(stage["status"] != "PASS" for stage in stages),
                "netron_exports_pass": manifest["result"]["netron_exports_pass"],
                "validation_checks_passed": manifest["result"]["validation_checks_passed"],
                "execution_manifest": relative(run_dir / "execution_manifest.json", root),
            },
            sort_keys=True,
        )
    )
    return 0 if status == "PASS" else 1


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