FrameProcessor: allow pausing/resuming system frames

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.
 

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()