From cc90f5ab9f21dc558f826f600456838f3acebe4f Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 12 Feb 2025 11:46:49 -0300 Subject: [PATCH] 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")