diff --git a/docs/calls.md b/docs/calls.md index c9e9fb7..a8c8af0 100644 --- a/docs/calls.md +++ b/docs/calls.md @@ -6,6 +6,14 @@ ::: voip.audio.AudioCall +## Voice Activity Detection + +::: voip.audio.VoiceActivityCall + +## Echo Call + +::: voip.audio.EchoCall + ## AI / Agentic Calls ::: voip.ai.TranscribeCall diff --git a/tests/test_ai.py b/tests/test_ai.py index b88d4da..ab74e90 100644 --- a/tests/test_ai.py +++ b/tests/test_ai.py @@ -114,12 +114,12 @@ def test_audio_received__initializes_vad_state(self): """WhisperCall starts with an empty speech buffer and no timer.""" call = make_whisper_call(MagicMock()) assert call.speech_buffer == [] - assert call.transcription_handle is None + assert call.silence_handle is None def test_audio_received__silence_audio_accumulates_in_buffer(self): """Silence audio (below speech_threshold) is still buffered for transcription.""" call = make_whisper_call(MagicMock()) - with patch("voip.ai.asyncio.get_event_loop"): + with patch("voip.audio.asyncio.get_running_loop"): call.audio_received(audio=np.zeros(320, dtype=np.float32), rms=0.0) assert len(call.speech_buffer) == 1 @@ -133,20 +133,20 @@ def test_audio_received__speech_audio_accumulates_in_buffer(self): def test_audio_received__silence_arms_transcription_timer(self): """Silence arms the transcription debounce timer.""" call = make_whisper_call(MagicMock()) - with patch("voip.ai.asyncio.get_event_loop") as mock_loop: + with patch("voip.audio.asyncio.get_running_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 ) - assert call.transcription_handle is handle + assert call.silence_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.transcription_handle = MagicMock() - with patch("voip.ai.asyncio.get_event_loop") as mock_loop: + call.silence_handle = MagicMock() + with patch("voip.audio.asyncio.get_running_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() @@ -154,15 +154,15 @@ def test_audio_received__speech_cancels_pending_timer(self): """Speech audio cancels any running transcription debounce timer.""" call = make_whisper_call(MagicMock()) handle = MagicMock() - call.transcription_handle = handle + call.silence_handle = handle call.audio_received(audio=np.ones(320, dtype=np.float32) * 0.6, rms=0.6) handle.cancel.assert_called_once() - assert call.transcription_handle is None + assert call.silence_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.ai.asyncio.get_event_loop"): + with patch("voip.audio.asyncio.get_running_loop"): call.audio_received(audio=np.zeros(0, dtype=np.float32), rms=0.0) assert len(call.speech_buffer) == 1 @@ -181,8 +181,6 @@ def transcription_received(self, text: str) -> None: call = make_whisper_call(model_mock, Capture) chunk = np.ones(320, dtype=np.float32) call.speech_buffer = [chunk] * 60 - # Set silence_gap=0 so the minimum-length check passes with one chunk. - call.silence_gap = 0 call.flush_speech_buffer() await asyncio.sleep(0.1) assert transcriptions == ["hello"] @@ -191,18 +189,34 @@ def transcription_received(self, text: str) -> None: def test_flush_speech_buffer__no_op_when_buffer_empty(self): """_flush_speech_buffer does nothing when the speech buffer is empty.""" call = make_whisper_call(MagicMock()) - with patch("voip.ai.asyncio.create_task") as mock_ct: + with patch("voip.audio.asyncio.create_task") as mock_ct: call.flush_speech_buffer() mock_ct.assert_not_called() def test_flush_speech_buffer__resets_state(self): - """_flush_speech_buffer clears _transcription_handle and the speech buffer.""" + """_flush_speech_buffer clears silence_handle and the speech buffer.""" call = make_whisper_call(MagicMock()) - call.transcription_handle = MagicMock() + call.silence_handle = MagicMock() call.speech_buffer = [np.zeros(1, dtype=np.float32)] - with patch("voip.ai.asyncio.create_task"): + with patch("voip.audio.asyncio.create_task", side_effect=lambda c: c.close()): call.flush_speech_buffer() - assert call.transcription_handle is None + assert call.silence_handle is None + + async def test_speech_buffer_ready__skips_short_audio(self): + """speech_buffer_ready discards audio shorter than one second.""" + transcriptions = [] + 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) + model_mock.transcribe.assert_not_called() + assert transcriptions == [] async def test_transcribe__strips_whitespace(self): """Strip leading and trailing whitespace from the transcription text.""" diff --git a/tests/test_audio.py b/tests/test_audio.py index 9a6c988..36d6815 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -3,14 +3,14 @@ from __future__ import annotations import asyncio -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest np = pytest.importorskip("numpy") av = pytest.importorskip("av") -from voip.audio import AudioCall # noqa: E402 +from voip.audio import AudioCall, EchoCall, VoiceActivityCall # noqa: E402 from voip.codecs.g722 import G722 # noqa: E402 from voip.codecs.opus import Opus # noqa: E402 from voip.codecs.pcma import PCMA # noqa: E402 @@ -448,3 +448,173 @@ async def capture_sleep(delay: float) -> None: assert len(sleep_calls) == 2 assert all(s == call.RTP_PACKET_DURATION_SECS for s in sleep_calls) + + +def make_echo_call(**kwargs) -> EchoCall: + """Create an EchoCall with mock rtp/sip for unit testing.""" + defaults: dict = { + "rtp": MagicMock(spec=RealtimeTransportProtocol), + "sip": MagicMock(), + "media": PCMU_MEDIA, + "caller": CallerID(""), + } + defaults.update(kwargs) + return EchoCall(**defaults) + + +def make_vac_call(**kwargs) -> VoiceActivityCall: + """Create a VoiceActivityCall with mock rtp/sip for unit testing.""" + defaults: dict = { + "rtp": MagicMock(spec=RealtimeTransportProtocol), + "sip": MagicMock(), + "media": PCMU_MEDIA, + "caller": CallerID(""), + } + defaults.update(kwargs) + return VoiceActivityCall(**defaults) + + +class TestVoiceActivityCall: + """Tests for the shared VAD infrastructure in VoiceActivityCall.""" + + 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.""" + call = make_vac_call() + assert call.collect_audio(np.ones(160, dtype=np.float32), rms=1.0) is True + + def test_collect_audio__returns_false_for_silence(self): + """collect_audio returns False when RMS is at or below speech_threshold.""" + call = make_vac_call() + assert call.collect_audio(np.zeros(160, dtype=np.float32), rms=0.0) is False + + def test_audio_received__buffers_speech_frames(self): + """audio_received buffers frames with RMS above speech_threshold.""" + call = make_vac_call() + call.audio_received(audio=np.ones(160, dtype=np.float32), rms=1.0) + assert len(call.speech_buffer) == 1 + + def test_audio_received__does_not_buffer_silence_frames(self): + """audio_received skips frames with RMS at or below speech_threshold.""" + call = make_vac_call() + call.audio_received(audio=np.zeros(160, dtype=np.float32), rms=0.0) + assert len(call.speech_buffer) == 0 + + def test_on_audio_speech__cancels_silence_timer(self): + """on_audio_speech cancels a running silence timer.""" + call = make_vac_call() + handle = MagicMock() + call.silence_handle = handle + call.on_audio_speech() + handle.cancel.assert_called_once() + assert call.silence_handle is None + + def test_on_audio_speech__noop_when_no_timer(self): + """on_audio_speech does nothing when no silence timer is running.""" + call = make_vac_call() + call.on_audio_speech() # must not raise + assert call.silence_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.""" + 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() + + @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.""" + call = make_vac_call() + call.on_audio_silence() + assert call.silence_handle is None + + @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.""" + 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() + + @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.""" + 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.""" + call = make_vac_call() + with patch("voip.audio.asyncio.create_task") as mock_ct: + call.flush_speech_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.""" + call = make_vac_call() + await call.speech_buffer_ready( + np.zeros(160, dtype=np.float32) + ) # must not raise + + +class TestEchoCall: + """Tests for EchoCall speech echo playback.""" + + def test_echo_call__is_voice_activity_call(self): + """EchoCall is a subclass of VoiceActivityCall.""" + 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.""" + 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) + mock_send.assert_awaited_once() + sent_audio = mock_send.call_args[0][0] + # PCMU sample_rate_hz == 8000; RESAMPLING_RATE_HZ == 16000 → half length + expected_len = round( + len(audio) * call.codec.sample_rate_hz / call.RESAMPLING_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) + remote_addr = ("10.0.0.1", 5004) + call.rtp.calls = {remote_addr: call} + + speech = np.ones(160, dtype=np.float32) * 0.5 + silence = np.zeros(160, dtype=np.float32) + + sent: list[np.ndarray] = [] + + async def capture_send(audio: np.ndarray) -> None: + sent.append(audio) + + with patch.object(call, "send_rtp_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) + + assert len(sent) == 1 diff --git a/tests/test_main.py b/tests/test_main.py index 5b88ded..218326b 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -20,7 +20,9 @@ "faster_whisper": MagicMock(), "ollama": MagicMock(), "pocket_tts": MagicMock(), - "voip.audio": MagicMock(), + "voip.audio": MagicMock( + EchoCall=MagicMock, VoiceActivityCall=MagicMock, AudioCall=MagicMock + ), "voip.ai": MagicMock(TranscribeCall=MagicMock, AgentCall=MagicMock), } @@ -754,3 +756,116 @@ async def _run_voice(): assert call_cls.voice == "ellie" asyncio.run(_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): + captured["host"] = host + captured["port"] = port + captured["ssl"] = ssl + raise KeyboardInterrupt + + with ( + patch.dict(sys.modules, _WHISPER_STUBS), + patch("asyncio.get_event_loop"), + patch("voip.__main__.asyncio.get_running_loop") as mock_loop, + ): + mock_loop.return_value.create_connection = fake_connection + make_runner().invoke( + voip, + [ + "sip", + "--password=secret", + "sips:alice@sip.example.com", + "echo", + ], + catch_exceptions=False, + ) + assert captured.get("host") == "sip.example.com" + assert captured.get("port") == 5061 + assert captured.get("ssl") is not None + + 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): + captured["ssl"] = ssl + captured["port"] = port + raise KeyboardInterrupt + + with ( + patch.dict(sys.modules, _WHISPER_STUBS), + patch("asyncio.get_event_loop"), + patch("voip.__main__.asyncio.get_running_loop") as mock_loop, + ): + mock_loop.return_value.create_connection = fake_connection + make_runner().invoke( + voip, + [ + "sip", + "--password=secret", + "sip:alice@example.com:5060", + "echo", + ], + catch_exceptions=False, + ) + assert captured.get("ssl") is None + assert captured.get("port") == 5060 + + 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): + protocol = factory() + protocol_holder["protocol"] = protocol + raise KeyboardInterrupt + + with ( + patch.dict(sys.modules, _WHISPER_STUBS), + patch("asyncio.get_event_loop"), + patch("voip.__main__.asyncio.get_running_loop") as mock_loop, + ): + mock_loop.return_value.create_connection = fake_connection + make_runner().invoke( + voip, + [ + "sip", + "--password=p", + "--stun-server=none", + "sips:alice@example.com", + "echo", + ], + catch_exceptions=False, + ) + protocol = protocol_holder["protocol"] + from voip.sip.messages import Request + + request = Request( + method="INVITE", + uri="sip:u@example.com", + headers={"From": "sip:caller@example.com", "Call-ID": "test@pc"}, + ) + + async def run(): + with patch.object(protocol, "answer") as mock_answer: + protocol.connection_made(MagicMock()) + protocol._pending_invites.add(request.headers["Call-ID"]) + protocol.call_received(request) + mock_answer.assert_called_once() + _, kwargs = mock_answer.call_args + assert "call_class" in kwargs + assert isinstance(kwargs["call_class"], type) + + asyncio.run(run()) diff --git a/voip/__main__.py b/voip/__main__.py index f69c6de..86aab51 100644 --- a/voip/__main__.py +++ b/voip/__main__.py @@ -270,6 +270,60 @@ async def _connect_sip( await asyncio.Future() +@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 + + 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(*bases): + def call_received(self, request) -> None: + self.ringing(request=request) + asyncio.create_task(self.answer(request=request, call_class=EchoCall)) + + async def run(): + await _connect_sip( + lambda: EchoSession( + outbound_proxy=proxy_addr, + aor=obj["aor"], + username=obj["username"], + password=obj["password"], + rtp_stun_server_address=obj["stun_server"], + ), + proxy_addr, + obj["use_tls"], + obj["no_verify_tls"], + ) + + try: + asyncio.run(run()) + except KeyboardInterrupt: + pass + + @sip.command() @click.option( "--model", diff --git a/voip/ai.py b/voip/ai.py index 70c56a5..c52562b 100644 --- a/voip/ai.py +++ b/voip/ai.py @@ -19,7 +19,7 @@ from faster_whisper import WhisperModel from pocket_tts import TTSModel -from voip.audio import AudioCall +from voip.audio import VoiceActivityCall from voip.codecs import Codec from voip.codecs.g722 import G722 from voip.codecs.opus import Opus @@ -32,17 +32,17 @@ @dataclasses.dataclass(kw_only=True) -class TranscribeCall(AudioCall): +class TranscribeCall(VoiceActivityCall): """RTP call handler that transcribes audio with faster-whisper. Audio is decoded by [`AudioCall`][voip.audio.AudioCall] on a per-packet basis and delivered to [`audio_received`][voip.audio.AudioCall.audio_received], - which applies an energy-based voice activity detector (VAD). Speech - packets are accumulated until silence is sustained for - `silence_gap` seconds, then the - entire utterance is sent to Whisper as one chunk. This avoids cutting - sentences in the middle and prevents background microphone noise from - being passed to Whisper as spurious audio. + which applies an energy-based voice activity detector (VAD) from + [`VoiceActivityCall`][voip.audio.VoiceActivityCall]. All audio frames + (speech and silence) are accumulated until silence is sustained for + `silence_gap` seconds, then the entire utterance is sent to Whisper as + 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: @@ -68,16 +68,6 @@ class MyCall(TranscribeCall): model: str | WhisperModel = dataclasses.field(default="kyutai/stt-1b-en_fr-trfs") whisper_model: WhisperModel = dataclasses.field(init=False, repr=False) - speech_threshold: float = dataclasses.field(default=0.001) - silence_gap: float = dataclasses.field(default=0.5) - - speech_buffer: list[np.ndarray] = dataclasses.field( - init=False, repr=False, default_factory=list - ) - transcription_handle: asyncio.TimerHandle | None = dataclasses.field( - init=False, repr=False, default=None - ) - def __post_init__(self) -> None: super().__post_init__() if isinstance(self.model, str): @@ -85,45 +75,31 @@ def __post_init__(self) -> None: self.whisper_model = WhisperModel(self.model) else: self.whisper_model = self.model - self.speech_buffer = [] - self.transcription_handle = None - def audio_received(self, *, audio: np.ndarray, rms: float) -> None: - self.speech_buffer.append(audio) - if rms > self.speech_threshold: - self.on_audio_speech() - else: - self.on_audio_silence() - - def on_audio_speech(self) -> None: - """Cancel any pending transcription timer when speech is detected.""" - if self.transcription_handle is not None: - self.transcription_handle.cancel() - self.transcription_handle = None - - def on_audio_silence(self) -> None: - """Arm the transcription debounce timer on silence if not already running.""" - if self.transcription_handle is None: - logger.debug("Silence detected") - loop = asyncio.get_event_loop() - self.transcription_handle = loop.call_later( - self.silence_gap, - self.flush_speech_buffer, - ) + def collect_audio(self, audio: np.ndarray, rms: float) -> bool: + """Buffer all audio frames (speech and silence) for transcription. + + Args: + audio: Decoded float32 PCM frame. + rms: Root mean square of *audio*. + + Returns: + Always `True` so that intra-utterance silences are preserved. + """ + return True - def flush_speech_buffer(self) -> None: - """Concatenate buffered audio and schedule async transcription. + async def speech_buffer_ready(self, audio: np.ndarray) -> None: + """Transcribe the buffered utterance when it meets the minimum length. - Resets speech state so the next utterance starts with a clean buffer. + Skips utterances shorter than one second to avoid passing fragments + to Whisper that would produce low-quality transcriptions. + + Args: + audio: Float32 mono PCM array at `RESAMPLING_RATE_HZ` Hz. """ - self.transcription_handle = None - # Ensure at least one second of audio to avoid cutting words in half. - if sum(len(c) for c in self.speech_buffer) < self.RESAMPLING_RATE_HZ: - self.speech_buffer.clear() + if len(audio) < self.RESAMPLING_RATE_HZ: return - audio = np.concatenate(self.speech_buffer) - self.speech_buffer.clear() - asyncio.create_task(self.transcribe(audio)) + await self.transcribe(audio) async def transcribe(self, audio: np.ndarray) -> None: """Transcribe decoded audio and deliver non-empty text to the handler. diff --git a/voip/audio.py b/voip/audio.py index 5b39fe6..d9256ee 100644 --- a/voip/audio.py +++ b/voip/audio.py @@ -22,10 +22,14 @@ import voip.codecs as codecs from voip.codecs import Codec +from voip.codecs.g722 import G722 # noqa: E402 +from voip.codecs.opus import Opus # noqa: E402 +from voip.codecs.pcma import PCMA # noqa: E402 +from voip.codecs.pcmu import PCMU # noqa: E402 from voip.rtp import RTPCall, RTPPacket from voip.sdp.types import MediaDescription -__all__ = ["AudioCall"] +__all__ = ["AudioCall", "EchoCall", "VoiceActivityCall"] logger = logging.getLogger(__name__) @@ -51,7 +55,7 @@ class is stored on `codec` after `__post_init__` and used for all """ #: Preferred codecs in priority order (highest priority first). - PREFERRED_CODECS: ClassVar[list[type[Codec]]] = [] + PREFERRED_CODECS: ClassVar[list[type[Codec]]] = [Opus, G722, PCMA, PCMU] #: Target sample rate for decoded audio delivered to `audio_received`. RESAMPLING_RATE_HZ: ClassVar[int] = 16000 @@ -280,14 +284,127 @@ def resample( ).astype(np.float32) -# Populate PREFERRED_CODECS after all codec imports settle. -from voip.codecs.g722 import G722 # noqa: E402 -from voip.codecs.opus import Opus # noqa: E402 -from voip.codecs.pcma import PCMA # noqa: E402 -from voip.codecs.pcmu import PCMU # noqa: E402 +@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`. + + Attributes: + speech_threshold: RMS level below which audio is treated as silence. + silence_gap: Seconds of sustained silence required to flush the buffer. + """ + + speech_threshold: float = dataclasses.field(default=0.001) + silence_gap: float = dataclasses.field(default=0.5) + + speech_buffer: list[np.ndarray] = dataclasses.field( + init=False, repr=False, default_factory=list + ) + silence_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.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*. -AudioCall.PREFERRED_CODECS = [Opus, G722, PCMA, PCMU] + 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 + + 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, + ) + + 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: + """Handle the flushed speech buffer. Override in subclasses. + + This base implementation is a no-op. Subclasses must override this + method to process the buffered utterance (e.g. echo it back, transcribe + it, etc.). + + Args: + audio: Float32 mono PCM array at `RESAMPLING_RATE_HZ` Hz + containing the full buffered utterance. + """ + + +@dataclasses.dataclass(kw_only=True) +class EchoCall(VoiceActivityCall): + """RTP call handler that echoes 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. + + Example: + ```python + class MySession(SessionInitiationProtocol): + def call_received(self, request: Request) -> None: + self.answer(request=request, call_class=EchoCall) + ``` + """ -# Re-export RTPPayloadType so existing importers that do -# ``from voip.audio import ...`` continue to work. -from voip.rtp import RTPPayloadType # noqa: E402, F401 + 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. + """ + resampled = self.resample( + audio, self.RESAMPLING_RATE_HZ, self.codec.sample_rate_hz + ) + await self.send_rtp_audio(resampled)