# Copyright 2023 LiveKit, Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from __future__ import annotations import asyncio import contextlib import dataclasses import time import weakref from collections.abc import AsyncGenerator, AsyncIterable, Callable from dataclasses import dataclass from datetime import timedelta from typing import cast, get_args from grpc.aio import StreamStreamCall import google.auth from google.api_core.client_options import ClientOptions from google.api_core.exceptions import DeadlineExceeded, GoogleAPICallError from google.auth import default as gauth_default from google.auth.exceptions import DefaultCredentialsError from google.cloud.speech_v1 import SpeechAsyncClient as SpeechAsyncClientV1 from google.cloud.speech_v1.types import cloud_speech as cloud_speech_v1, resource as resource_v1 from google.cloud.speech_v2 import SpeechAsyncClient as SpeechAsyncClientV2 from google.cloud.speech_v2.types import cloud_speech as cloud_speech_v2 from google.protobuf.duration_pb2 import Duration from livekit import rtc from livekit.agents import ( DEFAULT_API_CONNECT_OPTIONS, APIConnectionError, APIConnectOptions, APIStatusError, APITimeoutError, LanguageCode, stt, utils, ) from livekit.agents.types import ( NOT_GIVEN, NotGivenOr, ) from livekit.agents.utils import is_given from livekit.agents.utils.aio import ChanClosed from livekit.agents.voice.io import TimedString from .log import logger from .models import EndpointingSensitivity, SpeechLanguages, SpeechModels, SpeechModelsV2 LgType = SpeechLanguages | str LanguagesInput = LgType | list[LgType] # Google STT has a timeout of 5 mins, we'll attempt to restart the session # before that timeout is reached _max_session_duration = 240 # Google is very sensitive to background noise, so we'll ignore results with low confidence _default_min_confidence = 0.65 # Default boost applied to keyterms set via the provider-agnostic keyterm hook. # Google accepts boosts in roughly 0-20; a moderate value biases toward the terms without # over-triggering false positives. _DEFAULT_KEYTERM_BOOST = 10.0 # This class is only be used internally to encapsulate the options @dataclass class STTOptions: languages: list[LgType] detect_language: bool interim_results: bool punctuate: bool spoken_punctuation: bool enable_word_time_offsets: bool enable_word_confidence: bool enable_voice_activity_events: bool model: SpeechModels | str sample_rate: int min_confidence_threshold: float profanity_filter: bool denoiser_config: NotGivenOr[cloud_speech_v2.DenoiserConfig] = NOT_GIVEN adaptation: NotGivenOr[cloud_speech_v2.SpeechAdaptation | resource_v1.SpeechAdaptation] = ( NOT_GIVEN ) keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN speech_start_timeout: NotGivenOr[float] = NOT_GIVEN speech_end_timeout: NotGivenOr[float] = NOT_GIVEN endpointing_sensitivity: NotGivenOr[EndpointingSensitivity] = NOT_GIVEN @property def version(self) -> int: return 2 if self.model in get_args(SpeechModelsV2) else 1 def build_adaptation( self, ) -> cloud_speech_v2.SpeechAdaptation | resource_v1.SpeechAdaptation | None: if is_given(self.adaptation): return self.adaptation if is_given(self.keywords): if self.version == 2: return cloud_speech_v2.SpeechAdaptation( phrase_sets=[ cloud_speech_v2.SpeechAdaptation.AdaptationPhraseSet( inline_phrase_set=cloud_speech_v2.PhraseSet( phrases=[ cloud_speech_v2.PhraseSet.Phrase(value=keyword, boost=boost) for keyword, boost in self.keywords ] ) ) ] ) return resource_v1.SpeechAdaptation( phrase_sets=[ resource_v1.PhraseSet( name="keywords", phrases=[ resource_v1.PhraseSet.Phrase(value=keyword, boost=boost) for keyword, boost in self.keywords ], ) ] ) return None class STT(stt.STT): def __init__( self, *, languages: LanguagesInput = "en-US", # Google STT can accept multiple languages detect_language: bool = True, interim_results: bool = True, punctuate: bool = True, spoken_punctuation: bool = False, enable_word_time_offsets: NotGivenOr[bool] = NOT_GIVEN, enable_word_confidence: bool = False, enable_voice_activity_events: bool = False, model: SpeechModels | str = "latest_long", location: str = "global", profanity_filter: bool = False, sample_rate: int = 16000, min_confidence_threshold: float = _default_min_confidence, denoiser_config: NotGivenOr[cloud_speech_v2.DenoiserConfig] = NOT_GIVEN, adaptation: NotGivenOr[ cloud_speech_v2.SpeechAdaptation | resource_v1.SpeechAdaptation ] = NOT_GIVEN, credentials_info: NotGivenOr[dict] = NOT_GIVEN, credentials_file: NotGivenOr[str] = NOT_GIVEN, keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN, speech_start_timeout: NotGivenOr[float] = NOT_GIVEN, speech_end_timeout: NotGivenOr[float] = NOT_GIVEN, endpointing_sensitivity: NotGivenOr[EndpointingSensitivity] = NOT_GIVEN, use_streaming: NotGivenOr[bool] = NOT_GIVEN, ): """ Create a new instance of Google STT. Credentials must be provided, either by using the ``credentials_info`` dict, or reading from the file specified in ``credentials_file`` or via Application Default Credentials as described in https://cloud.google.com/docs/authentication/application-default-credentials args: languages(LanguagesInput): list of language codes to recognize (default: "en-US") detect_language(bool): whether to detect the language of the audio (default: True) interim_results(bool): whether to return interim results (default: True) punctuate(bool): whether to punctuate the audio (default: True) spoken_punctuation(bool): whether to use spoken punctuation (default: False) enable_word_time_offsets(bool): whether to enable word time offsets (default: None) enable_word_confidence(bool): whether to enable word confidence (default: False) enable_voice_activity_events(bool): whether to enable voice activity events (default: False) model(SpeechModels): the model to use for recognition default: "latest_long" location(str): the location to use for recognition default: "global" profanity_filter(bool): whether to filter out profanities default: False sample_rate(int): the sample rate of the audio default: 16000 min_confidence_threshold(float): minimum confidence threshold for recognition (default: 0.65) denoiser_config (DenoiserConfig): the denoiser configuration (default: None) adaptation (SpeechAdaptation): speech adaptation for biasing specific words and phrases (default: None) credentials_info(dict): the credentials info to use for recognition (default: None) credentials_file(str): the credentials file to use for recognition (default: None) keywords(List[tuple[str, float]]): list of keywords to recognize (default: None) speech_start_timeout(float): maximum seconds to wait for speech to begin before timeout (default: None) speech_end_timeout(float): seconds of silence before marking utterance as complete (default: None) endpointing_sensitivity(EndpointingSensitivity): controls the trade-off between latency and accuracy when detecting end-of-speech. Only supported with chirp_3. Options: ENDPOINTING_SENSITIVITY_STANDARD (default), ENDPOINTING_SENSITIVITY_SHORT, ENDPOINTING_SENSITIVITY_SUPERSHORT (default: None) use_streaming(bool): whether to use streaming for recognition (default: True) """ if is_given(endpointing_sensitivity) and model != "chirp_3": logger.warning( "endpointing_sensitivity is only supported with the chirp_3 model; ignoring." ) endpointing_sensitivity = NOT_GIVEN if is_given(adaptation): if is_given(keywords): logger.warning( "Both 'adaptation' and 'keywords' are set; 'keywords' will be ignored." ) self._validate_adaptation(adaptation, 2 if model in get_args(SpeechModelsV2) else 1) if not is_given(use_streaming): use_streaming = True if model == "chirp_3": if is_given(enable_word_time_offsets) and enable_word_time_offsets: logger.warning( "Chirp 3 does not support word timestamps, setting 'enable_word_time_offsets' to False." ) enable_word_time_offsets = False elif is_given(enable_word_time_offsets): enable_word_time_offsets = enable_word_time_offsets else: enable_word_time_offsets = True super().__init__( capabilities=stt.STTCapabilities( streaming=use_streaming, interim_results=True, aligned_transcript="word" if enable_word_time_offsets and use_streaming else False, # adaptation shadows keywords (see build_adaptation), so keyterms can't be applied keyterms=not is_given(adaptation), ) ) self._location = location self._credentials_info = credentials_info self._credentials_file = credentials_file self._project_id: str | None = None if not is_given(credentials_file) and not is_given(credentials_info): try: gauth_default() except DefaultCredentialsError: raise ValueError( "Application default credentials must be available " "when using Google STT without explicitly passing " "credentials through credentials_info or credentials_file." ) from None if isinstance(languages, str): languages = [LanguageCode(languages)] else: languages = [LanguageCode(lg) for lg in languages] self._config = STTOptions( languages=languages, detect_language=detect_language, interim_results=interim_results, punctuate=punctuate, spoken_punctuation=spoken_punctuation, enable_word_time_offsets=enable_word_time_offsets, enable_word_confidence=enable_word_confidence, enable_voice_activity_events=enable_voice_activity_events, model=model, profanity_filter=profanity_filter, sample_rate=sample_rate, min_confidence_threshold=min_confidence_threshold, adaptation=adaptation, keywords=keywords, denoiser_config=denoiser_config, speech_start_timeout=speech_start_timeout, speech_end_timeout=speech_end_timeout, endpointing_sensitivity=endpointing_sensitivity, ) # user-tuned (phrase, boost) pairs, kept separate so keyterm updates can't clobber them self._user_keywords: list[tuple[str, float]] = list(keywords) if is_given(keywords) else [] self._session_keyterms: list[str] = [] # framework-managed; merged with user keywords self._streams = weakref.WeakSet[SpeechStream]() self._pool = utils.ConnectionPool[SpeechAsyncClientV2 | SpeechAsyncClientV1]( max_session_duration=_max_session_duration, connect_cb=self._create_client, ) @property def model(self) -> str: return self._config.model @property def provider(self) -> str: return "Google Cloud Platform" async def _create_client(self, timeout: float) -> SpeechAsyncClientV2 | SpeechAsyncClientV1: # Add support for passing a specific location that matches recognizer # see: https://cloud.google.com/speech-to-text/v2/docs/speech-to-text-supported-languages # TODO(long): how to set timeout? client_options = None client: SpeechAsyncClientV2 | SpeechAsyncClientV1 | None = None client_cls = SpeechAsyncClientV2 if self._config.version == 2 else SpeechAsyncClientV1 if self._location != "global": client_options = ClientOptions(api_endpoint=f"{self._location}-speech.googleapis.com") if is_given(self._credentials_info): client = client_cls.from_service_account_info( self._credentials_info, client_options=client_options ) elif is_given(self._credentials_file): credentials, project_id = google.auth.load_credentials_from_file( # type: ignore[no-untyped-call] self._credentials_file, scopes=["https://www.googleapis.com/auth/cloud-platform"], ) self._project_id = project_id client = client_cls(credentials=credentials, client_options=client_options) else: client = client_cls(client_options=client_options) assert client is not None return client def _get_recognizer(self, client: SpeechAsyncClientV2) -> str: # TODO(theomonnom): should we use recognizers? # recognizers may improve latency https://cloud.google.com/speech-to-text/v2/docs/recognizers#understand_recognizers if self._project_id is not None: return f"projects/{self._project_id}/locations/{self._location}/recognizers/_" # TODO(theomonnom): find a better way to access the project_id try: project_id = client.transport._credentials.project_id # type: ignore except AttributeError: from google.auth import default as ga_default _, project_id = ga_default() return f"projects/{project_id}/locations/{self._location}/recognizers/_" def _sanitize_options(self, *, language: NotGivenOr[str] = NOT_GIVEN) -> STTOptions: config = dataclasses.replace(self._config) if is_given(language): config.languages = [LanguageCode(language)] if not isinstance(config.languages, list): config.languages = [config.languages] elif not config.detect_language: if len(config.languages) > 1: logger.warning("multiple languages provided, but language detection is disabled") config.languages = [config.languages[0]] return config def _build_recognition_config( self, sample_rate: int, num_channels: int, language: NotGivenOr[SpeechLanguages | str] = NOT_GIVEN, ) -> cloud_speech_v2.RecognitionConfig | cloud_speech_v1.RecognitionConfig: config = self._sanitize_options(language=language) if self._config.version == 2: return cloud_speech_v2.RecognitionConfig( explicit_decoding_config=cloud_speech_v2.ExplicitDecodingConfig( encoding=cloud_speech_v2.ExplicitDecodingConfig.AudioEncoding.LINEAR16, sample_rate_hertz=sample_rate, audio_channel_count=num_channels, ), adaptation=config.build_adaptation(), features=cloud_speech_v2.RecognitionFeatures( enable_automatic_punctuation=config.punctuate, enable_spoken_punctuation=config.spoken_punctuation, enable_word_time_offsets=config.enable_word_time_offsets, enable_word_confidence=config.enable_word_confidence, profanity_filter=config.profanity_filter, ), denoiser_config=config.denoiser_config if is_given(config.denoiser_config) else None, model=config.model, language_codes=config.languages, ) return cloud_speech_v1.RecognitionConfig( encoding=cloud_speech_v1.RecognitionConfig.AudioEncoding.LINEAR16, sample_rate_hertz=sample_rate, audio_channel_count=num_channels, adaptation=config.build_adaptation(), language_code=config.languages[0], alternative_language_codes=config.languages[1:], enable_word_time_offsets=config.enable_word_time_offsets, enable_word_confidence=config.enable_word_confidence, enable_automatic_punctuation=config.punctuate, enable_spoken_punctuation=config.spoken_punctuation, profanity_filter=config.profanity_filter, model=config.model, ) def _build_recognition_request( self, client: SpeechAsyncClientV2 | SpeechAsyncClientV1, config: cloud_speech_v2.RecognitionConfig | cloud_speech_v1.RecognitionConfig, content: bytes, ) -> cloud_speech_v2.RecognizeRequest | cloud_speech_v1.RecognizeRequest: if self._config.version == 2: return cloud_speech_v2.RecognizeRequest( recognizer=self._get_recognizer(cast(SpeechAsyncClientV2, client)), config=config, content=content, ) return cloud_speech_v1.RecognizeRequest( config=config, audio=cloud_speech_v1.RecognitionAudio(content=content), ) async def _recognize_impl( self, buffer: utils.AudioBuffer, *, language: NotGivenOr[SpeechLanguages | str] = NOT_GIVEN, conn_options: APIConnectOptions, ) -> stt.SpeechEvent: frame = rtc.combine_audio_frames(buffer) config = self._build_recognition_config( sample_rate=frame.sample_rate, num_channels=frame.num_channels, language=language, ) try: async with self._pool.connection(timeout=conn_options.timeout) as client: raw = await client.recognize( self._build_recognition_request(client, config, frame.data.tobytes()), timeout=conn_options.timeout, ) return _recognize_response_to_speech_event(raw) except DeadlineExceeded: raise APITimeoutError() from None except GoogleAPICallError as e: raise APIStatusError(f"{e.message} {e.details}", status_code=e.code or -1) from e except Exception as e: raise APIConnectionError() from e def stream( self, *, language: NotGivenOr[SpeechLanguages | str] = NOT_GIVEN, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS, ) -> SpeechStream: config = self._sanitize_options(language=language) stream = SpeechStream( stt=self, pool=self._pool, recognizer_cb=self._get_recognizer, config=config, conn_options=conn_options, ) self._streams.add(stream) return stream def update_options( self, *, languages: NotGivenOr[LanguagesInput] = NOT_GIVEN, detect_language: NotGivenOr[bool] = NOT_GIVEN, interim_results: NotGivenOr[bool] = NOT_GIVEN, punctuate: NotGivenOr[bool] = NOT_GIVEN, spoken_punctuation: NotGivenOr[bool] = NOT_GIVEN, profanity_filter: NotGivenOr[bool] = NOT_GIVEN, model: NotGivenOr[SpeechModels] = NOT_GIVEN, location: NotGivenOr[str] = NOT_GIVEN, denoiser_config: NotGivenOr[cloud_speech_v2.DenoiserConfig] = NOT_GIVEN, adaptation: NotGivenOr[ cloud_speech_v2.SpeechAdaptation | resource_v1.SpeechAdaptation ] = NOT_GIVEN, keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN, speech_start_timeout: NotGivenOr[float] = NOT_GIVEN, speech_end_timeout: NotGivenOr[float] = NOT_GIVEN, endpointing_sensitivity: NotGivenOr[EndpointingSensitivity] = NOT_GIVEN, ) -> None: if is_given(languages): if isinstance(languages, str): self._config.languages = [LanguageCode(languages)] else: self._config.languages = [LanguageCode(lg) for lg in languages] if is_given(detect_language): self._config.detect_language = detect_language if is_given(interim_results): self._config.interim_results = interim_results if is_given(punctuate): self._config.punctuate = punctuate if is_given(spoken_punctuation): self._config.spoken_punctuation = spoken_punctuation if is_given(profanity_filter): self._config.profanity_filter = profanity_filter new_version = ( (2 if model in get_args(SpeechModelsV2) else 1) if is_given(model) else self._config.version ) effective_adaptation = adaptation if is_given(adaptation) else self._config.adaptation if is_given(effective_adaptation) and (is_given(adaptation) or is_given(model)): self._validate_adaptation(effective_adaptation, new_version) if is_given(model): old_version = self._config.version self._config.model = model if self._config.version != old_version: self._pool.invalidate() if is_given(location): self._location = location # if location is changed, fetch a new client and recognizer as per the new location self._pool.invalidate() if is_given(denoiser_config): self._config.denoiser_config = denoiser_config if is_given(adaptation): if is_given(keywords) or is_given(self._config.keywords): logger.warning( "Both 'adaptation' and 'keywords' are set; 'keywords' will be ignored." ) self._config.adaptation = adaptation if is_given(keywords): if is_given(self._config.adaptation) and not is_given(adaptation): logger.warning( "Both 'adaptation' and 'keywords' are set; 'keywords' will be ignored." ) self._user_keywords = list(keywords) # re-merge with the active session keyterms so a user update doesn't drop them, # and forward the merged value to the streams below (not the raw user keywords) keywords = self._get_merged_keywords() self._config.keywords = keywords if is_given(speech_start_timeout): self._config.speech_start_timeout = speech_start_timeout if is_given(speech_end_timeout): self._config.speech_end_timeout = speech_end_timeout if is_given(endpointing_sensitivity): if self._config.model != "chirp_3": logger.warning( "endpointing_sensitivity is only supported with the chirp_3 model; ignoring." ) endpointing_sensitivity = NOT_GIVEN else: self._config.endpointing_sensitivity = endpointing_sensitivity for stream in self._streams: stream.update_options( languages=languages, detect_language=detect_language, interim_results=interim_results, punctuate=punctuate, spoken_punctuation=spoken_punctuation, profanity_filter=profanity_filter, model=model, denoiser_config=denoiser_config, adaptation=adaptation, keywords=keywords, speech_start_timeout=speech_start_timeout, speech_end_timeout=speech_end_timeout, endpointing_sensitivity=endpointing_sensitivity, ) def _get_merged_keywords(self) -> list[tuple[str, float]]: # Google biases via (phrase, boost) pairs; the session hook carries no per-term weight, # so keep the user keyword boosts and bias session terms no stronger than the weakest # user term (or a moderate default when the user gave none). user_phrases = {phrase for phrase, _ in self._user_keywords} session_boost = ( min(boost for _, boost in self._user_keywords) if self._user_keywords else _DEFAULT_KEYTERM_BOOST ) return self._user_keywords + [ (term, session_boost) for term in self._session_keyterms if term not in user_phrases ] def _update_session_keyterms(self, keyterms: list[str]) -> None: if is_given(self._config.adaptation): logger.warning("'adaptation' is set; ignoring keyterms update") return if keyterms == self._session_keyterms: return self._session_keyterms = list(keyterms) merged = self._get_merged_keywords() self._config.keywords = merged for stream in self._streams: if stream._speaking: # defer the reconnect to the end of the utterance so we don't cut it off stream._pending_keywords = merged else: stream.update_options(keywords=merged) async def aclose(self) -> None: await self._pool.aclose() await super().aclose() def _validate_adaptation( self, adaptation: cloud_speech_v2.SpeechAdaptation | resource_v1.SpeechAdaptation, api_version: int, ) -> None: if api_version == 2 and not isinstance(adaptation, cloud_speech_v2.SpeechAdaptation): raise ValueError( "adaptation must be cloud_speech_v2.SpeechAdaptation for v2 models, " f"got {type(adaptation).__name__}" ) if api_version == 1 and not isinstance(adaptation, resource_v1.SpeechAdaptation): raise ValueError( "adaptation must be resource_v1.SpeechAdaptation for v1 models, " f"got {type(adaptation).__name__}" ) class SpeechStream(stt.SpeechStream): def __init__( self, *, stt: STT, conn_options: APIConnectOptions, pool: utils.ConnectionPool[SpeechAsyncClientV2 | SpeechAsyncClientV1], recognizer_cb: Callable[[SpeechAsyncClientV2], str], config: STTOptions, ) -> None: super().__init__(stt=stt, conn_options=conn_options, sample_rate=config.sample_rate) self._pool = pool self._recognizer_cb = recognizer_cb self._config = config self._reconnect_event = asyncio.Event() self._session_connected_at: float = 0 self._speaking = False # keywords set while the user is speaking; applied at END_OF_SPEECH (latest wins) self._pending_keywords: list[tuple[str, float]] | None = None def update_options( self, *, languages: NotGivenOr[LanguagesInput] = NOT_GIVEN, detect_language: NotGivenOr[bool] = NOT_GIVEN, interim_results: NotGivenOr[bool] = NOT_GIVEN, punctuate: NotGivenOr[bool] = NOT_GIVEN, spoken_punctuation: NotGivenOr[bool] = NOT_GIVEN, profanity_filter: NotGivenOr[bool] = NOT_GIVEN, model: NotGivenOr[SpeechModels] = NOT_GIVEN, min_confidence_threshold: NotGivenOr[float] = NOT_GIVEN, denoiser_config: NotGivenOr[cloud_speech_v2.DenoiserConfig] = NOT_GIVEN, adaptation: NotGivenOr[ cloud_speech_v2.SpeechAdaptation | resource_v1.SpeechAdaptation ] = NOT_GIVEN, keywords: NotGivenOr[list[tuple[str, float]]] = NOT_GIVEN, speech_start_timeout: NotGivenOr[float] = NOT_GIVEN, speech_end_timeout: NotGivenOr[float] = NOT_GIVEN, endpointing_sensitivity: NotGivenOr[EndpointingSensitivity] = NOT_GIVEN, ) -> None: if is_given(languages): if isinstance(languages, str): self._config.languages = [LanguageCode(languages)] else: self._config.languages = [LanguageCode(lg) for lg in languages] if is_given(detect_language): self._config.detect_language = detect_language if is_given(interim_results): self._config.interim_results = interim_results if is_given(punctuate): self._config.punctuate = punctuate if is_given(spoken_punctuation): self._config.spoken_punctuation = spoken_punctuation if is_given(profanity_filter): self._config.profanity_filter = profanity_filter if is_given(model): old_version = self._config.version self._config.model = model if self._config.version != old_version: self._pool.invalidate() if is_given(min_confidence_threshold): self._config.min_confidence_threshold = min_confidence_threshold if is_given(denoiser_config): self._config.denoiser_config = denoiser_config if is_given(adaptation): self._config.adaptation = adaptation if is_given(keywords): self._config.keywords = keywords self._pending_keywords = None if is_given(speech_start_timeout): self._config.speech_start_timeout = speech_start_timeout if is_given(speech_end_timeout): self._config.speech_end_timeout = speech_end_timeout if is_given(endpointing_sensitivity): self._config.endpointing_sensitivity = endpointing_sensitivity self._reconnect_event.set() def _on_end_of_speech(self) -> None: if self._pending_keywords is not None: self.update_options(keywords=self._pending_keywords) self._pending_keywords = None def _build_streaming_config( self, ) -> cloud_speech_v2.StreamingRecognitionConfig | cloud_speech_v1.StreamingRecognitionConfig: if self._config.version == 2: # Build voice activity timeout if either timeout is specified voice_activity_timeout = None if is_given(self._config.speech_start_timeout) or is_given( self._config.speech_end_timeout ): voice_activity_timeout = ( cloud_speech_v2.StreamingRecognitionFeatures.VoiceActivityTimeout() ) if is_given(self._config.speech_start_timeout): voice_activity_timeout.speech_start_timeout = Duration( seconds=int(self._config.speech_start_timeout), nanos=int((self._config.speech_start_timeout % 1) * 1e9), ) if is_given(self._config.speech_end_timeout): voice_activity_timeout.speech_end_timeout = Duration( seconds=int(self._config.speech_end_timeout), nanos=int((self._config.speech_end_timeout % 1) * 1e9), ) return cloud_speech_v2.StreamingRecognitionConfig( config=cloud_speech_v2.RecognitionConfig( explicit_decoding_config=cloud_speech_v2.ExplicitDecodingConfig( encoding=cloud_speech_v2.ExplicitDecodingConfig.AudioEncoding.LINEAR16, sample_rate_hertz=self._config.sample_rate, audio_channel_count=1, ), adaptation=self._config.build_adaptation(), language_codes=self._config.languages, model=self._config.model, features=cloud_speech_v2.RecognitionFeatures( enable_automatic_punctuation=self._config.punctuate, enable_word_time_offsets=self._config.enable_word_time_offsets, enable_spoken_punctuation=self._config.spoken_punctuation, enable_word_confidence=self._config.enable_word_confidence, profanity_filter=self._config.profanity_filter, ), denoiser_config=self._config.denoiser_config if is_given(self._config.denoiser_config) else None, ), streaming_features=cloud_speech_v2.StreamingRecognitionFeatures( interim_results=self._config.interim_results, # Auto-enable voice activity events when voice_activity_timeout is specified, # as per Google API documentation requirements enable_voice_activity_events=self._config.enable_voice_activity_events or (voice_activity_timeout is not None), voice_activity_timeout=voice_activity_timeout, endpointing_sensitivity=getattr( cloud_speech_v2.StreamingRecognitionFeatures.EndpointingSensitivity, self._config.endpointing_sensitivity, ) if is_given(self._config.endpointing_sensitivity) else None, ), ) return cloud_speech_v1.StreamingRecognitionConfig( config=cloud_speech_v1.RecognitionConfig( encoding=cloud_speech_v1.RecognitionConfig.AudioEncoding.LINEAR16, sample_rate_hertz=self._config.sample_rate, audio_channel_count=1, adaptation=self._config.build_adaptation(), language_code=self._config.languages[0], alternative_language_codes=self._config.languages[1:], enable_word_time_offsets=self._config.enable_word_time_offsets, enable_word_confidence=self._config.enable_word_confidence, enable_automatic_punctuation=self._config.punctuate, enable_spoken_punctuation=self._config.spoken_punctuation, profanity_filter=self._config.profanity_filter, model=self._config.model, ), interim_results=self._config.interim_results, enable_voice_activity_events=self._config.enable_voice_activity_events, ) def _build_init_request( self, client: SpeechAsyncClientV2 | SpeechAsyncClientV1, ) -> cloud_speech_v2.StreamingRecognizeRequest | cloud_speech_v1.StreamingRecognizeRequest: if self._config.version == 2: return cloud_speech_v2.StreamingRecognizeRequest( recognizer=self._recognizer_cb(cast(SpeechAsyncClientV2, client)), streaming_config=self._streaming_config, ) return cloud_speech_v1.StreamingRecognizeRequest( streaming_config=self._streaming_config, ) def _build_audio_request( self, frame: rtc.AudioFrame, ) -> cloud_speech_v2.StreamingRecognizeRequest | cloud_speech_v1.StreamingRecognizeRequest: if self._config.version == 2: return cloud_speech_v2.StreamingRecognizeRequest(audio=frame.data.tobytes()) return cloud_speech_v1.StreamingRecognizeRequest(audio_content=frame.data.tobytes()) async def _run(self) -> None: audio_pushed = False # google requires a async generator when calling streaming_recognize # this function basically convert the queue into a async generator async def input_generator( client: SpeechAsyncClientV2 | SpeechAsyncClientV1, should_stop: asyncio.Event ) -> AsyncGenerator[ cloud_speech_v2.StreamingRecognizeRequest | cloud_speech_v1.StreamingRecognizeRequest, None, ]: nonlocal audio_pushed stop_task = asyncio.create_task(should_stop.wait()) frame_task: asyncio.Task[object] | None = None try: yield self._build_init_request(client) while True: # Race the next-frame await against should_stop so this generator # can exit even when no audio is flowing. Without this, on reconnect # the generator stays parked on _input_ch and pins the previous # gRPC streaming call, leaking it across iterations. frame_task = asyncio.create_task(self._input_ch.recv()) done, _ = await asyncio.wait( [frame_task, stop_task], return_when=asyncio.FIRST_COMPLETED ) if stop_task in done: return try: frame = frame_task.result() except ChanClosed: return finally: frame_task = None if isinstance(frame, rtc.AudioFrame): yield self._build_audio_request(frame) if not audio_pushed: audio_pushed = True except Exception: logger.exception("an error occurred while streaming input to google STT") finally: await utils.aio.gracefully_cancel( stop_task, *([frame_task] if frame_task is not None else []) ) async def process_stream( client: SpeechAsyncClientV2 | SpeechAsyncClientV1, stream: AsyncIterable[ cloud_speech_v2.StreamingRecognizeResponse | cloud_speech_v1.StreamingRecognizeResponse ], ) -> None: self._speaking = False last_usage_event_time: float = 0.0 async for resp in stream: if resp.speech_event_type == ( cloud_speech_v2.StreamingRecognizeResponse.SpeechEventType.SPEECH_ACTIVITY_BEGIN if self._config.version == 2 else cloud_speech_v1.StreamingRecognizeResponse.SpeechEventType.SPEECH_ACTIVITY_BEGIN ): self._event_ch.send_nowait( stt.SpeechEvent(type=stt.SpeechEventType.START_OF_SPEECH) ) self._speaking = True if ( resp.speech_event_type == ( cloud_speech_v2.StreamingRecognizeResponse.SpeechEventType.SPEECH_EVENT_TYPE_UNSPECIFIED if self._config.version == 2 else cloud_speech_v1.StreamingRecognizeResponse.SpeechEventType.SPEECH_EVENT_UNSPECIFIED ) and resp.results ): result = resp.results[0] speech_data = _streaming_recognize_response_to_speech_data( resp, min_confidence_threshold=self._config.min_confidence_threshold, start_time_offset=self.start_time_offset, ) if speech_data is None: continue if not result.is_final: self._event_ch.send_nowait( stt.SpeechEvent( type=stt.SpeechEventType.INTERIM_TRANSCRIPT, alternatives=[speech_data], ) ) else: self._event_ch.send_nowait( stt.SpeechEvent( type=stt.SpeechEventType.FINAL_TRANSCRIPT, alternatives=[speech_data], ) ) if time.time() - self._session_connected_at > _max_session_duration: logger.debug( "Google STT maximum connection time reached. Reconnecting..." ) self._pool.remove(client) if self._speaking: self._event_ch.send_nowait( stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH) ) self._speaking = False self._on_end_of_speech() self._reconnect_event.set() return if resp.speech_event_type == ( cloud_speech_v2.StreamingRecognizeResponse.SpeechEventType.SPEECH_ACTIVITY_END if self._config.version == 2 else cloud_speech_v1.StreamingRecognizeResponse.SpeechEventType.SPEECH_ACTIVITY_END ): self._event_ch.send_nowait( stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH) ) self._speaking = False self._on_end_of_speech() if (audio_duration := _get_audio_duration(resp, last_usage_event_time)) > 0: self._event_ch.send_nowait( stt.SpeechEvent( type=stt.SpeechEventType.RECOGNITION_USAGE, request_id=_get_request_id(resp), recognition_usage=stt.RecognitionUsage(audio_duration=audio_duration), ) ) last_usage_event_time += audio_duration while True: audio_pushed = False try: async with self._pool.connection(timeout=self._conn_options.timeout) as client: self._report_connection_acquired( self._pool.last_acquire_time, self._pool.last_connection_reused ) self._streaming_config = self._build_streaming_config() should_stop = asyncio.Event() stream = await client.streaming_recognize( requests=input_generator(client, should_stop), ) self._session_connected_at = time.time() process_stream_task = asyncio.create_task(process_stream(client, stream)) wait_reconnect_task = asyncio.create_task(self._reconnect_event.wait()) try: done, _ = await asyncio.wait( [process_stream_task, wait_reconnect_task], return_when=asyncio.FIRST_COMPLETED, ) for task in done: if task != wait_reconnect_task: task.result() if wait_reconnect_task not in done: break self._reconnect_event.clear() finally: should_stop.set() # Cancel the streaming RPC so its underlying call object releases # its read/write tasks and request iterator. Without this the # call (and the input_generator that yielded into it) stays # pinned across reconnects and leaks ~0.4 MB per cycle. with contextlib.suppress(Exception): cast(StreamStreamCall, stream).cancel() if not process_stream_task.done() and not wait_reconnect_task.done(): # try to gracefully stop the process_stream_task try: await asyncio.wait_for(process_stream_task, timeout=1.0) except asyncio.TimeoutError: pass await utils.aio.gracefully_cancel(process_stream_task, wait_reconnect_task) except DeadlineExceeded: raise APITimeoutError() from None except GoogleAPICallError as e: if e.code == 409: if audio_pushed: logger.debug("stream timed out, restarting.") else: raise APIStatusError( f"{e.message} {e.details}", status_code=e.code or -1 ) from e except Exception as e: raise APIConnectionError() from e def _duration_to_seconds(duration: Duration | timedelta) -> float: # Proto Plus may auto-convert Duration to timedelta; handle both. # https://proto-plus-python.readthedocs.io/en/latest/marshal.html if isinstance(duration, timedelta): return duration.total_seconds() return duration.seconds + duration.nanos / 1e9 def _get_start_time(word: cloud_speech_v2.WordInfo | cloud_speech_v1.WordInfo) -> float: if hasattr(word, "start_offset"): return _duration_to_seconds(word.start_offset) return _duration_to_seconds(word.start_time) def _get_end_time(word: cloud_speech_v2.WordInfo | cloud_speech_v1.WordInfo) -> float: if hasattr(word, "end_offset"): return _duration_to_seconds(word.end_offset) return _duration_to_seconds(word.end_time) def _recognize_response_to_speech_event( resp: cloud_speech_v2.RecognizeResponse | cloud_speech_v1.RecognizeResponse, ) -> stt.SpeechEvent: text = "" confidence = 0.0 for result in resp.results: text += result.alternatives[0].transcript confidence += result.alternatives[0].confidence alternatives = [] # Google STT may return empty results when spoken_lang != stt_lang if resp.results: try: start_time = _get_start_time(resp.results[0].alternatives[0].words[0]) end_time = _get_end_time(resp.results[-1].alternatives[0].words[-1]) except IndexError: # When enable_word_time_offsets=False, there are no "words" to access start_time = end_time = 0 confidence /= len(resp.results) lg = LanguageCode(resp.results[0].language_code) alternatives = [ stt.SpeechData( language=lg, start_time=start_time, end_time=end_time, confidence=confidence, text=text, words=[ TimedString( text=word.word, start_time=_get_start_time(word), end_time=_get_end_time(word), ) for word in resp.results[0].alternatives[0].words ] if resp.results[0].alternatives[0].words else None, ) ] return stt.SpeechEvent(type=stt.SpeechEventType.FINAL_TRANSCRIPT, alternatives=alternatives) @utils.log_exceptions(logger=logger) def _streaming_recognize_response_to_speech_data( resp: cloud_speech_v2.StreamingRecognizeResponse | cloud_speech_v1.StreamingRecognizeResponse, *, min_confidence_threshold: float, start_time_offset: float, ) -> stt.SpeechData | None: text = "" confidence = 0.0 final_result = None words: list[cloud_speech_v2.WordInfo | cloud_speech_v1.WordInfo] = [] for result in resp.results: if len(result.alternatives) == 0: continue else: if result.is_final: final_result = result break else: text += result.alternatives[0].transcript confidence += result.alternatives[0].confidence words.extend(result.alternatives[0].words) if final_result is not None: text = final_result.alternatives[0].transcript confidence = final_result.alternatives[0].confidence words = list(final_result.alternatives[0].words) lg = LanguageCode(final_result.language_code) else: confidence /= len(resp.results) if confidence < min_confidence_threshold: return None lg = LanguageCode(resp.results[0].language_code) if text == "" or not words: if text and not words: data = stt.SpeechData( language=lg, start_time=start_time_offset, end_time=start_time_offset, confidence=confidence, text=text, ) return data return None data = stt.SpeechData( language=lg, start_time=_get_start_time(words[0]) + start_time_offset, end_time=_get_end_time(words[-1]) + start_time_offset, confidence=confidence, text=text, words=[ TimedString( text=word.word, start_time=_get_start_time(word) + start_time_offset, end_time=_get_end_time(word) + start_time_offset, start_time_offset=start_time_offset, confidence=word.confidence, ) for word in words ], ) return data def _get_audio_duration( resp: cloud_speech_v2.StreamingRecognizeResponse | cloud_speech_v1.StreamingRecognizeResponse, last_usage_event_time: float, ) -> float: """Calculate the audio duration from the response. References: - https://docs.cloud.google.com/python/docs/reference/speech/latest/google.cloud.speech_v1.types.StreamingRecognizeResponse - https://docs.cloud.google.com/speech-to-text/docs/reference/rest/v2/StreamingRecognitionResult """ # total_billed_time is only set "if this is the last response in the stream" # use speech event time/offset before the last response is received if isinstance(resp, cloud_speech_v2.StreamingRecognizeResponse): if resp.metadata.total_billed_duration: return _duration_to_seconds(resp.metadata.total_billed_duration) - last_usage_event_time return _duration_to_seconds(resp.speech_event_offset) - last_usage_event_time if resp.total_billed_time: return _duration_to_seconds(resp.total_billed_time) - last_usage_event_time return _duration_to_seconds(resp.speech_event_time) - last_usage_event_time def _get_request_id( resp: cloud_speech_v2.StreamingRecognizeResponse | cloud_speech_v1.StreamingRecognizeResponse, ) -> str: if isinstance(resp, cloud_speech_v2.StreamingRecognizeResponse): return resp.metadata.request_id return str(resp.request_id)