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