From 08f40d9179b398d84e21e2784031e0d926c6b6e9 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 12 Feb 2025 11:46:37 -0300 Subject: [PATCH 1/2] Adding support to DailyTransport receive raw-audio through appMessage --- src/pipecat/transports/services/daily.py | 29 +++++++++++++++++++----- 1 file changed, 23 insertions(+), 6 deletions(-) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index b71d1ce30..99071cf5c 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -5,6 +5,7 @@ # import asyncio +import base64 import time import warnings from concurrent.futures import ThreadPoolExecutor @@ -13,9 +14,6 @@ from typing import Any, Awaitable, Callable, Mapping, Optional import aiohttp from daily import ( - CallClient, - Daily, - EventHandler, VirtualCameraDevice, VirtualMicrophoneDevice, VirtualSpeakerDevice, @@ -291,6 +289,7 @@ class DailyTransportClient(EventHandler): self._transcription_ids = [] self._transcription_status = None + self._joining = False self._joined = False self._joined_event = asyncio.Event() self._leave_counter = 0 @@ -423,13 +422,14 @@ class DailyTransportClient(EventHandler): ) async def join(self): - # Transport already joined, ignore. - if self._joined: + # Transport already joined or joining, ignore. + if self._joined or self._joining: # Increment leave counter if we already joined. self._leave_counter += 1 return logger.info(f"Joining {self._room_url}") + self._joining = True # For performance reasons, never subscribe to video streams (unless a # video renderer is registered). @@ -444,6 +444,7 @@ class DailyTransportClient(EventHandler): if not error: self._joined = True + self._joining = False # Increment leave counter if we successfully joined. self._leave_counter += 1 @@ -462,6 +463,7 @@ class DailyTransportClient(EventHandler): except asyncio.TimeoutError: error_msg = f"Time out joining {self._room_url}" logger.error(error_msg) + self._joining = False await self._callbacks.on_error(error_msg) async def _start_transcription(self): @@ -1198,7 +1200,22 @@ class DailyTransport(BaseTransport): async def _on_app_message(self, message: Any, sender: str): if self._input: - await self._input.push_app_message(message, sender) + if message["type"] in {"raw-audio", "raw-audio-batch"}: + data = message["data"] + audio_list = data.get( + "base64AudioBatch", [data.get("base64Audio")] + ) # Ensure a list + + for base64_audio in filter(None, audio_list): # Filter out None values + pcm_bytes = base64.b64decode(base64_audio) + frame = InputAudioRawFrame( + audio=pcm_bytes, + sample_rate=data["sampleRate"], + num_channels=data["numChannels"], + ) + await self._input.push_audio_frame(frame) + else: + await self._input.push_app_message(message, sender) await self._call_event_handler("on_app_message", message, sender) async def _on_call_state_updated(self, state: str): From cc90f5ab9f21dc558f826f600456838f3acebe4f Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 12 Feb 2025 11:46:49 -0300 Subject: [PATCH 2/2] Sending the RTVI messages to the websocket --- .../transports/network/websocket_server.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index 7dae04c7f..234f85119 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -6,6 +6,7 @@ import asyncio import io +import json import time import wave from typing import Awaitable, Callable, Optional @@ -21,6 +22,9 @@ from pipecat.frames.frames import ( OutputAudioRawFrame, StartFrame, StartInterruptionFrame, + TextFrame, + TransportMessageFrame, + TransportMessageUrgentFrame, ) from pipecat.processors.frame_processor import FrameDirection from pipecat.serializers.base_serializer import FrameSerializer @@ -46,6 +50,7 @@ class WebsocketServerCallbacks(BaseModel): on_client_connected: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]] on_client_disconnected: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]] on_session_timeout: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]] + on_websocket_ready: Callable[[], Awaitable[None]] class WebsocketServerInputTransport(BaseInputTransport): @@ -96,6 +101,7 @@ class WebsocketServerInputTransport(BaseInputTransport): async def _server_task_handler(self): logger.info(f"Starting websocket server on {self._host}:{self._port}") async with websockets.serve(self._client_handler, self._host, self._port) as server: + await self._callbacks.on_websocket_ready() await self._stop_server_event.wait() async def _client_handler(self, websocket: websockets.WebSocketServerProtocol, path): @@ -186,6 +192,12 @@ class WebsocketServerOutputTransport(BaseOutputTransport): await self._write_frame(frame) self._next_send_time = 0 + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + message_frame = TextFrame( + text=json.dumps(frame.message), + ) + await self._write_frame(message_frame) + async def write_raw_audio_frames(self, frames: bytes): if not self._websocket: # Simulate audio playback with a sleep. @@ -254,6 +266,7 @@ class WebsocketServerTransport(BaseTransport): on_client_connected=self._on_client_connected, on_client_disconnected=self._on_client_disconnected, on_session_timeout=self._on_session_timeout, + on_websocket_ready=self._on_websocket_ready, ) self._input: Optional[WebsocketServerInputTransport] = None self._output: Optional[WebsocketServerOutputTransport] = None @@ -264,6 +277,7 @@ class WebsocketServerTransport(BaseTransport): self._register_event_handler("on_client_connected") self._register_event_handler("on_client_disconnected") self._register_event_handler("on_session_timeout") + self._register_event_handler("on_websocket_ready") def input(self) -> WebsocketServerInputTransport: if not self._input: @@ -293,3 +307,6 @@ class WebsocketServerTransport(BaseTransport): async def _on_session_timeout(self, websocket): await self._call_event_handler("on_session_timeout", websocket) + + async def _on_websocket_ready(self): + await self._call_event_handler("on_websocket_ready")