music-assistant-server

1"""Helpers to render text into playable speech audio through the plugin TTS engines."""
2
3from __future__ import annotations
4
5import asyncio
6from pathlib import Path
7from typing import TYPE_CHECKING
8
9from music_assistant_models.enums import StreamType
10from music_assistant_models.errors import InvalidDataError, MusicAssistantError
11
12if TYPE_CHECKING:
13    from music_assistant_models.streamdetails import StreamDetails
14
15    from music_assistant.models.plugin import TTSEngine
16
17# last-resort guard so a wedged engine fails the call instead of hanging its caller.
18# Kept above the deadlines the engines apply themselves (120s in the OpenAI-compatible
19# providers), so their own, more specific error is the one that surfaces.
20TTS_QUERY_TIMEOUT_SECONDS = 180
21
22REMOTE_STREAM_SCHEMES = ("http://", "https://", "rtsp://", "rtmp://")
23
24
25async def query_tts_engine(
26    engine: TTSEngine,
27    message: str,
28    language: str | None = None,
29    timeout: float | None = None,
30) -> StreamDetails:
31    """
32    Render a message through a TTS engine.
33
34    :param engine: The TTS engine to speak the message.
35    :param message: The text to speak.
36    :param language: Optional language code, omit to use the engine's own default voice.
37    :param timeout: Seconds to wait for the engine, defaults to TTS_QUERY_TIMEOUT_SECONDS.
38        Lower it for a caller that is holding something up while it waits.
39    """
40    if timeout is None:
41        timeout = TTS_QUERY_TIMEOUT_SECONDS
42    try:
43        async with asyncio.timeout(timeout) as query_timeout:
44            return await engine.provider.get_tts_message(
45                message, language=language, engine_id=engine.id
46            )
47    except TimeoutError as err:
48        # expired() tells our own cap apart from a timeout raised inside the engine
49        if not query_timeout.expired():
50            raise
51        raise MusicAssistantError(
52            f"TTS engine '{engine.uid}' did not respond within {timeout}s"
53        ) from err
54
55
56async def resolve_tts_stream_path(
57    engine: TTSEngine, stream_details: StreamDetails
58) -> tuple[str, StreamType]:
59    """
60    Return the playable path of a rendered clip and the way to stream it.
61
62    :param engine: The TTS engine that produced the clip, named in the error raised when it
63        did not return anything playable.
64    :param stream_details: The StreamDetails the engine returned.
65    """
66    path = str(stream_details.path or "").strip()
67    if path.startswith(REMOTE_STREAM_SCHEMES):
68        return path, StreamType.HTTP
69    if path and Path(path).is_absolute() and await asyncio.to_thread(Path(path).is_file):
70        return path, StreamType.LOCAL_FILE
71    raise InvalidDataError(
72        f"TTS engine '{engine.uid}' returned an unusable stream path: "
73        f"{path or '<empty>'}. StreamDetails.path must be a fetchable "
74        "http(s)/rtsp/rtmp URL or the absolute path of an existing local file."
75    )
76