music-assistant-server

1"""Model/base for a Provider implementation within Music Assistant."""
2
3from __future__ import annotations
4
5import logging
6from typing import TYPE_CHECKING, Any, final
7
8from music_assistant_models.errors import UnsupportedFeaturedException
9
10from music_assistant.constants import CONF_LOG_LEVEL, MASS_LOGGER_NAME
11
12if TYPE_CHECKING:
13    from music_assistant_models.config_entries import ProviderConfig
14    from music_assistant_models.enums import ProviderFeature, ProviderStage, ProviderType
15    from music_assistant_models.provider import ProviderManifest
16    from zeroconf import ServiceStateChange
17    from zeroconf.asyncio import AsyncServiceInfo
18
19    from music_assistant.mass import MusicAssistant
20
21
22class Provider:
23    """Base representation of a Provider implementation within Music Assistant."""
24
25    def __init__(
26        self,
27        mass: MusicAssistant,
28        manifest: ProviderManifest,
29        config: ProviderConfig,
30        supported_features: set[ProviderFeature] | None = None,
31    ) -> None:
32        """Initialize MusicProvider."""
33        self.mass = mass
34        self.manifest = manifest
35        self.config = config
36        self._supported_features = supported_features or set()
37        self._set_log_level_from_config(config)
38        self.cache = mass.cache
39        self.available = False
40
41    @property
42    def supported_features(self) -> set[ProviderFeature]:
43        """Return the features supported by this Provider."""
44        # should not be overridden in normal circumstances
45        return self._supported_features
46
47    async def handle_async_init(self) -> None:
48        """Handle async initialization of the provider."""
49
50    async def loaded_in_mass(self) -> None:
51        """Call after the provider has been loaded."""
52
53    async def unload(self, is_removed: bool = False) -> None:
54        """
55        Handle unload/close of the provider.
56
57        Called when provider is deregistered (e.g. MA exiting or config reloading).
58        is_removed will be set to True when the provider is removed from the configuration.
59        """
60
61    async def update_config(self, config: ProviderConfig, changed_keys: set[str]) -> None:
62        """
63        Handle logic when the config is updated.
64
65        Override this method in your provider implementation if you need
66        to perform any additional setup logic after the provider is registered and
67        the self.config was loaded, and whenever the config changes.
68        """
69        # default implementation: perform a full reload on any config change
70        # override in your provider if you need more fine-grained control
71        # such as checking the changed_keys set and only reload when 'requires_reload' keys changed
72        if changed_keys == {f"values/{CONF_LOG_LEVEL}"}:
73            # only log level changed, no need to reload
74            self._set_log_level_from_config(config)
75        else:
76            self.logger.info(
77                "Config updated, reloading provider %s (instance_id=%s)",
78                self.domain,
79                self.instance_id,
80            )
81            task_id = f"provider_reload_{self.instance_id}"
82            self.mass.call_later(
83                1, self.mass.load_provider_config, config, self.instance_id, task_id=task_id
84            )
85
86    async def on_mdns_service_state_change(
87        self, name: str, state_change: ServiceStateChange, info: AsyncServiceInfo | None
88    ) -> None:
89        """Handle MDNS service state callback."""
90
91    @property
92    @final
93    def type(self) -> ProviderType:
94        """Return type of this provider."""
95        return self.manifest.type
96
97    @property
98    @final
99    def domain(self) -> str:
100        """Return domain for this provider."""
101        return self.manifest.domain
102
103    @property
104    @final
105    def instance_id(self) -> str:
106        """Return instance_id for this provider(instance)."""
107        return self.config.instance_id
108
109    @property
110    @final
111    def name(self) -> str:
112        """Return (custom) friendly name for this provider instance."""
113        if self.config.name:
114            # always prefer user-set name from config
115            return self.config.name
116        return self.default_name
117
118    @property
119    @final
120    def default_name(self) -> str:
121        """Return a default friendly name for this provider instance."""
122        # create default name based on instance count
123        prov_confs = self.mass.config.get("providers", {}).values()
124        instances = [x["instance_id"] for x in prov_confs if x["domain"] == self.domain]
125        if len(instances) <= 1:
126            # only one instance (or no instances yet at all) - return provider name
127            return self.manifest.name
128        instance_name_postfix = self.instance_name_postfix
129        if not instance_name_postfix:
130            # default implementation - simply use the instance number/index
131            instance_name_postfix = str(instances.index(self.instance_id) + 1)
132        # append instance name to provider name
133        return f"{self.manifest.name} [{self.instance_name_postfix}]"
134
135    @property
136    def instance_name_postfix(self) -> str | None:
137        """Return a (default) instance name postfix for this provider instance."""
138        return None
139
140    @property
141    @final
142    def stage(self) -> ProviderStage:
143        """Return the stage of this provider."""
144        return self.manifest.stage
145
146    def unload_with_error(self, error: str) -> None:
147        """Unload provider with error message."""
148        self.mass.call_later(1, self.mass.unload_provider, self.instance_id, error)
149
150    def to_dict(self) -> dict[str, Any]:
151        """Return Provider(instance) as serializable dict."""
152        return {
153            "type": self.type.value,
154            "domain": self.domain,
155            "name": self.name,
156            "default_name": self.default_name,
157            "instance_name_postfix": self.instance_name_postfix,
158            "instance_id": self.instance_id,
159            "lookup_key": self.instance_id,  # include for backwards compatibility
160            "supported_features": [x.value for x in self.supported_features],
161            "available": self.available,
162            "is_streaming_provider": getattr(self, "is_streaming_provider", None),
163        }
164
165    def supports_feature(self, feature: ProviderFeature) -> bool:
166        """Return True if this provider supports the given feature."""
167        return feature in self.supported_features
168
169    def check_feature(self, feature: ProviderFeature) -> None:
170        """Check if this provider supports the given feature."""
171        if not self.supports_feature(feature):
172            raise UnsupportedFeaturedException(
173                f"Provider {self.name} does not support feature {feature.name}"
174            )
175
176    def _update_config_value(self, key: str, value: Any, encrypted: bool = False) -> None:
177        """Update a config value."""
178        self.mass.config.set_raw_provider_config_value(self.instance_id, key, value, encrypted)
179        # also update the cached copy within the provider instance
180        self.config.values[key].value = value
181
182    def _set_log_level_from_config(self, config: ProviderConfig) -> None:
183        """Set log level from config."""
184        mass_logger = logging.getLogger(MASS_LOGGER_NAME)
185        self.logger = mass_logger.getChild(self.domain)
186        log_level = str(config.get_value(CONF_LOG_LEVEL))
187        if log_level == "GLOBAL":
188            self.logger.setLevel(mass_logger.level)
189        else:
190            self.logger.setLevel(log_level)
191        if logging.getLogger().level > self.logger.level:
192            # if the root logger's level is higher, we need to adjust that too
193            logging.getLogger().setLevel(self.logger.level)
194        self.logger.debug("Log level configured to %s", log_level)
195