diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index ad4636a74..e4d0f00ef 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -7,6 +7,7 @@ import asyncio import base64 import json +import random import uuid from typing import AsyncGenerator, List, Optional, Union @@ -222,6 +223,10 @@ class CartesiaTTSService(WordTTSService): async def _receive_task_handler(self): try: async for message in self._get_websocket(): + # Randomly cancel the asyncio task 1% of the time + if random.random() < 0.01: + logger.info(f"Cancelling task for {self} due to random chance") + asyncio.current_task().cancel() msg = json.loads(message) if not msg or msg["context_id"] != self._context_id: continue @@ -256,6 +261,7 @@ class CartesiaTTSService(WordTTSService): logger.error(f"Cartesia error, unknown message type: {msg}") except asyncio.CancelledError: pass + # await self.push_error(ErrorFrame(f"{self} cancelled", True)) except Exception as e: logger.error(f"{self} exception: {e}") diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 025a5bed2..6fc24fde7 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -71,6 +71,7 @@ class BaseInputTransport(FrameProcessor): return self._params.vad_analyzer async def push_audio_frame(self, frame: InputAudioRawFrame): + logger.info(f"Pushing audio qsize: {self._audio_in_queue.qsize()}") if self._params.audio_in_enabled or self._params.vad_enabled: await self._audio_in_queue.put(frame) @@ -167,6 +168,7 @@ class BaseInputTransport(FrameProcessor): return vad_state async def _audio_task_handler(self): + logger.info("_audio_task_handler started") vad_state: VADState = VADState.QUIET while True: try: diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index c0c8595e9..b330f61fa 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -106,6 +106,7 @@ class WebsocketServerInputTransport(BaseInputTransport): continue if isinstance(frame, AudioRawFrame): + logger.info("websocket_server") await self.push_audio_frame( InputAudioRawFrame( audio=frame.audio,