diff --git a/CHANGELOG.md b/CHANGELOG.md index 98cab2a05..c8fbf1515 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `BotBackgroundSound` processor. This processors allows you to add + background sound to the bots output. The background sound will always be + playing even if the bot is not talking. The volume of the background sound and + the sample rate can be configure. You can load any file format supported by + the `soundfile` library. + (see https://github.com/bastibe/python-soundfile) + - 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/src/pipecat/processors/audio/background_sound.py b/src/pipecat/processors/audio/background_sound.py new file mode 100644 index 000000000..98a951d75 --- /dev/null +++ b/src/pipecat/processors/audio/background_sound.py @@ -0,0 +1,134 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio + +import numpy as np + +from pipecat.audio.utils import resample_audio +from pipecat.processors.frame_processor import FrameProcessor, FrameDirection +from pipecat.frames.frames import ( + CancelFrame, + OutputAudioRawFrame, + Frame, + EndFrame, + StartFrame, + TTSAudioRawFrame, + TTSStartedFrame, + TTSStoppedFrame, +) + +from loguru import logger + +try: + import soundfile as sf +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error( + "In order to use background sound, you need to `pip install pipecat-ai[soundfile]`." + ) + raise Exception(f"Missing module: {e}") + + +class BotBackgroundSound(FrameProcessor): + def __init__( + self, + file_name: str, + volume: float = 0.4, + sample_rate: int = 24000, + **kwargs, + ): + super().__init__(**kwargs) + self._file_name = file_name + self._volume = volume + self._sample_rate = sample_rate + + self._sound = np.array([], dtype=np.int16) + self._sound_pos = 0 + + self._bot_speaking = False + self._sleep_time = 0.02 + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, StartFrame): + await self._start() + await self.push_frame(frame, direction) + elif isinstance(frame, (EndFrame, CancelFrame)): + await self._stop() + await self.push_frame(frame, direction) + elif isinstance(frame, TTSStartedFrame): + self._bot_speaking = True + elif isinstance(frame, TTSStoppedFrame): + self._bot_speaking = False + elif isinstance(frame, TTSAudioRawFrame): + frame.audio = self._mix_with_sound(frame.audio) + await self.push_frame(frame) + else: + await self.push_frame(frame, direction) + + async def _start(self): + try: + logger.debug(f"{self} loading background sound from {self._file_name}") + sound, sample_rate = sf.read(self._file_name, dtype="int16") + + audio = sound.tobytes() + if sample_rate != self._sample_rate: + logger.debug(f"{self} resampling background sound to {self._sample_rate}") + audio = resample_audio(audio, sample_rate, self._sample_rate) + + # Convert from np to bytes again. + self._sound = np.frombuffer(audio, dtype=np.int16) + + self._audio_queue = asyncio.Queue() + self._audio_task = self.get_event_loop().create_task(self._audio_task_handler()) + except Exception as ex: + logger.error(f"{self} unable to open file {self._file_name}") + + async def _stop(self): + self._audio_task.cancel() + await self._audio_task + + def _mix_with_sound(self, audio: bytes): + """Mixes raw audio frames with chunks of the same length from the sound + file. + + """ + if audio: + audio_np = np.frombuffer(audio, dtype=np.int16) + else: + num_samples = int(self._sleep_time * self._sample_rate) + audio_np = np.zeros(num_samples, dtype=np.int16) + + chunk_size = len(audio_np) + + # Go back to the beginning if we don't have enough data. + if self._sound_pos + chunk_size > len(self._sound): + self._sound_pos = 0 + + start_pos = self._sound_pos + end_pos = self._sound_pos + chunk_size + self._sound_pos = end_pos + + sound_np = self._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() + + async def _audio_task_handler(self): + while True: + try: + if not self._bot_speaking: + audio = self._mix_with_sound(b"") + frame = OutputAudioRawFrame( + audio=audio, sample_rate=self._sample_rate, num_channels=1 + ) + await self.push_frame(frame) + await asyncio.sleep(self._sleep_time) + except asyncio.CancelledError: + break diff --git a/src/pipecat/services/background_noise.py b/src/pipecat/services/background_noise.py deleted file mode 100644 index 8f502fec4..000000000 --- a/src/pipecat/services/background_noise.py +++ /dev/null @@ -1,120 +0,0 @@ -import asyncio -import json -import time -from asyncio import sleep -from io import BytesIO - -import loguru - -from pipecat.processors.frame_processor import FrameProcessor, FrameDirection -from pydub import AudioSegment -from pipecat.frames.frames import AudioRawFrame, OutputAudioRawFrame, Frame, BotStartedSpeakingFrame, \ - BotStoppedSpeakingFrame, EndFrame - - -class BackgroundNoiseEffect(FrameProcessor): - def __init__(self, websocket_client, stream_sid, music_path): - super().__init__(sync=False) - self._speaking = True - self._audio_task = self.get_event_loop().create_task(self._audio_task_handler()) - self._audio_queue = asyncio.Queue() - self._stop = False - self.stream_sid = stream_sid - self.websocket_client = websocket_client - self.music_path = music_path - self.get_music_part_gen = self._get_music_part() - self.emptied = False - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if isinstance(frame, BotStartedSpeakingFrame): - self._speaking = True - - if isinstance(frame, BotStoppedSpeakingFrame): - self._speaking = False - self.emptied = False - - if isinstance(frame, AudioRawFrame) and self._speaking: - if not self.emptied: - self.emptied = True - buffer_clear_message = {"event": "clear", "streamSid": self.stream_sid} - await self.websocket_client.send_text(json.dumps(buffer_clear_message)) - - frame.audio = self._combine_with_music(frame) - - if isinstance(frame, EndFrame): - self._stop = True - - await self.push_frame(frame, direction) - - def _combine_with_music(self, frame: AudioRawFrame): - """ - Combines small raw audio segments from the frame with chunks of a music file. - """ - small_audio_bytes = frame.audio - music_audio = AudioSegment.from_wav(self.music_path) - music_audio = music_audio - 15 - - music_position = 0 - small_audio = AudioSegment( - data=small_audio_bytes, - sample_width=2, - frame_rate=16000, - channels=1 - ) - - small_audio_length = len(small_audio) - music_chunk = music_audio[music_position:music_position + small_audio_length] - - if len(music_chunk) < small_audio_length: - music_position = 0 - music_chunk += music_audio[:small_audio_length - len(music_chunk)] - - combined_audio = music_chunk.overlay(small_audio) - music_position += small_audio_length - - output_buffer = BytesIO() - try: - combined_audio.export(output_buffer, format="raw") - return output_buffer.getvalue() - finally: - output_buffer.close() - - def _get_music_part(self): - """ - Generator that yields chunks of background music audio. - """ - music_audio = AudioSegment.from_wav(self.music_path) - music_audio = music_audio - 15 - - music_position = 0 - small_audio_length = 6400 - - while True: - if music_position + small_audio_length > len(music_audio): - music_chunk = music_audio[music_position:] + music_audio[ - :(music_position + small_audio_length) % len(music_audio)] - music_position = (music_position + small_audio_length) % len(music_audio) - else: - music_chunk = music_audio[music_position:music_position + small_audio_length] - music_position += small_audio_length - - output_buffer = BytesIO() - try: - music_chunk.export(output_buffer, format="raw") - frame = OutputAudioRawFrame(audio=output_buffer.getvalue(), sample_rate=16000, num_channels=1) - yield frame - finally: - output_buffer.close() - - async def _audio_task_handler(self): - while True: - await sleep(0.005) - if self._stop: - break - - if not self._speaking: - frame = next(self.get_music_part_gen) - await self.push_frame(frame, FrameDirection.DOWNSTREAM) -