mkbk
108 lines
4.3 KiB
Python
108 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from owner_voice_pet.config import AppConfig
|
|
from owner_voice_pet.conversation import ConversationContext
|
|
from owner_voice_pet.llm import MockLlmProvider
|
|
from owner_voice_pet.models import AudioFrame, AudioSegment, Transcript
|
|
from owner_voice_pet.runtime import LiveVoiceRuntime
|
|
from owner_voice_pet.transport import MemoryAudioTransport
|
|
from owner_voice_pet.tts import SineTtsProvider
|
|
from owner_voice_pet.vad import EnergyVadProvider, VadRecorder
|
|
|
|
|
|
def segment_frames(start_id: int, start_ms: int) -> list[AudioFrame]:
|
|
return [
|
|
AudioFrame(b"\xff\x7f", 16000, 1, start_ms, start_id, {"duration_ms": 20, "speech": True}),
|
|
AudioFrame(b"\xff\x7f", 16000, 1, start_ms + 20, start_id + 1, {"duration_ms": 20, "speech": True}),
|
|
AudioFrame(b"\x00\x00", 16000, 1, start_ms + 40, start_id + 2, {"duration_ms": 20, "speech": False}),
|
|
AudioFrame(b"\x00\x00", 16000, 1, start_ms + 60, start_id + 3, {"duration_ms": 20, "speech": False}),
|
|
]
|
|
|
|
|
|
class QueueSttProvider:
|
|
def __init__(self, texts: list[str]) -> None:
|
|
self.texts = list(texts)
|
|
self.calls: list[AudioSegment] = []
|
|
self.loaded = False
|
|
|
|
def load(self) -> None:
|
|
self.loaded = True
|
|
|
|
def transcribe(self, segment: AudioSegment) -> Transcript:
|
|
self.calls.append(segment)
|
|
text = self.texts.pop(0)
|
|
return Transcript(text, "zh", 1.0, segment.duration_ms, "queue-stt")
|
|
|
|
|
|
class RecordingReporter:
|
|
def __init__(self) -> None:
|
|
self.statuses: list[str] = []
|
|
self.errors: list[str] = []
|
|
|
|
def status(self, state: str, message: str, *, turn_id: int | None = None) -> None:
|
|
self.statuses.append(message)
|
|
|
|
def error(self, stage: str, code: str, message: str, *, turn_id: int | None = None) -> None:
|
|
self.errors.append(f"{stage}:{code}:{message}")
|
|
|
|
|
|
def make_runtime(texts: list[str], context: ConversationContext | None = None) -> tuple[LiveVoiceRuntime, QueueSttProvider, MockLlmProvider, MemoryAudioTransport, RecordingReporter]:
|
|
frames = []
|
|
for idx in range(4):
|
|
frames.extend(segment_frames(idx * 4, idx * 80))
|
|
transport = MemoryAudioTransport(frames)
|
|
stt = QueueSttProvider(texts)
|
|
llm = MockLlmProvider(["这是答复。"])
|
|
tts = SineTtsProvider()
|
|
reporter = RecordingReporter()
|
|
runtime = LiveVoiceRuntime(
|
|
config=AppConfig(llm_api_key="secret", speech_provider="cloud"),
|
|
transport=transport,
|
|
vad_recorder=VadRecorder(EnergyVadProvider(), min_duration_ms=40, end_silence_ms=40),
|
|
stt=stt,
|
|
llm=llm,
|
|
tts=tts,
|
|
context=context or ConversationContext(),
|
|
reporter=reporter,
|
|
)
|
|
return runtime, stt, llm, transport, reporter
|
|
|
|
|
|
class LiveRuntimeTests(unittest.TestCase):
|
|
def test_repeated_runtime_runs_two_turns_and_returns_to_standby(self) -> None:
|
|
runtime, stt, llm, transport, reporter = make_runtime(
|
|
["小杰小杰", "第一问", "小杰小杰", "第二问"]
|
|
)
|
|
summary = runtime.run(max_turns=2)
|
|
|
|
self.assertEqual(summary.completed_turns, 2)
|
|
self.assertEqual(len(stt.calls), 4)
|
|
self.assertEqual(len(llm.calls), 2)
|
|
self.assertEqual(len(transport.played_segments), 2)
|
|
self.assertIn("恢复待机：可继续唤醒", reporter.statuses[-1])
|
|
|
|
def test_temporary_context_is_sent_to_second_llm_call(self) -> None:
|
|
runtime, _, llm, _, _ = make_runtime(["小杰小杰", "第一问", "小杰小杰", "第二问"])
|
|
runtime.run(max_turns=2)
|
|
|
|
second_call_text = [message.content for message in llm.calls[1]]
|
|
self.assertIn("第一问", second_call_text)
|
|
self.assertIn("这是答复。", second_call_text)
|
|
self.assertEqual(second_call_text[-1], "第二问")
|
|
|
|
def test_new_runtime_context_starts_empty(self) -> None:
|
|
first_context = ConversationContext()
|
|
first_runtime, _, _, _, _ = make_runtime(["小杰小杰", "第一问"], context=first_context)
|
|
first_runtime.run(max_turns=1)
|
|
self.assertGreater(len(first_context.messages()), 0)
|
|
|
|
second_context = ConversationContext()
|
|
make_runtime(["小杰小杰", "第二问"], context=second_context)
|
|
self.assertEqual(second_context.messages(), ())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|