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