From 6e44a2ab493371e6c28ca823eb118abf3696bcac Mon Sep 17 00:00:00 2001 From: Om Chauhan Date: Wed, 21 Jan 2026 08:14:48 +0530 Subject: [PATCH] feat(task): add additive filter methods for frame monitoring --- src/pipecat/pipeline/task.py | 42 ++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 9a2f1d33d..20b4adf36 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -427,6 +427,24 @@ class PipelineTask(BasePipelineTask): raise Exception(f"{self} RTVI is not enabled.") return self._rtvi + @property + def reached_upstream_types(self) -> Tuple[Type[Frame], ...]: + """Get the currently configured upstream frame type filters. + + Returns: + Tuple of frame types that trigger the on_frame_reached_upstream event. + """ + return self._reached_upstream_types + + @property + def reached_downstream_types(self) -> Tuple[Type[Frame], ...]: + """Get the currently configured downstream frame type filters. + + Returns: + Tuple of frame types that trigger the on_frame_reached_downstream event. + """ + return self._reached_downstream_types + def event_handler(self, event_name: str): """Decorator for registering event handlers. @@ -480,6 +498,30 @@ class PipelineTask(BasePipelineTask): """ self._reached_downstream_types = types + def add_reached_upstream_filter(self, types: Tuple[Type[Frame], ...]): + """Add frame types to trigger the on_frame_reached_upstream event. + + Args: + types: Tuple of frame types to add to upstream monitoring. + """ + if not types: + return + current = set(self._reached_upstream_types) + current.update(types) + self._reached_upstream_types = tuple(current) + + def add_reached_downstream_filter(self, types: Tuple[Type[Frame], ...]): + """Add frame types to trigger the on_frame_reached_downstream event. + + Args: + types: Tuple of frame types to add to downstream monitoring. + """ + if not types: + return + current = set(self._reached_downstream_types) + current.update(types) + self._reached_downstream_types = tuple(current) + def has_finished(self) -> bool: """Check if the pipeline task has finished execution.