Merge pull request #1129 from pipecat-ai/instant_voice_demo

Pipecat improvements for the instant voice demo

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")

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):