File size: 3,086 Bytes
35d483e | 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 | from __future__ import annotations
import unittest
from turn_detection.runtime import ControllerConfig, Prediction, TurnController, TurnState
class TurnControllerTest(unittest.TestCase):
def setUp(self) -> None:
self.controller = TurnController(
ControllerConfig(
endpoint_threshold=0.7,
long_pause_threshold=0.4,
min_silence_ms=200,
relax_after_ms=800,
max_silence_ms=1800,
required_confirmations=2,
)
)
def test_minimum_silence_always_holds(self) -> None:
decision = self.controller.evaluate_pause(0.99, silence_ms=199)
self.assertEqual(decision.state, TurnState.HOLD)
self.assertEqual(decision.reason, "minimum_silence_not_reached")
def test_endpoint_requires_configured_confirmations(self) -> None:
first = self.controller.evaluate_pause(0.9, silence_ms=300, timestamp_ms=300)
second = self.controller.evaluate_pause(0.9, silence_ms=500, timestamp_ms=500)
self.assertEqual(first.state, TurnState.HOLD)
self.assertEqual(second.state, TurnState.END)
self.assertTrue(second.emit_response)
def test_endpoint_latches_without_duplicate_response(self) -> None:
self.controller.evaluate_pause(0.9, 300, timestamp_ms=300)
endpoint = self.controller.evaluate_pause(0.9, 500, timestamp_ms=500)
repeated = self.controller.evaluate_pause(0.01, 700, timestamp_ms=700)
self.assertTrue(endpoint.emit_response)
self.assertEqual(repeated.state, TurnState.END)
self.assertEqual(repeated.reason, "endpoint_latched")
self.assertFalse(repeated.emit_response)
def test_new_speech_resets_endpoint(self) -> None:
self.controller.evaluate_pause(0.9, 300)
self.controller.evaluate_pause(0.9, 500)
decision = self.controller.observe_speech(600)
self.assertEqual(decision.state, TurnState.SPEAKING)
def test_max_timeout_is_bounded(self) -> None:
decision = self.controller.evaluate_pause(0.01, silence_ms=1800)
self.assertEqual(decision.state, TurnState.END)
self.assertEqual(decision.reason, "maximum_timeout")
self.assertTrue(decision.emit_response)
def test_threshold_relaxes_linearly(self) -> None:
self.assertAlmostEqual(self.controller.threshold_for_silence(800), 0.7)
self.assertAlmostEqual(self.controller.threshold_for_silence(1300), 0.55)
self.assertAlmostEqual(self.controller.threshold_for_silence(1800), 0.4)
def test_prediction_validation(self) -> None:
with self.assertRaises(ValueError):
Prediction(endpoint_probability=1.1)
with self.assertRaises(ValueError):
self.controller.evaluate_pause(-0.1, 300)
def test_timestamps_must_be_monotonic(self) -> None:
self.controller.observe_speech(100)
with self.assertRaises(ValueError):
self.controller.evaluate_pause(0.5, 300, timestamp_ms=99)
if __name__ == "__main__":
unittest.main()
|