/
/
/
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(
54 [],
55 self.get(f"{CONF_CORE}/{core_controller}", {"domain": core_controller}),
56 ),
57 )
58 for core_controller in CONFIGURABLE_CORE_CONTROLLERS
59 ]
60
61 @api_command("config/core/get", required_scope=Scope.CONFIG_CORE_READ)
62 async def get_core_config(self, domain: str) -> CoreConfig:
63 """Return configuration for a single core controller."""
64 raw_conf = self.get(f"{CONF_CORE}/{domain}", {})
65 if not isinstance(raw_conf, dict):
66 raw_conf = {}
67 if "domain" not in raw_conf:
68 raw_conf = {**raw_conf, "domain": domain}
69 # build the schema straight from the controller (no dynamic UI options):
70 # CoreConfig.parse stamps the translation owner itself
71 controller: CoreController = getattr(self.mass, domain)
72 config_entries = list(await controller.get_config_entries() + DEFAULT_CORE_CONFIG_ENTRIES)
73 return cast("CoreConfig", CoreConfig.parse(config_entries, raw_conf))
74
75 @overload
76 async def get_core_config_value(
77 self,
78 domain: str,
79 key: str,
80 *,
81 default: _ConfigValueT,
82 return_type: type[_ConfigValueT] = ...,
83 ) -> _ConfigValueT: ...
84
85 @overload
86 async def get_core_config_value(
87 self,
88 domain: str,
89 key: str,
90 *,
91 default: ConfigValueType = ...,
92 return_type: type[_ConfigValueT] = ...,
93 ) -> _ConfigValueT: ...
94
95 @overload
96 async def get_core_config_value(
97 self,
98 domain: str,
99 key: str,
100 *,
101 default: ConfigValueType = ...,
102 return_type: None = ...,
103 ) -> ConfigValueType: ...
104
105 @api_command("config/core/get_value", required_scope=Scope.CONFIG_CORE_READ)
106 async def get_core_config_value(
107 self,
108 domain: str,
109 key: str,
110 *,
111 default: ConfigValueType = None,
112 return_type: type[_ConfigValueT | ConfigValueType] | None = None,
113 ) -> _ConfigValueT | ConfigValueType:
114 """
115 Return single configentry value for a core controller.
116
117 :param domain: The core controller domain.
118 :param key: The config key to retrieve.
119 :param default: Optional default value to return if key is not found.
120 :param return_type: Optional type hint for type inference (e.g., str, int, bool).
121 Note: This parameter is used purely for static type checking and does not
122 perform runtime type validation. Callers are responsible for ensuring the
123 specified type matches the actual config value type.
124 """
125 # prefer stored value so we don't have to retrieve all config entries every time
126 if (raw_value := self.get_raw_core_config_value(domain, key)) is not None:
127 return raw_value
128 conf = await self.get_core_config(domain)
129 if key not in conf.values:
130 if default is not None:
131 return default
132 msg = f"Config key {key} not found for core controller {domain}"
133 raise KeyError(msg)
134 return (
135 conf.values[key].value
136 if conf.values[key].value is not None
137 else conf.values[key].default_value
138 )
139
140 @api_command("config/core/get_entries", required_scope=Scope.CONFIG_CORE_READ)
141 async def get_core_config_entries(self, domain: str) -> list[ConfigEntry]:
142 """
143 Return Config entries to configure a core controller.
144
145 :param domain: The core controller domain.
146 """
147 controller: CoreController = getattr(self.mass, domain)
148 return await self._resolve_core_config_entries(
149 domain, await controller.get_config_entries()
150 )
151
152 @api_command("config/core/invoke_action", required_scope=Scope.CONFIG_CORE_WRITE)
153 async def invoke_core_config_action(
154 self, domain: str, action: str
155 ) -> list[ConfigEntry] | ConfigActionResult:
156 """
157 Run a one-shot action button from a core module's config.
158
159 A ``ConfigActionResult`` holds the outcome to report to the user; an empty list
160 means the action ran with nothing to report; a non-empty list holds the entries
161 the config form should re-render with.
162
163 :param domain: The core controller domain.
164 :param action: The action id of the pressed button.
165 """
166 controller: CoreController = getattr(self.mass, domain)
167 if (result := await controller.handle_config_action(action)) is None:
168 return []
169 if isinstance(result, ConfigActionResult):
170 result.translation_owner = result.translation_owner or f"core.{domain}"
171 return result
172 return await self._resolve_core_config_entries(domain, result)
173
174 @api_command("config/core/save", required_scope=Scope.CONFIG_CORE_WRITE)
175 async def save_core_config(
176 self,
177 domain: str,
178 values: dict[str, ConfigValueType],
179 ) -> CoreConfig:
180 """Save CoreController Config values."""
181 config = await self.get_core_config(domain)
182 prev_config = config.to_raw()
183 changed_keys = config.update(values)
184 # validate the new config
185 config.validate()
186 if not changed_keys:
187 # no changes
188 return config
189 # save the config first before reloading to avoid issues on reload
190 # for example when reloading the webserver we might be cancelled here
191 conf_key = f"{CONF_CORE}/{domain}"
192 self.set(conf_key, config.to_raw())
193 self.save(immediate=True)
194 try:
195 controller: CoreController = getattr(self.mass, domain)
196 await controller.update_config(config, changed_keys)
197 except asyncio.CancelledError:
198 pass
199 except Exception:
200 # revert to previous config on error
201 self.set(conf_key, prev_config)
202 self.save(immediate=True)
203 raise
204 # reload succeeded; clear last_error and persist the final state
205 config.last_error = None
206 # return full config
207 return await self.get_core_config(domain)
208
209 if TYPE_CHECKING:
210 # Overload for when default is provided - return type matches default type
211 @overload
212 def get_raw_core_config_value(
213 self, core_module: str, key: str, default: _ConfigValueT
214 ) -> _ConfigValueT: ...
215
216 # Overload for when no default is provided - return ConfigValueType | None
217 @overload
218 def get_raw_core_config_value(
219 self, core_module: str, key: str, default: None = None
220 ) -> ConfigValueType | None: ...
221
222 def get_raw_core_config_value(
223 self, core_module: str, key: str, default: ConfigValueType = None
224 ) -> ConfigValueType:
225 """
226 Return (raw) single configentry value for a core controller.
227
228 Note that this only returns the stored value without any validation or default.
229 """
230 return cast(
231 "ConfigValueType",
232 self.get(
233 f"{CONF_CORE}/{core_module}/values/{key}",
234 self.get(f"{CONF_CORE}/{core_module}/{key}", default),
235 ),
236 )
237
238 def set_raw_core_config_value(self, core_module: str, key: str, value: ConfigValueType) -> None:
239 """
240 Set (raw) single config(entry) value for a core controller.
241
242 Note that this only stores the (raw) value without any validation or default.
243 """
244 if not self.get(f"{CONF_CORE}/{core_module}"):
245 # create base object first if needed
246 self.set(f"{CONF_CORE}/{core_module}", CoreConfig({}, core_module).to_raw())
247 self.set(f"{CONF_CORE}/{core_module}/values/{key}", value)
248 # also update the controller's in-place config copy (if any) so
249 # object-local value reads stay in sync with raw writes
250 controller = getattr(self.mass, core_module, None)
251 if (config := getattr(controller, "config", None)) and (entry := config.values.get(key)):
252 entry.value = value
253
254 async def _resolve_core_config_entries(
255 self, domain: str, entries: tuple[ConfigEntry, ...]
256 ) -> list[ConfigEntry]:
257 """Append the server default entries, resolve dynamic options and stamp the owner."""
258 all_entries = list(entries + DEFAULT_CORE_CONFIG_ENTRIES)
259 if domain == CONF_PLAYER_QUEUES:
260 # populate the global autoplay playlist dropdown for the UI here (not in get_core_config),
261 # so the config value/parse path stays free of a library lookup
262 playlist_options = await self.mass.config._library_playlist_options()
263 for entry in all_entries:
264 if entry.key == CONF_AUTOPLAY_PLAYLIST:
265 entry.options = playlist_options
266 return _with_translation_owner(all_entries, f"core.{domain}")
267