diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index 276051919..fefbafc9b 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -229,9 +229,12 @@ class AudioBufferProcessor(FrameProcessor): # 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: + if self._buffer_size > 0 and ( + len(self._user_audio_buffer) >= self._buffer_size + or len(self._bot_audio_buffer) >= self._buffer_size + ): await self._call_on_audio_data_handler() - self._reset_recording() + self._clear_primary_audio_buffers() # Process turn recording with preprocessed data. if self._enable_turn_audio: @@ -272,9 +275,15 @@ class AudioBufferProcessor(FrameProcessor): async def _call_on_audio_data_handler(self): """Call the audio data event handlers with buffered audio.""" - if not self.has_audio() or not self._recording: + if not self._recording: return + if len(self._user_audio_buffer) == 0 and len(self._bot_audio_buffer) == 0: + return + + self._align_track_buffers() + flush_time = time.time() + # Call original handler with merged audio merged_audio = self.merge_audio_buffers() await self._call_event_handler( @@ -290,6 +299,9 @@ class AudioBufferProcessor(FrameProcessor): self._num_channels, ) + self._last_user_frame_at = flush_time + self._last_bot_frame_at = flush_time + def _buffer_has_audio(self, buffer: bytearray) -> bool: """Check if a buffer contains audio data.""" return buffer is not None and len(buffer) > 0 @@ -307,6 +319,26 @@ class AudioBufferProcessor(FrameProcessor): self._user_turn_audio_buffer = bytearray() self._bot_turn_audio_buffer = bytearray() + def _clear_primary_audio_buffers(self): + """Clear user and bot buffers while preserving turn buffers and timestamps.""" + self._user_audio_buffer = bytearray() + self._bot_audio_buffer = bytearray() + + def _align_track_buffers(self): + """Pad the shorter track with silence so both tracks stay in sync.""" + user_len = len(self._user_audio_buffer) + bot_len = len(self._bot_audio_buffer) + if user_len == bot_len: + return + + target_len = max(user_len, bot_len) + if user_len < target_len: + self._user_audio_buffer.extend(b"\x00" * (target_len - user_len)) + self._last_user_frame_at = max(self._last_user_frame_at, self._last_bot_frame_at) + if bot_len < target_len: + self._bot_audio_buffer.extend(b"\x00" * (target_len - bot_len)) + self._last_bot_frame_at = max(self._last_bot_frame_at, self._last_user_frame_at) + async def _resample_input_audio(self, frame: InputAudioRawFrame) -> bytes: """Resample audio frame to the target sample rate.""" return await self._input_resampler.resample( diff --git a/tests/test_audio_buffer_processor.py b/tests/test_audio_buffer_processor.py new file mode 100644 index 000000000..24ae42265 --- /dev/null +++ b/tests/test_audio_buffer_processor.py @@ -0,0 +1,111 @@ +# +# Copyright (c) 2024-2025 Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import struct +import unittest + +from pipecat.frames.frames import InputAudioRawFrame, OutputAudioRawFrame, StartFrame +from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor + + +class _PassthroughResampler: + async def resample(self, audio: bytes, in_rate: int, out_rate: int) -> bytes: # pragma: no cover - trivial + return audio + + +class TestAudioBufferProcessor(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.processor = AudioBufferProcessor(sample_rate=16000, num_channels=2, buffer_size=4) + self.processor._input_resampler = _PassthroughResampler() + self.processor._output_resampler = _PassthroughResampler() + self.processor._update_sample_rate(StartFrame(audio_out_sample_rate=16000)) + await self.processor.start_recording() + + async def asyncTearDown(self): + if getattr(self.processor, "_recording", False): + await self.processor.stop_recording() + await self.processor.cleanup() + + async def test_flush_user_audio_pads_bot_track(self): + user_audio = struct.pack("