From 72c27215b640de635c0a61356397dc1f3c3b6350 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 11 Jun 2024 23:10:37 -0700 Subject: [PATCH] transports(websocket): use push_audio_frame() --- CHANGELOG.md | 3 +++ src/pipecat/transports/base_input.py | 3 ++- src/pipecat/transports/network/websocket_server.py | 12 ++---------- 3 files changed, 7 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0cbb495f9..c499dcc1b 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 9496cd3b4..303979816 100644 --- a/src/pipecat/transports/base_input.py +++ b/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 diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index 8456c4536..bf51732e1 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/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: - self._client_audio_queue.put_nowait(frame) + if isinstance(frame, AudioRawFrame): + self.push_audio_frame(frame) else: await self._internal_push_frame(frame)