#!/usr/bin/env python3 """Independently validate the AD01 ONNX-MLIR numerical evidence package. The validator is intentionally read-only with respect to ``result-dir``. It does not load a model, invoke a runtime, measure speed, or extract features. It recomputes numerical comparisons from saved NPY arrays and DCASE AUC/pAUC from saved per-file scores, then atomically writes one validation report. Accepted quality score layouts ------------------------------ The preferred layout is ``compiled_file_scores.csv`` with these columns:: filename,machine_id,label,feature_vectors, fp32_onnxruntime_score,fp32_compiled_score, public_quantized_onnxruntime_score,public_quantized_compiled_score For interruption-friendly evaluators, two variant CSVs are also accepted. Their paths must identify ``fp32`` or ``public_quantized`` and each must have ``filename,machine_id,label,feature_vectors,onnxruntime_score,compiled_score``. The validator joins those files by filename and rejects missing/duplicate records. A few unambiguous ``ort``/``int8`` spelling aliases are accepted. ``compiled_quality_metrics.csv`` is required. It contains one row for every variant and machine ID (id_01..id_04 plus Average), with ``auc``, ``pauc`` and ``max_fpr``. All metric values are independently recomputed here. """ from __future__ import annotations import argparse import csv import hashlib import json import math import os import tempfile from dataclasses import dataclass, field from pathlib import Path from typing import Any, Iterable, Mapping, Sequence import numpy as np CONFIG_RELATIVE = Path("configs/mlir/runtime_validation/AD01_compiled_numerical.json") SOURCE_COMPARE_RELATIVE = { "fp32": Path("models/anomaly_detection/AD01/onnx/fp32/onnx_output_compare.json"), "public_quantized": Path( "models/anomaly_detection/AD01/onnx/quantized/onnx_output_compare.json" ), } CANONICAL_QUALITY_SCORES = Path( "models/anomaly_detection/AD01/quality_evaluation/results/file_scores.csv" ) VARIANTS = ( "fp32_onnxruntime", "fp32_compiled", "public_quantized_onnxruntime", "public_quantized_compiled", ) MACHINE_IDS = ("id_01", "id_02", "id_03", "id_04") class ValidationContractError(ValueError): """Raised when evidence cannot be interpreted without guessing.""" 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("--output", required=True, type=Path) return parser.parse_args() def read_json(path: Path) -> dict[str, Any]: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): raise ValidationContractError(f"expected JSON object: {path}") return value 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 array_sha256(value: np.ndarray[Any, Any]) -> str: return hashlib.sha256(np.ascontiguousarray(value).tobytes()).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 path_record(path: Path, root: Path) -> dict[str, Any]: resolved = path.resolve() try: display = str(resolved.relative_to(root.resolve())) except ValueError: display = str(resolved) return { "path": display, "bytes": resolved.stat().st_size, "sha256": sha256_file(resolved), } @dataclass class Checks: rows: list[dict[str, Any]] = field(default_factory=list) def add(self, name: str, ok: bool, detail: Any = None) -> bool: self.rows.append( { "check": name, "status": "PASS" if ok else "FAIL", "detail": detail, } ) return ok @property def passed(self) -> bool: return all(row["status"] == "PASS" for row in self.rows) @property def failure_count(self) -> int: return sum(row["status"] != "PASS" for row in self.rows) def _finite_numeric(value: np.ndarray[Any, Any]) -> bool: try: return bool(np.all(np.isfinite(value))) except TypeError: return False def compare_arrays( reference: np.ndarray[Any, Any], candidate: np.ndarray[Any, Any], *, atol: float, rtol: float, exact: bool, ) -> dict[str, Any]: shape_match = reference.shape == candidate.shape dtype_match = reference.dtype == candidate.dtype finite = _finite_numeric(reference) and _finite_numeric(candidate) if not shape_match: return { "status": "FAIL", "shape_match": False, "dtype_match": dtype_match, "finite": finite, "reference_shape": list(reference.shape), "candidate_shape": list(candidate.shape), "reference_dtype": str(reference.dtype), "candidate_dtype": str(candidate.dtype), "atol": atol, "rtol": rtol, "exact_required": exact, } left = reference.astype(np.float64) right = candidate.astype(np.float64) delta = right - left absolute = np.abs(delta) array_equal = bool(np.array_equal(reference, candidate)) allclose = bool(np.allclose(left, right, atol=atol, rtol=rtol, equal_nan=False)) accepted = array_equal if exact else allclose denominator = float(np.linalg.norm(left.ravel()) * np.linalg.norm(right.ravel())) if denominator == 0.0: cosine = 1.0 if array_equal else 0.0 else: cosine = float(np.dot(left.ravel(), right.ravel()) / denominator) mismatch_count = int(np.count_nonzero(reference != candidate)) return { "status": "PASS" if accepted and dtype_match and finite else "FAIL", "shape_match": True, "dtype_match": dtype_match, "finite": finite, "reference_shape": list(reference.shape), "candidate_shape": list(candidate.shape), "reference_dtype": str(reference.dtype), "candidate_dtype": str(candidate.dtype), "reference_array_sha256": array_sha256(reference), "candidate_array_sha256": array_sha256(candidate), "atol": atol, "rtol": rtol, "exact_required": exact, "array_equal": array_equal, "allclose": allclose, "total_element_count": int(reference.size), "mismatch_element_count": mismatch_count, "max_abs_error": float(absolute.max()) if absolute.size else 0.0, "mean_abs_error": float(absolute.mean()) if absolute.size else 0.0, "p95_abs_error": float(np.percentile(absolute, 95)) if absolute.size else 0.0, "rmse": float(math.sqrt(float(np.mean(np.square(delta))))) if delta.size else 0.0, "cosine_similarity": cosine, } def reported_comparison_matches( reported: Mapping[str, Any], recalculated: Mapping[str, Any], *, tolerance: float = 1e-15, ) -> bool: if reported.get("status") != recalculated.get("status"): return False if reported.get("reference_sha256") != recalculated.get("reference_array_sha256"): return False if reported.get("candidate_sha256") != recalculated.get("candidate_array_sha256"): return False for key in ("shape_match", "dtype_match", "allclose", "mismatch_element_count"): if reported.get(key) != recalculated.get(key): return False for key in ("max_abs_error", "mean_abs_error", "p95_abs_error", "rmse"): if not math.isclose( float(reported.get(key, math.nan)), float(recalculated.get(key, math.nan)), rel_tol=0.0, abs_tol=tolerance, ): return False return True def _roc_curve(labels: Sequence[int], scores: Sequence[float]) -> tuple[np.ndarray, np.ndarray]: y = np.asarray(labels, dtype=np.int8) values = np.asarray(scores, dtype=np.float64) if y.ndim != 1 or values.ndim != 1 or y.size != values.size or y.size == 0: raise ValidationContractError("labels and scores must be nonempty equal-length vectors") if not np.all(np.isfinite(values)): raise ValidationContractError("non-finite quality score") unique = set(int(item) for item in np.unique(y)) if unique != {0, 1}: raise ValidationContractError(f"binary labels require both 0 and 1, got {unique}") order = np.argsort(values, kind="mergesort")[::-1] ordered_scores = values[order] ordered_labels = y[order] distinct = np.where(np.diff(ordered_scores))[0] thresholds = np.r_[distinct, ordered_labels.size - 1] true_positives = np.cumsum(ordered_labels, dtype=np.float64)[thresholds] false_positives = 1.0 + thresholds - true_positives true_positives = np.r_[0.0, true_positives] false_positives = np.r_[0.0, false_positives] return false_positives / false_positives[-1], true_positives / true_positives[-1] def auc_pauc( labels: Sequence[int], scores: Sequence[float], max_fpr: float ) -> tuple[float, float]: """Return ROC AUC and sklearn-compatible standardized partial AUC.""" if not 0.0 < max_fpr <= 1.0: raise ValidationContractError(f"invalid max_fpr: {max_fpr}") fpr, tpr = _roc_curve(labels, scores) auc = float(np.trapezoid(tpr, fpr)) if max_fpr == 1.0: return auc, auc stop = int(np.searchsorted(fpr, max_fpr, side="right")) x = np.append(fpr[:stop], max_fpr) y = np.append(tpr[:stop], np.interp(max_fpr, fpr[stop - 1 : stop + 1], tpr[stop - 1 : stop + 1])) partial = float(np.trapezoid(y, x)) minimum = 0.5 * max_fpr**2 maximum = max_fpr standardized = 0.5 * (1.0 + (partial - minimum) / (maximum - minimum)) return auc, float(standardized) def recompute_metrics( rows: Sequence[Mapping[str, Any]], max_fpr: float ) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] for variant in VARIANTS: per_id: list[tuple[float, float]] = [] for machine_id in MACHINE_IDS: selected = [row for row in rows if row["machine_id"] == machine_id] labels = [int(row["label"]) for row in selected] scores = [float(row[variant]) for row in selected] auc, pauc = auc_pauc(labels, scores, max_fpr) per_id.append((auc, pauc)) result.append( { "variant": variant, "machine_id": machine_id, "normal_files": labels.count(0), "anomaly_files": labels.count(1), "auc": auc, "pauc": pauc, "max_fpr": max_fpr, } ) result.append( { "variant": variant, "machine_id": "Average", "normal_files": sum(int(row["label"]) == 0 for row in rows), "anomaly_files": sum(int(row["label"]) == 1 for row in rows), "auc": float(np.mean([value[0] for value in per_id])), "pauc": float(np.mean([value[1] for value in per_id])), "max_fpr": max_fpr, } ) return result ALIASES = { "filename": ("filename", "file", "file_name"), "machine_id": ("machine_id", "id"), "label": ("label", "target", "y_true"), "feature_vectors": ("feature_vectors", "vector_count", "vectors"), "fp32_onnxruntime": ( "fp32_onnxruntime_score", "fp32_ort_score", "fp32_onnx_score", ), "fp32_compiled": ("fp32_compiled_score", "fp32_so_score"), "public_quantized_onnxruntime": ( "public_quantized_onnxruntime_score", "public_quantized_ort_score", "quantized_onnxruntime_score", "int8_ort_score", ), "public_quantized_compiled": ( "public_quantized_compiled_score", "quantized_compiled_score", "int8_compiled_score", "public_int8_compiled_score", ), } def _field(row: Mapping[str, str], logical: str, required: bool = True) -> str | None: for name in ALIASES[logical]: if name in row and row[name] not in (None, ""): return row[name] if required: raise ValidationContractError( f"missing {logical}; accepted columns={list(ALIASES[logical])}" ) return None def _read_csv(path: Path) -> list[dict[str, str]]: with path.open(newline="", encoding="utf-8") as handle: rows = list(csv.DictReader(handle)) if not rows: raise ValidationContractError(f"empty CSV: {path}") return rows def _base_row(row: Mapping[str, str]) -> dict[str, Any]: label = int(str(_field(row, "label"))) if label not in (0, 1): raise ValidationContractError(f"non-binary label: {label}") return { "filename": str(_field(row, "filename")), "machine_id": str(_field(row, "machine_id")), "label": label, "feature_vectors": int(str(_field(row, "feature_vectors"))), } def _normalize_merged(rows: Sequence[Mapping[str, str]]) -> list[dict[str, Any]]: output: list[dict[str, Any]] = [] for raw in rows: row = _base_row(raw) for variant in VARIANTS: row[variant] = float(str(_field(raw, variant))) output.append(row) return output def _classify_variant_path(path: Path) -> str | None: text = "/".join(part.lower().replace("-", "_") for part in path.parts) if any(token in text for token in ("public_quantized", "quantized", "int8")): return "public_quantized" if "fp32" in text: return "fp32" return None def _normalize_variant_pair( fp32_path: Path, quantized_path: Path ) -> list[dict[str, Any]]: combined: dict[str, dict[str, Any]] = {} for prefix, path in (("fp32", fp32_path), ("public_quantized", quantized_path)): for raw in _read_csv(path): base = _base_row(raw) name = base["filename"] if name in combined: for key in ("machine_id", "label", "feature_vectors"): if combined[name][key] != base[key]: raise ValidationContractError( f"variant metadata mismatch for {name}: {key}" ) else: combined[name] = base ort = next( ( raw[key] for key in ("onnxruntime_score", "ort_score", "onnx_score") if key in raw and raw[key] != "" ), None, ) compiled = next( ( raw[key] for key in ("compiled_score", "so_score") if key in raw and raw[key] != "" ), None, ) if ort is None or compiled is None: raise ValidationContractError( f"{path}: variant CSV needs onnxruntime_score and compiled_score" ) combined[name][f"{prefix}_onnxruntime"] = float(ort) combined[name][f"{prefix}_compiled"] = float(compiled) missing = [name for name, row in combined.items() if any(key not in row for key in VARIANTS)] if missing: raise ValidationContractError( f"variant score files do not cover the same filenames: {missing[:5]}" ) return [combined[name] for name in sorted(combined)] def load_quality_scores(result_dir: Path) -> tuple[list[dict[str, Any]], list[Path]]: merged_candidates = sorted(result_dir.rglob("compiled_file_scores.csv")) for path in merged_candidates: raw = _read_csv(path) try: return _normalize_merged(raw), [path] except ValidationContractError: # A variant-specific file may share the preferred basename. continue candidates = sorted( path for path in result_dir.rglob("*.csv") if "file_scores" in path.name and path.name != "compiled_quality_metrics.csv" ) classified: dict[str, list[Path]] = {"fp32": [], "public_quantized": []} for path in candidates: variant = _classify_variant_path(path.relative_to(result_dir)) if variant: classified[variant].append(path) if len(classified["fp32"]) == 1 and len(classified["public_quantized"]) == 1: paths = [classified["fp32"][0], classified["public_quantized"][0]] return _normalize_variant_pair(paths[0], paths[1]), paths raise ValidationContractError( "quality scores missing: require one merged compiled_file_scores.csv or " "exactly one fp32 and one public_quantized variant file_scores CSV" ) def _metric_variant(value: str) -> str: normalized = value.strip().lower().replace("-", "_") aliases = { "fp32_ort": "fp32_onnxruntime", "fp32_onnx": "fp32_onnxruntime", "fp32_onnxruntime": "fp32_onnxruntime", "fp32_compiled": "fp32_compiled", "fp32_so": "fp32_compiled", "public_quantized_ort": "public_quantized_onnxruntime", "public_quantized_onnx": "public_quantized_onnxruntime", "public_quantized_onnxruntime": "public_quantized_onnxruntime", "quantized_onnxruntime": "public_quantized_onnxruntime", "int8_ort": "public_quantized_onnxruntime", "public_quantized_compiled": "public_quantized_compiled", "quantized_compiled": "public_quantized_compiled", "int8_compiled": "public_quantized_compiled", } if normalized not in aliases: raise ValidationContractError(f"unknown metric variant: {value}") return aliases[normalized] def find_quality_metrics(result_dir: Path) -> Path: candidates = sorted(result_dir.rglob("compiled_quality_metrics.csv")) if len(candidates) != 1: raise ValidationContractError( f"expected exactly one compiled_quality_metrics.csv, found {len(candidates)}" ) return candidates[0] def validate_metric_csv( path: Path, recomputed: Sequence[Mapping[str, Any]], *, tolerance: float = 1e-12 ) -> dict[str, Any]: reported_rows = _read_csv(path) reported: dict[tuple[str, str], Mapping[str, str]] = {} for row in reported_rows: variant = _metric_variant(row.get("variant", "")) machine_id = row.get("machine_id", "") key = (variant, machine_id) if key in reported: raise ValidationContractError(f"duplicate metric row: {key}") reported[key] = row expected = {(str(row["variant"]), str(row["machine_id"])): row for row in recomputed} if set(reported) != set(expected): raise ValidationContractError( f"metric row set mismatch: missing={sorted(set(expected)-set(reported))}, " f"extra={sorted(set(reported)-set(expected))}" ) maximum_delta = 0.0 for key, calculated in expected.items(): row = reported[key] for field_name in ("auc", "pauc", "max_fpr"): delta = abs(float(row[field_name]) - float(calculated[field_name])) maximum_delta = max(maximum_delta, delta) if delta > tolerance: raise ValidationContractError( f"metric mismatch {key} {field_name}: reported={row[field_name]} " f"recomputed={calculated[field_name]} delta={delta}" ) for field_name in ("normal_files", "anomaly_files"): if field_name in row and row[field_name] != "": if int(row[field_name]) != int(calculated[field_name]): raise ValidationContractError(f"metric count mismatch {key} {field_name}") return { "status": "PASS", "reported_rows": len(reported), "recomputed_rows": len(expected), "maximum_absolute_metric_delta": maximum_delta, "tolerance": tolerance, } def _evidence_path(value: str, base: Path) -> Path: path = Path(value) return path.resolve() if path.is_absolute() else (base / path).resolve() def validate_quality_manifest( result_dir: Path, score_paths: Sequence[Path], metrics_path: Path, artifacts: Mapping[str, Mapping[str, Mapping[str, Any]]], ) -> dict[str, Any]: """Validate the merge summary as the checksum-bearing Q1 manifest.""" candidates = sorted(result_dir.rglob("compiled_merge_summary.json")) if len(candidates) != 1: raise ValidationContractError( f"expected exactly one compiled_merge_summary.json, found {len(candidates)}" ) manifest_path = candidates[0] manifest = read_json(manifest_path) merge_fidelity = manifest.get("fidelity_status") merge_outcome_consistent = ( (merge_fidelity == "PASS" and manifest.get("status") == "PASS" and manifest.get("failure_code") is None) or ( merge_fidelity == "FAIL" and manifest.get("status") == "FAIL" and manifest.get("failure_code") == "FAIL_NUMERICAL_MISMATCH" ) ) if ( not merge_outcome_consistent or manifest.get("measurement_status") != "MEASURED" or manifest.get("acceptance_status") != "THRESHOLD_UNDEFINED" or manifest.get("latency_measured") is not False or int(manifest.get("file_score_rows", -1)) != 2459 or int(manifest.get("quality_metric_rows", -1)) != 20 ): raise ValidationContractError("compiled merge summary terminal contract mismatch") outputs = manifest.get("outputs", {}) score_output = outputs.get("compiled_file_scores", {}) metric_output = outputs.get("compiled_quality_metrics", {}) if len(score_paths) != 1: raise ValidationContractError( "compiled merge manifest requires the canonical merged score CSV" ) for name, record, actual in ( ("compiled_file_scores", score_output, score_paths[0]), ("compiled_quality_metrics", metric_output, metrics_path), ): recorded_path = _evidence_path(str(record.get("path", "")), manifest_path.parent) if recorded_path != actual.resolve(): raise ValidationContractError(f"merge output path mismatch: {name}") if record.get("sha256") != sha256_file(actual): raise ValidationContractError(f"merge output checksum mismatch: {name}") if int(record.get("bytes", -1)) != actual.stat().st_size: raise ValidationContractError(f"merge output byte count mismatch: {name}") input_records = manifest.get("inputs", {}) if set(input_records) != {"fp32", "public_quantized"}: raise ValidationContractError("merge summary must name both compiled variants") variant_manifests: dict[str, Any] = {} variant_score_paths: dict[str, Path] = {} for variant, record in input_records.items(): directory = _evidence_path(str(record.get("directory", "")), manifest_path.parent) summary_path = directory / "quality_summary.json" score_path = directory / "file_scores.csv" metric_path = directory / "quality_metrics.csv" for label, path, checksum_key in ( ("summary", summary_path, "summary_sha256"), ("file scores", score_path, "file_scores_sha256"), ("metrics", metric_path, "quality_metrics_sha256"), ): if not path.is_file() or sha256_file(path) != record.get(checksum_key): raise ValidationContractError( f"{variant} input {label} missing or checksum mismatch" ) summary = read_json(summary_path) variant_fidelity = summary.get("fidelity_status") variant_outcome_consistent = ( variant_fidelity == "PASS" and summary.get("status") == "PARTIAL" or ( variant_fidelity == "FAIL" and summary.get("status") == "FAIL" and summary.get("failure_code") == "FAIL_NUMERICAL_MISMATCH" ) ) if ( summary.get("variant") != variant or not variant_outcome_consistent or summary.get("measurement_status") != "MEASURED" or summary.get("acceptance_status") != "THRESHOLD_UNDEFINED" or summary.get("latency_measured") is not False or summary.get("evaluation_fingerprint") != record.get("evaluation_fingerprint") ): raise ValidationContractError(f"{variant} quality summary contract mismatch") inputs = summary.get("input_artifacts", {}) expected = artifacts[variant] for role, summary_role in ( ("onnx", "canonical_onnx"), ("compiled_library", "compiled_shared_library"), ("source_tflite", "source_tflite"), ): if inputs.get(summary_role, {}).get("sha256") != expected[role]["sha256"]: raise ValidationContractError( f"{variant} Q1 {summary_role} checksum differs from contract" ) variant_manifests[variant] = { "summary": path_record(summary_path, result_dir), "file_scores": path_record(score_path, result_dir), "quality_metrics": path_record(metric_path, result_dir), "fidelity_status": summary["fidelity_status"], "measurement_status": summary["measurement_status"], } variant_score_paths[variant] = score_path source_join = _normalize_variant_pair( variant_score_paths["fp32"], variant_score_paths["public_quantized"] ) merged_rows = _normalize_merged(_read_csv(score_paths[0])) source_by_name = {str(row["filename"]): row for row in source_join} merged_by_name = {str(row["filename"]): row for row in merged_rows} if source_by_name != merged_by_name: raise ValidationContractError( "compiled merged scores differ from checksum-pinned variant file scores" ) return { "status": "PASS", "manifest": path_record(manifest_path, result_dir), "variant_inputs": variant_manifests, "variant_score_rows_rejoined": len(source_join), "merged_scores_match_variant_inputs": True, "output_checksums_match": True, } observed = { str(row["filename"]): (str(row["machine_id"]), int(row["label"])) for row in rows } if observed != identity: raise ValidationContractError( "compiled quality dataset identity differs from canonical Q1" ) return { "status": "PASS", "test_files": len(rows), "feature_vectors": sum(int(row["feature_vectors"]) for row in rows), "machine_ids": sorted(set(str(row["machine_id"]) for row in rows)), "normal_files": sum(int(row["label"]) == 0 for row in rows), "anomaly_files": sum(int(row["label"]) == 1 for row in rows), "canonical_filename_label_machine_id_match": True, } observed = { str(row["filename"]): (str(row["machine_id"]), int(row["label"])) for row in rows } if observed != identity: raise ValidationContractError( "compiled quality dataset identity differs from canonical Q1" ) return { "status": "PASS", "test_files": len(rows), "feature_vectors": sum(int(row["feature_vectors"]) for row in rows), "machine_ids": sorted(set(str(row["machine_id"]) for row in rows)), "normal_files": sum(int(row["label"]) == 0 for row in rows), "anomaly_files": sum(int(row["label"]) == 1 for row in rows), "canonical_filename_label_machine_id_match": True, } observed = { str(row["filename"]): (str(row["machine_id"]), int(row["label"])) for row in rows } if observed != identity: raise ValidationContractError( "compiled quality dataset identity differs from canonical Q1" ) return { "status": "PASS", "test_files": len(rows), "feature_vectors": sum(int(row["feature_vectors"]) for row in rows), "machine_ids": sorted(set(str(row["machine_id"]) for row in rows)), "normal_files": sum(int(row["label"]) == 0 for row in rows), "anomaly_files": sum(int(row["label"]) == 1 for row in rows), "canonical_filename_label_machine_id_match": True, } observed = { str(row["filename"]): (str(row["machine_id"]), int(row["label"])) for row in rows } if observed != identity: raise ValidationContractError( "compiled quality dataset identity differs from canonical Q1" ) return { "status": "PASS", "test_files": len(rows), "feature_vectors": sum(int(row["feature_vectors"]) for row in rows), "machine_ids": sorted(set(str(row["machine_id"]) for row in rows)), "normal_files": sum(int(row["label"]) == 0 for row in rows), "anomaly_files": sum(int(row["label"]) == 1 for row in rows), "canonical_filename_label_machine_id_match": True, } observed = { str(row["filename"]): (str(row["machine_id"]), int(row["label"])) for row in rows } if observed != identity: raise ValidationContractError( "compiled quality dataset identity differs from canonical Q1" ) return { "status": "PASS", "test_files": len(rows), "feature_vectors": sum(int(row["feature_vectors"]) for row in rows), "machine_ids": sorted(set(str(row["machine_id"]) for row in rows)), "normal_files": sum(int(row["label"]) == 0 for row in rows), "anomaly_files": sum(int(row["label"]) == 1 for row in rows), "canonical_filename_label_machine_id_match": True, } def validate_quality_identity( rows: Sequence[Mapping[str, Any]], canonical_path: Path, config: Mapping[str, Any] ) -> dict[str, Any]: quality = config["quality"] expected_files = int(quality["expected_test_files"]) expected_vectors = int(quality["expected_total_feature_vectors"]) if len(rows) != expected_files: raise ValidationContractError( f"quality file count mismatch: {len(rows)} != {expected_files}" ) filenames = [str(row["filename"]) for row in rows] if len(set(filenames)) != len(filenames): raise ValidationContractError("duplicate filename in compiled quality scores") if set(str(row["machine_id"]) for row in rows) != set(MACHINE_IDS): raise ValidationContractError("quality scores must contain machine IDs id_01..id_04") if sum(int(row["feature_vectors"]) for row in rows) != expected_vectors: raise ValidationContractError("quality feature vector total mismatch") if not all( int(row["feature_vectors"]) == int(quality["expected_feature_vectors_per_file"]) for row in rows ): raise ValidationContractError("unexpected per-file feature vector count") for row in rows: for variant in VARIANTS: if not math.isfinite(float(row[variant])): raise ValidationContractError(f"non-finite file score: {row['filename']} {variant}") canonical = _read_csv(canonical_path) identity = { str(row["filename"]): (str(row["machine_id"]), int(row["label"])) for row in canonical } observed = { str(row["filename"]): (str(row["machine_id"]), int(row["label"])) for row in rows } if observed != identity: raise ValidationContractError( "compiled quality dataset identity differs from canonical Q1" ) return { "status": "PASS", "test_files": len(rows), "feature_vectors": sum(int(row["feature_vectors"]) for row in rows), "machine_ids": sorted(set(str(row["machine_id"]) for row in rows)), "normal_files": sum(int(row["label"]) == 0 for row in rows), "anomaly_files": sum(int(row["label"]) == 1 for row in rows), "canonical_filename_label_machine_id_match": True, } observed = { str(row["filename"]): (str(row["machine_id"]), int(row["label"])) for row in rows } if observed != identity: raise ValidationContractError( "compiled quality dataset identity differs from canonical Q1" ) return { "status": "PASS", "test_files": len(rows), "feature_vectors": sum(int(row["feature_vectors"]) for row in rows), "machine_ids": sorted(set(str(row["machine_id"]) for row in rows)), "normal_files": sum(int(row["label"]) == 0 for row in rows), "anomaly_files": sum(int(row["label"]) == 1 for row in rows), "canonical_filename_label_machine_id_match": True, } observed = { str(row["filename"]): (str(row["machine_id"]), int(row["label"])) for row in rows } if observed != identity: raise ValidationContractError( "compiled quality dataset identity differs from canonical Q1" ) return { "status": "PASS", "test_files": len(rows), "feature_vectors": sum(int(row["feature_vectors"]) for row in rows), "machine_ids": sorted(set(str(row["machine_id"]) for row in rows)), "normal_files": sum(int(row["label"]) == 0 for row in rows), "anomaly_files": sum(int(row["label"]) == 1 for row in rows), "canonical_filename_label_machine_id_match": True, } def quality_max_fpr(config: Mapping[str, Any], root: Path) -> float: """Resolve max_fpr from the compiled contract and cross-check base Q1.""" quality = config.get("quality", {}) configured = quality.get("max_fpr") base_relative = config.get("base_quality_config") base_value: float | None = None if base_relative: base_path = (root / str(base_relative)).resolve() base = read_json(base_path) base_value = float(base["metric"]["max_fpr"]) if configured is None and base_value is None: raise ValidationContractError("quality.max_fpr is missing from both contracts") resolved = float(base_value if configured is None else configured) if base_value is not None and not math.isclose( resolved, base_value, rel_tol=0.0, abs_tol=0.0 ): raise ValidationContractError( f"compiled/base quality max_fpr mismatch: {resolved} != {base_value}" ) return resolved SPEED_METRIC_KEYS = { "latency_ms", "mean_ms", "p50_ms", "p95_ms", "min_ms", "max_ms", "throughput", "throughput_per_second", "inferences_per_second", "tokens_per_second", "speedup", } SPEED_POLICY_BOOLEAN_KEYS = { "latency_measured", "speed_measured", "benchmark_run", "performance_measured", "target_performance_claimed", } def _walk_json(value: Any, location: str = "$") -> Iterable[tuple[str, str, Any]]: if isinstance(value, Mapping): for key, child in value.items(): child_location = f"{location}.{key}" yield child_location, str(key), child yield from _walk_json(child, child_location) elif isinstance(value, list): for index, child in enumerate(value): yield from _walk_json(child, f"{location}[{index}]") def validate_no_speed_policy(result_dir: Path, config: Mapping[str, Any]) -> dict[str, Any]: violations: list[str] = [] policy = config.get("policy", {}) if policy.get("latency_measured") is not False: violations.append("config policy.latency_measured must be false") json_files = sorted(result_dir.rglob("*.json")) for path in json_files: try: document = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError as error: violations.append(f"invalid JSON {path}: {error}") continue for location, key, value in _walk_json(document): normalized = key.lower() if normalized in SPEED_METRIC_KEYS: violations.append(f"speed metric present: {path}:{location}") if normalized in SPEED_POLICY_BOOLEAN_KEYS and value is not False: violations.append(f"speed policy is not false: {path}:{location}={value!r}") for path in result_dir.rglob("*"): if path.is_file() and "latencies" in path.name.lower(): violations.append(f"latency artifact present: {path}") return { "status": "PASS" if not violations else "FAIL", "policy": "NUMERICAL_AND_Q1_VALIDATION_ONLY_NO_SPEED_MEASUREMENT", "json_files_scanned": len(json_files), "violations": violations, } def _required_file(path: Path, label: str, checks: Checks) -> bool: return checks.add(label, path.is_file(), str(path)) def validate(repo_root: Path, result_dir: Path) -> dict[str, Any]: root = repo_root.resolve() results = result_dir.resolve() checks = Checks() config_path = root / CONFIG_RELATIVE if not _required_file(config_path, "contract_config_exists", checks): return _failed_report(root, results, checks, {}) config = read_json(config_path) checks.add("contract_model_id", config.get("model_id") == "AD01", config.get("model_id")) checks.add( "contract_quality_threshold_undefined", config.get("quality", {}).get("acceptance_threshold") is None and config.get("quality", {}).get("acceptance_policy") == "THRESHOLD_UNDEFINED", config.get("quality", {}).get("acceptance_policy"), ) checks.add("result_directory_exists", results.is_dir(), str(results)) evidence: dict[str, Any] = {"contract_config": path_record(config_path, root)} artifacts: dict[str, Any] = {} for variant in ("fp32", "public_quantized"): artifacts[variant] = {} for role in ("onnx", "compiled_library", "source_tflite"): contract = config["artifacts"][variant][role] path = (root / contract["path"]).resolve() exists = _required_file(path, f"artifact_exists:{variant}:{role}", checks) if not exists: continue record = path_record(path, root) record["expected_sha256"] = contract["sha256"] record["checksum_match"] = record["sha256"] == contract["sha256"] checks.add( f"artifact_checksum:{variant}:{role}", record["checksum_match"], record, ) artifacts[variant][role] = record evidence["model_artifacts"] = artifacts source_comparisons: dict[str, Any] = {} for variant, relative in SOURCE_COMPARE_RELATIVE.items(): path = root / relative if not _required_file(path, f"source_compare_exists:{variant}", checks): continue document = read_json(path) comparison = (document.get("output_comparisons") or [{}])[0] expected_domain = "real_value" if variant == "fp32" else "raw_integer" ok = ( document.get("runtime_output_match") is True and document.get("random_seed") == config["random_seed"] and comparison.get("comparison_domain") == expected_domain and comparison.get("match") is True ) checks.add(f"source_to_onnx_validation:{variant}", ok, document) source_comparisons[variant] = { "artifact": path_record(path, root), "runtime_output_match": document.get("runtime_output_match"), "comparison_domain": comparison.get("comparison_domain"), "onnx_output_array_sha256": comparison.get("onnx_output_sha256"), } evidence["source_to_onnx"] = source_comparisons report_path = results / "compiled_output_comparison.json" fixed: dict[str, Any] = {} if _required_file(report_path, "compiled_comparison_report_exists", checks): comparison_report = read_json(report_path) evidence["compiled_comparison_report"] = path_record(report_path, results) checks.add( "compiled_report_scope_not_latency", comparison_report.get("latency_measured") is False and comparison_report.get("dataset_quality_measured") is False and comparison_report.get("measurement_kind") == "COMPILER_OUTPUT_NUMERICAL_FIDELITY_NOT_LATENCY", { "latency_measured": comparison_report.get("latency_measured"), "dataset_quality_measured": comparison_report.get("dataset_quality_measured"), "measurement_kind": comparison_report.get("measurement_kind"), }, ) checks.add( "compiled_report_terminal_status", comparison_report.get("status") == "PASS" and comparison_report.get("failure_code") is None and all( comparison_report.get("abi_runtime_checks", {}).get(variant, {}).get("status") == "PASS" for variant in ("fp32", "public_quantized") ), { "status": comparison_report.get("status"), "failure_code": comparison_report.get("failure_code"), "abi_runtime_checks": comparison_report.get("abi_runtime_checks"), }, ) expected_report_artifacts = { "fp32_onnx": artifacts.get("fp32", {}).get("onnx"), "fp32_so": artifacts.get("fp32", {}).get("compiled_library"), "quantized_onnx": artifacts.get("public_quantized", {}).get("onnx"), "quantized_so": artifacts.get("public_quantized", {}).get("compiled_library"), "quantized_tflite": artifacts.get("public_quantized", {}).get("source_tflite"), } report_artifacts = comparison_report.get("input_artifacts", {}) report_hashes_ok = all( record is not None and report_artifacts.get(name, {}).get("sha256") == record["sha256"] for name, record in expected_report_artifacts.items() ) checks.add("compiled_report_input_checksums", report_hashes_ok, report_artifacts) array_paths = { "semantic_input": results / "semantic_input.npy", "quantized_input": results / "quantized_input.npy", "fp32_onnxruntime": results / "fp32_ort_output.npy", "fp32_compiled": results / "fp32_compiled_runtime/output_0.npy", "public_quantized_onnxruntime": results / "quantized_ort_output.npy", "public_quantized_compiled": results / "public_quantized_compiled_runtime/output_0.npy", } arrays: dict[str, np.ndarray[Any, Any]] = {} for name, path in array_paths.items(): if _required_file(path, f"array_exists:{name}", checks): try: arrays[name] = np.load(path, allow_pickle=False) evidence.setdefault("array_artifacts", {})[name] = { **path_record(path, results), "array_sha256": array_sha256(arrays[name]), "shape": list(arrays[name].shape), "dtype": str(arrays[name].dtype), } except (OSError, ValueError) as error: checks.add(f"array_load:{name}", False, f"{type(error).__name__}: {error}") if "semantic_input" in arrays: expected = config["fixed_fixture"]["fp32_array_sha256"] checks.add( "fixed_fixture_checksum:fp32", array_sha256(arrays["semantic_input"]) == expected, {"observed": array_sha256(arrays["semantic_input"]), "expected": expected}, ) if "quantized_input" in arrays: expected = config["fixed_fixture"]["public_quantized_array_sha256"] checks.add( "fixed_fixture_checksum:public_quantized", array_sha256(arrays["quantized_input"]) == expected, {"observed": array_sha256(arrays["quantized_input"]), "expected": expected}, ) if {"fp32_onnxruntime", "fp32_compiled"} <= arrays.keys(): fixed["fp32_compiled_vs_onnxruntime"] = compare_arrays( arrays["fp32_onnxruntime"], arrays["fp32_compiled"], atol=float(config["artifacts"]["fp32"]["atol"]), rtol=float(config["artifacts"]["fp32"]["rtol"]), exact=False, ) checks.add( "recomputed_fp32_allclose", fixed["fp32_compiled_vs_onnxruntime"]["status"] == "PASS", fixed["fp32_compiled_vs_onnxruntime"], ) checks.add( "reported_fp32_comparison_matches_arrays", reported_comparison_matches( comparison_report.get("comparisons", {}).get( "fp32_compiled_vs_onnxruntime", {} ), fixed["fp32_compiled_vs_onnxruntime"], ), comparison_report.get("comparisons", {}).get( "fp32_compiled_vs_onnxruntime" ), ) if { "public_quantized_onnxruntime", "public_quantized_compiled", } <= arrays.keys(): fixed["public_quantized_compiled_vs_onnxruntime"] = compare_arrays( arrays["public_quantized_onnxruntime"], arrays["public_quantized_compiled"], atol=0.0, rtol=0.0, exact=True, ) checks.add( "recomputed_public_quantized_exact_raw_integer", fixed["public_quantized_compiled_vs_onnxruntime"]["status"] == "PASS", fixed["public_quantized_compiled_vs_onnxruntime"], ) checks.add( "reported_public_quantized_comparison_matches_arrays", reported_comparison_matches( comparison_report.get("comparisons", {}).get( "public_quantized_compiled_vs_onnxruntime", {} ), fixed["public_quantized_compiled_vs_onnxruntime"], ), comparison_report.get("comparisons", {}).get( "public_quantized_compiled_vs_onnxruntime" ), ) for variant in ("fp32", "public_quantized"): invocation_path = results / f"{variant}_compiled_runtime/invoke.json" if _required_file(invocation_path, f"compiled_invoke_exists:{variant}", checks): invocation = read_json(invocation_path) expected_so = artifacts.get(variant, {}).get("compiled_library", {}) invocation_ok = ( invocation.get("status") == "PASS" and invocation.get("deterministic_outputs") is True and invocation.get("finite_outputs") is True and invocation.get("inputs_unchanged") is True and invocation.get("latency_measured") is False and invocation.get("shared_library", {}).get("sha256") == expected_so.get("sha256") ) output_records = invocation.get("outputs", []) expected_output = results / f"{variant}_compiled_runtime/output_0.npy" invocation_ok = invocation_ok and len(output_records) == 1 and ( output_records[0].get("sha256") == sha256_file(expected_output) if expected_output.is_file() else False ) checks.add(f"compiled_invoke_contract:{variant}", invocation_ok, invocation) else: comparison_report = {} quality_result: dict[str, Any] = {} quality_sources: list[Path] = [] try: quality_rows, score_paths = load_quality_scores(results) quality_sources.extend(score_paths) canonical_path = root / CANONICAL_QUALITY_SCORES identity = validate_quality_identity(quality_rows, canonical_path, config) checks.add("official_dcase_quality_identity", True, identity) maximum_fpr = quality_max_fpr(config, root) recomputed = recompute_metrics(quality_rows, maximum_fpr) metrics_path = find_quality_metrics(results) quality_sources.append(metrics_path) metric_validation = validate_metric_csv(metrics_path, recomputed) checks.add("official_dcase_auc_pauc_recomputed", True, metric_validation) manifest_validation = validate_quality_manifest( results, score_paths, metrics_path, artifacts ) checks.add("official_dcase_checksum_manifest", True, manifest_validation) quality_result = { "measurement_status": "MEASURED", "acceptance_status": "THRESHOLD_UNDEFINED", "reason": "The exact compiled artifacts have no official pass/fail task-quality threshold.", "identity": identity, "metric_validation": metric_validation, "manifest_validation": manifest_validation, "recomputed_metrics": recomputed, "score_artifacts": [path_record(path, results) for path in score_paths], "metric_artifact": path_record(metrics_path, results), } except (OSError, ValueError, KeyError, csv.Error) as error: checks.add( "official_dcase_quality_complete", False, f"{type(error).__name__}: {error}", ) quality_result = { "measurement_status": "FAIL", "acceptance_status": "NOT_EVALUATED", "error": f"{type(error).__name__}: {error}", } no_speed = validate_no_speed_policy(results, config) checks.add("no_speed_measurement_policy", no_speed["status"] == "PASS", no_speed) evidence["quality_source_artifacts"] = [ path_record(path, results) for path in quality_sources if path.is_file() ] status = "PASS" if checks.passed else "FAIL" failed_names = { str(row["check"]) for row in checks.rows if row["status"] == "FAIL" } if status == "PASS": failure_code = None elif failed_names & { "recomputed_fp32_allclose", "reported_fp32_comparison_matches_arrays", "recomputed_public_quantized_exact_raw_integer", "reported_public_quantized_comparison_matches_arrays", }: failure_code = "FAIL_NUMERICAL_MISMATCH" elif any(name.startswith("compiled_invoke_contract:") for name in failed_names): failure_code = "FAIL_RUNTIME" else: failure_code = "FAIL_ANALYSIS" return { "schema_version": "1.0", "model_id": "AD01", "stage": "INDEPENDENT_ONNX_MLIR_COMPILED_NUMERICAL_AND_Q1_VALIDATION", "status": status, "failure_code": failure_code, "validator_scope": { "runtime_invoked": False, "dataset_features_extracted": False, "speed_measured": False, "arrays_recomputed": True, "quality_metrics_recomputed": True, }, "acceptance": { "fp32_compiled_vs_onnxruntime": "float32 shape/dtype exact and numpy allclose(atol=1e-5, rtol=1e-5)", "public_quantized_compiled_vs_onnxruntime": "raw int8 shape/dtype and every output element exact; zero-LSB tolerance", "quality": "official DCASE file-level reconstruction scores with independently recomputed ROC AUC and standardized pAUC(max_fpr=0.1); threshold remains undefined", "speed": "not measured and no speed claim", "quantization_preservation": "not changed by numerical PASS; existing low-level float-compute PARTIAL remains authoritative", }, "checks": checks.rows, "check_summary": { "total": len(checks.rows), "passed": len(checks.rows) - checks.failure_count, "failed": checks.failure_count, }, "evidence": evidence, "fixed_fixture_recalculation": fixed, "official_dcase_q1_recalculation": quality_result, "no_speed_policy": no_speed, } def _failed_report( root: Path, results: Path, checks: Checks, evidence: Mapping[str, Any] ) -> dict[str, Any]: return { "schema_version": "1.0", "model_id": "AD01", "stage": "INDEPENDENT_ONNX_MLIR_COMPILED_NUMERICAL_AND_Q1_VALIDATION", "status": "FAIL", "failure_code": "FAIL_ANALYSIS", "repo_root": str(root), "result_dir": str(results), "checks": checks.rows, "check_summary": { "total": len(checks.rows), "passed": len(checks.rows) - checks.failure_count, "failed": checks.failure_count, }, "evidence": evidence, } def main() -> int: args = parse_args() try: report = validate(args.repo_root, args.result_dir) except Exception as error: # emit an auditable terminal report for contract failures report = { "schema_version": "1.0", "model_id": "AD01", "stage": "INDEPENDENT_ONNX_MLIR_COMPILED_NUMERICAL_AND_Q1_VALIDATION", "status": "FAIL", "failure_code": "FAIL_ANALYSIS", "error": f"{type(error).__name__}: {error}", } atomic_json(args.output.resolve(), report) print( json.dumps( { "status": report["status"], "failure_code": report.get("failure_code"), "output": str(args.output.resolve()), }, sort_keys=True, ) ) return 0 if report["status"] == "PASS" else 1 if __name__ == "__main__": raise SystemExit(main())