| |
| """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()) |
|
|
