From 9e22a8b4ffe49530345db929589d0997dc9a5577 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 15 May 2024 15:57:19 -0700 Subject: [PATCH] transports(daily): add receiving transport messages --- CHANGELOG.md | 8 +++--- src/pipecat/transports/services/daily.py | 33 ++++++++++++++---------- 2 files changed, 24 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3b2d777d9..24ce7459d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,15 +11,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added `DailyTransport` event `on_participant_left`. +- Added support for receiving `DailyTransportMessage`. + ### Fixed - Images are now resized to the size of the output camera. -- Fixed an issue in `DailyTransport` that would not allow the processor to +- Fixed an issue in `DailyTransport` that would not allow the input processor to shutdown if no participant ever joined the room. -- Fixed transport start and stop. In some situation processors would halt or not - shutdown properly. +- Fixed base transports start and stop. In some situation processors would halt + or not shutdown properly. ## [0.0.13] - 2024-05-14 diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 16a1f29ac..8ed75abec 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -104,6 +104,7 @@ class DailyCallbacks(BaseModel): on_participant_joined: Callable[[Mapping[str, Any]], None] on_participant_left: Callable[[Mapping[str, Any], str], None] on_first_participant_joined: Callable[[Mapping[str, Any]], None] + on_app_message: Callable[[Any, str], None] on_error: Callable[[str], None] @@ -400,6 +401,10 @@ class DailyTransportClient(EventHandler): def on_transcription_stopped(self, stopped_by, stopped_by_error): logger.debug("Transcription stopped") + def on_app_message(self, message: Any, sender: str): + self._callbacks.on_app_message(message, sender) + + # # Daily (CallClient callbacks) # @@ -468,6 +473,15 @@ class DailyInputTransport(BaseInputTransport): await super().process_frame(frame, direction) + # + # App Message + # + + def push_app_message(self, message: Any, sender: str): + frame = DailyTransportMessageFrame(message=message, participant_id=sender) + future = asyncio.run_coroutine_threadsafe(self.push_frame(frame), self.get_event_loop()) + future.result() + # # Camera in # @@ -576,6 +590,7 @@ class DailyTransport(BaseTransport): on_first_participant_joined=self._on_first_participant_joined, on_participant_joined=self._on_participant_joined, on_participant_left=self._on_participant_left, + on_app_message=self._on_app_message, on_error=self._on_error, ) self._params = params @@ -694,6 +709,10 @@ class DailyTransport(BaseTransport): def on_first_participant_joined(self, participant): pass + def on_app_message(self, message: Any, sender: str): + if self._input: + self._input.push_app_message(message, sender) + def event_handler(self, event_name: str): def decorator(handler): self._add_event_handler(event_name, handler) @@ -738,17 +757,3 @@ class DailyTransport(BaseTransport): # def start_recording(self): # self.client.start_recording() - - # def on_error(self, error): - # self._logger.error(f"on_error: {error}") - - # def on_participant_left(self, participant, reason): - # if len(self.client.participants()) < self._min_others_count + 1: - # self._stop_threads.set() - - # def on_app_message(self, message: Any, sender: str): - # if self._loop: - # frame = ReceivedAppMessageFrame(message, sender) - # asyncio.run_coroutine_threadsafe( - # self.receive_queue.put(frame), self._loop - # )