From bb29e50adb36c67efa1335959ff01c7295f844d0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 4 Mar 2025 23:03:27 -0800 Subject: [PATCH] introduce BaseObject class --- CHANGELOG.md | 46 ++++++++++-------- src/pipecat/audio/vad/vad_analyzer.py | 4 +- src/pipecat/pipeline/base_task.py | 17 ++----- src/pipecat/pipeline/runner.py | 10 ++-- src/pipecat/pipeline/task.py | 18 +------ src/pipecat/pipeline/task_observer.py | 16 +------ src/pipecat/processors/frame_processor.py | 40 ++-------------- src/pipecat/transports/base_transport.py | 54 +++------------------ src/pipecat/utils/base_object.py | 58 +++++++++++++++++++++++ 9 files changed, 103 insertions(+), 160 deletions(-) create mode 100644 src/pipecat/utils/base_object.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d5e4f862..4ff473ea2 100644 --- a/CHANGELOG.md +++ b/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. - ```python - weather_function = FunctionSchema( - name="get_current_weather", - description="Get the current weather", - properties={ - "location": { - "type": "string", - "description": "The city and state, e.g. San Francisco, CA", - }, - "format": { - "type": "string", - "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 new base class `BaseObject` which is now the base class of + `FrameProcessor`, `PipelineRunner`, `PipelineTask` and `BaseTransport`. The + new `BaseObject` adds supports for event handlers. + +- Added support for a unified format for specifying function calling across all + LLM services. + +```python + weather_function = FunctionSchema( + name="get_current_weather", + description="Get the current weather", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "format": { + "type": "string", + "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`. diff --git a/src/pipecat/audio/vad/vad_analyzer.py b/src/pipecat/audio/vad/vad_analyzer.py index b5d9b0ba2..3ca21a208 100644 --- a/src/pipecat/audio/vad/vad_analyzer.py +++ b/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 diff --git a/src/pipecat/pipeline/base_task.py b/src/pipecat/pipeline/base_task.py index b8622bac2..278709b7e 100644 --- a/src/pipecat/pipeline/base_task.py +++ b/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): - @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 - +class BaseTask(BaseObject): @abstractmethod def set_event_loop(self, loop: asyncio.AbstractEventLoop): """Sets the event loop that this task will run on.""" diff --git a/src/pipecat/pipeline/runner.py b/src/pipecat/pipeline/runner.py index acdbab562..3209fa92a 100644 --- a/src/pipecat/pipeline/runner.py +++ b/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() - self.name: str = name or f"{self.__class__.__name__}#{obj_count(self)}" + super().__init__(name=name) 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 diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 9e6b313a6..1c9d2dff9 100644 --- a/src/pipecat/pipeline/task.py +++ b/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() - self._name: str = f"{self.__class__.__name__}#{obj_count(self)}" - + super().__init__() 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 diff --git a/src/pipecat/pipeline/task_observer.py b/src/pipecat/pipeline/task_observer.py index 122038386..dd805032c 100644 --- a/src/pipecat/pipeline/task_observer.py +++ b/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 diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 335585ec3..6a1669ff1 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/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() - self._name = name or f"{self.__class__.__name__}#{obj_count(self)}" + super().__init__(name=name) 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 diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index c28579784..06f6cb920 100644 --- a/src/pipecat/transports/base_transport.py +++ b/src/pipecat/transports/base_transport.py @@ -4,18 +4,16 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import inspect -from abc import ABC, abstractmethod +from abc import 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() - self._name = name or f"{self.__class__.__name__}#{obj_count(self)}" + super().__init__(name=name) 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 - - 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 + pass diff --git a/src/pipecat/utils/base_object.py b/src/pipecat/utils/base_object.py new file mode 100644 index 000000000..e51eac35d --- /dev/null +++ b/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