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            try:
1030                if is_player_provider(provider):
1031                    # unregister all players of this provider, straight from the registry: the
1032                    # provider's own players listing hides disabled and still-initializing
1033                    # players, which must be unregistered here too so their on_unload runs
1034                    # and no stale entry is left behind
1035                    for player in list(self.players):
1036                        if player.provider.instance_id != instance_id:
1037                            continue
1038                        await self.players.unregister(player.player_id, permanent=is_removed)
1039                await provider.unload(is_removed)
1040            except Exception as err:
1041                LOGGER.warning(
1042                    "Error while unloading provider %s: %s", provider.name, str(err), exc_info=err
1043                )
1044            finally:
1045                if provider.domain in self._provider_ready_events:
1046                    self._provider_ready_events[provider.domain].clear()
1047                self._providers.pop(instance_id, None)
1048                self.discovery.on_provider_unload(instance_id)
1049                await self._update_available_providers_cache()
1050                self.signal_event(EventType.PROVIDERS_UPDATED, data=self.get_providers())
1051
1052    async def unload_provider_with_error(self, instance_id: str, error: str | Exception) -> None:
1053        """
1054        Unload a provider that hit a problem which needs user interaction.
1055
1056        :param error: The originating exception (preferred, so e.g. a LoginFailed surfaces as an
1057            auth-required status with a localized message) or a plain string for a generic error.
1058        """
1059        prov_error = (
1060            _provider_error_from_exc(error)
1061            if isinstance(error, Exception)
1062            else ProviderError(error_code=999, message=error)
1063        )
1064        self.config.update_provider_last_error(instance_id, prov_error)
1065        await self.unload_provider(instance_id)
1066
1067    async def run_provider_discovery(self, instance_id: str) -> None:
1068        """
1069        Run shared discovery for a given provider.
1070
1071        In case of a PlayerProvider, will also call its own discovery method.
1072        """
1073        provider = self.get_provider(instance_id, return_unavailable=False)
1074        if not provider:
1075            raise KeyError(f"Provider with instance ID {instance_id} not found")
1076        await self.discovery.run_provider_discovery(provider)
1077        if isinstance(provider, PlayerProvider):
1078            await provider.discover_players()
1079
1080    def verify_event_loop_thread(self, what: str) -> None:
1081        """Report and raise if we are not running in the event loop thread."""
1082        if self.loop_thread_id != threading.get_ident():
1083            raise RuntimeError(
1084                f"Non-Async operation detected: {what} may only be called from the eventloop."
1085            )
1086
1087    async def __aenter__(self) -> Self:
1088        """Return Context manager."""
1089        await self.start()
1090        return self
1091
1092    async def __aexit__(
1093        self,
1094        exc_type: type[BaseException] | None,
1095        exc_val: BaseException | None,
1096        exc_tb: TracebackType | None,
1097    ) -> bool | None:
1098        """Exit context manager."""
1099        await self.stop()
1100        return None
1101
1102    def _register_api_commands(self) -> None:
1103        """Register all methods decorated as api_command within a class(instance)."""
1104        for cls in (
1105            self,
1106            self.config,
1107            self.metadata,
1108            self.tasks,
1109            self.music,
1110            self.players,
1111            self.player_queues,
1112            self.translations,
1113            self.webserver,
1114            self.webserver.auth,
1115            self.streams.audio_analysis,
1116            self.diagnostics,
1117            self.dashboard,
1118        ):
1119            for attr_name in dir(cls):
1120                if attr_name.startswith("__"):
1121                    continue
1122                # Skip properties to avoid triggering lazy initialization side effects
1123                # (e.g. http_session creating an aiohttp connector during registration)
1124                if isinstance(getattr(type(cls), attr_name, None), property):
1125                    continue
1126                try:
1127                    obj = getattr(cls, attr_name)
1128                except AttributeError, RuntimeError:
1129                    # Skip attributes that fail during initialization
1130                    continue
1131                if hasattr(obj, "api_cmd"):
1132                    # method is decorated with our api decorator
1133                    authenticated = getattr(obj, "api_authenticated", True)
1134                    required_scope = getattr(obj, "api_required_scope", None)
1135                    allow_impersonation = getattr(obj, "api_allow_impersonation", False)
1136                    alias = getattr(obj, "api_alias", False)
1137                    self.register_api_command(
1138                        obj.api_cmd, obj, authenticated, required_scope, allow_impersonation, alias
1139                    )
1140
1141    async def _load_core_controllers(self) -> None:
1142        """Instantiate the core controllers and register their manifests and icons."""
1143        self.cache = CacheController(self)
1144        self.tasks = TasksController(self)
1145        self.webserver = WebserverController(self)
1146        self.metadata = MetaDataController(self)
1147        self.music = MusicController(self)
1148        self.players = PlayerController(self)
1149        self.player_queues = PlayerQueuesController(self)
1150        self.streams = StreamsController(self)
1151        self.translations = TranslationController(self)
1152        self.diagnostics = DiagnosticsController(self)
1153        self.dashboard = DashboardController(self)
1154        # add manifests for core controllers
1155        for controller_name in CONFIGURABLE_CORE_CONTROLLERS:
1156            controller: CoreController = getattr(self, controller_name)
1157            self._provider_manifests[controller.domain] = controller.manifest
1158            # load icon image(s) shipped alongside the controller module
1159            controller_dir = os.path.dirname(inspect.getfile(type(controller)))
1160            if icons := await detect_provider_icons(controller_dir):
1161                self._provider_icons[controller.domain] = icons
1162                controller.manifest.icon_images = list(icons)
1163
1164    async def _load_builtin_providers(self) -> None:
1165        """
1166        Load all builtin providers.
1167
1168        Builtin providers are always needed (also in safe mode) and are fully awaited.
1169        On error, setup will fail.
1170        """
1171        # create default config for any 'builtin' providers
1172        for prov_manifest in self._provider_manifests.values():
1173            if prov_manifest.type == ProviderType.CORE:
1174                # core controllers are not real providers
1175                continue
1176            if not prov_manifest.builtin:
1177                continue
1178            await self.config.create_builtin_provider_config(prov_manifest.domain)
1179
1180        # load all configured (and enabled) builtin providers
1181        # (only manifest-related fields are read here, so the option values are not resolved)
1182        prov_configs = await self.config.get_provider_configs()
1183        builtin_configs: list[ProviderConfig] = [
1184            prov_conf
1185            for prov_conf in prov_configs
1186            if (manifest := self._provider_manifests.get(prov_conf.domain))
1187            and manifest.builtin
1188            and (prov_conf.enabled or manifest.allow_disable is False)
1189        ]
1190
1191        # load builtin providers and wait for them to complete
1192        async with asyncio.TaskGroup() as tg:
1193            for conf in builtin_configs:
1194                tg.create_task(self.load_provider(conf.instance_id, allow_retry=True))
1195
1196    async def _load_providers(self) -> None:
1197        """
1198        Load regular (non-builtin) providers from config.
1199
1200        Regular providers are loaded in background tasks
1201        and can fail without affecting core setup.
1202        """
1203        # handle default providers setup
1204        self.config.set_default(CONF_DEFAULT_PROVIDERS_SETUP, set())
1205        default_providers_setup = set(self.config.get(CONF_DEFAULT_PROVIDERS_SETUP))
1206        changes_made = False
1207        newly_created_defaults: set[str] = set()
1208        for default_provider, require_mdns in DEFAULT_PROVIDERS:
1209            if default_provider in default_providers_setup:
1210                # already processed/setup before, skip
1211                continue
1212            if not (manifest := self._provider_manifests.get(default_provider)):
1213                continue
1214            if require_mdns:
1215                # if mdns discovery is required, check if we have seen any mdns entries
1216                # for this provider before setting it up
1217                for mdns_name in set(self.discovery.aiozc.zeroconf.cache.cache):
1218                    if manifest.mdns_discovery and any(
1219                        mdns_type in mdns_name for mdns_type in manifest.mdns_discovery
1220                    ):
1221                        break
1222                else:
1223                    continue
1224            await self.config.create_builtin_provider_config(manifest.domain)
1225            changes_made = True
1226            newly_created_defaults.add(manifest.domain)
1227            # TEMP: migration - to be removed after 2.8 release
1228            # enable all existing players of the default providers if they are not already enabled
1229            # due to the linked protocol feature we introduced
1230            for player_config in await self.config.get_player_configs(
1231                provider=default_provider, include_disabled=True
1232            ):
1233                if player_config.enabled:
1234                    continue
1235                await self.config.save_player_config(player_config.player_id, {"enabled": True})
1236            default_providers_setup.add(default_provider)
1237        if changes_made:
1238            self.config.set(CONF_DEFAULT_PROVIDERS_SETUP, default_providers_setup)
1239            self.config.save(True)
1240        # load all configured (and enabled) regular (non-builtin) providers
1241        # (only manifest-related fields are read here, so the option values are not resolved)
1242        prov_configs = await self.config.get_provider_configs()
1243        other_configs: list[ProviderConfig] = [
1244            prov_conf
1245            for prov_conf in prov_configs
1246            if prov_conf.enabled
1247            and (
1248                not (manifest := self._provider_manifests.get(prov_conf.domain))
1249                or not manifest.builtin
1250            )
1251        ]
1252        # load providers concurrently via tasks, bounded so a host with many providers does
1253        # not import every provider module at once (a torch-backed one costs hundreds of MB)
1254        async with TaskManager(self, PROVIDER_LOAD_CONCURRENCY) as tg:
1255            for prov_conf in other_configs:
1256                # Use a task so we can load multiple providers at once.
1257                # If a provider fails, that will not block the loading of other providers.
1258                # For providers just auto-set-up as a default, drop the config again if the
1259                # host does not meet their requirements (rather than retry a broken provider).
1260                await tg.create_task_with_limit(
1261                    self.load_provider(
1262                        prov_conf.instance_id,
1263                        allow_retry=True,
1264                        remove_if_unsupported=prov_conf.domain in newly_created_defaults,
1265                    )
1266                )
1267
1268    async def _load_provider(self, conf: ProviderConfig) -> None:
1269        """Load (or reload) a provider."""
1270        # if provider is already loaded, stop and unload it first
1271        await self.unload_provider(conf.instance_id)
1272        LOGGER.debug("Loading provider %s", conf.name or conf.domain)
1273        if not conf.enabled:
1274            msg = "Provider is disabled"
1275            raise SetupFailedError(msg)
1276
1277        # The config is validated after the instance is created and its config rehydrated
1278        # (see below): the full options entries - and thus which values are required - are
1279        # only known once the instance exists.
1280
1281        domain = conf.domain
1282        prov_manifest = self._provider_manifests.get(domain)
1283        # check for other instances of this provider
1284        existing = next((x for x in self.providers if x.domain == domain), None)
1285        if existing and prov_manifest and not prov_manifest.multi_instance:
1286            msg = f"Provider {domain} already loaded and only one instance allowed."
1287            raise SetupFailedError(msg)
1288        # check valid manifest (just in case)
1289        if not prov_manifest:
1290            msg = f"Provider {domain} manifest not found"
1291            raise SetupFailedError(msg)
1292
1293        # handle dependency on other provider
1294        if prov_manifest.depends_on and not self.get_provider(prov_manifest.depends_on):
1295            # we can safely ignore this completely as the setup will be retried later
1296            # automatically when the dependency is loaded
1297            return
1298
1299        # seed the config with its stored raw values so any construction-time option reads
1300        # in setup()/__init__ see them (the fully-typed entries are only resolvable once the
1301        # instance exists, and are applied by rehydrate_provider_config just below)
1302        self.config.seed_stored_config_values(conf)
1303
1304        # try to setup the module
1305        # (unbounded: this may still have to install the provider's requirements)
1306        async with _provider_load_step(domain, "import its module"):
1307            prov_mod = await load_provider_module(domain, prov_manifest.requirements)
1308        async with _provider_load_step(domain, "load", PROVIDER_SETUP_TIMEOUT):
1309            provider = await prov_mod.setup(self, prov_manifest, conf)
1310
1311        # The instance now exists, so its full (options) config entries can be resolved
1312        # (get_config_entries is an instance method). Rehydrate the config values from
1313        # storage against those entries and validate the complete config, before async
1314        # init so get_config_value reads there see the stored values.
1315        async with _provider_load_step(domain, "resolve its configuration", PROVIDER_SETUP_TIMEOUT):
1316            await self.config.rehydrate_provider_config(provider)
1317        try:
1318            provider.config.validate()
1319        except (KeyError, ValueError, AttributeError, TypeError) as err:
1320            # name the offending entry: the generic message alone gives no clue which
1321            # value is missing or malformed when a provider refuses to load
1322            msg = f"Configuration is invalid: {err}"
1323            raise SetupFailedError(msg) from err
1324
1325        # run async setup
1326        async with _provider_load_step(domain, "initialize", PROVIDER_ASYNC_INIT_TIMEOUT):
1327            await provider.handle_async_init()
1328
1329        await self._register_loaded_provider(provider, conf)
1330
1331    async def _register_loaded_provider(
1332        self, provider: ProviderInstanceType, conf: ProviderConfig
1333    ) -> None:
1334        """Register a provider that finished its setup and run its post-load steps."""
1335        # the instance is now live: register it so the post-load steps below can resolve it
1336        self._providers[provider.instance_id] = provider
1337        provider.available = True
1338
1339        # adapt logging name if needed
1340        provider._set_log_level_from_config(provider.config)
1341
1342        try:
1343            async with _provider_load_step(
1344                provider.domain, "finish loading", PROVIDER_SETUP_TIMEOUT
1345            ):
1346                await self._update_available_providers_cache()
1347                if isinstance(provider, MusicProvider):
1348                    await self.music.on_provider_loaded(provider)
1349                if isinstance(provider, PlayerProvider):
1350                    await self.players.on_provider_loaded(provider)
1351        except Exception:
1352            # a provider that did not finish loading must not stay registered: it would
1353            # report status LOADED while an error is recorded against it, which leaves the
1354            # user with a warning they can only find by opening the provider's own settings
1355            try:
1356                await self.unload_provider(provider.instance_id)
1357            except Exception as unload_err:
1358                # the load failure is the one worth reporting, so keep it as the raised error
1359                LOGGER.warning(
1360                    "Error unloading provider %s: %s",
1361                    provider.name,
1362                    unload_err,
1363                    exc_info=unload_err,
1364                )
1365            raise
1366
1367        # if we reach this point, the provider loaded successfully
1368        LOGGER.info(
1369            "Loaded %s provider %s",
1370            provider.type.value,
1371            provider.name,
1372        )
1373
1374        # execute post load actions
1375        async def _on_provider_loaded() -> None:
1376            try:
1377                await provider.loaded_in_mass()
1378            except Exception as err:
1379                # the provider stays registered and available either way, so the steps
1380                # below still run: an event left unset makes every waiter pay the full
1381                # timeout, on every attempt, until the provider reloads
1382                LOGGER.warning(
1383                    "Error in the post load step of provider %s: %s",
1384                    provider.name,
1385                    str(err) or err.__class__.__name__,
1386                    exc_info=err,
1387                )
1388            provider.initialized.set()
1389            self.get_provider_ready_event(provider.domain).set()
1390            await self.run_provider_discovery(provider.instance_id)
1391            # push instance name to config (to persist it if it was autogenerated)
1392            if provider.default_name != conf.default_name:
1393                self.config.set_provider_default_name(provider.instance_id, provider.default_name)
1394
1395        self.create_task(_on_provider_loaded())
1396
1397        # clear any previous error in config and signal update
1398        self.config.set(f"{CONF_PROVIDERS}/{conf.instance_id}/last_error", None)
1399        self.signal_event(EventType.PROVIDERS_UPDATED, data=self.get_providers())
1400
1401    async def __load_provider_manifests(self) -> None:
1402        """Preload all available provider manifest files."""
1403
1404        async def load_provider_manifest(provider_domain: str, provider_path: str) -> None:
1405            """Preload all available provider manifest files."""
1406            # get files in subdirectory
1407            for file_str in await asyncio.to_thread(os.listdir, provider_path):  # noqa: PTH208, RUF100
1408                file_path = os.path.join(provider_path, file_str)
1409                if not await isfile(file_path):
1410                    continue
1411                if file_str != "manifest.json":
1412                    continue
1413                try:
1414                    provider_manifest: ProviderManifest = await ProviderManifest.parse(file_path)
1415                    # detect provider icon image variants (svg preferred over png)
1416                    icons = await detect_provider_icons(provider_path)
1417                    if icons:
1418                        self._provider_icons[provider_manifest.domain] = icons
1419                        provider_manifest.icon_images = list(icons)
1420                    # detect a setup_flow.py module by its mere presence: importing it
1421                    # here would trigger installing the provider's requirements
1422                    provider_manifest.has_setup_flow = await isfile(
1423                        os.path.join(provider_path, "setup_flow.py")
1424                    )
1425                    # override Home Assistant provider if we're running as add-on
1426                    if provider_manifest.domain == "hass" and self.running_as_hass_addon:
1427                        provider_manifest.builtin = True
1428                        provider_manifest.allow_disable = False
1429
1430                    self._provider_manifests[provider_manifest.domain] = provider_manifest
1431                    LOGGER.log(
1432                        VERBOSE_LOG_LEVEL, "Loaded manifest for provider %s", provider_manifest.name
1433                    )
1434                except Exception as exc:
1435                    LOGGER.exception(
1436                        "Error while loading manifest for provider %s",
1437                        provider_domain,
1438                        exc_info=exc,
1439                    )
1440
1441        async with TaskManager(self) as tg:
1442            for dir_str in await asyncio.to_thread(os.listdir, PROVIDERS_PATH):  # noqa: PTH208, RUF100
1443                if dir_str.startswith("."):
1444                    # skip hidden directories
1445                    continue
1446                dir_path = os.path.join(PROVIDERS_PATH, dir_str)
1447                if dir_str.startswith("_") and not self.dev_mode:
1448                    # only load demo/test providers if debug mode is enabled (e.g. for development)
1449                    continue
1450                if not await isdir(dir_path):
1451                    continue
1452                tg.create_task(load_provider_manifest(dir_str, dir_path))
1453        self.logger.debug("Loaded %s provider manifests", len(self._provider_manifests))
1454
1455    async def _update_available_providers_cache(self) -> None:
1456        """Update the global cache variable of loaded/available providers."""
1457        await set_global_cache_values(
1458            {
1459                "provider_domains": {x.domain for x in self.providers},
1460                "provider_instance_ids": {x.instance_id for x in self.providers},
1461                "available_providers": {
1462                    *{x.domain for x in self.providers},
1463                    *{x.instance_id for x in self.providers},
1464                },
1465                "unique_providers": self.music.get_unique_providers(),
1466                "streaming_providers": {
1467                    x.domain
1468                    for x in self.providers
1469                    if is_music_provider(x) and x.is_streaming_provider
1470                },
1471                "non_streaming_providers": {
1472                    x.instance_id
1473                    for x in self.providers
1474                    if not (is_music_provider(x) and x.is_streaming_provider)
1475                },
1476            }
1477        )
1478
1479    async def _setup_storage(self) -> None:
1480        """Handle Setup of storage/cache folder(s)."""
1481        if not await isdir(self.storage_path):
1482            await mkdirs(self.storage_path)
1483        if not await isdir(self.cache_path):
1484            await mkdirs(self.cache_path)
1485
1486    def _set_state(self, new_state: CoreState) -> None:
1487        """Set new state and signal state change."""
1488        if self._state == new_state:
1489            return
1490        self._state = new_state
1491        if not hasattr(self, "webserver"):
1492            # a startup that failed before the core controllers were created has no
1493            # server info to report and no subscribers to report it to, while the state
1494            # itself must still change so that shutdown can run to completion
1495            return
1496        self.signal_event(EventType.CORE_STATE_UPDATED, data=self.get_server_info())
1497