Merge pull request #1332 from pipecat-ai/aleix/base-object-and-event-handlers

introduce BaseObject class

CHANGELOG.md

@@ -9,26 +9,32 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- Added support for a unified format for specifying function calling across all LLM services.
+- Added new base class `BaseObject` which is now the base class of
-  ```python
+  `FrameProcessor`, `PipelineRunner`, `PipelineTask` and `BaseTransport`. The
-    weather_function = FunctionSchema(
+  new `BaseObject` adds supports for event handlers.
-        name="get_current_weather",
+
-        description="Get the current weather",
+- Added support for a unified format for specifying function calling across all
-        properties={
+  LLM services.
-            "location": {
+
-                "type": "string",
+```python
-                "description": "The city and state, e.g. San Francisco, CA",
+  weather_function = FunctionSchema(
-            },
+      name="get_current_weather",
-            "format": {
+      description="Get the current weather",
-                "type": "string",
+      properties={
-                "enum": ["celsius", "fahrenheit"],
+          "location": {
-                "description": "The temperature unit to use. Infer this from the user's location.",
+              "type": "string",
-            },
+              "description": "The city and state, e.g. San Francisco, CA",
-        },
+          },
-        required=["location"],
+          "format": {
-    )
+              "type": "string",
-    tools = ToolsSchema(standard_tools=[weather_function])
+              "enum": ["celsius", "fahrenheit"],
-  ```
+              "description": "The temperature unit to use. Infer this from the user's location.",
+          },
+      },
+      required=["location"],
+  )
+  tools = ToolsSchema(standard_tools=[weather_function])
+```
 
 - Added `speech_threshold` parameter to `GladiaSTTService`.
 

src/pipecat/audio/vad/vad_analyzer.py

@@ -4,7 +4,7 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-from abc import abstractmethod
+from abc import ABC, abstractmethod
 from enum import Enum
 from typing import Optional
 
@@ -33,7 +33,7 @@ class VADParams(BaseModel):
     min_volume: float = VAD_MIN_VOLUME
 
 
-class VADAnalyzer:
+class VADAnalyzer(ABC):
     def __init__(self, *, sample_rate: Optional[int] = None, params: VADParams):
         self._init_sample_rate = sample_rate
         self._sample_rate = 0

src/pipecat/pipeline/base_task.py

@@ -5,25 +5,14 @@
 #
 
 import asyncio
-from abc import ABC, abstractmethod
+from abc import abstractmethod
 from typing import AsyncIterable, Iterable
 
 from pipecat.frames.frames import Frame
+from pipecat.utils.base_object import BaseObject
 
 
-class BaseTask(ABC):
+class BaseTask(BaseObject):
-    @property
-    @abstractmethod
-    def id(self) -> int:
-        """Returns the unique indetifier for this task."""
-        pass
-
-    @property
-    @abstractmethod
-    def name(self) -> str:
-        """Returns the name of this task."""
-        pass
-
     @abstractmethod
     def set_event_loop(self, loop: asyncio.AbstractEventLoop):
         """Sets the event loop that this task will run on."""

src/pipecat/pipeline/runner.py

@@ -12,10 +12,10 @@ from typing import Optional
 from loguru import logger
 
 from pipecat.pipeline.task import PipelineTask
-from pipecat.utils.utils import obj_count, obj_id
+from pipecat.utils.base_object import BaseObject
 
 
-class PipelineRunner:
+class PipelineRunner(BaseObject):
     def __init__(
         self,
         *,
@@ -24,8 +24,7 @@ class PipelineRunner:
         force_gc: bool = False,
         loop: Optional[asyncio.AbstractEventLoop] = None,
     ):
-        self.id: int = obj_id()
+        super().__init__(name=name)
-        self.name: str = name or f"{self.__class__.__name__}#{obj_count(self)}"
 
         self._tasks = {}
         self._sig_task = None
@@ -74,6 +73,3 @@ class PipelineRunner:
         collected = gc.collect()
         logger.debug(f"Garbage collector: collected {collected} objects.")
         logger.debug(f"Garbage collector: uncollectable objects {gc.garbage}")
-
-    def __str__(self):
-        return self.name

src/pipecat/pipeline/task.py

@@ -32,7 +32,6 @@ from pipecat.pipeline.base_task import BaseTask
 from pipecat.pipeline.task_observer import TaskObserver
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.utils.asyncio import BaseTaskManager, TaskManager
-from pipecat.utils.utils import obj_count, obj_id
 
 HEARTBEAT_SECONDS = 1.0
 HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 5
@@ -138,9 +137,7 @@ class PipelineTask(BaseTask):
         task_manager: Optional[BaseTaskManager] = None,
         check_dangling_tasks: bool = True,
     ):
-        self._id: int = obj_id()
+        super().__init__()
-        self._name: str = f"{self.__class__.__name__}#{obj_count(self)}"
-
         self._pipeline = pipeline
         self._clock = clock
         self._params = params
@@ -180,16 +177,6 @@ class PipelineTask(BaseTask):
 
         self._observer = TaskObserver(observers=observers, task_manager=self._task_manager)
 
-    @property
-    def id(self) -> int:
-        """Returns the unique indetifier for this task."""
-        return self._id
-
-    @property
-    def name(self) -> str:
-        """Returns the name of this task."""
-        return self._name
-
     @property
     def params(self) -> PipelineParams:
         """Returns the pipeline parameters of this task."""
@@ -434,6 +421,3 @@ class PipelineTask(BaseTask):
         tasks = [t.get_name() for t in self._task_manager.current_tasks()]
         if tasks:
             logger.warning(f"Dangling tasks detected: {tasks}")
-
-    def __str__(self):
-        return self.name

src/pipecat/pipeline/task_observer.py

@@ -13,7 +13,6 @@ from pipecat.frames.frames import Frame
 from pipecat.observers.base_observer import BaseObserver
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.utils.asyncio import BaseTaskManager
-from pipecat.utils.utils import obj_count, obj_id
 
 
 @dataclass
@@ -56,20 +55,10 @@ class TaskObserver(BaseObserver):
     """
 
     def __init__(self, *, observers: List[BaseObserver] = [], task_manager: BaseTaskManager):
-        self._id: int = obj_id()
-        self._name: str = f"{self.__class__.__name__}#{obj_count(self)}"
         self._observers = observers
         self._task_manager = task_manager
         self._proxies: List[Proxy] = []
 
-    @property
-    def id(self) -> int:
-        return self._id
-
-    @property
-    def name(self) -> str:
-        return self._name
-
     async def start(self):
         """Starts all proxy observer tasks."""
         self._proxies = self._create_proxies(self._observers)
@@ -100,7 +89,7 @@ class TaskObserver(BaseObserver):
             queue = asyncio.Queue()
             task = self._task_manager.create_task(
                 self._proxy_task_handler(queue, observer),
-                f"{self}::{observer.__class__.__name__}::_proxy_task_handler",
+                f"TaskObserver::{observer.__class__.__name__}::_proxy_task_handler",
             )
             proxy = Proxy(queue=queue, task=task, observer=observer)
             proxies.append(proxy)
@@ -112,6 +101,3 @@ class TaskObserver(BaseObserver):
             await observer.on_push_frame(
                 data.src, data.dst, data.frame, data.direction, data.timestamp
             )
-
-    def __str__(self):
-        return self.name

src/pipecat/processors/frame_processor.py

@@ -5,7 +5,6 @@
 #
 
 import asyncio
-import inspect
 from enum import Enum
 from typing import Awaitable, Callable, Coroutine, Optional
 
@@ -24,7 +23,7 @@ from pipecat.frames.frames import (
 from pipecat.metrics.metrics import LLMTokenUsage, MetricsData
 from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics
 from pipecat.utils.asyncio import BaseTaskManager
-from pipecat.utils.utils import obj_count, obj_id
+from pipecat.utils.base_object import BaseObject
 
 
 class FrameDirection(Enum):
@@ -32,7 +31,7 @@ class FrameDirection(Enum):
     UPSTREAM = 2
 
 
-class FrameProcessor:
+class FrameProcessor(BaseObject):
     def __init__(
         self,
         *,
@@ -40,14 +39,11 @@ class FrameProcessor:
         metrics: Optional[FrameProcessorMetrics] = None,
         **kwargs,
     ):
-        self._id: int = obj_id()
+        super().__init__(name=name)
-        self._name = name or f"{self.__class__.__name__}#{obj_count(self)}"
         self._parent: Optional["FrameProcessor"] = None
         self._prev: Optional["FrameProcessor"] = None
         self._next: Optional["FrameProcessor"] = None
 
-        self._event_handlers: dict = {}
-
         # Clock
         self._clock: Optional[BaseClock] = None
 
@@ -254,23 +250,6 @@ class FrameProcessor:
         else:
             await self.__push_queue.put((frame, direction))
 
-    def event_handler(self, event_name: str):
-        def decorator(handler):
-            self.add_event_handler(event_name, handler)
-            return handler
-
-        return decorator
-
-    def add_event_handler(self, event_name: str, handler):
-        if event_name not in self._event_handlers:
-            raise Exception(f"Event handler {event_name} not registered")
-        self._event_handlers[event_name].append(handler)
-
-    def _register_event_handler(self, event_name: str):
-        if event_name in self._event_handlers:
-            raise Exception(f"Event handler {event_name} already registered")
-        self._event_handlers[event_name] = []
-
     async def __start(self, frame: StartFrame):
         self.__create_input_task()
         self.__create_push_task()
@@ -385,16 +364,3 @@ class FrameProcessor:
             (frame, direction) = await self.__push_queue.get()
             await self.__internal_push_frame(frame, direction)
             self.__push_queue.task_done()
-
-    async def _call_event_handler(self, event_name: str, *args, **kwargs):
-        try:
-            for handler in self._event_handlers[event_name]:
-                if inspect.iscoroutinefunction(handler):
-                    await handler(self, *args, **kwargs)
-                else:
-                    handler(self, *args, **kwargs)
-        except Exception as e:
-            logger.exception(f"Exception in event handler {event_name}: {e}")
-
-    def __str__(self):
-        return self.name

src/pipecat/transports/base_transport.py

@@ -4,18 +4,16 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import inspect
+from abc import abstractmethod
-from abc import ABC, abstractmethod
 from typing import Optional
 
-from loguru import logger
 from pydantic import BaseModel, ConfigDict
 
 from pipecat.audio.filters.base_audio_filter import BaseAudioFilter
 from pipecat.audio.mixers.base_audio_mixer import BaseAudioMixer
 from pipecat.audio.vad.vad_analyzer import VADAnalyzer
 from pipecat.processors.frame_processor import FrameProcessor
-from pipecat.utils.utils import obj_count, obj_id
+from pipecat.utils.base_object import BaseObject
 
 
 class TransportParams(BaseModel):
@@ -43,7 +41,7 @@ class TransportParams(BaseModel):
     vad_analyzer: Optional[VADAnalyzer] = None
 
 
-class BaseTransport(ABC):
+class BaseTransport(BaseObject):
     def __init__(
         self,
         *,
@@ -51,54 +49,14 @@ class BaseTransport(ABC):
         input_name: Optional[str] = None,
         output_name: Optional[str] = None,
     ):
-        self._id: int = obj_id()
+        super().__init__(name=name)
-        self._name = name or f"{self.__class__.__name__}#{obj_count(self)}"
         self._input_name = input_name
         self._output_name = output_name
-        self._event_handlers: dict = {}
-
-    @property
-    def id(self) -> int:
-        return self._id
-
-    @property
-    def name(self) -> str:
-        return self._name
 
     @abstractmethod
     def input(self) -> FrameProcessor:
-        raise NotImplementedError
+        pass
 
     @abstractmethod
     def output(self) -> FrameProcessor:
-        raise NotImplementedError
+        pass
-
-    def event_handler(self, event_name: str):
-        def decorator(handler):
-            self.add_event_handler(event_name, handler)
-            return handler
-
-        return decorator
-
-    def add_event_handler(self, event_name: str, handler):
-        if event_name not in self._event_handlers:
-            raise Exception(f"Event handler {event_name} not registered")
-        self._event_handlers[event_name].append(handler)
-
-    def _register_event_handler(self, event_name: str):
-        if event_name in self._event_handlers:
-            raise Exception(f"Event handler {event_name} already registered")
-        self._event_handlers[event_name] = []
-
-    async def _call_event_handler(self, event_name: str, *args, **kwargs):
-        try:
-            for handler in self._event_handlers[event_name]:
-                if inspect.iscoroutinefunction(handler):
-                    await handler(self, *args, **kwargs)
-                else:
-                    handler(self, *args, **kwargs)
-        except Exception as e:
-            logger.exception(f"Exception in event handler {event_name}: {e}")
-
-    def __str__(self):
-        return self.name

src/pipecat/utils/base_object.py

@@ -0,0 +1,58 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import inspect
+from abc import ABC
+from typing import Optional
+
+from loguru import logger
+
+from pipecat.utils.utils import obj_count, obj_id
+
+
+class BaseObject(ABC):
+    def __init__(self, *, name: Optional[str] = None):
+        self._id: int = obj_id()
+        self._name = name or f"{self.__class__.__name__}#{obj_count(self)}"
+        self._event_handlers: dict = {}
+
+    @property
+    def id(self) -> int:
+        return self._id
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    def event_handler(self, event_name: str):
+        def decorator(handler):
+            self.add_event_handler(event_name, handler)
+            return handler
+
+        return decorator
+
+    def add_event_handler(self, event_name: str, handler):
+        if event_name not in self._event_handlers:
+            raise Exception(f"Event handler {event_name} not registered")
+        self._event_handlers[event_name].append(handler)
+
+    def _register_event_handler(self, event_name: str):
+        if event_name in self._event_handlers:
+            raise Exception(f"Event handler {event_name} already registered")
+        self._event_handlers[event_name] = []
+
+    async def _call_event_handler(self, event_name: str, *args, **kwargs):
+        try:
+            for handler in self._event_handlers[event_name]:
+                if inspect.iscoroutinefunction(handler):
+                    await handler(self, *args, **kwargs)
+                else:
+                    handler(self, *args, **kwargs)
+        except Exception as e:
+            logger.exception(f"Exception in event handler {event_name}: {e}")
+
+    def __str__(self):
+        return self.name