Add DTMFAggregator

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).
@@ -45,8 +50,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [0.0.68] - 2025-05-28
 
-### Added
-
 - Added `GoogleHttpTTSService` which uses Google's HTTP TTS API.
 
 - Added `TavusTransport`, a new transport implementation compatible with any

src/pipecat/processors/aggregators/dtmf_aggregator.py

@@ -0,0 +1,128 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+from typing import Optional
+
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    Frame,
+    InputDTMFFrame,
+    KeypadEntry,
+    StartInterruptionFrame,
+    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: '#')
+    - Interruption occurs
+
+    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 = None
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, InputDTMFFrame):
+            await self._handle_dtmf_frame(frame, direction)
+        elif isinstance(frame, StartInterruptionFrame):
+            # Flush on interruption
+            if self._aggregation:
+                await self._flush_aggregation(direction)
+        elif isinstance(frame, (EndFrame, CancelFrame)):
+            # Flush any pending aggregation
+            if self._aggregation:
+                await self._flush_aggregation(direction)
+            await self._stop_aggregation_task()
+
+        # Push frame
+        await self.push_frame(frame, direction)
+
+    async def _handle_dtmf_frame(self, frame: InputDTMFFrame, direction: FrameDirection):
+        # Start aggregation task if not running
+        if self._aggregation_task is None:
+            self._aggregation_task = self.create_task(self._aggregation_task_handler(direction))
+
+        # Add digit to aggregation
+        digit_value = frame.button.value
+        self._aggregation += digit_value
+
+        # Check for immediate flush conditions
+        if frame.button == self._termination_digit:
+            await self._flush_aggregation(direction)
+        else:
+            # Signal new digit received
+            self._digit_event.set()
+
+    async def _aggregation_task_handler(self, direction: FrameDirection):
+        """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(direction)
+
+    async def _flush_aggregation(self, direction: FrameDirection):
+        """Flush the current aggregation as a TranscriptionFrame."""
+        if not self._aggregation:
+            return
+
+        sequence = self._aggregation
+
+        # Create transcription with prefix for LLM context
+        transcription_text = f"{self._prefix}{sequence}"
+
+        # Create and push transcription frame
+        transcription_frame = TranscriptionFrame(
+            text=transcription_text, user_id="", timestamp=time_now_iso8601()
+        )
+
+        await self.push_frame(transcription_frame, direction)
+
+        # Reset aggregation
+        self._aggregation = ""
+
+    async def _stop_aggregation_task(self):
+        if self._aggregation_task:
+            await self.cancel_task(self._aggregation_task)
+            self._aggregation_task = None
+
+    async def cleanup(self) -> None:
+        await super().cleanup()
+        await self._stop_aggregation_task()