transports(websocket): use push_audio_frame()

CHANGELOG.md

@@ -19,6 +19,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed a couple of issues with `WebsocketServerTransport`. It needed to use
+  `push_audio_frame()` and also VAD was not working properly.
+
 - Fixed an issue that would cause LLM aggregator to fail with small
   `VADParams.stop_secs` values.
 

src/pipecat/transports/base_input.py

@@ -70,7 +70,8 @@ class BaseInputTransport(FrameProcessor):
         return self._params.vad_analyzer
 
     def push_audio_frame(self, frame: AudioRawFrame):
-        self._audio_in_queue.put_nowait(frame)
+        if self._params.audio_in_enabled or self._params.vad_enabled:
+            self._audio_in_queue.put_nowait(frame)
 
     #
     # Frame processor

src/pipecat/transports/network/websocket_server.py

@@ -7,7 +7,6 @@
 
 import asyncio
 import io
-import queue
 import wave
 import websockets
 
@@ -53,7 +52,6 @@ class WebsocketServerInputTransport(BaseInputTransport):
 
         self._websocket: websockets.WebSocketServerProtocol | None = None
 
-        self._client_audio_queue = queue.Queue()
         self._stop_server_event = asyncio.Event()
 
     async def start(self, frame: StartFrame):
@@ -65,12 +63,6 @@ class WebsocketServerInputTransport(BaseInputTransport):
         await self._server_task
         await super().stop()
 
-    def read_next_audio_frame(self) -> AudioRawFrame | None:
-        try:
-            return self._client_audio_queue.get(timeout=1)
-        except queue.Empty:
-            return None
-
     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:
@@ -90,8 +82,8 @@ class WebsocketServerInputTransport(BaseInputTransport):
         # Handle incoming messages
         async for message in websocket:
             frame = self._params.serializer.deserialize(message)
-            if isinstance(frame, AudioRawFrame) and self._params.audio_in_enabled:
+            if isinstance(frame, AudioRawFrame):
-                self._client_audio_queue.put_nowait(frame)
+                self.push_audio_frame(frame)
             else:
                 await self._internal_push_frame(frame)