wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

GooglSTTService: cleanup request queues

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-06-25 10:22:29 -07:00
parent eb5ecab104
commit d2730e6741
1 changed files with 15 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
src/pipecat/services/google/stt.py
View File

@@ -437,7 +437,6 @@ class GoogleSTTService(STTService):
self._location = location self._location = location
self._stream = None self._stream = None
self._config = None self._config = None
self._request_queue = asyncio.Queue()
self._streaming_task = None self._streaming_task = None
# Used for keep-alive logic # Used for keep-alive logic
@@ -684,23 +683,15 @@ class GoogleSTTService(STTService):
), ),
) )
self._request_queue = asyncio.Queue()
self._streaming_task = self.create_task(self._stream_audio()) self._streaming_task = self.create_task(self._stream_audio())
async def _disconnect(self): async def _disconnect(self):
"""Clean up streaming recognition resources.""" """Clean up streaming recognition resources."""
if self._streaming_task: if self._streaming_task:
logger.debug("Disconnecting from Google Speech-to-Text") logger.debug("Disconnecting from Google Speech-to-Text")
# Send sentinel value to stop request generator
await self._request_queue.put(None)
await self.cancel_task(self._streaming_task) await self.cancel_task(self._streaming_task)
self._streaming_task = None self._streaming_task = None
# Clear any remaining items in the queue
while not self._request_queue.empty():
try:
self._request_queue.get_nowait()
self._request_queue.task_done()
except asyncio.QueueEmpty:
break
async def _request_generator(self): async def _request_generator(self):
"""Generates requests for the streaming recognize method.""" """Generates requests for the streaming recognize method."""
@@ -715,29 +706,23 @@ class GoogleSTTService(STTService):
) )
while True: while True:
try: audio_data = await self._request_queue.get()
audio_data = await self._request_queue.get()
if audio_data is None: # Sentinel value to stop
break
# Check streaming limit self._request_queue.task_done()
if (int(time.time() * 1000) - self._stream_start_time) > self.STREAMING_LIMIT:
logger.debug("Streaming limit reached, initiating graceful reconnection")
# Instead of immediate reconnection, we'll break and let the stream close naturally
self._last_audio_input = self._audio_input
self._audio_input = []
self._restart_counter += 1
# Put the current audio chunk back in the queue
await self._request_queue.put(audio_data)
break
self._audio_input.append(audio_data) # Check streaming limit
yield cloud_speech.StreamingRecognizeRequest(audio=audio_data) if (int(time.time() * 1000) - self._stream_start_time) > self.STREAMING_LIMIT:
logger.debug("Streaming limit reached, initiating graceful reconnection")
except asyncio.CancelledError: # Instead of immediate reconnection, we'll break and let the stream close naturally
self._last_audio_input = self._audio_input
self._audio_input = []
self._restart_counter += 1
# Put the current audio chunk back in the queue
await self._request_queue.put(audio_data)
break break
finally:
self._request_queue.task_done() self._audio_input.append(audio_data)
yield cloud_speech.StreamingRecognizeRequest(audio=audio_data)
except Exception as e: except Exception as e:
logger.error(f"Error in request generator: {e}") logger.error(f"Error in request generator: {e}")