File size: 2,566 Bytes
0348402 | 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 | 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: # pragma: no cover - import machinery failure
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()
|