tests/test_kaggle_runtime.py

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