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

Sending the RTVI messages to the websocket

Browse Source
This commit is contained in:
Filipi Fuchter
2025-02-12 11:46:49 -03:00
parent 08f40d9179
commit cc90f5ab9f
1 changed files with 17 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -6,6 +6,7 @@
import asyncio import asyncio
import io import io
import json
import time import time
import wave import wave
from typing import Awaitable, Callable, Optional from typing import Awaitable, Callable, Optional
@@ -21,6 +22,9 @@ from pipecat.frames.frames import (
OutputAudioRawFrame, OutputAudioRawFrame,
StartFrame, StartFrame,
StartInterruptionFrame, StartInterruptionFrame,
TextFrame,
TransportMessageFrame,
TransportMessageUrgentFrame,
) )
from pipecat.processors.frame_processor import FrameDirection from pipecat.processors.frame_processor import FrameDirection
from pipecat.serializers.base_serializer import FrameSerializer from pipecat.serializers.base_serializer import FrameSerializer
@@ -46,6 +50,7 @@ class WebsocketServerCallbacks(BaseModel):
on_client_connected: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]] on_client_connected: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]]
on_client_disconnected: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]] on_client_disconnected: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]]
on_session_timeout: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]] on_session_timeout: Callable[[websockets.WebSocketServerProtocol], Awaitable[None]]
on_websocket_ready: Callable[[], Awaitable[None]]
class WebsocketServerInputTransport(BaseInputTransport): class WebsocketServerInputTransport(BaseInputTransport):
@@ -96,6 +101,7 @@ class WebsocketServerInputTransport(BaseInputTransport):
async def _server_task_handler(self): async def _server_task_handler(self):
logger.info(f"Starting websocket server on {self._host}:{self._port}") logger.info(f"Starting websocket server on {self._host}:{self._port}")
async with websockets.serve(self._client_handler, self._host, self._port) as server: 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() await self._stop_server_event.wait()
async def _client_handler(self, websocket: websockets.WebSocketServerProtocol, path): async def _client_handler(self, websocket: websockets.WebSocketServerProtocol, path):
@@ -186,6 +192,12 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
await self._write_frame(frame) await self._write_frame(frame)
self._next_send_time = 0 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): async def write_raw_audio_frames(self, frames: bytes):
if not self._websocket: if not self._websocket:
# Simulate audio playback with a sleep. # Simulate audio playback with a sleep.
@@ -254,6 +266,7 @@ class WebsocketServerTransport(BaseTransport):
on_client_connected=self._on_client_connected, on_client_connected=self._on_client_connected,
on_client_disconnected=self._on_client_disconnected, on_client_disconnected=self._on_client_disconnected,
on_session_timeout=self._on_session_timeout, on_session_timeout=self._on_session_timeout,
on_websocket_ready=self._on_websocket_ready,
) )
self._input: Optional[WebsocketServerInputTransport] = None self._input: Optional[WebsocketServerInputTransport] = None
self._output: Optional[WebsocketServerOutputTransport] = 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_connected")
self._register_event_handler("on_client_disconnected") self._register_event_handler("on_client_disconnected")
self._register_event_handler("on_session_timeout") self._register_event_handler("on_session_timeout")
self._register_event_handler("on_websocket_ready")
def input(self) -> WebsocketServerInputTransport: def input(self) -> WebsocketServerInputTransport:
if not self._input: if not self._input:
@@ -293,3 +307,6 @@ class WebsocketServerTransport(BaseTransport):
async def _on_session_timeout(self, websocket): async def _on_session_timeout(self, websocket):
await self._call_event_handler("on_session_timeout", websocket) await self._call_event_handler("on_session_timeout", websocket)
async def _on_websocket_ready(self):
await self._call_event_handler("on_websocket_ready")