Merge pull request #2676 from golbin/main

Fix audio buffer flush and silence handling

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(

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("<hh", 1000, -1000)
+        audio_event = asyncio.Event()
+        track_event = asyncio.Event()
+        captured = {}
+
+        async def on_audio_data(_, audio: bytes, sample_rate: int, num_channels: int):
+            captured["merged"] = (audio, sample_rate, num_channels)
+            audio_event.set()
+
+        async def on_track_audio_data(_, user: bytes, bot: bytes, sample_rate: int, num_channels: int):
+            captured["tracks"] = (user, bot, sample_rate, num_channels)
+            track_event.set()
+
+        self.processor.add_event_handler("on_audio_data", on_audio_data)
+        self.processor.add_event_handler("on_track_audio_data", on_track_audio_data)
+
+        frame = InputAudioRawFrame(audio=user_audio, sample_rate=16000, num_channels=1)
+        await self.processor._process_recording(frame)
+
+        await asyncio.wait_for(audio_event.wait(), timeout=1)
+        await asyncio.wait_for(track_event.wait(), timeout=1)
+
+        merged_audio, merged_sr, merged_channels = captured["merged"]
+        user_track, bot_track, track_sr, track_channels = captured["tracks"]
+
+        self.assertEqual(merged_sr, 16000)
+        self.assertEqual(merged_channels, 2)
+        self.assertEqual(track_sr, 16000)
+        self.assertEqual(track_channels, 2)
+        self.assertEqual(user_track, user_audio)
+        self.assertEqual(bot_track, b"\x00" * len(user_audio))
+        self.assertEqual(len(merged_audio), len(user_audio) * 2)
+        self.assertEqual(merged_audio[0:2], user_audio[0:2])
+        self.assertEqual(merged_audio[2:4], b"\x00\x00")
+        self.assertEqual(merged_audio[4:6], user_audio[2:4])
+        self.assertEqual(merged_audio[6:8], b"\x00\x00")
+        self.assertEqual(len(self.processor._user_audio_buffer), 0)
+        self.assertEqual(len(self.processor._bot_audio_buffer), 0)
+
+    async def test_flush_bot_audio_pads_user_track(self):
+        bot_audio = struct.pack("<hh", -800, 400)
+        audio_event = asyncio.Event()
+        track_event = asyncio.Event()
+        captured = {}
+
+        async def on_audio_data(_, audio: bytes, sample_rate: int, num_channels: int):
+            captured["merged"] = (audio, sample_rate, num_channels)
+            audio_event.set()
+
+        async def on_track_audio_data(_, user: bytes, bot: bytes, sample_rate: int, num_channels: int):
+            captured["tracks"] = (user, bot, sample_rate, num_channels)
+            track_event.set()
+
+        self.processor.add_event_handler("on_audio_data", on_audio_data)
+        self.processor.add_event_handler("on_track_audio_data", on_track_audio_data)
+
+        frame = OutputAudioRawFrame(audio=bot_audio, sample_rate=16000, num_channels=1)
+        await self.processor._process_recording(frame)
+
+        await asyncio.wait_for(audio_event.wait(), timeout=1)
+        await asyncio.wait_for(track_event.wait(), timeout=1)
+
+        merged_audio, merged_sr, merged_channels = captured["merged"]
+        user_track, bot_track, track_sr, track_channels = captured["tracks"]
+
+        self.assertEqual(merged_sr, 16000)
+        self.assertEqual(merged_channels, 2)
+        self.assertEqual(track_sr, 16000)
+        self.assertEqual(track_channels, 2)
+        self.assertEqual(user_track, b"\x00" * len(bot_audio))
+        self.assertEqual(bot_track, bot_audio)
+        self.assertEqual(len(merged_audio), len(bot_audio) * 2)
+        self.assertEqual(merged_audio[0:2], b"\x00\x00")
+        self.assertEqual(merged_audio[2:4], bot_audio[0:2])
+        self.assertEqual(merged_audio[4:6], b"\x00\x00")
+        self.assertEqual(merged_audio[6:8], bot_audio[2:4])
+        self.assertEqual(len(self.processor._user_audio_buffer), 0)
+        self.assertEqual(len(self.processor._bot_audio_buffer), 0)