music-assistant-server

1"""Model/base for a Core controller within Music Assistant."""
2
3from __future__ import annotations
4
5import logging
6from typing import TYPE_CHECKING
7
8from music_assistant_models.enums import ProviderStage, ProviderType
9from music_assistant_models.provider import ProviderManifest
10
11from music_assistant.constants import CONF_LOG_LEVEL, MASS_LOGGER_NAME
12
13if TYPE_CHECKING:
14    from music_assistant_models.config_entries import ConfigEntry, ConfigValueType, CoreConfig
15
16    from music_assistant.mass import MusicAssistant
17
18
19class CoreController:
20    """Base representation of a Core controller within Music Assistant."""
21
22    domain: str  # used as identifier (=name of the module)
23    manifest: ProviderManifest  # some info for the UI only
24
25    def __init__(self, mass: MusicAssistant) -> None:
26        """Initialize MusicProvider."""
27        self.mass = mass
28        self._set_logger()
29        self.manifest = ProviderManifest(
30            type=ProviderType.CORE,
31            domain=self.domain,
32            name=f"{self.domain.title()} Core controller",
33            description=f"{self.domain.title()} Core controller",
34            codeowners=["@music-assistant"],
35            stage=ProviderStage.STABLE,
36            icon="puzzle-outline",
37            builtin=True,
38            allow_disable=False,
39        )
40
41    async def get_config_entries(
42        self,
43        action: str | None = None,
44        values: dict[str, ConfigValueType] | None = None,
45    ) -> tuple[ConfigEntry, ...]:
46        """Return all Config Entries for this core module (if any)."""
47        return ()
48
49    async def setup(self, config: CoreConfig) -> None:
50        """Async initialize of module."""
51
52    async def close(self) -> None:
53        """Handle logic on server stop."""
54
55    async def reload(self, config: CoreConfig | None = None) -> None:
56        """Reload this core controller."""
57        await self.close()
58        if config is None:
59            config = await self.mass.config.get_core_config(self.domain)
60        log_level = str(config.get_value(CONF_LOG_LEVEL))
61        self._set_logger(log_level)
62        await self.setup(config)
63
64    async def update_config(self, config: CoreConfig, changed_keys: set[str]) -> None:
65        """Handle logic when the config is updated."""
66        # default implementation: perform a full reload on any config change
67        # TODO: only reload when 'requires_reload' keys changed
68        if changed_keys == {f"values/{CONF_LOG_LEVEL}"}:
69            # only log level changed, no need to reload
70            log_value = str(config.get_value(CONF_LOG_LEVEL))
71            self._set_logger(log_value)
72        else:
73            self.logger.info(
74                "Config updated, reloading %s core controller",
75                self.manifest.name,
76            )
77            task_id = f"core_reload_{self.domain}"
78            self.mass.call_later(1, self.reload, config, task_id=task_id)
79
80    def _set_logger(self, log_level: str | None = None) -> None:
81        """Set the logger settings."""
82        mass_logger = logging.getLogger(MASS_LOGGER_NAME)
83        self.logger = mass_logger.getChild(self.domain)
84        if log_level is None:
85            log_level = str(
86                self.mass.config.get_raw_core_config_value(self.domain, CONF_LOG_LEVEL, "GLOBAL")
87            )
88        if log_level == "GLOBAL":
89            self.logger.setLevel(mass_logger.level)
90        else:
91            self.logger.setLevel(log_level)
92        if logging.getLogger().level > self.logger.level:
93            # if the root logger's level is higher, we need to adjust that too
94            logging.getLogger().setLevel(self.logger.level)
95