Expose on-disk file path when resolving TTS media source (#172884)

Co-authored-by: Claude <noreply@anthropic.com>

homeassistant/components/tts/__init__.py

@@ -574,23 +574,43 @@ class ResultStream:
         """Override the TTS stream with a different media path."""
         self._override_media_path = Path(media_path)
 
+    @property
+    def _needs_conversion(self) -> bool:
+        """Return if the result requires conversion to a preferred format."""
+        return any(
+            self.options.get(option) is not None
+            for option in (
+                ATTR_PREFERRED_FORMAT,
+                ATTR_PREFERRED_SAMPLE_RATE,
+                ATTR_PREFERRED_SAMPLE_CHANNELS,
+                ATTR_PREFERRED_SAMPLE_BYTES,
+            )
+        )
+
+    @callback
+    def async_get_media_path(self) -> Path | None:
+        """Return the path to the result on disk, if available."""
+        if self._override_media_path is not None:
+            # An override that needs conversion no longer matches the file on
+            # disk, so the result is only available through the stream.
+            if self._needs_conversion:
+                return None
+            return self._override_media_path
+
+        if not self.use_file_cache or not self._result_cache.done():
+            return None
+
+        return self._manager.async_get_cache_file_path(
+            self._result_cache.result().cache_key
+        )
+
     async def _async_stream_override_result(self) -> AsyncGenerator[bytes]:
         """Get the stream of the overridden result."""
         assert self._override_media_path is not None
 
         preferred_format = self.options.get(ATTR_PREFERRED_FORMAT)
-        to_sample_rate = self.options.get(ATTR_PREFERRED_SAMPLE_RATE)
-        to_sample_channels = self.options.get(ATTR_PREFERRED_SAMPLE_CHANNELS)
-        to_sample_bytes = self.options.get(ATTR_PREFERRED_SAMPLE_BYTES)
 
-        needs_conversion = (
-            (preferred_format is not None)
-            or (to_sample_rate is not None)
-            or (to_sample_channels is not None)
-            or (to_sample_bytes is not None)
-        )
-
-        if not needs_conversion:
+        if not self._needs_conversion:
             # Read file directly (no conversion)
             yield await self.hass.async_add_executor_job(
                 self._override_media_path.read_bytes
@@ -749,6 +769,13 @@ class SpeechManager:
         self.file_cache.clear()
         await task
 
+    @callback
+    def async_get_cache_file_path(self, cache_key: str) -> Path | None:
+        """Return the path to a cached TTS file, if it is in the file cache."""
+        if not (filename := self.file_cache.get(cache_key)):
+            return None
+        return Path(self.cache_dir) / filename
+
     @callback
     def async_register_legacy_engine(
         self, engine: str, provider: Provider, config: ConfigType

homeassistant/components/tts/media_source.py

@@ -150,7 +150,9 @@ class TTSMediaSource(MediaSource):
         if stream is None:
             raise Unresolvable("Stream not found")
 
-        return PlayMedia(stream.url, stream.content_type)
+        return PlayMedia(
+            stream.url, stream.content_type, path=stream.async_get_media_path()
+        )
 
     async def async_browse_media(
         self,

tests/components/tts/test_init.py

@@ -2136,6 +2136,10 @@ async def test_stream_override(
         wav_file.seek(0)
 
         stream.async_override_result(wav_file.name)
+
+        # An override without conversion is available directly on disk.
+        assert stream.async_get_media_path() == Path(wav_file.name)
+
         result_data = b"".join([chunk async for chunk in stream.async_stream_result()])
 
     # Verify the result
@@ -2175,6 +2179,11 @@ async def test_stream_override_with_conversion(
 
         wav_file.seek(0)
         stream.async_override_result(wav_file.name)
+
+        # An override that needs conversion no longer matches the file on disk,
+        # so no path is exposed.
+        assert stream.async_get_media_path() is None
+
         result_data = b"".join([chunk async for chunk in stream.async_stream_result()])
 
     # Verify the result has the preferred format

tests/components/tts/test_media_source.py

@@ -1,6 +1,7 @@
 """Tests for TTS media source."""
 
 from http import HTTPStatus
+from pathlib import Path
 import re
 from unittest.mock import MagicMock
 
@@ -204,6 +205,7 @@ async def test_resolving(
     media = await media_source.async_resolve_media(hass, stream.media_source_id, None)
     assert media.url == stream.url
     assert media.mime_type == stream.content_type
+    assert media.path is None
 
     with pytest.raises(media_source.Unresolvable):
         await media_source.async_resolve_media(
@@ -211,6 +213,55 @@ async def test_resolving(
         )
 
 
+@pytest.mark.parametrize(
+    "mock_tts_entity",
+    [MSEntity(DEFAULT_LANG)],
+)
+async def test_resolving_sets_path_when_cached_on_disk(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    mock_tts_cache_dir: Path,
+    mock_tts_entity: MSEntity,
+) -> None:
+    """Test resolving exposes the on-disk file path once cached."""
+    await mock_config_entry_setup(hass, mock_tts_entity)
+
+    media_id = "media-source://tts/tts.test?message=Hello%20World&cache=true"
+
+    # Generate and persist the file to disk.
+    assert await retrieve_media(hass, hass_client, media_id) == HTTPStatus.OK
+    await hass.async_block_till_done(wait_background_tasks=True)
+
+    # Resolving now exposes the cached file on disk.
+    media = await media_source.async_resolve_media(hass, media_id, None)
+    assert media.url.startswith("/api/tts_proxy/")
+    assert media.path is not None
+    assert media.path.parent == mock_tts_cache_dir
+    assert media.path.is_file()
+
+
+@pytest.mark.parametrize(
+    "mock_tts_entity",
+    [MSEntity(DEFAULT_LANG)],
+)
+async def test_resolving_no_path_without_file_cache(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    mock_tts_cache_dir: Path,
+    mock_tts_entity: MSEntity,
+) -> None:
+    """Test resolving does not expose a path when file caching is disabled."""
+    await mock_config_entry_setup(hass, mock_tts_entity)
+
+    media_id = "media-source://tts/tts.test?message=Hello%20World&cache=false"
+
+    assert await retrieve_media(hass, hass_client, media_id) == HTTPStatus.OK
+    await hass.async_block_till_done(wait_background_tasks=True)
+
+    media = await media_source.async_resolve_media(hass, media_id, None)
+    assert media.path is None
+
+
 @pytest.mark.parametrize(
     ("mock_provider", "mock_tts_entity"),
     [(MSProvider(DEFAULT_LANG), MSEntity(DEFAULT_LANG))],