wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

transports: simplify and fix async and nested decorators

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-05-29 12:47:09 -07:00
parent 4099256c18
commit 8f7b3e1d5d
3 changed files with 31 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
src/pipecat/transports/base_transport.py
View File

@@ -6,10 +6,8 @@
import asyncio import asyncio
import inspect import inspect
import types
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from functools import partial
from pydantic import ConfigDict from pydantic import ConfigDict
from pydantic.main import BaseModel from pydantic.main import BaseModel
@@ -62,34 +60,22 @@ class BaseTransport(ABC):
return decorator return decorator
def _register_event_handler(self, event_name: str): def _register_event_handler(self, event_name: str):
methods = inspect.getmembers(self, predicate=inspect.ismethod) if event_name in self._event_handlers:
if event_name not in [method[0] for method in methods]: raise Exception(f"Event handler {event_name} already registered")
raise Exception(f"Event handler {event_name} not found") self._event_handlers[event_name] = []
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)
def _add_event_handler(self, event_name: str, handler): def _add_event_handler(self, event_name: str, handler):
if event_name not in self._event_handlers: if event_name not in self._event_handlers:
raise Exception(f"Event handler {event_name} not registered") 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: try:
for handler in self._event_handlers[event_name]: for handler in self._event_handlers[event_name]:
if inspect.iscoroutinefunction(handler): if inspect.iscoroutinefunction(handler):
# Beware, if handler() calls another event handler it await handler(self, *args, **kwargs)
# 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()
else: else:
handler(*args[1:], **kwargs) handler(self, *args, **kwargs)
except Exception as e: except Exception as e:
logger.error(f"Exception in event handler {event_name}: {e}") logger.error(f"Exception in event handler {event_name}: {e}")
raise e raise e

26
src/pipecat/transports/network/websocket_server.py
View File

@@ -33,7 +33,7 @@ class WebsocketServerParams(TransportParams):
class WebsocketServerCallbacks(BaseModel): class WebsocketServerCallbacks(BaseModel):
on_connection: Callable[[websockets.WebSocketServerProtocol], None] on_connection: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]]
class WebsocketServerInputTransport(FrameProcessor): class WebsocketServerInputTransport(FrameProcessor):
@@ -87,7 +87,7 @@ class WebsocketServerInputTransport(FrameProcessor):
self._websocket = websocket self._websocket = websocket
# Notify # Notify
self._callbacks.on_connection(websocket) await self._callbacks.on_connection(websocket)
# Handle incoming messages # Handle incoming messages
async for message in websocket: async for message in websocket:
@@ -148,11 +148,9 @@ class WebsocketServerOutputTransport(FrameProcessor):
self._audio_buffer = bytes() self._audio_buffer = bytes()
self._in_tts_audio = False 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: if self._websocket:
loop = self.get_event_loop() await self._websocket.close()
future = asyncio.run_coroutine_threadsafe(self._websocket.close(), loop)
future.result()
logger.warning("Only one client allowed, using new connection") logger.warning("Only one client allowed, using new connection")
self._websocket = websocket self._websocket = websocket
@@ -231,19 +229,9 @@ class WebsocketServerTransport(BaseTransport):
self._output = WebsocketServerOutputTransport(self._params) self._output = WebsocketServerOutputTransport(self._params)
return self._output return self._output
def _on_connection(self, websocket): async def _on_connection(self, websocket):
if self._output: if self._output:
print("000 AAAAAAAAAAAAAAAAAAA") await self._output.set_client_connection(websocket)
self._output.set_client_connection(websocket) await self._call_event_handler("on_client_connected", websocket)
print("111 AAAAAAAAAAAAAAAAAAA")
self.on_client_connected(websocket)
print("222 AAAAAAAAAAAAAAAAAAA")
else: else:
logger.error("A WebsocketServerTransport output is missing in the pipeline") logger.error("A WebsocketServerTransport output is missing in the pipeline")
#
# Decorators (event handlers)
#
def on_client_connected(self, client):
pass

68
src/pipecat/transports/services/daily.py
View File

@@ -741,10 +741,10 @@ class DailyTransport(BaseTransport):
participant_id, framerate, video_source, color_format) participant_id, framerate, video_source, color_format)
def _on_joined(self, participant): def _on_joined(self, participant):
self.on_joined(participant) self._call_async_event_handler("on_joined", participant)
def _on_left(self): def _on_left(self):
self.on_left() self._call_async_event_handler("on_left")
def _on_error(self, error): def _on_error(self, error):
# TODO(aleix): Report error to input/output transports. The one managing # 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): def _on_app_message(self, message: Any, sender: str):
if self._input: if self._input:
self._input.push_app_message(message, sender) 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): 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): async def _handle_dialin_ready(self, sip_endpoint: str):
if not self._params.dialin_settings: if not self._params.dialin_settings:
@@ -793,28 +793,28 @@ class DailyTransport(BaseTransport):
def _on_dialin_ready(self, sip_endpoint): def _on_dialin_ready(self, sip_endpoint):
if self._params.dialin_settings: if self._params.dialin_settings:
asyncio.run_coroutine_threadsafe(self._handle_dialin_ready(sip_endpoint), self._loop) 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): 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): 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): 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): 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): 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): 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): 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): def _on_transcription_message(self, participant_id, message):
text = message["text"] text = message["text"]
@@ -829,42 +829,8 @@ class DailyTransport(BaseTransport):
if self._input: if self._input:
self._input.push_transcription_frame(frame) self._input.push_transcription_frame(frame)
# def _call_async_event_handler(self, event_name: str, *args, **kwargs):
# Decorators (event handlers) future = asyncio.run_coroutine_threadsafe(
# self._call_event_handler(
event_name, args, kwargs), self._loop)
def on_joined(self, participant): future.result()
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