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
6import logging
7from pathlib import Path
8from typing import TYPE_CHECKING
9
10from music_assistant_models.enums import StreamType
11from music_assistant_models.errors import InvalidDataError, MusicAssistantError
12
13from music_assistant.constants import MASS_LOGGER_NAME
14
15if TYPE_CHECKING:
16    from music_assistant_models.streamdetails import StreamDetails
17
18    from music_assistant.mass import MusicAssistant
19    from music_assistant.models.plugin import TTSEngine
20
21LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.helpers.tts")
22
23# last-resort guard so a wedged engine fails the call instead of hanging its caller.
24# Kept above the deadlines the engines apply themselves (120s in the OpenAI-compatible
25# providers), so their own, more specific error is the one that surfaces.
26TTS_QUERY_TIMEOUT_SECONDS = 180
27
28REMOTE_STREAM_SCHEMES = ("http://", "https://", "rtsp://", "rtmp://")
29
30
31async def query_tts_engine(
32    engine: TTSEngine,
33    message: str,
34    language: str | None = None,
35    timeout: float | None = None,
36) -> StreamDetails:
37    """
38    Render a message through a TTS engine.
39
40    :param engine: The TTS engine to speak the message.
41    :param message: The text to speak.
42    :param language: Optional language code, omit to use the engine's own default voice.
43    :param timeout: Seconds to wait for the engine, defaults to TTS_QUERY_TIMEOUT_SECONDS.
44        Lower it for a caller that is holding something up while it waits.
45    """
46    if timeout is None:
47        timeout = TTS_QUERY_TIMEOUT_SECONDS
48    try:
49        async with asyncio.timeout(timeout) as query_timeout:
50            return await engine.provider.get_tts_message(
51                message, language=language, engine_id=engine.id
52            )
53    except TimeoutError as err:
54        # expired() tells our own cap apart from a timeout raised inside the engine
55        if not query_timeout.expired():
56            raise
57        raise MusicAssistantError(
58            f"TTS engine '{engine.uid}' did not respond within {timeout}s"
59        ) from err
60
61
62async def query_tts_engine_with_language_fallback(
63    engine: TTSEngine,
64    message: str,
65    language: str | None = None,
66    timeout: float | None = None,
67    logger: logging.Logger | None = None,
68) -> StreamDetails:
69    """
70    Render a message through a TTS engine, retrying without the language if it is rejected.
71
72    :param engine: The TTS engine to speak the message.
73    :param message: The text to speak.
74    :param language: Optional language code, omit to use the engine's own default voice.
75    :param timeout: Seconds to wait for the engine, defaults to TTS_QUERY_TIMEOUT_SECONDS.
76        Lower it for a caller that is holding something up while it waits.
77    :param logger: Optional logger to report a rejected language on.
78    """
79    try:
80        return await query_tts_engine(engine, message, language, timeout)
81    except TimeoutError, MusicAssistantError:
82        # a timeout or our own structured failure is not a language rejection, so a
83        # language-less retry would not help and would only double the wait
84        raise
85    except Exception as err:
86        if language is None:
87            raise
88        # some engines reject a language they don't support; fall back to the engine's
89        # own default voice rather than losing the audio entirely
90        (logger or LOGGER).warning(
91            "TTS engine '%s' rejected language '%s' (%s), retrying with its default voice",
92            engine.uid,
93            language,
94            err,
95        )
96        return await query_tts_engine(engine, message, None, timeout)
97
98
99def resolve_tts_language(mass: MusicAssistant) -> str | None:
100    """
101    Return the language a TTS engine should speak in, as a hyphenated code like 'en-US'.
102
103    Returns None when no language is configured, leaving the engine on its own default voice.
104
105    :param mass: The Music Assistant instance holding the configured locale.
106    """
107    locale = mass.metadata.locale
108    return locale.replace("_", "-") if locale else None
109
110
111async def resolve_tts_stream_path(
112    engine: TTSEngine, stream_details: StreamDetails
113) -> tuple[str, StreamType]:
114    """
115    Return the playable path of a rendered clip and the way to stream it.
116
117    :param engine: The TTS engine that produced the clip, named in the error raised when it
118        did not return anything playable.
119    :param stream_details: The StreamDetails the engine returned.
120    """
121    path = str(stream_details.path or "").strip()
122    if path.startswith(REMOTE_STREAM_SCHEMES):
123        return path, StreamType.HTTP
124    if path and Path(path).is_absolute() and await asyncio.to_thread(Path(path).is_file):
125        return path, StreamType.LOCAL_FILE
126    raise InvalidDataError(
127        f"TTS engine '{engine.uid}' returned an unusable stream path: "
128        f"{path or '<empty>'}. StreamDetails.path must be a fetchable "
129        "http(s)/rtsp/rtmp URL or the absolute path of an existing local file."
130    )
131