Add EchoCall: VAD-based speech echo handler + VoiceActivityCall base + echo CLI

docs/calls.md

@@ -6,6 +6,10 @@
 
 ::: voip.audio.AudioCall
 
+## Echo Call
+
+::: voip.audio.EchoCall
+
 ## AI / Agentic Calls
 
 ::: voip.ai.TranscribeCall

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  # 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,133 @@ 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)
+
+
+class TestEchoCall:
+    """Tests for EchoCall VAD-based echo playback."""
+
+    def test_echo_call__is_audio_call(self):
+        """EchoCall is a subclass of AudioCall."""
+        assert issubclass(EchoCall, AudioCall)
+
+    def test_audio_received__buffers_speech_frames(self):
+        """audio_received buffers frames with RMS above speech_threshold."""
+        call = make_echo_call()
+        audio = np.ones(160, dtype=np.float32)
+        call.audio_received(audio=audio, 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_echo_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_echo_timer(self):
+        """on_audio_speech cancels a running echo timer."""
+        call = make_echo_call()
+        handle = MagicMock()
+        call.echo_handle = handle
+        call.on_audio_speech()
+        handle.cancel.assert_called_once()
+        assert call.echo_handle is None
+
+    def test_on_audio_speech__noop_when_no_timer(self):
+        """on_audio_speech does nothing when no echo timer is running."""
+        call = make_echo_call()
+        call.on_audio_speech()  # must not raise
+        assert call.echo_handle is None
+
+    @pytest.mark.asyncio
+    async def test_on_audio_silence__arms_timer_when_buffer_has_speech(self):
+        """on_audio_silence schedules the echo timer when speech is buffered."""
+        call = make_echo_call()
+        call.speech_buffer.append(np.ones(160, dtype=np.float32))
+        call.on_audio_silence()
+        assert call.echo_handle is not None
+        call.echo_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_echo_call()
+        call.on_audio_silence()
+        assert call.echo_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 echo timer."""
+        call = make_echo_call()
+        call.speech_buffer.append(np.ones(160, dtype=np.float32))
+        call.on_audio_silence()
+        first_handle = call.echo_handle
+        call.on_audio_silence()
+        assert call.echo_handle is first_handle
+        call.echo_handle.cancel()
+
+    @pytest.mark.asyncio
+    async def test_flush_speech_buffer__resets_handle_and_schedules_echo(self):
+        """flush_speech_buffer clears the buffer and schedules echo playback."""
+        call = make_echo_call()
+        call.speech_buffer.append(np.ones(160, dtype=np.float32))
+        with patch.object(call, "echo", new_callable=AsyncMock) as mock_echo:
+            call.flush_speech_buffer()
+            await asyncio.sleep(0)
+        assert call.echo_handle is None
+        assert len(call.speech_buffer) == 0
+        mock_echo.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_echo_call()
+        call.flush_speech_buffer()  # must not raise or create a task
+        assert call.echo_handle is None
+
+    @pytest.mark.asyncio
+    async def test_echo__sends_resampled_audio(self):
+        """echo 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.echo(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

voip/audio.py

@@ -25,7 +25,7 @@
 from voip.rtp import RTPCall, RTPPacket
 from voip.sdp.types import MediaDescription
 
-__all__ = ["AudioCall"]
+__all__ = ["AudioCall", "EchoCall"]
 
 
 logger = logging.getLogger(__name__)
@@ -280,6 +280,83 @@ def resample(
         ).astype(np.float32)
 
 
+@dataclasses.dataclass(kw_only=True)
+class EchoCall(AudioCall):
+    """RTP call handler that echoes the caller's speech back after they finish speaking.
+
+    Accumulates speech audio frames (RMS above `speech_threshold`) 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)
+        ```
+
+    Attributes:
+        speech_threshold: RMS level below which audio is treated as silence.
+        silence_gap: Seconds of sustained silence required to trigger echo.
+    """
+
+    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
+    )
+    echo_handle: asyncio.TimerHandle | None = dataclasses.field(
+        init=False, repr=False, default=None
+    )
+
+    def audio_received(self, *, audio: np.ndarray, rms: float) -> None:
+        if rms > self.speech_threshold:
+            self.speech_buffer.append(audio)
+            self.on_audio_speech()
+        else:
+            self.on_audio_silence()
+
+    def on_audio_speech(self) -> None:
+        """Cancel any pending echo timer when speech is detected."""
+        if self.echo_handle is not None:
+            self.echo_handle.cancel()
+            self.echo_handle = None
+
+    def on_audio_silence(self) -> None:
+        """Arm the echo debounce timer on silence if not already running."""
+        if self.echo_handle is None and self.speech_buffer:
+            loop = asyncio.get_running_loop()
+            self.echo_handle = loop.call_later(
+                self.silence_gap,
+                self.flush_speech_buffer,
+            )
+
+    def flush_speech_buffer(self) -> None:
+        """Concatenate buffered speech and schedule async echo playback.
+
+        Resets speech state so the next utterance starts with a clean buffer.
+        """
+        self.echo_handle = None
+        if not self.speech_buffer:
+            return
+        audio = np.concatenate(self.speech_buffer)
+        self.speech_buffer.clear()
+        asyncio.create_task(self.echo(audio))
+
+    async def echo(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`][voip.audio.AudioCall.RESAMPLING_RATE_HZ] Hz.
+        """
+        resampled = self.resample(
+            audio, self.RESAMPLING_RATE_HZ, self.codec.sample_rate_hz
+        )
+        await self.send_rtp_audio(resampled)
+
+
 # Populate PREFERRED_CODECS after all codec imports settle.
 from voip.codecs.g722 import G722  # noqa: E402
 from voip.codecs.opus import Opus  # noqa: E402