audio: add BotBackgroundSound processor

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.

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

src/pipecat/services/background_noise.py

@@ -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)
-