/
/
/
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 name=self.webserver.server_name,
433 base_url=self.webserver.base_url,
434 internal_url=self.webserver.base_url,
435 external_url=self.webserver.external_url,
436 has_remote_access=self.webserver.remote_access.is_enabled,
437 homeassistant_addon=self.running_as_hass_addon,
438 onboard_done=self.config.onboard_done,
439 status=self._state,
440 )
441
442 @api_command("time", authenticated=False)
443 def get_server_time(self) -> float:
444 """
445 Return the current server time as UTC timestamp (seconds since epoch).
446
447 Clients compare server-provided timestamps (such as `elapsed_time_last_updated`)
448 against their own clock. Round-tripping this command lets a client estimate the
449 offset between the two clocks and correct for it, so a device with an unsynced
450 clock still renders playback progress and countdowns correctly.
451 """
452 return time.time()
453
454 @api_command("providers/manifests", required_scope=Scope.PROVIDERS_READ)
455 def get_provider_manifests(self) -> list[ProviderManifest]:
456 """Return all Provider manifests."""
457 return list(self._provider_manifests.values())
458
459 @api_command("providers/manifests/get", required_scope=Scope.PROVIDERS_READ)
460 def get_provider_manifest(self, instance_id_or_domain: str) -> ProviderManifest:
461 """Return Provider manifests of single provider(domain)."""
462 if instance_id_or_domain in self._provider_manifests:
463 return self._provider_manifests[instance_id_or_domain]
464 if provider := self.get_provider(instance_id_or_domain, return_unavailable=True):
465 return provider.manifest
466 raise KeyError(f"Provider manifest not found for {instance_id_or_domain}")
467
468 @api_command("providers/icon", required_scope=Scope.PROVIDERS_READ)
469 def get_provider_icon_data(
470 self,
471 provider: str,
472 variant: ProviderIconVariant = ProviderIconVariant.DEFAULT,
473 ) -> str | None:
474 """
475 Return a provider icon variant as a base64 data URI.
476
477 :param provider: A provider domain or instance id.
478 :param variant: Which icon variant to return.
479 """
480 icon = self.get_provider_icon(provider, variant)
481 if icon is None:
482 return None
483 mime, data = icon
484 return f"data:{mime};base64,{b64encode(data).decode('ascii')}"
485
486 def get_provider_icon(
487 self,
488 provider: str,
489 variant: ProviderIconVariant = ProviderIconVariant.DEFAULT,
490 ) -> tuple[str, bytes] | None:
491 """
492 Return the (mime, bytes) for a provider icon variant.
493
494 :param provider: A provider domain or instance id.
495 :param variant: Which icon variant to return.
496 """
497 domain = provider
498 if domain not in self._provider_icons:
499 try:
500 domain = self.get_provider_manifest(provider).domain
501 except KeyError:
502 return None
503 icons = self._provider_icons.get(domain)
504 if not icons:
505 return None
506 return icons.get(variant)
507
508 @api_command("providers", required_scope=Scope.PROVIDERS_READ)
509 def get_providers(
510 self, provider_type: ProviderType | None = None
511 ) -> list[ProviderInstanceType]:
512 """
513 Return all loaded/running Providers (instances).
514
515 Optionally filtered by ProviderType.
516 Note that this applies user filters for music providers (for non admin users).
517 """
518 user = get_current_user()
519 user_provider_filter = (
520 user.provider_filter if user and not has_scope(user, Scope.ALL) else None
521 )
522 return [
523 x
524 for x in list(self._providers.values())
525 if (provider_type is None or provider_type == x.type)
526 # apply user provider filter
527 and (
528 not user_provider_filter
529 or x.instance_id in user_provider_filter
530 or x.type != ProviderType.MUSIC
531 )
532 ]
533
534 @api_command("logging/get", required_scope=Scope.SYSTEM_MANAGE)
535 async def get_application_log(self) -> str:
536 """Return the application log from file."""
537 logfile = os.path.join(self.storage_path, "musicassistant.log")
538 async with aiofiles.open(logfile) as _file:
539 return str(await _file.read())
540
541 @property
542 def providers(self) -> list[ProviderInstanceType]:
543 """
544 Return all loaded/running Providers (instances).
545
546 Note that this skips user filters so may only be called from internal code.
547 """
548 return list(self._providers.values())
549
550 @overload
551 def get_provider(
552 self,
553 provider_instance_or_domain: str,
554 return_unavailable: bool = False,
555 provider_type: None = None,
556 ) -> ProviderInstanceType | None: ...
557
558 @overload
559 def get_provider(
560 self,
561 provider_instance_or_domain: str,
562 return_unavailable: bool = False,
563 *,
564 provider_type: type[_ProviderT],
565 ) -> _ProviderT | None: ...
566
567 def get_provider(
568 self,
569 provider_instance_or_domain: str,
570 return_unavailable: bool = False,
571 provider_type: type[_ProviderT] | None = None,
572 ) -> ProviderInstanceType | _ProviderT | None:
573 """
574 Return provider by instance id or domain.
575
576 :param provider_instance_or_domain: Instance ID or domain of the provider.
577 :param return_unavailable: Also return unavailable providers.
578 :param provider_type: Optional type hint for the expected provider type (unused at runtime).
579 """
580 # lookup by instance_id first
581 if prov := self._providers.get(provider_instance_or_domain):
582 if return_unavailable or prov.available:
583 return prov
584 if not getattr(prov, "is_streaming_provider", None):
585 # no need to lookup other instances because this provider has unique data
586 return None
587 provider_instance_or_domain = prov.domain
588 # fallback to match on domain
589 for prov in list(self._providers.values()):
590 if prov.domain != provider_instance_or_domain:
591 continue
592 if return_unavailable or prov.available:
593 return prov
594 return None
595
596 def get_provider_ready_event(self, domain: str) -> asyncio.Event:
597 """Get (or create) an asyncio.Event that is set when a provider of the given domain is loaded."""
598 if domain not in self._provider_ready_events:
599 self._provider_ready_events[domain] = asyncio.Event()
600 return self._provider_ready_events[domain]
601
602 def get_provider_instances(
603 self,
604 domain: str,
605 return_unavailable: bool = False,
606 provider_type: ProviderType | None = None,
607 ) -> list[ProviderInstanceType]:
608 """
609 Return all provider instances for a given domain.
610
611 Note that this skips user filters so may only be called from internal code.
612 """
613 return [
614 prov
615 for prov in list(self._providers.values())
616 if (provider_type is None or provider_type == prov.type)
617 and prov.domain == domain
618 and (return_unavailable or prov.available)
619 ]
620
621 def get_providers_supporting_feature(
622 self,
623 feature: ProviderFeature,
624 priority: tuple[ProviderType, ...] = (
625 ProviderType.MUSIC,
626 ProviderType.METADATA,
627 ProviderType.PLUGIN,
628 ),
629 ) -> list[ProviderInstanceType]:
630 """
631 Return all available providers that support the given feature.
632
633 Results are grouped by provider type in the order given by ``priority``,
634 and sorted within each tier by the provider's ``priority`` attribute
635 (lower value = higher priority).
636
637 :param feature: The ProviderFeature to query for.
638 :param priority: Ordered tuple of ProviderType values indicating tier order.
639 Types omitted from this tuple are excluded from the results.
640 """
641 by_tier: dict[ProviderType, list[ProviderInstanceType]] = {ptype: [] for ptype in priority}
642 for prov in self.get_providers():
643 if not prov.available:
644 continue
645 if prov.type not in by_tier:
646 continue
647 if feature not in prov.supported_features:
648 continue
649 by_tier[prov.type].append(prov)
650 result: list[ProviderInstanceType] = []
651 for ptype in priority:
652 result.extend(sorted(by_tier[ptype], key=lambda p: getattr(p, "priority", 50)))
653 return result
654
655 def signal_event(
656 self,
657 event: EventType,
658 object_id: str | None = None,
659 data: Any = None,
660 ) -> None:
661 """Signal event to subscribers."""
662 if self.closing:
663 return
664
665 self.verify_event_loop_thread("signal_event")
666
667 if LOGGER.isEnabledFor(VERBOSE_LOG_LEVEL):
668 # do not log queue time updated events because that is too chatty
669 LOGGER.getChild("event").log(VERBOSE_LOG_LEVEL, "%s %s", event.value, object_id or "")
670
671 event_obj = MassEvent(event=event, object_id=object_id, data=data)
672 for cb_func, event_filter, id_filter, is_coro in list(self._subscribers):
673 if not (event_filter is None or event in event_filter):
674 continue
675 if not (id_filter is None or object_id in id_filter):
676 continue
677 if is_coro:
678 if TYPE_CHECKING:
679 cb_func = cast("Callable[[MassEvent], Coroutine[Any, Any, None]]", cb_func)
680 self.create_task(cb_func, event_obj)
681 else:
682 if TYPE_CHECKING:
683 cb_func = cast("Callable[[MassEvent], None]", cb_func)
684 self.loop.call_soon(cb_func, event_obj)
685
686 def subscribe(
687 self,
688 cb_func: EventCallBackType,
689 event_filter: EventType | tuple[EventType, ...] | None = None,
690 id_filter: str | tuple[str, ...] | None = None,
691 ) -> Callable[[], None]:
692 """
693 Add callback to event listeners.
694
695 Returns function to remove the listener.
696 :param cb_func: callback function or coroutine
697 :param event_filter: Optionally only listen for these events
698 :param id_filter: Optionally only listen for these id's (player_id, queue_id, uri)
699 """
700 if isinstance(event_filter, EventType):
701 event_filter = (event_filter,)
702 if isinstance(id_filter, str):
703 id_filter = (id_filter,)
704 # precompute whether the callback is a coroutine so signal_event does not have to
705 # re-derive it via reflection for every subscriber on every (high-frequency) event
706 listener = (cb_func, event_filter, id_filter, inspect.iscoroutinefunction(cb_func))
707 self._subscribers.add(listener)
708
709 def remove_listener() -> None:
710 self._subscribers.remove(listener)
711
712 return remove_listener
713
714 def create_task(
715 self,
716 target: Callable[..., Coroutine[Any, Any, _R]] | Awaitable[_R],
717 *args: Any,
718 task_id: str | None = None,
719 abort_existing: bool = False,
720 eager_start: bool = True,
721 log_exceptions: bool = True,
722 **kwargs: Any,
723 ) -> asyncio.Task[_R]:
724 """
725 Create Task on (main) event loop from Coroutine(function).
726
727 Tasks created by this helper will be properly cancelled on stop.
728
729 :param target: Coroutine function or awaitable to run as a task.
730 :param args: Arguments to pass to the coroutine function.
731 :param task_id: Optional ID to track and deduplicate tasks.
732 :param abort_existing: If True, cancel existing task with same task_id.
733 :param eager_start: If True (default), start task immediately without waiting
734 for next event loop iteration. This ensures proper ordering
735 when creating multiple tasks in sequence.
736 :param log_exceptions: Set to False when the caller awaits the task and reports
737 its failures itself; the task then logs at debug level
738 instead of warning.
739 :param kwargs: Keyword arguments to pass to the coroutine function.
740 """
741 if task_id and (existing := self._tracked_tasks.get(task_id)) and not existing.done():
742 # prevent duplicate tasks if task_id is given and already present
743 if abort_existing:
744 existing.cancel()
745 else:
746 # close any already-constructed coroutine to avoid "never awaited" warning
747 if inspect.iscoroutine(target):
748 target.close()
749 return existing
750 self.verify_event_loop_thread("create_task")
751
752 if inspect.iscoroutinefunction(target):
753 # coroutine function
754 coro = target(*args, **kwargs)
755 elif inspect.iscoroutine(target):
756 # coroutine
757 coro = target
758 elif callable(target):
759 raise RuntimeError("Function is not a coroutine or coroutine function")
760 else:
761 raise RuntimeError("Target is missing")
762
763 # Use asyncio.Task directly with eager_start for immediate execution
764 task: asyncio.Task[_R] = asyncio.Task(coro, loop=self.loop, eager_start=eager_start)
765
766 if task_id is None:
767 task_id = uuid4().hex
768
769 def task_done_callback(_task: asyncio.Task[Any]) -> None:
770 # done callbacks run one event loop iteration after the task finished, so a
771 # caller may already have replaced the entry with a new task under the same
772 # task_id - only untrack when the entry still points at this task
773 if self._tracked_tasks.get(task_id) is _task:
774 del self._tracked_tasks[task_id]
775 if _task.cancelled():
776 return
777 # always retrieve the exception, otherwise asyncio logs a noisy
778 # "Task exception was never retrieved" error at garbage collection time
779 if err := _task.exception():
780 task_name = _task.get_name() if hasattr(_task, "get_name") else str(_task)
781 # a failure the waiters report themselves is demoted rather than dropped:
782 # work that outlives every waiter (join_task keeps it running) would
783 # otherwise fail without a trace anywhere
784 LOGGER.log(
785 logging.WARNING if log_exceptions else logging.DEBUG,
786 "Exception in task %s - target: %s: %s",
787 task_name,
788 str(target),
789 str(err),
790 exc_info=err if LOGGER.isEnabledFor(logging.DEBUG) else None,
791 )
792
793 self._tracked_tasks[task_id] = task
794 task.add_done_callback(task_done_callback)
795 return task
796
797 def call_later(
798 self,
799 delay: float,
800 target: Coroutine[Any, Any, _R] | Awaitable[_R] | Callable[..., _R],
801 *args: Any,
802 task_id: str | None = None,
803 **kwargs: Any,
804 ) -> asyncio.TimerHandle:
805 """
806 Run callable/awaitable after given delay.
807
808 Use task_id for debouncing.
809 """
810 self.verify_event_loop_thread("call_later")
811
812 if not task_id:
813 task_id = uuid4().hex
814
815 if existing := self._tracked_timers.get(task_id):
816 existing.cancel()
817
818 def _create_task(_target: Coroutine[Any, Any, _R]) -> None:
819 self._tracked_timers.pop(task_id)
820 self.create_task(_target, *args, task_id=task_id, abort_existing=True, **kwargs)
821
822 def _call_sync(_target: Callable[..., _R]) -> None:
823 self._tracked_timers.pop(task_id)
824 _target(*args, **kwargs)
825
826 if inspect.iscoroutinefunction(target) or inspect.iscoroutine(target):
827 # coroutine function
828 if TYPE_CHECKING:
829 target = cast("Coroutine[Any, Any, _R]", target)
830 handle = self.loop.call_later(delay, _create_task, target)
831 else:
832 # regular sync callable
833 if TYPE_CHECKING:
834 target = cast("Callable[..., _R]", target)
835 handle = self.loop.call_later(delay, _call_sync, target)
836 self._tracked_timers[task_id] = handle
837 return handle
838
839 def get_task(self, task_id: str) -> asyncio.Task[Any] | None:
840 """Get existing scheduled task."""
841 if existing := self._tracked_tasks.get(task_id):
842 # prevent duplicate tasks if task_id is given and already present
843 return existing
844 return None
845
846 def cancel_task(self, task_id: str) -> None:
847 """Cancel existing scheduled task."""
848 if existing := self._tracked_tasks.pop(task_id, None):
849 existing.cancel()
850
851 def cancel_timer(self, task_id: str) -> None:
852 """Cancel existing scheduled timer."""
853 if existing := self._tracked_timers.pop(task_id, None):
854 existing.cancel()
855
856 def register_api_command(
857 self,
858 command: str,
859 handler: Callable[..., Coroutine[Any, Any, Any] | AsyncGenerator[Any, Any]],
860 authenticated: bool = True,
861 required_scope: Scope | None = None,
862 allow_impersonation: bool = False,
863 alias: bool = False,
864 ) -> Callable[[], None]:
865 """
866 Dynamically register a command on the API.
867
868 :param command: The command name/path.
869 :param handler: The function to handle the command.
870 :param authenticated: Whether authentication is required (default: True).
871 :param required_scope: Scope required to execute the command,
872 None means any authenticated user.
873 :param allow_impersonation: Whether the command accepts a 'user' argument
874 to execute the command on behalf of another user (default: False).
875 :param alias: Whether this is an alias for backward compatibility (default: False).
876 Aliases are not shown in API documentation but remain functional.
877
878 Returns handle to unregister.
879 """
880 if command in self.command_handlers:
881 msg = f"Command {command} is already registered"
882 raise RuntimeError(msg)
883 self.command_handlers[command] = APICommandHandler.parse(
884 command, handler, authenticated, required_scope, allow_impersonation, alias
885 )
886
887 def unregister() -> None:
888 self.command_handlers.pop(command, None)
889
890 return unregister
891
892 async def load_provider_config(
893 self,
894 prov_conf: ProviderConfig,
895 ) -> None:
896 """Load (or reload) a provider from its config, recording any load failure."""
897 # cancel existing (re)load timer if needed
898 task_id = f"load_provider_{prov_conf.instance_id}"
899 if existing := self._tracked_timers.pop(task_id, None):
900 existing.cancel()
901
902 try:
903 await self._load_provider(prov_conf)
904 except Exception as exc:
905 # persist the failure so the provider surfaces a clear status (e.g. auth_required)
906 # to the UI instead of appearing stuck loading, then propagate to the caller
907 self.config.update_provider_last_error(
908 prov_conf.instance_id, _provider_error_from_exc(exc)
909 )
910 raise
911
912 # (re)load any dependents. The provider itself is loaded at this point, so a problem
913 # in this scan belongs to a dependent (or to nothing at all) and must never be
914 # recorded against - and thus flag - the provider we just loaded successfully.
915 try:
916 # resolving option values here would call get_config_entries() on every loaded
917 # provider (some of which do network i/o), for values _load_provider does not
918 # read: it seeds the stored raw values itself and rehydrates once the instance
919 # exists. Only the manifest-related fields below are needed to spot a dependent.
920 prov_configs = await self.config.get_provider_configs()
921 except Exception as exc:
922 LOGGER.warning(
923 "Error looking up dependents of provider(instance) %s: %s",
924 prov_conf.name or prov_conf.instance_id,
925 str(exc) or exc.__class__.__name__,
926 exc_info=_provider_error_traceback(exc),
927 )
928 return
929 for dep_prov_conf in prov_configs:
930 if not dep_prov_conf.enabled:
931 continue
932 manifest = self.get_provider_manifest(dep_prov_conf.domain)
933 if not manifest.depends_on:
934 continue
935 if manifest.depends_on != prov_conf.domain:
936 continue
937 try:
938 # the scan above skipped the config values, but the load path does need them:
939 # a provider reads config (e.g. its log level) while it is being constructed.
940 # Resolve them here, for this single dependent instead of for every provider.
941 dep_conf = await self.config.get_provider_config(dep_prov_conf.instance_id)
942 except KeyError:
943 # config was removed while we were scanning
944 continue
945 try:
946 await self._load_provider(dep_conf)
947 except Exception as exc:
948 # record the failure against the provider that hit it: attributing it to the
949 # provider we just loaded (which is fine) flags the wrong one in the UI
950 self.config.update_provider_last_error(
951 dep_prov_conf.instance_id, _provider_error_from_exc(exc)
952 )
953 LOGGER.warning(
954 "Error loading provider(instance) %s: %s",
955 dep_prov_conf.name or dep_prov_conf.instance_id,
956 str(exc) or exc.__class__.__name__,
957 exc_info=_provider_error_traceback(exc),
958 )
959
960 async def load_provider(
961 self,
962 instance_id: str,
963 allow_retry: bool = False,
964 remove_if_unsupported: bool = False,
965 ) -> None:
966 """Try to load a provider and catch errors."""
967 try:
968 prov_conf = await self.config.get_provider_config(instance_id)
969 except KeyError:
970 # Was deleted before we could run
971 return
972
973 if not prov_conf.enabled:
974 # Was disabled before we could run
975 return
976
977 # cancel existing (re)load timer if needed
978 task_id = f"load_provider_{instance_id}"
979 if existing := self._tracked_timers.pop(task_id, None):
980 existing.cancel()
981
982 try:
983 await self.load_provider_config(prov_conf)
984 except UnsupportedSystemError as exc:
985 # The host does not meet this provider's hardware requirements. This is a
986 # permanent condition, so we never retry. For a provider that was just
987 # auto-set-up as a default, drop the config again so it does not linger as a
988 # broken provider (it stays marked done so it is not auto-created again).
989 if remove_if_unsupported:
990 LOGGER.info(
991 "Not enabling default provider %s: %s",
992 prov_conf.name or prov_conf.instance_id,
993 exc,
994 )
995 # The provider never loaded, so just drop its auto-created config key.
996 # (remove_provider_config refuses builtin providers and runs loaded-provider
997 # cleanup we don't need here; a direct remove persists and is guard-free.)
998 self.config.remove(f"{CONF_PROVIDERS}/{instance_id}")
999 return
1000 prov_conf.last_error = _provider_error_from_exc(exc)
1001 self.config.update_provider_last_error(instance_id, prov_conf.last_error)
1002 LOGGER.warning(
1003 "Provider(instance) %s can not run on this system: %s",
1004 prov_conf.name or prov_conf.instance_id,
1005 exc,
1006 )
1007 return
1008 except Exception as exc:
1009 # if loading failed, we store the error in the config object
1010 # so we can show something useful to the user
1011 prov_conf.last_error = _provider_error_from_exc(exc)
1012 self.config.update_provider_last_error(instance_id, prov_conf.last_error)
1013
1014 # auto schedule a retry if the (re)load failed with a handled exception
1015 # unhandled exceptions (e.g. ValueError) are likely bugs that won't resolve themselves
1016 will_retry = (
1017 allow_retry
1018 and isinstance(exc, MusicAssistantError)
1019 and not isinstance(
1020 exc,
1021 (AuthenticationRequired, AuthenticationFailed, LoginFailed, InvalidToken),
1022 )
1023 )
1024 if will_retry:
1025 self.call_later(
1026 120,
1027 self.load_provider,
1028 instance_id,
1029 allow_retry,
1030 task_id=task_id,
1031 )
1032 LOGGER.warning(
1033 "Error loading provider(instance) %s: %s%s",
1034 prov_conf.name or prov_conf.instance_id,
1035 str(exc) or exc.__class__.__name__,
1036 " (will be retried later)" if will_retry else "",
1037 exc_info=_provider_error_traceback(exc),
1038 )
1039 return
1040
1041 # (re)load any dependents if needed
1042 for dep_prov in self.providers:
1043 if dep_prov.available:
1044 continue
1045 if dep_prov.manifest.depends_on == prov_conf.domain:
1046 await self.unload_provider(dep_prov.instance_id)
1047
1048 async def unload_provider(self, instance_id: str, is_removed: bool = False) -> None:
1049 """Unload a provider."""
1050 # this waits (bounded) for a running sync to unwind: provider.unload() below tears
1051 # down state the sync may still be using, such as the mount of a network share
1052 await self.music.unschedule_provider_sync(instance_id, clear_persisted_state=is_removed)
1053 if provider := self._providers.get(instance_id):
1054 # mark the provider as on its way out before anything is torn down: the steps
1055 # below have await points, so without this a callback that is still in flight
1056 # could register a player back onto a provider that is already gone
1057 provider.unloading = True
1058 if isinstance(provider, PluginProvider):
1059 # a live source cannot outlive the plugin exposing it: the player would go
1060 # on naming a source that can no longer be streamed, its queue held inactive
1061 await self.players.release_provider_sources(instance_id)
1062 if isinstance(provider, PlayerProvider):
1063 await self.players.on_provider_unload(provider)
1064 if isinstance(provider, MusicProvider):
1065 await self.music.on_provider_unload(provider)
1066 # check if there are no other providers dependent of this provider
1067 for dep_prov in self.providers:
1068 if dep_prov.manifest.depends_on == provider.domain:
1069 await self.unload_provider(dep_prov.instance_id)
1070 try:
1071 if is_player_provider(provider):
1072 # unregister all players of this provider, straight from the registry: the
1073 # provider's own players listing hides disabled and still-initializing
1074 # players, which must be unregistered here too so their on_unload runs
1075 # and no stale entry is left behind
1076 for player in list(self.players):
1077 if player.provider.instance_id != instance_id:
1078 continue
1079 await self.players.unregister(player.player_id, permanent=is_removed)
1080 await provider.unload(is_removed)
1081 except Exception as err:
1082 LOGGER.warning(
1083 "Error while unloading provider %s: %s", provider.name, str(err), exc_info=err
1084 )
1085 finally:
1086 if provider.domain in self._provider_ready_events:
1087 self._provider_ready_events[provider.domain].clear()
1088 self._providers.pop(instance_id, None)
1089 self.discovery.on_provider_unload(instance_id)
1090 await self._update_available_providers_cache()
1091 self.signal_event(EventType.PROVIDERS_UPDATED, data=self.get_providers())
1092
1093 async def unload_provider_with_error(self, instance_id: str, error: str | Exception) -> None:
1094 """
1095 Unload a provider that hit a problem which needs user interaction.
1096
1097 :param error: The originating exception (preferred, so e.g. a LoginFailed surfaces as an
1098 auth-required status with a localized message) or a plain string for a generic error.
1099 """
1100 prov_error = (
1101 _provider_error_from_exc(error)
1102 if isinstance(error, Exception)
1103 else ProviderError(error_code=999, message=error)
1104 )
1105 self.config.update_provider_last_error(instance_id, prov_error)
1106 await self.unload_provider(instance_id)
1107
1108 async def run_provider_discovery(self, instance_id: str) -> None:
1109 """
1110 Run shared discovery for a given provider.
1111
1112 In case of a PlayerProvider, will also call its own discovery method.
1113 """
1114 provider = self.get_provider(instance_id, return_unavailable=False)
1115 if not provider:
1116 raise KeyError(f"Provider with instance ID {instance_id} not found")
1117 await self.discovery.run_provider_discovery(provider)
1118 if isinstance(provider, PlayerProvider):
1119 await provider.discover_players()
1120
1121 def verify_event_loop_thread(self, what: str) -> None:
1122 """Report and raise if we are not running in the event loop thread."""
1123 if self.loop_thread_id != threading.get_ident():
1124 raise RuntimeError(
1125 f"Non-Async operation detected: {what} may only be called from the eventloop."
1126 )
1127
1128 async def __aenter__(self) -> Self:
1129 """Return Context manager."""
1130 await self.start()
1131 return self
1132
1133 async def __aexit__(
1134 self,
1135 exc_type: type[BaseException] | None,
1136 exc_val: BaseException | None,
1137 exc_tb: TracebackType | None,
1138 ) -> bool | None:
1139 """Exit context manager."""
1140 await self.stop()
1141 return None
1142
1143 def _register_api_commands(self) -> None:
1144 """Register all methods decorated as api_command within a class(instance)."""
1145 for cls in (
1146 self,
1147 self.config,
1148 self.metadata,
1149 self.tasks,
1150 self.music,
1151 self.players,
1152 self.player_queues,
1153 self.translations,
1154 self.webserver,
1155 self.webserver.auth,
1156 self.streams.audio_analysis,
1157 self.diagnostics,
1158 self.dashboard,
1159 ):
1160 for attr_name in dir(cls):
1161 if attr_name.startswith("__"):
1162 continue
1163 # Skip properties to avoid triggering lazy initialization side effects
1164 # (e.g. http_session creating an aiohttp connector during registration)
1165 if isinstance(getattr(type(cls), attr_name, None), property):
1166 continue
1167 try:
1168 obj = getattr(cls, attr_name)
1169 except AttributeError, RuntimeError:
1170 # Skip attributes that fail during initialization
1171 continue
1172 if hasattr(obj, "api_cmd"):
1173 # method is decorated with our api decorator
1174 authenticated = getattr(obj, "api_authenticated", True)
1175 required_scope = getattr(obj, "api_required_scope", None)
1176 allow_impersonation = getattr(obj, "api_allow_impersonation", False)
1177 alias = getattr(obj, "api_alias", False)
1178 self.register_api_command(
1179 obj.api_cmd, obj, authenticated, required_scope, allow_impersonation, alias
1180 )
1181
1182 async def _load_core_controllers(self) -> None:
1183 """Instantiate the core controllers and register their manifests and icons."""
1184 self.cache = CacheController(self)
1185 self.tasks = TasksController(self)
1186 self.webserver = WebserverController(self)
1187 self.metadata = MetaDataController(self)
1188 self.music = MusicController(self)
1189 self.players = PlayerController(self)
1190 self.player_queues = PlayerQueuesController(self)
1191 self.streams = StreamsController(self)
1192 self.translations = TranslationController(self)
1193 self.diagnostics = DiagnosticsController(self)
1194 self.dashboard = DashboardController(self)
1195 # add manifests for core controllers
1196 for controller_name in CONFIGURABLE_CORE_CONTROLLERS:
1197 controller: CoreController = getattr(self, controller_name)
1198 self._provider_manifests[controller.domain] = controller.manifest
1199 # load icon image(s) shipped alongside the controller module
1200 controller_dir = os.path.dirname(inspect.getfile(type(controller)))
1201 if icons := await detect_provider_icons(controller_dir):
1202 self._provider_icons[controller.domain] = icons
1203 controller.manifest.icon_images = list(icons)
1204
1205 async def _load_builtin_providers(self) -> None:
1206 """
1207 Load all builtin providers.
1208
1209 Builtin providers are always needed (also in safe mode) and are fully awaited.
1210 On error, setup will fail.
1211 """
1212 # create default config for any 'builtin' providers
1213 for prov_manifest in self._provider_manifests.values():
1214 if prov_manifest.type == ProviderType.CORE:
1215 # core controllers are not real providers
1216 continue
1217 if not prov_manifest.builtin:
1218 continue
1219 await self.config.create_builtin_provider_config(prov_manifest.domain)
1220
1221 # load all configured (and enabled) builtin providers
1222 # (only manifest-related fields are read here, so the option values are not resolved)
1223 prov_configs = await self.config.get_provider_configs()
1224 builtin_configs: list[ProviderConfig] = [
1225 prov_conf
1226 for prov_conf in prov_configs
1227 if (manifest := self._provider_manifests.get(prov_conf.domain))
1228 and manifest.builtin
1229 and (prov_conf.enabled or manifest.allow_disable is False)
1230 ]
1231
1232 # load builtin providers and wait for them to complete
1233 async with asyncio.TaskGroup() as tg:
1234 for conf in builtin_configs:
1235 tg.create_task(self.load_provider(conf.instance_id, allow_retry=True))
1236
1237 async def _load_providers(self) -> None:
1238 """
1239 Load regular (non-builtin) providers from config.
1240
1241 Regular providers are loaded in background tasks
1242 and can fail without affecting core setup.
1243 """
1244 # handle default providers setup
1245 self.config.set_default(CONF_DEFAULT_PROVIDERS_SETUP, set())
1246 default_providers_setup = set(self.config.get(CONF_DEFAULT_PROVIDERS_SETUP))
1247 changes_made = False
1248 newly_created_defaults: set[str] = set()
1249 for default_provider, require_mdns in DEFAULT_PROVIDERS:
1250 if default_provider in default_providers_setup:
1251 # already processed/setup before, skip
1252 continue
1253 if not (manifest := self._provider_manifests.get(default_provider)):
1254 continue
1255 if require_mdns:
1256 # if mdns discovery is required, check if we have seen any mdns entries
1257 # for this provider before setting it up
1258 for mdns_name in set(self.discovery.aiozc.zeroconf.cache.cache):
1259 if manifest.mdns_discovery and any(
1260 mdns_type in mdns_name for mdns_type in manifest.mdns_discovery
1261 ):
1262 break
1263 else:
1264 continue
1265 await self.config.create_builtin_provider_config(manifest.domain)
1266 changes_made = True
1267 newly_created_defaults.add(manifest.domain)
1268 # TEMP: migration - to be removed after 2.8 release
1269 # enable all existing players of the default providers if they are not already enabled
1270 # due to the linked protocol feature we introduced
1271 for player_config in await self.config.get_player_configs(
1272 provider=default_provider, include_disabled=True
1273 ):
1274 if player_config.enabled:
1275 continue
1276 await self.config.save_player_config(player_config.player_id, {"enabled": True})
1277 default_providers_setup.add(default_provider)
1278 if changes_made:
1279 self.config.set(CONF_DEFAULT_PROVIDERS_SETUP, default_providers_setup)
1280 self.config.save(True)
1281 # load all configured (and enabled) regular (non-builtin) providers
1282 # (only manifest-related fields are read here, so the option values are not resolved)
1283 prov_configs = await self.config.get_provider_configs()
1284 other_configs: list[ProviderConfig] = [
1285 prov_conf
1286 for prov_conf in prov_configs
1287 if prov_conf.enabled
1288 and (
1289 not (manifest := self._provider_manifests.get(prov_conf.domain))
1290 or not manifest.builtin
1291 )
1292 ]
1293 # load providers concurrently via tasks, bounded so a host with many providers does
1294 # not import every provider module at once (a torch-backed one costs hundreds of MB)
1295 async with TaskManager(self, PROVIDER_LOAD_CONCURRENCY) as tg:
1296 for prov_conf in other_configs:
1297 # Use a task so we can load multiple providers at once.
1298 # If a provider fails, that will not block the loading of other providers.
1299 # For providers just auto-set-up as a default, drop the config again if the
1300 # host does not meet their requirements (rather than retry a broken provider).
1301 await tg.create_task_with_limit(
1302 self.load_provider(
1303 prov_conf.instance_id,
1304 allow_retry=True,
1305 remove_if_unsupported=prov_conf.domain in newly_created_defaults,
1306 )
1307 )
1308
1309 async def _load_provider(self, conf: ProviderConfig) -> None:
1310 """Load (or reload) a provider."""
1311 # if provider is already loaded, stop and unload it first
1312 await self.unload_provider(conf.instance_id)
1313 LOGGER.debug("Loading provider %s", conf.name or conf.domain)
1314 if not conf.enabled:
1315 msg = "Provider is disabled"
1316 raise SetupFailedError(msg)
1317
1318 # The config is validated after the instance is created and its config rehydrated
1319 # (see below): the full options entries - and thus which values are required - are
1320 # only known once the instance exists.
1321
1322 domain = conf.domain
1323 prov_manifest = self._provider_manifests.get(domain)
1324 # check for other instances of this provider
1325 existing = next((x for x in self.providers if x.domain == domain), None)
1326 if existing and prov_manifest and not prov_manifest.multi_instance:
1327 msg = f"Provider {domain} already loaded and only one instance allowed."
1328 raise SetupFailedError(msg)
1329 # check valid manifest (just in case)
1330 if not prov_manifest:
1331 msg = f"Provider {domain} manifest not found"
1332 raise SetupFailedError(msg)
1333
1334 # handle dependency on other provider
1335 if prov_manifest.depends_on and not self.get_provider(prov_manifest.depends_on):
1336 # we can safely ignore this completely as the setup will be retried later
1337 # automatically when the dependency is loaded
1338 return
1339
1340 # seed the config with its stored raw values so any construction-time option reads
1341 # in setup()/__init__ see them (the fully-typed entries are only resolvable once the
1342 # instance exists, and are applied by rehydrate_provider_config just below)
1343 self.config.seed_stored_config_values(conf)
1344
1345 # try to setup the module
1346 # (unbounded: this may still have to install the provider's requirements)
1347 async with _provider_load_step(domain, "import its module"):
1348 prov_mod = await load_provider_module(domain, prov_manifest.requirements)
1349 async with _provider_load_step(domain, "load", PROVIDER_SETUP_TIMEOUT):
1350 provider = await prov_mod.setup(self, prov_manifest, conf)
1351
1352 # The instance now exists, so its full (options) config entries can be resolved
1353 # (get_config_entries is an instance method). Rehydrate the config values from
1354 # storage against those entries and validate the complete config, before async
1355 # init so get_config_value reads there see the stored values.
1356 async with _provider_load_step(domain, "resolve its configuration", PROVIDER_SETUP_TIMEOUT):
1357 await self.config.rehydrate_provider_config(provider)
1358 try:
1359 provider.config.validate()
1360 except (KeyError, ValueError, AttributeError, TypeError) as err:
1361 # name the offending entry: the generic message alone gives no clue which
1362 # value is missing or malformed when a provider refuses to load
1363 msg = f"Configuration is invalid: {err}"
1364 raise SetupFailedError(msg) from err
1365
1366 # run async setup
1367 async with _provider_load_step(domain, "initialize", PROVIDER_ASYNC_INIT_TIMEOUT):
1368 await provider.handle_async_init()
1369
1370 await self._register_loaded_provider(provider, conf)
1371
1372 async def _register_loaded_provider(
1373 self, provider: ProviderInstanceType, conf: ProviderConfig
1374 ) -> None:
1375 """Register a provider that finished its setup and run its post-load steps."""
1376 # the instance is now live: register it so the post-load steps below can resolve it
1377 self._providers[provider.instance_id] = provider
1378 provider.available = True
1379
1380 # adapt logging name if needed
1381 provider._set_log_level_from_config(provider.config)
1382
1383 try:
1384 async with _provider_load_step(
1385 provider.domain, "finish loading", PROVIDER_SETUP_TIMEOUT
1386 ):
1387 await self._update_available_providers_cache()
1388 if isinstance(provider, MusicProvider):
1389 await self.music.on_provider_loaded(provider)
1390 if isinstance(provider, PlayerProvider):
1391 await self.players.on_provider_loaded(provider)
1392 except Exception:
1393 # a provider that did not finish loading must not stay registered: it would
1394 # report status LOADED while an error is recorded against it, which leaves the
1395 # user with a warning they can only find by opening the provider's own settings
1396 try:
1397 await self.unload_provider(provider.instance_id)
1398 except Exception as unload_err:
1399 # the load failure is the one worth reporting, so keep it as the raised error
1400 LOGGER.warning(
1401 "Error unloading provider %s: %s",
1402 provider.name,
1403 unload_err,
1404 exc_info=unload_err,
1405 )
1406 raise
1407
1408 # if we reach this point, the provider loaded successfully
1409 LOGGER.info(
1410 "Loaded %s provider %s",
1411 provider.type.value,
1412 provider.name,
1413 )
1414
1415 # execute post load actions
1416 async def _on_provider_loaded() -> None:
1417 try:
1418 await provider.loaded_in_mass()
1419 except Exception as err:
1420 # the provider stays registered and available either way, so the steps
1421 # below still run: an event left unset makes every waiter pay the full
1422 # timeout, on every attempt, until the provider reloads
1423 LOGGER.warning(
1424 "Error in the post load step of provider %s: %s",
1425 provider.name,
1426 str(err) or err.__class__.__name__,
1427 exc_info=err,
1428 )
1429 provider.initialized.set()
1430 self.get_provider_ready_event(provider.domain).set()
1431 await self.run_provider_discovery(provider.instance_id)
1432 # push instance name to config (to persist it if it was autogenerated)
1433 if provider.default_name != conf.default_name:
1434 self.config.set_provider_default_name(provider.instance_id, provider.default_name)
1435
1436 self.create_task(_on_provider_loaded())
1437
1438 # clear any previous error in config and signal update
1439 self.config.set(f"{CONF_PROVIDERS}/{conf.instance_id}/last_error", None)
1440 self.signal_event(EventType.PROVIDERS_UPDATED, data=self.get_providers())
1441
1442 async def __load_provider_manifests(self) -> None:
1443 """Preload all available provider manifest files."""
1444
1445 async def load_provider_manifest(provider_domain: str, provider_path: str) -> None:
1446 """Preload all available provider manifest files."""
1447 # get files in subdirectory
1448 for file_str in await asyncio.to_thread(os.listdir, provider_path): # noqa: PTH208, RUF100
1449 file_path = os.path.join(provider_path, file_str)
1450 if not await isfile(file_path):
1451 continue
1452 if file_str != "manifest.json":
1453 continue
1454 try:
1455 provider_manifest: ProviderManifest = await ProviderManifest.parse(file_path)
1456 # detect provider icon image variants (svg preferred over png)
1457 icons = await detect_provider_icons(provider_path)
1458 if icons:
1459 self._provider_icons[provider_manifest.domain] = icons
1460 provider_manifest.icon_images = list(icons)
1461 # detect a setup_flow.py module by its mere presence: importing it
1462 # here would trigger installing the provider's requirements
1463 provider_manifest.has_setup_flow = await isfile(
1464 os.path.join(provider_path, "setup_flow.py")
1465 )
1466 # override Home Assistant provider if we're running as add-on
1467 if provider_manifest.domain == "hass" and self.running_as_hass_addon:
1468 provider_manifest.builtin = True
1469 provider_manifest.allow_disable = False
1470
1471 self._provider_manifests[provider_manifest.domain] = provider_manifest
1472 LOGGER.log(
1473 VERBOSE_LOG_LEVEL, "Loaded manifest for provider %s", provider_manifest.name
1474 )
1475 except Exception as exc:
1476 LOGGER.exception(
1477 "Error while loading manifest for provider %s",
1478 provider_domain,
1479 exc_info=exc,
1480 )
1481
1482 async with TaskManager(self) as tg:
1483 for dir_str in await asyncio.to_thread(os.listdir, PROVIDERS_PATH): # noqa: PTH208, RUF100
1484 if dir_str.startswith("."):
1485 # skip hidden directories
1486 continue
1487 dir_path = os.path.join(PROVIDERS_PATH, dir_str)
1488 if dir_str.startswith("_") and not self.dev_mode:
1489 # only load demo/test providers if debug mode is enabled (e.g. for development)
1490 continue
1491 if not await isdir(dir_path):
1492 continue
1493 tg.create_task(load_provider_manifest(dir_str, dir_path))
1494 self.logger.debug("Loaded %s provider manifests", len(self._provider_manifests))
1495
1496 async def _update_available_providers_cache(self) -> None:
1497 """Update the global cache variable of loaded/available providers."""
1498 await set_global_cache_values(
1499 {
1500 "provider_domains": {x.domain for x in self.providers},
1501 "provider_instance_ids": {x.instance_id for x in self.providers},
1502 "available_providers": {
1503 *{x.domain for x in self.providers},
1504 *{x.instance_id for x in self.providers},
1505 },
1506 "unique_providers": self.music.get_unique_providers(),
1507 "streaming_providers": {
1508 x.domain
1509 for x in self.providers
1510 if is_music_provider(x) and x.is_streaming_provider
1511 },
1512 "non_streaming_providers": {
1513 x.instance_id
1514 for x in self.providers
1515 if not (is_music_provider(x) and x.is_streaming_provider)
1516 },
1517 }
1518 )
1519
1520 async def _setup_storage(self) -> None:
1521 """Handle Setup of storage/cache folder(s)."""
1522 if not await isdir(self.storage_path):
1523 await mkdirs(self.storage_path)
1524 if not await isdir(self.cache_path):
1525 await mkdirs(self.cache_path)
1526
1527 def _set_state(self, new_state: CoreState) -> None:
1528 """Set new state and signal state change."""
1529 if self._state == new_state:
1530 return
1531 self._state = new_state
1532 if not hasattr(self, "webserver"):
1533 # a startup that failed before the core controllers were created has no
1534 # server info to report and no subscribers to report it to, while the state
1535 # itself must still change so that shutdown can run to completion
1536 return
1537 self.signal_event(EventType.CORE_STATE_UPDATED, data=self.get_server_info())
1538