diff --git a/CHANGELOG.md b/CHANGELOG.md index 98cab2a05..dd56eb011 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/pyproject.toml b/pyproject.toml index d0422dacd..115822cbf 100644 --- a/pyproject.toml +++ b/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" ] diff --git a/src/pipecat/audio/mixers/__init__.py b/src/pipecat/audio/mixers/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/audio/mixers/base_audio_mixer.py b/src/pipecat/audio/mixers/base_audio_mixer.py new file mode 100644 index 000000000..bd9d03b89 --- /dev/null +++ b/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 diff --git a/src/pipecat/audio/mixers/soundfile_mixer.py b/src/pipecat/audio/mixers/soundfile_mixer.py new file mode 100644 index 000000000..a9a17a106 --- /dev/null +++ b/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() diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 517918a52..faa029635 100644 --- a/src/pipecat/frames/frames.py +++ b/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 diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index e604fd289..15091255d 100644 --- a/src/pipecat/transports/base_output.py +++ b/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: - # If we don't have an audio frame for VAD stop secs we will - # consider the bot is not speaking. - frame = await asyncio.wait_for(self._audio_out_queue.get(), timeout=wait_time) - - # Notify the bot started speaking upstream if necessary. + try: + async for frame in self._next_audio_frame(): + # Notify the bot started speaking upstream if necessary and that + # it's actually speaking. 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) - - # Notify the bot is speaking upstream. - if isinstance(frame, TTSAudioRawFrame): - await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM) - - # 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}") + except asyncio.CancelledError: + pass + except Exception as e: + logger.exception(f"{self} error writing to microphone: {e}") diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index 2ca2b6470..946a8fb90 100644 --- a/src/pipecat/transports/base_transport.py +++ b/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