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

Add _reconnect_if_needed to simplify reconnect logic

Browse Source
This commit is contained in:
Mark Backman
2025-02-12 10:00:48 -05:00
parent 9f728aa623
commit 5d6370690c
1 changed files with 13 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
src/pipecat/services/google/google.py
View File

@@ -1566,29 +1566,33 @@ class GoogleSTTService(STTService):
return [language_to_google_stt_language(lang) or "en-US" for lang in language]
return language_to_google_stt_language(language) or "en-US"
async def _reconnect_if_needed(self):
"""Reconnect the stream if it's currently active."""
if self._streaming_task:
logger.debug("Reconnecting stream due to configuration changes")
await self._disconnect()
await self._connect()
async def set_languages(self, languages: List[Language]):
"""Update the service's recognition languages.
Args:
languages: List of languages for recognition. First language is primary.
"""
logger.info(f"Switching STT languages to: {languages}")
logger.debug(f"Switching STT languages to: {languages}")
self._settings["language_codes"] = [
self.language_to_service_language(lang) for lang in languages
]
# Recreate stream with new languages
if self._streaming_task:
await self._disconnect()
await self._connect()
await self._reconnect_if_needed()
async def set_model(self, model: str):
"""Update the service's recognition model."""
logger.debug(f"Switching STT model to: {model}")
await super().set_model(model)
self._settings["model"] = model
# Recreate stream with new model
if self._streaming_task:
await self._disconnect()
await self._connect()
await self._reconnect_if_needed()
async def start(self, frame: StartFrame):
await super().start(frame)
@@ -1636,71 +1640,55 @@ class GoogleSTTService(STTService):
Changes that affect the streaming configuration will cause
the stream to be reconnected.
"""
needs_reconnect = False
# Update settings with new values
if languages is not None:
logger.debug(f"Updating language to: {languages}")
self._settings["language_codes"] = [
self.language_to_service_language(lang) for lang in languages
]
needs_reconnect = True
if model is not None:
logger.debug(f"Updating model to: {model}")
self._settings["model"] = model
needs_reconnect = True
if enable_automatic_punctuation is not None:
logger.debug(f"Updating automatic punctuation to: {enable_automatic_punctuation}")
self._settings["enable_automatic_punctuation"] = enable_automatic_punctuation
needs_reconnect = True
if enable_spoken_punctuation is not None:
logger.debug(f"Updating spoken punctuation to: {enable_spoken_punctuation}")
self._settings["enable_spoken_punctuation"] = enable_spoken_punctuation
needs_reconnect = True
if enable_spoken_emojis is not None:
logger.debug(f"Updating spoken emojis to: {enable_spoken_emojis}")
self._settings["enable_spoken_emojis"] = enable_spoken_emojis
needs_reconnect = True
if profanity_filter is not None:
logger.debug(f"Updating profanity filter to: {profanity_filter}")
self._settings["profanity_filter"] = profanity_filter
needs_reconnect = True
if enable_word_time_offsets is not None:
logger.debug(f"Updating word time offsets to: {enable_word_time_offsets}")
self._settings["enable_word_time_offsets"] = enable_word_time_offsets
needs_reconnect = True
if enable_word_confidence is not None:
logger.debug(f"Updating word confidence to: {enable_word_confidence}")
self._settings["enable_word_confidence"] = enable_word_confidence
needs_reconnect = True
if enable_interim_results is not None:
logger.debug(f"Updating interim results to: {enable_interim_results}")
self._settings["enable_interim_results"] = enable_interim_results
needs_reconnect = True
if enable_voice_activity_events is not None:
logger.debug(f"Updating voice activity events to: {enable_voice_activity_events}")
self._settings["enable_voice_activity_events"] = enable_voice_activity_events
needs_reconnect = True
if location is not None:
logger.debug(f"Updating location to: {location}")
self._location = location
needs_reconnect = True
# Reconnect the stream if necessary
if needs_reconnect and self._streaming_task:
logger.debug("Reconnecting stream due to configuration changes")
await self._disconnect()
await self._connect()
# Reconnect the stream for updates
await self._reconnect_if_needed()
async def _connect(self):
"""Initialize streaming recognition config and stream."""