/
/
/
1"""Helpers to discover, select and configure the AI/TTS engines exposed by plugins."""
2
3from __future__ import annotations
4
5import logging
6from typing import TYPE_CHECKING, Any
7
8from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
9from music_assistant_models.enums import ConfigEntryType, ProviderFeature, ProviderType
10
11from music_assistant.constants import MASS_LOGGER_NAME
12from music_assistant.models.plugin import (
13 AIEngine,
14 PluginEngine,
15 PluginProvider,
16 TTSEngine,
17)
18
19if TYPE_CHECKING:
20 from collections.abc import Callable, Coroutine
21
22 from music_assistant_models.config_entries import ConfigValueType
23
24 from music_assistant.mass import MusicAssistant
25 from music_assistant.models.provider import Provider
26
27LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.helpers.plugin_engines")
28
29
30async def get_ai_engines(mass: MusicAssistant) -> list[AIEngine]:
31 """
32 Return all AI engines currently exposed by the loaded plugins, in a stable order.
33
34 :param mass: The Music Assistant instance to query.
35 """
36 return await _collect_engines(
37 mass, ProviderFeature.AI_QUERY, lambda provider: provider.get_ai_engines()
38 )
39
40
41async def get_tts_engines(mass: MusicAssistant) -> list[TTSEngine]:
42 """
43 Return all TTS engines currently exposed by the loaded plugins, in a stable order.
44
45 :param mass: The Music Assistant instance to query.
46 """
47 return await _collect_engines(
48 mass, ProviderFeature.TTS, lambda provider: provider.get_tts_engines()
49 )
50
51
52def engine_display_name(engine: PluginEngine) -> str:
53 """
54 Return the name to show for an engine, naming the plugin it comes from.
55
56 :param engine: The engine to name. Engine names alone are ambiguous, since the same
57 name can be exposed by more than one plugin.
58 """
59 return f"{engine.provider.name} | {engine.name}"
60
61
62async def resolve_ai_engine(mass: MusicAssistant, selected: str | None) -> AIEngine | None:
63 """
64 Return the AI engine for a configured selection.
65
66 :param mass: The Music Assistant instance to query.
67 :param selected: The configured engine uid. Returns None when it is empty/unset or names
68 an engine that no longer exists - another engine is never substituted for it.
69 """
70 return _resolve(await get_ai_engines(mass), selected)
71
72
73async def resolve_tts_engine(mass: MusicAssistant, selected: str | None) -> TTSEngine | None:
74 """
75 Return the TTS engine for a configured selection.
76
77 :param mass: The Music Assistant instance to query.
78 :param selected: The configured engine uid. Returns None when it is empty/unset or names
79 an engine that no longer exists - another engine is never substituted for it.
80 """
81 return _resolve(await get_tts_engines(mass), selected)
82
83
84async def select_ai_engine(
85 provider: Provider, key: str, *, in_setup_data: bool = False
86) -> AIEngine | None:
87 """
88 Return the provider's AI engine, adopting the first available one when it has none yet.
89
90 A selection that was made explicitly is honoured or, when that engine no longer exists,
91 reported as missing - another engine is never substituted for it.
92
93 :param provider: The provider whose selection to read and, if unset, persist.
94 :param key: The config key holding the selected engine uid.
95 :param in_setup_data: Store in (and read from) the provider's setup_data instead of its
96 regular config values.
97 """
98 return _select_provider_engine(
99 provider, key, await get_ai_engines(provider.mass), in_setup_data=in_setup_data
100 )
101
102
103async def select_tts_engine(
104 provider: Provider, key: str, *, in_setup_data: bool = False
105) -> TTSEngine | None:
106 """
107 Return the provider's TTS engine, adopting the first available one when it has none yet.
108
109 A selection that was made explicitly is honoured or, when that engine no longer exists,
110 reported as missing - another engine is never substituted for it.
111
112 :param provider: The provider whose selection to read and, if unset, persist.
113 :param key: The config key holding the selected engine uid.
114 :param in_setup_data: Store in (and read from) the provider's setup_data instead of its
115 regular config values.
116 """
117 return _select_provider_engine(
118 provider, key, await get_tts_engines(provider.mass), in_setup_data=in_setup_data
119 )
120
121
122async def select_core_tts_engine(mass: MusicAssistant, domain: str, key: str) -> TTSEngine | None:
123 """
124 Return a core controller's TTS engine, adopting the first available one when it has none yet.
125
126 A selection that was made explicitly is honoured or, when that engine no longer exists,
127 reported as missing - another engine is never substituted for it.
128
129 :param mass: The Music Assistant instance to query.
130 :param domain: The domain of the core controller whose selection to read and, if unset, persist.
131 :param key: The config key holding the selected engine uid.
132 """
133 config = mass.config
134 return _select_engine(
135 await get_tts_engines(mass),
136 config.get_raw_core_config_value(domain, key),
137 lambda uid: config.set_raw_core_config_value(domain, key, uid),
138 domain,
139 key,
140 )
141
142
143async def create_ai_engine_config_entries(
144 mass: MusicAssistant,
145 key: str,
146 depends_on: str | None = None,
147 required: bool = False,
148 category: str = "features",
149) -> tuple[ConfigEntry, ...]:
150 """
151 Return the config entries letting the user pick an AI engine.
152
153 Adds a second (alert) entry with key ``<key>_unavailable`` when no engine is available.
154
155 :param mass: The Music Assistant instance to query.
156 :param key: The config entry key holding the selected engine uid.
157 :param depends_on: Optional key of the entry this picker should be shown for.
158 :param required: Reject a submission that leaves the picker empty. Only for setup flows;
159 an options entry must stay optional so a provider with no selection still loads.
160 :param category: The settings category to show the entries under.
161 """
162 return _create_engine_config_entries(
163 await get_ai_engines(mass),
164 key,
165 depends_on,
166 required,
167 category,
168 unavailable_translation_key="ai_engine_unavailable",
169 )
170
171
172async def create_tts_engine_config_entries(
173 mass: MusicAssistant,
174 key: str,
175 depends_on: str | None = None,
176 required: bool = False,
177 category: str = "features",
178) -> tuple[ConfigEntry, ...]:
179 """
180 Return the config entries letting the user pick a TTS engine.
181
182 Adds a second (alert) entry with key ``<key>_unavailable`` when no engine is available.
183
184 :param mass: The Music Assistant instance to query.
185 :param key: The config entry key holding the selected engine uid.
186 :param depends_on: Optional key of the entry this picker should be shown for.
187 :param required: Reject a submission that leaves the picker empty. Only for setup flows;
188 an options entry must stay optional so a provider with no selection still loads.
189 :param category: The settings category to show the entries under.
190 """
191 return _create_engine_config_entries(
192 await get_tts_engines(mass),
193 key,
194 depends_on,
195 required,
196 category,
197 unavailable_translation_key="tts_engine_unavailable",
198 )
199
200
201async def _collect_engines[EngineT: PluginEngine](
202 mass: MusicAssistant,
203 feature: ProviderFeature,
204 fetch: Callable[[PluginProvider], Coroutine[Any, Any, list[EngineT]]],
205) -> list[EngineT]:
206 """Collect the engines of every available plugin declaring the given feature."""
207 result: list[EngineT] = []
208 for provider in mass.get_providers_supporting_feature(feature, priority=(ProviderType.PLUGIN,)):
209 if not isinstance(provider, PluginProvider):
210 continue
211 try:
212 engines = await fetch(provider)
213 except Exception as err:
214 LOGGER.warning(
215 "Could not retrieve %s engines from %s: %s",
216 feature,
217 provider.instance_id,
218 err,
219 )
220 continue
221 result.extend(sorted(engines, key=lambda engine: engine.name))
222 return result
223
224
225def _resolve[EngineT: PluginEngine](engines: list[EngineT], selected: str | None) -> EngineT | None:
226 """Return the available engine matching a configured selection, if any."""
227 if not selected:
228 return None
229 return next((engine for engine in engines if engine.uid == selected), None)
230
231
232def _select_engine[EngineT: PluginEngine](
233 engines: list[EngineT],
234 stored: ConfigValueType,
235 persist: Callable[[str], None],
236 owner: str,
237 key: str,
238) -> EngineT | None:
239 """Return the stored engine, or persist and return the first available one."""
240 if isinstance(stored, str) and stored:
241 return _resolve(engines, stored)
242 if not engines:
243 return None
244 engine = engines[0]
245 persist(engine.uid)
246 LOGGER.debug("Selected engine %s as %s for %s", engine.uid, key, owner)
247 return engine
248
249
250def _select_provider_engine[EngineT: PluginEngine](
251 provider: Provider,
252 key: str,
253 engines: list[EngineT],
254 *,
255 in_setup_data: bool,
256) -> EngineT | None:
257 """Return the provider's stored engine, or persist and return the first available one."""
258 config = provider.mass.config
259 stored = (
260 config.get_provider_setup_value(provider.instance_id, key)
261 if in_setup_data
262 else config.get_raw_provider_config_value(provider.instance_id, key)
263 )
264 persist = provider._update_setup_data if in_setup_data else provider._update_config_value
265 return _select_engine(engines, stored, lambda uid: persist(key, uid), provider.instance_id, key)
266
267
268def _create_engine_config_entries(
269 engines: list[AIEngine] | list[TTSEngine],
270 key: str,
271 depends_on: str | None,
272 required: bool = False,
273 category: str = "features",
274 *,
275 unavailable_translation_key: str,
276) -> tuple[ConfigEntry, ...]:
277 """Build the picker (and unavailable alert) config entries for the given engines."""
278 entry = ConfigEntry(
279 key=key,
280 type=ConfigEntryType.STRING,
281 required=required,
282 # a concrete default would make the seeded selection equal to it and therefore
283 # not persisted (to_raw only stores values differing from the default)
284 default_value=None,
285 options=[
286 ConfigValueOption(engine.uid, title=engine_display_name(engine)) for engine in engines
287 ],
288 depends_on=depends_on,
289 category=category,
290 read_only=not engines,
291 )
292 if engines:
293 return (entry,)
294 return (
295 entry,
296 ConfigEntry(
297 key=f"{key}_unavailable",
298 type=ConfigEntryType.ALERT,
299 depends_on=depends_on,
300 category=category,
301 # the alert text does not depend on what the picker is used for, so every
302 # caller shares one source string instead of authoring its own copy
303 translation_key=unavailable_translation_key,
304 ),
305 )
306