/
/
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 log_exceptions: bool = True,
708 **kwargs: Any,
709 ) -> asyncio.Task[_R]:
710 """
711 Create Task on (main) event loop from Coroutine(function).
712
713 Tasks created by this helper will be properly cancelled on stop.
714
715 :param target: Coroutine function or awaitable to run as a task.
716 :param args: Arguments to pass to the coroutine function.
717 :param task_id: Optional ID to track and deduplicate tasks.
718 :param abort_existing: If True, cancel existing task with same task_id.
719 :param eager_start: If True (default), start task immediately without waiting
720 for next event loop iteration. This ensures proper ordering
721 when creating multiple tasks in sequence.
722 :param log_exceptions: Set to False when the caller awaits the task and reports
723 its failures itself; the task then logs at debug level
724 instead of warning.
725 :param kwargs: Keyword arguments to pass to the coroutine function.
726 """
727 if task_id and (existing := self._tracked_tasks.get(task_id)) and not existing.done():
728 # prevent duplicate tasks if task_id is given and already present
729 if abort_existing:
730 existing.cancel()
731 else:
732 # close any already-constructed coroutine to avoid "never awaited" warning
733 if inspect.iscoroutine(target):
734 target.close()
735 return existing
736 self.verify_event_loop_thread("create_task")
737
738 if inspect.iscoroutinefunction(target):
739 # coroutine function
740 coro = target(*args, **kwargs)
741 elif inspect.iscoroutine(target):
742 # coroutine
743 coro = target
744 elif callable(target):
745 raise RuntimeError("Function is not a coroutine or coroutine function")
746 else:
747 raise RuntimeError("Target is missing")
748
749 # Use asyncio.Task directly with eager_start for immediate execution
750 task: asyncio.Task[_R] = asyncio.Task(coro, loop=self.loop, eager_start=eager_start)
751
752 if task_id is None:
753 task_id = uuid4().hex
754
755 def task_done_callback(_task: asyncio.Task[Any]) -> None:
756 # done callbacks run one event loop iteration after the task finished, so a
757 # caller may already have replaced the entry with a new task under the same
758 # task_id - only untrack when the entry still points at this task
759 if self._tracked_tasks.get(task_id) is _task:
760 del self._tracked_tasks[task_id]
761 if _task.cancelled():
762 return
763 # always retrieve the exception, otherwise asyncio logs a noisy
764 # "Task exception was never retrieved" error at garbage collection time
765 if err := _task.exception():
766 task_name = _task.get_name() if hasattr(_task, "get_name") else str(_task)
767 # a failure the waiters report themselves is demoted rather than dropped:
768 # work that outlives every waiter (join_task keeps it running) would
769 # otherwise fail without a trace anywhere
770 LOGGER.log(
771 logging.WARNING if log_exceptions else logging.DEBUG,
772 "Exception in task %s - target: %s: %s",
773 task_name,
774 str(target),
775 str(err),
776 exc_info=err if LOGGER.isEnabledFor(logging.DEBUG) else None,
777 )
778
779 self._tracked_tasks[task_id] = task
780 task.add_done_callback(task_done_callback)
781 return task
782
783 def call_later(
784 self,
785 delay: float,
786 target: Coroutine[Any, Any, _R] | Awaitable[_R] | Callable[..., _R],
787 *args: Any,
788 task_id: str | None = None,
789 **kwargs: Any,
790 ) -> asyncio.TimerHandle:
791 """
792 Run callable/awaitable after given delay.
793
794 Use task_id for debouncing.
795 """
796 self.verify_event_loop_thread("call_later")
797
798 if not task_id:
799 task_id = uuid4().hex
800
801 if existing := self._tracked_timers.get(task_id):
802 existing.cancel()
803
804 def _create_task(_target: Coroutine[Any, Any, _R]) -> None:
805 self._tracked_timers.pop(task_id)
806 self.create_task(_target, *args, task_id=task_id, abort_existing=True, **kwargs)
807
808 def _call_sync(_target: Callable[..., _R]) -> None:
809 self._tracked_timers.pop(task_id)
810 _target(*args, **kwargs)
811
812 if inspect.iscoroutinefunction(target) or inspect.iscoroutine(target):
813 # coroutine function
814 if TYPE_CHECKING:
815 target = cast("Coroutine[Any, Any, _R]", target)
816 handle = self.loop.call_later(delay, _create_task, target)
817 else:
818 # regular sync callable
819 if TYPE_CHECKING:
820 target = cast("Callable[..., _R]", target)
821 handle = self.loop.call_later(delay, _call_sync, target)
822 self._tracked_timers[task_id] = handle
823 return handle
824
825 def get_task(self, task_id: str) -> asyncio.Task[Any] | None:
826 """Get existing scheduled task."""
827 if existing := self._tracked_tasks.get(task_id):
828 # prevent duplicate tasks if task_id is given and already present
829 return existing
830 return None
831
832 def cancel_task(self, task_id: str) -> None:
833 """Cancel existing scheduled task."""
834 if existing := self._tracked_tasks.pop(task_id, None):
835 existing.cancel()
836
837 def cancel_timer(self, task_id: str) -> None:
838 """Cancel existing scheduled timer."""
839 if existing := self._tracked_timers.pop(task_id, None):
840 existing.cancel()
841
842 def register_api_command(
843 self,
844 command: str,
845 handler: Callable[..., Coroutine[Any, Any, Any] | AsyncGenerator[Any, Any]],
846 authenticated: bool = True,
847 required_scope: Scope | None = None,
848 allow_impersonation: bool = False,
849 alias: bool = False,
850 ) -> Callable[[], None]:
851 """
852 Dynamically register a command on the API.
853
854 :param command: The command name/path.
855 :param handler: The function to handle the command.
856 :param authenticated: Whether authentication is required (default: True).
857 :param required_scope: Scope required to execute the command,
858 None means any authenticated user.
859 :param allow_impersonation: Whether the command accepts a 'user' argument
860 to execute the command on behalf of another user (default: False).
861 :param alias: Whether this is an alias for backward compatibility (default: False).
862 Aliases are not shown in API documentation but remain functional.
863
864 Returns handle to unregister.
865 """
866 if command in self.command_handlers:
867 msg = f"Command {command} is already registered"
868 raise RuntimeError(msg)
869 self.command_handlers[command] = APICommandHandler.parse(
870 command, handler, authenticated, required_scope, allow_impersonation, alias
871 )
872
873 def unregister() -> None:
874 self.command_handlers.pop(command, None)
875
876 return unregister
877
878 async def load_provider_config(
879 self,
880 prov_conf: ProviderConfig,
881 ) -> None:
882 """Load (or reload) a provider from its config, recording any load failure."""
883 # cancel existing (re)load timer if needed
884 task_id = f"load_provider_{prov_conf.instance_id}"
885 if existing := self._tracked_timers.pop(task_id, None):
886 existing.cancel()
887
888 try:
889 await self._load_provider(prov_conf)
890 except Exception as exc:
891 # persist the failure so the provider surfaces a clear status (e.g. auth_required)
892 # to the UI instead of appearing stuck loading, then propagate to the caller
893 self.config.update_provider_last_error(
894 prov_conf.instance_id, _provider_error_from_exc(exc)
895 )
896 raise
897
898 # (re)load any dependents. The provider itself is loaded at this point, so a problem
899 # in this scan belongs to a dependent (or to nothing at all) and must never be
900 # recorded against - and thus flag - the provider we just loaded successfully.
901 try:
902 # resolving option values here would call get_config_entries() on every loaded
903 # provider (some of which do network i/o), for values _load_provider does not
904 # read: it seeds the stored raw values itself and rehydrates once the instance
905 # exists. Only the manifest-related fields below are needed to spot a dependent.
906 prov_configs = await self.config.get_provider_configs()
907 except Exception as exc:
908 LOGGER.warning(
909 "Error looking up dependents of provider(instance) %s: %s",
910 prov_conf.name or prov_conf.instance_id,
911 str(exc) or exc.__class__.__name__,
912 exc_info=_provider_error_traceback(exc),
913 )
914 return
915 for dep_prov_conf in prov_configs:
916 if not dep_prov_conf.enabled:
917 continue
918 manifest = self.get_provider_manifest(dep_prov_conf.domain)
919 if not manifest.depends_on:
920 continue
921 if manifest.depends_on != prov_conf.domain:
922 continue
923 try:
924 # the scan above skipped the config values, but the load path does need them:
925 # a provider reads config (e.g. its log level) while it is being constructed.
926 # Resolve them here, for this single dependent instead of for every provider.
927 dep_conf = await self.config.get_provider_config(dep_prov_conf.instance_id)
928 except KeyError:
929 # config was removed while we were scanning
930 continue
931 try:
932 await self._load_provider(dep_conf)
933 except Exception as exc:
934 # record the failure against the provider that hit it: attributing it to the
935 # provider we just loaded (which is fine) flags the wrong one in the UI
936 self.config.update_provider_last_error(
937 dep_prov_conf.instance_id, _provider_error_from_exc(exc)
938 )
939 LOGGER.warning(
940 "Error loading provider(instance) %s: %s",
941 dep_prov_conf.name or dep_prov_conf.instance_id,
942 str(exc) or exc.__class__.__name__,
943 exc_info=_provider_error_traceback(exc),
944 )
945
946 async def load_provider(
947 self,
948 instance_id: str,
949 allow_retry: bool = False,
950 remove_if_unsupported: bool = False,
951 ) -> None:
952 """Try to load a provider and catch errors."""
953 try:
954 prov_conf = await self.config.get_provider_config(instance_id)
955 except KeyError:
956 # Was deleted before we could run
957 return
958
959 if not prov_conf.enabled:
960 # Was disabled before we could run
961 return
962
963 # cancel existing (re)load timer if needed
964 task_id = f"load_provider_{instance_id}"
965 if existing := self._tracked_timers.pop(task_id, None):
966 existing.cancel()
967
968 try:
969 await self.load_provider_config(prov_conf)
970 except UnsupportedSystemError as exc:
971 # The host does not meet this provider's hardware requirements. This is a
972 # permanent condition, so we never retry. For a provider that was just
973 # auto-set-up as a default, drop the config again so it does not linger as a
974 # broken provider (it stays marked done so it is not auto-created again).
975 if remove_if_unsupported:
976 LOGGER.info(
977 "Not enabling default provider %s: %s",
978 prov_conf.name or prov_conf.instance_id,
979 exc,
980 )
981 # The provider never loaded, so just drop its auto-created config key.
982 # (remove_provider_config refuses builtin providers and runs loaded-provider
983 # cleanup we don't need here; a direct remove persists and is guard-free.)
984 self.config.remove(f"{CONF_PROVIDERS}/{instance_id}")
985 return
986 prov_conf.last_error = _provider_error_from_exc(exc)
987 self.config.update_provider_last_error(instance_id, prov_conf.last_error)
988 LOGGER.warning(
989 "Provider(instance) %s can not run on this system: %s",
990 prov_conf.name or prov_conf.instance_id,
991 exc,
992 )
993 return
994 except Exception as exc:
995 # if loading failed, we store the error in the config object
996 # so we can show something useful to the user
997 prov_conf.last_error = _provider_error_from_exc(exc)
998 self.config.update_provider_last_error(instance_id, prov_conf.last_error)
999
1000 # auto schedule a retry if the (re)load failed with a handled exception
1001 # unhandled exceptions (e.g. ValueError) are likely bugs that won't resolve themselves
1002 will_retry = (
1003 allow_retry
1004 and isinstance(exc, MusicAssistantError)
1005 and not isinstance(
1006 exc,
1007 (AuthenticationRequired, AuthenticationFailed, LoginFailed, InvalidToken),
1008 )
1009 )
1010 if will_retry:
1011 self.call_later(
1012 120,
1013 self.load_provider,
1014 instance_id,
1015 allow_retry,
1016 task_id=task_id,
1017 )
1018 LOGGER.warning(
1019 "Error loading provider(instance) %s: %s%s",
1020 prov_conf.name or prov_conf.instance_id,
1021 str(exc) or exc.__class__.__name__,
1022 " (will be retried later)" if will_retry else "",
1023 exc_info=_provider_error_traceback(exc),
1024 )
1025 return
1026
1027 # (re)load any dependents if needed
1028 for dep_prov in self.providers:
1029 if dep_prov.available:
1030 continue
1031 if dep_prov.manifest.depends_on == prov_conf.domain:
1032 await self.unload_provider(dep_prov.instance_id)
1033
1034 async def unload_provider(self, instance_id: str, is_removed: bool = False) -> None:
1035 """Unload a provider."""
1036 # this waits (bounded) for a running sync to unwind: provider.unload() below tears
1037 # down state the sync may still be using, such as the mount of a network share
1038 await self.music.unschedule_provider_sync(instance_id, clear_persisted_state=is_removed)
1039 if provider := self._providers.get(instance_id):
1040 # mark the provider as on its way out before anything is torn down: the steps
1041 # below have await points, so without this a callback that is still in flight
1042 # could register a player back onto a provider that is already gone
1043 provider.unloading = True
1044 if isinstance(provider, PlayerProvider):
1045 await self.players.on_provider_unload(provider)
1046 if isinstance(provider, MusicProvider):
1047 await self.music.on_provider_unload(provider)
1048 # check if there are no other providers dependent of this provider
1049 for dep_prov in self.providers:
1050 if dep_prov.manifest.depends_on == provider.domain:
1051 await self.unload_provider(dep_prov.instance_id)
1052 try:
1053 if is_player_provider(provider):
1054 # unregister all players of this provider, straight from the registry: the
1055 # provider's own players listing hides disabled and still-initializing
1056 # players, which must be unregistered here too so their on_unload runs
1057 # and no stale entry is left behind
1058 for player in list(self.players):
1059 if player.provider.instance_id != instance_id:
1060 continue
1061 await self.players.unregister(player.player_id, permanent=is_removed)
1062 await provider.unload(is_removed)
1063 except Exception as err:
1064 LOGGER.warning(
1065 "Error while unloading provider %s: %s", provider.name, str(err), exc_info=err
1066 )
1067 finally:
1068 if provider.domain in self._provider_ready_events:
1069 self._provider_ready_events[provider.domain].clear()
1070 self._providers.pop(instance_id, None)
1071 self.discovery.on_provider_unload(instance_id)
1072 await self._update_available_providers_cache()
1073 self.signal_event(EventType.PROVIDERS_UPDATED, data=self.get_providers())
1074
1075 async def unload_provider_with_error(self, instance_id: str, error: str | Exception) -> None:
1076 """
1077 Unload a provider that hit a problem which needs user interaction.
1078
1079 :param error: The originating exception (preferred, so e.g. a LoginFailed surfaces as an
1080 auth-required status with a localized message) or a plain string for a generic error.
1081 """
1082 prov_error = (
1083 _provider_error_from_exc(error)
1084 if isinstance(error, Exception)
1085 else ProviderError(error_code=999, message=error)
1086 )
1087 self.config.update_provider_last_error(instance_id, prov_error)
1088 await self.unload_provider(instance_id)
1089
1090 async def run_provider_discovery(self, instance_id: str) -> None:
1091 """
1092 Run shared discovery for a given provider.
1093
1094 In case of a PlayerProvider, will also call its own discovery method.
1095 """
1096 provider = self.get_provider(instance_id, return_unavailable=False)
1097 if not provider:
1098 raise KeyError(f"Provider with instance ID {instance_id} not found")
1099 await self.discovery.run_provider_discovery(provider)
1100 if isinstance(provider, PlayerProvider):
1101 await provider.discover_players()
1102
1103 def verify_event_loop_thread(self, what: str) -> None:
1104 """Report and raise if we are not running in the event loop thread."""
1105 if self.loop_thread_id != threading.get_ident():
1106 raise RuntimeError(
1107 f"Non-Async operation detected: {what} may only be called from the eventloop."
1108 )
1109
1110 async def __aenter__(self) -> Self:
1111 """Return Context manager."""
1112 await self.start()
1113 return self
1114
1115 async def __aexit__(
1116 self,
1117 exc_type: type[BaseException] | None,
1118 exc_val: BaseException | None,
1119 exc_tb: TracebackType | None,
1120 ) -> bool | None:
1121 """Exit context manager."""
1122 await self.stop()
1123 return None
1124
1125 def _register_api_commands(self) -> None:
1126 """Register all methods decorated as api_command within a class(instance)."""
1127 for cls in (
1128 self,
1129 self.config,
1130 self.metadata,
1131 self.tasks,
1132 self.music,
1133 self.players,
1134 self.player_queues,
1135 self.translations,
1136 self.webserver,
1137 self.webserver.auth,
1138 self.streams.audio_analysis,
1139 self.diagnostics,
1140 self.dashboard,
1141 ):
1142 for attr_name in dir(cls):
1143 if attr_name.startswith("__"):
1144 continue
1145 # Skip properties to avoid triggering lazy initialization side effects
1146 # (e.g. http_session creating an aiohttp connector during registration)
1147 if isinstance(getattr(type(cls), attr_name, None), property):
1148 continue
1149 try:
1150 obj = getattr(cls, attr_name)
1151 except AttributeError, RuntimeError:
1152 # Skip attributes that fail during initialization
1153 continue
1154 if hasattr(obj, "api_cmd"):
1155 # method is decorated with our api decorator
1156 authenticated = getattr(obj, "api_authenticated", True)
1157 required_scope = getattr(obj, "api_required_scope", None)
1158 allow_impersonation = getattr(obj, "api_allow_impersonation", False)
1159 alias = getattr(obj, "api_alias", False)
1160 self.register_api_command(
1161 obj.api_cmd, obj, authenticated, required_scope, allow_impersonation, alias
1162 )
1163
1164 async def _load_core_controllers(self) -> None:
1165 """Instantiate the core controllers and register their manifests and icons."""
1166 self.cache = CacheController(self)
1167 self.tasks = TasksController(self)
1168 self.webserver = WebserverController(self)
1169 self.metadata = MetaDataController(self)
1170 self.music = MusicController(self)
1171 self.players = PlayerController(self)
1172 self.player_queues = PlayerQueuesController(self)
1173 self.streams = StreamsController(self)
1174 self.translations = TranslationController(self)
1175 self.diagnostics = DiagnosticsController(self)
1176 self.dashboard = DashboardController(self)
1177 # add manifests for core controllers
1178 for controller_name in CONFIGURABLE_CORE_CONTROLLERS:
1179 controller: CoreController = getattr(self, controller_name)
1180 self._provider_manifests[controller.domain] = controller.manifest
1181 # load icon image(s) shipped alongside the controller module
1182 controller_dir = os.path.dirname(inspect.getfile(type(controller)))
1183 if icons := await detect_provider_icons(controller_dir):
1184 self._provider_icons[controller.domain] = icons
1185 controller.manifest.icon_images = list(icons)
1186
1187 async def _load_builtin_providers(self) -> None:
1188 """
1189 Load all builtin providers.
1190
1191 Builtin providers are always needed (also in safe mode) and are fully awaited.
1192 On error, setup will fail.
1193 """
1194 # create default config for any 'builtin' providers
1195 for prov_manifest in self._provider_manifests.values():
1196 if prov_manifest.type == ProviderType.CORE:
1197 # core controllers are not real providers
1198 continue
1199 if not prov_manifest.builtin:
1200 continue
1201 await self.config.create_builtin_provider_config(prov_manifest.domain)
1202
1203 # load all configured (and enabled) builtin providers
1204 # (only manifest-related fields are read here, so the option values are not resolved)
1205 prov_configs = await self.config.get_provider_configs()
1206 builtin_configs: list[ProviderConfig] = [
1207 prov_conf
1208 for prov_conf in prov_configs
1209 if (manifest := self._provider_manifests.get(prov_conf.domain))
1210 and manifest.builtin
1211 and (prov_conf.enabled or manifest.allow_disable is False)
1212 ]
1213
1214 # load builtin providers and wait for them to complete
1215 async with asyncio.TaskGroup() as tg:
1216 for conf in builtin_configs:
1217 tg.create_task(self.load_provider(conf.instance_id, allow_retry=True))
1218
1219 async def _load_providers(self) -> None:
1220 """
1221 Load regular (non-builtin) providers from config.
1222
1223 Regular providers are loaded in background tasks
1224 and can fail without affecting core setup.
1225 """
1226 # handle default providers setup
1227 self.config.set_default(CONF_DEFAULT_PROVIDERS_SETUP, set())
1228 default_providers_setup = set(self.config.get(CONF_DEFAULT_PROVIDERS_SETUP))
1229 changes_made = False
1230 newly_created_defaults: set[str] = set()
1231 for default_provider, require_mdns in DEFAULT_PROVIDERS:
1232 if default_provider in default_providers_setup:
1233 # already processed/setup before, skip
1234 continue
1235 if not (manifest := self._provider_manifests.get(default_provider)):
1236 continue
1237 if require_mdns:
1238 # if mdns discovery is required, check if we have seen any mdns entries
1239 # for this provider before setting it up
1240 for mdns_name in set(self.discovery.aiozc.zeroconf.cache.cache):
1241 if manifest.mdns_discovery and any(
1242 mdns_type in mdns_name for mdns_type in manifest.mdns_discovery
1243 ):
1244 break
1245 else:
1246 continue
1247 await self.config.create_builtin_provider_config(manifest.domain)
1248 changes_made = True
1249 newly_created_defaults.add(manifest.domain)
1250 # TEMP: migration - to be removed after 2.8 release
1251 # enable all existing players of the default providers if they are not already enabled
1252 # due to the linked protocol feature we introduced
1253 for player_config in await self.config.get_player_configs(
1254 provider=default_provider, include_disabled=True
1255 ):
1256 if player_config.enabled:
1257 continue
1258 await self.config.save_player_config(player_config.player_id, {"enabled": True})
1259 default_providers_setup.add(default_provider)
1260 if changes_made:
1261 self.config.set(CONF_DEFAULT_PROVIDERS_SETUP, default_providers_setup)
1262 self.config.save(True)
1263 # load all configured (and enabled) regular (non-builtin) providers
1264 # (only manifest-related fields are read here, so the option values are not resolved)
1265 prov_configs = await self.config.get_provider_configs()
1266 other_configs: list[ProviderConfig] = [
1267 prov_conf
1268 for prov_conf in prov_configs
1269 if prov_conf.enabled
1270 and (
1271 not (manifest := self._provider_manifests.get(prov_conf.domain))
1272 or not manifest.builtin
1273 )
1274 ]
1275 # load providers concurrently via tasks, bounded so a host with many providers does
1276 # not import every provider module at once (a torch-backed one costs hundreds of MB)
1277 async with TaskManager(self, PROVIDER_LOAD_CONCURRENCY) as tg:
1278 for prov_conf in other_configs:
1279 # Use a task so we can load multiple providers at once.
1280 # If a provider fails, that will not block the loading of other providers.
1281 # For providers just auto-set-up as a default, drop the config again if the
1282 # host does not meet their requirements (rather than retry a broken provider).
1283 await tg.create_task_with_limit(
1284 self.load_provider(
1285 prov_conf.instance_id,
1286 allow_retry=True,
1287 remove_if_unsupported=prov_conf.domain in newly_created_defaults,
1288 )
1289 )
1290
1291 async def _load_provider(self, conf: ProviderConfig) -> None:
1292 """Load (or reload) a provider."""
1293 # if provider is already loaded, stop and unload it first
1294 await self.unload_provider(conf.instance_id)
1295 LOGGER.debug("Loading provider %s", conf.name or conf.domain)
1296 if not conf.enabled:
1297 msg = "Provider is disabled"
1298 raise SetupFailedError(msg)
1299
1300 # The config is validated after the instance is created and its config rehydrated
1301 # (see below): the full options entries - and thus which values are required - are
1302 # only known once the instance exists.
1303
1304 domain = conf.domain
1305 prov_manifest = self._provider_manifests.get(domain)
1306 # check for other instances of this provider
1307 existing = next((x for x in self.providers if x.domain == domain), None)
1308 if existing and prov_manifest and not prov_manifest.multi_instance:
1309 msg = f"Provider {domain} already loaded and only one instance allowed."
1310 raise SetupFailedError(msg)
1311 # check valid manifest (just in case)
1312 if not prov_manifest:
1313 msg = f"Provider {domain} manifest not found"
1314 raise SetupFailedError(msg)
1315
1316 # handle dependency on other provider
1317 if prov_manifest.depends_on and not self.get_provider(prov_manifest.depends_on):
1318 # we can safely ignore this completely as the setup will be retried later
1319 # automatically when the dependency is loaded
1320 return
1321
1322 # seed the config with its stored raw values so any construction-time option reads
1323 # in setup()/__init__ see them (the fully-typed entries are only resolvable once the
1324 # instance exists, and are applied by rehydrate_provider_config just below)
1325 self.config.seed_stored_config_values(conf)
1326
1327 # try to setup the module
1328 # (unbounded: this may still have to install the provider's requirements)
1329 async with _provider_load_step(domain, "import its module"):
1330 prov_mod = await load_provider_module(domain, prov_manifest.requirements)
1331 async with _provider_load_step(domain, "load", PROVIDER_SETUP_TIMEOUT):
1332 provider = await prov_mod.setup(self, prov_manifest, conf)
1333
1334 # The instance now exists, so its full (options) config entries can be resolved
1335 # (get_config_entries is an instance method). Rehydrate the config values from
1336 # storage against those entries and validate the complete config, before async
1337 # init so get_config_value reads there see the stored values.
1338 async with _provider_load_step(domain, "resolve its configuration", PROVIDER_SETUP_TIMEOUT):
1339 await self.config.rehydrate_provider_config(provider)
1340 try:
1341 provider.config.validate()
1342 except (KeyError, ValueError, AttributeError, TypeError) as err:
1343 # name the offending entry: the generic message alone gives no clue which
1344 # value is missing or malformed when a provider refuses to load
1345 msg = f"Configuration is invalid: {err}"
1346 raise SetupFailedError(msg) from err
1347
1348 # run async setup
1349 async with _provider_load_step(domain, "initialize", PROVIDER_ASYNC_INIT_TIMEOUT):
1350 await provider.handle_async_init()
1351
1352 await self._register_loaded_provider(provider, conf)
1353
1354 async def _register_loaded_provider(
1355 self, provider: ProviderInstanceType, conf: ProviderConfig
1356 ) -> None:
1357 """Register a provider that finished its setup and run its post-load steps."""
1358 # the instance is now live: register it so the post-load steps below can resolve it
1359 self._providers[provider.instance_id] = provider
1360 provider.available = True
1361
1362 # adapt logging name if needed
1363 provider._set_log_level_from_config(provider.config)
1364
1365 try:
1366 async with _provider_load_step(
1367 provider.domain, "finish loading", PROVIDER_SETUP_TIMEOUT
1368 ):
1369 await self._update_available_providers_cache()
1370 if isinstance(provider, MusicProvider):
1371 await self.music.on_provider_loaded(provider)
1372 if isinstance(provider, PlayerProvider):
1373 await self.players.on_provider_loaded(provider)
1374 except Exception:
1375 # a provider that did not finish loading must not stay registered: it would
1376 # report status LOADED while an error is recorded against it, which leaves the
1377 # user with a warning they can only find by opening the provider's own settings
1378 try:
1379 await self.unload_provider(provider.instance_id)
1380 except Exception as unload_err:
1381 # the load failure is the one worth reporting, so keep it as the raised error
1382 LOGGER.warning(
1383 "Error unloading provider %s: %s",
1384 provider.name,
1385 unload_err,
1386 exc_info=unload_err,
1387 )
1388 raise
1389
1390 # if we reach this point, the provider loaded successfully
1391 LOGGER.info(
1392 "Loaded %s provider %s",
1393 provider.type.value,
1394 provider.name,
1395 )
1396
1397 # execute post load actions
1398 async def _on_provider_loaded() -> None:
1399 try:
1400 await provider.loaded_in_mass()
1401 except Exception as err:
1402 # the provider stays registered and available either way, so the steps
1403 # below still run: an event left unset makes every waiter pay the full
1404 # timeout, on every attempt, until the provider reloads
1405 LOGGER.warning(
1406 "Error in the post load step of provider %s: %s",
1407 provider.name,
1408 str(err) or err.__class__.__name__,
1409 exc_info=err,
1410 )
1411 provider.initialized.set()
1412 self.get_provider_ready_event(provider.domain).set()
1413 await self.run_provider_discovery(provider.instance_id)
1414 # push instance name to config (to persist it if it was autogenerated)
1415 if provider.default_name != conf.default_name:
1416 self.config.set_provider_default_name(provider.instance_id, provider.default_name)
1417
1418 self.create_task(_on_provider_loaded())
1419
1420 # clear any previous error in config and signal update
1421 self.config.set(f"{CONF_PROVIDERS}/{conf.instance_id}/last_error", None)
1422 self.signal_event(EventType.PROVIDERS_UPDATED, data=self.get_providers())
1423
1424 async def __load_provider_manifests(self) -> None:
1425 """Preload all available provider manifest files."""
1426
1427 async def load_provider_manifest(provider_domain: str, provider_path: str) -> None:
1428 """Preload all available provider manifest files."""
1429 # get files in subdirectory
1430 for file_str in await asyncio.to_thread(os.listdir, provider_path): # noqa: PTH208, RUF100
1431 file_path = os.path.join(provider_path, file_str)
1432 if not await isfile(file_path):
1433 continue
1434 if file_str != "manifest.json":
1435 continue
1436 try:
1437 provider_manifest: ProviderManifest = await ProviderManifest.parse(file_path)
1438 # detect provider icon image variants (svg preferred over png)
1439 icons = await detect_provider_icons(provider_path)
1440 if icons:
1441 self._provider_icons[provider_manifest.domain] = icons
1442 provider_manifest.icon_images = list(icons)
1443 # detect a setup_flow.py module by its mere presence: importing it
1444 # here would trigger installing the provider's requirements
1445 provider_manifest.has_setup_flow = await isfile(
1446 os.path.join(provider_path, "setup_flow.py")
1447 )
1448 # override Home Assistant provider if we're running as add-on
1449 if provider_manifest.domain == "hass" and self.running_as_hass_addon:
1450 provider_manifest.builtin = True
1451 provider_manifest.allow_disable = False
1452
1453 self._provider_manifests[provider_manifest.domain] = provider_manifest
1454 LOGGER.log(
1455 VERBOSE_LOG_LEVEL, "Loaded manifest for provider %s", provider_manifest.name
1456 )
1457 except Exception as exc:
1458 LOGGER.exception(
1459 "Error while loading manifest for provider %s",
1460 provider_domain,
1461 exc_info=exc,
1462 )
1463
1464 async with TaskManager(self) as tg:
1465 for dir_str in await asyncio.to_thread(os.listdir, PROVIDERS_PATH): # noqa: PTH208, RUF100
1466 if dir_str.startswith("."):
1467 # skip hidden directories
1468 continue
1469 dir_path = os.path.join(PROVIDERS_PATH, dir_str)
1470 if dir_str.startswith("_") and not self.dev_mode:
1471 # only load demo/test providers if debug mode is enabled (e.g. for development)
1472 continue
1473 if not await isdir(dir_path):
1474 continue
1475 tg.create_task(load_provider_manifest(dir_str, dir_path))
1476 self.logger.debug("Loaded %s provider manifests", len(self._provider_manifests))
1477
1478 async def _update_available_providers_cache(self) -> None:
1479 """Update the global cache variable of loaded/available providers."""
1480 await set_global_cache_values(
1481 {
1482 "provider_domains": {x.domain for x in self.providers},
1483 "provider_instance_ids": {x.instance_id for x in self.providers},
1484 "available_providers": {
1485 *{x.domain for x in self.providers},
1486 *{x.instance_id for x in self.providers},
1487 },
1488 "unique_providers": self.music.get_unique_providers(),
1489 "streaming_providers": {
1490 x.domain
1491 for x in self.providers
1492 if is_music_provider(x) and x.is_streaming_provider
1493 },
1494 "non_streaming_providers": {
1495 x.instance_id
1496 for x in self.providers
1497 if not (is_music_provider(x) and x.is_streaming_provider)
1498 },
1499 }
1500 )
1501
1502 async def _setup_storage(self) -> None:
1503 """Handle Setup of storage/cache folder(s)."""
1504 if not await isdir(self.storage_path):
1505 await mkdirs(self.storage_path)
1506 if not await isdir(self.cache_path):
1507 await mkdirs(self.cache_path)
1508
1509 def _set_state(self, new_state: CoreState) -> None:
1510 """Set new state and signal state change."""
1511 if self._state == new_state:
1512 return
1513 self._state = new_state
1514 if not hasattr(self, "webserver"):
1515 # a startup that failed before the core controllers were created has no
1516 # server info to report and no subscribers to report it to, while the state
1517 # itself must still change so that shutdown can run to completion
1518 return
1519 self.signal_event(EventType.CORE_STATE_UPDATED, data=self.get_server_info())
1520