TTS Cleanup and expose get audio (#79065)

homeassistant/components/tts/media_source.py

@@ -2,17 +2,18 @@
 from __future__ import annotations
 
 import mimetypes
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, TypedDict
 
 from yarl import URL
 
 from homeassistant.components.media_player import BrowseError, MediaClass
-from homeassistant.components.media_source.error import Unresolvable
-from homeassistant.components.media_source.models import (
+from homeassistant.components.media_source import (
     BrowseMediaSource,
     MediaSource,
     MediaSourceItem,
     PlayMedia,
+    Unresolvable,
+    generate_media_source_id as ms_generate_media_source_id,
 )
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.exceptions import HomeAssistantError
@@ -29,6 +30,75 @@ async def async_get_media_source(hass: HomeAssistant) -> TTSMediaSource:
     return TTSMediaSource(hass)
 
 
+@callback
+def generate_media_source_id(
+    hass: HomeAssistant,
+    message: str,
+    engine: str | None = None,
+    language: str | None = None,
+    options: dict | None = None,
+    cache: bool | None = None,
+) -> str:
+    """Generate a media source ID for text-to-speech."""
+    manager: SpeechManager = hass.data[DOMAIN]
+
+    if engine is not None:
+        pass
+    elif not manager.providers:
+        raise HomeAssistantError("No TTS providers available")
+    elif "cloud" in manager.providers:
+        engine = "cloud"
+    else:
+        engine = next(iter(manager.providers))
+
+    manager.process_options(engine, language, options)
+    params = {
+        "message": message,
+    }
+    if cache is not None:
+        params["cache"] = "true" if cache else "false"
+    if language is not None:
+        params["language"] = language
+    if options is not None:
+        params.update(options)
+
+    return ms_generate_media_source_id(
+        DOMAIN,
+        str(URL.build(path=engine, query=params)),
+    )
+
+
+class MediaSourceOptions(TypedDict):
+    """Media source options."""
+
+    engine: str
+    message: str
+    language: str | None
+    options: dict | None
+    cache: bool | None
+
+
+@callback
+def media_source_id_to_kwargs(media_source_id: str) -> MediaSourceOptions:
+    """Turn a media source ID into options."""
+    parsed = URL(media_source_id)
+    if "message" not in parsed.query:
+        raise Unresolvable("No message specified.")
+
+    options = dict(parsed.query)
+    kwargs: MediaSourceOptions = {
+        "engine": parsed.name,
+        "message": options.pop("message"),
+        "language": options.pop("language", None),
+        "options": options,
+        "cache": None,
+    }
+    if "cache" in options:
+        kwargs["cache"] = options.pop("cache") == "true"
+
+    return kwargs
+
+
 class TTSMediaSource(MediaSource):
     """Provide text-to-speech providers as media sources."""
 
@@ -41,24 +111,12 @@ class TTSMediaSource(MediaSource):
 
     async def async_resolve_media(self, item: MediaSourceItem) -> PlayMedia:
         """Resolve media to a url."""
-        parsed = URL(item.identifier)
-        if "message" not in parsed.query:
-            raise Unresolvable("No message specified.")
-
-        options = dict(parsed.query)
-        kwargs: dict[str, Any] = {
-            "engine": parsed.name,
-            "message": options.pop("message"),
-            "language": options.pop("language", None),
-            "options": options,
-        }
-        if "cache" in options:
-            kwargs["cache"] = options.pop("cache") == "true"
-
         manager: SpeechManager = self.hass.data[DOMAIN]
 
         try:
-            url = await manager.async_get_url_path(**kwargs)
+            url = await manager.async_get_url_path(
+                **media_source_id_to_kwargs(item.identifier)
+            )
         except HomeAssistantError as err:
             raise Unresolvable(str(err)) from err