diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d9a402eb..7ef8af2b3 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 @@ -41,6 +47,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- `GatedOpenAILLMContextAggregator` now require keyword arguments. Also, a new + `start_open` argument has been added to set the initial state of the gate. + - Added `organization` and `project` level authentication to `OpenAILLMService`. @@ -56,8 +65,27 @@ 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. + +### Removed + +- `AudioBufferProcessor.reset_audio_buffers()` has been removed, use + `AudioBufferProcessor.start_recording()` and + ``AudioBufferProcessor.stop_recording()` instead. + ### Fixed +- Fixed a `AudioBufferProcessor` that would cause crackling in some recordings. + +- Fixed an issue in `AudioBufferProcessor` where user callback would not be + called on task cancellation. + +- Fixed an issue in `AudioBufferProcessor` that would cause wrong silence + padding in some cases. + - Fixed an issue where `ElevenLabsTTSService` messages would return a 1009 websocket error by increasing the max message size limit to 16MB. @@ -1527,6 +1555,9 @@ async def on_connected(processor): ### Changed +- `FrameSerializer.serialize()` and `FrameSerializer.deserialize()` are now + `async`. + - `Filter` has been renamed to `FrameFilter` and it's now under `processors/filters`. diff --git a/examples/canonical-metrics/bot.py b/examples/canonical-metrics/bot.py index 945792b25..e2673bb3e 100644 --- a/examples/canonical-metrics/bot.py +++ b/examples/canonical-metrics/bot.py @@ -124,6 +124,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): + await audio_buffer_processor.start_recording() await transport.capture_participant_transcription(participant["id"]) await task.queue_frames([context_aggregator.user().get_context_frame()]) diff --git a/examples/chatbot-audio-recording/bot.py b/examples/chatbot-audio-recording/bot.py index 7cf750c62..62b7b29c8 100644 --- a/examples/chatbot-audio-recording/bot.py +++ b/examples/chatbot-audio-recording/bot.py @@ -109,8 +109,9 @@ async def main(): context = OpenAILLMContext(messages) context_aggregator = llm.create_context_aggregator(context) - # Save audio every 10 seconds. - audiobuffer = AudioBufferProcessor(buffer_size=480000) + # NOTE: Watch out! This will save all the conversation in memory. You + # can pass `buffer_size` to get periodic callbacks. + audiobuffer = AudioBufferProcessor() pipeline = Pipeline( [ @@ -132,6 +133,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): + await audiobuffer.start_recording() await transport.capture_participant_transcription(participant["id"]) await task.queue_frames([context_aggregator.user().get_context_frame()]) diff --git a/examples/foundational/22-natural-conversation.py b/examples/foundational/22-natural-conversation.py index 9b6471f7c..5e6b0e501 100644 --- a/examples/foundational/22-natural-conversation.py +++ b/examples/foundational/22-natural-conversation.py @@ -104,8 +104,11 @@ async def main(): ) # This processor keeps the last context and will let it through once the - # notifier is woken up. - gated_context_aggregator = GatedOpenAILLMContextAggregator(notifier) + # notifier is woken up. We start with the gate open because we send an + # initial context frame to start the conversation. + gated_context_aggregator = GatedOpenAILLMContextAggregator( + notifier=notifier, start_open=True + ) # Notify if the user hasn't said anything. async def user_idle_notifier(frame): diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index 8f19618a8..57639fe20 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -129,9 +129,9 @@ class CompletenessCheck(FrameProcessor): class OutputGate(FrameProcessor): - def __init__(self, notifier: BaseNotifier, **kwargs): + def __init__(self, *, notifier: BaseNotifier, start_open: bool = False, **kwargs): super().__init__(**kwargs) - self._gate_open = False + self._gate_open = start_open self._frames_buffer = [] self._notifier = notifier @@ -252,7 +252,9 @@ async def main(): # sentence, this will wake up the notifier if that happens. user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0) - bot_output_gate = OutputGate(notifier=notifier) + # We start with the gate open because we send an initial context frame + # to start the conversation. + bot_output_gate = OutputGate(notifier=notifier, start_open=True) async def block_user_stopped_speaking(frame): return not isinstance(frame, UserStoppedSpeakingFrame) diff --git a/examples/foundational/22c-natural-conversation-mixed-llms.py b/examples/foundational/22c-natural-conversation-mixed-llms.py index e4509da9c..1a24af21c 100644 --- a/examples/foundational/22c-natural-conversation-mixed-llms.py +++ b/examples/foundational/22c-natural-conversation-mixed-llms.py @@ -333,9 +333,9 @@ class CompletenessCheck(FrameProcessor): class OutputGate(FrameProcessor): - def __init__(self, notifier: BaseNotifier, **kwargs): + def __init__(self, *, notifier: BaseNotifier, start_open: bool = False, **kwargs): super().__init__(**kwargs) - self._gate_open = False + self._gate_open = start_open self._frames_buffer = [] self._notifier = notifier @@ -461,7 +461,9 @@ async def main(): # sentence, this will wake up the notifier if that happens. user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0) - bot_output_gate = OutputGate(notifier=notifier) + # We start with the gate open because we send an initial context frame + # to start the conversation. + bot_output_gate = OutputGate(notifier=notifier, start_open=True) async def block_user_stopped_speaking(frame): return not isinstance(frame, UserStoppedSpeakingFrame) 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..6afbbcbfe --- /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 + async 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..8c053fc3b --- /dev/null +++ b/src/pipecat/audio/resamplers/resampy_resampler.py @@ -0,0 +1,25 @@ +# +# 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 + + async 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") + result = resampled_audio.astype(np.int16).tobytes() + return result diff --git a/src/pipecat/audio/resamplers/soxr_resampler.py b/src/pipecat/audio/resamplers/soxr_resampler.py new file mode 100644 index 000000000..88edb84eb --- /dev/null +++ b/src/pipecat/audio/resamplers/soxr_resampler.py @@ -0,0 +1,25 @@ +# +# 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 + + async 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") + result = resampled_audio.astype(np.int16).tobytes() + return result diff --git a/src/pipecat/audio/utils.py b/src/pipecat/audio/utils.py index 8e95ebc31..7c4c25fcc 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,41 +91,45 @@ 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): +async 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 = await 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): +async 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 = await resampler.resample(pcm_bytes, in_rate, out_rate) # Convert PCM to μ-law - ulaw_bytes = audioop.lin2ulaw(in_pcm_bytes, 2) + out_ulaw_bytes = audioop.lin2ulaw(in_pcm_bytes, 2) - return ulaw_bytes + return out_ulaw_bytes -def alaw_to_pcm(alaw_bytes: bytes, in_sample_rate: int, out_sample_rate: int) -> bytes: +async 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 = await 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): +async 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 = await resampler.resample(pcm_bytes, in_rate, out_rate) # Convert PCM to μ-law - alaw_bytes = audioop.lin2alaw(in_pcm_bytes, 2) + out_alaw_bytes = audioop.lin2alaw(in_pcm_bytes, 2) - return alaw_bytes + return out_alaw_bytes diff --git a/src/pipecat/processors/aggregators/gated_openai_llm_context.py b/src/pipecat/processors/aggregators/gated_openai_llm_context.py index 8c3f496eb..f3ae78121 100644 --- a/src/pipecat/processors/aggregators/gated_openai_llm_context.py +++ b/src/pipecat/processors/aggregators/gated_openai_llm_context.py @@ -16,9 +16,10 @@ class GatedOpenAILLMContextAggregator(FrameProcessor): """ - def __init__(self, notifier: BaseNotifier, **kwargs): + def __init__(self, *, notifier: BaseNotifier, start_open: bool = False, **kwargs): super().__init__(**kwargs) self._notifier = notifier + self._start_open = start_open self._last_context_frame = None async def process_frame(self, frame: Frame, direction: FrameDirection): @@ -31,7 +32,11 @@ class GatedOpenAILLMContextAggregator(FrameProcessor): await self._stop() await self.push_frame(frame) elif isinstance(frame, OpenAILLMContextFrame): - self._last_context_frame = frame + if self._start_open: + self._start_open = False + await self.push_frame(frame, direction) + else: + self._last_context_frame = frame else: await self.push_frame(frame, direction) diff --git a/src/pipecat/processors/async_generator.py b/src/pipecat/processors/async_generator.py index e13026cdb..142d9cb47 100644 --- a/src/pipecat/processors/async_generator.py +++ b/src/pipecat/processors/async_generator.py @@ -30,7 +30,7 @@ class AsyncGeneratorProcessor(FrameProcessor): if isinstance(frame, (CancelFrame, EndFrame)): await self._data_queue.put(None) else: - data = self._serializer.serialize(frame) + data = await self._serializer.serialize(frame) if data: await self._data_queue.put(data) diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index fed4db3c8..d56604330 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -4,8 +4,12 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from pipecat.audio.utils import interleave_stereo_audio, mix_audio, resample_audio +import time + +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 +43,13 @@ class AudioBufferProcessor(FrameProcessor): self._user_audio_buffer = bytearray() self._bot_audio_buffer = bytearray() + self._last_user_frame_at = 0 + self._last_bot_frame_at = 0 + + self._recording = False + + self._resampler = create_default_resampler() + self._register_event_handler("on_audio_data") @property @@ -64,43 +75,76 @@ class AudioBufferProcessor(FrameProcessor): else: return b"" - def reset_audio_buffers(self): - self._user_audio_buffer = bytearray() - self._bot_audio_buffer = bytearray() + async def start_recording(self): + self._recording = True + self._reset_recording() + + async def stop_recording(self): + await self._call_on_audio_data_handler() + self._recording = False async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) - # Include all audio from the user. - if isinstance(frame, InputAudioRawFrame): - resampled = resample_audio(frame.audio, frame.sample_rate, self._sample_rate) + if self._recording and isinstance(frame, InputAudioRawFrame): + # Add silence if we need to. + silence = self._compute_silence(self._last_user_frame_at) + self._user_audio_buffer.extend(silence) + # Add user audio. + resampled = await 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): - silence = b"\x00" * len(resampled) - 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) + # Save time of frame so we can compute silence. + self._last_user_frame_at = time.time() + elif self._recording and isinstance(frame, OutputAudioRawFrame): + # Add silence if we need to. + silence = self._compute_silence(self._last_bot_frame_at) + self._bot_audio_buffer.extend(silence) + # Add bot audio. + resampled = await self._resample_audio(frame) self._bot_audio_buffer.extend(resampled) + # Save time of frame so we can compute silence. + self._last_bot_frame_at = time.time() if self._buffer_size > 0 and len(self._user_audio_buffer) > self._buffer_size: await self._call_on_audio_data_handler() - if isinstance(frame, EndFrame): - await self._call_on_audio_data_handler() + if isinstance(frame, (CancelFrame, EndFrame)): + await self.stop_recording() await self.push_frame(frame, direction) async def _call_on_audio_data_handler(self): - if not self.has_audio(): + if not self.has_audio() or not self._recording: return merged_audio = self.merge_audio_buffers() await self._call_event_handler( "on_audio_data", merged_audio, self._sample_rate, self._num_channels ) - self.reset_audio_buffers() + self._reset_audio_buffers() def _buffer_has_audio(self, buffer: bytearray) -> bool: return buffer is not None and len(buffer) > 0 + + def _reset_recording(self): + self._reset_audio_buffers() + self._last_user_frame_at = time.time() + self._last_bot_frame_at = time.time() + + def _reset_audio_buffers(self): + self._user_audio_buffer = bytearray() + self._bot_audio_buffer = bytearray() + + async def _resample_audio(self, frame: AudioRawFrame) -> bytes: + return await self._resampler.resample(frame.audio, frame.sample_rate, self._sample_rate) + + def _compute_silence(self, from_time: float) -> bytes: + quiet_time = time.time() - from_time + # We should get audio frames very frequently. We pick 100ms because + # that's big enough, but it could be even a bit slower since we usually + # do 20ms audio frames. + if from_time == 0 or quiet_time < 0.1: + return b"" + num_bytes = int(quiet_time * self._sample_rate) * 2 + silence = b"\x00" * num_bytes + return silence diff --git a/src/pipecat/serializers/base_serializer.py b/src/pipecat/serializers/base_serializer.py index 428649131..8a0efc49b 100644 --- a/src/pipecat/serializers/base_serializer.py +++ b/src/pipecat/serializers/base_serializer.py @@ -22,9 +22,9 @@ class FrameSerializer(ABC): pass @abstractmethod - def serialize(self, frame: Frame) -> str | bytes | None: + async def serialize(self, frame: Frame) -> str | bytes | None: pass @abstractmethod - def deserialize(self, data: str | bytes) -> Frame | None: + async def deserialize(self, data: str | bytes) -> Frame | None: pass diff --git a/src/pipecat/serializers/livekit.py b/src/pipecat/serializers/livekit.py index 760097ffe..d856a7a56 100644 --- a/src/pipecat/serializers/livekit.py +++ b/src/pipecat/serializers/livekit.py @@ -25,7 +25,7 @@ class LivekitFrameSerializer(FrameSerializer): def type(self) -> FrameSerializerType: return FrameSerializerType.BINARY - def serialize(self, frame: Frame) -> str | bytes | None: + async def serialize(self, frame: Frame) -> str | bytes | None: if not isinstance(frame, OutputAudioRawFrame): return None audio_frame = AudioFrame( @@ -36,7 +36,7 @@ class LivekitFrameSerializer(FrameSerializer): ) return pickle.dumps(audio_frame) - def deserialize(self, data: str | bytes) -> Frame | None: + async def deserialize(self, data: str | bytes) -> Frame | None: audio_frame: AudioFrame = pickle.loads(data)["frame"] return InputAudioRawFrame( audio=bytes(audio_frame.data), diff --git a/src/pipecat/serializers/protobuf.py b/src/pipecat/serializers/protobuf.py index 13d7ded34..125f2037f 100644 --- a/src/pipecat/serializers/protobuf.py +++ b/src/pipecat/serializers/protobuf.py @@ -41,7 +41,7 @@ class ProtobufFrameSerializer(FrameSerializer): def type(self) -> FrameSerializerType: return FrameSerializerType.BINARY - def serialize(self, frame: Frame) -> str | bytes | None: + async def serialize(self, frame: Frame) -> str | bytes | None: proto_frame = frame_protos.Frame() if type(frame) not in self.SERIALIZABLE_TYPES: logger.warning(f"Frame type {type(frame)} is not serializable") @@ -57,26 +57,7 @@ class ProtobufFrameSerializer(FrameSerializer): return proto_frame.SerializeToString() - def deserialize(self, data: str | bytes) -> Frame | None: - """Returns a Frame object from a Frame protobuf. - - Used to convert frames - passed over the wire as protobufs to Frame objects used in pipelines - and frame processors. - - >>> serializer = ProtobufFrameSerializer() - >>> serializer.deserialize( - ... serializer.serialize(OutputAudioFrame(data=b'1234567890'))) - InputAudioFrame(data=b'1234567890') - - >>> serializer.deserialize( - ... serializer.serialize(TextFrame(text='hello world'))) - TextFrame(text='hello world') - - >>> serializer.deserialize(serializer.serialize(TranscriptionFrame( - ... text="Hello there!", participantId="123", timestamp="2021-01-01"))) - TranscriptionFrame(text='Hello there!', participantId='123', timestamp='2021-01-01') - """ + async def deserialize(self, data: str | bytes) -> Frame | None: proto = frame_protos.Frame.FromString(data) which = proto.WhichOneof("frame") if which not in self.DESERIALIZABLE_FIELDS: diff --git a/src/pipecat/serializers/telnyx.py b/src/pipecat/serializers/telnyx.py index 14482531d..79c6f42ff 100644 --- a/src/pipecat/serializers/telnyx.py +++ b/src/pipecat/serializers/telnyx.py @@ -9,7 +9,13 @@ import json 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,21 +46,23 @@ class TelnyxFrameSerializer(FrameSerializer): params.inbound_encoding = inbound_encoding self._params = params + self._resampler = create_default_resampler() + @property def type(self) -> FrameSerializerType: return FrameSerializerType.TEXT - def serialize(self, frame: Frame) -> str | bytes | None: + async def serialize(self, frame: Frame) -> str | bytes | None: if isinstance(frame, AudioRawFrame): data = frame.audio if self._params.inbound_encoding == "PCMU": - serialized_data = pcm_to_ulaw( - data, frame.sample_rate, self._params.telnyx_sample_rate + serialized_data = await pcm_to_ulaw( + 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 + serialized_data = await pcm_to_alaw( + data, frame.sample_rate, self._params.telnyx_sample_rate, self._resampler ) else: raise ValueError(f"Unsupported encoding: {self._params.inbound_encoding}") @@ -71,7 +79,7 @@ class TelnyxFrameSerializer(FrameSerializer): answer = {"event": "clear"} return json.dumps(answer) - def deserialize(self, data: str | bytes) -> Frame | None: + async def deserialize(self, data: str | bytes) -> Frame | None: message = json.loads(data) if message["event"] == "media": @@ -79,12 +87,18 @@ class TelnyxFrameSerializer(FrameSerializer): payload = base64.b64decode(payload_base64) if self._params.outbound_encoding == "PCMU": - deserialized_data = ulaw_to_pcm( - payload, self._params.telnyx_sample_rate, self._params.sample_rate + deserialized_data = await ulaw_to_pcm( + 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 + deserialized_data = await alaw_to_pcm( + 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..7862fa0db 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,18 +32,22 @@ class TwilioFrameSerializer(FrameSerializer): self._stream_sid = stream_sid self._params = params + self._resampler = create_default_resampler() + @property def type(self) -> FrameSerializerType: return FrameSerializerType.TEXT - def serialize(self, frame: Frame) -> str | bytes | None: + async def serialize(self, frame: Frame) -> str | bytes | None: if isinstance(frame, StartInterruptionFrame): answer = {"event": "clear", "streamSid": self._stream_sid} return json.dumps(answer) elif isinstance(frame, AudioRawFrame): data = frame.audio - serialized_data = pcm_to_ulaw(data, frame.sample_rate, self._params.twilio_sample_rate) + serialized_data = await 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", @@ -55,15 +59,15 @@ class TwilioFrameSerializer(FrameSerializer): elif isinstance(frame, (TransportMessageFrame, TransportMessageUrgentFrame)): return json.dumps(frame.message) - def deserialize(self, data: str | bytes) -> Frame | None: + async def deserialize(self, data: str | bytes) -> Frame | None: message = json.loads(data) if message["event"] == "media": payload_base64 = message["media"]["payload"] payload = base64.b64decode(payload_base64) - deserialized_data = ulaw_to_pcm( - payload, self._params.twilio_sample_rate, self._params.sample_rate + deserialized_data = await ulaw_to_pcm( + 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..4444f07b7 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,8 +195,7 @@ 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"]) - return resampled + return audio_data return None logger.debug(f"Generating TTS: [{text}]") @@ -225,6 +226,10 @@ class PollyTTSService(TTSService): yield None return + audio_data = await self._resampler.resample( + audio_data, 16000, self._settings["sample_rate"] + ) + await self.start_tts_usage_metrics(text) yield TTSStartedFrame() diff --git a/src/pipecat/services/canonical.py b/src/pipecat/services/canonical.py index 884171b10..31bd6e89a 100644 --- a/src/pipecat/services/canonical.py +++ b/src/pipecat/services/canonical.py @@ -117,7 +117,6 @@ class CanonicalMetricsService(AIService): try: await self._multipart_upload(filename) await aiofiles.os.remove(filename) - audio_buffer_processor.reset_audio_buffers() except FileNotFoundError: pass except Exception as e: diff --git a/src/pipecat/services/tavus.py b/src/pipecat/services/tavus.py index b1dfef9e4..b0ca699cb 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 = await 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..a10247acd 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 = await self._resampler.resample( bytes(process_data), 24000, self._sample_rate ) # Create the frame with the resampled audio @@ -170,7 +172,9 @@ 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 = await 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/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index d4d458927..b53daa3b4 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -91,7 +91,7 @@ class FastAPIWebsocketInputTransport(BaseInputTransport): async def _receive_messages(self): try: async for message in self._iter_data(): - frame = self._params.serializer.deserialize(message) + frame = await self._params.serializer.deserialize(message) if not frame: continue @@ -163,7 +163,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): async def _write_frame(self, frame: Frame): try: - payload = self._params.serializer.serialize(frame) + payload = await self._params.serializer.serialize(frame) if payload and self._websocket.client_state == WebSocketState.CONNECTED: await self._send_data(payload) except Exception as e: diff --git a/src/pipecat/transports/network/websocket_client.py b/src/pipecat/transports/network/websocket_client.py index f2e26c4b5..56a7b57ac 100644 --- a/src/pipecat/transports/network/websocket_client.py +++ b/src/pipecat/transports/network/websocket_client.py @@ -138,7 +138,7 @@ class WebsocketClientInputTransport(BaseInputTransport): await self._session.disconnect() async def on_message(self, websocket, message): - frame = self._params.serializer.deserialize(message) + frame = await self._params.serializer.deserialize(message) if not frame: return if isinstance(frame, InputAudioRawFrame) and self._params.audio_in_enabled: @@ -200,7 +200,7 @@ class WebsocketClientOutputTransport(BaseOutputTransport): await self._write_audio_sleep() async def _write_frame(self, frame: Frame): - payload = self._params.serializer.serialize(frame) + payload = await self._params.serializer.serialize(frame) if payload: await self._session.send(payload) diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index c0dd299f6..684e030ad 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -105,7 +105,7 @@ class WebsocketServerInputTransport(BaseInputTransport): # Handle incoming messages try: async for message in websocket: - frame = self._params.serializer.deserialize(message) + frame = await self._params.serializer.deserialize(message) if not frame: continue @@ -193,7 +193,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport): async def _write_frame(self, frame: Frame): try: - payload = self._params.serializer.serialize(frame) + payload = await self._params.serializer.serialize(frame) if payload and self._websocket: await self._websocket.send(payload) except Exception as e: diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index 49f177179..11926f86c 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) @@ -384,7 +385,9 @@ class LiveKitInputTransport(BaseInputTransport): audio_data = await self._client.get_next_audio_frame() if audio_data: audio_frame_event, participant_id = audio_data - pipecat_audio_frame = self._convert_livekit_audio_to_pipecat(audio_frame_event) + pipecat_audio_frame = await self._convert_livekit_audio_to_pipecat( + audio_frame_event + ) input_audio_frame = InputAudioRawFrame( audio=pipecat_audio_frame.audio, sample_rate=pipecat_audio_frame.sample_rate, @@ -392,12 +395,12 @@ class LiveKitInputTransport(BaseInputTransport): ) await self.push_audio_frame(input_audio_frame) - def _convert_livekit_audio_to_pipecat( + async def _convert_livekit_audio_to_pipecat( self, audio_frame_event: rtc.AudioFrameEvent ) -> AudioRawFrame: audio_frame = audio_frame_event.frame - audio_data = resample_audio( + audio_data = await self._resampler.resample( audio_frame.data.tobytes(), audio_frame.sample_rate, self._params.audio_in_sample_rate ) diff --git a/tests/test_protobuf_serializer.py b/tests/test_protobuf_serializer.py index 88fd7867e..7d45800a8 100644 --- a/tests/test_protobuf_serializer.py +++ b/tests/test_protobuf_serializer.py @@ -20,17 +20,19 @@ class TestProtobufFrameSerializer(unittest.IsolatedAsyncioTestCase): async def test_roundtrip(self): text_frame = TextFrame(text="hello world") - frame = self.serializer.deserialize(self.serializer.serialize(text_frame)) + frame = await self.serializer.deserialize(await self.serializer.serialize(text_frame)) self.assertEqual(text_frame, frame) transcription_frame = TranscriptionFrame( text="Hello there!", user_id="123", timestamp="2021-01-01" ) - frame = self.serializer.deserialize(self.serializer.serialize(transcription_frame)) + frame = await self.serializer.deserialize( + await self.serializer.serialize(transcription_frame) + ) self.assertEqual(frame, transcription_frame) audio_frame = OutputAudioRawFrame(audio=b"1234567890", sample_rate=16000, num_channels=1) - frame = self.serializer.deserialize(self.serializer.serialize(audio_frame)) + frame = await self.serializer.deserialize(await self.serializer.serialize(audio_frame)) self.assertEqual(frame.audio, audio_frame.audio) self.assertEqual(frame.sample_rate, audio_frame.sample_rate) self.assertEqual(frame.num_channels, audio_frame.num_channels)