/
/
/
1"""Core controller configuration handling for the ConfigController."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import TYPE_CHECKING, Any, cast, overload
7
8from music_assistant_models.auth import Scope
9from music_assistant_models.config_entries import (
10 ConfigActionResult,
11 ConfigEntry,
12 ConfigValueType,
13 CoreConfig,
14)
15
16from music_assistant.constants import (
17 CONF_CORE,
18 CONF_PLAYER_QUEUES,
19 CONFIGURABLE_CORE_CONTROLLERS,
20 DEFAULT_CORE_CONFIG_ENTRIES,
21)
22from music_assistant.controllers.config.constants import _ConfigValueT
23from music_assistant.controllers.config.helpers import _with_translation_owner
24from music_assistant.controllers.player_queues.constants import CONF_AUTOPLAY_PLAYLIST
25from music_assistant.helpers.api import api_command
26
27if TYPE_CHECKING:
28 from music_assistant import MusicAssistant
29 from music_assistant.models.core_controller import CoreController
30
31
32class CoreConfigMixin:
33 """Mixin providing core controller configuration handling for the ConfigController."""
34
35 # Type hints for attributes/methods provided by the class this mixin is used with
36 if TYPE_CHECKING:
37 mass: MusicAssistant
38
39 def get(self, key: str, default: Any = None) -> Any: ... # noqa: D102
40
41 def set(self, key: str, value: Any) -> None: ... # noqa: D102
42
43 def save(self, immediate: bool = False) -> None: ... # noqa: D102
44
45 @api_command("config/core", required_scope=Scope.CONFIG_CORE_READ)
46 async def get_core_configs(self, include_values: bool = False) -> list[CoreConfig]:
47 """Return all core controllers config options."""
48 return [
49 await self.get_core_config(core_controller)
50 if include_values
51 else cast(
52 "CoreConfig",
53 CoreConfig.parse([], self._get_raw_core_config(core_controller)),
54 )
55 for core_controller in CONFIGURABLE_CORE_CONTROLLERS
56 ]
57
58 @api_command("config/core/get", required_scope=Scope.CONFIG_CORE_READ)
59 async def get_core_config(self, domain: str) -> CoreConfig:
60 """Return configuration for a single core controller."""
61 raw_conf = self._get_raw_core_config(domain)
62 # build the schema straight from the controller (no dynamic UI options):
63 # CoreConfig.parse stamps the translation owner itself
64 controller: CoreController = getattr(self.mass, domain)
65 config_entries = list(await controller.get_config_entries() + DEFAULT_CORE_CONFIG_ENTRIES)
66 return cast("CoreConfig", CoreConfig.parse(config_entries, raw_conf))
67
68 @overload
69 async def get_core_config_value(
70 self,
71 domain: str,
72 key: str,
73 *,
74 default: _ConfigValueT,
75 return_type: type[_ConfigValueT] = ...,
76 ) -> _ConfigValueT: ...
77
78 @overload
79 async def get_core_config_value(
80 self,
81 domain: str,
82 key: str,
83 *,
84 default: ConfigValueType = ...,
85 return_type: type[_ConfigValueT] = ...,
86 ) -> _ConfigValueT: ...
87
88 @overload
89 async def get_core_config_value(
90 self,
91 domain: str,
92 key: str,
93 *,
94 default: ConfigValueType = ...,
95 return_type: None = ...,
96 ) -> ConfigValueType: ...
97
98 @api_command("config/core/get_value", required_scope=Scope.CONFIG_CORE_READ)
99 async def get_core_config_value(
100 self,
101 domain: str,
102 key: str,
103 *,
104 default: ConfigValueType = None,
105 return_type: type[_ConfigValueT | ConfigValueType] | None = None,
106 ) -> _ConfigValueT | ConfigValueType:
107 """
108 Return single configentry value for a core controller.
109
110 :param domain: The core controller domain.
111 :param key: The config key to retrieve.
112 :param default: Optional default value to return if key is not found.
113 :param return_type: Optional type hint for type inference (e.g., str, int, bool).
114 Note: This parameter is used purely for static type checking and does not
115 perform runtime type validation. Callers are responsible for ensuring the
116 specified type matches the actual config value type.
117 """
118 # prefer stored value so we don't have to retrieve all config entries every time
119 if (raw_value := self.get_raw_core_config_value(domain, key)) is not None:
120 return raw_value
121 conf = await self.get_core_config(domain)
122 if key not in conf.values:
123 if default is not None:
124 return default
125 msg = f"Config key {key} not found for core controller {domain}"
126 raise KeyError(msg)
127 return (
128 conf.values[key].value
129 if conf.values[key].value is not None
130 else conf.values[key].default_value
131 )
132
133 @api_command("config/core/get_entries", required_scope=Scope.CONFIG_CORE_READ)
134 async def get_core_config_entries(self, domain: str) -> list[ConfigEntry]:
135 """
136 Return Config entries to configure a core controller.
137
138 :param domain: The core controller domain.
139 """
140 controller: CoreController = getattr(self.mass, domain)
141 return await self._resolve_core_config_entries(
142 domain, await controller.get_config_entries()
143 )
144
145 @api_command("config/core/invoke_action", required_scope=Scope.CONFIG_CORE_WRITE)
146 async def invoke_core_config_action(
147 self, domain: str, action: str
148 ) -> list[ConfigEntry] | ConfigActionResult:
149 """
150 Run a one-shot action button from a core module's config.
151
152 A ``ConfigActionResult`` holds the outcome to report to the user; an empty list
153 means the action ran with nothing to report; a non-empty list holds the entries
154 the config form should re-render with.
155
156 :param domain: The core controller domain.
157 :param action: The action id of the pressed button.
158 """
159 controller: CoreController = getattr(self.mass, domain)
160 if (result := await controller.handle_config_action(action)) is None:
161 return []
162 if isinstance(result, ConfigActionResult):
163 result.translation_owner = result.translation_owner or f"core.{domain}"
164 return result
165 return await self._resolve_core_config_entries(domain, result)
166
167 @api_command("config/core/save", required_scope=Scope.CONFIG_CORE_WRITE)
168 async def save_core_config(
169 self,
170 domain: str,
171 values: dict[str, ConfigValueType],
172 ) -> CoreConfig:
173 """Save CoreController Config values."""
174 config = await self.get_core_config(domain)
175 prev_config = config.to_raw()
176 changed_keys = config.update(values)
177 # validate the new config
178 config.validate()
179 if not changed_keys:
180 # no changes
181 return config
182 # save the config first before reloading to avoid issues on reload
183 # for example when reloading the webserver we might be cancelled here
184 conf_key = f"{CONF_CORE}/{domain}"
185 self.set(conf_key, config.to_raw())
186 self.save(immediate=True)
187 try:
188 controller: CoreController = getattr(self.mass, domain)
189 await controller.update_config(config, changed_keys)
190 except asyncio.CancelledError:
191 pass
192 except Exception:
193 # revert to previous config on error
194 self.set(conf_key, prev_config)
195 self.save(immediate=True)
196 raise
197 # reload succeeded; clear last_error and persist the final state
198 config.last_error = None
199 # return full config
200 return await self.get_core_config(domain)
201
202 if TYPE_CHECKING:
203 # Overload for when default is provided - return type matches default type
204 @overload
205 def get_raw_core_config_value(
206 self, core_module: str, key: str, default: _ConfigValueT
207 ) -> _ConfigValueT: ...
208
209 # Overload for when no default is provided - return ConfigValueType | None
210 @overload
211 def get_raw_core_config_value(
212 self, core_module: str, key: str, default: None = None
213 ) -> ConfigValueType | None: ...
214
215 def get_raw_core_config_value(
216 self, core_module: str, key: str, default: ConfigValueType = None
217 ) -> ConfigValueType:
218 """
219 Return (raw) single configentry value for a core controller.
220
221 Note that this only returns the stored value without any validation or default.
222 """
223 return cast(
224 "ConfigValueType",
225 self.get(
226 f"{CONF_CORE}/{core_module}/values/{key}",
227 self.get(f"{CONF_CORE}/{core_module}/{key}", default),
228 ),
229 )
230
231 def set_raw_core_config_value(self, core_module: str, key: str, value: ConfigValueType) -> None:
232 """
233 Set (raw) single config(entry) value for a core controller.
234
235 Note that this only stores the (raw) value without any validation or default.
236 """
237 self.ensure_core_config_base(core_module)
238 self.set(f"{CONF_CORE}/{core_module}/values/{key}", value)
239 # also update the controller's in-place config copy (if any) so
240 # object-local value reads stay in sync with raw writes
241 controller = getattr(self.mass, core_module, None)
242 if (config := getattr(controller, "config", None)) and (entry := config.values.get(key)):
243 entry.value = value
244
245 def ensure_core_config_base(self, core_module: str) -> None:
246 """
247 Create or repair the stored config block of a core controller.
248
249 Call this before storing a raw value in the block, so the block is left in a state
250 that still parses as a CoreConfig.
251
252 :param core_module: The domain of the core controller.
253 """
254 raw_conf = self.get(f"{CONF_CORE}/{core_module}")
255 if not isinstance(raw_conf, dict) or not raw_conf:
256 self.set(f"{CONF_CORE}/{core_module}", CoreConfig({}, core_module).to_raw())
257 elif "domain" not in raw_conf:
258 self.set(f"{CONF_CORE}/{core_module}/domain", core_module)
259
260 def _get_raw_core_config(self, domain: str) -> dict[str, Any]:
261 """
262 Return the stored raw config of a core controller, ready to parse as a CoreConfig.
263
264 :param domain: The domain of the core controller.
265 """
266 raw_conf = self.get(f"{CONF_CORE}/{domain}", {})
267 if not isinstance(raw_conf, dict):
268 raw_conf = {}
269 if "domain" not in raw_conf:
270 # older versions could store the block without its mandatory domain key
271 return {**raw_conf, "domain": domain}
272 return raw_conf
273
274 async def _resolve_core_config_entries(
275 self, domain: str, entries: tuple[ConfigEntry, ...]
276 ) -> list[ConfigEntry]:
277 """Append the server default entries, resolve dynamic options and stamp the owner."""
278 all_entries = list(entries + DEFAULT_CORE_CONFIG_ENTRIES)
279 if domain == CONF_PLAYER_QUEUES:
280 # populate the global autoplay playlist dropdown for the UI here (not in get_core_config),
281 # so the config value/parse path stays free of a library lookup
282 playlist_options = await self.mass.config._library_playlist_options()
283 for entry in all_entries:
284 if entry.key == CONF_AUTOPLAY_PLAYLIST:
285 entry.options = playlist_options
286 return _with_translation_owner(all_entries, f"core.{domain}")
287