/
/
/
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, Any
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
31class TTSLanguageNotSupportedError(MusicAssistantError):
32 """The TTS engine rejected (or likely rejected) the requested language."""
33
34
35async def query_tts_engine(
36 engine: TTSEngine,
37 message: str,
38 language: str | None = None,
39 timeout: float | None = None,
40 options: dict[str, Any] | None = None,
41) -> StreamDetails:
42 """
43 Render a message through a TTS engine.
44
45 :param engine: The TTS engine to speak the message.
46 :param message: The text to speak.
47 :param language: Optional language code, omit to use the engine's own default voice.
48 :param timeout: Seconds to wait for the engine, defaults to TTS_QUERY_TIMEOUT_SECONDS.
49 Lower it for a caller that is holding something up while it waits.
50 :param options: Optional integration-specific TTS options, passed through to the
51 engine as-is. Ignored by engines that have none.
52 """
53 if timeout is None:
54 timeout = TTS_QUERY_TIMEOUT_SECONDS
55 try:
56 async with asyncio.timeout(timeout) as query_timeout:
57 return await engine.provider.get_tts_message(
58 message, language=language, engine_id=engine.id, options=options
59 )
60 except TimeoutError as err:
61 # expired() tells our own cap apart from a timeout raised inside the engine
62 if not query_timeout.expired():
63 raise
64 raise MusicAssistantError(
65 f"TTS engine '{engine.uid}' did not respond within {timeout}s"
66 ) from err
67
68
69async def query_tts_engine_with_language_fallback(
70 engine: TTSEngine,
71 message: str,
72 language: str | None = None,
73 timeout: float | None = None,
74 logger: logging.Logger | None = None,
75 options: dict[str, Any] | None = None,
76) -> StreamDetails:
77 """
78 Render a message through a TTS engine, retrying without the language if the engine rejects it.
79
80 :param engine: The TTS engine to speak the message.
81 :param message: The text to speak.
82 :param language: Optional language code, omit to use the engine's own default voice.
83 :param timeout: Seconds to wait for the engine, defaults to TTS_QUERY_TIMEOUT_SECONDS.
84 Lower it for a caller that is holding something up while it waits.
85 :param logger: Optional logger to report a rejected language on.
86 :param options: Optional integration-specific TTS options, passed through to the
87 engine as-is. Ignored by engines that have none.
88 """
89 try:
90 return await query_tts_engine(engine, message, language, timeout, options)
91 except TTSLanguageNotSupportedError as err:
92 if not language:
93 raise
94 # the error message carries the classifier's certainty, so it is logged as-is
95 (logger or LOGGER).warning("%s, retrying with the engine's default voice", err)
96 return await query_tts_engine(engine, message, None, timeout, options)
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