#!/usr/bin/env python3 """Generate and validate the public 42-graph ONNX Dialect image set.""" from __future__ import annotations import argparse import hashlib import json import platform import shlex import subprocess import sys 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="seconds").replace("+00:00", "Z") def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for block in iter(lambda: stream.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def record(path: Path, root: Path) -> dict[str, Any]: return { "path": str(path.relative_to(root)), "bytes": path.stat().st_size, "sha256": sha256(path), } def write_text(path: Path, text: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(text, encoding="utf-8") temporary.replace(path) def write_json(path: Path, value: Any) -> None: write_text(path, json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n") def run_stage(name: str, command: list[str], root: Path, run_dir: Path) -> dict[str, Any]: command_path = run_dir / f"{name}.command.txt" stdout_path = run_dir / f"{name}.stdout.log" stderr_path = run_dir / f"{name}.stderr.log" exit_path = run_dir / f"{name}.exit_code.txt" write_text(command_path, shlex.join(command) + "\n") with stdout_path.open("wb") as stdout, stderr_path.open("wb") as stderr: try: completed = subprocess.run(command, cwd=root, stdout=stdout, stderr=stderr, check=False) exit_code = completed.returncode except OSError as error: exit_code = 127 stderr.write((f"{type(error).__name__}: {error}\n").encode()) write_text(exit_path, f"{exit_code}\n") return { "stage": name, "status": "PASS" if exit_code == 0 else "FAIL", "exit_code": exit_code, "command": shlex.join(command), "logs": { "command": record(command_path, root), "stdout": record(stdout_path, root), "stderr": record(stderr_path, root), "exit_code": record(exit_path, root), }, } def resolve(root: Path, path: Path) -> Path: resolved = path.resolve() if path.is_absolute() else (root / path).resolve() try: resolved.relative_to(root) except ValueError as error: raise SystemExit(f"path outside repository root: {resolved}") from error return resolved 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) parser.add_argument("--report-dir", type=Path, default=Path("reports/graphs/mlir")) parser.add_argument("--coverage-matrix", type=Path, default=Path("reports/conversion/ir_stage_coverage.csv")) parser.add_argument("--render-workers", type=int, default=4) args = parser.parse_args() root = args.repo_root.resolve() run_dir = resolve(root, args.run_dir) report_dir = resolve(root, args.report_dir) matrix = resolve(root, args.coverage_matrix) execution_manifest = run_dir / "execution_manifest.json" if execution_manifest.exists(): raise SystemExit(f"completed run directory already exists: {run_dir}") run_dir.mkdir(parents=True, exist_ok=True) python = root / ".venv/bin/python" builder = root / "scripts/build_mlir_ir_graphs.py" common = root / "scripts/mlir_graph_common.py" validator = root / "scripts/validate_mlir_ir_graphs.py" runner = root / "scripts/run_mlir_graph_checks.py" test_builder = root / "tests/test_mlir_graph_builder_resilience.py" test_graph = root / "tests/test_mlir_ir_graphs.py" test_validator = root / "tests/test_validate_mlir_ir_graphs.py" validation = run_dir / "validation.json" checkpoint = run_dir / "checkpoint.json" render_logs = run_dir / "renders" started_at = utc_now() stages = [ run_stage( "build", [ str(python), "scripts/build_mlir_ir_graphs.py", "--repo-root", str(root), "--coverage-matrix", str(matrix), "--output-dir", str(report_dir), "--render-log-dir", str(render_logs), "--checkpoint", str(checkpoint), "--render-workers", str(max(1, min(args.render_workers, 4))), "--primary-only", "--images-only", ], root, run_dir, ), run_stage( "validate", [ str(python), "scripts/validate_mlir_ir_graphs.py", "--repo-root", str(root), "--report-dir", str(report_dir), "--coverage-matrix", str(matrix), "--output", str(validation), ], root, run_dir, ), run_stage( "targeted_tests", [ str(python), "-m", "pytest", "-q", str(test_graph.relative_to(root)), str(test_builder.relative_to(root)), str(test_validator.relative_to(root)), "-k", "not runner and not generated_mlir_graphs_package_contract", ], root, run_dir, ), run_stage( "compile", [ str(python), "-m", "py_compile", str(common), str(builder), str(validator), str(runner), str(test_graph), str(test_builder), str(test_validator), ], root, run_dir, ), ] svg_files = sorted(report_dir.glob("graphs/*/*/onnx/execution_dependency_graph.svg")) png_files = sorted(report_dir.glob("graphs/*/*/onnx/execution_dependency_graph.png")) status = "PASS" if all(stage["status"] == "PASS" for stage in stages) else "FAIL" manifest = { "schema_version": "1.0", "stage": "ONNX_DIALECT_PRIMARY_GRAPH_WORKFLOW", "status": status, "started_at": started_at, "finished_at": utc_now(), "tool_versions": {"python": platform.python_version()}, "inputs": [record(path, root) for path in (matrix, builder, common, validator, runner, test_graph, test_builder, test_validator)], "stages": stages, "outputs": { "svg": [record(path, root) for path in svg_files], "png": [record(path, root) for path in png_files], "validation": record(validation, root) if validation.is_file() else None, }, "counts": {"svg": len(svg_files), "png": len(png_files)}, "policy": { "primary_graph_stage": "ONNX_DIALECT", "supplemental_lower_graphs_included": False, "internal_graph_package_published": False, "model_runtime_run": False, "mlir_toolchain_run": False, }, } write_json(execution_manifest, manifest) print(json.dumps({"status": status, "counts": manifest["counts"], "execution_manifest": str(execution_manifest)}, ensure_ascii=False, sort_keys=True)) return 0 if status == "PASS" else 1 if __name__ == "__main__": raise SystemExit(main())