update

homeassistant/components/google_cloud/helpers.py

@@ -162,4 +162,4 @@ def validate_service_account_info(info: Mapping[str, str]) -> None:
         ValueError: If the info is not in the expected format.
 
     """
-    Credentials.from_service_account_info(info)  # type: ignore[no-untyped-call]
+    Credentials.from_service_account_info(info)  # type:ignore[no-untyped-call]

homeassistant/components/google_cloud/tts.py

@@ -75,7 +75,6 @@ async def async_get_engine(
         _LOGGER.error("Error from calling list_voices: %s", err)
         return None
     return GoogleCloudTTSProvider(
-        hass,
         client,
         voices,
         config.get(CONF_LANG, DEFAULT_LANG),
@@ -117,28 +116,17 @@ async def async_setup_entry(
     )
 
 
-class GoogleCloudTTSEntity(TextToSpeechEntity):
-    """The Google Cloud TTS entity."""
+class BaseGoogleCloudProvider:
+    """The Google Cloud TTS base provider."""
 
     def __init__(
         self,
-        entry: ConfigEntry,
         client: texttospeech.TextToSpeechAsyncClient,
         voices: dict[str, list[str]],
         language: str,
         options_schema: vol.Schema,
     ) -> None:
-        """Init Google Cloud TTS entity."""
-        self._attr_unique_id = f"{entry.entry_id}-tts"
-        self._attr_name = entry.title
-        self._attr_device_info = dr.DeviceInfo(
-            identifiers={(DOMAIN, entry.entry_id)},
-            name=entry.title,
-            manufacturer="Google",
-            model="Cloud",
-            entry_type=dr.DeviceEntryType.SERVICE,
-        )
-        self._entry = entry
+        """Init Google Cloud TTS base provider."""
         self._client = client
         self._voices = voices
         self._language = language
@@ -171,129 +159,116 @@ def async_get_supported_voices(self, language: str) -> list[Voice] | None:
             return None
         return [Voice(voice, voice) for voice in voices]
 
+    async def _async_get_tts_audio(
+        self,
+        message: str,
+        language: str,
+        options: dict[str, Any],
+    ) -> TtsAudioType:
+        """Load TTS from Google Cloud."""
+        try:
+            options = self._options_schema(options)
+        except vol.Invalid as err:
+            _LOGGER.error("Error: %s when validating options: %s", err, options)
+            return None, None
+
+        encoding: texttospeech.AudioEncoding = texttospeech.AudioEncoding[
+            options[CONF_ENCODING]
+        ]  # type: ignore[misc]
+        gender: texttospeech.SsmlVoiceGender | None = texttospeech.SsmlVoiceGender[
+            options[CONF_GENDER]
+        ]  # type: ignore[misc]
+        voice = options[CONF_VOICE]
+        if voice:
+            gender = None
+            if not voice.startswith(language):
+                language = voice[:5]
+
+        request = texttospeech.SynthesizeSpeechRequest(
+            input=texttospeech.SynthesisInput(**{options[CONF_TEXT_TYPE]: message}),
+            voice=texttospeech.VoiceSelectionParams(
+                language_code=language,
+                ssml_gender=gender,
+                name=voice,
+            ),
+            audio_config=texttospeech.AudioConfig(
+                audio_encoding=encoding,
+                speaking_rate=options[CONF_SPEED],
+                pitch=options[CONF_PITCH],
+                volume_gain_db=options[CONF_GAIN],
+                effects_profile_id=options[CONF_PROFILES],
+            ),
+        )
+
+        response = await self._client.synthesize_speech(request, timeout=10)
+
+        if encoding == texttospeech.AudioEncoding.MP3:
+            extension = "mp3"
+        elif encoding == texttospeech.AudioEncoding.OGG_OPUS:
+            extension = "ogg"
+        else:
+            extension = "wav"
+
+        return extension, response.audio_content
+
+
+class GoogleCloudTTSEntity(BaseGoogleCloudProvider, TextToSpeechEntity):
+    """The Google Cloud TTS entity."""
+
+    def __init__(
+        self,
+        entry: ConfigEntry,
+        client: texttospeech.TextToSpeechAsyncClient,
+        voices: dict[str, list[str]],
+        language: str,
+        options_schema: vol.Schema,
+    ) -> None:
+        """Init Google Cloud TTS entity."""
+        super().__init__(client, voices, language, options_schema)
+        self._attr_unique_id = f"{entry.entry_id}-tts"
+        self._attr_name = entry.title
+        self._attr_device_info = dr.DeviceInfo(
+            identifiers={(DOMAIN, entry.entry_id)},
+            name=entry.title,
+            manufacturer="Google",
+            model="Cloud",
+            entry_type=dr.DeviceEntryType.SERVICE,
+        )
+        self._entry = entry
+
     async def async_get_tts_audio(
         self, message: str, language: str, options: dict[str, Any]
     ) -> TtsAudioType:
         """Load TTS from Google Cloud."""
         try:
-            return await _async_get_tts_audio(
-                message, language, options, self._client, self._options_schema
-            )
+            return await self._async_get_tts_audio(message, language, options)
         except GoogleAPIError as err:
             _LOGGER.error("Error occurred during Google Cloud TTS call: %s", err)
             if isinstance(err, Unauthenticated):
                 self._entry.async_start_reauth(self.hass)
             return None, None
 
 
-class GoogleCloudTTSProvider(Provider):
+class GoogleCloudTTSProvider(BaseGoogleCloudProvider, Provider):
     """The Google Cloud TTS API provider."""
 
     def __init__(
         self,
-        hass: HomeAssistant,
         client: texttospeech.TextToSpeechAsyncClient,
         voices: dict[str, list[str]],
         language: str,
         options_schema: vol.Schema,
     ) -> None:
         """Init Google Cloud TTS service."""
-        self.hass = hass
+        super().__init__(client, voices, language, options_schema)
         self.name = "Google Cloud TTS"
-        self._client = client
-        self._voices = voices
-        self._language = language
-        self._options_schema = options_schema
-
-    @property
-    def supported_languages(self) -> list[str]:
-        """Return a list of supported languages."""
-        return list(self._voices)
-
-    @property
-    def default_language(self) -> str:
-        """Return the default language."""
-        return self._language
-
-    @property
-    def supported_options(self) -> list[str]:
-        """Return a list of supported options."""
-        return [option.schema for option in self._options_schema.schema]
-
-    @property
-    def default_options(self) -> dict[str, Any]:
-        """Return a dict including default options."""
-        return cast(dict[str, Any], self._options_schema({}))
-
-    @callback
-    def async_get_supported_voices(self, language: str) -> list[Voice] | None:
-        """Return a list of supported voices for a language."""
-        if not (voices := self._voices.get(language)):
-            return None
-        return [Voice(voice, voice) for voice in voices]
 
     async def async_get_tts_audio(
         self, message: str, language: str, options: dict[str, Any]
     ) -> TtsAudioType:
         """Load TTS from Google Cloud."""
         try:
-            return await _async_get_tts_audio(
-                message, language, options, self._client, self._options_schema
-            )
+            return await self._async_get_tts_audio(message, language, options)
         except GoogleAPIError as err:
             _LOGGER.error("Error occurred during Google Cloud TTS call: %s", err)
             return None, None
-
-
-async def _async_get_tts_audio(
-    message: str,
-    language: str,
-    options: dict[str, Any],
-    client: texttospeech.TextToSpeechAsyncClient,
-    options_schema: vol.Schema,
-) -> TtsAudioType:
-    """Load TTS from Google Cloud."""
-    try:
-        options = options_schema(options)
-    except vol.Invalid as err:
-        _LOGGER.error("Error: %s when validating options: %s", err, options)
-        return None, None
-
-    encoding: texttospeech.AudioEncoding = texttospeech.AudioEncoding[
-        options[CONF_ENCODING]
-    ]  # type: ignore[misc]
-    gender: texttospeech.SsmlVoiceGender | None = texttospeech.SsmlVoiceGender[
-        options[CONF_GENDER]
-    ]  # type: ignore[misc]
-    voice = options[CONF_VOICE]
-    if voice:
-        gender = None
-        if not voice.startswith(language):
-            language = voice[:5]
-
-    request = texttospeech.SynthesizeSpeechRequest(
-        input=texttospeech.SynthesisInput(**{options[CONF_TEXT_TYPE]: message}),
-        voice=texttospeech.VoiceSelectionParams(
-            language_code=language,
-            ssml_gender=gender,
-            name=voice,
-        ),
-        audio_config=texttospeech.AudioConfig(
-            audio_encoding=encoding,
-            speaking_rate=options[CONF_SPEED],
-            pitch=options[CONF_PITCH],
-            volume_gain_db=options[CONF_GAIN],
-            effects_profile_id=options[CONF_PROFILES],
-        ),
-    )
-
-    response = await client.synthesize_speech(request, timeout=10)
-
-    if encoding == texttospeech.AudioEncoding.MP3:
-        extension = "mp3"
-    elif encoding == texttospeech.AudioEncoding.OGG_OPUS:
-        extension = "ogg"
-    else:
-        extension = "wav"
-
-    return extension, response.audio_content