diff --git a/tests/test_audio.py b/tests/test_audio.py index 253d4af..9bd9688 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -462,7 +462,7 @@ async def capture_sleep(delay: float) -> None: await call.send_audio(audio) assert len(sleep_calls) == 2 - assert all(s == call.rpt_packet_duration.total_seconds() 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: diff --git a/voip/audio.py b/voip/audio.py index 050b35a..7652002 100644 --- a/voip/audio.py +++ b/voip/audio.py @@ -17,6 +17,7 @@ import json import logging import secrets +import time from typing import ClassVar import numpy as np @@ -73,6 +74,13 @@ 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, + ) def __post_init__(self) -> None: fmt = self.media.fmt[0] @@ -207,9 +215,17 @@ async def send_audio(self, audio: np.ndarray) -> None: if remote_addr is None: logger.warning("No remote RTP address for this call; dropping audio") return - for payload in self.codec.packetize(audio): - self.send_packet(self.next_rtp_packet(payload), remote_addr) - await asyncio.sleep(self.rpt_packet_duration.total_seconds()) + 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), + ) + ) + self.send_packet(self.next_rtp_packet(payload), remote_addr) + self.last_packet_time = time.perf_counter() def audio_received(self, *, audio: np.ndarray, rms: float) -> None: """ @@ -310,7 +326,7 @@ def flush_voice_buffer(self) -> None: < 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[:])) + asyncio.create_task(self.voice_received(self._speech_buffer.copy())) self._speech_buffer = np.empty((0,), dtype=np.float32) async def voice_received(self, audio: np.ndarray) -> None: