/
/
/
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 mass: MusicAssistant
26 manifest: ProviderManifest
27 config: ProviderConfig
28
29 def __init__(
30 self,
31 mass: MusicAssistant,
32 manifest: ProviderManifest,
33 config: ProviderConfig,
34 supported_features: set[ProviderFeature] | None = None,
35 ) -> None:
36 """Initialize MusicProvider."""
37 self.mass = mass
38 self.manifest = manifest
39 self.config = config
40 self._supported_features = supported_features or set()
41 self._set_log_level_from_config(config)
42 self.cache = mass.cache
43 self.available = False
44
45 @property
46 def supported_features(self) -> set[ProviderFeature]:
47 """Return the features supported by this Provider."""
48 # should not be overridden in normal circumstances
49 return self._supported_features
50
51 async def handle_async_init(self) -> None:
52 """Handle async initialization of the provider."""
53
54 async def loaded_in_mass(self) -> None:
55 """Call after the provider has been loaded."""
56
57 async def unload(self, is_removed: bool = False) -> None:
58 """
59 Handle unload/close of the provider.
60
61 Called when provider is deregistered (e.g. MA exiting or config reloading).
62 is_removed will be set to True when the provider is removed from the configuration.
63 """
64
65 async def update_config(self, config: ProviderConfig, changed_keys: set[str]) -> None:
66 """
67 Handle logic when the config is updated.
68
69 Override this method in your provider implementation if you need
70 to perform any additional setup logic after the provider is registered and
71 the self.config was loaded, and whenever the config changes.
72
73 The default implementation reloads the provider on any config change
74 (except log-level-only changes), since provider reloads are lightweight
75 and most providers cache config values at setup time.
76 """
77 # always update the stored config so dynamic reads pick up new values
78 self.config = config
79
80 # update log level if changed
81 if f"values/{CONF_LOG_LEVEL}" in changed_keys or "name" in changed_keys:
82 self._set_log_level_from_config(config)
83
84 # reload if any non-log-level value keys changed
85 value_keys_changed = {
86 k for k in changed_keys if k.startswith("values/") and k != f"values/{CONF_LOG_LEVEL}"
87 }
88 if value_keys_changed:
89 self.logger.info(
90 "Config updated, reloading provider %s (instance_id=%s)",
91 self.domain,
92 self.instance_id,
93 )
94 task_id = f"provider_reload_{self.instance_id}"
95 self.mass.call_later(1, self.mass.load_provider_config, config, task_id=task_id)
96
97 async def on_mdns_service_state_change(
98 self, name: str, state_change: ServiceStateChange, info: AsyncServiceInfo | None
99 ) -> None:
100 """Handle MDNS service state callback."""
101
102 @property
103 @final
104 def type(self) -> ProviderType:
105 """Return type of this provider."""
106 return self.manifest.type
107
108 @property
109 @final
110 def domain(self) -> str:
111 """Return domain for this provider."""
112 return self.manifest.domain
113
114 @property
115 @final
116 def instance_id(self) -> str:
117 """Return instance_id for this provider(instance)."""
118 return self.config.instance_id
119
120 @property
121 @final
122 def name(self) -> str:
123 """Return (custom) friendly name for this provider instance."""
124 if self.config.name:
125 # always prefer user-set name from config
126 return self.config.name
127 return self.default_name
128
129 @property
130 @final
131 def default_name(self) -> str:
132 """Return a default friendly name for this provider instance."""
133 # create default name based on instance count
134 prov_confs = self.mass.config.get("providers", {}).values()
135 instances = [x["instance_id"] for x in prov_confs if x["domain"] == self.domain]
136 if len(instances) <= 1:
137 # only one instance (or no instances yet at all) - return provider name
138 return self.manifest.name
139 instance_name_postfix = self.instance_name_postfix
140 if not instance_name_postfix:
141 # default implementation - simply use the instance number/index
142 instance_name_postfix = str(instances.index(self.instance_id) + 1)
143 # append instance name to provider name
144 return f"{self.manifest.name} [{self.instance_name_postfix}]"
145
146 @property
147 def instance_name_postfix(self) -> str | None:
148 """Return a (default) instance name postfix for this provider instance."""
149 return None
150
151 @property
152 @final
153 def stage(self) -> ProviderStage:
154 """Return the stage of this provider."""
155 return self.manifest.stage
156
157 def unload_with_error(self, error: str) -> None:
158 """Unload provider with error message."""
159 self.mass.call_later(1, self.mass.unload_provider, self.instance_id, error)
160
161 def to_dict(self) -> dict[str, Any]:
162 """Return Provider(instance) as serializable dict."""
163 return {
164 "type": self.type.value,
165 "domain": self.domain,
166 "name": self.name,
167 "default_name": self.default_name,
168 "instance_name_postfix": self.instance_name_postfix,
169 "instance_id": self.instance_id,
170 "lookup_key": self.instance_id, # include for backwards compatibility
171 "supported_features": [x.value for x in self.supported_features],
172 "available": self.available,
173 "is_streaming_provider": getattr(self, "is_streaming_provider", None),
174 }
175
176 def supports_feature(self, feature: ProviderFeature) -> bool:
177 """Return True if this provider supports the given feature."""
178 return feature in self.supported_features
179
180 def check_feature(self, feature: ProviderFeature) -> None:
181 """Check if this provider supports the given feature."""
182 if not self.supports_feature(feature):
183 raise UnsupportedFeaturedException(
184 f"Provider {self.name} does not support feature {feature.name}"
185 )
186
187 def _update_config_value(self, key: str, value: Any, encrypted: bool = False) -> None:
188 """Update a config value."""
189 self.mass.config.set_raw_provider_config_value(self.instance_id, key, value, encrypted)
190 # also update the cached copy within the provider instance
191 self.config.values[key].value = value
192
193 def _set_log_level_from_config(self, config: ProviderConfig) -> None:
194 """Set log level from config."""
195 mass_logger = logging.getLogger(MASS_LOGGER_NAME)
196 # self.name is only available after async_init. Otherwise we run into a race condition.
197 # see https://github.com/music-assistant/support/issues/4801
198 logging_name = self.domain
199 if getattr(self, "available", False):
200 # async_init completed
201 logging_name = self.name
202 self.logger = mass_logger.getChild(logging_name)
203 log_level = str(config.get_value(CONF_LOG_LEVEL))
204 if log_level == "GLOBAL":
205 self.logger.setLevel(mass_logger.level)
206 else:
207 self.logger.setLevel(log_level)
208 if logging.getLogger().level > self.logger.level:
209 # if the root logger's level is higher, we need to adjust that too
210 logging.getLogger().setLevel(self.logger.level)
211 self.logger.debug("Log level configured to %s", log_level)
212