80 lines
3.1 KiB
Python
80 lines
3.1 KiB
Python
"""Verify replay timing assumptions against Pipecat's actual buffer logic."""
|
|
|
|
import asyncio
|
|
import time
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
from pipecat.audio.turn.base_turn_analyzer import EndOfTurnState
|
|
from pipecat.audio.turn.smart_turn.base_smart_turn import BaseSmartTurn, SmartTurnParams
|
|
|
|
import evaluate_work_data as evaluation
|
|
|
|
|
|
class RecordingTurn(BaseSmartTurn):
|
|
prediction = 0
|
|
|
|
def _predict_endpoint(self, audio):
|
|
self.last_audio = audio.copy()
|
|
return {"prediction": self.prediction, "probability": 0.9 if self.prediction else 0.1}
|
|
|
|
|
|
class ReplayTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.clock = evaluation.MediaClock()
|
|
self.time_patch = patch.object(evaluation.base_module, "time", SimpleNamespace(
|
|
monotonic=self.clock.monotonic, perf_counter=time.perf_counter))
|
|
self.time_patch.start()
|
|
self.turn = RecordingTurn(sample_rate=8000, params=SmartTurnParams())
|
|
self.turn.set_sample_rate(8000)
|
|
self.turn.update_vad_start_secs(0.2)
|
|
self.chunk = np.full(256, 1234, dtype=np.int16).tobytes()
|
|
|
|
def tearDown(self):
|
|
if self.turn._executor:
|
|
self.turn._executor.shutdown(wait=True)
|
|
self.time_patch.stop()
|
|
|
|
def feed(self, count, speaking):
|
|
for _ in range(count):
|
|
self.clock.seconds += 0.032
|
|
state = self.turn.append_audio(self.chunk, speaking)
|
|
return state
|
|
|
|
def test_incomplete_keeps_first_start_and_resumed_speech_resets_silence(self):
|
|
self.feed(10, False)
|
|
self.feed(20, True)
|
|
first_start = self.turn._speech_start_time
|
|
self.feed(60, False) # 1.92 s pause, below the timeout.
|
|
state, _ = asyncio.run(self.turn.analyze_end_of_turn())
|
|
self.assertEqual(state, EndOfTurnState.INCOMPLETE)
|
|
preserved_count = len(self.turn._audio_buffer)
|
|
self.feed(1, True)
|
|
self.assertEqual(self.turn._speech_start_time, first_start)
|
|
self.assertEqual(len(self.turn._audio_buffer), preserved_count + 1)
|
|
self.assertEqual(self.turn._silence_ms, 0)
|
|
|
|
def test_model_only_gets_last_eight_seconds_and_probe_does_not_clear(self):
|
|
self.feed(320, True) # 10.24 s accumulated speech.
|
|
before_start = self.turn._speech_start_time
|
|
before_count = len(self.turn._audio_buffer)
|
|
self.turn.prediction = 1
|
|
state, _ = self.turn._process_speech_segment(self.turn._audio_buffer)
|
|
self.assertEqual(state, EndOfTurnState.COMPLETE)
|
|
self.assertEqual(len(self.turn.last_audio), 8 * 8000)
|
|
self.assertEqual(self.turn._speech_start_time, before_start)
|
|
self.assertEqual(len(self.turn._audio_buffer), before_count)
|
|
|
|
def test_three_second_silence_timeout_clears_old_origin(self):
|
|
self.feed(20, True)
|
|
state = self.feed(94, False) # 3.008 s at 32 ms frame granularity.
|
|
self.assertEqual(state, EndOfTurnState.COMPLETE)
|
|
self.assertFalse(self.turn.speech_triggered)
|
|
self.assertEqual(self.turn._speech_start_time, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|