Use high precision scheduler to send RTP packets

tests/test_audio.py

@@ -414,7 +414,7 @@ class TestSendRTPAudio:
     """Tests for AudioCall.send_rtp_audio."""
 
     async def test_send_rtp_audio__sends_to_remote_addr(self):
-        """send_rtp_audio sends RTP packets to the caller's registered address."""
+        """The first RTP packet is transmitted synchronously inside send_audio."""
         call = make_audio_call(media=PCMU_MEDIA)
         remote_addr = ("10.0.0.1", 5004)
         call.rtp.calls = {remote_addr: call}
@@ -442,27 +442,47 @@ async def test_send_rtp_audio__drops_audio_when_no_remote_addr(self, caplog):
         assert any("dropping audio" in r.message for r in caplog.records)
 
     async def test_send_rtp_audio__paces_packets_at_20ms_intervals(self):
-        """send_rtp_audio sleeps RTP_PACKET_DURATION_SECS between each packet."""
+        """Subsequent packets are scheduled at rpt_packet_duration intervals via call_later."""
         call = make_audio_call(media=PCMU_MEDIA)
         remote_addr = ("10.0.0.2", 5006)
         call.rtp.calls = {remote_addr: call}
 
-        audio = np.zeros(320, dtype=np.float32)
-        sleep_calls: list[float] = []
-        original_sleep = asyncio.sleep
+        with patch.object(call, "send_packet"):
+            await call.send_audio(np.zeros(320, dtype=np.float32))
 
-        async def capture_sleep(delay: float) -> None:
-            sleep_calls.append(delay)
-            await original_sleep(0)
+        loop = asyncio.get_event_loop()
+        assert call.outbound_handle is not None
+        assert call.outbound_handle.when() == pytest.approx(
+            loop.time() + call.rpt_packet_duration.total_seconds(), abs=0.01
+        )
 
-        with (
-            patch("voip.audio.asyncio.sleep", side_effect=capture_sleep),
-            patch.object(call, "send_packet"),
-        ):
-            await call.send_audio(audio)
+    async def test_cancel_outbound_audio__cancels_handle_and_clears(self):
+        """cancel_outbound_audio cancels the pending handle and sets outbound_handle to None."""
+        call = make_audio_call(media=PCMU_MEDIA)
+        remote_addr = ("10.0.0.1", 5004)
+        call.rtp.calls = {remote_addr: call}
+
+        with patch.object(call, "send_packet"):
+            await call.send_audio(np.zeros(320, dtype=np.float32))
+
+        handle = call.outbound_handle
+        call.cancel_outbound_audio()
+
+        assert handle.cancelled()
+        assert call.outbound_handle is None
+
+    async def test_send_audio__preempts_pending_handle(self):
+        """A second send_audio cancels the pending handle from the first call."""
+        call = make_audio_call(media=PCMU_MEDIA)
+        remote_addr = ("10.0.0.1", 5004)
+        call.rtp.calls = {remote_addr: call}
+
+        with patch.object(call, "send_packet"):
+            await call.send_audio(np.zeros(320, dtype=np.float32))
+            first_handle = call.outbound_handle
+            await call.send_audio(np.zeros(320, dtype=np.float32))
 
-        assert len(sleep_calls) == 2
-        assert all(s <= call.rpt_packet_duration.total_seconds() for s in sleep_calls)
+        assert first_handle.cancelled()
 
 
 def make_echo_call(**kwargs) -> EchoCall:

voip/ai.py

@@ -12,6 +12,7 @@
 import asyncio
 import dataclasses
 import logging
+import re
 import typing
 
 import numpy as np
@@ -128,6 +129,21 @@ class AgentCall(TranscribeCall):
     _response_task: asyncio.Task | None = dataclasses.field(
         init=False, repr=False, default=None
     )
+    _cancel_audio_handle: asyncio.Handle | None = dataclasses.field(
+        init=False, repr=False, default=None
+    )
+
+    emoji_pattern: typing.ClassVar[typing.Pattern[str]] = re.compile(
+        "["
+        "\U0001f600-\U0001f64f"  # emoticons
+        "\U0001f300-\U0001f5ff"  # symbols & pictographs
+        "\U0001f680-\U0001f6ff"  # transport & map symbols
+        "\U0001f1e0-\U0001f1ff"  # flags (iOS)
+        "\U00002702-\U000027b0"
+        "\U000024c2-\U0001f251"
+        "]+",
+        flags=re.UNICODE,
+    )
 
     def __post_init__(self) -> None:
         super().__post_init__()
@@ -141,6 +157,7 @@ def __post_init__(self) -> None:
         ]
 
     def transcription_received(self, text: str) -> None:
+        self.cancel_outbound_audio()
         self._messages.append({"role": "user", "content": text})
         if self._response_task is not None and not self._response_task.done():
             self._response_task.cancel()
@@ -152,7 +169,7 @@ async def respond(self) -> None:
             messages=self._messages,
         )
         # clean non-ascii characters from the response for TTS processing
-        reply = (response.message.content or "").encode("ascii", "ignore").decode()
+        reply = self.emoji_pattern.sub("", response.message.content or "")
         self._messages.append({"role": "assistant", "content": reply})
         logger.debug("Agent reply: %r", reply)
         await self.send_speech(reply)
@@ -167,3 +184,18 @@ async def send_speech(self, text: str) -> None:
                 audio.numpy(), self.tts_model.sample_rate, self.codec.sample_rate_hz
             )
         )
+
+    def on_audio_speech(self) -> None:
+        loop = asyncio.get_event_loop()
+        if self._cancel_audio_handle is None:
+            self._cancel_audio_handle = loop.call_later(0.5, self.cancel_outbound_audio)
+        super().on_audio_speech()
+
+    def on_audio_silence(self) -> None:
+        super().on_audio_silence()
+        try:
+            self._cancel_audio_handle.cancel()
+        except AttributeError:
+            pass
+        else:
+            self._cancel_audio_handle = None

voip/audio.py

@@ -17,10 +17,11 @@
 import json
 import logging
 import secrets
-import time
+from collections.abc import Iterator
 from typing import ClassVar
 
 import numpy as np
+import pytest
 
 import voip.codecs as codecs
 from voip.codecs import RTPCodec
@@ -74,13 +75,15 @@ class AudioCall(RTPCall):
     rtp_ssrc: int = dataclasses.field(
         init=False, repr=False, default_factory=generate_ssrc
     )
-    last_packet_time: float = dataclasses.field(
-        init=False, repr=False, default_factory=time.perf_counter
-    )
     send_audio_lock: asyncio.Lock = dataclasses.field(
         default_factory=asyncio.Lock,
         init=False,
     )
+    outbound_handle: asyncio.TimerHandle | None = dataclasses.field(
+        default=None,
+        init=False,
+        repr=False,
+    )
 
     def __post_init__(self) -> None:
         fmt = self.media.fmt[0]
@@ -173,8 +176,8 @@ def decode_payload(self, payload: bytes) -> np.ndarray:
     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,
+            sequence_number=self.rtp_sequence_number,
+            timestamp=self.rtp_timestamp,
             ssrc=self.rtp_ssrc,
             payload=payload,
         )
@@ -201,9 +204,41 @@ def rms(audio: np.ndarray) -> float:
         """
         return float(np.sqrt(np.mean(np.square(audio))))
 
+    def cancel_outbound_audio(self) -> None:
+        """Stop the current outbound audio while it is being sent."""
+        try:
+            self.outbound_handle.cancel()
+        except AttributeError:
+            pass
+        else:
+            self.outbound_handle = None
+
+    def _dispatch_next_packet(
+        self,
+        packets: Iterator[bytes],
+        remote_addr: tuple[str, int],
+        next_send_at: float,
+    ) -> None:
+        try:
+            payload = next(packets)
+        except StopIteration:
+            self.outbound_handle = None
+        else:
+            self.send_packet(self.next_rtp_packet(payload), remote_addr)
+            duration_seconds = self.rpt_packet_duration.total_seconds()
+            next_deadline = next_send_at + duration_seconds
+            loop = asyncio.get_running_loop()
+            self.outbound_handle = loop.call_at(
+                next_deadline,
+                self._dispatch_next_packet,
+                packets,
+                remote_addr,
+                next_deadline,
+            )
+
     async def send_audio(self, audio: np.ndarray) -> None:
         """
-        Encode *audio* with the negotiated codec and transmit via RTP.
+        Encode `audio` with the negotiated codec and transmit via RTP.
 
         Args:
             audio: Float32 mono PCM at `codec.sample_rate_hz` Hz.
@@ -212,20 +247,23 @@ async def send_audio(self, audio: np.ndarray) -> None:
             (addr for addr, call in self.rtp.calls.items() if call is self),
             None,
         )
-        if remote_addr is None:
-            logger.warning("No remote RTP address for this call; dropping audio")
-            return
-        async with self.send_audio_lock:
-            for payload in self.codec.packetize(audio):
-                await asyncio.sleep(
-                    max(
-                        0.0,
-                        self.rpt_packet_duration.total_seconds()
-                        - (time.perf_counter() - self.last_packet_time),
-                    )
+        match remote_addr:
+            case None:
+                logger.warning(
+                    "No remote RTP address for this call; dropping audio",
                 )
-                self.send_packet(self.next_rtp_packet(payload), remote_addr)
-                self.last_packet_time = time.perf_counter()
+                return
+            case _:
+                pass
+        async with self.send_audio_lock:
+            self.cancel_outbound_audio()
+            loop = asyncio.get_running_loop()
+            next_send_at = loop.time()
+            self._dispatch_next_packet(
+                self.codec.packetize(audio),
+                remote_addr,
+                next_send_at,
+            )
 
     def audio_received(self, *, audio: np.ndarray, rms: float) -> None:
         """
@@ -237,6 +275,49 @@ def audio_received(self, *, audio: np.ndarray, rms: float) -> None:
         """
 
 
+@pytest.mark.asyncio
+async def test_send_audio_with_empty_packet_iterator_does_not_schedule_packets() -> (
+    None
+):
+    empty_audio = np.array([], dtype=np.float32)
+
+    class EmptyPacketCodec:
+        def __init__(self) -> None:
+            self.payload_type = 0
+            self.timestamp_increment = 160
+            self.sample_rate_hz = 8000
+
+        def packetize(self, audio: np.ndarray) -> Iterator[bytes]:
+            return iter(())
+
+    class SingleCallRtp:
+        def __init__(self, call: AudioCall, remote_addr: tuple[str, int]) -> None:
+            self.calls = {remote_addr: call}
+
+    call = object.__new__(AudioCall)
+    remote_addr = ("127.0.0.1", 4000)
+    codec = EmptyPacketCodec()
+    send_calls: list[tuple[RTPPacket, tuple[str, int]]] = []
+
+    def send_packet(packet: RTPPacket, addr: tuple[str, int]) -> None:
+        send_calls.append((packet, addr))
+
+    call.codec = codec
+    call.rtp_sequence_number = 0
+    call.rtp_timestamp = 0
+    call.rtp_ssrc = 1
+    call.rpt_packet_duration = datetime.timedelta(milliseconds=20)
+    call.outbound_handle = None
+    call.rtp = SingleCallRtp(call, remote_addr)
+    call.send_audio_lock = asyncio.Lock()
+    call.send_packet = send_packet
+
+    await AudioCall.send_audio(call, empty_audio)
+
+    assert call.outbound_handle is None
+    assert send_calls == []
+
+
 @dataclasses.dataclass(kw_only=True)
 class VoiceActivityCall(AudioCall):
     """