/
/
/
1"""Model/base for a Provider implementation within Music Assistant."""
2
3from __future__ import annotations
4
5import asyncio
6import builtins
7import logging
8from typing import TYPE_CHECKING, Any, TypeVar, final, overload
9
10from music_assistant_models.config_entries import UI_ONLY, ConfigValueType
11from music_assistant_models.enums import ConfigEntryType, EventType
12from music_assistant_models.errors import ActionUnavailable, UnsupportedFeaturedException
13
14from music_assistant.constants import CONF_LOG_LEVEL, CONF_PROVIDERS, MASS_LOGGER_NAME
15
16if TYPE_CHECKING:
17 from async_upnp_client.utils import CaseInsensitiveDict
18 from music_assistant_models.config_entries import (
19 ConfigActionResult,
20 ConfigEntry,
21 ProviderConfig,
22 )
23 from music_assistant_models.enums import ProviderFeature, ProviderStage, ProviderType
24 from music_assistant_models.provider import ProviderManifest
25 from zeroconf import ServiceStateChange
26 from zeroconf.asyncio import AsyncServiceInfo
27
28 from music_assistant.helpers.json import SerializableType
29 from music_assistant.mass import MusicAssistant
30
31# TypeVar for config value type inference
32_ConfigValueT = TypeVar("_ConfigValueT", bound=ConfigValueType)
33
34
35class Provider:
36 """Base representation of a Provider implementation within Music Assistant."""
37
38 mass: MusicAssistant
39 manifest: ProviderManifest
40 config: ProviderConfig
41 # set to True in providers that capture a mass.streams address or port while loading,
42 # to have them reloaded onto the new one when the streamserver network changes
43 reload_on_streams_network_change: bool = False
44
45 def __init__(
46 self,
47 mass: MusicAssistant,
48 manifest: ProviderManifest,
49 config: ProviderConfig,
50 supported_features: set[ProviderFeature] | None = None,
51 ) -> None:
52 """Initialize MusicProvider."""
53 self.mass = mass
54 self.manifest = manifest
55 self.config = config
56 self._supported_features = supported_features or set()
57 self._set_log_level_from_config(config)
58 self.cache = mass.cache
59 self.available = False
60 # set by the controller once teardown of this provider starts, so work that is
61 # already in flight (e.g. a discovery running in a worker thread) can tell a
62 # provider on its way out apart from one that is not loaded yet
63 self.unloading = False
64 self.initialized = asyncio.Event()
65
66 @property
67 def supported_features(self) -> set[ProviderFeature]:
68 """Return the features supported by this Provider."""
69 # should not be overridden in normal circumstances
70 return self._supported_features
71
72 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
73 """
74 Return the (options) config entries to configure this provider instance.
75
76 Resolved on every load - before ``handle_async_init`` - as well as whenever the
77 options page is opened, so this may not read state that async init assigns. Read
78 the current values via ``self.config``/``self.get_config_value`` and the
79 capabilities via ``self.supported_features``. One-time setup input is collected by
80 the setup flow (see ``setup_flow.py``), not here. Include ``ConfigEntryType.ACTION``
81 entries for one-shot buttons and handle their presses in ``handle_config_action``.
82 """
83 return ()
84
85 async def handle_config_action(
86 self, action: str
87 ) -> tuple[ConfigEntry, ...] | ConfigActionResult | None:
88 """
89 Run the one-shot side effect for a pressed action button from this provider's options.
90
91 Override to run the side effect for each ``ConfigEntryType.ACTION`` entry this
92 provider declares. Return a ``ConfigActionResult`` to report the outcome (a message
93 to show and/or a url to open), or None when there is nothing to report. Raise to
94 report failure to the caller. Returning config entries re-renders the options page
95 with those entries instead.
96
97 :param action: The action id of the pressed button (an entry's ``action`` key).
98 """
99 raise ActionUnavailable(f"Unknown action: {action}")
100
101 async def handle_async_init(self) -> None:
102 """
103 Handle async initialization of the provider.
104
105 Runs after ``get_config_entries`` was already resolved, so state assigned here
106 is not available to it.
107 """
108
109 async def loaded_in_mass(self) -> None:
110 """Call after the provider has been loaded."""
111
112 async def unload(self, is_removed: bool = False) -> None:
113 """
114 Handle unload/close of the provider.
115
116 Called when provider is deregistered (e.g. MA exiting or config reloading).
117 is_removed will be set to True when the provider is removed from the configuration.
118 """
119
120 async def update_config(self, config: ProviderConfig, changed_keys: set[str]) -> None:
121 """
122 Handle logic when the config is updated.
123
124 Override this method in your provider implementation if you need
125 to perform any additional setup logic after the provider is registered and
126 the self.config was loaded, and whenever the config changes.
127
128 The default implementation reloads the provider on any config change
129 (except log-level-only changes), since provider reloads are lightweight
130 and most providers cache config values at setup time.
131 """
132 # always update the stored config so dynamic reads pick up new values
133 self.config = config
134
135 # update log level if changed
136 if f"values/{CONF_LOG_LEVEL}" in changed_keys or "name" in changed_keys:
137 self._set_log_level_from_config(config)
138
139 # reload if any non-log-level value keys changed
140 value_keys_changed = {
141 k for k in changed_keys if k.startswith("values/") and k != f"values/{CONF_LOG_LEVEL}"
142 }
143 if value_keys_changed:
144 self.logger.info(
145 "Config updated, reloading provider %s (instance_id=%s)",
146 self.domain,
147 self.instance_id,
148 )
149 # armed under the load path's task id so any (re)load starting before it fires
150 # cancels it
151 task_id = f"load_provider_{self.instance_id}"
152 self.mass.call_later(1, self.mass.load_provider_config, config, task_id=task_id)
153
154 async def get_diagnostics(self) -> dict[str, SerializableType] | None:
155 """
156 Return optional diagnostics info for this provider to include in diagnostics reports.
157
158 Return None (the default) when this provider has nothing to contribute.
159 Keep the returned data small, JSON serializable and free of sensitive values.
160 """
161 return None
162
163 async def on_mdns_service_state_change(
164 self, name: str, state_change: ServiceStateChange, info: AsyncServiceInfo | None
165 ) -> None:
166 """Handle MDNS service state callback."""
167
168 async def on_upnp_service_discovered(
169 self, search_target: str, discovery_info: CaseInsensitiveDict
170 ) -> None:
171 """Handle UPNP/SSDP discovery callback."""
172
173 @property
174 @final
175 def type(self) -> ProviderType:
176 """Return type of this provider."""
177 return self.manifest.type
178
179 @property
180 @final
181 def domain(self) -> str:
182 """Return domain for this provider."""
183 return self.manifest.domain
184
185 @property
186 @final
187 def instance_id(self) -> str:
188 """Return instance_id for this provider(instance)."""
189 return self.config.instance_id
190
191 @property
192 @final
193 def translation_owner(self) -> str:
194 """Return the "provider.<domain>" namespace this provider's translation strings resolve under."""
195 return f"provider.{self.domain}"
196
197 @property
198 @final
199 def name(self) -> str:
200 """Return (custom) friendly name for this provider instance."""
201 if self.config.name:
202 # always prefer user-set name from config
203 return self.config.name
204 return self.default_name
205
206 @property
207 @final
208 def default_name(self) -> str:
209 """Return a default friendly name for this provider instance."""
210 # create default name based on instance count
211 prov_confs = self.mass.config.get("providers", {}).values()
212 instances = [x["instance_id"] for x in prov_confs if x["domain"] == self.domain]
213 if len(instances) <= 1:
214 # only one instance (or no instances yet at all) - return provider name
215 return self.manifest.name
216 instance_name_postfix = self.instance_name_postfix
217 if not instance_name_postfix:
218 # default implementation - simply use the instance number/index
219 instance_name_postfix = str(instances.index(self.instance_id) + 1)
220 # append instance name to provider name
221 return f"{self.manifest.name} [{instance_name_postfix}]"
222
223 @property
224 def instance_name_postfix(self) -> str | None:
225 """Return a (default) instance name postfix for this provider instance."""
226 return None
227
228 @property
229 @final
230 def stage(self) -> ProviderStage:
231 """Return the stage of this provider."""
232 return self.manifest.stage
233
234 def unload_with_error(self, error: str | Exception) -> None:
235 """
236 Unload this provider and record an error for the user to act on.
237
238 :param error: The originating exception (preferred, so its error code and localized
239 message are preserved) or a plain string for a generic error message.
240 """
241 self.mass.call_later(1, self.mass.unload_provider_with_error, self.instance_id, error)
242
243 def to_dict(self) -> dict[str, Any]:
244 """Return Provider(instance) as serializable dict."""
245 return {
246 "type": self.type.value,
247 "domain": self.domain,
248 "name": self.name,
249 "instance_id": self.instance_id,
250 "supported_features": [x.value for x in self.supported_features],
251 "available": self.available,
252 "is_streaming_provider": getattr(self, "is_streaming_provider", None),
253 "lookup_key": self.instance_id, # include for backwards compatibility
254 }
255
256 def supports_feature(self, feature: ProviderFeature) -> bool:
257 """Return True if this provider supports the given feature."""
258 return feature in self.supported_features
259
260 def check_feature(self, feature: ProviderFeature) -> None:
261 """Check if this provider supports the given feature."""
262 if not self.supports_feature(feature):
263 raise UnsupportedFeaturedException(
264 f"Provider {self.name} does not support feature {feature.name}"
265 )
266
267 @final
268 def signal_provider_event(self, data: SerializableType, sub_scope: str | None = None) -> None:
269 """
270 Signal a custom provider event to all subscribers (e.g. connected clients).
271
272 Emits a PROVIDER_EVENT with this provider's instance_id as object_id,
273 optionally suffixed with /sub_scope to allow clients to distinguish
274 multiple event streams from the same provider.
275
276 :param data: The JSON serializable event payload, defined by the provider.
277 :param sub_scope: Optional sub scope to append to the object_id.
278 """
279 object_id = f"{self.instance_id}/{sub_scope}" if sub_scope else self.instance_id
280 self.mass.signal_event(EventType.PROVIDER_EVENT, object_id=object_id, data=data)
281
282 @overload
283 def get_config_value(
284 self, key: str, default: _ConfigValueT, *, return_type: builtins.type[_ConfigValueT] = ...
285 ) -> _ConfigValueT: ...
286
287 @overload
288 def get_config_value(
289 self, key: str, default: ConfigValueType = ..., *, return_type: builtins.type[_ConfigValueT]
290 ) -> _ConfigValueT: ...
291
292 @overload
293 def get_config_value(
294 self, key: str, default: ConfigValueType = ..., *, return_type: None = ...
295 ) -> ConfigValueType: ...
296
297 def get_config_value(
298 self,
299 key: str,
300 default: ConfigValueType = None,
301 *,
302 return_type: builtins.type[_ConfigValueT | ConfigValueType] | None = None,
303 ) -> _ConfigValueT | ConfigValueType:
304 """
305 Return the current persisted config value for this provider.
306
307 Falls back to the active config entry value or default when no value is persisted.
308
309 :param key: The config key to retrieve.
310 :param default: Value to return when the key is not present in the active config.
311 :param return_type: Optional type hint for type inference (e.g., str, int, bool).
312 Note: This parameter is used purely for static type checking and does not
313 perform runtime type validation. Callers are responsible for ensuring the
314 specified type matches the actual config value type.
315 """
316 if (entry := self.config.values.get(key)) is None:
317 return self.config.get_value(key, default)
318 if entry.type in UI_ONLY:
319 # a display-only entry holds label text rather than a value, so reading
320 # through it would shadow the caller's default
321 return default
322 value = self.mass.config.get_raw_provider_config_value(self.instance_id, key)
323 if value is None:
324 return self.config.get_value(key, default)
325 if entry.type == ConfigEntryType.SECURE_STRING:
326 assert isinstance(value, str)
327 return self.mass.config.decrypt_string(value)
328 return value
329
330 def get_setup_value(self, key: str, default: ConfigValueType = None) -> ConfigValueType:
331 """
332 Return a value collected by this provider's setup flow (from setup_data).
333
334 Encrypted (string) values are decrypted transparently. When the key is not
335 present in setup_data, the active config entry value or the given default is
336 returned.
337
338 :param key: The setup data key to retrieve.
339 :param default: Value to return when the key is not present anywhere.
340 """
341 setup_data = self.mass.config.get(f"{CONF_PROVIDERS}/{self.instance_id}/setup_data") or {}
342 if key in setup_data:
343 value = setup_data[key]
344 return self.mass.config.decrypt_string(value) if isinstance(value, str) else value
345 return self.get_config_value(key, default)
346
347 def _update_setup_data(self, key: str, value: ConfigValueType, immediate: bool = True) -> None:
348 """
349 Update a single setup_data value for this provider (e.g. a rotated auth token).
350
351 :param key: The setup data key to update.
352 :param value: The new value; strings are encrypted at rest.
353 :param immediate: Persist to disk right away (the default) instead of on the
354 debounced save timer, so a critical value survives a crash.
355 """
356 if not self.mass.config.get(f"{CONF_PROVIDERS}/{self.instance_id}"):
357 # only allow setting setup data if the main config entry exists
358 msg = f"Invalid provider instance: {self.instance_id}"
359 raise KeyError(msg)
360 stored_value = self.mass.config.encrypt_string(value) if isinstance(value, str) else value
361 self.mass.config.set(
362 f"{CONF_PROVIDERS}/{self.instance_id}/setup_data/{key}",
363 stored_value,
364 immediate=immediate,
365 )
366 # keep the in-memory config copy in sync with storage
367 self.config.setup_data[key] = stored_value
368
369 def _update_config_value(
370 self, key: str, value: ConfigValueType, encrypted: bool = False, immediate: bool = False
371 ) -> None:
372 """
373 Update a config value.
374
375 :param immediate: Persist to disk right away instead of on the debounced save timer;
376 use for critical values (e.g. a rotated auth token) that must survive a crash.
377 """
378 self.mass.config.set_raw_provider_config_value(
379 self.instance_id, key, value, encrypted=encrypted, immediate=immediate
380 )
381 if (entry := self.config.values.get(key)) is not None:
382 entry.value = self.mass.config.get_raw_provider_config_value(self.instance_id, key)
383
384 def _set_log_level_from_config(self, config: ProviderConfig) -> None:
385 """Set log level from config."""
386 mass_logger = logging.getLogger(MASS_LOGGER_NAME)
387 # self.name is only available after async_init. Otherwise we run into a race condition.
388 # see https://github.com/music-assistant/support/issues/4801
389 logging_name = self.domain
390 if getattr(self, "available", False):
391 # async_init completed
392 logging_name = self.name
393 self.logger = mass_logger.getChild(logging_name)
394 # fall back to the entry's own default: a config that reaches us without its
395 # entries resolved must not take the whole provider down over a log level
396 log_level = str(config.get_value(CONF_LOG_LEVEL) or "GLOBAL")
397 if log_level == "GLOBAL":
398 self.logger.setLevel(mass_logger.level)
399 else:
400 self.logger.setLevel(log_level)
401 self.logger.debug("Log level configured to %s", log_level)
402