| from __future__ import annotations |
|
|
| import importlib.util |
| import sys |
| import unittest |
| from pathlib import Path |
|
|
| import numpy as np |
|
|
| from turn_detection.runtime.features import ( |
| FrontendConfig as CanonicalFrontendConfig, |
| ) |
| from turn_detection.runtime.features import ( |
| log_mel_spectrogram as canonical_log_mel_spectrogram, |
| ) |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| MODULE_PATH = ROOT / "deployment" / "kaggle" / "turn_detector.py" |
| SPEC = importlib.util.spec_from_file_location("kaggle_turn_detector", MODULE_PATH) |
| if SPEC is None or SPEC.loader is None: |
| raise RuntimeError(f"cannot load {MODULE_PATH}") |
| KAGGLE_RUNTIME = importlib.util.module_from_spec(SPEC) |
| sys.modules[SPEC.name] = KAGGLE_RUNTIME |
| SPEC.loader.exec_module(KAGGLE_RUNTIME) |
|
|
|
|
| class KaggleRuntimeParityTest(unittest.TestCase): |
| def test_flat_bundle_frontend_matches_canonical_runtime(self) -> None: |
| canonical_config = CanonicalFrontendConfig(max_seconds=4.0) |
| kaggle_config = KAGGLE_RUNTIME.FrontendConfig(**canonical_config.to_dict()) |
| for sample_rate in (8_000, 16_000, 48_000): |
| timeline = np.arange(round(sample_rate * 1.37), dtype=np.float32) / sample_rate |
| signal = 0.13 * np.sin(2.0 * np.pi * 337.0 * timeline) |
| stereo = np.stack((signal, signal * 0.7), axis=1) |
| expected_features, expected_mask = canonical_log_mel_spectrogram( |
| stereo, |
| sample_rate, |
| canonical_config, |
| ) |
| actual_features, actual_mask = KAGGLE_RUNTIME.log_mel_spectrogram( |
| stereo, |
| sample_rate, |
| kaggle_config, |
| ) |
| np.testing.assert_array_equal(actual_features, expected_features) |
| np.testing.assert_array_equal(actual_mask, expected_mask) |
|
|
| def test_controller_threshold_relaxation_matches_serialized_policy(self) -> None: |
| detector = object.__new__(KAGGLE_RUNTIME.TurnDetector) |
| detector.controller = KAGGLE_RUNTIME.ControllerConfig( |
| endpoint_threshold=0.74, |
| long_pause_threshold=0.56, |
| min_silence_ms=200.0, |
| relax_after_ms=800.0, |
| max_silence_ms=1_800.0, |
| required_confirmations=1, |
| ) |
| self.assertEqual(detector.threshold_for_silence(300.0), 0.74) |
| self.assertAlmostEqual(detector.threshold_for_silence(1_300.0), 0.65) |
| self.assertEqual(detector.threshold_for_silence(2_000.0), 0.56) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
