music-assistant-server

1"""Main Music Assistant class."""
2
3from __future__ import annotations
4
5import asyncio
6import inspect
7import logging
8import os
9import pathlib
10import threading
11import time
12from base64 import b64encode
13from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable, Coroutine
14from contextlib import asynccontextmanager
15from typing import TYPE_CHECKING, Any, Self, TypeGuard, TypeVar, cast, overload
16from uuid import uuid4
17
18import aiofiles
19from aiofiles.os import wrap
20from music_assistant_models.api import ServerInfoMessage
21from music_assistant_models.auth import Scope
22from music_assistant_models.config_entries import ProviderError
23from music_assistant_models.enums import (
24    CoreState,
25    EventType,
26    ProviderFeature,
27    ProviderIconVariant,
28    ProviderType,
29)
30from music_assistant_models.errors import (
31    MusicAssistantError,
32    SetupFailedError,
33    UnsupportedSystemError,
34)
35from music_assistant_models.event import MassEvent
36from music_assistant_models.helpers import set_global_cache_values
37from music_assistant_models.provider import ProviderManifest
38
39from music_assistant.constants import (
40    API_SCHEMA_VERSION,
41    CONF_DEFAULT_PROVIDERS_SETUP,
42    CONF_PROVIDERS,
43    CONF_SERVER_ID,
44    CONFIGURABLE_CORE_CONTROLLERS,
45    DEFAULT_PROVIDERS,
46    MASS_LOGGER_NAME,
47    MIN_SCHEMA_VERSION,
48    VERBOSE_LOG_LEVEL,
49)
50from music_assistant.controllers.cache import CacheController
51from music_assistant.controllers.config import ConfigController
52from music_assistant.controllers.dashboard import DashboardController
53from music_assistant.controllers.diagnostics import DiagnosticsController
54from music_assistant.controllers.discovery import DiscoveryController
55from music_assistant.controllers.metadata import MetaDataController
56from music_assistant.controllers.music import MusicController
57from music_assistant.controllers.player_queues import PlayerQueuesController
58from music_assistant.controllers.players import PlayerController
59from music_assistant.controllers.streams import StreamsController
60from music_assistant.controllers.tasks import TasksController
61from music_assistant.controllers.translations import TranslationController
62from music_assistant.controllers.webserver import WebserverController
63from music_assistant.controllers.webserver.helpers.auth_middleware import (
64    get_current_user,
65    has_scope,
66)
67from music_assistant.helpers.aiohttp_client import create_clientsession
68from music_assistant.helpers.api import APICommandHandler, api_command
69from music_assistant.helpers.diagnostics import install_diagnostics_log_handler
70from music_assistant.helpers.images import detect_provider_icons
71from music_assistant.helpers.util import (
72    TaskManager,
73    get_package_version,
74    is_hass_supervisor,
75    load_provider_module,
76    warn_if_missing_x86_64_v2,
77)
78from music_assistant.models import ProviderInstanceType
79from music_assistant.models.audio_analysis_provider import AudioAnalysisProvider
80from music_assistant.models.music_provider import MusicProvider
81from music_assistant.models.player_provider import PlayerProvider
82
83if TYPE_CHECKING:
84    from types import TracebackType
85
86    from aiohttp import ClientSession
87    from music_assistant_models.config_entries import ProviderConfig
88
89    from music_assistant.models.core_controller import CoreController
90
91isdir = wrap(os.path.isdir)
92isfile = wrap(os.path.isfile)
93mkdirs = wrap(os.makedirs)
94rmfile = wrap(os.remove)
95listdir = wrap(os.listdir)
96rename = wrap(os.rename)
97
98EventCallBackType = Callable[[MassEvent], None] | Callable[[MassEvent], Coroutine[Any, Any, None]]
99EventSubscriptionType = tuple[
100    EventCallBackType, tuple[EventType, ...] | None, tuple[str, ...] | None, bool
101]
102
103LOGGER = logging.getLogger(MASS_LOGGER_NAME)
104
105BASE_DIR = os.path.dirname(os.path.abspath(__file__))
106PROVIDERS_PATH = os.path.join(BASE_DIR, "providers")
107# These bounds guard against a wedged provider, they are not a performance budget: several
108# providers load at once on a busy event loop, so a step can take much longer in wall clock
109# time than it takes on its own. Keep them generous enough that a slow host never trips them.
110PROVIDER_SETUP_TIMEOUT = 120
111# Generous enough for the slowest hosts to load their ML models, but bounded so a wedged
112# provider fails to load instead of holding up startup forever.
113PROVIDER_ASYNC_INIT_TIMEOUT = 300
114PROVIDER_LOAD_CONCURRENCY = 8
115
116_R = TypeVar("_R")
117_ProviderT = TypeVar("_ProviderT", bound=ProviderInstanceType)
118
119
120def is_music_provider(provider: ProviderInstanceType) -> TypeGuard[MusicProvider]:
121    """Type guard that returns true if a provider is a music provider."""
122    return provider.type == ProviderType.MUSIC
123
124
125def is_player_provider(provider: ProviderInstanceType) -> TypeGuard[PlayerProvider]:
126    """Type guard that returns true if a provider is a player provider."""
127    return provider.type == ProviderType.PLAYER
128
129
130def is_audio_analysis_provider(
131    provider: ProviderInstanceType,
132) -> TypeGuard[AudioAnalysisProvider]:
133    """Type guard that returns true if a provider is an audio analysis provider."""
134    return provider.type == ProviderType.AUDIO_ANALYSIS
135
136
137def _provider_error_from_exc(exc: BaseException) -> ProviderError:
138    """Build a serializable, localizable ProviderError from a provider setup exception."""
139    message = str(exc) or type(exc).__name__
140    if isinstance(exc, MusicAssistantError):
141        return ProviderError(
142            error_code=exc.error_code,
143            message=message,
144            translation_key=exc.translation_key,
145            translation_args=list(exc.translation_args),
146            translation_owner=exc.translation_owner,
147        )
148    return ProviderError(error_code=999, message=message)
149
150
151def _provider_error_traceback(exc: BaseException) -> BaseException | None:
152    """Return the exception to log a traceback for, or None when its message says enough."""
153    # a handled condition (auth required, unsupported system, ...) explains itself, but anything
154    # unexpected - or a setup failure wrapping an underlying error - can only be diagnosed from a
155    # traceback, and by the time it is reported the user rarely still has verbose logging on
156    if not isinstance(exc, MusicAssistantError) or exc.__cause__ is not None:
157        return exc
158    return exc if LOGGER.isEnabledFor(VERBOSE_LOG_LEVEL) else None
159
160
161@asynccontextmanager
162async def _provider_load_step(
163    domain: str, action: str, timeout: int | None = None
164) -> AsyncIterator[None]:
165    """
166    Name a provider load step, so any failure in it surfaces as a usable setup failure.
167
168    :param domain: Domain of the provider being loaded, used in the error message.
169    :param action: Verb describing the step, used in the error message.
170    :param timeout: Seconds to allow the step before it is treated as failed, if bounded.
171    """
172    timeout_cm: asyncio.Timeout | None = None
173    try:
174        if timeout is None:
175            yield
176        else:
177            async with asyncio.timeout(timeout) as timeout_cm:
178                yield
179    except TimeoutError as err:
180        if timeout_cm is not None and timeout_cm.expired():
181            msg = f"Provider {domain} did not {action} within {timeout} seconds"
182        else:
183            # a timeout from the provider's own code (an http call, say) carries no message
184            # of its own: name the step it happened in instead of blaming our own bound
185            msg = f"Provider {domain} timed out while trying to {action}"
186        raise SetupFailedError(msg) from err
187    except MusicAssistantError:
188        # already carries a message (and a translation key) meant for the user
189        raise
190    except Exception as err:
191        if str(err):
192            raise
193        # an exception without a message (a bare TimeoutError from an http call, say) would
194        # otherwise reach the user as nothing but its class name, with no hint of what failed
195        msg = f"Provider {domain} failed to {action}: {type(err).__name__}"
196        raise SetupFailedError(msg) from err
197
198
199class MusicAssistant:
200    """Main MusicAssistant (Server) object."""
201
202    loop: asyncio.AbstractEventLoop
203    config: ConfigController
204    webserver: WebserverController
205    cache: CacheController
206    metadata: MetaDataController
207    tasks: TasksController
208    music: MusicController
209    players: PlayerController
210    player_queues: PlayerQueuesController
211    discovery: DiscoveryController
212    streams: StreamsController
213    translations: TranslationController
214    diagnostics: DiagnosticsController
215    dashboard: DashboardController
216
217    def __init__(self, storage_path: str, cache_path: str, safe_mode: bool = False) -> None:
218        """Initialize the MusicAssistant Server."""
219        self._state = CoreState.STARTING
220        self.storage_path = storage_path
221        self.cache_path = cache_path
222        # Sqlite spills temp files (sort scratch, the VACUUM rebuild copy) to /tmp by
223        # default, which is a RAM-backed tmpfs on HAOS - redirect to the data volume.
224        os.environ.setdefault("SQLITE_TMPDIR", storage_path)
225        self.safe_mode = safe_mode
226        # we dynamically register command handlers which can be consumed by the apis
227        self.command_handlers: dict[str, APICommandHandler] = {}
228        self._subscribers: set[EventSubscriptionType] = set()
229        self._provider_manifests: dict[str, ProviderManifest] = {}
230        self._provider_icons: dict[str, dict[ProviderIconVariant, tuple[str, bytes]]] = {}
231        self._providers: dict[str, ProviderInstanceType] = {}
232        self._tracked_tasks: dict[str, asyncio.Task[Any]] = {}
233        self._tracked_timers: dict[str, asyncio.TimerHandle] = {}
234        self._provider_ready_events: dict[str, asyncio.Event] = {}
235        self.running_as_hass_addon: bool = False
236        self.version: str = "0.0.0"
237        self.logger = LOGGER
238        self.dev_mode = (
239            os.environ.get("PYTHONDEVMODE") == "1"
240            or pathlib.Path(__file__).parent.resolve().parent.resolve().joinpath(".venv").exists()
241        )
242        self._http_session: ClientSession | None = None
243        self._http_session_no_ssl: ClientSession | None = None
244
245    async def start(self) -> None:
246        """Start running the Music Assistant server."""
247        self.loop = asyncio.get_running_loop()
248        # start() runs on the event loop thread, so this is the loop's thread id.
249        self.loop_thread_id = threading.get_ident()
250        # ensure the always-on diagnostics capture handler is installed as early as
251        # possible so boot-time errors are captured (idempotent, also installed by
252        # __main__ but embedded usage boots the server directly)
253        install_diagnostics_log_handler()
254        self.running_as_hass_addon = await is_hass_supervisor()
255        self.version = await get_package_version("music_assistant") or "0.0.0"
256        # setup config controller first and fetch important config values
257        self.config = ConfigController(self)
258        await self.config.setup()
259        self.discovery = DiscoveryController(self)
260        # load all available providers from manifest files
261        await self.__load_provider_manifests()
262        # setup/migrate storage
263        await self._setup_storage()
264        LOGGER.info(
265            "Starting Music Assistant Server (%s) version %s - HA add-on: %s - Safe mode: %s",
266            self.server_id,
267            self.version,
268            self.running_as_hass_addon,
269            self.safe_mode,
270        )
271        await warn_if_missing_x86_64_v2(LOGGER)
272        # setup other core controllers
273        await self._load_core_controllers()
274
275        # setup all core controllers in parallel
276        async def setup_controller(controller: CoreController) -> None:
277            config = await self.config.get_core_config(controller.domain)
278            # keep the active config on the controller so internal code can read
279            # config values without rebuilding the config entries
280            controller.config = config
281            await controller.setup(config)
282            controller.initialized.set()
283
284        # set up the translations catalog first so it is ready before any object is serialized
285        await setup_controller(self.translations)
286
287        async with asyncio.TaskGroup() as tg:
288            tg.create_task(setup_controller(self.cache))
289            tg.create_task(setup_controller(self.tasks))
290            tg.create_task(setup_controller(self.streams))
291            tg.create_task(setup_controller(self.music))
292            tg.create_task(setup_controller(self.metadata))
293            tg.create_task(setup_controller(self.players))
294            tg.create_task(setup_controller(self.player_queues))
295            tg.create_task(setup_controller(self.diagnostics))
296            tg.create_task(setup_controller(self.dashboard))
297
298        for controller_name in (
299            "cache",
300            "tasks",
301            "streams",
302            "music",
303            "metadata",
304            "players",
305            "player_queues",
306        ):
307            await cast("CoreController", getattr(self, controller_name)).post_setup()
308
309        # load webserver/api now that the core controllers are setup and ready to be used
310        self._register_api_commands()
311        webserver_config = await self.config.get_core_config("webserver")
312        self.webserver.config = webserver_config
313        await self.webserver.setup(webserver_config)
314        await setup_controller(self.discovery)
315        # load builtin providers (always needed, also in safe mode)
316        await self._load_builtin_providers()
317        # load regular providers (skip when in safe mode)
318        # providers are loaded in background tasks so they won't block
319        # the startup if they fail or take a long time to load
320        if not self.safe_mode:
321            await self._load_providers()
322        # at this point we are fully up and running,
323        # set state to running to signal we're ready
324        self._set_state(CoreState.RUNNING)
325
326    async def stop(self) -> None:
327        """Stop running the music assistant server."""
328        LOGGER.info("Stop called, cleaning up...")
329        # set state to stopping to signal we're shutting down
330        self._set_state(CoreState.STOPPING)
331        # cancel all running tasks
332        for task in list(self._tracked_tasks.values()):
333            task.cancel()
334        # cleanup all providers
335        await asyncio.gather(
336            *[self.unload_provider(prov_id) for prov_id in list(self._providers.keys())],
337            return_exceptions=True,
338        )
339        # stop core controllers, cache and config last because the others rely on them.
340        # a failed startup may not have created (or fully set up) every controller, so
341        # each one is closed independently: leaving a database open here would keep its
342        # worker thread alive and stop the process from ever exiting.
343        for controller_name in (
344            "discovery",
345            "streams",
346            "webserver",
347            "tasks",
348            "metadata",
349            "music",
350            "player_queues",
351            "players",
352            "translations",
353            "diagnostics",
354            "dashboard",
355            "config",
356            "cache",
357        ):
358            if (controller := getattr(self, controller_name, None)) is None:
359                continue
360            try:
361                await controller.close()
362            except Exception:
363                LOGGER.exception("Error while closing the %s controller", controller_name)
364        # close/cleanup shared http sessions
365        if self._http_session and not self._http_session.closed:
366            await self._http_session.close()
367        if self._http_session_no_ssl and not self._http_session_no_ssl.closed:
368            await self._http_session_no_ssl.close()
369        self._set_state(CoreState.STOPPED)
370
371    @property
372    def state(self) -> CoreState:
373        """Return current state of the core."""
374        return self._state
375
376    @property
377    def closing(self) -> bool:
378        """Return true if the server is (in the process of) closing."""
379        return self._state in (CoreState.STOPPING, CoreState.STOPPED)
380
381    @property
382    def server_id(self) -> str:
383        """Return unique ID of this server."""
384        if not self.config.initialized:
385            return ""
386        return self.config.get(CONF_SERVER_ID)  # type: ignore[no-any-return]
387
388    @property
389    def http_session(self) -> ClientSession:
390        """
391        Return the shared HTTP Client session (with SSL).
392
393        NOTE: May only be called from the event loop.
394        """
395        if self._http_session is None:
396            self._http_session = create_clientsession(self, verify_ssl=True)
397        return self._http_session
398
399    @property
400    def http_session_no_ssl(self) -> ClientSession:
401        """
402        Return the shared HTTP Client session (without SSL).
403
404        NOTE: May only be called from the event loop thread.
405        """
406        if self._http_session_no_ssl is None:
407            self._http_session_no_ssl = create_clientsession(self, verify_ssl=False)
408        return self._http_session_no_ssl
409
410    @api_command("info")
411    def get_server_info(self) -> ServerInfoMessage:
412        """Return Info of this server."""
413        return ServerInfoMessage(
414            server_id=self.server_id,
415            server_version=self.version,
416            schema_version=API_SCHEMA_VERSION,
417            min_supported_schema_version=MIN_SCHEMA_VERSION,
418            base_url=self.webserver.base_url,
419            homeassistant_addon=self.running_as_hass_addon,
420            onboard_done=self.config.onboard_done,
421            status=self._state,
422        )
423
424    @api_command("time", authenticated=False)
425    def get_server_time(self) -> float:
426        """
427        Return the current server time as UTC timestamp (seconds since epoch).
428
429        Clients compare server-provided timestamps (such as `elapsed_time_last_updated`)
430        against their own clock. Round-tripping this command lets a client estimate the
431        offset between the two clocks and correct for it, so a device with an unsynced
432        clock still renders playback progress and countdowns correctly.
433        """
434        return time.time()
435
436    @api_command("providers/manifests", required_scope=Scope.PROVIDERS_READ)
437    def get_provider_manifests(self) -> list[ProviderManifest]:
438        """Return all Provider manifests."""
439        return list(self._provider_manifests.values())
440
441    @api_command("providers/manifests/get", required_scope=Scope.PROVIDERS_READ)
442    def get_provider_manifest(self, instance_id_or_domain: str) -> ProviderManifest:
443        """Return Provider manifests of single provider(domain)."""
444        if instance_id_or_domain in self._provider_manifests:
445            return self._provider_manifests[instance_id_or_domain]
446        if provider := self.get_provider(instance_id_or_domain, return_unavailable=True):
447            return provider.manifest
448        raise KeyError(f"Provider manifest not found for {instance_id_or_domain}")
449
450    @api_command("providers/icon", required_scope=Scope.PROVIDERS_READ)
451    def get_provider_icon_data(
452        self,
453        provider: str,
454        variant: ProviderIconVariant = ProviderIconVariant.DEFAULT,
455    ) -> str | None:
456        """
457        Return a provider icon variant as a base64 data URI.
458
459        :param provider: A provider domain or instance id.
460        :param variant: Which icon variant to return.
461        """
462        icon = self.get_provider_icon(provider, variant)
463        if icon is None:
464            return None
465        mime, data = icon
466        return f"data:{mime};base64,{b64encode(data).decode('ascii')}"
467
468    def get_provider_icon(
469        self,
470        provider: str,
471        variant: ProviderIconVariant = ProviderIconVariant.DEFAULT,
472    ) -> tuple[str, bytes] | None:
473        """
474        Return the (mime, bytes) for a provider icon variant.
475
476        :param provider: A provider domain or instance id.
477        :param variant: Which icon variant to return.
478        """
479        domain = provider
480        if domain not in self._provider_icons:
481            try:
482                domain = self.get_provider_manifest(provider).domain
483            except KeyError:
484                return None
485        icons = self._provider_icons.get(domain)
486        if not icons:
487            return None
488        return icons.get(variant)
489
490    @api_command("providers", required_scope=Scope.PROVIDERS_READ)
491    def get_providers(
492        self, provider_type: ProviderType | None = None
493    ) -> list[ProviderInstanceType]:
494        """
495        Return all loaded/running Providers (instances).
496
497        Optionally filtered by ProviderType.
498        Note that this applies user filters for music providers (for non admin users).
499        """
500        user = get_current_user()
501        user_provider_filter = (
502            user.provider_filter if user and not has_scope(user, Scope.ALL) else None
503        )
504        return [
505            x
506            for x in list(self._providers.values())
507            if (provider_type is None or provider_type == x.type)
508            # apply user provider filter
509            and (
510                not user_provider_filter
511                or x.instance_id in user_provider_filter
512                or x.type != ProviderType.MUSIC
513            )
514        ]
515
516    @api_command("logging/get", required_scope=Scope.SYSTEM_MANAGE)
517    async def get_application_log(self) -> str:
518        """Return the application log from file."""
519        logfile = os.path.join(self.storage_path, "musicassistant.log")
520        async with aiofiles.open(logfile) as _file:
521            return str(await _file.read())
522
523    @property
524    def providers(self) -> list[ProviderInstanceType]:
525        """
526        Return all loaded/running Providers (instances).
527
528        Note that this skips user filters so may only be called from internal code.
529        """
530        return list(self._providers.values())
531
532    @overload
533    def get_provider(
534        self,
535        provider_instance_or_domain: str,
536        return_unavailable: bool = False,
537        provider_type: None = None,
538    ) -> ProviderInstanceType | None: ...
539
540    @overload
541    def get_provider(
542        self,
543        provider_instance_or_domain: str,
544        return_unavailable: bool = False,
545        *,
546        provider_type: type[_ProviderT],
547    ) -> _ProviderT | None: ...
548
549    def get_provider(
550        self,
551        provider_instance_or_domain: str,
552        return_unavailable: bool = False,
553        provider_type: type[_ProviderT] | None = None,
554    ) -> ProviderInstanceType | _ProviderT | None:
555        """
556        Return provider by instance id or domain.
557
558        :param provider_instance_or_domain: Instance ID or domain of the provider.
559        :param return_unavailable: Also return unavailable providers.
560        :param provider_type: Optional type hint for the expected provider type (unused at runtime).
561        """
562        # lookup by instance_id first
563        if prov := self._providers.get(provider_instance_or_domain):
564            if return_unavailable or prov.available:
565                return prov
566            if not getattr(prov, "is_streaming_provider", None):
567                # no need to lookup other instances because this provider has unique data
568                return None
569            provider_instance_or_domain = prov.domain
570        # fallback to match on domain
571        for prov in list(self._providers.values()):
572            if prov.domain != provider_instance_or_domain:
573                continue
574            if return_unavailable or prov.available:
575                return prov
576        return None
577
578    def get_provider_ready_event(self, domain: str) -> asyncio.Event:
579        """Get (or create) an asyncio.Event that is set when a provider of the given domain is loaded."""
580        if domain not in self._provider_ready_events:
581            self._provider_ready_events[domain] = asyncio.Event()
582        return self._provider_ready_events[domain]
583
584    def get_provider_instances(
585        self,
586        domain: str,
587        return_unavailable: bool = False,
588        provider_type: ProviderType | None = None,
589    ) -> list[ProviderInstanceType]:
590        """
591        Return all provider instances for a given domain.
592
593        Note that this skips user filters so may only be called from internal code.
594        """
595        return [
596            prov
597            for prov in list(self._providers.values())
598            if (provider_type is None or provider_type == prov.type)
599            and prov.domain == domain
600            and (return_unavailable or prov.available)
601        ]
602
603    def get_providers_supporting_feature(
604        self,
605        feature: ProviderFeature,
606        priority: tuple[ProviderType, ...] = (
607            ProviderType.MUSIC,
608            ProviderType.METADATA,
609            ProviderType.PLUGIN,
610        ),
611    ) -> list[ProviderInstanceType]:
612        """
613        Return all available providers that support the given feature.
614
615        Results are grouped by provider type in the order given by ``priority``,
616        and sorted within each tier by the provider's ``priority`` attribute
617        (lower value = higher priority).
618
619        :param feature: The ProviderFeature to query for.
620        :param priority: Ordered tuple of ProviderType values indicating tier order.
621            Types omitted from this tuple are excluded from the results.
622        """
623        by_tier: dict[ProviderType, list[ProviderInstanceType]] = {ptype: [] for ptype in priority}
624        for prov in self.get_providers():
625            if not prov.available:
626                continue
627            if prov.type not in by_tier:
628                continue
629            if feature not in prov.supported_features:
630                continue
631            by_tier[prov.type].append(prov)
632        result: list[ProviderInstanceType] = []
633        for ptype in priority:
634            result.extend(sorted(by_tier[ptype], key=lambda p: getattr(p, "priority", 50)))
635        return result
636
637    def signal_event(
638        self,
639        event: EventType,
640        object_id: str | None = None,
641        data: Any = None,
642    ) -> None:
643        """Signal event to subscribers."""
644        if self.closing:
645            return
646
647        self.verify_event_loop_thread("signal_event")
648
649        if LOGGER.isEnabledFor(VERBOSE_LOG_LEVEL):
650            # do not log queue time updated events because that is too chatty
651            LOGGER.getChild("event").log(VERBOSE_LOG_LEVEL, "%s %s", event.value, object_id or "")
652
653        event_obj = MassEvent(event=event, object_id=object_id, data=data)
654        for cb_func, event_filter, id_filter, is_coro in list(self._subscribers):
655            if not (event_filter is None or event in event_filter):
656                continue
657            if not (id_filter is None or object_id in id_filter):
658                continue
659            if is_coro:
660                if TYPE_CHECKING:
661                    cb_func = cast("Callable[[MassEvent], Coroutine[Any, Any, None]]", cb_func)
662                self.create_task(cb_func, event_obj)
663            else:
664                if TYPE_CHECKING:
665                    cb_func = cast("Callable[[MassEvent], None]", cb_func)
666                self.loop.call_soon(cb_func, event_obj)
667
668    def subscribe(
669        self,
670        cb_func: EventCallBackType,
671        event_filter: EventType | tuple[EventType, ...] | None = None,
672        id_filter: str | tuple[str, ...] | None = None,
673    ) -> Callable[[], None]:
674        """
675        Add callback to event listeners.
676
677        Returns function to remove the listener.
678            :param cb_func: callback function or coroutine
679            :param event_filter: Optionally only listen for these events
680            :param id_filter: Optionally only listen for these id's (player_id, queue_id, uri)
681        """
682        if isinstance(event_filter, EventType):
683            event_filter = (event_filter,)
684        if isinstance(id_filter, str):
685            id_filter = (id_filter,)
686        # precompute whether the callback is a coroutine so signal_event does not have to
687        # re-derive it via reflection for every subscriber on every (high-frequency) event
688        listener = (cb_func, event_filter, id_filter, inspect.iscoroutinefunction(cb_func))
689        self._subscribers.add(listener)
690
691        def remove_listener() -> None:
692            self._subscribers.remove(listener)
693
694        return remove_listener
695
696    def create_task(
697        self,
698        target: Callable[..., Coroutine[Any, Any, _R]] | Awaitable[_R],
699        *args: Any,
700        task_id: str | None = None,
701        abort_existing: bool = False,
702        eager_start: bool = True,
703        **kwargs: Any,
704    ) -> asyncio.Task[_R]:
705        """
706        Create Task on (main) event loop from Coroutine(function).
707
708        Tasks created by this helper will be properly cancelled on stop.
709
710        :param target: Coroutine function or awaitable to run as a task.
711        :param args: Arguments to pass to the coroutine function.
712        :param task_id: Optional ID to track and deduplicate tasks.
713        :param abort_existing: If True, cancel existing task with same task_id.
714        :param eager_start: If True (default), start task immediately without waiting
715                           for next event loop iteration. This ensures proper ordering
716                           when creating multiple tasks in sequence.
717        :param kwargs: Keyword arguments to pass to the coroutine function.
718        """
719        if task_id and (existing := self._tracked_tasks.get(task_id)) and not existing.done():
720            # prevent duplicate tasks if task_id is given and already present
721            if abort_existing:
722                existing.cancel()
723            else:
724                # close any already-constructed coroutine to avoid "never awaited" warning
725                if inspect.iscoroutine(target):
726                    target.close()
727                return existing
728        self.verify_event_loop_thread("create_task")
729
730        if inspect.iscoroutinefunction(target):
731            # coroutine function
732            coro = target(*args, **kwargs)
733        elif inspect.iscoroutine(target):
734            # coroutine
735            coro = target
736        elif callable(target):
737            raise RuntimeError("Function is not a coroutine or coroutine function")
738        else:
739            raise RuntimeError("Target is missing")
740
741        # Use asyncio.Task directly with eager_start for immediate execution
742        task: asyncio.Task[_R] = asyncio.Task(coro, loop=self.loop, eager_start=eager_start)
743
744        if task_id is None:
745            task_id = uuid4().hex
746
747        def task_done_callback(_task: asyncio.Task[Any]) -> None:
748            # done callbacks run one event loop iteration after the task finished, so a
749            # caller may already have replaced the entry with a new task under the same
750            # task_id - only untrack when the entry still points at this task
751            if self._tracked_tasks.get(task_id) is _task:
752                del self._tracked_tasks[task_id]
753            if _task.cancelled():
754                return
755            # always retrieve the exception, otherwise asyncio logs a noisy
756            # "Task exception was never retrieved" error at garbage collection time
757            if err := _task.exception():
758                task_name = _task.get_name() if hasattr(_task, "get_name") else str(_task)
759                LOGGER.warning(
760                    "Exception in task %s - target: %s: %s",
761                    task_name,
762                    str(target),
763                    str(err),
764                    exc_info=err if LOGGER.isEnabledFor(logging.DEBUG) else None,
765                )
766
767        self._tracked_tasks[task_id] = task
768        task.add_done_callback(task_done_callback)
769        return task
770
771    def call_later(
772        self,
773        delay: float,
774        target: Coroutine[Any, Any, _R] | Awaitable[_R] | Callable[..., _R],
775        *args: Any,
776        task_id: str | None = None,
777        **kwargs: Any,
778    ) -> asyncio.TimerHandle:
779        """
780        Run callable/awaitable after given delay.
781
782        Use task_id for debouncing.
783        """
784        self.verify_event_loop_thread("call_later")
785
786        if not task_id:
787            task_id = uuid4().hex
788
789        if existing := self._tracked_timers.get(task_id):
790            existing.cancel()
791
792        def _create_task(_target: Coroutine[Any, Any, _R]) -> None:
793            self._tracked_timers.pop(task_id)
794            self.create_task(_target, *args, task_id=task_id, abort_existing=True, **kwargs)
795
796        def _call_sync(_target: Callable[..., _R]) -> None:
797            self._tracked_timers.pop(task_id)
798            _target(*args, **kwargs)
799
800        if inspect.iscoroutinefunction(target) or inspect.iscoroutine(target):
801            # coroutine function
802            if TYPE_CHECKING:
803                target = cast("Coroutine[Any, Any, _R]", target)
804            handle = self.loop.call_later(delay, _create_task, target)
805        else:
806            # regular sync callable
807            if TYPE_CHECKING:
808                target = cast("Callable[..., _R]", target)
809            handle = self.loop.call_later(delay, _call_sync, target)
810        self._tracked_timers[task_id] = handle
811        return handle
812
813    def get_task(self, task_id: str) -> asyncio.Task[Any] | None:
814        """Get existing scheduled task."""
815        if existing := self._tracked_tasks.get(task_id):
816            # prevent duplicate tasks if task_id is given and already present
817            return existing
818        return None
819
820    def cancel_task(self, task_id: str) -> None:
821        """Cancel existing scheduled task."""
822        if existing := self._tracked_tasks.pop(task_id, None):
823            existing.cancel()
824
825    def cancel_timer(self, task_id: str) -> None:
826        """Cancel existing scheduled timer."""
827        if existing := self._tracked_timers.pop(task_id, None):
828            existing.cancel()
829
830    def register_api_command(
831        self,
832        command: str,
833        handler: Callable[..., Coroutine[Any, Any, Any] | AsyncGenerator[Any, Any]],
834        authenticated: bool = True,
835        required_scope: Scope | None = None,
836        allow_impersonation: bool = False,
837        alias: bool = False,
838    ) -> Callable[[], None]:
839        """
840        Dynamically register a command on the API.
841
842        :param command: The command name/path.
843        :param handler: The function to handle the command.
844        :param authenticated: Whether authentication is required (default: True).
845        :param required_scope: Scope required to execute the command,
846            None means any authenticated user.
847        :param allow_impersonation: Whether the command accepts a 'user' argument
848            to execute the command on behalf of another user (default: False).
849        :param alias: Whether this is an alias for backward compatibility (default: False).
850            Aliases are not shown in API documentation but remain functional.
851
852        Returns handle to unregister.
853        """
854        if command in self.command_handlers:
855            msg = f"Command {command} is already registered"
856            raise RuntimeError(msg)
857        self.command_handlers[command] = APICommandHandler.parse(
858            command, handler, authenticated, required_scope, allow_impersonation, alias
859        )
860
861        def unregister() -> None:
862            self.command_handlers.pop(command, None)
863
864        return unregister
865
866    async def load_provider_config(
867        self,
868        prov_conf: ProviderConfig,
869    ) -> None:
870        """Load (or reload) a provider from its config, recording any load failure."""
871        # cancel existing (re)load timer if needed
872        task_id = f"load_provider_{prov_conf.instance_id}"
873        if existing := self._tracked_timers.pop(task_id, None):
874            existing.cancel()
875
876        try:
877            await self._load_provider(prov_conf)
878        except Exception as exc:
879            # persist the failure so the provider surfaces a clear status (e.g. auth_required)
880            # to the UI instead of appearing stuck loading, then propagate to the caller
881            self.config.update_provider_last_error(
882                prov_conf.instance_id, _provider_error_from_exc(exc)
883            )
884            raise
885
886        # (re)load any dependents. The provider itself is loaded at this point, so a problem
887        # in this scan belongs to a dependent (or to nothing at all) and must never be
888        # recorded against - and thus flag - the provider we just loaded successfully.
889        try:
890            # resolving option values here would call get_config_entries() on every loaded
891            # provider (some of which do network i/o), for values _load_provider does not
892            # read: it seeds the stored raw values itself and rehydrates once the instance
893            # exists. Only the manifest-related fields below are needed to spot a dependent.
894            prov_configs = await self.config.get_provider_configs()
895        except Exception as exc:
896            LOGGER.warning(
897                "Error looking up dependents of provider(instance) %s: %s",
898                prov_conf.name or prov_conf.instance_id,
899                str(exc) or exc.__class__.__name__,
900                exc_info=_provider_error_traceback(exc),
901            )
902            return
903        for dep_prov_conf in prov_configs:
904            if not dep_prov_conf.enabled:
905                continue
906            manifest = self.get_provider_manifest(dep_prov_conf.domain)
907            if not manifest.depends_on:
908                continue
909            if manifest.depends_on != prov_conf.domain:
910                continue
911            try:
912                # the scan above skipped the config values, but the load path does need them:
913                # a provider reads config (e.g. its log level) while it is being constructed.
914                # Resolve them here, for this single dependent instead of for every provider.
915                dep_conf = await self.config.get_provider_config(dep_prov_conf.instance_id)
916            except KeyError:
917                # config was removed while we were scanning
918                continue
919            try:
920                await self._load_provider(dep_conf)
921            except Exception as exc:
922                # record the failure against the provider that hit it: attributing it to the
923                # provider we just loaded (which is fine) flags the wrong one in the UI
924                self.config.update_provider_last_error(
925                    dep_prov_conf.instance_id, _provider_error_from_exc(exc)
926                )
927                LOGGER.warning(
928                    "Error loading provider(instance) %s: %s",
929                    dep_prov_conf.name or dep_prov_conf.instance_id,
930                    str(exc) or exc.__class__.__name__,
931                    exc_info=_provider_error_traceback(exc),
932                )
933
934    async def load_provider(
935        self,
936        instance_id: str,
937        allow_retry: bool = False,
938        remove_if_unsupported: bool = False,
939    ) -> None:
940        """Try to load a provider and catch errors."""
941        try:
942            prov_conf = await self.config.get_provider_config(instance_id)
943        except KeyError:
944            # Was deleted before we could run
945            return
946
947        if not prov_conf.enabled:
948            # Was disabled before we could run
949            return
950
951        # cancel existing (re)load timer if needed
952        task_id = f"load_provider_{instance_id}"
953        if existing := self._tracked_timers.pop(task_id, None):
954            existing.cancel()
955
956        try:
957            await self.load_provider_config(prov_conf)
958        except UnsupportedSystemError as exc:
959            # The host does not meet this provider's hardware requirements. This is a
960            # permanent condition, so we never retry. For a provider that was just
961            # auto-set-up as a default, drop the config again so it does not linger as a
962            # broken provider (it stays marked done so it is not auto-created again).
963            if remove_if_unsupported:
964                LOGGER.info(
965                    "Not enabling default provider %s: %s",
966                    prov_conf.name or prov_conf.instance_id,
967                    exc,
968                )
969                # The provider never loaded, so just drop its auto-created config key.
970                # (remove_provider_config refuses builtin providers and runs loaded-provider
971                # cleanup we don't need here; a direct remove persists and is guard-free.)
972                self.config.remove(f"{CONF_PROVIDERS}/{instance_id}")
973                return
974            prov_conf.last_error = _provider_error_from_exc(exc)
975            self.config.update_provider_last_error(instance_id, prov_conf.last_error)
976            LOGGER.warning(
977                "Provider(instance) %s can not run on this system: %s",
978                prov_conf.name or prov_conf.instance_id,
979                exc,
980            )
981            return
982        except Exception as exc:
983            # if loading failed, we store the error in the config object
984            # so we can show something useful to the user
985            prov_conf.last_error = _provider_error_from_exc(exc)
986            self.config.update_provider_last_error(instance_id, prov_conf.last_error)
987
988            # auto schedule a retry if the (re)load failed with a handled exception
989            # unhandled exceptions (e.g. ValueError) are likely bugs that won't resolve themselves
990            will_retry = allow_retry and isinstance(exc, MusicAssistantError)
991            if will_retry:
992                self.call_later(
993                    120,
994                    self.load_provider,
995                    instance_id,
996                    allow_retry,
997                    task_id=task_id,
998                )
999            LOGGER.warning(
1000                "Error loading provider(instance) %s: %s%s",
1001                prov_conf.name or prov_conf.instance_id,
1002                str(exc) or exc.__class__.__name__,
1003                " (will be retried later)" if will_retry else "",
1004                exc_info=_provider_error_traceback(exc),
1005            )
1006            return
1007
1008        # (re)load any dependents if needed
1009        for dep_prov in self.providers:
1010            if dep_prov.available:
1011                continue
1012            if dep_prov.manifest.depends_on == prov_conf.domain:
1013                await self.unload_provider(dep_prov.instance_id)
1014
1015    async def unload_provider(self, instance_id: str, is_removed: bool = False) -> None:
1016        """Unload a provider."""
1017        # this waits (bounded) for a running sync to unwind: provider.unload() below tears
1018        # down state the sync may still be using, such as the mount of a network share
1019        await self.music.unschedule_provider_sync(instance_id, clear_persisted_state=is_removed)
1020        if provider := self._providers.get(instance_id):
1021            if isinstance(provider, PlayerProvider):
1022                await self.players.on_provider_unload(provider)
1023            if isinstance(provider, MusicProvider):
1024                await self.music.on_provider_unload(provider)
1025            # check if there are no other providers dependent of this provider
1026            for dep_prov in self.providers:
1027                if dep_prov.manifest.depends_on == provider.domain:
1028                    await self.unload_provider(dep_prov.instance_id)
1029            if is_player_provider(provider):
1030                # unregister all players of this provider
1031                for player in provider.players:
1032                    await self.players.unregister(player.player_id, permanent=is_removed)
1033            try:
1034                await provider.unload(is_removed)
1035            except Exception as err:
1036                LOGGER.warning(
1037                    "Error while unloading provider %s: %s", provider.name, str(err), exc_info=err
1038                )
1039            finally:
1040                if provider.domain in self._provider_ready_events:
1041                    self._provider_ready_events[provider.domain].clear()
1042                self._providers.pop(instance_id, None)
1043                self.discovery.on_provider_unload(instance_id)
1044                await self._update_available_providers_cache()
1045                self.signal_event(EventType.PROVIDERS_UPDATED, data=self.get_providers())
1046
1047    async def unload_provider_with_error(self, instance_id: str, error: str | Exception) -> None:
1048        """
1049        Unload a provider that hit a problem which needs user interaction.
1050
1051        :param error: The originating exception (preferred, so e.g. a LoginFailed surfaces as an
1052            auth-required status with a localized message) or a plain string for a generic error.
1053        """
1054        prov_error = (
1055            _provider_error_from_exc(error)
1056            if isinstance(error, Exception)
1057            else ProviderError(error_code=999, message=error)
1058        )
1059        self.config.update_provider_last_error(instance_id, prov_error)
1060        await self.unload_provider(instance_id)
1061
1062    async def run_provider_discovery(self, instance_id: str) -> None:
1063        """
1064        Run shared discovery for a given provider.
1065
1066        In case of a PlayerProvider, will also call its own discovery method.
1067        """
1068        provider = self.get_provider(instance_id, return_unavailable=False)
1069        if not provider:
1070            raise KeyError(f"Provider with instance ID {instance_id} not found")
1071        await self.discovery.run_provider_discovery(provider)
1072        if isinstance(provider, PlayerProvider):
1073            await provider.discover_players()
1074
1075    def verify_event_loop_thread(self, what: str) -> None:
1076        """Report and raise if we are not running in the event loop thread."""
1077        if self.loop_thread_id != threading.get_ident():
1078            raise RuntimeError(
1079                f"Non-Async operation detected: {what} may only be called from the eventloop."
1080            )
1081
1082    async def __aenter__(self) -> Self:
1083        """Return Context manager."""
1084        await self.start()
1085        return self
1086
1087    async def __aexit__(
1088        self,
1089        exc_type: type[BaseException] | None,
1090        exc_val: BaseException | None,
1091        exc_tb: TracebackType | None,
1092    ) -> bool | None:
1093        """Exit context manager."""
1094        await self.stop()
1095        return None
1096
1097    def _register_api_commands(self) -> None:
1098        """Register all methods decorated as api_command within a class(instance)."""
1099        for cls in (
1100            self,
1101            self.config,
1102            self.metadata,
1103            self.tasks,
1104            self.music,
1105            self.players,
1106            self.player_queues,
1107            self.translations,
1108            self.webserver,
1109            self.webserver.auth,
1110            self.streams.audio_analysis,
1111            self.diagnostics,
1112            self.dashboard,
1113        ):
1114            for attr_name in dir(cls):
1115                if attr_name.startswith("__"):
1116                    continue
1117                # Skip properties to avoid triggering lazy initialization side effects
1118                # (e.g. http_session creating an aiohttp connector during registration)
1119                if isinstance(getattr(type(cls), attr_name, None), property):
1120                    continue
1121                try:
1122                    obj = getattr(cls, attr_name)
1123                except AttributeError, RuntimeError:
1124                    # Skip attributes that fail during initialization
1125                    continue
1126                if hasattr(obj, "api_cmd"):
1127                    # method is decorated with our api decorator
1128                    authenticated = getattr(obj, "api_authenticated", True)
1129                    required_scope = getattr(obj, "api_required_scope", None)
1130                    allow_impersonation = getattr(obj, "api_allow_impersonation", False)
1131                    alias = getattr(obj, "api_alias", False)
1132                    self.register_api_command(
1133                        obj.api_cmd, obj, authenticated, required_scope, allow_impersonation, alias
1134                    )
1135
1136    async def _load_core_controllers(self) -> None:
1137        """Instantiate the core controllers and register their manifests and icons."""
1138        self.cache = CacheController(self)
1139        self.tasks = TasksController(self)
1140        self.webserver = WebserverController(self)
1141        self.metadata = MetaDataController(self)
1142        self.music = MusicController(self)
1143        self.players = PlayerController(self)
1144        self.player_queues = PlayerQueuesController(self)
1145        self.streams = StreamsController(self)
1146        self.translations = TranslationController(self)
1147        self.diagnostics = DiagnosticsController(self)
1148        self.dashboard = DashboardController(self)
1149        # add manifests for core controllers
1150        for controller_name in CONFIGURABLE_CORE_CONTROLLERS:
1151            controller: CoreController = getattr(self, controller_name)
1152            self._provider_manifests[controller.domain] = controller.manifest
1153            # load icon image(s) shipped alongside the controller module
1154            controller_dir = os.path.dirname(inspect.getfile(type(controller)))
1155            if icons := await detect_provider_icons(controller_dir):
1156                self._provider_icons[controller.domain] = icons
1157                controller.manifest.icon_images = list(icons)
1158
1159    async def _load_builtin_providers(self) -> None:
1160        """
1161        Load all builtin providers.
1162
1163        Builtin providers are always needed (also in safe mode) and are fully awaited.
1164        On error, setup will fail.
1165        """
1166        # create default config for any 'builtin' providers
1167        for prov_manifest in self._provider_manifests.values():
1168            if prov_manifest.type == ProviderType.CORE:
1169                # core controllers are not real providers
1170                continue
1171            if not prov_manifest.builtin:
1172                continue
1173            await self.config.create_builtin_provider_config(prov_manifest.domain)
1174
1175        # load all configured (and enabled) builtin providers
1176        # (only manifest-related fields are read here, so the option values are not resolved)
1177        prov_configs = await self.config.get_provider_configs()
1178        builtin_configs: list[ProviderConfig] = [
1179            prov_conf
1180            for prov_conf in prov_configs
1181            if (manifest := self._provider_manifests.get(prov_conf.domain))
1182            and manifest.builtin
1183            and (prov_conf.enabled or manifest.allow_disable is False)
1184        ]
1185
1186        # load builtin providers and wait for them to complete
1187        async with asyncio.TaskGroup() as tg:
1188            for conf in builtin_configs:
1189                tg.create_task(self.load_provider(conf.instance_id, allow_retry=True))
1190
1191    async def _load_providers(self) -> None:
1192        """
1193        Load regular (non-builtin) providers from config.
1194
1195        Regular providers are loaded in background tasks
1196        and can fail without affecting core setup.
1197        """
1198        # handle default providers setup
1199        self.config.set_default(CONF_DEFAULT_PROVIDERS_SETUP, set())
1200        default_providers_setup = set(self.config.get(CONF_DEFAULT_PROVIDERS_SETUP))
1201        changes_made = False
1202        newly_created_defaults: set[str] = set()
1203        for default_provider, require_mdns in DEFAULT_PROVIDERS:
1204            if default_provider in default_providers_setup:
1205                # already processed/setup before, skip
1206                continue
1207            if not (manifest := self._provider_manifests.get(default_provider)):
1208                continue
1209            if require_mdns:
1210                # if mdns discovery is required, check if we have seen any mdns entries
1211                # for this provider before setting it up
1212                for mdns_name in set(self.discovery.aiozc.zeroconf.cache.cache):
1213                    if manifest.mdns_discovery and any(
1214                        mdns_type in mdns_name for mdns_type in manifest.mdns_discovery
1215                    ):
1216                        break
1217                else:
1218                    continue
1219            await self.config.create_builtin_provider_config(manifest.domain)
1220            changes_made = True
1221            newly_created_defaults.add(manifest.domain)
1222            # TEMP: migration - to be removed after 2.8 release
1223            # enable all existing players of the default providers if they are not already enabled
1224            # due to the linked protocol feature we introduced
1225            for player_config in await self.config.get_player_configs(
1226                provider=default_provider, include_disabled=True
1227            ):
1228                if player_config.enabled:
1229                    continue
1230                await self.config.save_player_config(player_config.player_id, {"enabled": True})
1231            default_providers_setup.add(default_provider)
1232        if changes_made:
1233            self.config.set(CONF_DEFAULT_PROVIDERS_SETUP, default_providers_setup)
1234            self.config.save(True)
1235        # load all configured (and enabled) regular (non-builtin) providers
1236        # (only manifest-related fields are read here, so the option values are not resolved)
1237        prov_configs = await self.config.get_provider_configs()
1238        other_configs: list[ProviderConfig] = [
1239            prov_conf
1240            for prov_conf in prov_configs
1241            if prov_conf.enabled
1242            and (
1243                not (manifest := self._provider_manifests.get(prov_conf.domain))
1244                or not manifest.builtin
1245            )
1246        ]
1247        # load providers concurrently via tasks, bounded so a host with many providers does
1248        # not import every provider module at once (a torch-backed one costs hundreds of MB)
1249        async with TaskManager(self, PROVIDER_LOAD_CONCURRENCY) as tg:
1250            for prov_conf in other_configs:
1251                # Use a task so we can load multiple providers at once.
1252                # If a provider fails, that will not block the loading of other providers.
1253                # For providers just auto-set-up as a default, drop the config again if the
1254                # host does not meet their requirements (rather than retry a broken provider).
1255                await tg.create_task_with_limit(
1256                    self.load_provider(
1257                        prov_conf.instance_id,
1258                        allow_retry=True,
1259                        remove_if_unsupported=prov_conf.domain in newly_created_defaults,
1260                    )
1261                )
1262
1263    async def _load_provider(self, conf: ProviderConfig) -> None:
1264        """Load (or reload) a provider."""
1265        # if provider is already loaded, stop and unload it first
1266        await self.unload_provider(conf.instance_id)
1267        LOGGER.debug("Loading provider %s", conf.name or conf.domain)
1268        if not conf.enabled:
1269            msg = "Provider is disabled"
1270            raise SetupFailedError(msg)
1271
1272        # The config is validated after the instance is created and its config rehydrated
1273        # (see below): the full options entries - and thus which values are required - are
1274        # only known once the instance exists.
1275
1276        domain = conf.domain
1277        prov_manifest = self._provider_manifests.get(domain)
1278        # check for other instances of this provider
1279        existing = next((x for x in self.providers if x.domain == domain), None)
1280        if existing and prov_manifest and not prov_manifest.multi_instance:
1281            msg = f"Provider {domain} already loaded and only one instance allowed."
1282            raise SetupFailedError(msg)
1283        # check valid manifest (just in case)
1284        if not prov_manifest:
1285            msg = f"Provider {domain} manifest not found"
1286            raise SetupFailedError(msg)
1287
1288        # handle dependency on other provider
1289        if prov_manifest.depends_on and not self.get_provider(prov_manifest.depends_on):
1290            # we can safely ignore this completely as the setup will be retried later
1291            # automatically when the dependency is loaded
1292            return
1293
1294        # seed the config with its stored raw values so any construction-time option reads
1295        # in setup()/__init__ see them (the fully-typed entries are only resolvable once the
1296        # instance exists, and are applied by rehydrate_provider_config just below)
1297        self.config.seed_stored_config_values(conf)
1298
1299        # try to setup the module
1300        # (unbounded: this may still have to install the provider's requirements)
1301        async with _provider_load_step(domain, "import its module"):
1302            prov_mod = await load_provider_module(domain, prov_manifest.requirements)
1303        async with _provider_load_step(domain, "load", PROVIDER_SETUP_TIMEOUT):
1304            provider = await prov_mod.setup(self, prov_manifest, conf)
1305
1306        # The instance now exists, so its full (options) config entries can be resolved
1307        # (get_config_entries is an instance method). Rehydrate the config values from
1308        # storage against those entries and validate the complete config, before async
1309        # init so get_config_value reads there see the stored values.
1310        async with _provider_load_step(domain, "resolve its configuration", PROVIDER_SETUP_TIMEOUT):
1311            await self.config.rehydrate_provider_config(provider)
1312        try:
1313            provider.config.validate()
1314        except (KeyError, ValueError, AttributeError, TypeError) as err:
1315            # name the offending entry: the generic message alone gives no clue which
1316            # value is missing or malformed when a provider refuses to load
1317            msg = f"Configuration is invalid: {err}"
1318            raise SetupFailedError(msg) from err
1319
1320        # run async setup
1321        async with _provider_load_step(domain, "initialize", PROVIDER_ASYNC_INIT_TIMEOUT):
1322            await provider.handle_async_init()
1323
1324        await self._register_loaded_provider(provider, conf)
1325
1326    async def _register_loaded_provider(
1327        self, provider: ProviderInstanceType, conf: ProviderConfig
1328    ) -> None:
1329        """Register a provider that finished its setup and run its post-load steps."""
1330        # the instance is now live: register it so the post-load steps below can resolve it
1331        self._providers[provider.instance_id] = provider
1332        provider.available = True
1333
1334        # adapt logging name if needed
1335        provider._set_log_level_from_config(provider.config)
1336
1337        try:
1338            async with _provider_load_step(
1339                provider.domain, "finish loading", PROVIDER_SETUP_TIMEOUT
1340            ):
1341                await self._update_available_providers_cache()
1342                if isinstance(provider, MusicProvider):
1343                    await self.music.on_provider_loaded(provider)
1344                if isinstance(provider, PlayerProvider):
1345                    await self.players.on_provider_loaded(provider)
1346        except Exception:
1347            # a provider that did not finish loading must not stay registered: it would
1348            # report status LOADED while an error is recorded against it, which leaves the
1349            # user with a warning they can only find by opening the provider's own settings
1350            try:
1351                await self.unload_provider(provider.instance_id)
1352            except Exception as unload_err:
1353                # the load failure is the one worth reporting, so keep it as the raised error
1354                LOGGER.warning(
1355                    "Error unloading provider %s: %s",
1356                    provider.name,
1357                    unload_err,
1358                    exc_info=unload_err,
1359                )
1360            raise
1361
1362        # if we reach this point, the provider loaded successfully
1363        LOGGER.info(
1364            "Loaded %s provider %s",
1365            provider.type.value,
1366            provider.name,
1367        )
1368
1369        # execute post load actions
1370        async def _on_provider_loaded() -> None:
1371            await provider.loaded_in_mass()
1372            provider.initialized.set()
1373            self.get_provider_ready_event(provider.domain).set()
1374            await self.run_provider_discovery(provider.instance_id)
1375            # push instance name to config (to persist it if it was autogenerated)
1376            if provider.default_name != conf.default_name:
1377                self.config.set_provider_default_name(provider.instance_id, provider.default_name)
1378
1379        self.create_task(_on_provider_loaded())
1380
1381        # clear any previous error in config and signal update
1382        self.config.set(f"{CONF_PROVIDERS}/{conf.instance_id}/last_error", None)
1383        self.signal_event(EventType.PROVIDERS_UPDATED, data=self.get_providers())
1384
1385    async def __load_provider_manifests(self) -> None:
1386        """Preload all available provider manifest files."""
1387
1388        async def load_provider_manifest(provider_domain: str, provider_path: str) -> None:
1389            """Preload all available provider manifest files."""
1390            # get files in subdirectory
1391            for file_str in await asyncio.to_thread(os.listdir, provider_path):  # noqa: PTH208, RUF100
1392                file_path = os.path.join(provider_path, file_str)
1393                if not await isfile(file_path):
1394                    continue
1395                if file_str != "manifest.json":
1396                    continue
1397                try:
1398                    provider_manifest: ProviderManifest = await ProviderManifest.parse(file_path)
1399                    # detect provider icon image variants (svg preferred over png)
1400                    icons = await detect_provider_icons(provider_path)
1401                    if icons:
1402                        self._provider_icons[provider_manifest.domain] = icons
1403                        provider_manifest.icon_images = list(icons)
1404                    # detect a setup_flow.py module by its mere presence: importing it
1405                    # here would trigger installing the provider's requirements
1406                    provider_manifest.has_setup_flow = await isfile(
1407                        os.path.join(provider_path, "setup_flow.py")
1408                    )
1409                    # override Home Assistant provider if we're running as add-on
1410                    if provider_manifest.domain == "hass" and self.running_as_hass_addon:
1411                        provider_manifest.builtin = True
1412                        provider_manifest.allow_disable = False
1413
1414                    self._provider_manifests[provider_manifest.domain] = provider_manifest
1415                    LOGGER.log(
1416                        VERBOSE_LOG_LEVEL, "Loaded manifest for provider %s", provider_manifest.name
1417                    )
1418                except Exception as exc:
1419                    LOGGER.exception(
1420                        "Error while loading manifest for provider %s",
1421                        provider_domain,
1422                        exc_info=exc,
1423                    )
1424
1425        async with TaskManager(self) as tg:
1426            for dir_str in await asyncio.to_thread(os.listdir, PROVIDERS_PATH):  # noqa: PTH208, RUF100
1427                if dir_str.startswith("."):
1428                    # skip hidden directories
1429                    continue
1430                dir_path = os.path.join(PROVIDERS_PATH, dir_str)
1431                if dir_str.startswith("_") and not self.dev_mode:
1432                    # only load demo/test providers if debug mode is enabled (e.g. for development)
1433                    continue
1434                if not await isdir(dir_path):
1435                    continue
1436                tg.create_task(load_provider_manifest(dir_str, dir_path))
1437        self.logger.debug("Loaded %s provider manifests", len(self._provider_manifests))
1438
1439    async def _update_available_providers_cache(self) -> None:
1440        """Update the global cache variable of loaded/available providers."""
1441        await set_global_cache_values(
1442            {
1443                "provider_domains": {x.domain for x in self.providers},
1444                "provider_instance_ids": {x.instance_id for x in self.providers},
1445                "available_providers": {
1446                    *{x.domain for x in self.providers},
1447                    *{x.instance_id for x in self.providers},
1448                },
1449                "unique_providers": self.music.get_unique_providers(),
1450                "streaming_providers": {
1451                    x.domain
1452                    for x in self.providers
1453                    if is_music_provider(x) and x.is_streaming_provider
1454                },
1455                "non_streaming_providers": {
1456                    x.instance_id
1457                    for x in self.providers
1458                    if not (is_music_provider(x) and x.is_streaming_provider)
1459                },
1460            }
1461        )
1462
1463    async def _setup_storage(self) -> None:
1464        """Handle Setup of storage/cache folder(s)."""
1465        if not await isdir(self.storage_path):
1466            await mkdirs(self.storage_path)
1467        if not await isdir(self.cache_path):
1468            await mkdirs(self.cache_path)
1469
1470    def _set_state(self, new_state: CoreState) -> None:
1471        """Set new state and signal state change."""
1472        if self._state == new_state:
1473            return
1474        self._state = new_state
1475        if not hasattr(self, "webserver"):
1476            # a startup that failed before the core controllers were created has no
1477            # server info to report and no subscribers to report it to, while the state
1478            # itself must still change so that shutdown can run to completion
1479            return
1480        self.signal_event(EventType.CORE_STATE_UPDATED, data=self.get_server_info())
1481