music-assistant-server

1"""Local Audio Out player provider for Music Assistant."""
2
3from __future__ import annotations
4
5import ctypes
6import sys
7
8from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
9from music_assistant_models.enums import ConfigEntryType
10from music_assistant_models.errors import SetupFailedError
11
12from music_assistant.models.player_provider import PlayerProvider
13
14from .constants import (
15    AUDIO_BACKEND_ALSA,
16    AUDIO_BACKEND_AUTO,
17    AUDIO_BACKEND_PULSEAUDIO,
18    CONF_AUDIO_BACKEND,
19)
20from .sendspin_bridge import LocalAudioBridgeManager
21
22
23class LocalAudioProvider(PlayerProvider):
24    """Player provider that exposes locally attached soundcards as Sendspin players."""
25
26    _bridge_manager: LocalAudioBridgeManager
27
28    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
29        """Return Config entries to configure this provider."""
30        entries: list[ConfigEntry] = []
31
32        if sys.platform == "linux":
33            entries.append(
34                ConfigEntry(
35                    key=CONF_AUDIO_BACKEND,
36                    type=ConfigEntryType.STRING,
37                    options=[
38                        ConfigValueOption(AUDIO_BACKEND_AUTO),
39                        ConfigValueOption(AUDIO_BACKEND_PULSEAUDIO),
40                        ConfigValueOption(AUDIO_BACKEND_ALSA),
41                    ],
42                    default_value=AUDIO_BACKEND_AUTO,
43                )
44            )
45
46        return tuple(entries)
47
48    async def handle_async_init(self) -> None:
49        """Handle async initialization of the provider."""
50        if sys.platform == "linux":
51            configured_backend: str = str(
52                self.config.get_value(CONF_AUDIO_BACKEND) or AUDIO_BACKEND_AUTO
53            )
54            needs_pulse = configured_backend in (AUDIO_BACKEND_PULSEAUDIO, AUDIO_BACKEND_AUTO)
55            if needs_pulse:
56                # Verify libpulse-simple is present before attempting PA output.
57                # On AUTO we probe but don't hard-fail — bridge manager will fall
58                # back to ALSA if pactl returns no sinks at enumeration time.
59                try:
60                    ctypes.CDLL("libpulse-simple.so.0")
61                except OSError:
62                    if configured_backend == AUDIO_BACKEND_PULSEAUDIO:
63                        raise SetupFailedError(
64                            "libpulse-simple.so.0 not found — is PulseAudio installed?"
65                        ) from None
66                    # AUTO: libpulse absent, bridge manager will use ALSA
67
68        self._bridge_manager = LocalAudioBridgeManager(self)
69
70    async def discover_players(self) -> None:
71        """Discover local audio output devices and register their players."""
72        await self._bridge_manager.discover_and_register()
73
74    async def unload(self, is_removed: bool = False) -> None:
75        """Handle unload/removal of the provider."""
76        if bridge_manager := getattr(self, "_bridge_manager", None):
77            await bridge_manager.close()
78