File size: 3,360 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 | from __future__ import annotations
import ast
import importlib.util
import json
from pathlib import Path
import numpy as np
ROOT = Path(__file__).resolve().parents[1]
EVALUATOR = ROOT / "scripts/stages/evaluate_sp02_mlcommons_streaming.py"
CONFIG = (
ROOT
/ "configs/evaluation/speech_kws/SP02_mlcommons_streaming_host_quality_eval.json"
)
def load_evaluator():
spec = importlib.util.spec_from_file_location("evaluate_sp02", EVALUATOR)
assert spec and spec.loader
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def test_detection_alignment_matches_pinned_training_padding() -> None:
module = load_evaluator()
signal = np.asarray([1, 0, 1], dtype=np.int8)
times = module.detection_timestamps_seconds(
signal,
total_samples=5 * 512,
sample_rate_hz=16000,
stride_samples=512,
)
assert times.tolist() == [0.064, 0.128]
def test_readme_one_second_semantics_and_debounce() -> None:
module = load_evaluator()
truth = [[10.0, 11.0]]
result = module.count_readme_spec(
np.asarray([11.75, 13.0, 13.5, 14.6]), truth, post_end_seconds=1.0
)
assert result["true_positives"] == 1
assert result["false_negatives"] == 0
assert result["false_positives"] == 2
assert result["false_positive_seconds"] == [13.0, 14.6]
def test_reporter_quality_schema_is_numeric() -> None:
module = load_evaluator()
protocols = {
"training_evaluator_0p5s": {
"true_positives": 40,
"false_positives": 4,
"false_negatives": 10,
"acceptance_status": "FAIL",
}
}
payload = module.reporter_quality_schema(
"keras_h5",
37470,
{
"raw_positive_frames": 123,
"protocols": protocols,
},
)
assert payload["metric_name"] == "max_false_positives_false_negatives"
assert payload["metric_value"] == 10
assert isinstance(payload["metric_value"], int)
assert payload["threshold"] == 8
def test_evaluator_has_no_prohibited_model_mutation_calls() -> None:
tree = ast.parse(EVALUATOR.read_text())
prohibited = []
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
if isinstance(node.func, ast.Attribute) and node.func.attr in {
"set_weights",
"fit",
"save_weights",
}:
prohibited.append(node.func.attr)
if isinstance(node.func, ast.Name) and node.func.id in {"get_model", "clone_model"}:
prohibited.append(node.func.id)
assert prohibited == []
def test_config_declares_three_separate_protocols_and_no_prohibited_work() -> None:
config = json.loads(CONFIG.read_text())
protocols = config["detector"]["protocols"]
assert set(protocols) == {
"training_evaluator_0p5s",
"runner_readme_spec_1p0s",
"runner_code_observed_2p0s",
}
assert protocols["training_evaluator_0p5s"]["result_id"] == "PINNED_TRAINING_HOST_SOURCE_0P5"
assert protocols["runner_readme_spec_1p0s"]["result_id"] == "PINNED_RUNNER_README_1P0"
assert protocols["runner_code_observed_2p0s"]["result_id"] == "PINNED_RUNNER_CODE_OBSERVED_2P0"
assert all(value is False for value in config["prohibited_operations"].values())
|