diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index ef24a9f5a..921e08c83 100644 --- a/src/pipecat/transports/base_transport.py +++ b/src/pipecat/transports/base_transport.py @@ -6,10 +6,8 @@ import asyncio import inspect -import types from abc import ABC, abstractmethod -from functools import partial from pydantic import ConfigDict from pydantic.main import BaseModel @@ -62,34 +60,22 @@ class BaseTransport(ABC): return decorator def _register_event_handler(self, event_name: str): - methods = inspect.getmembers(self, predicate=inspect.ismethod) - if event_name not in [method[0] for method in methods]: - raise Exception(f"Event handler {event_name} not found") - - self._event_handlers[event_name] = [getattr(self, event_name)] - - patch_method = types.MethodType(partial(self._patch_method, event_name), self) - setattr(self, event_name, patch_method) + if event_name in self._event_handlers: + raise Exception(f"Event handler {event_name} already registered") + self._event_handlers[event_name] = [] 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(types.MethodType(handler, self)) + self._event_handlers[event_name].append(handler) - def _patch_method(self, event_name, *args, **kwargs): + async def _call_event_handler(self, event_name: str, *args, **kwargs): try: for handler in self._event_handlers[event_name]: if inspect.iscoroutinefunction(handler): - # Beware, if handler() calls another event handler it - # will deadlock. You shouldn't do that anyways. - future = asyncio.run_coroutine_threadsafe( - handler(*args[1:], **kwargs), self._loop) - - # wait for the coroutine to finish. This will also - # raise any exceptions raised by the coroutine. - future.result() + await handler(self, *args, **kwargs) else: - handler(*args[1:], **kwargs) + handler(self, *args, **kwargs) except Exception as e: logger.error(f"Exception in event handler {event_name}: {e}") raise e diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index f6e034afb..24560ea2a 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -33,7 +33,7 @@ class WebsocketServerParams(TransportParams): class WebsocketServerCallbacks(BaseModel): - on_connection: Callable[[websockets.WebSocketServerProtocol], None] + on_connection: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]] class WebsocketServerInputTransport(FrameProcessor): @@ -87,7 +87,7 @@ class WebsocketServerInputTransport(FrameProcessor): self._websocket = websocket # Notify - self._callbacks.on_connection(websocket) + await self._callbacks.on_connection(websocket) # Handle incoming messages async for message in websocket: @@ -148,11 +148,9 @@ class WebsocketServerOutputTransport(FrameProcessor): self._audio_buffer = bytes() self._in_tts_audio = False - def set_client_connection(self, websocket: websockets.WebSocketServerProtocol): + async def set_client_connection(self, websocket: websockets.WebSocketServerProtocol): if self._websocket: - loop = self.get_event_loop() - future = asyncio.run_coroutine_threadsafe(self._websocket.close(), loop) - future.result() + await self._websocket.close() logger.warning("Only one client allowed, using new connection") self._websocket = websocket @@ -231,19 +229,9 @@ class WebsocketServerTransport(BaseTransport): self._output = WebsocketServerOutputTransport(self._params) return self._output - def _on_connection(self, websocket): + async def _on_connection(self, websocket): if self._output: - print("000 AAAAAAAAAAAAAAAAAAA") - self._output.set_client_connection(websocket) - print("111 AAAAAAAAAAAAAAAAAAA") - self.on_client_connected(websocket) - print("222 AAAAAAAAAAAAAAAAAAA") + await self._output.set_client_connection(websocket) + await self._call_event_handler("on_client_connected", websocket) else: logger.error("A WebsocketServerTransport output is missing in the pipeline") - - # - # Decorators (event handlers) - # - - def on_client_connected(self, client): - pass diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 04d24667c..99421f2ef 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -741,10 +741,10 @@ class DailyTransport(BaseTransport): participant_id, framerate, video_source, color_format) def _on_joined(self, participant): - self.on_joined(participant) + self._call_async_event_handler("on_joined", participant) def _on_left(self): - self.on_left() + self._call_async_event_handler("on_left") def _on_error(self, error): # TODO(aleix): Report error to input/output transports. The one managing @@ -754,10 +754,10 @@ class DailyTransport(BaseTransport): def _on_app_message(self, message: Any, sender: str): if self._input: self._input.push_app_message(message, sender) - self.on_app_message(message, sender) + self._call_async_event_handler("on_app_message", message, sender) def _on_call_state_updated(self, state: str): - self.on_call_state_updated(state) + self._call_async_event_handler("on_call_state_updated", state) async def _handle_dialin_ready(self, sip_endpoint: str): if not self._params.dialin_settings: @@ -793,28 +793,28 @@ class DailyTransport(BaseTransport): def _on_dialin_ready(self, sip_endpoint): if self._params.dialin_settings: asyncio.run_coroutine_threadsafe(self._handle_dialin_ready(sip_endpoint), self._loop) - self.on_dialin_ready(sip_endpoint) + self._call_async_event_handler("on_dialin_ready", sip_endpoint) def _on_dialout_connected(self, data): - self.on_dialout_connected(data) + self._call_async_event_handler("on_dialout_connected", data) def _on_dialout_stopped(self, data): - self.on_dialout_stopped(data) + self._call_async_event_handler("on_dialout_stopped", data) def _on_dialout_error(self, data): - self.on_dialout_error(data) + self._call_async_event_handler("on_dialout_error", data) def _on_dialout_warning(self, data): - self.on_dialout_warning(data) + self._call_async_event_handler("on_dialout_warning", data) def _on_participant_joined(self, participant): - self.on_participant_joined(participant) + self._call_async_event_handler("on_participant_joined", participant) def _on_participant_left(self, participant, reason): - self.on_participant_left(participant, reason) + self._call_async_event_handler("on_participant_left", participant, reason) def _on_first_participant_joined(self, participant): - self.on_first_participant_joined(participant) + self._call_async_event_handler("on_first_participant_joined", participant) def _on_transcription_message(self, participant_id, message): text = message["text"] @@ -829,42 +829,8 @@ class DailyTransport(BaseTransport): if self._input: self._input.push_transcription_frame(frame) - # - # Decorators (event handlers) - # - - def on_joined(self, participant): - pass - - def on_left(self): - pass - - def on_app_message(self, message, sender): - pass - - def on_call_state_updated(self, state): - pass - - def on_dialin_ready(self, sip_endpoint): - pass - - def on_dialout_connected(self, data): - pass - - def on_dialout_stopped(self, data): - pass - - def on_dialout_error(self, data): - pass - - def on_dialout_warning(self, data): - pass - - def on_first_participant_joined(self, participant): - pass - - def on_participant_joined(self, participant): - pass - - def on_participant_left(self, participant, reason): - pass + def _call_async_event_handler(self, event_name: str, *args, **kwargs): + future = asyncio.run_coroutine_threadsafe( + self._call_event_handler( + event_name, args, kwargs), self._loop) + future.result()