introduce output transport audio mixers

CHANGELOG.md

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Introduce output transport audio mixers. Output transport audio mixers can be
+  used, for example, to add background sounds or any other audio mixing
+  functionality before the output audio is actually written to the transport.
+
 - Added `GatedOpenAILLMContextAggregator`. This aggregator keeps the last
   received OpenAI LLM context frame and it doesn't let it through until the
   notifier is notified.

pyproject.toml

@@ -60,6 +60,7 @@ openai = [ "openai~=1.50.2", "websockets~=13.1", "python-deepcompare~=1.0.1" ]
 openpipe = [ "openpipe~=4.24.0" ]
 playht = [ "pyht~=0.1.4", "websockets~=13.1" ]
 silero = [ "onnxruntime~=1.19.2" ]
+soundfile = [ "soundfile~=0.12.1" ]
 together = [ "openai~=1.50.2" ]
 websocket = [ "websockets~=13.1", "fastapi~=0.115.0" ]
 whisper = [ "faster-whisper~=1.0.3" ]

src/pipecat/audio/mixers/base_audio_mixer.py

@@ -0,0 +1,53 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from abc import ABC, abstractmethod
+
+from pipecat.frames.frames import Frame
+
+
+class BaseAudioMixer(ABC):
+    """This is a base class for output transport audio mixers. If an audio mixer
+    is provided to the output transport it will be used to mix the audio frames
+    coming into to the transport with the audio generated from the mixer. There
+    are control frames to update mixer settings or to enable or disable the
+    mixer at runtime.
+
+    """
+
+    @abstractmethod
+    async def start(self, sample_rate: int):
+        """This will be called from the output transport when the transport is
+        started. It can be used to initialize the mixer. The output transport
+        sample rate is provided so the mixer can adjust to that sample rate.
+
+        """
+        pass
+
+    @abstractmethod
+    async def stop(self):
+        """This will be called from the output transport when the transport is
+        stopping.
+
+        """
+        pass
+
+    @abstractmethod
+    async def process_frame(self, frame: Frame):
+        """This will be called when the output transport receives a
+        MixerControlFrame.
+
+        """
+        pass
+
+    @abstractmethod
+    async def mix(self, audio: bytes) -> bytes:
+        """This is called with the audio that is about to be sent from the
+        output transport and that should be mixed with the mixer audio if the
+        mixer is enabled.
+
+        """
+        pass

src/pipecat/audio/mixers/soundfile_mixer.py

@@ -0,0 +1,140 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+
+from typing import Any, Dict, Mapping
+
+import numpy as np
+
+from pipecat.audio.mixers.base_audio_mixer import BaseAudioMixer
+from pipecat.audio.utils import resample_audio
+from pipecat.frames.frames import Frame, MixerUpdateSettingsFrame, MixerEnableFrame
+
+from loguru import logger
+
+try:
+    import soundfile as sf
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error(
+        "In order to use the soundfile mixer, you need to `pip install pipecat-ai[soundfile]`."
+    )
+    raise Exception(f"Missing module: {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.
+
+    Multiple files can be loaded, each with a different name. The
+    `MixerUpdateSettingsFrame` has the following settings available: `sound`
+    (str) and `volume` (float) to be able to update to a different sound file or
+    to change the volume at runtime.
+
+    """
+
+    def __init__(
+        self,
+        sound_files: Mapping[str, str],
+        default_sound: str,
+        volume: float = 0.4,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self._sound_files = sound_files
+        self._volume = volume
+        self._sample_rate = 0
+
+        self._sound_pos = 0
+        self._sounds: Dict[str, Any] = {}
+        self._current_sound = default_sound
+        self._mixing = True
+
+    async def start(self, sample_rate: int):
+        self._sample_rate = sample_rate
+        for sound_name, file_name in self._sound_files.items():
+            await asyncio.to_thread(self._load_sound_file, sound_name, file_name)
+
+    async def stop(self):
+        pass
+
+    async def process_frame(self, frame: Frame):
+        if isinstance(frame, MixerUpdateSettingsFrame):
+            await self._update_settings(frame)
+        elif isinstance(frame, MixerEnableFrame):
+            await self._enable_mixing(frame.enable)
+        pass
+
+    async def mix(self, audio: bytes) -> bytes:
+        return self._mix_with_sound(audio)
+
+    async def _enable_mixing(self, enable: bool):
+        self._mixing = enable
+
+    async def _update_settings(self, frame: MixerUpdateSettingsFrame):
+        for setting, value in frame.settings.items():
+            match setting:
+                case "sound":
+                    await self._change_sound(value)
+                case "volume":
+                    await self._update_volume(value)
+
+    async def _change_sound(self, sound: str):
+        if sound in self._sound_files:
+            self._current_sound = sound
+            self._sound_pos = 0
+        else:
+            logger.error(f"Sound {sound} is not available")
+
+    async def _update_volume(self, volume: float):
+        self._volume = volume
+
+    def _load_sound_file(self, sound_name: str, file_name: str):
+        try:
+            logger.debug(f"Loading background 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)
+        except Exception as ex:
+            logger.error(f"Unable to open file {file_name}")
+
+    def _mix_with_sound(self, audio: bytes):
+        """Mixes raw audio frames with chunks of the same length from the sound
+        file.
+
+        """
+        if not self._mixing:
+            return audio
+
+        audio_np = np.frombuffer(audio, dtype=np.int16)
+        chunk_size = len(audio_np)
+
+        # Sound currently playing.
+        sound = self._sounds[self._current_sound]
+
+        # Go back to the beginning if we don't have enough data.
+        if self._sound_pos + chunk_size > len(sound):
+            self._sound_pos = 0
+
+        start_pos = self._sound_pos
+        end_pos = self._sound_pos + chunk_size
+        self._sound_pos = end_pos
+
+        sound_np = sound[start_pos:end_pos]
+
+        mixed_audio = np.clip(audio_np + sound_np * self._volume, -32768, 32767).astype(np.int16)
+
+        return mixed_audio.astype(np.int16).tobytes()

src/pipecat/frames/frames.py

@@ -5,7 +5,7 @@
 #
 
 from dataclasses import dataclass, field
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, List, Mapping, Optional, Tuple
 
 from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.clocks.base_clock import BaseClock
@@ -557,7 +557,7 @@ class TTSStoppedFrame(ControlFrame):
 class ServiceUpdateSettingsFrame(ControlFrame):
     """A control frame containing a request to update service settings."""
 
-    settings: Dict[str, Any]
+    settings: Mapping[str, Any]
 
 
 @dataclass
@@ -582,3 +582,24 @@ class VADParamsUpdateFrame(ControlFrame):
     """
 
     params: VADParams
+
+
+@dataclass
+class MixerControlFrame(ControlFrame):
+    """Base control frame for other mixer frames."""
+
+    pass
+
+
+@dataclass
+class MixerUpdateSettingsFrame(MixerControlFrame):
+    """Control frame to update mixer settings."""
+
+    settings: Mapping[str, Any]
+
+
+@dataclass
+class MixerEnableFrame(MixerControlFrame):
+    """Control frame to enable or disable the mixer at runtime."""
+
+    enable: bool

src/pipecat/transports/base_output.py

@@ -8,19 +8,21 @@ import asyncio
 import itertools
 import sys
 import time
-from typing import List
+from typing import AsyncGenerator, List
 
 from loguru import logger
 from PIL import Image
 
 from pipecat.audio.vad.vad_analyzer import VAD_STOP_SECS
 from pipecat.frames.frames import (
+    AudioRawFrame,
     BotSpeakingFrame,
     BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
     Frame,
+    MixerControlFrame,
     OutputAudioRawFrame,
     OutputImageRawFrame,
     SpriteFrame,
@@ -73,11 +75,15 @@ class BaseOutputTransport(FrameProcessor):
         self._bot_speaking = False
 
     async def start(self, frame: StartFrame):
+        if self._params.audio_out_mixer:
+            await self._params.audio_out_mixer.start(self._params.audio_out_sample_rate)
         self._create_output_tasks()
         self._create_sink_tasks()
 
     async def stop(self, frame: EndFrame):
         await self._cancel_output_tasks()
+        if self._params.audio_out_mixer:
+            await self._params.audio_out_mixer.stop()
 
     async def cancel(self, frame: CancelFrame):
         # Since we are cancelling everything it doesn't matter if we cancel sink
@@ -129,6 +135,8 @@ class BaseOutputTransport(FrameProcessor):
             await self.stop(frame)
             # We finally push EndFrame down so PipelineTask stops nicely.
             await self.push_frame(frame, direction)
+        elif isinstance(frame, MixerControlFrame) and self._params.audio_out_mixer:
+            await self._params.audio_out_mixer.process_frame(frame)
         # Other frames.
         elif isinstance(frame, OutputAudioRawFrame):
             await self._handle_audio(frame)
@@ -386,35 +394,70 @@ class BaseOutputTransport(FrameProcessor):
     async def send_audio(self, frame: OutputAudioRawFrame):
         await self.process_frame(frame, FrameDirection.DOWNSTREAM)
 
+    def _next_audio_frame(self) -> AsyncGenerator[AudioRawFrame, None]:
+        async def without_mixer(vad_stop_secs: float) -> AsyncGenerator[AudioRawFrame, None]:
+            while True:
+                try:
+                    frame = await asyncio.wait_for(
+                        self._audio_out_queue.get(), timeout=vad_stop_secs
+                    )
+                    yield frame
+                except asyncio.TimeoutError:
+                    # Notify the bot stopped speaking upstream if necessary.
+                    await self._bot_stopped_speaking()
+
+        async def with_mixer(vad_stop_secs: float) -> AsyncGenerator[AudioRawFrame, None]:
+            last_frame_time = 0
+            silence = b"\x00" * self._audio_chunk_size
+            while True:
+                try:
+                    frame = self._audio_out_queue.get_nowait()
+                    frame.audio = await self._params.audio_out_mixer.mix(frame.audio)
+                    last_frame_time = time.time()
+                    yield frame
+                except asyncio.QueueEmpty:
+                    # Notify the bot stopped speaking upstream if necessary.
+                    diff_time = time.time() - last_frame_time
+                    if diff_time > vad_stop_secs:
+                        await self._bot_stopped_speaking()
+                    # Generate an audio frame with only the mixer's part.
+                    frame = OutputAudioRawFrame(
+                        audio=await self._params.audio_out_mixer.mix(silence),
+                        sample_rate=self._params.audio_out_sample_rate,
+                        num_channels=self._params.audio_out_channels,
+                    )
+                    yield frame
+
+        vad_stop_secs = (
+            self._params.vad_analyzer.params.stop_secs
+            if self._params.vad_analyzer
+            else VAD_STOP_SECS
+        )
+        if self._params.audio_out_mixer:
+            return with_mixer(vad_stop_secs)
+        else:
+            return without_mixer(vad_stop_secs)
+
     async def _audio_out_task_handler(self):
         wait_time = (
             self._params.vad_analyzer.params.stop_secs
             if self._params.vad_analyzer
             else VAD_STOP_SECS
         )
-        while True:
+        try:
-            try:
+            async for frame in self._next_audio_frame():
-                # If we don't have an audio frame for VAD stop secs we will
+                # Notify the bot started speaking upstream if necessary and that
-                # consider the bot is not speaking.
+                # it's actually speaking.
-                frame = await asyncio.wait_for(self._audio_out_queue.get(), timeout=wait_time)
-
-                # Notify the bot started speaking upstream if necessary.
                 if isinstance(frame, TTSAudioRawFrame):
                     await self._bot_started_speaking()
+                    await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM)
+
+                # Also, push frame downstream in case anyone else needs it.
+                await self.push_frame(frame)
 
                 # Send audio.
                 await self.write_raw_audio_frames(frame.audio)
-
+        except asyncio.CancelledError:
-                # Notify the bot is speaking upstream.
+            pass
-                if isinstance(frame, TTSAudioRawFrame):
+        except Exception as e:
-                    await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM)
+            logger.exception(f"{self} error writing to microphone: {e}")
-
-                # Push frame downstream in case anyone else needs it.
-                await self.push_frame(frame)
-            except asyncio.TimeoutError:
-                # Notify the bot stopped speaking upstream if necessary.
-                await self._bot_stopped_speaking()
-            except asyncio.CancelledError:
-                break
-            except Exception as e:
-                logger.exception(f"{self} error writing to camera: {e}")

src/pipecat/transports/base_transport.py

@@ -8,10 +8,12 @@ import asyncio
 import inspect
 
 from abc import ABC, abstractmethod
+from typing import Optional
 
 from pydantic import ConfigDict
 from pydantic.main import BaseModel
 
+from pipecat.audio.mixers.base_audio_mixer import BaseAudioMixer
 from pipecat.audio.vad.vad_analyzer import VADAnalyzer
 from pipecat.processors.frame_processor import FrameProcessor
 
@@ -33,6 +35,7 @@ class TransportParams(BaseModel):
     audio_out_sample_rate: int = 24000
     audio_out_channels: int = 1
     audio_out_bitrate: int = 96000
+    audio_out_mixer: Optional[BaseAudioMixer] = None
     audio_in_enabled: bool = False
     audio_in_sample_rate: int = 16000
     audio_in_channels: int = 1