diff --git a/tests/test_audio.py b/tests/test_audio.py index 9bd9688..bd2954f 100644 --- a/tests/test_audio.py +++ b/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: diff --git a/voip/ai.py b/voip/ai.py index 76d9fad..7941346 100644 --- a/voip/ai.py +++ b/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 diff --git a/voip/audio.py b/voip/audio.py index 7652002..5f20228 100644 --- a/voip/audio.py +++ b/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): """