From d161e2767f1b94f024acf3098576a70882e80fa0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 20 Aug 2025 07:27:42 -0700 Subject: [PATCH] FrameProcessor: allow pausing/resuming system frames --- CHANGELOG.md | 4 ++++ src/pipecat/processors/frame_processor.py | 26 +++++++++++++++++++++-- 2 files changed, 28 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index df75b8f7d..54494e043 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `FrameProcessor.pause_processing_system_frames()` and + `FrameProcessor.resume_processing_system_frames()`. These allow to pause and + resume the processing of system frame. + - Added new `on_process_frame()` observer method which makes it possible to know when a frame is being processed. diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 47612ed89..95b300fce 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -165,7 +165,7 @@ class FrameProcessor(BaseObject): watchdog_timeout_secs: Timeout in seconds for watchdog operations. **kwargs: Additional arguments passed to parent class. """ - super().__init__(name=name) + super().__init__(name=name, **kwargs) self._prev: Optional["FrameProcessor"] = None self._next: Optional["FrameProcessor"] = None @@ -222,6 +222,8 @@ class FrameProcessor(BaseObject): # The input task that handles all types of frames. It processes system # frames right away and queues non-system frames for later processing. + self.__should_block_system_frames = False + self.__input_event: Optional[asyncio.Event] = None self.__input_frame_task: Optional[asyncio.Task] = None # The process task processes non-system frames. Non-system frames will @@ -230,6 +232,7 @@ class FrameProcessor(BaseObject): # called. To resume processing frames we need to call # `resume_processing_frames()` which will wake up the event. self.__should_block_frames = False + self.__process_event: Optional[asyncio.Event] = None self.__process_frame_task: Optional[asyncio.Task] = None @property @@ -581,12 +584,23 @@ class FrameProcessor(BaseObject): logger.trace(f"{self}: pausing frame processing") self.__should_block_frames = True + async def pause_processing_system_frames(self): + """Pause processing of queued system frames.""" + logger.trace(f"{self}: pausing system frame processing") + self.__should_block_system_frames = True + async def resume_processing_frames(self): """Resume processing of queued frames.""" logger.trace(f"{self}: resuming frame processing") if self.__process_event: self.__process_event.set() + async def resume_processing_system_frames(self): + """Resume processing of queued system frames.""" + logger.trace(f"{self}: resuming system frame processing") + if self.__input_event: + self.__input_event.set() + async def process_frame(self, frame: Frame, direction: FrameDirection): """Process a frame. @@ -760,6 +774,7 @@ class FrameProcessor(BaseObject): return if not self.__input_frame_task: + self.__input_event = WatchdogEvent(self.task_manager) self.__input_queue = FrameProcessorQueue(self.task_manager) self.__input_frame_task = self.create_task(self.__input_frame_task_handler()) @@ -809,6 +824,13 @@ class FrameProcessor(BaseObject): """ while True: + if self.__should_block_system_frames and self.__input_event: + logger.trace(f"{self}: system frame processing paused") + await self.__input_event.wait() + self.__input_event.clear() + self.__should_block_system_frames = False + logger.trace(f"{self}: system frame processing resumed") + (frame, direction, callback) = await self.__input_queue.get() if isinstance(frame, SystemFrame): @@ -825,7 +847,7 @@ class FrameProcessor(BaseObject): async def __process_frame_task_handler(self): """Handle non-system frames from the process queue.""" while True: - if self.__should_block_frames: + if self.__should_block_frames and self.__process_event: logger.trace(f"{self}: frame processing paused") await self.__process_event.wait() self.__process_event.clear()