From 1377dec01b47c911c9deea5d0a95763bd529ce48 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 20 Aug 2025 14:32:12 -0700 Subject: [PATCH 1/2] DTMFAggregator: no need for interruption task Now that system frames are queued there's no need to have an additional task to push a `BotInterruptionFrame`. --- .../processors/aggregators/dtmf_aggregator.py | 33 ++++++++----------- 1 file changed, 13 insertions(+), 20 deletions(-) diff --git a/src/pipecat/processors/aggregators/dtmf_aggregator.py b/src/pipecat/processors/aggregators/dtmf_aggregator.py index e55ae66bc..3676824fd 100644 --- a/src/pipecat/processors/aggregators/dtmf_aggregator.py +++ b/src/pipecat/processors/aggregators/dtmf_aggregator.py @@ -24,8 +24,9 @@ from pipecat.frames.frames import ( StartFrame, TranscriptionFrame, ) -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup from pipecat.utils.asyncio.timeout import wait_for +from pipecat.utils.asyncio.watchdog_event import WatchdogEvent from pipecat.utils.time import time_now_iso8601 @@ -63,9 +64,17 @@ class DTMFAggregator(FrameProcessor): self._termination_digit = termination_digit self._prefix = prefix - self._digit_event = asyncio.Event() self._aggregation_task: Optional[asyncio.Task] = None - self._interruption_task: Optional[asyncio.Task] = None + + async def setup(self, setup: FrameProcessorSetup): + """Setup resources.""" + await super().setup(setup) + self._digit_event = WatchdogEvent(setup.task_manager) + + async def cleanup(self) -> None: + """Clean up resources.""" + await super().cleanup() + await self._stop_aggregation_task() async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: """Process incoming frames and handle DTMF aggregation. @@ -83,7 +92,6 @@ class DTMFAggregator(FrameProcessor): if self._aggregation: await self._flush_aggregation() await self._stop_aggregation_task() - await self._stop_interruption_task() await self.push_frame(frame, direction) elif isinstance(frame, InputDTMFFrame): # Push the DTMF frame downstream first @@ -103,7 +111,7 @@ class DTMFAggregator(FrameProcessor): # For first digit, schedule interruption in separate task if is_first_digit: - self._interruption_task = self.create_task(self._send_interruption_task()) + await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) # Check for immediate flush conditions if frame.button == self._termination_digit: @@ -112,16 +120,6 @@ class DTMFAggregator(FrameProcessor): # Signal digit received for timeout handling self._digit_event.set() - async def _send_interruption_task(self): - """Send interruption frame safely in a separate task.""" - await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) - - async def _stop_interruption_task(self) -> None: - """Stops the interruption task.""" - if self._interruption_task: - await self.cancel_task(self._interruption_task) - self._interruption_task = None - def _create_aggregation_task(self) -> None: """Creates the aggregation task if it hasn't been created yet.""" if not self._aggregation_task: @@ -158,8 +156,3 @@ class DTMFAggregator(FrameProcessor): await self.push_frame(transcription_frame) self._aggregation = "" - - async def cleanup(self) -> None: - """Clean up resources.""" - await super().cleanup() - await self._stop_aggregation_task() From 4ff0567025027b6c03de27c9801fdfb38d17421a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 20 Aug 2025 15:03:21 -0700 Subject: [PATCH 2/2] BaseObject: allow keyword arguments --- src/pipecat/utils/base_object.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/pipecat/utils/base_object.py b/src/pipecat/utils/base_object.py index 51eb1195b..bf34515a6 100644 --- a/src/pipecat/utils/base_object.py +++ b/src/pipecat/utils/base_object.py @@ -29,12 +29,13 @@ class BaseObject(ABC): classes in the framework should inherit from this base class. """ - def __init__(self, *, name: Optional[str] = None): + def __init__(self, *, name: Optional[str] = None, **kwargs): """Initialize the base object. Args: name: Optional custom name for the object. If not provided, generates a name using the class name and instance count. + **kwargs: Additional arguments passed to parent class. """ self._id: int = obj_id() self._name = name or f"{self.__class__.__name__}#{obj_count(self)}"