/
/
/
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