diff --git a/CHANGELOG.md b/CHANGELOG.md index 24502827d..de1a9c1f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added `OutputDTMFUrgentFrame` to send a DTMF keypress quickly. The previous `OutputDTMFFrame` queues the keypress with the rest of data frames. +- Added `DTMFAggregator`, which aggregates keypad presses into + `TranscriptionFrame`s. Aggregation occurs after a timeout, termination key + press, or user interruption. You can specify the prefix of the + `TranscriptionFrame`. + - Added new functions `DailyTransport.start_transcription()` and `DailyTransport.stop_transcription()` to be able to start and stop Daily transcription dynamically (maybe with different settings). diff --git a/src/pipecat/processors/aggregators/dtmf_aggregator.py b/src/pipecat/processors/aggregators/dtmf_aggregator.py new file mode 100644 index 000000000..cc6218a1f --- /dev/null +++ b/src/pipecat/processors/aggregators/dtmf_aggregator.py @@ -0,0 +1,143 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +from typing import Optional + +from pipecat.frames.frames import ( + BotInterruptionFrame, + CancelFrame, + EndFrame, + Frame, + InputDTMFFrame, + KeypadEntry, + StartFrame, + TranscriptionFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.utils.time import time_now_iso8601 + + +class DTMFAggregator(FrameProcessor): + """Aggregates DTMF frames into meaningful sequences for LLM processing. + + The aggregator accumulates digits from InputDTMFFrame instances and flushes + when: + - Timeout occurs (configurable idle period) + - Termination digit is received (default: '#') + - EndFrame or CancelFrame is received + + Emits TranscriptionFrame for compatibility with existing LLM context aggregators. + + Args: + timeout: Idle timeout in seconds before flushing + termination_digit: Digit that triggers immediate flush + prefix: Prefix added to DTMF sequence in transcription + """ + + def __init__( + self, + timeout: float = 2.0, + termination_digit: KeypadEntry = KeypadEntry.POUND, + prefix: str = "DTMF: ", + **kwargs, + ): + super().__init__(**kwargs) + self._aggregation = "" + self._idle_timeout = timeout + self._termination_digit = termination_digit + self._prefix = prefix + + self._digit_event = asyncio.Event() + self._aggregation_task: Optional[asyncio.Task] = None + + async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: + await super().process_frame(frame, direction) + + if isinstance(frame, StartFrame): + self._create_aggregation_task() + await self.push_frame(frame, direction) + elif isinstance(frame, (EndFrame, CancelFrame)): + if self._aggregation: + await self._flush_aggregation() + await self._stop_aggregation_task() + await self.push_frame(frame, direction) + elif isinstance(frame, InputDTMFFrame): + # Push the DTMF frame downstream first + await self.push_frame(frame, direction) + # Then handle it in order for the TranscriptionFrame to be emitted + # after the InputDTMFFrame + await self._handle_dtmf_frame(frame) + else: + await self.push_frame(frame, direction) + + async def _handle_dtmf_frame(self, frame: InputDTMFFrame): + """Handle DTMF input frame.""" + is_first_digit = not self._aggregation + + digit_value = frame.button.value + self._aggregation += digit_value + + # For first digit, schedule interruption in separate task + if is_first_digit: + asyncio.create_task(self._send_interruption_task()) + + # Check for immediate flush conditions + if frame.button == self._termination_digit: + await self._flush_aggregation() + else: + # Signal digit received for timeout handling + self._digit_event.set() + + async def _send_interruption_task(self): + """Send interruption frame safely in a separate task.""" + try: + # Send the interruption frame + await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) + except Exception as e: + # Log error but don't propagate + print(f"Error sending interruption: {e}") + + def _create_aggregation_task(self) -> None: + """Creates the aggregation task if it hasn't been created yet.""" + if not self._aggregation_task: + self._aggregation_task = self.create_task(self._aggregation_task_handler()) + + async def _stop_aggregation_task(self) -> None: + """Stops the aggregation task.""" + if self._aggregation_task: + await self.cancel_task(self._aggregation_task) + self._aggregation_task = None + + async def _aggregation_task_handler(self): + """Background task that handles timeout-based flushing.""" + while True: + try: + await asyncio.wait_for(self._digit_event.wait(), timeout=self._idle_timeout) + self._digit_event.clear() + except asyncio.TimeoutError: + if self._aggregation: + await self._flush_aggregation() + + async def _flush_aggregation(self): + """Flush the current aggregation as a TranscriptionFrame.""" + if not self._aggregation: + return + + sequence = self._aggregation + transcription_text = f"{self._prefix}{sequence}" + + transcription_frame = TranscriptionFrame( + text=transcription_text, user_id="", timestamp=time_now_iso8601() + ) + await self.push_frame(transcription_frame) + + self._aggregation = "" + + async def cleanup(self) -> None: + """Clean up resources.""" + await super().cleanup() + await self._stop_aggregation_task() diff --git a/tests/test_dtmf_aggregator.py b/tests/test_dtmf_aggregator.py new file mode 100644 index 000000000..d2e1cc9aa --- /dev/null +++ b/tests/test_dtmf_aggregator.py @@ -0,0 +1,229 @@ +# +# Copyright (c) 2024-2025 Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import unittest + +from pipecat.frames.frames import ( + EndFrame, + InputDTMFFrame, + KeypadEntry, + TranscriptionFrame, +) +from pipecat.processors.aggregators.dtmf_aggregator import DTMFAggregator +from pipecat.tests.utils import SleepFrame, run_test + + +class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase): + async def test_basic_aggregation_with_pound(self): + """Test basic DTMF aggregation ending with pound key.""" + aggregator = DTMFAggregator() + frames_to_send = [ + InputDTMFFrame(button=KeypadEntry.ONE), + InputDTMFFrame(button=KeypadEntry.TWO), + InputDTMFFrame(button=KeypadEntry.THREE), + InputDTMFFrame(button=KeypadEntry.POUND), + ] + expected_down_frames = [ + InputDTMFFrame, + InputDTMFFrame, + InputDTMFFrame, + InputDTMFFrame, + TranscriptionFrame, + ] + + received_down_frames, _ = await run_test( + aggregator, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + # Find and verify the TranscriptionFrame + transcription_frames = [ + f for f in received_down_frames if isinstance(f, TranscriptionFrame) + ] + self.assertEqual(len(transcription_frames), 1) + self.assertEqual(transcription_frames[0].text, "DTMF: 123#") + + async def test_timeout_aggregation(self): + """Test DTMF aggregation with timeout flush.""" + aggregator = DTMFAggregator(timeout=0.1) + frames_to_send = [ + InputDTMFFrame(button=KeypadEntry.ONE), + InputDTMFFrame(button=KeypadEntry.TWO), + SleepFrame(sleep=0.2), # This should trigger timeout + InputDTMFFrame(button=KeypadEntry.THREE), + SleepFrame(sleep=0.2), # This should trigger another timeout + ] + expected_down_frames = [ + InputDTMFFrame, + InputDTMFFrame, + TranscriptionFrame, # First aggregation "12" + InputDTMFFrame, + TranscriptionFrame, # Second aggregation "3" + ] + + received_down_frames, _ = await run_test( + aggregator, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + # Find the TranscriptionFrames + transcription_frames = [ + f for f in received_down_frames if isinstance(f, TranscriptionFrame) + ] + self.assertEqual(len(transcription_frames), 2) + self.assertEqual(transcription_frames[0].text, "DTMF: 12") + self.assertEqual(transcription_frames[1].text, "DTMF: 3") + + async def test_multiple_aggregations(self): + """Test multiple DTMF sequences with pound termination.""" + aggregator = DTMFAggregator(timeout=0.2) + frames_to_send = [ + InputDTMFFrame(button=KeypadEntry.ONE), + InputDTMFFrame(button=KeypadEntry.TWO), + InputDTMFFrame(button=KeypadEntry.POUND), # First sequence + SleepFrame(sleep=0.1), + InputDTMFFrame(button=KeypadEntry.FOUR), + InputDTMFFrame(button=KeypadEntry.FIVE), + SleepFrame(sleep=0.3), # Second sequence via timeout + ] + expected_down_frames = [ + InputDTMFFrame, + InputDTMFFrame, + InputDTMFFrame, + TranscriptionFrame, # "12#" + InputDTMFFrame, + InputDTMFFrame, + TranscriptionFrame, # "45" + ] + + received_down_frames, _ = await run_test( + aggregator, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + transcription_frames = [ + f for f in received_down_frames if isinstance(f, TranscriptionFrame) + ] + self.assertEqual(len(transcription_frames), 2) + self.assertEqual(transcription_frames[0].text, "DTMF: 12#") + self.assertEqual(transcription_frames[1].text, "DTMF: 45") + + async def test_end_frame_flush(self): + """Test that EndFrame flushes pending aggregation.""" + aggregator = DTMFAggregator(timeout=1.0) + frames_to_send = [ + InputDTMFFrame(button=KeypadEntry.ONE), + InputDTMFFrame(button=KeypadEntry.TWO), + SleepFrame(sleep=0.1), # Allow time for aggregation + EndFrame(), + ] + expected_down_frames = [ + InputDTMFFrame, + InputDTMFFrame, + TranscriptionFrame, # Should flush before EndFrame + EndFrame, + ] + + received_down_frames, _ = await run_test( + aggregator, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + send_end_frame=False, # We're sending one in the test to test EndFrame logic + ) + + transcription_frames = [ + f for f in received_down_frames if isinstance(f, TranscriptionFrame) + ] + self.assertEqual(len(transcription_frames), 1) + self.assertEqual(transcription_frames[0].text, "DTMF: 12") + + async def test_custom_prefix(self): + """Test custom prefix configuration.""" + aggregator = DTMFAggregator(prefix="Menu: ") + frames_to_send = [ + InputDTMFFrame(button=KeypadEntry.ONE), + InputDTMFFrame(button=KeypadEntry.POUND), + ] + expected_down_frames = [ + InputDTMFFrame, + InputDTMFFrame, + TranscriptionFrame, + ] + + received_down_frames, _ = await run_test( + aggregator, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + transcription_frames = [ + f for f in received_down_frames if isinstance(f, TranscriptionFrame) + ] + self.assertEqual(len(transcription_frames), 1) + self.assertEqual(transcription_frames[0].text, "Menu: 1#") + + async def test_custom_termination_digit(self): + """Test custom termination digit configuration.""" + aggregator = DTMFAggregator(termination_digit=KeypadEntry.STAR) + frames_to_send = [ + InputDTMFFrame(button=KeypadEntry.ONE), + InputDTMFFrame(button=KeypadEntry.TWO), + InputDTMFFrame(button=KeypadEntry.STAR), # Custom terminator + ] + expected_down_frames = [ + InputDTMFFrame, + InputDTMFFrame, + InputDTMFFrame, + TranscriptionFrame, + ] + + received_down_frames, _ = await run_test( + aggregator, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + transcription_frames = [ + f for f in received_down_frames if isinstance(f, TranscriptionFrame) + ] + self.assertEqual(len(transcription_frames), 1) + self.assertEqual(transcription_frames[0].text, "DTMF: 12*") + + async def test_all_keypad_entries(self): + """Test all possible keypad entries.""" + aggregator = DTMFAggregator() + frames_to_send = [ + InputDTMFFrame(button=KeypadEntry.ZERO), + InputDTMFFrame(button=KeypadEntry.ONE), + InputDTMFFrame(button=KeypadEntry.TWO), + InputDTMFFrame(button=KeypadEntry.THREE), + InputDTMFFrame(button=KeypadEntry.FOUR), + InputDTMFFrame(button=KeypadEntry.FIVE), + InputDTMFFrame(button=KeypadEntry.SIX), + InputDTMFFrame(button=KeypadEntry.SEVEN), + InputDTMFFrame(button=KeypadEntry.EIGHT), + InputDTMFFrame(button=KeypadEntry.NINE), + InputDTMFFrame(button=KeypadEntry.STAR), + InputDTMFFrame(button=KeypadEntry.POUND), + ] + + # All the InputDTMFFrames plus one TranscriptionFrame + expected_down_frames = [InputDTMFFrame] * 12 + [TranscriptionFrame] + + received_down_frames, _ = await run_test( + aggregator, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + transcription_frames = [ + f for f in received_down_frames if isinstance(f, TranscriptionFrame) + ] + self.assertEqual(len(transcription_frames), 1) + self.assertEqual(transcription_frames[0].text, "DTMF: 0123456789*#")