/
/
/
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
8
9from music_assistant_models.enums import ProviderStage, ProviderType
10from music_assistant_models.provider import ProviderManifest
11
12from music_assistant.constants import CONF_LOG_LEVEL, MASS_LOGGER_NAME
13
14if TYPE_CHECKING:
15 from music_assistant_models.config_entries import ConfigEntry, ConfigValueType, CoreConfig
16
17 from music_assistant.mass import MusicAssistant
18
19
20class CoreController:
21 """Base representation of a Core controller within Music Assistant."""
22
23 domain: str # used as identifier (=name of the module)
24 manifest: ProviderManifest # some info for the UI only
25
26 def __init__(self, mass: MusicAssistant) -> None:
27 """Initialize core controller."""
28 self.mass = mass
29 self.initialized = asyncio.Event()
30 self._set_logger()
31 self.manifest = ProviderManifest(
32 type=ProviderType.CORE,
33 domain=self.domain,
34 name=f"{self.domain.title()} Core controller",
35 description=f"{self.domain.title()} Core controller",
36 codeowners=["@music-assistant"],
37 stage=ProviderStage.STABLE,
38 icon="puzzle-outline",
39 builtin=True,
40 allow_disable=False,
41 )
42
43 async def get_config_entries(
44 self,
45 action: str | None = None,
46 values: dict[str, ConfigValueType] | None = None,
47 ) -> tuple[ConfigEntry, ...]:
48 """Return all Config Entries for this core module (if any)."""
49 return ()
50
51 async def setup(self, config: CoreConfig) -> None:
52 """Async initialize of module."""
53
54 async def close(self) -> None:
55 """Handle logic on server stop."""
56
57 async def reload(self, config: CoreConfig | None = None) -> None:
58 """Reload this core controller."""
59 await self.close()
60 if config is None:
61 config = await self.mass.config.get_core_config(self.domain)
62 log_level = str(config.get_value(CONF_LOG_LEVEL))
63 self._set_logger(log_level)
64 await self.setup(config)
65
66 async def update_config(self, config: CoreConfig, changed_keys: set[str]) -> None:
67 """Handle logic when the config is updated."""
68 # always update the stored config so dynamic reads pick up new values
69 self.config = config
70
71 # apply log level change dynamically (doesn't require reload)
72 if f"values/{CONF_LOG_LEVEL}" in changed_keys:
73 log_value = str(config.get_value(CONF_LOG_LEVEL))
74 self._set_logger(log_value)
75
76 # reload if any changed value entry has requires_reload set to True
77 needs_reload = any(
78 (entry := config.values.get(key.removeprefix("values/"))) is not None
79 and entry.requires_reload is True
80 for key in changed_keys
81 if key.startswith("values/")
82 )
83 if needs_reload:
84 self.logger.info(
85 "Config updated, reloading %s core controller",
86 self.manifest.name,
87 )
88 task_id = f"core_reload_{self.domain}"
89 self.mass.call_later(1, self.reload, config, task_id=task_id)
90
91 def _set_logger(self, log_level: str | None = None) -> None:
92 """Set the logger settings."""
93 mass_logger = logging.getLogger(MASS_LOGGER_NAME)
94 self.logger = mass_logger.getChild(self.domain)
95 if log_level is None:
96 log_level = str(
97 self.mass.config.get_raw_core_config_value(self.domain, CONF_LOG_LEVEL, "GLOBAL")
98 )
99 if log_level == "GLOBAL":
100 self.logger.setLevel(mass_logger.level)
101 else:
102 self.logger.setLevel(log_level)
103 if logging.getLogger().level > self.logger.level:
104 # if the root logger's level is higher, we need to adjust that too
105 logging.getLogger().setLevel(self.logger.level)
106