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