music-assistant-server

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