From c0b3fe3985aba168a7f6b99f3741138ca53e70c7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 17 Apr 2025 11:54:07 -0700 Subject: [PATCH] services: only read from TTS websocket if websocket connection established --- src/pipecat/services/cartesia/tts.py | 3 ++- src/pipecat/services/elevenlabs/tts.py | 4 ++-- src/pipecat/services/fish/tts.py | 3 ++- src/pipecat/services/gladia/stt.py | 2 +- src/pipecat/services/lmnt/tts.py | 2 +- src/pipecat/services/neuphonic/tts.py | 11 ++++++++--- src/pipecat/services/playht/tts.py | 2 +- src/pipecat/services/rime/tts.py | 2 +- 8 files changed, 18 insertions(+), 11 deletions(-) diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py index 3fc70c5cd..2acabd541 100644 --- a/src/pipecat/services/cartesia/tts.py +++ b/src/pipecat/services/cartesia/tts.py @@ -185,7 +185,8 @@ class CartesiaTTSService(AudioContextWordTTSService): async def _connect(self): await self._connect_websocket() - if not self._receive_task: + + if self._websocket and not self._receive_task: self._receive_task = self.create_task(self._receive_task_handler(self._report_error)) async def _disconnect(self): diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index 54ba99fb7..4efaa8807 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -308,10 +308,10 @@ class ElevenLabsTTSService(InterruptibleWordTTSService): async def _connect(self): await self._connect_websocket() - if not self._receive_task: + if self._websocket and not self._receive_task: self._receive_task = self.create_task(self._receive_task_handler(self._report_error)) - if not self._keepalive_task: + if self._websocket and not self._keepalive_task: self._keepalive_task = self.create_task(self._keepalive_task_handler()) async def _disconnect(self): diff --git a/src/pipecat/services/fish/tts.py b/src/pipecat/services/fish/tts.py index 177444f35..abf609177 100644 --- a/src/pipecat/services/fish/tts.py +++ b/src/pipecat/services/fish/tts.py @@ -104,7 +104,8 @@ class FishAudioTTSService(InterruptibleTTSService): async def _connect(self): await self._connect_websocket() - if not self._receive_task: + + if self._websocket and not self._receive_task: self._receive_task = self.create_task(self._receive_task_handler(self._report_error)) async def _disconnect(self): diff --git a/src/pipecat/services/gladia/stt.py b/src/pipecat/services/gladia/stt.py index 03dd35ff2..55581b3e6 100644 --- a/src/pipecat/services/gladia/stt.py +++ b/src/pipecat/services/gladia/stt.py @@ -285,7 +285,7 @@ class GladiaSTTService(STTService): settings = self._prepare_settings() response = await self._setup_gladia(settings) self._websocket = await websockets.connect(response["url"]) - if not self._receive_task: + if self._websocket and not self._receive_task: self._receive_task = self.create_task(self._receive_task_handler()) async def stop(self, frame: EndFrame): diff --git a/src/pipecat/services/lmnt/tts.py b/src/pipecat/services/lmnt/tts.py index eb07c9e8a..b9e0bc79f 100644 --- a/src/pipecat/services/lmnt/tts.py +++ b/src/pipecat/services/lmnt/tts.py @@ -109,7 +109,7 @@ class LmntTTSService(InterruptibleTTSService): async def _connect(self): await self._connect_websocket() - if not self._receive_task: + if self._websocket and not self._receive_task: self._receive_task = self.create_task(self._receive_task_handler(self._report_error)) async def _disconnect(self): diff --git a/src/pipecat/services/neuphonic/tts.py b/src/pipecat/services/neuphonic/tts.py index 0d6a2de44..1819482b8 100644 --- a/src/pipecat/services/neuphonic/tts.py +++ b/src/pipecat/services/neuphonic/tts.py @@ -106,6 +106,9 @@ class NeuphonicTTSService(InterruptibleTTSService): self._started = False self._cumulative_time = 0 + self._receive_task = None + self._keepalive_task = None + def can_generate_metrics(self) -> bool: return True @@ -159,8 +162,11 @@ class NeuphonicTTSService(InterruptibleTTSService): async def _connect(self): await self._connect_websocket() - self._receive_task = self.create_task(self._receive_task_handler(self._report_error)) - self._keepalive_task = self.create_task(self._keepalive_task_handler()) + if self._websocket and not self._receive_task: + self._receive_task = self.create_task(self._receive_task_handler(self._report_error)) + + if self._websocket and not self._keepalive_task: + self._keepalive_task = self.create_task(self._keepalive_task_handler()) async def _disconnect(self): if self._receive_task: @@ -193,7 +199,6 @@ class NeuphonicTTSService(InterruptibleTTSService): url = f"{self._url}/speak/{self._settings['lang_code']}?{'&'.join(query_params)}" self._websocket = await websockets.connect(url) - except Exception as e: logger.error(f"{self} initialization error: {e}") self._websocket = None diff --git a/src/pipecat/services/playht/tts.py b/src/pipecat/services/playht/tts.py index 96bf1c75b..eb1b8d3f1 100644 --- a/src/pipecat/services/playht/tts.py +++ b/src/pipecat/services/playht/tts.py @@ -157,7 +157,7 @@ class PlayHTTTSService(InterruptibleTTSService): async def _connect(self): await self._connect_websocket() - if not self._receive_task: + if self._websocket and not self._receive_task: self._receive_task = self.create_task(self._receive_task_handler(self._report_error)) async def _disconnect(self): diff --git a/src/pipecat/services/rime/tts.py b/src/pipecat/services/rime/tts.py index abf08bbaa..d3e8e5adb 100644 --- a/src/pipecat/services/rime/tts.py +++ b/src/pipecat/services/rime/tts.py @@ -168,7 +168,7 @@ class RimeTTSService(AudioContextWordTTSService): """Establish websocket connection and start receive task.""" await self._connect_websocket() - if not self._receive_task: + if self._websocket and not self._receive_task: self._receive_task = self.create_task(self._receive_task_handler(self._report_error)) async def _disconnect(self):