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