music-assistant-server

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