mkbk
101 lines
4.6 KiB
Python
101 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from owner_voice_pet.models import AudioFrame, AudioSegment, Message, PlaybackResult, ReplyDelta, TransportHealth
|
|
from owner_voice_pet.real_live_check import (
|
|
FixtureLiveAudioTransport,
|
|
RecordingLlmProvider,
|
|
TimingRecorder,
|
|
_stage_timings_from_events,
|
|
)
|
|
|
|
|
|
def frame(idx: int) -> AudioFrame:
|
|
return AudioFrame(b"\x00\x00", 16000, 1, idx * 20, idx, {"duration_ms": 20})
|
|
|
|
|
|
class FakeOutputTransport:
|
|
def __init__(self) -> None:
|
|
self.played: list[AudioSegment] = []
|
|
|
|
def play_pcm(self, segment: AudioSegment, interrupt: bool = False) -> PlaybackResult:
|
|
self.played.append(segment)
|
|
return PlaybackResult(True, segment.duration_ms)
|
|
|
|
def health(self) -> TransportHealth:
|
|
return TransportHealth(True, True, "fake")
|
|
|
|
|
|
class RealLiveCheckTests(unittest.TestCase):
|
|
def test_fixture_transport_flush_preserves_generated_future_frames(self) -> None:
|
|
transport = FixtureLiveAudioTransport([frame(1), frame(2)], play_audio=False)
|
|
transport.start_input()
|
|
|
|
self.assertEqual(transport.flush_input(), 0)
|
|
self.assertEqual(transport.flush_count, 1)
|
|
self.assertEqual([item.frame_id for item in transport.read_frames(10)], [1])
|
|
|
|
def test_fixture_transport_can_delegate_real_playback(self) -> None:
|
|
output = FakeOutputTransport()
|
|
transport = FixtureLiveAudioTransport([], play_audio=True, output_transport=output)
|
|
segment = AudioSegment(b"\x00\x00", 16000, 1, 0, 20)
|
|
|
|
result = transport.play_pcm(segment)
|
|
|
|
self.assertTrue(result.played)
|
|
self.assertEqual(output.played, [segment])
|
|
self.assertEqual(transport.played_segments, [segment])
|
|
|
|
def test_stage_timings_are_built_from_pipeline_events(self) -> None:
|
|
events = [
|
|
{"type": "wake_listening", "turn": 1, "at": "t0", "offset_ms": 0, "message": ""},
|
|
{"type": "wake_detected", "turn": 1, "at": "t1", "offset_ms": 120, "message": ""},
|
|
{"type": "ack_started", "turn": 1, "at": "t2", "offset_ms": 130, "message": ""},
|
|
{"type": "question_prompt", "turn": 1, "at": "t3", "offset_ms": 250, "message": ""},
|
|
{"type": "capture_started", "turn": 1, "at": "t4", "offset_ms": 260, "message": ""},
|
|
{"type": "speech_started", "turn": 1, "at": "t5", "offset_ms": 300, "message": ""},
|
|
{"type": "speech_ended", "turn": 1, "at": "t6", "offset_ms": 700, "message": ""},
|
|
{"type": "stt_started", "turn": 1, "at": "t7", "offset_ms": 720, "message": ""},
|
|
{"type": "transcript_final", "turn": 1, "at": "t8", "offset_ms": 900, "message": ""},
|
|
{"type": "llm_started", "turn": 1, "at": "t9", "offset_ms": 910, "message": ""},
|
|
{"type": "tts_started", "turn": 1, "at": "t10", "offset_ms": 1300, "message": ""},
|
|
{"type": "playback_finished", "turn": 1, "at": "t11", "offset_ms": 1500, "message": ""},
|
|
{"type": "standby_resumed", "turn": 1, "at": "t12", "offset_ms": 1520, "message": ""},
|
|
]
|
|
|
|
timings = _stage_timings_from_events(events)
|
|
by_name = {str(item["name"]): item for item in timings if "segment_index" not in item}
|
|
|
|
self.assertEqual(by_name["wake_wait"]["duration_ms"], 120)
|
|
self.assertEqual(by_name["acknowledgement"]["duration_ms"], 120)
|
|
self.assertEqual(by_name["capture"]["duration_ms"], 400)
|
|
self.assertEqual(by_name["final_stt"]["duration_ms"], 180)
|
|
self.assertEqual(by_name["llm_to_first_tts"]["duration_ms"], 390)
|
|
self.assertEqual(by_name["turn_total"]["duration_ms"], 1520)
|
|
tts = [item for item in timings if item["name"] == "tts_playback"]
|
|
self.assertEqual(tts[0]["duration_ms"], 200)
|
|
|
|
def test_recording_llm_provider_records_sent_time_and_response_duration(self) -> None:
|
|
class FakeLlm:
|
|
def stream_reply(self, messages):
|
|
yield ReplyDelta("你好", finish_reason="stop")
|
|
|
|
timing = TimingRecorder()
|
|
provider = RecordingLlmProvider(FakeLlm(), timing) # type: ignore[arg-type]
|
|
messages = [Message("user", "现在几点", 0.0)]
|
|
|
|
deltas = list(provider.stream_reply(messages))
|
|
|
|
self.assertEqual(deltas[0].text_delta, "你好")
|
|
self.assertEqual(len(provider.request_timings), 1)
|
|
request = provider.request_timings[0]
|
|
self.assertIn("sent_at", request)
|
|
self.assertIn("finished_at", request)
|
|
self.assertIsInstance(request["duration_ms"], int)
|
|
self.assertEqual(request["last_user_preview"], "现在几点")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|