From 87c12f3098c4176788a8090c1890bdf7dbd00bf0 Mon Sep 17 00:00:00 2001 From: Om Chauhan Date: Thu, 22 Jan 2026 08:39:31 +0530 Subject: [PATCH] changed frame filter storage type from tuples to sets --- changelog/3510.changed.3.md | 1 + src/pipecat/pipeline/task.py | 30 +++++++++++------------------- 2 files changed, 12 insertions(+), 19 deletions(-) create mode 100644 changelog/3510.changed.3.md 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 20b4adf36..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") @@ -434,7 +434,7 @@ class PipelineTask(BasePipelineTask): Returns: Tuple of frame types that trigger the on_frame_reached_upstream event. """ - return self._reached_upstream_types + return tuple(self._reached_upstream_types) @property def reached_downstream_types(self) -> Tuple[Type[Frame], ...]: @@ -443,7 +443,7 @@ class PipelineTask(BasePipelineTask): Returns: Tuple of frame types that trigger the on_frame_reached_downstream event. """ - return self._reached_downstream_types + return tuple(self._reached_downstream_types) def event_handler(self, event_name: str): """Decorator for registering event handlers. @@ -488,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. @@ -496,7 +496,7 @@ 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. @@ -504,11 +504,7 @@ class PipelineTask(BasePipelineTask): 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) + 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. @@ -516,11 +512,7 @@ class PipelineTask(BasePipelineTask): 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) + self._reached_downstream_types.update(types) def has_finished(self) -> bool: """Check if the pipeline task has finished execution. @@ -820,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): @@ -859,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):