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