/
/
/
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 The default implementation reloads the provider on any config change
70 (except log-level-only changes), since provider reloads are lightweight
71 and most providers cache config values at setup time.
72 """
73 # always update the stored config so dynamic reads pick up new values
74 self.config = config
75
76 # update log level if changed
77 if f"values/{CONF_LOG_LEVEL}" in changed_keys:
78 self._set_log_level_from_config(config)
79
80 # reload if any non-log-level value keys changed
81 value_keys_changed = {
82 k for k in changed_keys if k.startswith("values/") and k != f"values/{CONF_LOG_LEVEL}"
83 }
84 if value_keys_changed:
85 self.logger.info(
86 "Config updated, reloading provider %s (instance_id=%s)",
87 self.domain,
88 self.instance_id,
89 )
90 task_id = f"provider_reload_{self.instance_id}"
91 self.mass.call_later(
92 1, self.mass.load_provider_config, config, self.instance_id, task_id=task_id
93 )
94
95 async def on_mdns_service_state_change(
96 self, name: str, state_change: ServiceStateChange, info: AsyncServiceInfo | None
97 ) -> None:
98 """Handle MDNS service state callback."""
99
100 @property
101 @final
102 def type(self) -> ProviderType:
103 """Return type of this provider."""
104 return self.manifest.type
105
106 @property
107 @final
108 def domain(self) -> str:
109 """Return domain for this provider."""
110 return self.manifest.domain
111
112 @property
113 @final
114 def instance_id(self) -> str:
115 """Return instance_id for this provider(instance)."""
116 return self.config.instance_id
117
118 @property
119 @final
120 def name(self) -> str:
121 """Return (custom) friendly name for this provider instance."""
122 if self.config.name:
123 # always prefer user-set name from config
124 return self.config.name
125 return self.default_name
126
127 @property
128 @final
129 def default_name(self) -> str:
130 """Return a default friendly name for this provider instance."""
131 # create default name based on instance count
132 prov_confs = self.mass.config.get("providers", {}).values()
133 instances = [x["instance_id"] for x in prov_confs if x["domain"] == self.domain]
134 if len(instances) <= 1:
135 # only one instance (or no instances yet at all) - return provider name
136 return self.manifest.name
137 instance_name_postfix = self.instance_name_postfix
138 if not instance_name_postfix:
139 # default implementation - simply use the instance number/index
140 instance_name_postfix = str(instances.index(self.instance_id) + 1)
141 # append instance name to provider name
142 return f"{self.manifest.name} [{self.instance_name_postfix}]"
143
144 @property
145 def instance_name_postfix(self) -> str | None:
146 """Return a (default) instance name postfix for this provider instance."""
147 return None
148
149 @property
150 @final
151 def stage(self) -> ProviderStage:
152 """Return the stage of this provider."""
153 return self.manifest.stage
154
155 def unload_with_error(self, error: str) -> None:
156 """Unload provider with error message."""
157 self.mass.call_later(1, self.mass.unload_provider, self.instance_id, error)
158
159 def to_dict(self) -> dict[str, Any]:
160 """Return Provider(instance) as serializable dict."""
161 return {
162 "type": self.type.value,
163 "domain": self.domain,
164 "name": self.name,
165 "default_name": self.default_name,
166 "instance_name_postfix": self.instance_name_postfix,
167 "instance_id": self.instance_id,
168 "lookup_key": self.instance_id, # include for backwards compatibility
169 "supported_features": [x.value for x in self.supported_features],
170 "available": self.available,
171 "is_streaming_provider": getattr(self, "is_streaming_provider", None),
172 }
173
174 def supports_feature(self, feature: ProviderFeature) -> bool:
175 """Return True if this provider supports the given feature."""
176 return feature in self.supported_features
177
178 def check_feature(self, feature: ProviderFeature) -> None:
179 """Check if this provider supports the given feature."""
180 if not self.supports_feature(feature):
181 raise UnsupportedFeaturedException(
182 f"Provider {self.name} does not support feature {feature.name}"
183 )
184
185 def _update_config_value(self, key: str, value: Any, encrypted: bool = False) -> None:
186 """Update a config value."""
187 self.mass.config.set_raw_provider_config_value(self.instance_id, key, value, encrypted)
188 # also update the cached copy within the provider instance
189 self.config.values[key].value = value
190
191 def _set_log_level_from_config(self, config: ProviderConfig) -> None:
192 """Set log level from config."""
193 mass_logger = logging.getLogger(MASS_LOGGER_NAME)
194 self.logger = mass_logger.getChild(self.domain)
195 log_level = str(config.get_value(CONF_LOG_LEVEL))
196 if log_level == "GLOBAL":
197 self.logger.setLevel(mass_logger.level)
198 else:
199 self.logger.setLevel(log_level)
200 if logging.getLogger().level > self.logger.level:
201 # if the root logger's level is higher, we need to adjust that too
202 logging.getLogger().setLevel(self.logger.level)
203 self.logger.debug("Log level configured to %s", log_level)
204