/
/
/
1"""Model/base for a Core controller within Music Assistant."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from typing import TYPE_CHECKING, TypeVar, overload
8
9from music_assistant_models.config_entries import ConfigValueType
10from music_assistant_models.enums import ProviderStage, ProviderType
11from music_assistant_models.errors import ActionUnavailable
12from music_assistant_models.provider import ProviderManifest
13
14from music_assistant.constants import CONF_LOG_LEVEL, MASS_LOGGER_NAME
15
16if TYPE_CHECKING:
17 from music_assistant_models.config_entries import ConfigActionResult, ConfigEntry, CoreConfig
18
19 from music_assistant.helpers.json import SerializableType
20 from music_assistant.mass import MusicAssistant
21
22# TypeVar for config value type inference
23_ConfigValueT = TypeVar("_ConfigValueT", bound=ConfigValueType)
24
25
26class CoreController:
27 """Base representation of a Core controller within Music Assistant."""
28
29 domain: str # used as identifier (=name of the module)
30 manifest: ProviderManifest # some info for the UI only
31 # config: the controller's active configuration, assigned at startup/reload and kept
32 # up to date by the config controller, so internal code can read config values
33 # (including entry defaults) without rebuilding the config entries
34 config: CoreConfig
35
36 def __init__(self, mass: MusicAssistant) -> None:
37 """Initialize core controller."""
38 self.mass = mass
39 self.initialized = asyncio.Event()
40 self._set_logger()
41 self.manifest = ProviderManifest(
42 type=ProviderType.CORE,
43 domain=self.domain,
44 name=f"{self.domain.title()} Core controller",
45 description=f"{self.domain.title()} Core controller",
46 codeowners=["@music-assistant"],
47 stage=ProviderStage.STABLE,
48 icon="puzzle-outline",
49 builtin=True,
50 allow_disable=False,
51 )
52
53 @property
54 def translation_owner(self) -> str:
55 """Return the "core.<domain>" namespace this module's translation strings resolve under."""
56 return f"core.{self.domain}"
57
58 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
59 """
60 Return all Config Entries for this core module (if any).
61
62 Include ``ConfigEntryType.ACTION`` entries for one-shot buttons and handle
63 their presses in ``handle_config_action``.
64 """
65 return ()
66
67 async def handle_config_action(
68 self, action: str
69 ) -> tuple[ConfigEntry, ...] | ConfigActionResult | None:
70 """
71 Run the one-shot side effect for a pressed action button from this module's config.
72
73 Override to run the side effect for each ``ConfigEntryType.ACTION`` entry this
74 module declares. Return a ``ConfigActionResult`` to report the outcome (a message
75 to show and/or a url to open), or None when there is nothing to report. Raise to
76 report failure to the caller. Returning config entries re-renders the config form
77 with those entries instead.
78
79 :param action: The action id of the pressed button (an entry's ``action`` key).
80 """
81 raise ActionUnavailable(f"Unknown action: {action}")
82
83 @overload
84 def get_config_value(
85 self, key: str, default: _ConfigValueT, *, return_type: type[_ConfigValueT] = ...
86 ) -> _ConfigValueT: ...
87
88 @overload
89 def get_config_value(
90 self, key: str, default: ConfigValueType = ..., *, return_type: type[_ConfigValueT]
91 ) -> _ConfigValueT: ...
92
93 @overload
94 def get_config_value(
95 self, key: str, default: ConfigValueType = ..., *, return_type: None = ...
96 ) -> ConfigValueType: ...
97
98 def get_config_value(
99 self,
100 key: str,
101 default: ConfigValueType = None,
102 *,
103 return_type: type[_ConfigValueT | ConfigValueType] | None = None,
104 ) -> _ConfigValueT | ConfigValueType:
105 """
106 Return a single config value from this core controller's active configuration.
107
108 Entry defaults are already applied to the active configuration, so the
109 default is only returned when the key itself is not present.
110
111 :param key: The config key to retrieve.
112 :param default: Value to return when the key is not present in the config.
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 return self.config.get_value(key, default)
119
120 async def get_diagnostics(self) -> dict[str, SerializableType] | None:
121 """
122 Return optional diagnostics info for this controller to include in diagnostics reports.
123
124 Return None (the default) when this controller has nothing to contribute.
125 Keep the returned data small, JSON serializable and free of sensitive values.
126 """
127 return None
128
129 async def setup(self, config: CoreConfig) -> None:
130 """Async initialize of module."""
131
132 async def post_setup(self) -> None:
133 """Handle logic after all core controllers have been set up."""
134
135 async def close(self) -> None:
136 """Handle logic on server stop."""
137
138 async def reload(self, config: CoreConfig | None = None) -> None:
139 """Reload this core controller."""
140 await self.close()
141 if config is None:
142 config = await self.mass.config.get_core_config(self.domain)
143 log_level = str(config.get_value(CONF_LOG_LEVEL))
144 self._set_logger(log_level)
145 self.config = config
146 await self.setup(config)
147 await self.post_setup()
148
149 async def update_config(self, config: CoreConfig, changed_keys: set[str]) -> None:
150 """Handle logic when the config is updated."""
151 # always update the stored config so dynamic reads pick up new values
152 self.config = config
153
154 # apply log level change dynamically (doesn't require reload)
155 if f"values/{CONF_LOG_LEVEL}" in changed_keys:
156 log_value = str(config.get_value(CONF_LOG_LEVEL))
157 self._set_logger(log_value)
158
159 # reload if any changed value entry has requires_reload set to True
160 needs_reload = any(
161 (entry := config.values.get(key.removeprefix("values/"))) is not None
162 and entry.requires_reload is True
163 for key in changed_keys
164 if key.startswith("values/")
165 )
166 if needs_reload:
167 self.logger.info(
168 "Config updated, reloading %s core controller",
169 self.manifest.name,
170 )
171 task_id = f"core_reload_{self.domain}"
172 self.mass.call_later(1, self.reload, config, task_id=task_id)
173
174 def _set_logger(self, log_level: str | None = None) -> None:
175 """Set the logger settings."""
176 mass_logger = logging.getLogger(MASS_LOGGER_NAME)
177 self.logger = mass_logger.getChild(self.domain)
178 if log_level is None:
179 log_level = str(
180 self.mass.config.get_raw_core_config_value(self.domain, CONF_LOG_LEVEL, "GLOBAL")
181 )
182 if log_level == "GLOBAL":
183 self.logger.setLevel(mass_logger.level)
184 else:
185 self.logger.setLevel(log_level)
186