music-assistant-server

1"""VBAN protocol receiver plugin for Music Assistant."""
2
3from __future__ import annotations
4
5import asyncio
6import re
7from collections.abc import AsyncGenerator
8from contextlib import suppress
9from typing import TYPE_CHECKING, cast
10
11from aiovban.asyncio.util import BackPressureStrategy
12from aiovban.enums import VBANSampleRate
13from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
14from music_assistant_models.enums import (
15    ConfigEntryType,
16    ContentType,
17    ProviderFeature,
18    StreamType,
19)
20from music_assistant_models.errors import SetupFailedError
21from music_assistant_models.media_items import AudioFormat
22from music_assistant_models.streamdetails import StreamMetadata
23
24from music_assistant.constants import (
25    CONF_BIND_IP,
26    CONF_BIND_PORT,
27    CONF_ENTRY_WARN_PREVIEW,
28)
29from music_assistant.helpers.util import (
30    get_ip_addresses,
31)
32from music_assistant.models.plugin import PluginProvider, PluginSource
33
34from .vban import AsyncVBANClientMod
35
36if TYPE_CHECKING:
37    from aiovban.asyncio.device import VBANDevice
38    from aiovban.asyncio.streams import VBANIncomingStream
39    from music_assistant_models.config_entries import ConfigValueType, ProviderConfig
40    from music_assistant_models.provider import ProviderManifest
41
42    from music_assistant.mass import MusicAssistant
43    from music_assistant.models import ProviderInstanceType
44
45DEFAULT_UDP_PORT = 6980
46DEFAULT_PCM_AUDIO_FORMAT = "S16LE"
47DEFAULT_PCM_SAMPLE_RATE = 44100
48DEFAULT_AUDIO_CHANNELS = 2
49
50CONF_VBAN_STREAM_NAME = "vban_stream_name"
51CONF_SENDER_HOST = "sender_host"
52CONF_PCM_AUDIO_FORMAT = "audio_format"
53CONF_PCM_SAMPLE_RATE = "sample_rate"
54CONF_AUDIO_CHANNELS = "audio_channels"
55CONF_VBAN_QUEUE_STRATEGY = "vban_queue_strategy"
56CONF_VBAN_QUEUE_SIZE = "vban_queue_size"
57
58VBAN_QUEUE_STRATEGIES = {
59    "Clear entire queue": BackPressureStrategy.DROP,
60    "Clear the oldest half of the queue": BackPressureStrategy.DRAIN_OLDEST,
61    "Remove single oldest queue entry": BackPressureStrategy.POP,
62}
63
64SUPPORTED_FEATURES = {ProviderFeature.AUDIO_SOURCE}
65
66
67def _get_supported_pcm_formats() -> dict[str, int]:
68    """Return supported PCM formats."""
69    pcm_formats = {}
70    for content_type in ContentType.__members__:
71        if match := re.match(r"PCM_([S|F](\d{2})LE)", content_type):
72            pcm_formats[match.group(1)] = int(match.group(2))
73    return pcm_formats
74
75
76def _get_vban_sample_rates() -> list[int]:
77    """Return supported VBAN sample rates."""
78    return [int(member.split("_")[1]) for member in VBANSampleRate.__members__]
79
80
81async def setup(
82    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
83) -> ProviderInstanceType:
84    """Initialize provider(instance) with given configuration."""
85    return VBANReceiverProvider(mass, manifest, config)
86
87
88async def get_config_entries(
89    mass: MusicAssistant,  # noqa: ARG001
90    instance_id: str | None = None,  # noqa: ARG001
91    action: str | None = None,  # noqa: ARG001
92    values: dict[str, ConfigValueType] | None = None,  # noqa: ARG001
93) -> tuple[ConfigEntry, ...]:
94    """
95    Return Config entries to setup this provider.
96
97    instance_id: id of an existing provider instance (None if new instance setup).
98    action: [optional] action key called from config entries UI.
99    values: the (intermediate) raw values for config entries sent with the action.
100    """
101    ip_addresses = await get_ip_addresses()
102
103    def _validate_stream_name(config_value: str) -> bool:
104        """Validate stream name."""
105        try:
106            config_value.encode("ascii")
107        except UnicodeEncodeError:
108            return False
109        return len(config_value) < 17
110
111    return (
112        CONF_ENTRY_WARN_PREVIEW,
113        ConfigEntry(
114            key=CONF_BIND_PORT,
115            type=ConfigEntryType.INTEGER,
116            default_value=DEFAULT_UDP_PORT,
117            label="Receiver: UDP Port",
118            description="The UDP port the VBAN receiver will listen on for connections. "
119            "Make sure that this server can be reached "
120            "on the given IP and UDP port by remote VBAN senders.",
121        ),
122        ConfigEntry(
123            key=CONF_VBAN_STREAM_NAME,
124            type=ConfigEntryType.STRING,
125            label="Sender: VBAN Stream Name",
126            default_value="Network AUX",
127            description="Max 16 ASCII chars.\n"
128            "The VBAN stream name to expect from the remote VBAN sender.\n"
129            "This MUST match what the remote VBAN sender has set for the session name "
130            "otherwise audio streaming will not work.",
131            required=True,
132            validate=_validate_stream_name,  # type: ignore[arg-type]
133        ),
134        ConfigEntry(
135            key=CONF_SENDER_HOST,
136            type=ConfigEntryType.STRING,
137            default_value="127.0.0.1",
138            label="Sender: VBAN Sender hostname/IP address",
139            description="The hostname/IP Address of the remote VBAN SENDER.",
140            required=True,
141        ),
142        ConfigEntry(
143            key=CONF_PCM_AUDIO_FORMAT,
144            type=ConfigEntryType.STRING,
145            default_value=DEFAULT_PCM_AUDIO_FORMAT,
146            options=[ConfigValueOption(x, x) for x in _get_supported_pcm_formats()],
147            label="PCM audio format",
148            description="The VBAN PCM audio format to expect from the remote VBAN sender. "
149            "This MUST match what the remote VBAN sender has set otherwise audio streaming "
150            "will not work.",
151            required=True,
152        ),
153        ConfigEntry(
154            key=CONF_PCM_SAMPLE_RATE,
155            type=ConfigEntryType.INTEGER,
156            default_value=DEFAULT_PCM_SAMPLE_RATE,
157            options=[ConfigValueOption(str(x), x) for x in _get_vban_sample_rates()],
158            label="PCM sample rate",
159            description="The VBAN PCM sample rate to expect from the remote VBAN sender. "
160            "This MUST match what the remote VBAN sender has set otherwise audio streaming "
161            "will not work.",
162            required=True,
163        ),
164        ConfigEntry(
165            key=CONF_AUDIO_CHANNELS,
166            type=ConfigEntryType.INTEGER,
167            default_value=DEFAULT_AUDIO_CHANNELS,
168            options=[ConfigValueOption(str(x), x) for x in list(range(1, 9))],
169            label="Channels",
170            description="The number of audio channels",
171            required=True,
172        ),
173        ConfigEntry(
174            key=CONF_BIND_IP,
175            type=ConfigEntryType.STRING,
176            default_value="0.0.0.0",
177            options=[ConfigValueOption(x, x) for x in {"0.0.0.0", *ip_addresses}],
178            label="Receiver: Bind to IP/interface",
179            description="Start the VBAN receiver on this specific interface. \n"
180            "Use 0.0.0.0 to bind to all interfaces, which is the default. \n"
181            "This is an advanced setting that should normally "
182            "not be adjusted in regular setups.",
183            category="advanced",
184            required=True,
185        ),
186        ConfigEntry(
187            key=CONF_VBAN_QUEUE_STRATEGY,
188            type=ConfigEntryType.STRING,
189            default_value=next(iter(VBAN_QUEUE_STRATEGIES)),
190            options=[ConfigValueOption(x, x) for x in VBAN_QUEUE_STRATEGIES],
191            label="Receiver: VBAN queue strategy",
192            description="What should happen if the receiving queue fills up?",
193            category="advanced",
194            required=True,
195        ),
196        ConfigEntry(
197            key=CONF_VBAN_QUEUE_SIZE,
198            type=ConfigEntryType.INTEGER,
199            default_value=AsyncVBANClientMod.default_queue_size,
200            label="Receiver: VBAN packets queue size",
201            description="This can be increased if MA is running on a very low power device, "
202            "otherwise this should not need to be changed.",
203            category="advanced",
204            required=True,
205        ),
206    )
207
208
209class VBANReceiverProvider(PluginProvider):
210    """Implementation of a VBAN protocol receiver plugin."""
211
212    def __init__(
213        self, mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
214    ) -> None:
215        """Initialize MusicProvider."""
216        super().__init__(mass, manifest, config, SUPPORTED_FEATURES)
217        self._bind_port: int = cast("int", self.config.get_value(CONF_BIND_PORT))
218        self._bind_ip: str = cast("str", self.config.get_value(CONF_BIND_IP))
219        self._sender_host: str = cast("str", self.config.get_value(CONF_SENDER_HOST))
220        self._vban_stream_name: str = cast("str", self.config.get_value(CONF_VBAN_STREAM_NAME))
221        self._pcm_audio_format: str = cast("str", self.config.get_value(CONF_PCM_AUDIO_FORMAT))
222        self._pcm_sample_rate: int = cast("int", self.config.get_value(CONF_PCM_SAMPLE_RATE))
223        self._audio_channels: int = cast("int", self.config.get_value(CONF_AUDIO_CHANNELS))
224        self._vban_queue_strategy: BackPressureStrategy = VBAN_QUEUE_STRATEGIES[
225            cast("str", self.config.get_value(CONF_VBAN_QUEUE_STRATEGY))
226        ]
227        self._vban_queue_size: int = cast("int", self.config.get_value(CONF_VBAN_QUEUE_SIZE))
228
229        self._vban_receiver: AsyncVBANClientMod | None = None
230        self._vban_sender: VBANDevice | None = None
231        self._vban_stream: VBANIncomingStream | None = None
232
233        self._source_details = PluginSource(
234            id=self.instance_id,
235            name=f"{self.manifest.name}: {self._vban_stream_name}",
236            passive=False,
237            can_play_pause=False,
238            can_seek=False,
239            can_next_previous=False,
240            audio_format=AudioFormat(
241                content_type=ContentType(self._pcm_audio_format.lower()),
242                codec_type=ContentType(self._pcm_audio_format.lower()),
243                sample_rate=self._pcm_sample_rate,
244                bit_depth=_get_supported_pcm_formats()[self._pcm_audio_format],
245                channels=self._audio_channels,
246            ),
247            metadata=StreamMetadata(
248                title=self._vban_stream_name,
249                artist=self._sender_host,
250            ),
251            stream_type=StreamType.CUSTOM,
252        )
253
254    @property
255    def instance_name_postfix(self) -> str | None:
256        """Return a (default) instance name postfix for this provider instance."""
257        return self._vban_stream_name
258
259    async def handle_async_init(self) -> None:
260        """Handle async initialization of the provider."""
261        self._vban_receiver = AsyncVBANClientMod(
262            default_queue_size=self._vban_queue_size, ignore_audio_streams=False
263        )
264        try:
265            self._udp_socket_task = asyncio.create_task(
266                self._vban_receiver.listen(
267                    address=self._bind_ip, port=self._bind_port, controller=self
268                )
269            )
270        except OSError as err:
271            raise SetupFailedError(f"Failed to start VBAN receiver plugin: {err}") from err
272
273        self._vban_sender = self._vban_receiver.register_device(self._sender_host)
274        if self._vban_sender:
275            self._vban_stream = self._vban_sender.receive_stream(
276                self._vban_stream_name, back_pressure_strategy=self._vban_queue_strategy
277            )
278
279    async def unload(self, is_removed: bool = False) -> None:
280        """Handle close/cleanup of the provider."""
281        self.logger.debug("Unloading plugin")
282        if self._vban_receiver:
283            self.logger.debug("Closing UDP transport")
284            self._vban_receiver.close()
285            with suppress(asyncio.CancelledError):
286                await self._udp_socket_task
287
288        self._vban_receiver = None
289        self._vban_sender = None
290        self._vban_stream = None
291        await asyncio.sleep(0.1)
292
293    def get_source(self) -> PluginSource:
294        """Get (audio)source details for this plugin."""
295        return self._source_details
296
297    @property
298    def active_player(self) -> bool:
299        """Report the active player status."""
300        return bool(self._source_details.in_use_by)
301
302    async def get_audio_stream(self, player_id: str) -> AsyncGenerator[bytes, None]:
303        """Yield raw PCM chunks from the VBANIncomingStream queue."""
304        self.logger.debug(
305            "Getting VBAN PCM audio stream for Player: %s//Stream: %s//Config: %s",
306            player_id,
307            self._vban_stream_name,
308            self._source_details.audio_format.output_format_str,
309        )
310        while (
311            self._source_details.in_use_by
312            and self._vban_stream
313            and not self._udp_socket_task.done()
314        ):
315            try:
316                packet = await self._vban_stream.get_packet()
317            except asyncio.QueueShutDown:  # type: ignore[attr-defined]
318                self.logger.error(
319                    "Found VBANIncomingStream queue shut down when attempting to get VBAN packet"
320                )
321                break
322
323            yield packet.body.data
324