#!/usr/bin/env python3 """Run and log the AD01 compiled-MLIR numerical/Q1 validation workflow.""" from __future__ import annotations import argparse import hashlib import json import os import platform import shlex import subprocess import sys import tempfile from pathlib import Path from typing import Any def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", required=True, type=Path) parser.add_argument("--result-dir", required=True, type=Path) parser.add_argument("--run-dir", required=True, type=Path) parser.add_argument("--report-csv", required=True, type=Path) parser.add_argument("--report-md", required=True, type=Path) parser.add_argument("--resume", action="store_true") return parser.parse_args() def sha256_file(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 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) handle.write("\n") temporary = Path(handle.name) os.replace(temporary, path) def run_stage( name: str, command: list[str], *, cwd: Path, run_dir: Path, environment: dict[str, str], ) -> 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" command_path.write_text(shlex.join(command) + "\n", encoding="utf-8") completed = subprocess.run( command, cwd=cwd, env=environment, text=True, capture_output=True, check=False, ) stdout_path.write_text(completed.stdout, encoding="utf-8") stderr_path.write_text(completed.stderr, encoding="utf-8") exit_path.write_text(f"{completed.returncode}\n", encoding="utf-8") return { "name": name, "status": "PASS" if completed.returncode == 0 else "FAIL", "failure_code": None if completed.returncode == 0 else "FAIL_ANALYSIS", "command": shlex.join(command), "exit_code": completed.returncode, "command_log": str(command_path), "stdout_log": str(stdout_path), "stderr_log": str(stderr_path), "exit_code_log": str(exit_path), "command_log_sha256": sha256_file(command_path), "stdout_log_sha256": sha256_file(stdout_path), "stderr_log_sha256": sha256_file(stderr_path), "exit_code_log_sha256": sha256_file(exit_path), } def attach_result_outcome(stage: dict[str, Any], result_path: Path) -> None: if not result_path.is_file(): return result = json.loads(result_path.read_text()) stage["result"] = str(result_path) stage["result_sha256"] = sha256_file(result_path) stage["result_status"] = result.get("status") stage["failure_code"] = result.get("failure_code") def main() -> int: args = parse_args() root = args.repo_root.resolve() result_dir = args.result_dir.resolve() run_dir = args.run_dir.resolve() report_csv = args.report_csv.resolve() report_md = args.report_md.resolve() if run_dir.exists() and any(run_dir.iterdir()): raise FileExistsError(f"refusing non-empty run directory: {run_dir}") if result_dir.exists() and any(result_dir.iterdir()) and not args.resume: raise FileExistsError(f"refusing non-empty result directory: {result_dir}") run_dir.mkdir(parents=True, exist_ok=True) result_dir.mkdir(parents=True, exist_ok=True) python = root / ".venv/bin/python" quality_site = root / "environment/quality/ad01_dcase/.venv/lib/python3.12/site-packages" environment = dict(os.environ) existing_pythonpath = environment.get("PYTHONPATH") environment["PYTHONPATH"] = ( str(quality_site) if not existing_pythonpath else str(quality_site) + os.pathsep + existing_pythonpath ) for name in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS", "NUMBA_NUM_THREADS"): environment[name] = "1" config = root / "configs/mlir/runtime_validation/AD01_compiled_numerical.json" quality_config = root / "configs/evaluation/anomaly_detection/AD01_dcase_quality_eval.json" fp32_dir = result_dir / "q1/fp32" quant_dir = result_dir / "q1/public_quantized" stages: list[dict[str, Any]] = [] fixed_report = result_dir / "compiled_output_comparison.json" if args.resume and fixed_report.is_file() and json.loads(fixed_report.read_text()).get("status") == "PASS": stages.append( { "name": "fixed_fixture", "status": "PASS", "failure_code": None, "reused": True, "report": str(fixed_report), "report_sha256": sha256_file(fixed_report), } ) else: stages.append( run_stage( "fixed_fixture", [ str(python), "scripts/stages/compare_onnx_mlir_compiled.py", "--repo-root", str(root), "--config", str(config), "--output-dir", str(result_dir), ], cwd=root, run_dir=run_dir, environment=environment, ) ) eval_common = [ str(python), "scripts/stages/evaluate_ad01_compiled_mlir.py", "--project-root", str(root), "--config", str(quality_config), ] stages.append( run_stage( "q1_fp32", eval_common + [ "--variant", "fp32", "--onnx", str(root / "models/anomaly_detection/AD01/onnx/fp32/model_fp32.onnx"), "--shared-library", str(root / "models/anomaly_detection/AD01/mlir/fp32/codegen/model.so"), "--expected-onnx-sha256", "4b23496a4b93b28a950ccb969e64dbbd1e16d7994d642ef6c292e230958e116f", "--expected-shared-library-sha256", "7ca859be70e9d6d0915cc95a93e34878cbeccdaec940a1a5cc05e52cf097ff17", "--output-dir", str(fp32_dir), ], cwd=root, run_dir=run_dir, environment=environment, ) ) stages.append( run_stage( "q1_public_quantized", eval_common + [ "--variant", "public_quantized", "--onnx", str(root / "models/anomaly_detection/AD01/onnx/quantized/model_quantized.onnx"), "--shared-library", str(root / "models/anomaly_detection/AD01/mlir/quantized/codegen/model.so"), "--expected-onnx-sha256", "f918dbcb7669ba49c9d782b440587343e25c3fba3fa2eb2b8052552a7355ce60", "--expected-shared-library-sha256", "cdb1cc4fc98e59b972d4fc555a0966a3e08e7e2a6f330aeb89c1d0966c5a5fd0", "--output-dir", str(quant_dir), ], cwd=root, run_dir=run_dir, environment=environment, ) ) attach_result_outcome(stages[-2], fp32_dir / "quality_summary.json") attach_result_outcome(stages[-1], quant_dir / "quality_summary.json") q1_outputs_complete = all( (directory / name).is_file() for directory in (fp32_dir, quant_dir) for name in ("quality_summary.json", "file_scores.csv", "quality_metrics.csv") ) if q1_outputs_complete: stages.append( run_stage( "merge_quality", [ str(python), "scripts/stages/evaluate_ad01_compiled_mlir.py", "--merge-input", str(fp32_dir), "--merge-input", str(quant_dir), "--output-dir", str(result_dir), ], cwd=root, run_dir=run_dir, environment=environment, ) ) else: stages.append({"name": "merge_quality", "status": "BLOCKED", "failure_code": "FAIL_PREREQUISITE"}) merge_summary_path = result_dir / "compiled_merge_summary.json" attach_result_outcome(stages[-1], merge_summary_path) if stages[0]["status"] == "PASS" and merge_summary_path.is_file(): stages.append( run_stage( "validate", [ str(python), "scripts/validate_ad01_compiled_numerical.py", "--repo-root", str(root), "--result-dir", str(result_dir), "--output", str(result_dir / "validation.json"), ], cwd=root, run_dir=run_dir, environment=environment, ) ) else: stages.append({"name": "validate", "status": "BLOCKED", "failure_code": "FAIL_PREREQUISITE"}) validation_path = result_dir / "validation.json" attach_result_outcome(stages[-1], validation_path) if validation_path.is_file(): stages.append( run_stage( "build_report", [ str(python), "scripts/build_ad01_compiled_numerical_report.py", "--repo-root", str(root), "--result-dir", str(result_dir), "--report-csv", str(report_csv), "--report-md", str(report_md), ], cwd=root, run_dir=run_dir, environment=environment, ) ) else: stages.append({"name": "build_report", "status": "BLOCKED", "failure_code": "FAIL_PREREQUISITE"}) if validation_path.is_file() and report_csv.is_file() and report_md.is_file(): stages.append( run_stage( "build_manifest", [ str(python), "scripts/build_ad01_compiled_numerical_manifest.py", "--repo-root", str(root), "--result-dir", str(result_dir), "--report", str(report_csv), "--report", str(report_md), ], cwd=root, run_dir=run_dir, environment=environment, ) ) else: stages.append({"name": "build_manifest", "status": "BLOCKED", "failure_code": "FAIL_PREREQUISITE"}) if stages[-1]["status"] == "PASS": stages.append( run_stage( "checksum_manifest", ["sha256sum", "-c", str(result_dir / "artifacts.sha256")], cwd=root, run_dir=run_dir, environment=environment, ) ) else: stages.append({"name": "checksum_manifest", "status": "BLOCKED", "failure_code": "FAIL_PREREQUISITE"}) stages.append( run_stage( "targeted_tests", [ str(python), "-m", "pytest", "-q", "tests/test_onnx_mlir_compiled_runtime.py", "tests/test_ad01_compiled_mlir_evaluation.py", "tests/test_validate_ad01_compiled_numerical.py", ], cwd=root, run_dir=run_dir, environment=environment, ) ) stages.append( run_stage( "compileall", [ str(python), "-m", "py_compile", "scripts/stages/onnx_mlir_compiled_runtime.py", "scripts/stages/compare_onnx_mlir_compiled.py", "scripts/stages/evaluate_ad01_compiled_mlir.py", "scripts/validate_ad01_compiled_numerical.py", "scripts/build_ad01_compiled_numerical_report.py", "scripts/build_ad01_compiled_numerical_manifest.py", "scripts/run_ad01_compiled_numerical_validation.py", ], cwd=root, run_dir=run_dir, environment=environment, ) ) evidence_complete = all( path.is_file() for path in ( fixed_report, fp32_dir / "quality_summary.json", quant_dir / "quality_summary.json", result_dir / "compiled_file_scores.csv", result_dir / "compiled_quality_metrics.csv", validation_path, report_csv, report_md, result_dir / "artifact_manifest.json", result_dir / "artifacts.sha256", ) ) validation_status = ( json.loads(validation_path.read_text()).get("status") if validation_path.is_file() else None ) numerical_fidelity_status = ( json.loads(merge_summary_path.read_text()).get("fidelity_status") if merge_summary_path.is_file() else None ) overall = "PASS" if ( evidence_complete and validation_status == "PASS" and numerical_fidelity_status == "PASS" ) else ( "PARTIAL" if evidence_complete else "FAIL" ) manifest = { "schema_version": "1.0", "model_id": "AD01", "stage": "S6-NV_COMPILED_MLIR_NUMERICAL_AND_Q1_VALIDATION", "status": overall, "numerical_fidelity_status": numerical_fidelity_status, "independent_validation_status": validation_status, "stages": stages, "environment": { "python": sys.version, "platform": platform.platform(), "pythonpath_quality_site": str(quality_site), "threads": 1, }, "policy": { "latency_measured": False, "converter_run": False, "lowering_run": False, "codegen_run": False, "allocator_run": False, "training_or_calibration": False, "model_or_weight_modified": False, "quantized_model_generated": False, }, "outputs": { "result_dir": str(result_dir), "report_csv": str(report_csv), "report_md": str(report_md), }, } atomic_json(run_dir / "execution_manifest.json", manifest) print(json.dumps({"status": overall, "run_dir": str(run_dir)}, sort_keys=True)) return 0 if overall in {"PASS", "PARTIAL"} else 1 if __name__ == "__main__": raise SystemExit(main())