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