515 lines
19 KiB
Python
515 lines
19 KiB
Python
import asyncio
|
|
import threading
|
|
import time
|
|
import uuid
|
|
|
|
from services.ai_voice_runtime_service.audiosocket import (
|
|
AUDIO_SOCKET_PACKET_PCM16,
|
|
AUDIO_SOCKET_PACKET_UUID,
|
|
EnergyVAD,
|
|
encode_audio_packet,
|
|
encode_packet,
|
|
normalize_media_uuid,
|
|
read_packet,
|
|
resample_pcm16le,
|
|
)
|
|
from services.ai_voice_runtime_service.media_runtime import AudioSocketMediaRuntime, MediaActor, MediaRegistration
|
|
from services.ai_voice_runtime_service.providers.asr import ASRProvider, ASRTranscription
|
|
from services.ai_voice_runtime_service.providers.tts import TTSProvider, TTSSynthesis
|
|
from services.shared.models import VoiceAITurnDecisionOut
|
|
|
|
|
|
class _StubASRProvider(ASRProvider):
|
|
name = "stub-asr"
|
|
|
|
def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
|
|
assert audio_bytes
|
|
return ASRTranscription(text="hello", language=language_hint or "ru", confidence=0.9)
|
|
|
|
|
|
class _StubTTSProvider(TTSProvider):
|
|
name = "stub-tts"
|
|
|
|
def synthesize(self, text: str, *, language: str | None = None) -> TTSSynthesis:
|
|
del language
|
|
assert text
|
|
return TTSSynthesis(
|
|
text=text,
|
|
audio_bytes=(b"\x10\x00" * 960),
|
|
sample_rate_hz=24000,
|
|
)
|
|
|
|
|
|
def test_resample_pcm16le_downsamples_to_8khz():
|
|
source = b"\x20\x00" * 2400
|
|
converted = resample_pcm16le(source, input_rate_hz=24000, output_rate_hz=8000)
|
|
assert converted
|
|
assert len(converted) < len(source)
|
|
|
|
|
|
def test_normalize_media_uuid_accepts_text_bytes():
|
|
media_uuid = str(uuid.uuid4())
|
|
assert normalize_media_uuid(media_uuid.encode("utf-8")) == media_uuid
|
|
|
|
|
|
def test_energy_vad_emits_utterance_after_trailing_silence():
|
|
vad = EnergyVAD(frame_ms=20, min_speech_ms=40, trailing_silence_ms=40, max_turn_ms=400)
|
|
speech_frame = (1000).to_bytes(2, "little", signed=True) * 160
|
|
silence_frame = b"\x00\x00" * 160
|
|
|
|
first = vad.feed(speech_frame)
|
|
second = vad.feed(speech_frame)
|
|
third = vad.feed(silence_frame)
|
|
fourth = vad.feed(silence_frame)
|
|
|
|
assert first.speech_started is False
|
|
assert second.speech_started is True
|
|
assert third.utterance_pcm is None
|
|
assert fourth.utterance_pcm is not None
|
|
|
|
|
|
def test_media_runtime_streams_greeting_and_turn():
|
|
registrations: dict[str, MediaRegistration] = {}
|
|
states: list[tuple[str, str]] = []
|
|
delivered: list[tuple[str, str, bool]] = []
|
|
turns: list[tuple[str, str, bool]] = []
|
|
touch_calls: list[str] = []
|
|
handoffs: list[str] = []
|
|
media_ended: list[tuple[str, str]] = []
|
|
errors: list[tuple[str, str]] = []
|
|
|
|
media_uuid = str(uuid.uuid4())
|
|
registrations[media_uuid] = MediaRegistration(
|
|
voice_session_id="avs_media_runtime",
|
|
call_id="call_media_runtime",
|
|
interaction_id="int_media_runtime",
|
|
ai_session_id="ais_media_runtime",
|
|
language="ru",
|
|
media_uuid=media_uuid,
|
|
)
|
|
|
|
runtime = AudioSocketMediaRuntime(
|
|
enabled=True,
|
|
host="127.0.0.1",
|
|
port=0,
|
|
frame_ms=20,
|
|
idle_timeout_seconds=2.0,
|
|
registration_wait_timeout_seconds=0.5,
|
|
min_speech_ms=40,
|
|
trailing_silence_ms=40,
|
|
max_turn_ms=400,
|
|
asr_provider=_StubASRProvider(),
|
|
tts_provider=_StubTTSProvider(),
|
|
load_registration_by_media_uuid=lambda value: registrations.get(value),
|
|
mark_media_connected=lambda session_id, value: None,
|
|
mark_media_ended=lambda session_id, reason: media_ended.append((session_id, reason)),
|
|
touch_media_frame=lambda session_id: touch_calls.append(session_id),
|
|
set_state=lambda session_id, state, handoff_reason, metadata: states.append((session_id, state)),
|
|
get_pending_greeting=lambda session_id: "greeting" if session_id == "avs_media_runtime" else None,
|
|
mark_reply_delivered=lambda session_id, text, is_greeting: delivered.append((session_id, text, is_greeting)),
|
|
process_turn=lambda session_id, transcript_text, language, barge_in, metadata: (
|
|
turns.append((session_id, transcript_text, barge_in))
|
|
or VoiceAITurnDecisionOut(
|
|
language=language or "ru",
|
|
intent="answer",
|
|
reply_text="reply",
|
|
confidence=0.9,
|
|
needs_handoff=False,
|
|
handoff_reason=None,
|
|
case_action="keep_open",
|
|
kb_refs=[],
|
|
summary_text="reply ready",
|
|
model="stub-voice",
|
|
latency_ms=1,
|
|
status="active",
|
|
)
|
|
),
|
|
request_handoff=lambda session_id, customer_request_text, decision: handoffs.append(session_id),
|
|
handle_media_error=lambda session_id, message, metadata: errors.append((session_id, message)),
|
|
)
|
|
|
|
async def _scenario() -> None:
|
|
await runtime.start()
|
|
port = runtime._server.sockets[0].getsockname()[1]
|
|
|
|
reader, writer = await asyncio.open_connection("127.0.0.1", port)
|
|
writer.write(encode_packet(AUDIO_SOCKET_PACKET_UUID, uuid.UUID(media_uuid).bytes))
|
|
await writer.drain()
|
|
|
|
packet_type, _ = await read_packet(reader, timeout=2.0)
|
|
assert packet_type == AUDIO_SOCKET_PACKET_PCM16
|
|
await asyncio.sleep(0.15)
|
|
|
|
speech_frame = (1000).to_bytes(2, "little", signed=True) * 160
|
|
silence_frame = b"\x00\x00" * 160
|
|
for _ in range(2):
|
|
writer.write(encode_audio_packet(speech_frame))
|
|
for _ in range(2):
|
|
writer.write(encode_audio_packet(silence_frame))
|
|
await writer.drain()
|
|
|
|
packet_type, _ = await read_packet(reader, timeout=2.0)
|
|
assert packet_type == AUDIO_SOCKET_PACKET_PCM16
|
|
|
|
for _ in range(20):
|
|
if len(delivered) >= 2 and turns:
|
|
break
|
|
await asyncio.sleep(0.05)
|
|
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
await asyncio.sleep(0.2)
|
|
await runtime.stop()
|
|
|
|
asyncio.run(_scenario())
|
|
|
|
assert ("avs_media_runtime", "speaking") in states
|
|
assert ("avs_media_runtime", "thinking") in states
|
|
assert ("avs_media_runtime", "listening") in states
|
|
assert delivered[0] == ("avs_media_runtime", "greeting", True)
|
|
assert delivered[-1] == ("avs_media_runtime", "reply", False)
|
|
assert turns == [("avs_media_runtime", "hello", False)]
|
|
assert handoffs == []
|
|
assert touch_calls
|
|
assert media_ended
|
|
assert errors == []
|
|
|
|
|
|
def test_media_runtime_waits_for_late_registration():
|
|
registrations: dict[str, MediaRegistration] = {}
|
|
delivered: list[tuple[str, str, bool]] = []
|
|
errors: list[tuple[str, str]] = []
|
|
|
|
media_uuid = str(uuid.uuid4())
|
|
|
|
runtime = AudioSocketMediaRuntime(
|
|
enabled=True,
|
|
host="127.0.0.1",
|
|
port=0,
|
|
frame_ms=20,
|
|
idle_timeout_seconds=2.0,
|
|
registration_wait_timeout_seconds=1.0,
|
|
min_speech_ms=40,
|
|
trailing_silence_ms=40,
|
|
max_turn_ms=400,
|
|
asr_provider=_StubASRProvider(),
|
|
tts_provider=_StubTTSProvider(),
|
|
load_registration_by_media_uuid=lambda value: registrations.get(value),
|
|
mark_media_connected=lambda session_id, value: None,
|
|
mark_media_ended=lambda session_id, reason: None,
|
|
touch_media_frame=lambda session_id: None,
|
|
set_state=lambda session_id, state, handoff_reason, metadata: None,
|
|
get_pending_greeting=lambda session_id: "greeting" if session_id == "avs_media_runtime_late" else None,
|
|
mark_reply_delivered=lambda session_id, text, is_greeting: delivered.append((session_id, text, is_greeting)),
|
|
process_turn=lambda session_id, transcript_text, language, barge_in, metadata: VoiceAITurnDecisionOut(
|
|
language=language or "ru",
|
|
intent="answer",
|
|
reply_text="reply",
|
|
confidence=0.9,
|
|
needs_handoff=False,
|
|
handoff_reason=None,
|
|
case_action="keep_open",
|
|
kb_refs=[],
|
|
summary_text="reply ready",
|
|
model="stub-voice",
|
|
latency_ms=1,
|
|
status="active",
|
|
),
|
|
request_handoff=lambda session_id, customer_request_text, decision: None,
|
|
handle_media_error=lambda session_id, message, metadata: errors.append((session_id, message)),
|
|
)
|
|
|
|
async def _scenario() -> None:
|
|
await runtime.start()
|
|
port = runtime._server.sockets[0].getsockname()[1]
|
|
|
|
async def _late_register() -> None:
|
|
await asyncio.sleep(0.15)
|
|
registrations[media_uuid] = MediaRegistration(
|
|
voice_session_id="avs_media_runtime_late",
|
|
call_id="call_media_runtime_late",
|
|
interaction_id="int_media_runtime_late",
|
|
ai_session_id="ais_media_runtime_late",
|
|
language="ru",
|
|
media_uuid=media_uuid,
|
|
)
|
|
|
|
late_task = asyncio.create_task(_late_register())
|
|
reader, writer = await asyncio.open_connection("127.0.0.1", port)
|
|
writer.write(encode_packet(AUDIO_SOCKET_PACKET_UUID, uuid.UUID(media_uuid).bytes))
|
|
await writer.drain()
|
|
|
|
packet_type, _ = await read_packet(reader, timeout=2.0)
|
|
assert packet_type == AUDIO_SOCKET_PACKET_PCM16
|
|
for _ in range(20):
|
|
if delivered:
|
|
break
|
|
await asyncio.sleep(0.05)
|
|
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
await late_task
|
|
await asyncio.sleep(0.2)
|
|
await runtime.stop()
|
|
|
|
asyncio.run(_scenario())
|
|
|
|
assert delivered == [("avs_media_runtime_late", "greeting", True)]
|
|
assert errors == []
|
|
|
|
|
|
def test_media_runtime_sends_keepalive_while_tts_is_slow():
|
|
registrations: dict[str, MediaRegistration] = {}
|
|
|
|
media_uuid = str(uuid.uuid4())
|
|
registrations[media_uuid] = MediaRegistration(
|
|
voice_session_id="avs_media_runtime_keepalive",
|
|
call_id="call_media_runtime_keepalive",
|
|
interaction_id="int_media_runtime_keepalive",
|
|
ai_session_id="ais_media_runtime_keepalive",
|
|
language="ru",
|
|
media_uuid=media_uuid,
|
|
)
|
|
|
|
class _SlowTTSProvider(TTSProvider):
|
|
name = "slow-stub-tts"
|
|
|
|
def synthesize(self, text: str, *, language: str | None = None) -> TTSSynthesis:
|
|
del language
|
|
assert text
|
|
time.sleep(1.2)
|
|
return TTSSynthesis(
|
|
text=text,
|
|
audio_bytes=(b"\x20\x00" * 960),
|
|
sample_rate_hz=24000,
|
|
)
|
|
|
|
runtime = AudioSocketMediaRuntime(
|
|
enabled=True,
|
|
host="127.0.0.1",
|
|
port=0,
|
|
frame_ms=20,
|
|
idle_timeout_seconds=2.0,
|
|
registration_wait_timeout_seconds=0.5,
|
|
min_speech_ms=40,
|
|
trailing_silence_ms=40,
|
|
max_turn_ms=400,
|
|
asr_provider=_StubASRProvider(),
|
|
tts_provider=_SlowTTSProvider(),
|
|
load_registration_by_media_uuid=lambda value: registrations.get(value),
|
|
mark_media_connected=lambda session_id, value: None,
|
|
mark_media_ended=lambda session_id, reason: None,
|
|
touch_media_frame=lambda session_id: None,
|
|
set_state=lambda session_id, state, handoff_reason, metadata: None,
|
|
get_pending_greeting=lambda session_id: "greeting" if session_id == "avs_media_runtime_keepalive" else None,
|
|
mark_reply_delivered=lambda session_id, text, is_greeting: None,
|
|
process_turn=lambda session_id, transcript_text, language, barge_in, metadata: VoiceAITurnDecisionOut(
|
|
language=language or "ru",
|
|
intent="answer",
|
|
reply_text="reply",
|
|
confidence=0.9,
|
|
needs_handoff=False,
|
|
handoff_reason=None,
|
|
case_action="keep_open",
|
|
kb_refs=[],
|
|
summary_text="reply ready",
|
|
model="stub-voice",
|
|
latency_ms=1,
|
|
status="active",
|
|
),
|
|
request_handoff=lambda session_id, customer_request_text, decision: None,
|
|
handle_media_error=lambda session_id, message, metadata: None,
|
|
)
|
|
|
|
async def _scenario() -> None:
|
|
await runtime.start()
|
|
port = runtime._server.sockets[0].getsockname()[1]
|
|
|
|
reader, writer = await asyncio.open_connection("127.0.0.1", port)
|
|
writer.write(encode_packet(AUDIO_SOCKET_PACKET_UUID, uuid.UUID(media_uuid).bytes))
|
|
await writer.drain()
|
|
|
|
started_at = time.monotonic()
|
|
packet_type, payload = await read_packet(reader, timeout=1.0)
|
|
elapsed = time.monotonic() - started_at
|
|
assert packet_type == AUDIO_SOCKET_PACKET_PCM16
|
|
assert payload == (b"\x00" * 320)
|
|
assert elapsed < 1.0
|
|
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
await asyncio.sleep(0.2)
|
|
await runtime.stop()
|
|
|
|
asyncio.run(_scenario())
|
|
|
|
|
|
def test_media_runtime_sends_keepalive_before_registration_is_ready():
|
|
registrations: dict[str, MediaRegistration] = {}
|
|
delivered: list[tuple[str, str, bool]] = []
|
|
|
|
media_uuid = str(uuid.uuid4())
|
|
|
|
runtime = AudioSocketMediaRuntime(
|
|
enabled=True,
|
|
host="127.0.0.1",
|
|
port=0,
|
|
frame_ms=20,
|
|
idle_timeout_seconds=2.0,
|
|
registration_wait_timeout_seconds=2.0,
|
|
min_speech_ms=40,
|
|
trailing_silence_ms=40,
|
|
max_turn_ms=400,
|
|
asr_provider=_StubASRProvider(),
|
|
tts_provider=_StubTTSProvider(),
|
|
load_registration_by_media_uuid=lambda value: registrations.get(value),
|
|
mark_media_connected=lambda session_id, value: None,
|
|
mark_media_ended=lambda session_id, reason: None,
|
|
touch_media_frame=lambda session_id: None,
|
|
set_state=lambda session_id, state, handoff_reason, metadata: None,
|
|
get_pending_greeting=lambda session_id: "greeting" if session_id == "avs_media_runtime_prereg" else None,
|
|
mark_reply_delivered=lambda session_id, text, is_greeting: delivered.append((session_id, text, is_greeting)),
|
|
process_turn=lambda session_id, transcript_text, language, barge_in, metadata: VoiceAITurnDecisionOut(
|
|
language=language or "ru",
|
|
intent="answer",
|
|
reply_text="reply",
|
|
confidence=0.9,
|
|
needs_handoff=False,
|
|
handoff_reason=None,
|
|
case_action="keep_open",
|
|
kb_refs=[],
|
|
summary_text="reply ready",
|
|
model="stub-voice",
|
|
latency_ms=1,
|
|
status="active",
|
|
),
|
|
request_handoff=lambda session_id, customer_request_text, decision: None,
|
|
handle_media_error=lambda session_id, message, metadata: None,
|
|
)
|
|
|
|
async def _scenario() -> None:
|
|
await runtime.start()
|
|
port = runtime._server.sockets[0].getsockname()[1]
|
|
|
|
async def _late_register() -> None:
|
|
await asyncio.sleep(1.1)
|
|
registrations[media_uuid] = MediaRegistration(
|
|
voice_session_id="avs_media_runtime_prereg",
|
|
call_id="call_media_runtime_prereg",
|
|
interaction_id="int_media_runtime_prereg",
|
|
ai_session_id="ais_media_runtime_prereg",
|
|
language="ru",
|
|
media_uuid=media_uuid,
|
|
)
|
|
|
|
late_task = asyncio.create_task(_late_register())
|
|
reader, writer = await asyncio.open_connection("127.0.0.1", port)
|
|
writer.write(encode_packet(AUDIO_SOCKET_PACKET_UUID, uuid.UUID(media_uuid).bytes))
|
|
await writer.drain()
|
|
|
|
started_at = time.monotonic()
|
|
packet_type, payload = await read_packet(reader, timeout=1.0)
|
|
elapsed = time.monotonic() - started_at
|
|
assert packet_type == AUDIO_SOCKET_PACKET_PCM16
|
|
assert payload == (b"\x00" * 320)
|
|
assert elapsed < 1.0
|
|
|
|
for _ in range(30):
|
|
if delivered:
|
|
break
|
|
await asyncio.sleep(0.1)
|
|
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
await late_task
|
|
await asyncio.sleep(0.2)
|
|
await runtime.stop()
|
|
|
|
asyncio.run(_scenario())
|
|
|
|
assert delivered == [("avs_media_runtime_prereg", "greeting", True)]
|
|
|
|
|
|
def test_media_runtime_starts_handoff_before_handoff_tts_finishes():
|
|
handoff_started = threading.Event()
|
|
events: list[str] = []
|
|
|
|
runtime = AudioSocketMediaRuntime(
|
|
enabled=True,
|
|
host="127.0.0.1",
|
|
port=0,
|
|
frame_ms=20,
|
|
idle_timeout_seconds=2.0,
|
|
registration_wait_timeout_seconds=0.5,
|
|
min_speech_ms=40,
|
|
trailing_silence_ms=40,
|
|
max_turn_ms=400,
|
|
asr_provider=_StubASRProvider(),
|
|
tts_provider=_StubTTSProvider(),
|
|
load_registration_by_media_uuid=lambda value: None,
|
|
mark_media_connected=lambda session_id, value: None,
|
|
mark_media_ended=lambda session_id, reason: None,
|
|
touch_media_frame=lambda session_id: None,
|
|
set_state=lambda session_id, state, handoff_reason, metadata: None,
|
|
get_pending_greeting=lambda session_id: None,
|
|
mark_reply_delivered=lambda session_id, text, is_greeting: None,
|
|
process_turn=lambda session_id, transcript_text, language, barge_in, metadata: VoiceAITurnDecisionOut(
|
|
language=language or "ru",
|
|
intent="handoff_request",
|
|
reply_text="Соединяю с оператором",
|
|
confidence=0.9,
|
|
needs_handoff=True,
|
|
handoff_reason="Нужен живой оператор",
|
|
case_action="keep_open",
|
|
kb_refs=[],
|
|
summary_text="handoff requested",
|
|
model="stub-voice",
|
|
latency_ms=1,
|
|
status="handoff_requested",
|
|
),
|
|
request_handoff=lambda session_id, customer_request_text, decision: (
|
|
events.append("handoff"),
|
|
handoff_started.set()
|
|
),
|
|
handle_media_error=lambda session_id, message, metadata: None,
|
|
)
|
|
|
|
actor = MediaRegistration(
|
|
voice_session_id="avs_media_runtime_handoff",
|
|
call_id="call_media_runtime_handoff",
|
|
interaction_id="int_media_runtime_handoff",
|
|
ai_session_id="ais_media_runtime_handoff",
|
|
language="ru",
|
|
media_uuid=str(uuid.uuid4()),
|
|
)
|
|
|
|
media_actor = None
|
|
|
|
async def _scenario() -> None:
|
|
nonlocal media_actor
|
|
media_actor = MediaActor(
|
|
registration=actor,
|
|
reader=asyncio.StreamReader(),
|
|
writer=None, # type: ignore[arg-type]
|
|
vad=EnergyVAD(frame_ms=20, min_speech_ms=40, trailing_silence_ms=40, max_turn_ms=400),
|
|
frame_ms=20,
|
|
frame_bytes=320,
|
|
)
|
|
|
|
async def _fake_speak_text(current_actor, text: str, *, is_greeting: bool) -> None:
|
|
del current_actor, text, is_greeting
|
|
events.append("speak_start")
|
|
await asyncio.sleep(0)
|
|
assert handoff_started.wait(timeout=0.5)
|
|
await asyncio.sleep(0.05)
|
|
events.append("speak_end")
|
|
|
|
runtime._speak_text = _fake_speak_text # type: ignore[method-assign]
|
|
pcm_frame = (1000).to_bytes(2, "little", signed=True) * 160
|
|
await runtime._process_utterance(media_actor, pcm_frame, False)
|
|
|
|
asyncio.run(_scenario())
|
|
|
|
assert handoff_started.is_set()
|
|
assert events.index("handoff") < events.index("speak_end")
|