diff --git a/changelog/3510.added.2.md b/changelog/3510.added.2.md new file mode 100644 index 000000000..e03b81ee3 --- /dev/null +++ b/changelog/3510.added.2.md @@ -0,0 +1 @@ +- Added `add_reached_upstream_filter()` and `add_reached_downstream_filter()` methods to `PipelineTask` for appending frame types. \ No newline at end of file diff --git a/changelog/3510.added.md b/changelog/3510.added.md new file mode 100644 index 000000000..93d9ef179 --- /dev/null +++ b/changelog/3510.added.md @@ -0,0 +1 @@ +- Added `reached_upstream_types` and `reached_downstream_types` read-only properties to `PipelineTask` for inspecting current frame filters. \ No newline at end of file diff --git a/changelog/3510.changed.3.md b/changelog/3510.changed.3.md new file mode 100644 index 000000000..ac299f143 --- /dev/null +++ b/changelog/3510.changed.3.md @@ -0,0 +1 @@ +- Changed frame filter storage from tuples to sets in `PipelineTask`. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 9a2f1d33d..60df77ff9 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -15,7 +15,7 @@ import asyncio import importlib.util import os from pathlib import Path -from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Tuple, Type +from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Set, Tuple, Type from loguru import logger from pydantic import BaseModel, ConfigDict, Field @@ -366,8 +366,8 @@ class PipelineTask(BasePipelineTask): # in. This is mainly for efficiency reason because each event handler # creates a task and most likely you only care about one or two frame # types. - self._reached_upstream_types: Tuple[Type[Frame], ...] = () - self._reached_downstream_types: Tuple[Type[Frame], ...] = () + self._reached_upstream_types: Set[Type[Frame]] = set() + self._reached_downstream_types: Set[Type[Frame]] = set() self._register_event_handler("on_frame_reached_upstream") self._register_event_handler("on_frame_reached_downstream") self._register_event_handler("on_idle_timeout") @@ -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 tuple(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 tuple(self._reached_downstream_types) + def event_handler(self, event_name: str): """Decorator for registering event handlers. @@ -470,7 +488,7 @@ class PipelineTask(BasePipelineTask): Args: types: Tuple of frame types to monitor for upstream events. """ - self._reached_upstream_types = types + self._reached_upstream_types = set(types) def set_reached_downstream_filter(self, types: Tuple[Type[Frame], ...]): """Set which frame types trigger the on_frame_reached_downstream event. @@ -478,7 +496,23 @@ class PipelineTask(BasePipelineTask): Args: types: Tuple of frame types to monitor for downstream events. """ - self._reached_downstream_types = types + self._reached_downstream_types = set(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. + """ + self._reached_upstream_types.update(types) + + 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. + """ + self._reached_downstream_types.update(types) def has_finished(self) -> bool: """Check if the pipeline task has finished execution. @@ -778,7 +812,7 @@ class PipelineTask(BasePipelineTask): pipeline to be stopped (e.g. EndTaskFrame) in which case we would send an EndFrame down the pipeline. """ - if isinstance(frame, self._reached_upstream_types): + if isinstance(frame, tuple(self._reached_upstream_types)): await self._call_event_handler("on_frame_reached_upstream", frame) if isinstance(frame, EndTaskFrame): @@ -817,7 +851,7 @@ class PipelineTask(BasePipelineTask): processors have handled the EndFrame and therefore we can exit the task cleanly. """ - if isinstance(frame, self._reached_downstream_types): + if isinstance(frame, tuple(self._reached_downstream_types)): await self._call_event_handler("on_frame_reached_downstream", frame) if isinstance(frame, StartFrame):