diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d9a402eb..f9e1e4c34 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Introduce audio resamplers (`BaseAudioResampler`). This is just a base class + to implement audio resamplers. Currently, two implementations are provided + `SOXRAudioResampler` and `ResampyResampler`. A new + `create_default_resampler()` has been added (replacing the now deprecated + `resample_audio()`). + - It is now possible to specify the asyncio event loop that a `PipelineTask` and all the processors should run on by passing it as a new argument to the `PipelineRunner`. This could allow running pipelines in multiple threads each @@ -56,6 +62,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `InputDTMFFrame` is now based on `DTMFFrame`. There's also a new `OutputDTMFFrame` frame. +### Deprecated + +- `resample_audio()` is now deprecated, use `create_default_resampler()` + instead. + ### Fixed - Fixed an issue where `ElevenLabsTTSService` messages would return a 1009 diff --git a/pyproject.toml b/pyproject.toml index 801948c8f..ee911098f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,6 +32,7 @@ dependencies = [ "protobuf~=5.29.3", "pydantic~=2.10.5", "pyloudnorm~=0.1.1", + "resampy~=0.4.3", "soxr~=0.5.0" ] diff --git a/src/pipecat/audio/resamplers/__init__.py b/src/pipecat/audio/resamplers/__init__.py new file mode 100644 index 000000000..8b1378917 --- /dev/null +++ b/src/pipecat/audio/resamplers/__init__.py @@ -0,0 +1 @@ + diff --git a/src/pipecat/audio/resamplers/base_audio_resampler.py b/src/pipecat/audio/resamplers/base_audio_resampler.py new file mode 100644 index 000000000..0d7c3afef --- /dev/null +++ b/src/pipecat/audio/resamplers/base_audio_resampler.py @@ -0,0 +1,30 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from abc import ABC, abstractmethod + + +class BaseAudioResampler(ABC): + """Abstract base class for audio resampling. This class defines an + interface for audio resampling implementations. + """ + + @abstractmethod + def resample(self, audio: bytes, in_rate: int, out_rate: int) -> bytes: + """ + Resamples the given audio data to a different sample rate. + + This is an abstract method that must be implemented in subclasses. + + Parameters: + audio (bytes): The audio data to be resampled, represented as a byte string. + in_rate (int): The original sample rate of the audio data (in Hz). + out_rate (int): The desired sample rate for the resampled audio data (in Hz). + + Returns: + bytes: The resampled audio data as a byte string. + """ + pass diff --git a/src/pipecat/audio/resamplers/resampy_resampler.py b/src/pipecat/audio/resamplers/resampy_resampler.py new file mode 100644 index 000000000..be194b767 --- /dev/null +++ b/src/pipecat/audio/resamplers/resampy_resampler.py @@ -0,0 +1,24 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import numpy as np +import resampy + +from pipecat.audio.resamplers.base_audio_resampler import BaseAudioResampler + + +class ResampyResampler(BaseAudioResampler): + """Audio resampler implementation using the resampy library.""" + + def __init__(self, **kwargs): + pass + + def resample(self, audio: bytes, in_rate: int, out_rate: int) -> bytes: + if in_rate == out_rate: + return audio + audio_data = np.frombuffer(audio, dtype=np.int16) + resampled_audio = resampy.resample(audio_data, in_rate, out_rate, filter="kaiser_fast") + return resampled_audio.astype(np.int16).tobytes() diff --git a/src/pipecat/audio/resamplers/soxr_resampler.py b/src/pipecat/audio/resamplers/soxr_resampler.py new file mode 100644 index 000000000..eaa06ad4e --- /dev/null +++ b/src/pipecat/audio/resamplers/soxr_resampler.py @@ -0,0 +1,24 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import numpy as np +import soxr + +from pipecat.audio.resamplers.base_audio_resampler import BaseAudioResampler + + +class SOXRAudioResampler(BaseAudioResampler): + """Audio resampler implementation using the SoX resampler library.""" + + def __init__(self, **kwargs): + pass + + def resample(self, audio: bytes, in_rate: int, out_rate: int) -> bytes: + if in_rate == out_rate: + return audio + audio_data = np.frombuffer(audio, dtype=np.int16) + resampled_audio = soxr.resample(audio_data, in_rate, out_rate, quality="VHQ") + return resampled_audio.astype(np.int16).tobytes() diff --git a/src/pipecat/audio/utils.py b/src/pipecat/audio/utils.py index 8e95ebc31..9a9d442dd 100644 --- a/src/pipecat/audio/utils.py +++ b/src/pipecat/audio/utils.py @@ -10,8 +10,24 @@ import numpy as np import pyloudnorm as pyln import soxr +from pipecat.audio.resamplers.base_audio_resampler import BaseAudioResampler +from pipecat.audio.resamplers.soxr_resampler import SOXRAudioResampler + + +def create_default_resampler(**kwargs) -> BaseAudioResampler: + return SOXRAudioResampler(**kwargs) + def resample_audio(audio: bytes, original_rate: int, target_rate: int) -> bytes: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "'resample_audio()' is deprecated, use 'create_default_resampler()' instead.", + DeprecationWarning, + ) + if original_rate == target_rate: return audio audio_data = np.frombuffer(audio, dtype=np.int16) @@ -75,19 +91,19 @@ def exp_smoothing(value: float, prev_value: float, factor: float) -> float: return prev_value + factor * (value - prev_value) -def ulaw_to_pcm(ulaw_bytes: bytes, in_sample_rate: int, out_sample_rate: int): +def ulaw_to_pcm(ulaw_bytes: bytes, in_rate: int, out_rate: int, resampler: BaseAudioResampler): # Convert μ-law to PCM in_pcm_bytes = audioop.ulaw2lin(ulaw_bytes, 2) # Resample - out_pcm_bytes = resample_audio(in_pcm_bytes, in_sample_rate, out_sample_rate) + out_pcm_bytes = resampler.resample(in_pcm_bytes, in_rate, out_rate) return out_pcm_bytes -def pcm_to_ulaw(pcm_bytes: bytes, in_sample_rate: int, out_sample_rate: int): +def pcm_to_ulaw(pcm_bytes: bytes, in_rate: int, out_rate: int, resampler: BaseAudioResampler): # Resample - in_pcm_bytes = resample_audio(pcm_bytes, in_sample_rate, out_sample_rate) + in_pcm_bytes = resampler.resample(pcm_bytes, in_rate, out_rate) # Convert PCM to μ-law ulaw_bytes = audioop.lin2ulaw(in_pcm_bytes, 2) @@ -95,19 +111,21 @@ def pcm_to_ulaw(pcm_bytes: bytes, in_sample_rate: int, out_sample_rate: int): return ulaw_bytes -def alaw_to_pcm(alaw_bytes: bytes, in_sample_rate: int, out_sample_rate: int) -> bytes: +def alaw_to_pcm( + alaw_bytes: bytes, in_rate: int, out_rate: int, resampler: BaseAudioResampler +) -> bytes: # Convert a-law to PCM in_pcm_bytes = audioop.alaw2lin(alaw_bytes, 2) # Resample - out_pcm_bytes = resample_audio(in_pcm_bytes, in_sample_rate, out_sample_rate) + out_pcm_bytes = resampler.resample(in_pcm_bytes, in_rate, out_rate) return out_pcm_bytes -def pcm_to_alaw(pcm_bytes: bytes, in_sample_rate: int, out_sample_rate: int): +def pcm_to_alaw(pcm_bytes: bytes, in_rate: int, out_rate: int, resampler: BaseAudioResampler): # Resample - in_pcm_bytes = resample_audio(pcm_bytes, in_sample_rate, out_sample_rate) + in_pcm_bytes = resampler.resample(pcm_bytes, in_rate, out_rate) # Convert PCM to μ-law alaw_bytes = audioop.lin2alaw(in_pcm_bytes, 2) diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index fed4db3c8..7500a3f9e 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -4,8 +4,10 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from pipecat.audio.utils import interleave_stereo_audio, mix_audio, resample_audio +from pipecat.audio.utils import create_default_resampler, interleave_stereo_audio, mix_audio from pipecat.frames.frames import ( + AudioRawFrame, + CancelFrame, EndFrame, Frame, InputAudioRawFrame, @@ -39,6 +41,8 @@ class AudioBufferProcessor(FrameProcessor): self._user_audio_buffer = bytearray() self._bot_audio_buffer = bytearray() + self._resampler = create_default_resampler() + self._register_event_handler("on_audio_data") @property @@ -73,7 +77,7 @@ class AudioBufferProcessor(FrameProcessor): # Include all audio from the user. if isinstance(frame, InputAudioRawFrame): - resampled = resample_audio(frame.audio, frame.sample_rate, self._sample_rate) + resampled = self._resample_audio(frame) self._user_audio_buffer.extend(resampled) # Sync the bot's buffer to the user's buffer by adding silence if needed if len(self._user_audio_buffer) > len(self._bot_audio_buffer): @@ -81,7 +85,7 @@ class AudioBufferProcessor(FrameProcessor): self._bot_audio_buffer.extend(silence) # If the bot is speaking, include all audio from the bot. elif isinstance(frame, OutputAudioRawFrame): - resampled = resample_audio(frame.audio, frame.sample_rate, self._sample_rate) + resampled = self._resample_audio(frame) self._bot_audio_buffer.extend(resampled) if self._buffer_size > 0 and len(self._user_audio_buffer) > self._buffer_size: @@ -104,3 +108,6 @@ class AudioBufferProcessor(FrameProcessor): def _buffer_has_audio(self, buffer: bytearray) -> bool: return buffer is not None and len(buffer) > 0 + + def _resample_audio(self, frame: AudioRawFrame) -> bytes: + return self._resampler.resample(frame.audio, frame.sample_rate, self._sample_rate) diff --git a/src/pipecat/serializers/telnyx.py b/src/pipecat/serializers/telnyx.py index 14482531d..aacb6b646 100644 --- a/src/pipecat/serializers/telnyx.py +++ b/src/pipecat/serializers/telnyx.py @@ -6,10 +6,17 @@ import base64 import json +from typing import Optional from pydantic import BaseModel -from pipecat.audio.utils import alaw_to_pcm, pcm_to_alaw, pcm_to_ulaw, ulaw_to_pcm +from pipecat.audio.utils import ( + alaw_to_pcm, + create_default_resampler, + pcm_to_alaw, + pcm_to_ulaw, + ulaw_to_pcm, +) from pipecat.frames.frames import ( AudioRawFrame, Frame, @@ -40,6 +47,8 @@ class TelnyxFrameSerializer(FrameSerializer): params.inbound_encoding = inbound_encoding self._params = params + self._resampler = create_default_resampler() + @property def type(self) -> FrameSerializerType: return FrameSerializerType.TEXT @@ -50,11 +59,11 @@ class TelnyxFrameSerializer(FrameSerializer): if self._params.inbound_encoding == "PCMU": serialized_data = pcm_to_ulaw( - data, frame.sample_rate, self._params.telnyx_sample_rate + data, frame.sample_rate, self._params.telnyx_sample_rate, self._resampler ) elif self._params.inbound_encoding == "PCMA": serialized_data = pcm_to_alaw( - data, frame.sample_rate, self._params.telnyx_sample_rate + data, frame.sample_rate, self._params.telnyx_sample_rate, self._resampler ) else: raise ValueError(f"Unsupported encoding: {self._params.inbound_encoding}") @@ -80,11 +89,17 @@ class TelnyxFrameSerializer(FrameSerializer): if self._params.outbound_encoding == "PCMU": deserialized_data = ulaw_to_pcm( - payload, self._params.telnyx_sample_rate, self._params.sample_rate + payload, + self._params.telnyx_sample_rate, + self._params.sample_rate, + self._resampler, ) elif self._params.outbound_encoding == "PCMA": deserialized_data = alaw_to_pcm( - payload, self._params.telnyx_sample_rate, self._params.sample_rate + payload, + self._params.telnyx_sample_rate, + self._params.sample_rate, + self._resampler, ) else: raise ValueError(f"Unsupported encoding: {self._params.outbound_encoding}") diff --git a/src/pipecat/serializers/twilio.py b/src/pipecat/serializers/twilio.py index 40c8c726f..7da7a80c7 100644 --- a/src/pipecat/serializers/twilio.py +++ b/src/pipecat/serializers/twilio.py @@ -9,7 +9,7 @@ import json from pydantic import BaseModel -from pipecat.audio.utils import pcm_to_ulaw, ulaw_to_pcm +from pipecat.audio.utils import create_default_resampler, pcm_to_ulaw, ulaw_to_pcm from pipecat.frames.frames import ( AudioRawFrame, Frame, @@ -32,6 +32,8 @@ class TwilioFrameSerializer(FrameSerializer): self._stream_sid = stream_sid self._params = params + self._resampler = create_default_resampler() + @property def type(self) -> FrameSerializerType: return FrameSerializerType.TEXT @@ -43,7 +45,9 @@ class TwilioFrameSerializer(FrameSerializer): elif isinstance(frame, AudioRawFrame): data = frame.audio - serialized_data = pcm_to_ulaw(data, frame.sample_rate, self._params.twilio_sample_rate) + serialized_data = pcm_to_ulaw( + data, frame.sample_rate, self._params.twilio_sample_rate, self._resampler + ) payload = base64.b64encode(serialized_data).decode("utf-8") answer = { "event": "media", @@ -63,7 +67,7 @@ class TwilioFrameSerializer(FrameSerializer): payload = base64.b64decode(payload_base64) deserialized_data = ulaw_to_pcm( - payload, self._params.twilio_sample_rate, self._params.sample_rate + payload, self._params.twilio_sample_rate, self._params.sample_rate, self._resampler ) audio_frame = InputAudioRawFrame( audio=deserialized_data, num_channels=1, sample_rate=self._params.sample_rate diff --git a/src/pipecat/services/aws.py b/src/pipecat/services/aws.py index 3061e13a1..eb03bbc59 100644 --- a/src/pipecat/services/aws.py +++ b/src/pipecat/services/aws.py @@ -10,7 +10,7 @@ from typing import AsyncGenerator, Optional from loguru import logger from pydantic import BaseModel -from pipecat.audio.utils import resample_audio +from pipecat.audio.utils import create_default_resampler from pipecat.frames.frames import ( ErrorFrame, Frame, @@ -148,6 +148,8 @@ class PollyTTSService(TTSService): "volume": params.volume, } + self._resampler = create_default_resampler() + self.set_voice(voice_id) def can_generate_metrics(self) -> bool: @@ -193,7 +195,9 @@ class PollyTTSService(TTSService): response = self._polly_client.synthesize_speech(**args) if "AudioStream" in response: audio_data = response["AudioStream"].read() - resampled = resample_audio(audio_data, 16000, self._settings["sample_rate"]) + resampled = self._resampler.resample( + audio_data, 16000, self._settings["sample_rate"] + ) return resampled return None diff --git a/src/pipecat/services/tavus.py b/src/pipecat/services/tavus.py index b1dfef9e4..e6693ecc7 100644 --- a/src/pipecat/services/tavus.py +++ b/src/pipecat/services/tavus.py @@ -12,7 +12,7 @@ import base64 import aiohttp from loguru import logger -from pipecat.audio.utils import resample_audio +from pipecat.audio.utils import create_default_resampler from pipecat.frames.frames import ( CancelFrame, EndFrame, @@ -47,6 +47,8 @@ class TavusVideoService(AIService): self._conversation_id: str + self._resampler = create_default_resampler() + async def initialize(self) -> str: url = "https://tavusapi.com/v2/conversations" headers = {"Content-Type": "application/json", "x-api-key": self._api_key} @@ -89,12 +91,10 @@ class TavusVideoService(AIService): async with self._session.post(url, headers=headers) as r: r.raise_for_status() - async def _encode_audio_and_send( - self, audio: bytes, original_sample_rate: int, done: bool - ) -> None: + async def _encode_audio_and_send(self, audio: bytes, in_rate: int, done: bool) -> None: """Encodes audio to base64 and sends it to Tavus""" if not done: - audio = resample_audio(audio, original_sample_rate, 16000) + audio = self._resampler.resample(audio, in_rate, 16000) audio_base64 = base64.b64encode(audio).decode("utf-8") logger.trace(f"{self}: sending {len(audio)} bytes") await self._send_audio_message(audio_base64, done=done) diff --git a/src/pipecat/services/xtts.py b/src/pipecat/services/xtts.py index ba8803f51..b26660175 100644 --- a/src/pipecat/services/xtts.py +++ b/src/pipecat/services/xtts.py @@ -9,7 +9,7 @@ from typing import Any, AsyncGenerator, Dict import aiohttp from loguru import logger -from pipecat.audio.utils import resample_audio +from pipecat.audio.utils import create_default_resampler from pipecat.frames.frames import ( ErrorFrame, Frame, @@ -89,6 +89,8 @@ class XTTSService(TTSService): self._studio_speakers: Dict[str, Any] | None = None self._aiohttp_session = aiohttp_session + self._resampler = create_default_resampler() + def can_generate_metrics(self) -> bool: return True @@ -161,7 +163,7 @@ class XTTSService(TTSService): buffer = buffer[48000:] # XTTS uses 24000 so we need to resample to our desired rate. - resampled_audio = resample_audio( + resampled_audio = self._resampler.resample( bytes(process_data), 24000, self._sample_rate ) # Create the frame with the resampled audio @@ -170,7 +172,7 @@ class XTTSService(TTSService): # Process any remaining data in the buffer. if len(buffer) > 0: - resampled_audio = resample_audio(bytes(buffer), 24000, self._sample_rate) + resampled_audio = self._resampler.resample(bytes(buffer), 24000, self._sample_rate) frame = TTSAudioRawFrame(resampled_audio, self._sample_rate, 1) yield frame diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index 49f177179..a80e2fa32 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/src/pipecat/transports/services/livekit.py @@ -11,7 +11,7 @@ from typing import Any, Awaitable, Callable, List, Optional from loguru import logger from pydantic import BaseModel -from pipecat.audio.utils import resample_audio +from pipecat.audio.utils import create_default_resampler from pipecat.audio.vad.vad_analyzer import VADAnalyzer from pipecat.frames.frames import ( AudioRawFrame, @@ -349,6 +349,7 @@ class LiveKitInputTransport(BaseInputTransport): self._client = client self._audio_in_task = None self._vad_analyzer: VADAnalyzer | None = params.vad_analyzer + self._resampler = create_default_resampler() async def start(self, frame: StartFrame): await super().start(frame) @@ -397,7 +398,7 @@ class LiveKitInputTransport(BaseInputTransport): ) -> AudioRawFrame: audio_frame = audio_frame_event.frame - audio_data = resample_audio( + audio_data = self._resampler.resample( audio_frame.data.tobytes(), audio_frame.sample_rate, self._params.audio_in_sample_rate )