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