Merge pull request #781 from pipecat-ai/aleix/websocket-transports-mixer-fixes

websocket transports mixer fixes

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.
@@ -57,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.

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
+    formats. The audio files need to only have one channel (mono) and it needs
-    have any sample rate that will be resampled to the output transport sample
+    to match the sample rate of the output transport.
-    rate.
 
     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:
-            if sample_rate != self._sample_rate:
+                audio = sound.tobytes()
-                logger.debug(f"Resampling background sound to {self._sample_rate}")
+                # Convert from np to bytes again.
-                audio = resample_audio(audio, sample_rate, self._sample_rate)
+                self._sounds[sound_name] = np.frombuffer(audio, dtype=np.int16)
-
+            else:
-            # Convert from np to bytes again.
+                logger.warning(
-            self._sounds[sound_name] = np.frombuffer(audio, dtype=np.int16)
+                    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)

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__(

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__(