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