From e0f7a8a9f4a82bc0ca717fed8ea7a77af4a84853 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 4 Dec 2024 15:14:02 -0800 Subject: [PATCH 1/2] audio(mixer): SoundfileMixer doesn't resample files anymore --- CHANGELOG.md | 4 ++++ src/pipecat/audio/mixers/soundfile_mixer.py | 25 ++++++++++----------- 2 files changed, 16 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 400726a6d..69f1d1fa2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,10 @@ async def on_audio_data(processor, audio, sample_rate, num_channels): ### Changed +- `SoundfileMixer` doesn't resample input files anymore to avoid startup + delays. The sample rate of the provided sound files now need to match the + sample rate of the output transport. + - All input frames (text, audio, image, etc.) are now system frames. This means they are processed immediately by all processors instead of being queued internally. diff --git a/src/pipecat/audio/mixers/soundfile_mixer.py b/src/pipecat/audio/mixers/soundfile_mixer.py index 8965e2d53..659ab0507 100644 --- a/src/pipecat/audio/mixers/soundfile_mixer.py +++ b/src/pipecat/audio/mixers/soundfile_mixer.py @@ -11,7 +11,6 @@ import numpy as np from loguru import logger from pipecat.audio.mixers.base_audio_mixer import BaseAudioMixer -from pipecat.audio.utils import resample_audio from pipecat.frames.frames import MixerControlFrame, MixerEnableFrame, MixerUpdateSettingsFrame try: @@ -27,9 +26,8 @@ except ModuleNotFoundError as e: class SoundfileMixer(BaseAudioMixer): """This is an audio mixer that mixes incoming audio with audio from a file. It uses the soundfile library to load files so it supports multiple - formats. The audio files need to only have one channel (mono) but they can - have any sample rate that will be resampled to the output transport sample - rate. + formats. The audio files need to only have one channel (mono) and it needs + to match the sample rate of the output transport. Multiple files can be loaded, each with a different name. The `MixerUpdateSettingsFrame` has the following settings available: `sound` @@ -103,16 +101,17 @@ class SoundfileMixer(BaseAudioMixer): def _load_sound_file(self, sound_name: str, file_name: str): try: - logger.debug(f"Loading background sound from {file_name}") + logger.debug(f"Loading mixer sound from {file_name}") sound, sample_rate = sf.read(file_name, dtype="int16") - audio = sound.tobytes() - if sample_rate != self._sample_rate: - logger.debug(f"Resampling background sound to {self._sample_rate}") - audio = resample_audio(audio, sample_rate, self._sample_rate) - - # Convert from np to bytes again. - self._sounds[sound_name] = np.frombuffer(audio, dtype=np.int16) + if sample_rate == self._sample_rate: + audio = sound.tobytes() + # Convert from np to bytes again. + self._sounds[sound_name] = np.frombuffer(audio, dtype=np.int16) + else: + logger.warning( + f"Sound file {file_name} has incorrect sample rate {sample_rate} (should be {self._sample_rate})" + ) except Exception as e: logger.error(f"Unable to open file {file_name}: {e}") @@ -121,7 +120,7 @@ class SoundfileMixer(BaseAudioMixer): file. """ - if not self._mixing: + if not self._mixing or not self._current_sound in self._sounds: return audio audio_np = np.frombuffer(audio, dtype=np.int16) From 0935d773f5a1264e6388b077a183c5c92cd7fbff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 4 Dec 2024 15:15:43 -0800 Subject: [PATCH 2/2] transport(websockets): fix initial busy loop when using audio mixers --- CHANGELOG.md | 3 +++ .../transports/network/fastapi_websocket.py | 23 +++++++++++++------ .../transports/network/websocket_server.py | 20 ++++++++++------ 3 files changed, 32 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 69f1d1fa2..de29d0c74 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -61,6 +61,9 @@ async def on_audio_data(processor, audio, sample_rate, num_channels): ### Fixed +- Fixed an issue in `WebsocketServerTransport` and `FastAPIWebsocketTransport` + that would cause a busy loop when using audio mixer. + - Fixed a `DailyTransport` and `LiveKitTransport` issue where connections were being closed in the input transport prematurely. This was causing frames queued inside the pipeline being discarded. diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index db1dcccdd..0ca969463 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -105,6 +105,11 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): self._next_send_time = 0 async def write_raw_audio_frames(self, frames: bytes): + if self._websocket.client_state != WebSocketState.CONNECTED: + # Simulate audio playback with a sleep. + await self._write_audio_sleep() + return + frame = AudioRawFrame( audio=frames, sample_rate=self._params.audio_out_sample_rate, @@ -125,10 +130,21 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): ) frame = wav_frame + payload = self._params.serializer.serialize(frame) + if payload: + await self._websocket.send_text(payload) + + self._websocket_audio_buffer = bytes() + + # Simulate audio playback with a sleep. + await self._write_audio_sleep() + + async def _write_frame(self, frame: Frame): payload = self._params.serializer.serialize(frame) if payload and self._websocket.client_state == WebSocketState.CONNECTED: await self._websocket.send_text(payload) + async def _write_audio_sleep(self): # Simulate a clock. current_time = time.monotonic() sleep_duration = max(0, self._next_send_time - current_time) @@ -138,13 +154,6 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): else: self._next_send_time += self._send_interval - self._websocket_audio_buffer = bytes() - - async def _write_frame(self, frame: Frame): - payload = self._params.serializer.serialize(frame) - if payload and self._websocket.client_state == WebSocketState.CONNECTED: - await self._websocket.send_text(payload) - class FastAPIWebsocketTransport(BaseTransport): def __init__( diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index 10a42828b..2b88a7334 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -153,6 +153,8 @@ class WebsocketServerOutputTransport(BaseOutputTransport): async def write_raw_audio_frames(self, frames: bytes): if not self._websocket: + # Simulate audio playback with a sleep. + await self._write_audio_sleep() return frame = AudioRawFrame( @@ -179,6 +181,17 @@ class WebsocketServerOutputTransport(BaseOutputTransport): if proto: await self._websocket.send(proto) + self._websocket_audio_buffer = bytes() + + # Simulate audio playback with a sleep. + await self._write_audio_sleep() + + async def _write_frame(self, frame: Frame): + payload = self._params.serializer.serialize(frame) + if payload and self._websocket: + await self._websocket.send(payload) + + async def _write_audio_sleep(self): # Simulate a clock. current_time = time.monotonic() sleep_duration = max(0, self._next_send_time - current_time) @@ -188,13 +201,6 @@ class WebsocketServerOutputTransport(BaseOutputTransport): else: self._next_send_time += self._send_interval - self._websocket_audio_buffer = bytes() - - async def _write_frame(self, frame: Frame): - payload = self._params.serializer.serialize(frame) - if payload and self._websocket: - await self._websocket.send(payload) - class WebsocketServerTransport(BaseTransport): def __init__(