| |
| """Validate the 42 published ONNX Dialect execution-dependency graphs.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import csv |
| import json |
| import sys |
| import xml.etree.ElementTree as ET |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Any |
|
|
| REPO_ROOT = Path(__file__).resolve().parents[1] |
| if str(REPO_ROOT) not in sys.path: |
| sys.path.insert(0, str(REPO_ROOT)) |
|
|
| from scripts.mlir_graph_common import ( |
| VARIANTS, |
| atomic_json, |
| resolve_coverage_path, |
| sha256, |
| ) |
|
|
|
|
| def utc_now() -> str: |
| return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") |
|
|
|
|
| def normalize_variant(value: str) -> str: |
| return "public_quantized" if value in {"quantized", "public_quantized"} else value |
|
|
|
|
| def read_rows(path: Path) -> list[dict[str, str]]: |
| with path.open(newline="", encoding="utf-8") as stream: |
| return list(csv.DictReader(stream)) |
|
|
|
|
| def validate(args: argparse.Namespace) -> tuple[dict[str, Any], int]: |
| root = args.repo_root.resolve() |
| report_dir = (root / args.report_dir).resolve() if not args.report_dir.is_absolute() else args.report_dir.resolve() |
| matrix = (root / args.coverage_matrix).resolve() if not args.coverage_matrix.is_absolute() else args.coverage_matrix.resolve() |
| output = (root / args.output).resolve() if not args.output.is_absolute() else args.output.resolve() |
| for path in (report_dir, matrix, output.parent): |
| try: |
| path.relative_to(root) |
| except ValueError as error: |
| raise SystemExit(f"path outside repository root: {path}") from error |
|
|
| checks: list[dict[str, Any]] = [] |
|
|
| def check(name: str, passed: bool, detail: Any = None) -> None: |
| checks.append({"name": name, "status": "PASS" if passed else "FAIL", "detail": detail}) |
|
|
| rows = read_rows(matrix) |
| keyed: dict[tuple[str, str], dict[str, str]] = {} |
| for row in rows: |
| variant = normalize_variant(row.get("variant", "")) |
| key = (row.get("model_id", ""), variant) |
| check("unique_matrix_row", key not in keyed, {"key": key}) |
| keyed[key] = row |
|
|
| model_ids = sorted({model_id for model_id, _ in keyed}) |
| check("matrix_variant_count", len(keyed) == 42, len(keyed)) |
| check("matrix_model_count", len(model_ids) == 21, len(model_ids)) |
| check( |
| "matrix_pair_completeness", |
| all((model_id, variant) in keyed for model_id in model_ids for variant in VARIANTS), |
| ) |
|
|
| expected_svg: set[Path] = set() |
| expected_png: set[Path] = set() |
| for model_id in model_ids: |
| for variant in VARIANTS: |
| row = keyed.get((model_id, variant)) |
| if row is None: |
| continue |
| context = f"{model_id}:{variant}" |
| check("onnx_dialect_status", row.get("onnx_status") == "PASS", {"graph": context, "status": row.get("onnx_status")}) |
| try: |
| source = resolve_coverage_path(row.get("onnx_artifact", ""), root) |
| source_ok = source.is_file() and sha256(source) == row.get("onnx_sha256") |
| except (OSError, ValueError) as error: |
| source_ok = False |
| source = Path(row.get("onnx_artifact", "")) |
| check("onnx_source_resolution", False, {"graph": context, "error": str(error)}) |
| else: |
| check("onnx_source_checksum", source_ok, {"graph": context, "source": str(source)}) |
|
|
| graph_dir = report_dir / "graphs" / model_id / variant / "onnx" |
| svg = graph_dir / "execution_dependency_graph.svg" |
| png = graph_dir / "execution_dependency_graph.png" |
| expected_svg.add(svg) |
| expected_png.add(png) |
| try: |
| svg_text = svg.read_text(encoding="utf-8") |
| ET.fromstring(svg_text) |
| svg_ok = "STATIC_MLIR_PROGRAM_ORDER" in svg_text and f"{model_id}:{variant}:ONNX" in svg_text |
| except (OSError, ET.ParseError) as error: |
| svg_ok = False |
| check("svg_read", False, {"graph": context, "error": str(error)}) |
| else: |
| check("svg_semantics", svg_ok, context) |
| try: |
| with png.open("rb") as stream: |
| png_ok = stream.read(8) == b"\x89PNG\r\n\x1a\n" and png.stat().st_size > 8 |
| except OSError as error: |
| png_ok = False |
| check("png_read", False, {"graph": context, "error": str(error)}) |
| else: |
| check("png_signature", png_ok, context) |
|
|
| actual_svg = set(report_dir.glob("graphs/*/*/*/execution_dependency_graph.svg")) |
| actual_png = set(report_dir.glob("graphs/*/*/*/execution_dependency_graph.png")) |
| check("exact_svg_set", actual_svg == expected_svg, {"expected": 42, "actual": len(actual_svg)}) |
| check("exact_png_set", actual_png == expected_png, {"expected": 42, "actual": len(actual_png)}) |
| check("no_affine_graphs", not any("affine_scf_memref" in path.parts for path in report_dir.rglob("*"))) |
| check("no_internal_graph_records", not any(report_dir.rglob("graph_record*.json"))) |
|
|
| failed = [row for row in checks if row["status"] != "PASS"] |
| result = { |
| "schema_version": "1.0", |
| "stage": "ONNX_DIALECT_PRIMARY_GRAPH_VALIDATION", |
| "status": "PASS" if not failed else "FAIL", |
| "generated_at": utc_now(), |
| "counts": { |
| "models": len(model_ids), |
| "variants": len(keyed), |
| "svg": len(actual_svg), |
| "png": len(actual_png), |
| "checks": len(checks), |
| "failed": len(failed), |
| }, |
| "checks": checks, |
| "policy": { |
| "primary_graph_stage": "ONNX_DIALECT", |
| "order_semantics": "STATIC_MLIR_PROGRAM_ORDER", |
| "supplemental_lower_graphs_included": False, |
| "model_runtime_run": False, |
| "mlir_toolchain_run": False, |
| }, |
| } |
| output.parent.mkdir(parents=True, exist_ok=True) |
| atomic_json(output, result) |
| return result, 0 if result["status"] == "PASS" else 1 |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--repo-root", type=Path, default=REPO_ROOT) |
| 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("--output", type=Path, required=True) |
| args = parser.parse_args() |
| result, exit_code = validate(args) |
| print(json.dumps({"status": result["status"], "counts": result["counts"]}, ensure_ascii=False, sort_keys=True)) |
| return exit_code |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
