Merge pull request #1889 from pipecat-ai/mb/add-dtmf-aggregator
Add DTMFAggregator
This commit is contained in:
Mark Backman
@@ -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).
|
||||
|
||||
143
src/pipecat/processors/aggregators/dtmf_aggregator.py
Normal file
143
src/pipecat/processors/aggregators/dtmf_aggregator.py
Normal file
@@ -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()
|
||||
229
tests/test_dtmf_aggregator.py
Normal file
229
tests/test_dtmf_aggregator.py
Normal file
@@ -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*#")
|
||||
Reference in New Issue
Block a user