File size: 7,180 Bytes
ed3aeeb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | #!/usr/bin/env python3
"""Validate the compact MLIR graph package without the removed source MLIR.
The full source-reparse result is retained as
``source_validation_at_generation.json``. This validator checks that record,
then independently verifies the current package identities, graph artifacts,
CSV counts, and manifest hashes after repository-only naming changes.
"""
from __future__ import annotations
import argparse
import csv
import hashlib
import json
import os
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for block in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def rows(path: Path) -> list[dict[str, str]]:
with path.open(encoding="utf-8", newline="") as handle:
return list(csv.DictReader(handle))
def csv_data_rows(path: Path) -> int:
with path.open(encoding="utf-8", newline="") as handle:
reader = csv.reader(handle)
next(reader, None)
return sum(1 for _ in reader)
def atomic_json(path: Path, value: dict[str, 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 main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--repo-root", type=Path, default=Path("."))
parser.add_argument("--report-dir", type=Path, default=Path("reports/graphs/mlir"))
parser.add_argument("--output", type=Path)
args = parser.parse_args()
root = args.repo_root.resolve()
report_dir = args.report_dir if args.report_dir.is_absolute() else root / args.report_dir
output = args.output or report_dir / "validation.json"
if not output.is_absolute():
output = root / output
checks = 0
errors: list[str] = []
def check(condition: bool, detail: str) -> None:
nonlocal checks
checks += 1
if not condition:
errors.append(detail)
historical_path = report_dir / "source_validation_at_generation.json"
historical = json.loads(historical_path.read_text(encoding="utf-8"))
check(historical.get("status") == "PASS", "historical source validation is not PASS")
historical_counts = historical.get("counts", {})
check(historical_counts.get("failed_calls") == 0, "historical source validation has failures")
check(historical_counts.get("independently_parsed_graphs") == 56, "historical source parse count is not 56")
inventory = rows(report_dir / "operation_inventory.csv")
check(len(inventory) == 56, f"inventory rows {len(inventory)} != 56")
check(len({row["graph_id"] for row in inventory}) == 56, "duplicate graph_id")
check(sum(row["stage"] == "ONNX" for row in inventory) == 42, "ONNX graph count != 42")
check(sum(row["stage"] == "AFFINE_SCF_MEMREF" for row in inventory) == 14, "Affine graph count != 14")
for row in inventory:
check(row["analysis_status"] == "PASS", f"{row['graph_id']}: analysis status is not PASS")
for path_field, hash_field in (
("execution_dependency_graph_svg", "execution_dependency_graph_svg_sha256"),
("execution_dependency_graph_png", "execution_dependency_graph_png_sha256"),
("graph_record_json", "graph_record_json_sha256"),
):
path = root / row[path_field]
check(path.is_file(), f"{row['graph_id']}: missing {row[path_field]}")
if path.is_file():
check(sha256(path) == row[hash_field], f"{row['graph_id']}: checksum mismatch {path_field}")
record_path = root / row["graph_record_json"]
if record_path.is_file():
record = json.loads(record_path.read_text(encoding="utf-8"))
check(record.get("graph_id") == row["graph_id"], f"{row['graph_id']}: record identity mismatch")
check(record.get("stage") == "T85_MLIR_IR_GRAPH_RECORD", f"{row['graph_id']}: old record stage")
summary = json.loads((report_dir / "summary.json").read_text(encoding="utf-8"))
check(summary.get("status") == "PASS", "summary status is not PASS")
check(summary.get("stage") == "T85_MLIR_IR_GRAPH", "summary uses an old stage name")
counts = summary.get("counts", {})
check(csv_data_rows(report_dir / "operation_order.csv") == counts.get("operation_rows"), "operation row count mismatch")
check(csv_data_rows(report_dir / "ssa_edges.csv") == counts.get("ssa_edge_rows"), "SSA row count mismatch")
check(csv_data_rows(report_dir / "control_flow_edges.csv") == counts.get("control_relation_rows"), "control row count mismatch")
manifest = json.loads((report_dir / "artifact_manifest.json").read_text(encoding="utf-8"))
check(manifest.get("status") == "PASS", "artifact manifest is not PASS")
check(manifest.get("stage") == "T85_MLIR_IR_GRAPH_MANIFEST", "manifest uses an old stage name")
manifest_paths: set[str] = set()
for item in manifest.get("files", []):
path = root / item["path"]
manifest_paths.add(item["path"])
check(path.is_file(), f"manifest file missing: {item['path']}")
if path.is_file():
check(path.stat().st_size == item["bytes"], f"manifest size mismatch: {item['path']}")
check(sha256(path) == item["sha256"], f"manifest checksum mismatch: {item['path']}")
check(len(manifest_paths) == manifest.get("file_count"), "manifest duplicate path or count mismatch")
check(str(historical_path.relative_to(root)) in manifest_paths, "historical source validation is not pinned")
status = "PASS" if not errors else "FAIL"
result = {
"schema_version": "1.0",
"stage": "T85_RETAINED_MLIR_IR_GRAPH_VALIDATION",
"status": status,
"generated_at": datetime.now(timezone.utc).isoformat(),
"checks": checks,
"failed_checks": len(errors),
"errors": errors,
"validation_scope": "retained package integrity after repository naming cleanup",
"source_reparse_status": "PASS_AT_GENERATION_NOT_REPEATED_AFTER_COMPACT_CLEANUP",
"source_validation": {
"path": str(historical_path.relative_to(root)),
"sha256": sha256(historical_path),
"independently_parsed_graphs": historical_counts.get("independently_parsed_graphs"),
"check_calls": historical_counts.get("check_calls"),
"failed_calls": historical_counts.get("failed_calls"),
},
"policy": {
"model_runtime_run": False,
"compiler_or_lowering_run": False,
"dataset_run": False,
"allocator_run": False,
},
}
atomic_json(output, result)
print(f"{status}: checks={checks} failures={len(errors)}")
return 0 if status == "PASS" else 1
if __name__ == "__main__":
raise SystemExit(main())
|