ONNX
onnxruntime
onnx-mlir
quantization
fp32
ONNX_Models / tests /test_validate_ad01_compiled_numerical.py
purejomo's picture
Finalize public ONNX/ONNX-MLIR validation release
ed3aeeb
Raw
History Blame Contribute Delete
26.1 kB
from __future__ import annotations
import csv
import json
from pathlib import Path
import numpy as np
import pytest
from scripts import validate_ad01_compiled_numerical as validator
def _write_csv(path: Path, fieldnames: list[str], rows: list[dict[str, object]]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=fieldnames, extrasaction="ignore")
writer.writeheader()
writer.writerows(rows)
def test_compare_arrays_uses_allclose_for_fp32_and_exact_for_quantized() -> None:
reference = np.asarray([[1.0, 2.0]], dtype=np.float32)
close = np.asarray([[1.0, 2.0 + 1e-6]], dtype=np.float32)
assert validator.compare_arrays(
reference, close, atol=1e-5, rtol=1e-5, exact=False
)["status"] == "PASS"
raw = np.asarray([[1, 2]], dtype=np.int8)
changed = np.asarray([[1, 3]], dtype=np.int8)
exact = validator.compare_arrays(raw, raw.copy(), atol=0.0, rtol=0.0, exact=True)
mismatch = validator.compare_arrays(raw, changed, atol=0.0, rtol=0.0, exact=True)
assert exact["status"] == "PASS"
assert exact["mismatch_element_count"] == 0
assert mismatch["status"] == "FAIL"
assert mismatch["mismatch_element_count"] == 1
assert mismatch["max_abs_error"] == 1.0
def test_auc_pauc_matches_known_perfect_and_worst_rankings() -> None:
labels = [0, 0, 1, 1]
perfect_auc, perfect_pauc = validator.auc_pauc(labels, [0.1, 0.2, 0.8, 0.9], 0.1)
worst_auc, worst_pauc = validator.auc_pauc(labels, [0.9, 0.8, 0.2, 0.1], 0.1)
assert perfect_auc == pytest.approx(1.0)
assert perfect_pauc == pytest.approx(1.0)
assert worst_auc == pytest.approx(0.0)
# sklearn's standardized partial AUC maps random to 0.5 and worst-at-low-FPR
# below 0.5 rather than returning the raw partial area.
assert worst_pauc == pytest.approx((1.0 - 0.1) / (2.0 - 0.1))
def test_recompute_metrics_averages_four_machine_ids() -> None:
rows: list[dict[str, object]] = []
for machine_id in validator.MACHINE_IDS:
for label, score in ((0, 0.1), (0, 0.2), (1, 0.8), (1, 0.9)):
rows.append(
{
"machine_id": machine_id,
"label": label,
**{variant: score for variant in validator.VARIANTS},
}
)
metrics = validator.recompute_metrics(rows, max_fpr=0.1)
assert len(metrics) == 20
averages = [row for row in metrics if row["machine_id"] == "Average"]
assert len(averages) == 4
assert all(row["auc"] == pytest.approx(1.0) for row in averages)
assert all(row["pauc"] == pytest.approx(1.0) for row in averages)
def test_merged_quality_contract_and_metric_tamper_detection(tmp_path: Path) -> None:
score_rows: list[dict[str, object]] = []
for machine_id in validator.MACHINE_IDS:
for index, (label, score) in enumerate(((0, 0.1), (1, 0.9))):
score_rows.append(
{
"filename": f"{'normal' if label == 0 else 'anomaly'}_{machine_id}_{index}.wav",
"machine_id": machine_id,
"label": label,
"feature_vectors": 196,
"fp32_onnxruntime_score": score,
"fp32_compiled_score": score,
"public_quantized_onnxruntime_score": score,
"public_quantized_compiled_score": score,
}
)
score_path = tmp_path / "compiled_file_scores.csv"
_write_csv(score_path, list(score_rows[0]), score_rows)
loaded, paths = validator.load_quality_scores(tmp_path)
assert paths == [score_path]
assert len(loaded) == 8
metrics = validator.recompute_metrics(loaded, 0.1)
metric_path = tmp_path / "compiled_quality_metrics.csv"
_write_csv(
metric_path,
["variant", "machine_id", "auc", "pauc", "max_fpr"],
metrics,
)
assert validator.validate_metric_csv(metric_path, metrics)["status"] == "PASS"
tampered = list(metrics)
tampered[0] = {**tampered[0], "auc": 0.25}
_write_csv(metric_path, list(tampered[0]), tampered)
with pytest.raises(validator.ValidationContractError, match="metric mismatch"):
validator.validate_metric_csv(metric_path, metrics)
def test_variant_directory_layout_is_joined_by_filename(tmp_path: Path) -> None:
fields = [
"filename",
"machine_id",
"label",
"feature_vectors",
"onnxruntime_score",
"compiled_score",
]
fp32 = [{
"filename": "normal_id_01_a.wav", "machine_id": "id_01", "label": 0,
"feature_vectors": 196, "onnxruntime_score": 1.0, "compiled_score": 1.1,
}]
quantized = [{
"filename": "normal_id_01_a.wav", "machine_id": "id_01", "label": 0,
"feature_vectors": 196, "onnxruntime_score": 2.0, "compiled_score": 2.1,
}]
_write_csv(tmp_path / "fp32/file_scores.csv", fields, fp32)
_write_csv(tmp_path / "public_quantized/file_scores.csv", fields, quantized)
rows, paths = validator.load_quality_scores(tmp_path)
assert len(paths) == 2
assert rows == [{
"filename": "normal_id_01_a.wav",
"machine_id": "id_01",
"label": 0,
"feature_vectors": 196,
"fp32_onnxruntime": 1.0,
"fp32_compiled": 1.1,
"public_quantized_onnxruntime": 2.0,
"public_quantized_compiled": 2.1,
}]
def test_no_speed_policy_rejects_latency_claim_or_artifact(tmp_path: Path) -> None:
config = {"policy": {"latency_measured": False}}
(tmp_path / "summary.json").write_text(
json.dumps({"latency_measured": False, "measurement_kind": "NUMERICAL"})
)
assert validator.validate_no_speed_policy(tmp_path, config)["status"] == "PASS"
(tmp_path / "claim.json").write_text(json.dumps({"latency_measured": True}))
report = validator.validate_no_speed_policy(tmp_path, config)
assert report["status"] == "FAIL"
assert any("speed policy is not false" in value for value in report["violations"])
def test_quality_merge_manifest_pins_variant_inputs_and_outputs(tmp_path: Path) -> None:
result_dir = tmp_path / "result"
result_dir.mkdir()
merged_scores = result_dir / "compiled_file_scores.csv"
merged_metrics = result_dir / "compiled_quality_metrics.csv"
merged_score_rows = [{
"filename": "a.wav", "machine_id": "id_01", "label": 0,
"feature_vectors": 196,
"fp32_onnxruntime_score": 1.0, "fp32_compiled_score": 1.1,
"public_quantized_onnxruntime_score": 2.0,
"public_quantized_compiled_score": 2.1,
}]
_write_csv(merged_scores, list(merged_score_rows[0]), merged_score_rows)
merged_metrics.write_text("variant\nfp32_compiled\n")
artifact_hashes = {
"fp32": {
"onnx": {"sha256": "1" * 64},
"compiled_library": {"sha256": "2" * 64},
"source_tflite": {"sha256": "3" * 64},
},
"public_quantized": {
"onnx": {"sha256": "4" * 64},
"compiled_library": {"sha256": "5" * 64},
"source_tflite": {"sha256": "6" * 64},
},
}
inputs: dict[str, dict[str, object]] = {}
for variant in ("fp32", "public_quantized"):
directory = result_dir / variant
directory.mkdir()
variant_row = {
"filename": "a.wav", "machine_id": "id_01", "label": 0,
"feature_vectors": 196,
"onnxruntime_score": 1.0 if variant == "fp32" else 2.0,
"compiled_score": 1.1 if variant == "fp32" else 2.1,
}
_write_csv(directory / "file_scores.csv", list(variant_row), [variant_row])
(directory / "quality_metrics.csv").write_text("variant\ncompiled\n")
summary = {
"variant": variant,
"status": "PARTIAL",
"failure_code": None,
"measurement_status": "MEASURED",
"acceptance_status": "THRESHOLD_UNDEFINED",
"fidelity_status": "PASS",
"latency_measured": False,
"evaluation_fingerprint": f"fingerprint-{variant}",
"input_artifacts": {
"canonical_onnx": artifact_hashes[variant]["onnx"],
"compiled_shared_library": artifact_hashes[variant]["compiled_library"],
"source_tflite": artifact_hashes[variant]["source_tflite"],
},
}
summary_path = directory / "quality_summary.json"
summary_path.write_text(json.dumps(summary))
inputs[variant] = {
"directory": str(directory),
"summary_sha256": validator.sha256_file(summary_path),
"file_scores_sha256": validator.sha256_file(directory / "file_scores.csv"),
"quality_metrics_sha256": validator.sha256_file(directory / "quality_metrics.csv"),
"evaluation_fingerprint": f"fingerprint-{variant}",
}
manifest = {
"status": "PASS",
"measurement_status": "MEASURED",
"acceptance_status": "THRESHOLD_UNDEFINED",
"fidelity_status": "PASS",
"latency_measured": False,
"file_score_rows": 2459,
"quality_metric_rows": 20,
"inputs": inputs,
"outputs": {
"compiled_file_scores": {
"path": str(merged_scores),
"bytes": merged_scores.stat().st_size,
"sha256": validator.sha256_file(merged_scores),
},
"compiled_quality_metrics": {
"path": str(merged_metrics),
"bytes": merged_metrics.stat().st_size,
"sha256": validator.sha256_file(merged_metrics),
},
},
}
(result_dir / "compiled_merge_summary.json").write_text(json.dumps(manifest))
report = validator.validate_quality_manifest(
result_dir, [merged_scores], merged_metrics, artifact_hashes
)
assert report["status"] == "PASS"
assert report["output_checksums_match"] is True
merged_scores.write_text("filename\ntampered.wav\n")
with pytest.raises(validator.ValidationContractError, match="checksum mismatch"):
validator.validate_quality_manifest(
result_dir, [merged_scores], merged_metrics, artifact_hashes
)
def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None:
canonical = tmp_path / "canonical.csv"
rows = [
{
"filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav",
"machine_id": machine_id,
"label": index % 2,
"feature_vectors": 196,
**{variant: float(index) for variant in validator.VARIANTS},
}
for index, machine_id in enumerate(validator.MACHINE_IDS, start=1)
]
_write_csv(
canonical,
["filename", "machine_id", "label"],
[
{"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]}
for row in rows
],
)
config = {
"quality": {
"expected_test_files": 4,
"expected_total_feature_vectors": 784,
"expected_feature_vectors_per_file": 196,
}
}
report = validator.validate_quality_identity(rows, canonical, config)
assert report["canonical_filename_label_machine_id_match"] is True
tampered = [dict(row) for row in rows]
tampered[0]["label"] = 1 - int(tampered[0]["label"])
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
tampered = [dict(row) for row in rows]
tampered[0]["machine_id"], tampered[1]["machine_id"] = (
tampered[1]["machine_id"],
tampered[0]["machine_id"],
)
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None:
canonical = tmp_path / "canonical.csv"
rows = [
{
"filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav",
"machine_id": machine_id,
"label": index % 2,
"feature_vectors": 196,
**{variant: float(index) for variant in validator.VARIANTS},
}
for index, machine_id in enumerate(validator.MACHINE_IDS, start=1)
]
_write_csv(
canonical,
["filename", "machine_id", "label"],
[
{"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]}
for row in rows
],
)
config = {
"quality": {
"expected_test_files": 4,
"expected_total_feature_vectors": 784,
"expected_feature_vectors_per_file": 196,
}
}
report = validator.validate_quality_identity(rows, canonical, config)
assert report["canonical_filename_label_machine_id_match"] is True
tampered = [dict(row) for row in rows]
tampered[0]["label"] = 1 - int(tampered[0]["label"])
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
tampered = [dict(row) for row in rows]
tampered[0]["machine_id"], tampered[1]["machine_id"] = (
tampered[1]["machine_id"],
tampered[0]["machine_id"],
)
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None:
canonical = tmp_path / "canonical.csv"
rows = [
{
"filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav",
"machine_id": machine_id,
"label": index % 2,
"feature_vectors": 196,
**{variant: float(index) for variant in validator.VARIANTS},
}
for index, machine_id in enumerate(validator.MACHINE_IDS, start=1)
]
_write_csv(
canonical,
["filename", "machine_id", "label"],
[
{"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]}
for row in rows
],
)
config = {
"quality": {
"expected_test_files": 4,
"expected_total_feature_vectors": 784,
"expected_feature_vectors_per_file": 196,
}
}
report = validator.validate_quality_identity(rows, canonical, config)
assert report["canonical_filename_label_machine_id_match"] is True
tampered = [dict(row) for row in rows]
tampered[0]["label"] = 1 - int(tampered[0]["label"])
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
tampered = [dict(row) for row in rows]
tampered[0]["machine_id"], tampered[1]["machine_id"] = (
tampered[1]["machine_id"],
tampered[0]["machine_id"],
)
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None:
canonical = tmp_path / "canonical.csv"
rows = [
{
"filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav",
"machine_id": machine_id,
"label": index % 2,
"feature_vectors": 196,
**{variant: float(index) for variant in validator.VARIANTS},
}
for index, machine_id in enumerate(validator.MACHINE_IDS, start=1)
]
_write_csv(
canonical,
["filename", "machine_id", "label"],
[
{"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]}
for row in rows
],
)
config = {
"quality": {
"expected_test_files": 4,
"expected_total_feature_vectors": 784,
"expected_feature_vectors_per_file": 196,
}
}
report = validator.validate_quality_identity(rows, canonical, config)
assert report["canonical_filename_label_machine_id_match"] is True
tampered = [dict(row) for row in rows]
tampered[0]["label"] = 1 - int(tampered[0]["label"])
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
tampered = [dict(row) for row in rows]
tampered[0]["machine_id"], tampered[1]["machine_id"] = (
tampered[1]["machine_id"],
tampered[0]["machine_id"],
)
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None:
canonical = tmp_path / "canonical.csv"
rows = [
{
"filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav",
"machine_id": machine_id,
"label": index % 2,
"feature_vectors": 196,
**{variant: float(index) for variant in validator.VARIANTS},
}
for index, machine_id in enumerate(validator.MACHINE_IDS, start=1)
]
_write_csv(
canonical,
["filename", "machine_id", "label"],
[
{"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]}
for row in rows
],
)
config = {
"quality": {
"expected_test_files": 4,
"expected_total_feature_vectors": 784,
"expected_feature_vectors_per_file": 196,
}
}
report = validator.validate_quality_identity(rows, canonical, config)
assert report["canonical_filename_label_machine_id_match"] is True
tampered = [dict(row) for row in rows]
tampered[0]["label"] = 1 - int(tampered[0]["label"])
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
tampered = [dict(row) for row in rows]
tampered[0]["machine_id"], tampered[1]["machine_id"] = (
tampered[1]["machine_id"],
tampered[0]["machine_id"],
)
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None:
canonical = tmp_path / "canonical.csv"
rows = [
{
"filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav",
"machine_id": machine_id,
"label": index % 2,
"feature_vectors": 196,
**{variant: float(index) for variant in validator.VARIANTS},
}
for index, machine_id in enumerate(validator.MACHINE_IDS, start=1)
]
_write_csv(
canonical,
["filename", "machine_id", "label"],
[
{"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]}
for row in rows
],
)
config = {
"quality": {
"expected_test_files": 4,
"expected_total_feature_vectors": 784,
"expected_feature_vectors_per_file": 196,
}
}
report = validator.validate_quality_identity(rows, canonical, config)
assert report["canonical_filename_label_machine_id_match"] is True
tampered = [dict(row) for row in rows]
tampered[0]["label"] = 1 - int(tampered[0]["label"])
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
tampered = [dict(row) for row in rows]
tampered[0]["machine_id"], tampered[1]["machine_id"] = (
tampered[1]["machine_id"],
tampered[0]["machine_id"],
)
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None:
canonical = tmp_path / "canonical.csv"
rows = [
{
"filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav",
"machine_id": machine_id,
"label": index % 2,
"feature_vectors": 196,
**{variant: float(index) for variant in validator.VARIANTS},
}
for index, machine_id in enumerate(validator.MACHINE_IDS, start=1)
]
_write_csv(
canonical,
["filename", "machine_id", "label"],
[
{"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]}
for row in rows
],
)
config = {
"quality": {
"expected_test_files": 4,
"expected_total_feature_vectors": 784,
"expected_feature_vectors_per_file": 196,
}
}
report = validator.validate_quality_identity(rows, canonical, config)
assert report["canonical_filename_label_machine_id_match"] is True
tampered = [dict(row) for row in rows]
tampered[0]["label"] = 1 - int(tampered[0]["label"])
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
tampered = [dict(row) for row in rows]
tampered[0]["machine_id"], tampered[1]["machine_id"] = (
tampered[1]["machine_id"],
tampered[0]["machine_id"],
)
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
def test_quality_identity_rejects_label_or_machine_id_drift(tmp_path: Path) -> None:
canonical = tmp_path / "canonical.csv"
rows = [
{
"filename": f"{'normal' if index % 2 else 'anomaly'}_{machine_id}_00000000.wav",
"machine_id": machine_id,
"label": index % 2,
"feature_vectors": 196,
**{variant: float(index) for variant in validator.VARIANTS},
}
for index, machine_id in enumerate(validator.MACHINE_IDS, start=1)
]
_write_csv(
canonical,
["filename", "machine_id", "label"],
[
{"filename": row["filename"], "machine_id": row["machine_id"], "label": row["label"]}
for row in rows
],
)
config = {
"quality": {
"expected_test_files": 4,
"expected_total_feature_vectors": 784,
"expected_feature_vectors_per_file": 196,
}
}
report = validator.validate_quality_identity(rows, canonical, config)
assert report["canonical_filename_label_machine_id_match"] is True
tampered = [dict(row) for row in rows]
tampered[0]["label"] = 1 - int(tampered[0]["label"])
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
tampered = [dict(row) for row in rows]
tampered[0]["machine_id"], tampered[1]["machine_id"] = (
tampered[1]["machine_id"],
tampered[0]["machine_id"],
)
with pytest.raises(
validator.ValidationContractError,
match="dataset identity differs from canonical Q1",
):
validator.validate_quality_identity(tampered, canonical, config)
def test_fixed_fixture_only_directory_cannot_pass_official_q1(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(validator, "CONFIG_RELATIVE", Path("contract.json"))
artifact = tmp_path / "artifact.bin"
artifact.write_bytes(b"fixed")
artifact_contract = {
"path": "artifact.bin",
"sha256": validator.sha256_file(artifact),
}
contract = {
"model_id": "AD01",
"random_seed": 20260806,
"quality": {
"acceptance_threshold": None,
"acceptance_policy": "THRESHOLD_UNDEFINED",
"expected_test_files": 2459,
"expected_feature_vectors_per_file": 196,
"expected_total_feature_vectors": 481964,
"max_fpr": 0.1,
},
"policy": {"latency_measured": False},
"artifacts": {
variant: {
role: dict(artifact_contract)
for role in ("onnx", "compiled_library", "source_tflite")
}
for variant in ("fp32", "public_quantized")
},
}
(tmp_path / "contract.json").write_text(json.dumps(contract))
result_dir = tmp_path / "results"
result_dir.mkdir()
report = validator.validate(tmp_path, result_dir)
assert report["status"] == "FAIL"
assert report["official_dcase_q1_recalculation"]["measurement_status"] == "FAIL"
assert any(
row["check"] == "official_dcase_quality_complete" and row["status"] == "FAIL"
for row in report["checks"]
)