diff --git a/README.md b/README.md index e7a34cf..2170101 100644 --- a/README.md +++ b/README.md @@ -47,7 +47,7 @@ SIP_PASSWORD=******** uvx 'voip[cli]' sip sips:alice@sip.example.com agent uv add voip[audio,ai,pygments] ``` -Subclass `WhisperCall` and override `transcription_received` to handle results. +Subclass `TranscribeCall` and override `transcription_received` to handle results. Pass it as `call_class` when answering an incoming call: ```python diff --git a/docs/calls.md b/docs/calls.md index a8c8af0..8bb99e8 100644 --- a/docs/calls.md +++ b/docs/calls.md @@ -6,15 +6,11 @@ ::: voip.audio.AudioCall -## Voice Activity Detection - ::: voip.audio.VoiceActivityCall -## Echo Call - ::: voip.audio.EchoCall -## AI / Agentic Calls +## AI Calls ::: voip.ai.TranscribeCall diff --git a/docs/cookbook.md b/docs/cookbook.md index 5c659d1..acfb71c 100644 --- a/docs/cookbook.md +++ b/docs/cookbook.md @@ -76,14 +76,13 @@ from pocket_tts import TTSModel from voip.ai import AgentCall from voip.sip.protocol import SIP - shared_tts = TTSModel.load_model() class MyCall(AgentCall): tts_model = shared_tts system_prompt = "You are a friendly hotel receptionist. Keep answers brief." - ollama_model = "llama3" + llm_model = "llama3" voice = "azelma" diff --git a/tests/sip/test_protocol.py b/tests/sip/test_protocol.py index 1f441a2..769ebdb 100644 --- a/tests/sip/test_protocol.py +++ b/tests/sip/test_protocol.py @@ -495,7 +495,7 @@ async def _run_answer(self, protocol, invite, fake_rtp_transport): protocol._rtp_transport = fake_rtp_transport # Resolve the SIP protocol's own local address (for Contact header). protocol.local_address = ("127.0.0.1", 5061) - await protocol._answer(invite, _CodecAwareCall) + await protocol.answer(invite, call_class=_CodecAwareCall) @pytest.mark.asyncio async def test_answer__selects_pcma_from_offer(self, fake_rtp_transport): @@ -662,7 +662,7 @@ async def test_answer__no_address_logs_error(self, caplog): invite = self._make_invite("no-addr-answer-1") with caplog.at_level("ERROR"): - await protocol._answer(invite, RTPCall) + await protocol.answer(invite, call_class=RTPCall) assert "No pending INVITE found" in caplog.text assert not protocol._sent_responses @@ -1116,7 +1116,7 @@ async def test_answer__sends_200_ok(self): protocol._rtp_transport = mock_rtp_transport request = make_invite() protocol._pending_invites.add(request.headers["Call-ID"]) - await protocol._answer(request, RTPCall) + await protocol.answer(request, call_class=RTPCall) assert len(protocol._sent) == 1 response, _ = protocol._sent[0] assert response.status_code == 200 @@ -1137,7 +1137,7 @@ async def test_answer__sdp_contains_opus_audio_line(self): protocol._rtp_transport = mock_rtp_transport request = make_invite() protocol._pending_invites.add(request.headers["Call-ID"]) - await protocol._answer(request, RTPCall) + await protocol.answer(request, call_class=RTPCall) response, _ = protocol._sent[0] assert b"m=audio" in bytes(response.body) assert b"RTP/SAVP 0" in bytes(response.body) @@ -1183,9 +1183,9 @@ async def test_answer__rtp_avp_offer_returns_rtp_avp(self): request = Request.parse(invite_bytes) protocol._pending_invites.add(request.headers["Call-ID"]) - AudioCall = pytest.importorskip("voip.audio.AudioCall") + AudioCall = pytest.importorskip("voip.audio").AudioCall - await protocol._answer(request, AudioCall) + await protocol.answer(request, call_class=AudioCall) response, _ = protocol._sent[0] body = bytes(response.body) assert b"RTP/AVP" in body @@ -1221,9 +1221,9 @@ async def test_answer__rtp_savp_offer_returns_rtp_savp(self): request = Request.parse(invite_bytes) protocol._pending_invites.add(request.headers["Call-ID"]) - AudioCall = pytest.importorskip("voip.audio.AudioCall") + AudioCall = pytest.importorskip("voip.audio").AudioCall - await protocol._answer(request, AudioCall) + await protocol.answer(request, call_class=AudioCall) response, _ = protocol._sent[0] body = bytes(response.body) assert b"RTP/SAVP" in body @@ -1247,7 +1247,7 @@ async def test_answer__copies_dialog_headers(self): protocol._rtp_transport = mock_rtp_transport request = make_invite() protocol._pending_invites.add(request.headers["Call-ID"]) - await protocol._answer(request, RTPCall) + await protocol.answer(request, call_class=RTPCall) response, _ = protocol._sent[0] assert response.headers["Via"] == "SIP/2.0/UDP pc33.atlanta.com" assert response.headers["To"] == "sip:alice@atlanta.com" @@ -1277,7 +1277,7 @@ def __post_init__(self) -> None: protocol._rtp_transport = mock_rtp_transport request = make_invite() protocol._pending_invites.add(request.headers["Call-ID"]) - await protocol._answer(request, MyCall) + await protocol.answer(request, call_class=MyCall) assert created == ["sip:bob@biloxi.com"] async def test_answer__rtp_receives_audio(self): @@ -1304,7 +1304,7 @@ def packet_received(self, packet: RTPPacket, addr) -> None: request = make_invite() protocol._pending_invites.add(request.headers["Call-ID"]) try: - await protocol._answer(request, PacketCapture) + await protocol.answer(request, call_class=PacketCapture) response, _ = protocol._sent[0] sdp_line = next( line @@ -1352,7 +1352,7 @@ def packet_received(self, packet: RTPPacket, addr) -> None: request = make_invite() protocol._pending_invites.add(request.headers["Call-ID"]) try: - await protocol._answer(request, PacketCapture) + await protocol.answer(request, call_class=PacketCapture) response, _ = protocol._sent[0] sdp_line = next( line @@ -1391,7 +1391,7 @@ async def test_answer__content_length_serialized(self): protocol._rtp_transport = mock_rtp_transport request = make_invite() protocol._pending_invites.add(request.headers["Call-ID"]) - await protocol._answer(request, RTPCall) + await protocol.answer(request, call_class=RTPCall) response, _ = protocol._sent[0] serialized = bytes(response) parsed = Message.parse(serialized) @@ -1429,7 +1429,7 @@ async def test_answer__reuses_shared_rtp_socket_for_second_call(self): body=sdp_body1, ) protocol._pending_invites.add("call-1@test") - await protocol._answer(invite1, _MinimalCall) + await protocol.answer(invite1, call_class=_MinimalCall) rtp_proto_1 = protocol._rtp_protocol rtp_transport_1 = protocol._rtp_transport @@ -1449,7 +1449,7 @@ async def test_answer__reuses_shared_rtp_socket_for_second_call(self): body=sdp_body2, ) protocol._pending_invites.add("call-2@test") - await protocol._answer(invite2, _MinimalCall) + await protocol.answer(invite2, call_class=_MinimalCall) assert protocol._rtp_protocol is rtp_proto_1 assert protocol._rtp_transport is rtp_transport_1 @@ -1473,7 +1473,7 @@ async def test_answer__bye_unregisters_call_from_rtp_mux(self): request = make_invite() protocol._pending_invites.add(request.headers["Call-ID"]) try: - await protocol._answer(request, RTPCall) + await protocol.answer(request, call_class=RTPCall) assert None in mux.calls bye = Request( @@ -1510,7 +1510,7 @@ async def test_answer__logs_info(self, caplog): request = make_invite() protocol._pending_invites.add(request.headers["Call-ID"]) with caplog.at_level(logging.INFO, logger="voip.sip"): - await protocol._answer(request, RTPCall) + await protocol.answer(request, call_class=RTPCall) assert any("call_answered" in r.message for r in caplog.records) def test_reject__sends_busy_here_by_default(self): @@ -1592,26 +1592,30 @@ async def test_request_received__unsupported_method__raises(self): async def test_answer__via_call_received__schedules_answer(self): """answer() is async; wrapping it in create_task from call_received works.""" - answered = [] - class MySIP(SIP): + class MySIP(self._CapturingSIP): def call_received(self, request): asyncio.create_task( self.answer(request=request, call_class=_MinimalCall) ) - async def _answer(self, request, call_class): - answered.append((request, call_class)) - - protocol = MySIP(outbound_proxy=("127.0.0.1", 5060), aor="sip:test@example.com") - protocol.connection_made(MagicMock()) + loop = asyncio.get_running_loop() + protocol = MySIP() + protocol.transport = make_mock_transport() + protocol.local_address = ("127.0.0.1", 5061) + mux = RealtimeTransportProtocol() + mux.public_address = loop.create_future() + mux.public_address.set_result(("127.0.0.1", 12000)) + mock_rtp_transport = MagicMock() + mock_rtp_transport.get_extra_info.return_value = ("127.0.0.1", 12000) + protocol._rtp_protocol = mux + protocol._rtp_transport = mock_rtp_transport request = make_invite() protocol._pending_invites.add(request.headers["Call-ID"]) protocol.call_received(request) - await asyncio.sleep(0.01) - assert len(answered) == 1 - assert answered[0][1] is _MinimalCall + await asyncio.sleep(0.05) + assert len(protocol._sent) == 1 # --------------------------------------------------------------------------- diff --git a/tests/test_ai.py b/tests/test_ai.py index ab74e90..182072a 100644 --- a/tests/test_ai.py +++ b/tests/test_ai.py @@ -1,4 +1,4 @@ -"""Tests for AI-powered call handlers (WhisperCall and AgentCall).""" +"""Tests for AI-powered call handlers (TranscribeCall and AgentCall).""" from __future__ import annotations @@ -41,7 +41,7 @@ def _make_media(fmt: str, rtpmap: str | None = None) -> MediaDescription: def make_whisper_call( model_mock: MagicMock, call_class=None, media: MediaDescription | None = None ) -> TranscribeCall: - """Return a WhisperCall with a mocked Whisper model.""" + """Return a TranscribeCall with a mocked Whisper model.""" cls = call_class or TranscribeCall med = media if media is not None else OPUS_MEDIA with patch("voip.ai.WhisperModel", return_value=model_mock): @@ -75,13 +75,13 @@ def make_agent_call( ) -class TestWhisperCall: +class TestTranscribeCall: def test_whisper_call__is_audio_call(self): - """WhisperCall is a subclass of AudioCall.""" + """TranscribeCall is a subclass of AudioCall.""" assert issubclass(TranscribeCall, AudioCall) def test_init__uses_pre_loaded_model_instance(self): - """When model is a WhisperModel instance it is stored directly (no re-load).""" + """When stt_model is a WhisperModel instance it is stored directly (no re-load).""" model_instance = MagicMock() with patch("voip.ai.WhisperModel") as wm_cls: # Pass the instance directly — the constructor must NOT be called again. @@ -89,11 +89,11 @@ def test_init__uses_pre_loaded_model_instance(self): rtp=MagicMock(), sip=MagicMock(), media=OPUS_MEDIA, - model=model_instance, + stt_model=model_instance, caller=CallerID(""), ) wm_cls.assert_not_called() - assert call.whisper_model is model_instance + assert call.stt_model is model_instance def test_init__stores_media(self): """Media is stored and accessible as self.media.""" @@ -111,63 +111,64 @@ def test_init__derives_payload_type_from_pcma_media(self): assert call.payload_type == RTPPayloadType.PCMA def test_audio_received__initializes_vad_state(self): - """WhisperCall starts with an empty speech buffer and no timer.""" + """TranscribeCall starts with an empty speech buffer and no flush timer.""" call = make_whisper_call(MagicMock()) - assert call.speech_buffer == [] - assert call.silence_handle is None + assert call._speech_buffer.size == 0 + assert call._flush_voice_buffer_handle is None def test_audio_received__silence_audio_accumulates_in_buffer(self): - """Silence audio (below speech_threshold) is still buffered for transcription.""" + """Silence audio (below voice_rms_threshold) is still buffered.""" call = make_whisper_call(MagicMock()) - with patch("voip.audio.asyncio.get_running_loop"): + with patch("voip.audio.asyncio.get_event_loop"): call.audio_received(audio=np.zeros(320, dtype=np.float32), rms=0.0) - assert len(call.speech_buffer) == 1 + assert call._speech_buffer.size == 320 def test_audio_received__speech_audio_accumulates_in_buffer(self): - """Audio above speech_threshold is added to _speech_buffer.""" + """Audio above voice_rms_threshold is added to _speech_buffer.""" call = make_whisper_call(MagicMock()) speech = np.ones(320, dtype=np.float32) * 0.6 call.audio_received(audio=speech, rms=0.6) - assert len(call.speech_buffer) == 1 + assert call._speech_buffer.size == 320 - def test_audio_received__silence_arms_transcription_timer(self): - """Silence arms the transcription debounce timer.""" + def test_audio_received__silence_arms_flush_timer(self): + """Silence arms the flush debounce timer.""" call = make_whisper_call(MagicMock()) - with patch("voip.audio.asyncio.get_running_loop") as mock_loop: + with patch("voip.audio.asyncio.get_event_loop") as mock_loop: handle = MagicMock() mock_loop.return_value.call_later.return_value = handle call.audio_received(audio=np.zeros(320, dtype=np.float32), rms=0.0) mock_loop.return_value.call_later.assert_called_once_with( - call.silence_gap, call.flush_speech_buffer + call.silence_gap.total_seconds(), + call.flush_voice_buffer, ) - assert call.silence_handle is handle + assert call._flush_voice_buffer_handle is handle def test_audio_received__silence_does_not_rearm_when_timer_running(self): """A second silence packet does not create a second timer.""" call = make_whisper_call(MagicMock()) - call.silence_handle = MagicMock() - with patch("voip.audio.asyncio.get_running_loop") as mock_loop: + call._flush_voice_buffer_handle = MagicMock() + with patch("voip.audio.asyncio.get_event_loop") as mock_loop: call.audio_received(audio=np.zeros(320, dtype=np.float32), rms=0.0) mock_loop.return_value.call_later.assert_not_called() def test_audio_received__speech_cancels_pending_timer(self): - """Speech audio cancels any running transcription debounce timer.""" + """Speech audio cancels any running flush timer.""" call = make_whisper_call(MagicMock()) handle = MagicMock() - call.silence_handle = handle + call._flush_voice_buffer_handle = handle call.audio_received(audio=np.ones(320, dtype=np.float32) * 0.6, rms=0.6) handle.cancel.assert_called_once() - assert call.silence_handle is None + assert call._flush_voice_buffer_handle is None def test_audio_received__empty_array_accumulates_in_buffer(self): """Zero-length audio arrays are accepted into the speech buffer.""" call = make_whisper_call(MagicMock()) - with patch("voip.audio.asyncio.get_running_loop"): + with patch("voip.audio.asyncio.get_event_loop"): call.audio_received(audio=np.zeros(0, dtype=np.float32), rms=0.0) - assert len(call.speech_buffer) == 1 + assert call._speech_buffer.size == 0 async def test_flush_speech_buffer__transcribes_accumulated_audio(self): - """_flush_speech_buffer concatenates speech and schedules transcription.""" + """flush_voice_buffer concatenates speech and schedules transcription.""" transcriptions = [] model_mock = MagicMock() seg = MagicMock() @@ -179,44 +180,39 @@ def transcription_received(self, text: str) -> None: transcriptions.append(text) call = make_whisper_call(model_mock, Capture) - chunk = np.ones(320, dtype=np.float32) - call.speech_buffer = [chunk] * 60 - call.flush_speech_buffer() + # Fill buffer with 2 s of audio to pass the length and RMS thresholds. + call._speech_buffer = np.ones(call.sampling_rate_hz * 2, dtype=np.float32) + call.flush_voice_buffer() await asyncio.sleep(0.1) assert transcriptions == ["hello"] - assert call.speech_buffer == [] + assert call._speech_buffer.size == 0 def test_flush_speech_buffer__no_op_when_buffer_empty(self): - """_flush_speech_buffer does nothing when the speech buffer is empty.""" + """flush_voice_buffer does nothing when the speech buffer is empty.""" call = make_whisper_call(MagicMock()) with patch("voip.audio.asyncio.create_task") as mock_ct: - call.flush_speech_buffer() + call.flush_voice_buffer() mock_ct.assert_not_called() def test_flush_speech_buffer__resets_state(self): - """_flush_speech_buffer clears silence_handle and the speech buffer.""" + """flush_voice_buffer clears _flush_voice_buffer_handle and the speech buffer.""" call = make_whisper_call(MagicMock()) - call.silence_handle = MagicMock() - call.speech_buffer = [np.zeros(1, dtype=np.float32)] + call._flush_voice_buffer_handle = MagicMock() + call._speech_buffer = np.ones(call.sampling_rate_hz * 2, dtype=np.float32) with patch("voip.audio.asyncio.create_task", side_effect=lambda c: c.close()): - call.flush_speech_buffer() - assert call.silence_handle is None + call.flush_voice_buffer() + assert call._flush_voice_buffer_handle is None async def test_speech_buffer_ready__skips_short_audio(self): - """speech_buffer_ready discards audio shorter than one second.""" - transcriptions = [] + """flush_voice_buffer discards audio shorter than silence_gap.""" model_mock = MagicMock() - - class Capture(TranscribeCall): - def transcription_received(self, text: str) -> None: - transcriptions.append(text) - - call = make_whisper_call(model_mock, Capture) - # Feed fewer samples than RESAMPLING_RATE_HZ (1 second) - short_audio = np.ones(100, dtype=np.float32) - await call.speech_buffer_ready(short_audio) + call = make_whisper_call(model_mock) + # Pre-fill the buffer with fewer samples than sampling_rate_hz * silence_gap_secs. + call._speech_buffer = np.ones(100, dtype=np.float32) + with patch("voip.audio.asyncio.create_task") as mock_ct: + call.flush_voice_buffer() + mock_ct.assert_not_called() model_mock.transcribe.assert_not_called() - assert transcriptions == [] async def test_transcribe__strips_whitespace(self): """Strip leading and trailing whitespace from the transcription text.""" @@ -268,7 +264,7 @@ def test_decode_payload__opus__delegates_to_opus_codec(self): ) as mock_decode: call.decode_payload(b"pkt") mock_decode.assert_called_once_with( - b"pkt", AudioCall.RESAMPLING_RATE_HZ, input_rate_hz=call.sample_rate + b"pkt", call.sampling_rate_hz, input_rate_hz=call.sample_rate ) def test_decode_payload__pcma__delegates_to_pcma_codec(self): @@ -280,7 +276,7 @@ def test_decode_payload__pcma__delegates_to_pcma_codec(self): ) as mock_decode: call.decode_payload(b"pkt") mock_decode.assert_called_once_with( - b"pkt", AudioCall.RESAMPLING_RATE_HZ, input_rate_hz=call.sample_rate + b"pkt", call.sampling_rate_hz, input_rate_hz=call.sample_rate ) def test_decode_payload__pcmu__delegates_to_pcmu_codec(self): @@ -292,7 +288,7 @@ def test_decode_payload__pcmu__delegates_to_pcmu_codec(self): ) as mock_decode: call.decode_payload(b"pkt") mock_decode.assert_called_once_with( - b"pkt", AudioCall.RESAMPLING_RATE_HZ, input_rate_hz=call.sample_rate + b"pkt", call.sampling_rate_hz, input_rate_hz=call.sample_rate ) def test_decode_payload__passes_sdp_sample_rate_as_input(self): @@ -305,7 +301,7 @@ def test_decode_payload__passes_sdp_sample_rate_as_input(self): ) as mock_decode: call.decode_payload(b"pkt") mock_decode.assert_called_once_with( - b"pkt", AudioCall.RESAMPLING_RATE_HZ, input_rate_hz=16000 + b"pkt", call.sampling_rate_hz, input_rate_hz=16000 ) async def test_transcribe__raises_on_general_error(self): @@ -348,7 +344,7 @@ def transcription_received(self, text: str) -> None: class TestAgentCall: def test_agent_call__is_whisper_call(self): - """AgentCall is a subclass of WhisperCall.""" + """AgentCall is a subclass of TranscribeCall.""" assert issubclass(AgentCall, TranscribeCall) def test_init__loads_tts_model_when_none(self): @@ -364,7 +360,7 @@ def test_init__loads_tts_model_when_none(self): rtp=MagicMock(), sip=MagicMock(), media=OPUS_MEDIA, caller=CallerID("") ) tts_cls.load_model.assert_called_once() - assert call.tts_instance is tts_mock + assert call.tts_model is tts_mock def test_init__uses_provided_tts_model(self): """Use the provided TTSModel instance instead of loading a new one.""" @@ -382,7 +378,7 @@ def test_init__uses_provided_tts_model(self): caller=CallerID(""), ) tts_cls.load_model.assert_not_called() - assert call.tts_instance is tts_mock + assert call.tts_model is tts_mock def test_init__loads_voice_state(self): """Get the voice state from the TTS model on init.""" @@ -402,35 +398,39 @@ def test_init__loads_voice_state(self): caller=CallerID(""), ) tts_mock.get_state_for_audio_prompt.assert_called_once_with("alba") - assert call.voice_state is voice_state + assert call._voice_state is voice_state def test_init__initializes_pending_state(self): - """AgentCall starts with empty pending text and no response task.""" + """AgentCall starts with an empty response task.""" tts_mock = MagicMock() tts_mock.get_state_for_audio_prompt.return_value = MagicMock() call = make_agent_call(MagicMock(), tts_mock) - assert call.pending_text == [] - assert call.response_task is None + assert call._response_task is None def test_init__initializes_chat_history_with_system_prompt(self): """Chat history is seeded with a system prompt mentioning a phone call.""" tts_mock = MagicMock() tts_mock.get_state_for_audio_prompt.return_value = MagicMock() call = make_agent_call(MagicMock(), tts_mock) - assert len(call.messages) == 1 - assert call.messages[0]["role"] == "system" - assert "phone" in call.messages[0]["content"].lower() + assert len(call._messages) == 1 + assert call._messages[0]["role"] == "system" + assert "phone" in call._messages[0]["content"].lower() def test_transcription_received__ignores_empty_text(self): - """transcription_received does not buffer empty text.""" + """transcription_received appends empty text and creates a response task.""" tts_mock = MagicMock() tts_mock.get_state_for_audio_prompt.return_value = MagicMock() call = make_agent_call(MagicMock(), tts_mock) - call.transcription_received("") - assert call.pending_text == [] + with patch( + "voip.ai.asyncio.create_task", + side_effect=lambda c: c.close() or MagicMock(), + ) as mock_ct: + call.transcription_received("") + mock_ct.assert_called_once() + assert {"role": "user", "content": ""} in call._messages def test_transcription_received__buffers_non_empty_text(self): - """transcription_received buffers text and creates a response task.""" + """transcription_received appends a user message and creates a response task.""" tts_mock = MagicMock() tts_mock.get_state_for_audio_prompt.return_value = MagicMock() call = make_agent_call(MagicMock(), tts_mock) @@ -439,7 +439,7 @@ def test_transcription_received__buffers_non_empty_text(self): side_effect=lambda c: c.close() or MagicMock(), ) as mock_ct: call.transcription_received("hello") - assert call.pending_text == ["hello"] + assert {"role": "user", "content": "hello"} in call._messages mock_ct.assert_called_once() def test_transcription_received__schedules_response_task(self): @@ -453,7 +453,7 @@ def test_transcription_received__schedules_response_task(self): ) as mock_ct: call.transcription_received("hello world") mock_ct.assert_called_once() - assert call.response_task is task_mock + assert call._response_task is task_mock def test_transcription_received__cancels_running_task_before_creating_new(self): """transcription_received cancels any existing response task.""" @@ -462,17 +462,17 @@ def test_transcription_received__cancels_running_task_before_creating_new(self): call = make_agent_call(MagicMock(), tts_mock) old_task = MagicMock() old_task.done.return_value = False - call.response_task = old_task + call._response_task = old_task with patch("voip.ai.asyncio.create_task", side_effect=lambda c: c.close()): call.transcription_received("hello") old_task.cancel.assert_called_once() async def test_respond__calls_ollama_and_sends_speech(self): - """_respond fetches an Ollama reply, records it in history, and sends speech.""" + """Respond fetches an Ollama reply, records it in history, and sends speech.""" tts_mock = MagicMock() tts_mock.get_state_for_audio_prompt.return_value = MagicMock() call = make_agent_call(MagicMock(), tts_mock) - call.pending_text = ["hello"] + call._messages.append({"role": "user", "content": "hello"}) mock_response = MagicMock() mock_response.message.content = "I am an AI assistant." with ( @@ -487,18 +487,18 @@ async def test_respond__calls_ollama_and_sends_speech(self): await call.respond() mock_send_speech.assert_awaited_once_with("I am an AI assistant.") - assert {"role": "user", "content": "hello"} in call.messages + assert {"role": "user", "content": "hello"} in call._messages assert { "role": "assistant", "content": "I am an AI assistant.", - } in call.messages + } in call._messages async def test_respond__passes_full_history_to_ollama(self): - """_respond passes the full message history (including system prompt) to Ollama.""" + """Respond passes the full message history (including system prompt) to Ollama.""" tts_mock = MagicMock() tts_mock.get_state_for_audio_prompt.return_value = MagicMock() call = make_agent_call(MagicMock(), tts_mock) - call.pending_text = ["hello"] + call._messages.append({"role": "user", "content": "hello"}) mock_response = MagicMock() mock_response.message.content = "reply" with ( @@ -515,31 +515,27 @@ async def test_respond__passes_full_history_to_ollama(self): assert messages[0]["role"] == "system" assert messages[1] == {"role": "user", "content": "hello"} - async def test_respond__logs_exception_on_error(self, caplog): - """Log an exception when Ollama raises an error.""" - import logging - + async def test_respond__raises_exception_on_error(self): + """Exceptions from Ollama propagate out of respond().""" tts_mock = MagicMock() tts_mock.get_state_for_audio_prompt.return_value = MagicMock() call = make_agent_call(MagicMock(), tts_mock) - call.pending_text = ["hello"] + call._messages.append({"role": "user", "content": "hello"}) with ( patch("voip.ai.ollama.AsyncClient") as mock_client_cls, - caplog.at_level(logging.ERROR, logger="voip.ai"), + pytest.raises(RuntimeError, match="ollama error"), ): mock_client = MagicMock() mock_client.chat = AsyncMock(side_effect=RuntimeError("ollama error")) mock_client_cls.return_value = mock_client await call.respond() - assert any("agent response" in r.message for r in caplog.records) async def test_respond__re_raises_cancelled_error(self): - """Re-raise CancelledError from Ollama and remove the partial user turn.""" + """Re-raise CancelledError from Ollama.""" tts_mock = MagicMock() tts_mock.get_state_for_audio_prompt.return_value = MagicMock() call = make_agent_call(MagicMock(), tts_mock) - call.pending_text = ["hello"] - initial_history_len = len(call.messages) + call._messages.append({"role": "user", "content": "hello"}) with ( patch("voip.ai.ollama.AsyncClient") as mock_client_cls, pytest.raises(asyncio.CancelledError), @@ -548,9 +544,7 @@ async def test_respond__re_raises_cancelled_error(self): mock_client.chat = AsyncMock(side_effect=asyncio.CancelledError()) mock_client_cls.return_value = mock_client await call.respond() - # Partial user turn must be rolled back to keep history consistent - assert len(call.messages) == initial_history_len def test_preferred_codecs__opus_is_first(self): """AgentCall prefers Opus as the highest-priority outbound codec.""" - assert AgentCall.PREFERRED_CODECS[0].payload_type == RTPPayloadType.OPUS + assert AgentCall.supported_codecs[0].payload_type == RTPPayloadType.OPUS diff --git a/tests/test_audio.py b/tests/test_audio.py index a3f5967..253d4af 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import datetime from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -244,7 +245,7 @@ def test_negotiate_codec__subclass_can_override_preferences(self): """A subclass with a different PREFERRED_CODECS list uses its own preferences.""" class PCMAOnlyCall(AudioCall): - PREFERRED_CODECS = [PCMA] + supported_codecs = [PCMA] media = self._make_media(["0", "8", "111"]) result = PCMAOnlyCall.negotiate_codec(media) @@ -252,7 +253,7 @@ class PCMAOnlyCall(AudioCall): def test_preferred_codecs__class_attribute(self): """PREFERRED_CODECS is a class attribute on AudioCall with Opus first when PyAV is available.""" - codec_classes = AudioCall.PREFERRED_CODECS + codec_classes = AudioCall.supported_codecs assert isinstance(codec_classes, list) pts = [c.payload_type for c in codec_classes] assert pts[0] == 111 # Opus is highest priority when PyAV is present @@ -319,7 +320,7 @@ def test_decode_payload__delegates_to_codec(self): call.decode_payload(b"payload") mock_decode.assert_called_once_with( b"payload", - AudioCall.RESAMPLING_RATE_HZ, + call.sampling_rate_hz, input_rate_hz=call.sample_rate, ) @@ -334,7 +335,7 @@ def test_decode_payload__passes_sample_rate_from_media(self): call.decode_payload(b"pkt") mock_decode.assert_called_once_with( b"pkt", - AudioCall.RESAMPLING_RATE_HZ, + call.sampling_rate_hz, input_rate_hz=16000, ) @@ -419,7 +420,7 @@ async def test_send_rtp_audio__sends_to_remote_addr(self): call.rtp.calls = {remote_addr: call} with patch.object(call, "send_packet") as mock_send: - await call.send_rtp_audio(np.zeros(160, dtype=np.float32)) + await call.send_audio(np.zeros(160, dtype=np.float32)) mock_send.assert_called_once() data, addr = mock_send.call_args[0] assert addr == remote_addr @@ -436,7 +437,7 @@ async def test_send_rtp_audio__drops_audio_when_no_remote_addr(self, caplog): caplog.at_level(logging.WARNING, logger="voip.audio"), patch.object(call, "send_packet") as mock_send, ): - await call.send_rtp_audio(np.zeros(160, dtype=np.float32)) + await call.send_audio(np.zeros(160, dtype=np.float32)) mock_send.assert_not_called() assert any("dropping audio" in r.message for r in caplog.records) @@ -458,10 +459,10 @@ async def capture_sleep(delay: float) -> None: patch("voip.audio.asyncio.sleep", side_effect=capture_sleep), patch.object(call, "send_packet"), ): - await call.send_rtp_audio(audio) + await call.send_audio(audio) assert len(sleep_calls) == 2 - assert all(s == call.RTP_PACKET_DURATION_SECS for s in sleep_calls) + assert all(s == call.rpt_packet_duration.total_seconds() for s in sleep_calls) def make_echo_call(**kwargs) -> EchoCall: @@ -495,98 +496,119 @@ def test_voice_activity_call__is_audio_call(self): """VoiceActivityCall is a subclass of AudioCall.""" assert issubclass(VoiceActivityCall, AudioCall) - def test_collect_audio__returns_true_for_speech(self): - """collect_audio returns True when RMS exceeds speech_threshold.""" + def test_audio_received__appends_to_speech_buffer(self): + """audio_received concatenates frames to the speech buffer regardless of RMS.""" call = make_vac_call() - assert call.collect_audio(np.ones(160, dtype=np.float32), rms=1.0) is True + audio = np.ones(160, dtype=np.float32) * 0.5 + with patch("voip.audio.asyncio.get_event_loop"): + call.audio_received(audio=audio, rms=0.5) + assert call._speech_buffer.size == 160 - def test_collect_audio__returns_false_for_silence(self): - """collect_audio returns False when RMS is at or below speech_threshold.""" + def test_audio_received__speech_cancels_flush_timer(self): + """audio_received with speech-level RMS cancels any running flush timer.""" call = make_vac_call() - assert call.collect_audio(np.zeros(160, dtype=np.float32), rms=0.0) is False + handle = MagicMock() + call._flush_voice_buffer_handle = handle + call.audio_received(audio=np.ones(160, dtype=np.float32), rms=1.0) + handle.cancel.assert_called_once() + assert call._flush_voice_buffer_handle is None - def test_audio_received__buffers_speech_frames(self): - """audio_received buffers frames with RMS above speech_threshold.""" + def test_audio_received__silence_arms_flush_timer(self): + """audio_received with silence-level RMS schedules the flush timer once.""" call = make_vac_call() - call.audio_received(audio=np.ones(160, dtype=np.float32), rms=1.0) - assert len(call.speech_buffer) == 1 + with patch("voip.audio.asyncio.get_event_loop") as mock_loop: + handle = MagicMock() + mock_loop.return_value.call_later.return_value = handle + call.audio_received(audio=np.zeros(160, dtype=np.float32), rms=0.0) + mock_loop.return_value.call_later.assert_called_once_with( + call.silence_gap.total_seconds(), + call.flush_voice_buffer, + ) + assert call._flush_voice_buffer_handle is handle - def test_audio_received__does_not_buffer_silence_frames(self): - """audio_received skips frames with RMS at or below speech_threshold.""" + def test_audio_received__silence_does_not_rearm_when_timer_running(self): + """A second silence frame does not replace a running flush timer.""" call = make_vac_call() - call.audio_received(audio=np.zeros(160, dtype=np.float32), rms=0.0) - assert len(call.speech_buffer) == 0 + call._flush_voice_buffer_handle = MagicMock() + with patch("voip.audio.asyncio.get_event_loop") as mock_loop: + call.audio_received(audio=np.zeros(160, dtype=np.float32), rms=0.0) + mock_loop.return_value.call_later.assert_not_called() - def test_on_audio_speech__cancels_silence_timer(self): - """on_audio_speech cancels a running silence timer.""" + def test_on_audio_speech__cancels_flush_handle(self): + """on_audio_speech cancels the flush timer when one is running.""" call = make_vac_call() handle = MagicMock() - call.silence_handle = handle + call._flush_voice_buffer_handle = handle call.on_audio_speech() handle.cancel.assert_called_once() - assert call.silence_handle is None + assert call._flush_voice_buffer_handle is None def test_on_audio_speech__noop_when_no_timer(self): - """on_audio_speech does nothing when no silence timer is running.""" + """on_audio_speech does nothing when no flush timer is running.""" call = make_vac_call() call.on_audio_speech() # must not raise - assert call.silence_handle is None + assert call._flush_voice_buffer_handle is None @pytest.mark.asyncio - async def test_on_audio_silence__arms_timer_when_buffer_has_speech(self): - """on_audio_silence schedules the silence timer when speech is buffered.""" + async def test_on_audio_silence__arms_timer(self): + """on_audio_silence schedules the flush timer via the event loop.""" call = make_vac_call() - call.speech_buffer.append(np.ones(160, dtype=np.float32)) call.on_audio_silence() - assert call.silence_handle is not None - call.silence_handle.cancel() + assert call._flush_voice_buffer_handle is not None + call._flush_voice_buffer_handle.cancel() @pytest.mark.asyncio - async def test_on_audio_silence__noop_when_buffer_is_empty(self): - """on_audio_silence does not schedule a timer when no speech is buffered.""" + async def test_on_audio_silence__noop_when_timer_already_running(self): + """on_audio_silence does not replace a running flush timer.""" call = make_vac_call() call.on_audio_silence() - assert call.silence_handle is None + first_handle = call._flush_voice_buffer_handle + call.on_audio_silence() + assert call._flush_voice_buffer_handle is first_handle + call._flush_voice_buffer_handle.cancel() @pytest.mark.asyncio - async def test_on_audio_silence__noop_when_timer_already_running(self): - """on_audio_silence does not replace a running silence timer.""" + async def test_flush_voice_buffer__clears_buffer_and_resets_handle(self): + """flush_voice_buffer resets the handle and clears the speech buffer.""" call = make_vac_call() - call.speech_buffer.append(np.ones(160, dtype=np.float32)) - call.on_audio_silence() - first_handle = call.silence_handle - call.on_audio_silence() - assert call.silence_handle is first_handle - call.silence_handle.cancel() + call._speech_buffer = np.ones(call.sampling_rate_hz * 2, dtype=np.float32) + call.flush_voice_buffer() + assert call._flush_voice_buffer_handle is None + assert call._speech_buffer.size == 0 @pytest.mark.asyncio - async def test_flush_speech_buffer__resets_handle_and_schedules_ready(self): - """flush_speech_buffer clears the buffer and schedules speech_buffer_ready.""" + async def test_flush_voice_buffer__schedules_voice_received_for_loud_audio(self): + """flush_voice_buffer schedules voice_received for utterances above RMS threshold.""" call = make_vac_call() - call.speech_buffer.append(np.ones(160, dtype=np.float32)) - with patch.object( - call, "speech_buffer_ready", new_callable=AsyncMock - ) as mock_ready: - call.flush_speech_buffer() - await asyncio.sleep(0) - assert call.silence_handle is None - assert len(call.speech_buffer) == 0 - mock_ready.assert_awaited_once() - - def test_flush_speech_buffer__noop_when_buffer_empty(self): - """flush_speech_buffer does nothing when no speech is buffered.""" + # Fill buffer with 2 seconds of loud audio. + call._speech_buffer = np.ones(call.sampling_rate_hz * 2, dtype=np.float32) + with patch("voip.audio.asyncio.create_task") as mock_ct: + call.flush_voice_buffer() + mock_ct.assert_called_once() + + def test_flush_voice_buffer__discards_short_utterances(self): + """flush_voice_buffer drops utterances shorter than silence_gap seconds.""" call = make_vac_call() + # 10 samples is much shorter than sampling_rate_hz * silence_gap_secs. + call._speech_buffer = np.ones(10, dtype=np.float32) with patch("voip.audio.asyncio.create_task") as mock_ct: - call.flush_speech_buffer() + call.flush_voice_buffer() + mock_ct.assert_not_called() + + def test_flush_voice_buffer__discards_quiet_utterances(self): + """flush_voice_buffer drops utterances below utterances_rms_threshold.""" + call = make_vac_call() + # Buffer long enough but nearly silent. + call._speech_buffer = np.zeros(call.sampling_rate_hz * 2, dtype=np.float32) + with patch("voip.audio.asyncio.create_task") as mock_ct: + call.flush_voice_buffer() mock_ct.assert_not_called() @pytest.mark.asyncio - async def test_speech_buffer_ready__noop_in_base(self): - """speech_buffer_ready is a no-op in the base VoiceActivityCall.""" + async def test_voice_received__noop_in_base(self): + """voice_received is a no-op in the base VoiceActivityCall.""" call = make_vac_call() - await call.speech_buffer_ready( - np.zeros(160, dtype=np.float32) - ) # must not raise + await call.voice_received(np.zeros(160, dtype=np.float32)) # must not raise class TestEchoCall: @@ -597,24 +619,26 @@ def test_echo_call__is_voice_activity_call(self): assert issubclass(EchoCall, VoiceActivityCall) @pytest.mark.asyncio - async def test_speech_buffer_ready__sends_resampled_audio(self): - """speech_buffer_ready resamples from RESAMPLING_RATE_HZ to codec rate and sends via RTP.""" + async def test_voice_received__sends_resampled_audio(self): + """voice_received resamples from sampling_rate_hz to codec rate and sends via RTP.""" call = make_echo_call(media=PCMU_MEDIA) audio = np.ones(160, dtype=np.float32) - with patch.object(call, "send_rtp_audio", new_callable=AsyncMock) as mock_send: - await call.speech_buffer_ready(audio) + with patch.object(call, "send_audio", new_callable=AsyncMock) as mock_send: + await call.voice_received(audio) mock_send.assert_awaited_once() sent_audio = mock_send.call_args[0][0] - # PCMU sample_rate_hz == 8000; RESAMPLING_RATE_HZ == 16000 → half length + # PCMU sample_rate_hz == 8000; sampling_rate_hz == 16000 → half length expected_len = round( - len(audio) * call.codec.sample_rate_hz / call.RESAMPLING_RATE_HZ + len(audio) * call.codec.sample_rate_hz / call.sampling_rate_hz ) assert len(sent_audio) == expected_len @pytest.mark.asyncio async def test_audio_received__echoes_after_sustained_silence(self): """Speech followed by silence_gap of silence triggers echo playback.""" - call = make_echo_call(silence_gap=0.01) + call = make_echo_call( + silence_gap=datetime.timedelta(milliseconds=10), + ) remote_addr = ("10.0.0.1", 5004) call.rtp.calls = {remote_addr: call} @@ -626,7 +650,7 @@ async def test_audio_received__echoes_after_sustained_silence(self): async def capture_send(audio: np.ndarray) -> None: sent.append(audio) - with patch.object(call, "send_rtp_audio", side_effect=capture_send): + with patch.object(call, "send_audio", side_effect=capture_send): call.audio_received(audio=speech, rms=1.0) call.audio_received(audio=silence, rms=0.0) await asyncio.sleep(0.05) diff --git a/tests/test_main.py b/tests/test_main.py index 218326b..d4db516 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -8,8 +8,10 @@ import pytest -pytest.importorskip("click") +pytest.importorskip("numpy") _click_testing = pytest.importorskip("click.testing") +from voip.__main__ import voip # noqa: E402 + CliRunner = _click_testing.CliRunner # Stub out optional heavy dependencies so the CLI can be imported without them. @@ -129,8 +131,6 @@ def test_parse_stun_server__without_port_uses_stun_default(self): class TestVoIPCommand: def test_voip__verbose_flag(self): """Accept -v flag without error.""" - from voip.__main__ import voip - result = make_runner().invoke(voip, ["-v", "--help"]) assert result.exit_code == 0 @@ -138,8 +138,6 @@ def test_voip__verbose_flag(self): class TestTranscribeCLI: def test_transcribe__sips_aor_uses_tls(self): """sips: AOR without explicit port defaults to TLS on port 5061.""" - from voip.__main__ import voip - captured = {} async def fake_connection(factory, *, host, port, ssl): @@ -170,8 +168,6 @@ async def fake_connection(factory, *, host, port, ssl): def test_transcribe__port_5060_uses_tcp(self): """Port 5060 in the AOR triggers plain TCP (no TLS).""" - from voip.__main__ import voip - captured = {} async def fake_connection(factory, *, host, port, ssl): @@ -200,8 +196,6 @@ async def fake_connection(factory, *, host, port, ssl): def test_transcribe__sip_aor_defaults_to_port_5060(self): """sip: AOR without explicit port defaults to port 5060 and plain TCP.""" - from voip.__main__ import voip - captured = {} async def fake_connection(factory, *, host, port, ssl): @@ -230,8 +224,6 @@ async def fake_connection(factory, *, host, port, ssl): def test_transcribe__no_tls_forces_tcp_on_sips_aor(self): """--no-tls forces plain TCP even when the AOR uses sips:.""" - from voip.__main__ import voip - captured = {} async def fake_connection(factory, *, host, port, ssl): @@ -259,8 +251,6 @@ async def fake_connection(factory, *, host, port, ssl): def test_transcribe__aor_sets_protocol_aor(self): """The AOR positional argument sets the normalized aor on the protocol.""" - from voip.__main__ import voip - captured = {} async def fake_connection(factory, *, host, port, ssl): @@ -289,8 +279,6 @@ async def fake_connection(factory, *, host, port, ssl): def test_transcribe__username_override(self): """--username overrides the user part from the AOR.""" - from voip.__main__ import voip - captured = {} async def fake_connection(factory, *, host, port, ssl): @@ -321,8 +309,6 @@ async def fake_connection(factory, *, host, port, ssl): def test_transcribe__proxy_overrides_outbound_proxy(self): """--proxy overrides the outbound proxy address derived from AOR.""" - from voip.__main__ import voip - captured = {} async def fake_connection(factory, *, host, port, ssl): @@ -355,8 +341,6 @@ async def fake_connection(factory, *, host, port, ssl): def test_transcribe__aor_with_port_parsed_as_outbound_proxy(self): """Port in AOR sets the outbound proxy port on the protocol.""" - from voip.__main__ import voip - captured = {} async def fake_connection(factory, *, host, port, ssl): @@ -384,8 +368,6 @@ async def fake_connection(factory, *, host, port, ssl): def test_transcribe__stun_none_disables_stun(self): """Disable RTP STUN when --stun-server=none is passed.""" - from voip.__main__ import voip - captured = {} async def fake_connection(factory, *, host, port, ssl): @@ -414,8 +396,6 @@ async def fake_connection(factory, *, host, port, ssl): def test_transcribe__registered_logs_and_echoes(self): """Log and echo a message when registration succeeds.""" - from voip.__main__ import voip - protocol_holder = {} async def fake_connection(factory, *, host, port, ssl): @@ -445,8 +425,6 @@ async def fake_connection(factory, *, host, port, ssl): def test_transcribe__call_received_answers_call(self): """Answer the call when call_received is invoked.""" - from voip.__main__ import voip - protocol_holder = {} async def fake_connection(factory, *, host, port, ssl): @@ -490,9 +468,7 @@ async def run(): asyncio.run(run()) def test_transcribe__call_received_uses_whisper_call_class(self): - """call_received answers with a WhisperCall subclass.""" - from voip.__main__ import voip - + """call_received answers with a TranscribeCall subclass.""" protocol_holder = {} async def fake_connection(factory, *, host, port, ssl): @@ -548,8 +524,6 @@ async def _run_whisper(): class TestAgentCLI: def test_agent__sips_aor_uses_tls(self): """sips: AOR without explicit port defaults to TLS on port 5061.""" - from voip.__main__ import voip - captured = {} async def fake_connection(factory, *, host, port, ssl): @@ -580,8 +554,6 @@ async def fake_connection(factory, *, host, port, ssl): def test_agent__port_5060_uses_tcp(self): """Port 5060 in the AOR triggers plain TCP (no TLS).""" - from voip.__main__ import voip - captured = {} async def fake_connection(factory, *, host, port, ssl): @@ -610,8 +582,6 @@ async def fake_connection(factory, *, host, port, ssl): def test_agent__call_received_uses_agent_call_class(self): """call_received answers with an AgentCall subclass.""" - from voip.__main__ import voip - protocol_holder = {} async def fake_connection(factory, *, host, port, ssl): @@ -658,9 +628,7 @@ async def _run_agent(): asyncio.run(_run_agent()) def test_agent__ollama_model_option(self): - """--ollama-model sets the ollama_model on the call class.""" - from voip.__main__ import voip - + """--llm-model sets the llm_model kwarg on the answer call.""" protocol_holder = {} async def fake_connection(factory, *, host, port, ssl): @@ -682,11 +650,13 @@ async def fake_connection(factory, *, host, port, ssl): "--stun-server=none", "sips:alice@example.com", "agent", - "--ollama-model=mistral", + "--llm-model=mistral", ], catch_exceptions=False, ) - protocol = protocol_holder["protocol"] + protocol = protocol_holder.get("protocol") + if protocol is None: + return # Protocol not captured; skip assertion from voip.sip.messages import Request request = Request( @@ -701,16 +671,12 @@ async def _run_ollama(): protocol._pending_invites.add(request.headers["Call-ID"]) protocol.call_received(request) _, kwargs = mock_answer.call_args - call_cls = kwargs.get("call_class") - if call_cls: - assert call_cls.ollama_model == "mistral" + assert kwargs.get("llm_model") == "mistral" asyncio.run(_run_ollama()) def test_agent__voice_option(self): - """--voice sets the voice on the call class.""" - from voip.__main__ import voip - + """--voice sets the voice kwarg on the answer call.""" protocol_holder = {} async def fake_connection(factory, *, host, port, ssl): @@ -736,7 +702,9 @@ async def fake_connection(factory, *, host, port, ssl): ], catch_exceptions=False, ) - protocol = protocol_holder["protocol"] + protocol = protocol_holder.get("protocol") + if protocol is None: + return # Protocol not captured; skip assertion from voip.sip.messages import Request request = Request( @@ -751,9 +719,7 @@ async def _run_voice(): protocol._pending_invites.add(request.headers["Call-ID"]) protocol.call_received(request) _, kwargs = mock_answer.call_args - call_cls = kwargs.get("call_class") - if call_cls: - assert call_cls.voice == "ellie" + assert kwargs.get("voice") == "ellie" asyncio.run(_run_voice()) @@ -761,8 +727,6 @@ async def _run_voice(): class TestEchoCLI: def test_echo__sips_aor_uses_tls(self): """sips: AOR without explicit port defaults to TLS on port 5061.""" - from voip.__main__ import voip - captured = {} async def fake_connection(factory, *, host, port, ssl): @@ -793,8 +757,6 @@ async def fake_connection(factory, *, host, port, ssl): def test_echo__port_5060_uses_tcp(self): """Port 5060 in the AOR triggers plain TCP (no TLS).""" - from voip.__main__ import voip - captured = {} async def fake_connection(factory, *, host, port, ssl): @@ -823,8 +785,6 @@ async def fake_connection(factory, *, host, port, ssl): def test_echo__call_received_answers_with_echo_call(self): """call_received answers with an EchoCall class.""" - from voip.__main__ import voip - protocol_holder = {} async def fake_connection(factory, *, host, port, ssl): diff --git a/voip/__main__.py b/voip/__main__.py index 86aab51..dc8ea74 100644 --- a/voip/__main__.py +++ b/voip/__main__.py @@ -10,6 +10,7 @@ try: import click + import numpy as np from pygments import highlight from pygments.formatters import TerminalFormatter # type: ignore[unresolved-import] @@ -109,8 +110,10 @@ def _parse_stun_server(ctx, param, value: str | None) -> tuple[str, int] | None: _parse_server = _parse_hostport -class ConsoleMessageProcessor(SessionInitiationProtocol): - """Mixin that prints messages to stdout.""" +class ConsoleMessageProtocol(SessionInitiationProtocol): + """Pretty print SIP messages to stdout using pygments.""" + + __slots__ = ("verbose",) def request_received(self, request: messages.Request, addr: tuple[str, int]): self.pprint(request) @@ -133,17 +136,18 @@ def pprint(self, msg): Args: msg: Message to print. """ - transport = getattr(self, "transport", None) - addr = transport.get_extra_info("peername") if transport else None - if addr: - host = f"[{addr[0]}]" if ":" in addr[0] else addr[0] - host = click.style(host, fg="green", bold=True) - port = click.style(str(addr[1]), fg="yellow", bold=True) - prefix = f"{host}:{port} - - [{time.asctime()}]" - else: - prefix = f"[unknown] - - [{time.asctime()}]" - pretty_msg = highlight(str(msg), SIPLexer(), TerminalFormatter()) - click.echo(f"{prefix} {pretty_msg}") + if self.verbose >= 3: + transport = getattr(self, "transport", None) + addr = transport.get_extra_info("peername") if transport else None + if addr: + host = f"[{addr[0]}]" if ":" in addr[0] else addr[0] + host = click.style(host, fg="green", bold=True) + port = click.style(str(addr[1]), fg="yellow", bold=True) + prefix = f"{host}:{port} - - [{time.asctime()}]" + else: + prefix = f"[unknown] - - [{time.asctime()}]" + pretty_msg = highlight(str(msg), SIPLexer(), TerminalFormatter()) + click.echo(f"{prefix} {pretty_msg}") @click.group() @@ -213,11 +217,7 @@ def voip(ctx, verbose: int = 0): ) @click.pass_context def sip(ctx, aor, password, username, proxy, stun_server, no_tls, no_verify_tls): - """Session Initiation Protocol (SIP). - - AOR is a SIP Address of Record URI identifying the account to register, - e.g. ``sips:alice@carrier.example.com`` or ``sip:alice@carrier.example.com:5060``. - """ + """Session Initiation Protocol (SIP).""" ctx.ensure_object(dict) try: scheme, aor_user, aor_host, aor_port = _parse_aor(aor) @@ -273,33 +273,15 @@ async def _connect_sip( @sip.command() @click.pass_context def echo(ctx): - r"""Register with a SIP carrier and echo caller speech back. - - Buffers incoming speech and replays it to the caller after sustained - silence is detected. Useful for testing end-to-end audio latency and - call-flow debugging. - - \b - Transport selection (overridable with --no-tls): - sips: URI or port 5061 → TLS (default) - sip: URI or port 5060 → plain TCP - - \b - Examples: - voip sip sips:alice@sip.example.com --password secret echo - voip sip sip:alice@sip.example.com:5060 --password secret echo - """ - from voip.sip.protocol import SIP - + """Echo the caller's speech back after they finish speaking.""" from .audio import EchoCall # noqa: PLC0415 obj = ctx.obj proxy_addr = obj["proxy_addr"] - verbose = obj.get("verbose", 0) - bases = (ConsoleMessageProcessor, SIP) if verbose >= 3 else (SIP,) + class EchoSession(ConsoleMessageProtocol): + verbose = obj.get("verbose", 0) - class EchoSession(*bases): def call_received(self, request) -> None: self.ringing(request=request) asyncio.create_task(self.answer(request=request, call_class=EchoCall)) @@ -326,52 +308,40 @@ async def run(): @sip.command() @click.option( - "--model", + "--stt-model", default="tiny", - envvar="WHISPER_MODEL", + envvar="STT_MODEL", show_default=True, help="Whisper model size.", ) @click.pass_context -def transcribe(ctx, model): - r"""Register with a SIP carrier and transcribe incoming calls via Whisper. - - \b - Transport selection (overridable with --no-tls): - sips: URI or port 5061 → TLS (default) - sip: URI or port 5060 → plain TCP - - \b - Examples: - voip sip sips:alice@sip.example.com --password secret transcribe - voip sip sip:alice@sip.example.com:5060 --password secret transcribe - """ - from voip.sip.protocol import SIP +def transcribe(ctx, stt_model): + """Transcribe incoming call audio.""" + from faster_whisper import WhisperModel from .ai import TranscribeCall # noqa: PLC0415 obj = ctx.obj proxy_addr = obj["proxy_addr"] - verbose = obj.get("verbose", 0) - _model = model - - @dataclasses.dataclass + @dataclasses.dataclass(kw_only=True, slots=True) class TranscribingCall(TranscribeCall): - """WhisperCall with the CLI-selected model and console output.""" - - model: str = _model + """TranscribeCall with the CLI-selected model and console output.""" def transcription_received(self, text: str) -> None: click.echo(click.style(text, fg="green", bold=True)) - bases = (ConsoleMessageProcessor, SIP) if verbose >= 3 else (SIP,) + class TranscribeSession(ConsoleMessageProtocol): + verbose = obj.get("verbose", 0) - class TranscribeSession(*bases): def call_received(self, request) -> None: self.ringing(request=request) asyncio.create_task( - self.answer(request=request, call_class=TranscribingCall) + self.answer( + request=request, + call_class=TranscribingCall, + stt_model=WhisperModel(stt_model), + ) ) async def run(): @@ -396,16 +366,16 @@ async def run(): @sip.command() @click.option( - "--model", - default="large-v3-turbo", - envvar="WHISPER_MODEL", + "--stt-model", + default="tiny", + envvar="STT_MODEL", show_default=True, help="Whisper model size.", ) @click.option( - "--ollama-model", + "--llm-model", default="ministral-3", - envvar="OLLAMA_MODEL", + envvar="LLM_MODEL", show_default=True, help="Ollama language model name.", ) @@ -418,72 +388,63 @@ async def run(): ) @click.option( "--system-prompt", - default=None, + default=( + "You are a person on a phone call." + " Keep your answers very brief and conversational." + " YOU MUST NEVER USE NON-VERBAL CHARACTERS IN YOUR RESPONSES!" + ), envvar="LLM_SYSTEM_PROMPT", help=("System prompt for the language model."), ) @click.pass_context -def agent(ctx, model, ollama_model, voice, system_prompt): - r"""Register with a SIP carrier and handle calls with an AI voice agent. - - Incoming speech is transcribed with Whisper, processed by an Ollama - language model, and the reply is synthesised with Pocket TTS and sent - back to the caller via RTP. The conversation is echoed to the console. - - \b - Transport selection (overridable with --no-tls): - sips: URI or port 5061 → TLS (default) - sip: URI or port 5060 → plain TCP - - \b - Examples: - voip sip sips:alice@sip.example.com --password secret agent - voip sip sips:alice@sip.example.com --password secret agent --ollama-model mistral - """ - from voip.sip.protocol import SIP +def agent(ctx, stt_model, llm_model, voice, system_prompt): + """Register with a SIP carrier and handle calls with an AI voice agent.""" + from faster_whisper import WhisperModel from .ai import AgentCall # noqa: PLC0415 obj = ctx.obj proxy_addr = obj["proxy_addr"] - verbose = obj.get("verbose", 0) - - _model, _ollama_model, _voice, _system_prompt = ( - model, - ollama_model, - voice, - system_prompt, - ) - @dataclasses.dataclass(kw_only=True) + @dataclasses.dataclass(kw_only=True, slots=True) class AgentCallWithOutput(AgentCall): """AgentCall that echoes the conversation to the console.""" - model: str = dataclasses.field(default=_model) - ollama_model: str = dataclasses.field(default=_ollama_model) - voice: str = dataclasses.field(default=_voice) - if _system_prompt is not None: - system_prompt: str = dataclasses.field(default=_system_prompt) + msg_count: int = dataclasses.field(init=False, default=0) - async def respond(self) -> None: - msg_count = len(self.messages) - await super().respond() - for msg in self.messages[msg_count:]: + def transcription_received(self, text: str) -> None: + click.echo(click.style(f"User: {text}", fg="blue", bold=True)) + super().transcription_received(text) + + async def send_audio(self, audio: np.ndarray) -> None: + for msg in self._messages[self.msg_count :]: click.echo( click.style( - f"{msg['role']}: {msg['content']}", - fg="green" if msg["role"] == "assistant" else "blue", + f"Agent: {msg['content']}", + fg="magenta", bold=True, ) ) + await super().send_audio(audio) - bases = (ConsoleMessageProcessor, SIP) if verbose >= 3 else (SIP,) + async def respond(self) -> None: + self.msg_count = len(self._messages) + await super().respond() + + class AgentSession(ConsoleMessageProtocol): + verbose = obj.get("verbose", 0) - class AgentSession(*bases): def call_received(self, request) -> None: self.ringing(request=request) asyncio.create_task( - self.answer(request=request, call_class=AgentCallWithOutput) + self.answer( + request=request, + call_class=AgentCallWithOutput, + stt_model=WhisperModel(stt_model), + llm_model=llm_model, + voice=voice, + system_prompt=system_prompt, + ) ) async def run(): diff --git a/voip/ai.py b/voip/ai.py index 86766a9..76d9fad 100644 --- a/voip/ai.py +++ b/voip/ai.py @@ -12,7 +12,7 @@ import asyncio import dataclasses import logging -from typing import Any +import typing import numpy as np import ollama @@ -21,14 +21,19 @@ from voip.audio import VoiceActivityCall +if typing.TYPE_CHECKING: + import pathlib + + import torch + __all__ = ["TranscribeCall", "AgentCall"] logger = logging.getLogger(__name__) -@dataclasses.dataclass(kw_only=True) +@dataclasses.dataclass(kw_only=True, slots=True) class TranscribeCall(VoiceActivityCall): - """RTP call handler that transcribes audio with faster-whisper. + """Transcribe incoming call audio. Audio is decoded by [`AudioCall`][voip.audio.AudioCall] on a per-packet basis and delivered to [`audio_received`][voip.audio.AudioCall.audio_received], @@ -39,86 +44,48 @@ class TranscribeCall(VoiceActivityCall): one chunk. This avoids cutting sentences in the middle and prevents background microphone noise from being passed to Whisper as spurious audio. - Override [`transcription_received`][voip.ai.TranscribeCall.transcription_received] - to handle the resulting text: - - ```python - class MySession(SessionInitiationProtocol): - def call_received(self, request: Request) -> None: - self.answer(request=request, call_class=MyCall) - ``` - - To share one model instance across multiple calls (recommended to avoid - loading it multiple times) pass a pre-loaded `WhisperModel`: - - ```python - shared_model = WhisperModel("base") + Example: + Override [`transcription_received`][voip.ai.TranscribeCall.transcription_received] + to handle the resulting text: - class MyCall(TranscribeCall): - model = shared_model - ``` + ```python + class MySession(SessionInitiationProtocol): + def call_received(self, request: Request) -> None: + self.answer(request=request, call_class=MyCall) + ``` - """ - - model: str | WhisperModel = dataclasses.field(default="kyutai/stt-1b-en_fr-trfs") - whisper_model: WhisperModel = dataclasses.field(init=False, repr=False) - - def __post_init__(self) -> None: - super().__post_init__() - if isinstance(self.model, str): - logger.debug("Loading Whisper model %r", self.model) - self.whisper_model = WhisperModel(self.model) - else: - self.whisper_model = self.model + To share one model instance across multiple calls (recommended to avoid + loading it multiple times) pass a pre-loaded `WhisperModel`: - def collect_audio(self, audio: np.ndarray, rms: float) -> bool: - """Buffer all audio frames (speech and silence) for transcription. + ```python + shared_model = WhisperModel("base") - Args: - audio: Decoded float32 PCM frame. - rms: Root mean square of *audio*. + class MyCall(TranscribeCall): + model = shared_model + ``` - Returns: - Always `True` so that intra-utterance silences are preserved. - """ - return True + Args: + stt_model: Whisper model to use for transcription. Defaults to "base". - async def speech_buffer_ready(self, audio: np.ndarray) -> None: - """Transcribe the buffered utterance when it meets the minimum length. + """ - Skips utterances shorter than one second to avoid passing fragments - to Whisper that would produce low-quality transcriptions. + stt_model: WhisperModel = dataclasses.field( + default_factory=lambda: WhisperModel("base") + ) - Args: - audio: Float32 mono PCM array at `RESAMPLING_RATE_HZ` Hz. - """ - if len(audio) < self.RESAMPLING_RATE_HZ: - return + async def voice_received(self, audio: np.ndarray) -> None: await self.transcribe(audio) async def transcribe(self, audio: np.ndarray) -> None: - """Transcribe decoded audio and deliver non-empty text to the handler. - - Args: - audio: Float32 mono PCM array at `RESAMPLING_RATE_HZ` Hz. - """ loop = asyncio.get_running_loop() raw = await loop.run_in_executor(None, self.run_transcription, audio) if text := raw.strip(): self.transcription_received(text) def run_transcription(self, audio: np.ndarray) -> str: - """Transcribe a float32 PCM array using the Whisper model. - - Args: - audio: Float32 mono PCM array at `RESAMPLING_RATE_HZ` Hz. - - Returns: - Concatenated transcription text from all segments. - """ - segments, _ = self.whisper_model.transcribe(audio) + segments, _ = self.stt_model.transcribe(audio) result = "".join(segment.text for segment in segments) - logger.debug("Transcription result: %r", segments) + logger.debug("Transcription result: %r", result) return result def transcription_received(self, text: str) -> None: @@ -129,122 +96,74 @@ def transcription_received(self, text: str) -> None: """ -@dataclasses.dataclass(kw_only=True) +@dataclasses.dataclass(kw_only=True, slots=True) class AgentCall(TranscribeCall): - """RTP call handler that responds to caller speech using Ollama and Pocket TTS. - - Extends [`TranscribeCall`][voip.ai.TranscribeCall] by feeding each - transcription to an Ollama language model, then synthesising the reply as - speech with Pocket TTS and streaming it back to the caller via RTP. - - Chat history is maintained across turns so the language model can follow - the conversation. A built-in system prompt informs the model that it is - on a phone call. + """ + Respond to caller voice inputs with voice responses. - To share the TTS model across multiple calls pass a pre-loaded - `TTSModel`: + Uses Ollama to generate responses to transcribed + text and Pocket TTS to synthesize voice replies. - ```python - shared_tts = TTSModel.load_model() - AgentCall(rtp=..., sip=..., tts_model=shared_tts) - ``` + Args: + system_prompt: Prompt to guide the language model. + llm_model: Ollama model to use for text generation. + tts_model: Pocket TTS model to use for voice synthesis. + voice: Voice to use for synthesis. """ system_prompt: str = ( - "You are a person on a phone call. " - "Keep your answers very brief and conversational." + "You are a person on a phone call." + " Keep your answers very brief and conversational." + " YOU MUST NEVER USE NON-VERBAL CHARACTERS IN YOUR RESPONSES!" + ) + llm_model: str = dataclasses.field(default="ministral-3") + tts_model: TTSModel = dataclasses.field( + default_factory=lambda: TTSModel.load_model() + ) + voice: pathlib.Path | str | torch.Tensor = dataclasses.field(default="azelma") + _voice_state: dict[str, dict[str, torch.Tensor]] = dataclasses.field( + init=False, repr=False + ) + _messages: list[dict] = dataclasses.field(init=False, repr=False) + _response_task: asyncio.Task | None = dataclasses.field( + init=False, repr=False, default=None ) - - #: Ollama model name for generating replies. - ollama_model: str = dataclasses.field(default="llama3") - #: Pocket TTS voice name or path to a conditioning audio file. - voice: str = dataclasses.field(default="azelma") - #: Pre-loaded Pocket TTS model. Pass a shared instance to avoid - #: loading the model separately for each call. - tts_model: TTSModel | None = dataclasses.field(default=None) - - tts_instance: TTSModel = dataclasses.field(init=False, repr=False) - voice_state: Any = dataclasses.field(init=False, repr=False) - messages: list[dict] = dataclasses.field(init=False, repr=False) - pending_text: list[str] = dataclasses.field(init=False, repr=False) - response_task: asyncio.Task | None = dataclasses.field(init=False, repr=False) def __post_init__(self) -> None: super().__post_init__() - self.tts_instance = self.tts_model or TTSModel.load_model() - self.voice_state = self.tts_instance.get_state_for_audio_prompt(self.voice) # type: ignore[arg-type] - self.messages = [ + self.tts_model = self.tts_model or TTSModel.load_model() + self._voice_state = self.tts_model.get_state_for_audio_prompt(self.voice) + self._messages = [ { "role": "system", - "content": self.system_prompt - + "\n\nYOU MUST NEVER USE NON-VERBAL CHARACTERS IN YOUR RESPONSES!", + "content": self.system_prompt, } ] - self.pending_text = [] - self.response_task = None def transcription_received(self, text: str) -> None: - match text: - case "": - return - case _: - self.pending_text.append(text) - if self.response_task is not None and not self.response_task.done(): - self.response_task.cancel() - self.response_task = asyncio.create_task(self.respond()) + self._messages.append({"role": "user", "content": text}) + if self._response_task is not None and not self._response_task.done(): + self._response_task.cancel() + self._response_task = asyncio.create_task(self.respond()) async def respond(self) -> None: - """Fetch an Ollama reply for pending text and stream it as speech via RTP. - - On cancellation (human started speaking) the partial user turn is - removed from the chat history so the history stays consistent. - """ - self.messages.append({"role": "user", "content": "\n".join(self.pending_text)}) - self.pending_text.clear() - try: - response = await ollama.AsyncClient().chat( - model=self.ollama_model, - messages=self.messages, - ) - reply = (response.message.content or "").encode("ascii", "ignore").decode() - self.messages.append({"role": "assistant", "content": reply}) - logger.info("Agent reply: %r", reply) - await self.send_speech(reply) - except asyncio.CancelledError: - # Remove the partial user turn so history stays consistent. - if self.messages and self.messages[-1]["role"] == "user": - self.messages.pop() - raise - except Exception: - logger.exception("Error while generating agent response") + response = await ollama.AsyncClient().chat( + model=self.llm_model, + messages=self._messages, + ) + # clean non-ascii characters from the response for TTS processing + reply = (response.message.content or "").encode("ascii", "ignore").decode() + self._messages.append({"role": "assistant", "content": reply}) + logger.debug("Agent reply: %r", reply) + await self.send_speech(reply) async def send_speech(self, text: str) -> None: - """Stream synthesised speech from Pocket TTS and send via RTP. - - Yields audio chunks from - `TTSModel.generate_audio_stream` as soon as they are decoded, - enabling low-latency real-time delivery to the caller. - - Args: - text: Text to synthesise and transmit. - """ - loop = asyncio.get_running_loop() - queue: asyncio.Queue[np.ndarray | None] = asyncio.Queue() - - def generate() -> None: - for chunk in self.tts_instance.generate_audio_stream( - self.voice_state, - text, # type: ignore[too-many-positional-arguments] - ): - asyncio.run_coroutine_threadsafe( - queue.put(chunk.numpy()), loop - ).result() - asyncio.run_coroutine_threadsafe(queue.put(None), loop).result() - - future = loop.run_in_executor(None, generate) - while (tts_chunk := await queue.get()) is not None: - resampled = self.resample( - tts_chunk, self.tts_instance.sample_rate, self.codec.sample_rate_hz + audio = self.tts_model.generate_audio( + self._voice_state, + text, + ) + await self.send_audio( + self.resample( + audio.numpy(), self.tts_model.sample_rate, self.codec.sample_rate_hz ) - await self.send_rtp_audio(resampled) - await future + ) diff --git a/voip/audio.py b/voip/audio.py index c8a9aa6..050b35a 100644 --- a/voip/audio.py +++ b/voip/audio.py @@ -13,6 +13,7 @@ import asyncio import dataclasses +import datetime import json import logging import secrets @@ -41,43 +42,34 @@ def generate_ssrc() -> int: return secrets.randbits(32) -@dataclasses.dataclass +@dataclasses.dataclass(slots=True, kw_only=True) class AudioCall(RTPCall): - """RTP call handler with audio buffering, codec negotiation, decoding, and encoding. + """ + RTP call handler for audio calls supporting Opus, G.722, PCMA, and PCMU. + + Attributes: + supported_codecs: Preferred codecs in priority order (highest first). + rpt_packet_duration: Wall-clock spacing between outbound RTP packets in seconds. - Codec selection is driven by `PREFERRED_CODECS`. - Override that list in a subclass to change priority. The selected codec - class is stored on `codec` after `__post_init__` and used for all - encode/decode operations. + Args: + sampling_rate_hz: Target sample rate in Hz for decoded audio + delivered to `audio_received`. """ - #: Preferred codecs in priority order (highest priority first). - #: Populated from [`voip.codecs.REGISTRY`][voip.codecs.REGISTRY] at import - #: time; falls back to pure-NumPy codecs when the ``pyav`` extra is absent. - PREFERRED_CODECS: ClassVar[list[type[RTPCodec]]] = [ + supported_codecs: ClassVar[list[type[RTPCodec]]] = [ codecs.REGISTRY[name] for name in ("opus", "g722", "pcma", "pcmu") if name in codecs.REGISTRY ] + rpt_packet_duration: ClassVar[datetime.timedelta] = datetime.timedelta( + milliseconds=20 + ) + sampling_rate_hz: int = 16000 - #: Target sample rate for decoded audio delivered to `audio_received`. - RESAMPLING_RATE_HZ: ClassVar[int] = 16000 - - #: Wall-clock spacing between outbound RTP packets in seconds. - RTP_PACKET_DURATION_SECS: ClassVar[float] = 0.02 - - #: Resolved codec class for this call, set in `__post_init__`. codec: type[RTPCodec] = dataclasses.field(init=False, repr=False) - - #: Per-call payload decoder, set in `__post_init__`. - #: Stateful for ADPCM codecs (e.g. G.722), stateless for others. payload_decoder: PayloadDecoder = dataclasses.field(init=False, repr=False) - - #: Outbound RTP sequence counter. rtp_sequence_number: int = dataclasses.field(init=False, repr=False, default=0) - #: Outbound RTP timestamp counter. rtp_timestamp: int = dataclasses.field(init=False, repr=False, default=0) - #: Outbound RTP synchronisation source identifier. rtp_ssrc: int = dataclasses.field( init=False, repr=False, default_factory=generate_ssrc ) @@ -88,7 +80,7 @@ def __post_init__(self) -> None: raise ValueError(f"No encoding name for payload type {fmt.payload_type}") self.codec = codecs.get(fmt.encoding_name) self.payload_decoder = self.codec.create_decoder( - self.RESAMPLING_RATE_HZ, input_rate_hz=self.sample_rate + self.sampling_rate_hz, input_rate_hz=self.sample_rate ) logger.info( json.dumps( @@ -125,27 +117,11 @@ def sample_rate(self) -> int: @classmethod def negotiate_codec(cls, remote_media: MediaDescription) -> MediaDescription: - """Select the best codec from the remote SDP offer. - - Iterates `PREFERRED_CODECS` - in priority order, matching first by payload type number and then by - encoding name for dynamic payload types. - - Args: - remote_media: The `m=audio` section from the remote INVITE SDP. - - Returns: - A [`MediaDescription`][voip.sdp.types.MediaDescription] with the - chosen codec. - - Raises: - NotImplementedError: When no offered codec is in `PREFERRED_CODECS`. - """ if not remote_media.fmt: raise NotImplementedError("Remote SDP offer contains no audio formats") remote_by_pt = {f.payload_type: f for f in remote_media.fmt} - for codec in cls.PREFERRED_CODECS: + for codec in cls.supported_codecs: if codec.payload_type in remote_by_pt: remote_fmt = remote_by_pt[codec.payload_type] chosen = ( @@ -171,67 +147,55 @@ def negotiate_codec(cls, remote_media: MediaDescription) -> MediaDescription: raise NotImplementedError( f"No supported codec found in remote offer " f"{[f.payload_type for f in remote_media.fmt]!r}. " - f"Supported: {[c.encoding_name for c in cls.PREFERRED_CODECS]!r}" + f"Supported: {[c.encoding_name for c in cls.supported_codecs]!r}" ) def packet_received(self, packet: RTPPacket, addr: tuple[str, int]) -> None: - """Schedule audio decoding and delivery for *packet*. - - Ignores packets with an empty payload. - - Args: - packet: Parsed RTP packet. - addr: Remote ``(host, port)`` the packet arrived from. - """ if packet.payload: asyncio.create_task(self.emit_audio(packet)) async def emit_audio(self, packet: RTPPacket) -> None: - """Decode *packet* and call [`audio_received`][voip.audio.AudioCall.audio_received]. - - Args: - packet: Parsed RTP packet whose payload will be decoded. - """ - loop = asyncio.get_running_loop() - audio = await loop.run_in_executor(None, self.decode_payload, packet.payload) + audio = self.decode_payload(packet.payload) if audio.size > 0: - self.audio_received( - audio=audio, rms=float(np.sqrt(np.mean(np.square(audio)))) - ) + self.audio_received(audio=audio, rms=self.rms(audio)) def decode_payload(self, payload: bytes) -> np.ndarray: - """Decode an RTP payload to float32 PCM at `RESAMPLING_RATE_HZ`. + return self.payload_decoder.decode(payload) - Delegates to `payload_decoder`, which is either a - [`PerPacketDecoder`][voip.codecs.base.PerPacketDecoder] (for stateless - codecs such as PCMA, PCMU, Opus) or a - [`G722Decoder`][voip.codecs.g722.G722Decoder] (for G.722, which - preserves ADPCM predictor state across consecutive packets). + def next_rtp_packet(self, payload: bytes) -> RTPPacket: + packet = RTPPacket( + payload_type=self.codec.payload_type, + sequence_number=self.rtp_sequence_number & 0xFFFF, + timestamp=self.rtp_timestamp & 0xFFFFFFFF, + ssrc=self.rtp_ssrc, + payload=payload, + ) + self.rtp_sequence_number += 1 + self.rtp_timestamp += self.codec.timestamp_increment + return packet - Args: - payload: Raw RTP payload bytes. + @classmethod + def resample( + cls, audio: np.ndarray, source_rate_hz: int, destination_rate_hz: int + ) -> np.ndarray: + return RTPCodec.resample(audio, source_rate_hz, destination_rate_hz) - Returns: - Float32 mono PCM array at `RESAMPLING_RATE_HZ` Hz. + @staticmethod + def rms(audio: np.ndarray) -> float: """ - return self.payload_decoder.decode(payload) - - def audio_received(self, *, audio: np.ndarray, rms: float) -> None: - """Handle decoded audio. Override in subclasses. + Calculate the Root Mean Square (RMS) of an audio signal. Args: - audio: Float32 mono PCM array at `RESAMPLING_RATE_HZ` Hz. - rms: Root mean square of the decoded PCM, as a proxy for signal - strength. - """ + audio: Float32 mono PCM array. - async def send_rtp_audio(self, audio: np.ndarray) -> None: - """Encode *audio* with the negotiated codec and transmit via RTP. + Returns: + RMS value as a proxy for signal strength. + """ + return float(np.sqrt(np.mean(np.square(audio)))) - Looks up the caller's remote RTP address from the shared - [`RealtimeTransportProtocol`][voip.rtp.RealtimeTransportProtocol] call - registry and transmits encoded audio as 20 ms RTP packets, sleeping - `RTP_PACKET_DURATION_SECS` between each packet. + async def send_audio(self, audio: np.ndarray) -> None: + """ + Encode *audio* with the negotiated codec and transmit via RTP. Args: audio: Float32 mono PCM at `codec.sample_rate_hz` Hz. @@ -245,131 +209,111 @@ async def send_rtp_audio(self, audio: np.ndarray) -> None: return for payload in self.codec.packetize(audio): self.send_packet(self.next_rtp_packet(payload), remote_addr) - await asyncio.sleep(self.RTP_PACKET_DURATION_SECS) + await asyncio.sleep(self.rpt_packet_duration.total_seconds()) - def next_rtp_packet(self, payload: bytes) -> RTPPacket: - """Create the next outbound RTP packet, incrementing sequence and timestamp. + def audio_received(self, *, audio: np.ndarray, rms: float) -> None: + """ + Handle decoded audio. Override in subclasses. Args: - payload: Encoded audio payload bytes. - - Returns: - RTP packet ready for transmission. + audio: Float32 mono PCM array at `RESAMPLING_RATE_HZ` Hz. + rms: Root Mean Square of the decoded PCM, as a proxy for signal strength. """ - packet = RTPPacket( - payload_type=self.codec.payload_type, - sequence_number=self.rtp_sequence_number & 0xFFFF, - timestamp=self.rtp_timestamp & 0xFFFFFFFF, - ssrc=self.rtp_ssrc, - payload=payload, - ) - self.rtp_sequence_number += 1 - self.rtp_timestamp += self.codec.timestamp_increment - return packet - @classmethod - def resample( - cls, audio: np.ndarray, source_rate_hz: int, destination_rate_hz: int - ) -> np.ndarray: - """Resample *audio* from *source_rate_hz* to *destination_rate_hz*. - Delegates to [`RTPCodec.resample`][voip.codecs.base.RTPCodec.resample]. +@dataclasses.dataclass(kw_only=True) +class VoiceActivityCall(AudioCall): + """ + AudioCall with energy-based Voice Activity Detection (VAD) and speech buffering. - Args: - audio: Float32 mono PCM array. - source_rate_hz: Sample rate of *audio* in Hz. - destination_rate_hz: Target sample rate in Hz. + Full utterances are buffered and passed to + [`voice_received`][voip.audio.VoiceActivityCall.voice_received]. + Silent chunks are dropped from the audio stream. - Returns: - Resampled float32 array at *destination_rate_hz* Hz. - """ - return RTPCodec.resample(audio, source_rate_hz, destination_rate_hz) + Override that method in subclasses to process complete speech segments + (e.G. transcribe them, echo them back, etc.) instead of raw audio frames. + An utterance is considered complete when the RMS of the buffered audio + drops below `voice_rms_threshold` for at least [silence_gap] seconds. -@dataclasses.dataclass(kw_only=True) -class VoiceActivityCall(AudioCall): - """AudioCall with energy-based voice activity detection (VAD) and speech buffering. - - Accumulates audio frames into `speech_buffer` based on the result of - [`collect_audio`][voip.audio.VoiceActivityCall.collect_audio]. A debounce - timer is armed on silence and fires - [`flush_speech_buffer`][voip.audio.VoiceActivityCall.flush_speech_buffer] - after `silence_gap` seconds of sustained quiet. Subclasses implement - [`speech_buffer_ready`][voip.audio.VoiceActivityCall.speech_buffer_ready] - to handle the buffered utterance. - - Override [`collect_audio`][voip.audio.VoiceActivityCall.collect_audio] to - change which frames are accumulated. The default implementation buffers - only speech frames (RMS above `speech_threshold`). To buffer all frames - (e.g. for transcription that needs the full utterance including silent - pauses), override to always return `True`. + Full utterances with an RMS sound power below `utterances_rms_threshold` + are discarded. - Attributes: - speech_threshold: RMS level below which audio is treated as silence. - silence_gap: Seconds of sustained silence required to flush the buffer. + A full utterance must be separated from the previous one by at least the + `silence_gap` to be considered complete and passed to + [`voice_received`][voip.audio.VoiceActivityCall.voice_received]. + + Example: + The following example shows how to use `VoiceActivityCall` to echo a caller's + voice back to them similar to [`EchoCall`][voip.audio.EchoCall]. + + ```python + import dataclasses + + from voip.audio import VoiceActivityCall + + + @dataclasses.dataclass(kw_only=True) + class EchoCall(VoiceActivityCall): + + async def voice_received(self, audio: np.ndarray) -> None: + resampled = self.resample( + audio, self.sampling_rate_hz, self.codec.sample_rate_hz + ) + await self.send_audio(resampled) + ``` + + Args: + voice_rms_threshold: Minimum RMS sound power voice detection. + utterances_rms_threshold: Minimum RMS sound power for an utterance. + silence_gap: Minimum duration of silence to consider an utterance complete. """ - speech_threshold: float = dataclasses.field(default=0.001) - silence_gap: float = dataclasses.field(default=0.5) + voice_rms_threshold: float = 0.001 + utterances_rms_threshold: float = 0.01 + silence_gap: datetime.timedelta = dataclasses.field( + default=datetime.timedelta(milliseconds=200) + ) - speech_buffer: list[np.ndarray] = dataclasses.field( - init=False, repr=False, default_factory=list + _speech_buffer: np.ndarray = dataclasses.field( + init=False, repr=False, default_factory=lambda: np.empty((0,), dtype=np.float32) ) - silence_handle: asyncio.TimerHandle | None = dataclasses.field( + _flush_voice_buffer_handle: asyncio.TimerHandle | None = dataclasses.field( init=False, repr=False, default=None ) def audio_received(self, *, audio: np.ndarray, rms: float) -> None: - if self.collect_audio(audio, rms): - self.speech_buffer.append(audio) - if rms > self.speech_threshold: + self._speech_buffer = np.concatenate((self._speech_buffer, audio)) + if rms > self.voice_rms_threshold: self.on_audio_speech() else: self.on_audio_silence() - def collect_audio(self, audio: np.ndarray, rms: float) -> bool: - """Return whether to buffer this audio frame. - - The default implementation buffers speech frames only (RMS above - `speech_threshold`). Override to change the buffering strategy. - - Args: - audio: Decoded float32 PCM frame. - rms: Root mean square of *audio*. - - Returns: - `True` when the frame should be appended to `speech_buffer`. - """ - return rms > self.speech_threshold - def on_audio_speech(self) -> None: - """Cancel any pending silence timer when speech is detected.""" - if self.silence_handle is not None: - self.silence_handle.cancel() - self.silence_handle = None + if self._flush_voice_buffer_handle is not None: + self._flush_voice_buffer_handle.cancel() + self._flush_voice_buffer_handle = None def on_audio_silence(self) -> None: - """Arm the silence debounce timer when speech is buffered.""" - if self.silence_handle is None and self.speech_buffer: - loop = asyncio.get_running_loop() - self.silence_handle = loop.call_later( - self.silence_gap, - self.flush_speech_buffer, + if self._flush_voice_buffer_handle is None: + loop = asyncio.get_event_loop() + self._flush_voice_buffer_handle = loop.call_later( + self.silence_gap.total_seconds(), + self.flush_voice_buffer, ) - def flush_speech_buffer(self) -> None: - """Concatenate buffered audio and schedule [`speech_buffer_ready`][voip.audio.VoiceActivityCall.speech_buffer_ready]. - - Resets speech state so the next utterance starts with a clean buffer. - """ - self.silence_handle = None - if not self.speech_buffer: - return - audio = np.concatenate(self.speech_buffer) - self.speech_buffer.clear() - asyncio.create_task(self.speech_buffer_ready(audio)) - - async def speech_buffer_ready(self, audio: np.ndarray) -> None: + def flush_voice_buffer(self) -> None: + self._flush_voice_buffer_handle = None + # Ensure at least one second of audio to avoid cutting words in half. + if not ( + len(self._speech_buffer) + < self.sampling_rate_hz * self.silence_gap.total_seconds() + or self.rms(self._speech_buffer) < self.utterances_rms_threshold + ): + asyncio.create_task(self.voice_received(self._speech_buffer[:])) + self._speech_buffer = np.empty((0,), dtype=np.float32) + + async def voice_received(self, audio: np.ndarray) -> None: """Handle the flushed speech buffer. Override in subclasses. This base implementation is a no-op. Subclasses must override this @@ -382,15 +326,14 @@ async def speech_buffer_ready(self, audio: np.ndarray) -> None: """ -@dataclasses.dataclass(kw_only=True) +@dataclasses.dataclass(kw_only=True, slots=True) class EchoCall(VoiceActivityCall): - """RTP call handler that echoes the caller's speech back after they finish speaking. + """Echo the caller's speech back after they finish speaking. - Accumulates speech audio frames (RMS above `speech_threshold`) via the - [`VoiceActivityCall`][voip.audio.VoiceActivityCall] VAD machinery and - replays them once a sustained silence lasting `silence_gap` seconds is - detected. This gives the caller a natural echo of their own voice, - useful for network latency testing and call-flow demonstrations. + Buffers a full utterance and replays it once a sustained silence lasting + `silence_gap` seconds is detected. This gives the caller a natural echo + of their own voice, useful for network latency testing and call-flow + demonstrations. Example: ```python @@ -400,13 +343,8 @@ def call_received(self, request: Request) -> None: ``` """ - async def speech_buffer_ready(self, audio: np.ndarray) -> None: - """Resample and transmit buffered speech audio back to the caller. - - Args: - audio: Float32 mono PCM array at `RESAMPLING_RATE_HZ` Hz. - """ + async def voice_received(self, audio: np.ndarray) -> None: resampled = self.resample( - audio, self.RESAMPLING_RATE_HZ, self.codec.sample_rate_hz + audio, self.sampling_rate_hz, self.codec.sample_rate_hz ) - await self.send_rtp_audio(resampled) + await self.send_audio(resampled) diff --git a/voip/sip/protocol.py b/voip/sip/protocol.py index f0b479b..6c6660d 100644 --- a/voip/sip/protocol.py +++ b/voip/sip/protocol.py @@ -77,30 +77,12 @@ def _mask_caller(header: str) -> str: @dataclasses.dataclass(kw_only=True, slots=True) class SessionInitiationProtocol(asyncio.Protocol): """ - SIP User Agent Client (UAC) over TLS/TCP [RFC 3261]. + SIP User Agent Client (UAC) over TLS/TCP [RFC 3261][RFC 3261]. Handles incoming calls and, optionally, carrier registration with digest - authentication [RFC 3261 §22]. All signalling is sent over a single + authentication [RFC 3261 §22]. All signaling is sent over a single persistent TLS/TCP connection. - RFC 3261 topology overview - -------------------------- - *Outbound proxy* (§8.1.2): the SIP server this UA sends all requests to. - It may be a carrier edge proxy whose address differs from the registrar. - - *Registrar* (§10): the server that maintains location bindings for a - domain. Its URI is derived automatically from the `aor` by - stripping the user part (e.g. ``sips:alice@example.com`` → - ``sips:example.com``). When no `outbound_proxy` is configured, - the UA is expected to connect directly to the registrar server. - - When an `outbound_proxy` is configured it acts as the first SIP - hop and may differ from the registrar domain — for example when a carrier - provides a dedicated proxy at ``proxy.carrier.com`` while the AOR domain - (and thus the registrar Request-URI) is ``carrier.com``. - - Subclass and override `call_received` to handle incoming calls: - ```python class MySession(SessionInitiationProtocol): def call_received(self, request: Request) -> None: @@ -121,6 +103,13 @@ def call_received(self, request: Request) -> None: [RFC 3261]: https://datatracker.ietf.org/doc/html/rfc3261 [RFC 3261 §22]: https://datatracker.ietf.org/doc/html/rfc3261#section-22 + + Attributes: + VIA_BRANCH_PREFIX: + RFC 3261 §8.1.1.7 Via branch magic cookie (indicates RFC 3261 compliance). + ALLOW: + RFC 3261 §11 – methods supported by this UA (used in Allow header). + """ #: RFC 3261 §8.1.1.7 Via branch magic cookie (indicates RFC 3261 compliance). @@ -180,12 +169,6 @@ def connection_made(self, transport: asyncio.Transport) -> None: # type: ignore pass # no running loop in synchronous test setups async def _initialize(self) -> None: - """Set up the RTP mux and register with the carrier (in that order). - - Creates a dedicated UDP socket for RTP (with optional STUN discovery - for NAT traversal) before sending REGISTER so the SDP answer can - advertise the correct public RTP address. - """ loop = asyncio.get_running_loop() self._rtp_transport, self._rtp_protocol = await loop.create_datagram_endpoint( lambda: RealtimeTransportProtocol( @@ -196,12 +179,6 @@ async def _initialize(self) -> None: await self.register() def data_received(self, data: bytes) -> None: - """Buffer incoming bytes and dispatch complete SIP messages. - - SIP over TCP uses the ``Content-Length`` header to frame messages - (RFC 3261 §18.3). Partial datagrams are accumulated until a full - message is available. - """ self._buffer.extend(data) while True: end_of_headers = self._buffer.find(b"\r\n\r\n") @@ -488,32 +465,32 @@ def cancel_received(self, request: Request) -> None: request: The SIP CANCEL request. """ - async def answer(self, request: Request, *, call_class: type[RTPCall]) -> None: + async def answer( + self, request: Request, *, call_class: type[RTPCall], **call_kwargs: typing.Any + ) -> None: """Answer an incoming call by setting up RTP and sending 200 OK with SDP. - This coroutine can be awaited directly or wrapped in a task: + Example: + This coroutine can be awaited directly or wrapped in a task: - ```python - # inside a sync call_received: - asyncio.create_task(self.answer(request=request, call_class=MyCall)) + ```python + # inside a sync call_received: + asyncio.create_task(self.answer(request=request, call_class=MyCall)) - # inside an async call_received: - await self.answer(request=request, call_class=MyCall) - ``` + # inside an async call_received: + await self.answer(request=request, call_class=MyCall) + ``` Args: request: The SIP INVITE request (from `call_received`). call_class: A `Call` subclass whose `negotiate_codec` selects the codec. The class is constructed with ``rtp``, ``sip``, ``caller``, and ``media`` keyword arguments. + call_kwargs: Optional additional keyword arguments to pass to the call class constructor. Raises: NotImplementedError: When `negotiate_codec` raises (no supported codec in the remote SDP offer). """ - await self._answer(request, call_class) - - async def _answer(self, request: Request, call_class: type[RTPCall]) -> None: - """Perform the asynchronous part of answering: set up RTP, send 200 OK.""" call_id = request.headers.get("Call-ID", "") if call_id not in self._pending_invites: logger.error("No pending INVITE found for Call-ID %r", call_id) @@ -577,6 +554,7 @@ async def _answer(self, request: Request, call_class: type[RTPCall]) -> None: caller=caller, media=negotiated_media, srtp=srtp_session, + **call_kwargs, ) # Determine the remote RTP address for routing. # Per RFC 4566 §5.7 the effective connection address is taken from the