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