File size: 5,092 Bytes
ed3aeeb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 | from __future__ import annotations
from dataclasses import dataclass
import numpy as np
from scripts.stages.evaluate_ad01_compiled_mlir import (
compiled_metric_rows,
evaluate_feature_vectors,
merge_file_score_rows,
merge_metric_rows,
)
@dataclass
class _Input:
name: str = "input_1"
class _Reference:
def __init__(self) -> None:
self.input_shapes: list[tuple[int, ...]] = []
def get_inputs(self) -> list[_Input]:
return [_Input()]
def run(self, _outputs: object, feeds: dict[str, np.ndarray]) -> list[np.ndarray]:
self.input_shapes.append(feeds["input_1"].shape)
return [feeds["input_1"] * np.float32(0.5)]
class _Compiled:
def __init__(self) -> None:
self.input_shapes: list[tuple[int, ...]] = []
def run(self, inputs: list[np.ndarray]) -> list[np.ndarray]:
self.input_shapes.append(inputs[0].shape)
return [inputs[0] * np.float32(0.5)]
class _ChangedCompiled(_Compiled):
def run(self, inputs: list[np.ndarray]) -> list[np.ndarray]:
value = super().run(inputs)[0].copy()
value[0, 0] += np.float32(1.0)
return [value]
def test_compiled_rows_reuse_official_metric_and_preserve_fidelity() -> None:
# The official path invokes every one of a file's 196 feature vectors with
# batch one. A [196, 640] compiler batch is deliberately not substituted:
# it has a distinct FP32 numerical fidelity result at the pinned tolerance.
vectors = np.arange(196 * 640, dtype=np.float64).reshape(196, 640) / 1000.0
reference = _Reference()
compiled = _Compiled()
evaluated = evaluate_feature_vectors(
vectors,
filename="normal_id_01_00000000.wav",
variant="fp32",
reference=reference,
compiled=compiled,
quantization=None,
fp32_atol=1e-5,
fp32_rtol=1e-5,
)
assert evaluated["fidelity_rows"] == 196
assert evaluated["fidelity_matching_rows"] == 196
assert evaluated["fidelity_mismatching_rows"] == 0
assert evaluated["fidelity_bitwise_equal_rows"] == 196
assert evaluated["fidelity_total_elements"] == 196 * 640
assert len(evaluated["fidelity_digest"]) == 64
assert evaluated["fidelity_mismatches"] == []
assert reference.input_shapes == [(1, 640)] * 196
assert compiled.input_shapes == [(1, 640)] * 196
assert evaluated["compiled_score"] == evaluated["onnxruntime_score"]
mismatch = evaluate_feature_vectors(
vectors[:1],
filename="anomaly_id_01_00000000.wav",
variant="fp32",
reference=_Reference(),
compiled=_ChangedCompiled(),
quantization=None,
fp32_atol=1e-5,
fp32_rtol=1e-5,
)
assert mismatch["fidelity_rows"] == 1
assert mismatch["fidelity_matching_rows"] == 0
assert mismatch["fidelity_mismatching_rows"] == 1
assert len(mismatch["fidelity_mismatches"]) == 1
assert mismatch["fidelity_mismatches"][0]["status"] == "FAIL"
file_rows = [
{"machine_id": "id_01", "label": 0, "compiled_score": 0.1},
{"machine_id": "id_01", "label": 0, "compiled_score": 0.2},
{"machine_id": "id_01", "label": 1, "compiled_score": 0.8},
{"machine_id": "id_01", "label": 1, "compiled_score": 0.9},
]
metrics = compiled_metric_rows(
file_rows, "compiled_score", "fp32", max_fpr=0.1
)
average = next(row for row in metrics if row["machine_id"] == "Average")
assert average["variant"] == "fp32"
assert average["auc"] == 1.0
assert average["pauc"] == 1.0
fp32_scores = [{
"filename": "a.wav", "machine_id": "id_01", "label": "0",
"feature_vectors": "196", "onnxruntime_score": "0.1",
"compiled_score": "0.2",
}]
quantized_scores = [{
"filename": "a.wav", "machine_id": "id_01", "label": "0",
"feature_vectors": "196", "onnxruntime_score": "0.3",
"compiled_score": "0.4",
}]
merged_scores = merge_file_score_rows(fp32_scores, quantized_scores)
assert merged_scores == [{
"filename": "a.wav", "machine_id": "id_01", "label": "0",
"feature_vectors": "196", "fp32_onnxruntime_score": "0.1",
"fp32_compiled_score": "0.2",
"public_quantized_onnxruntime_score": "0.3",
"public_quantized_compiled_score": "0.4",
}]
metric_inputs = []
for runtime in ("onnxruntime_reference", "compiled_mlir"):
for machine_id in ("id_01", "id_02", "id_03", "id_04", "Average"):
metric_inputs.append({
"runtime": runtime, "variant": "fp32", "machine_id": machine_id,
"auc": "0.9", "pauc": "0.8", "max_fpr": "0.1",
})
quantized_metric_inputs = [
{**row, "variant": "public_quantized"} for row in metric_inputs
]
merged_metrics = merge_metric_rows(metric_inputs, quantized_metric_inputs)
assert len(merged_metrics) == 20
assert {row["variant"] for row in merged_metrics} == {
"fp32_onnxruntime", "fp32_compiled",
"public_quantized_onnxruntime", "public_quantized_compiled",
}
|