music-assistant-server

1"""VBAN receiver provider implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from collections.abc import AsyncGenerator
8from typing import TYPE_CHECKING, Any, cast
9from uuid import uuid4
10
11from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
12from music_assistant_models.enums import ConfigEntryType, ContentType, MediaType, StreamType
13from music_assistant_models.errors import AudioError, MediaNotFoundError, SetupFailedError
14from music_assistant_models.media_items import AudioFormat, AudioSource, ProviderMapping
15from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
16
17from music_assistant.constants import (
18    CONF_BIND_IP,
19    CONF_BIND_PORT,
20    CONF_ENTRY_WARN_PREVIEW,
21    VERBOSE_LOG_LEVEL,
22)
23from music_assistant.models.plugin import PluginProvider
24
25from .constants import (
26    CONF_AUDIO_CHANNELS,
27    CONF_LOG_VBAN_STREAM_STATS,
28    CONF_PCM_AUDIO_FORMAT,
29    CONF_PCM_SAMPLE_RATE,
30    CONF_SENDER_HOST,
31    CONF_VBAN_QUEUE_SIZE,
32    CONF_VBAN_QUEUE_STRATEGY,
33    CONF_VBAN_STREAM_NAME,
34    DEFAULT_AUDIO_CHANNELS,
35    DEFAULT_PCM_AUDIO_FORMAT,
36    DEFAULT_PCM_SAMPLE_RATE,
37    DEFAULT_UDP_PORT,
38    SUPPORTED_FEATURES,
39    VBAN_QUEUE_STRATEGIES,
40)
41from .helpers import get_supported_pcm_formats
42from .stats import VBANStatsReporter
43from .vban import AsyncVBANClientMod
44
45if TYPE_CHECKING:
46    from aiovban.asyncio.streams import VBANIncomingStream
47    from aiovban.asyncio.util import BackPressureStrategy
48    from music_assistant_models.config_entries import ProviderConfig
49    from music_assistant_models.provider import ProviderManifest
50
51    from music_assistant.mass import MusicAssistant
52
53
54# stable id for the single AudioSource this provider exposes;
55# combined with the provider instance_id this forms the persistent uri
56AUDIO_SOURCE_ID = "main"
57
58
59class VBANReceiverProvider(PluginProvider):
60    """Implementation of a VBAN protocol receiver plugin."""
61
62    def __init__(
63        self, mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
64    ) -> None:
65        """Initialize MusicProvider."""
66        super().__init__(mass, manifest, config, SUPPORTED_FEATURES)
67        # Setup values fall back to legacy option values for existing instances.
68        self._bind_port: int = cast("int", self.get_setup_value(CONF_BIND_PORT) or DEFAULT_UDP_PORT)
69        self._bind_ip: str = cast("str", self.get_setup_value(CONF_BIND_IP) or "0.0.0.0")
70        self._sender_host: str = cast("str", self.get_setup_value(CONF_SENDER_HOST) or "127.0.0.1")
71        self._vban_stream_name: str = cast(
72            "str", self.get_setup_value(CONF_VBAN_STREAM_NAME) or "Network AUX"
73        )
74        self._pcm_audio_format: str = cast(
75            "str", self.get_setup_value(CONF_PCM_AUDIO_FORMAT) or DEFAULT_PCM_AUDIO_FORMAT
76        )
77        self._pcm_sample_rate: int = cast(
78            "int", self.get_setup_value(CONF_PCM_SAMPLE_RATE) or DEFAULT_PCM_SAMPLE_RATE
79        )
80        self._audio_channels: int = cast(
81            "int", self.get_setup_value(CONF_AUDIO_CHANNELS) or DEFAULT_AUDIO_CHANNELS
82        )
83        self._vban_queue_strategy: BackPressureStrategy = VBAN_QUEUE_STRATEGIES[
84            cast(
85                "str",
86                self.config.get_value(CONF_VBAN_QUEUE_STRATEGY)
87                or next(iter(VBAN_QUEUE_STRATEGIES)),
88            )
89        ]
90        self._vban_queue_size: int = cast(
91            "int",
92            self.config.get_value(CONF_VBAN_QUEUE_SIZE) or AsyncVBANClientMod.default_queue_size,
93        )
94        self._log_stats: bool = cast("bool", self.config.get_value(CONF_LOG_VBAN_STREAM_STATS))
95
96        self._vban_receiver: AsyncVBANClientMod | None = None
97        self._vban_stream: VBANIncomingStream | None = None
98        self._udp_socket_fut: asyncio.Future[Any] | None = None
99        self._stats_reporter: VBANStatsReporter | None = None
100        self._active_stream_id: str = ""
101        self._in_use_by_player: str | None = None
102        # _active_session_id is the controller-provided token for the current
103        # stream request — used to reject stale on_source_unselected callbacks
104        # after a same-queue reconnect supersedes the previous request.
105        self._active_session_id: str | None = None
106
107        self._audio_format = AudioFormat(
108            content_type=ContentType(self._pcm_audio_format.lower()),
109            codec_type=ContentType(self._pcm_audio_format.lower()),
110            sample_rate=self._pcm_sample_rate,
111            bit_depth=get_supported_pcm_formats()[self._pcm_audio_format],
112            channels=self._audio_channels,
113        )
114        self._audio_source = AudioSource(
115            item_id=AUDIO_SOURCE_ID,
116            provider=self.instance_id,
117            name=f"{self.manifest.name}: {self._vban_stream_name}",
118            provider_mappings={
119                ProviderMapping(
120                    item_id=AUDIO_SOURCE_ID,
121                    provider_domain=self.domain,
122                    provider_instance=self.instance_id,
123                    audio_format=self._audio_format,
124                )
125            },
126            can_play_pause=False,
127            can_seek=False,
128            can_next_previous=False,
129            exclusive=True,
130            allow_external_trigger=False,
131            # MA opens the UDP listener on demand; get_audio_stream raises if
132            # the configured sender never sends
133            can_initiate=True,
134        )
135
136    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
137        """Return runtime options for this provider."""
138        return (
139            CONF_ENTRY_WARN_PREVIEW,
140            ConfigEntry(
141                key=CONF_VBAN_QUEUE_STRATEGY,
142                type=ConfigEntryType.STRING,
143                default_value=next(iter(VBAN_QUEUE_STRATEGIES)),
144                options=[ConfigValueOption(x, title=x) for x in VBAN_QUEUE_STRATEGIES],
145                advanced=True,
146                required=True,
147            ),
148            ConfigEntry(
149                key=CONF_VBAN_QUEUE_SIZE,
150                type=ConfigEntryType.INTEGER,
151                default_value=AsyncVBANClientMod.default_queue_size,
152                advanced=True,
153                required=True,
154            ),
155            ConfigEntry(
156                key=CONF_LOG_VBAN_STREAM_STATS,
157                type=ConfigEntryType.BOOLEAN,
158                default_value=False,
159                advanced=True,
160                required=True,
161            ),
162        )
163
164    @property
165    def instance_name_postfix(self) -> str | None:
166        """Return a (default) instance name postfix for this provider instance."""
167        return self._vban_stream_name
168
169    async def handle_async_init(self) -> None:
170        """Handle async initialization of the provider."""
171        # Set-up aiovban logging - DEBUG level is noisy
172        if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
173            logging.getLogger("aiovban").setLevel(logging.DEBUG)
174        else:
175            logging.getLogger("aiovban").setLevel(logging.INFO)
176
177        if self._log_stats and (
178            self.logger.isEnabledFor(logging.DEBUG) or self.logger.isEnabledFor(VERBOSE_LOG_LEVEL)
179        ):
180            self._stats_reporter = VBANStatsReporter(
181                pcm_sample_size=self._audio_format.pcm_sample_size
182            )
183
184        self._vban_receiver = AsyncVBANClientMod(default_queue_size=self._vban_queue_size)
185        try:
186            self._vban_stream = (
187                await self._vban_receiver.register_device(self._sender_host)
188            ).receive_stream(
189                self._vban_stream_name, back_pressure_strategy=self._vban_queue_strategy
190            )
191
192            self._udp_socket_fut = await self._vban_receiver.listen(
193                address=self._bind_ip, port=self._bind_port, controller=self
194            )
195        except (OSError, ValueError) as err:
196            raise SetupFailedError(f"Failed to start VBAN receiver plugin: {err}") from err
197
198    async def unload(self, is_removed: bool = False) -> None:
199        """Handle close/cleanup of the provider."""
200        self.logger.debug("Unloading plugin")
201
202        self._cancel_stats_reporter()
203
204        if self._in_use_by_player:
205            # Allow the running stream to stop cleanly
206            self._in_use_by_player = None
207            await asyncio.sleep(1)
208
209        if self._vban_receiver:
210            self.logger.debug("Closing UDP transport")
211            self._vban_receiver.close()
212            if self._udp_socket_fut:
213                try:
214                    await self._udp_socket_fut
215                except Exception as err:
216                    self.logger.debug("Error while closing UDP transport: %s", err)
217                else:
218                    self.logger.debug("Closed UDP transport")
219            self._vban_receiver = None
220
221        self._vban_stream = None
222
223    async def get_audio_sources(self) -> list[AudioSource]:
224        """Return the single AudioSource this VBAN receiver exposes."""
225        return [self._audio_source]
226
227    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
228        """
229        Return StreamDetails for streaming the VBAN PCM audio to a queue.
230
231        Side-effect-free: ownership is claimed in on_source_selected (which the
232        streams controller fires before this method on the actual stream
233        request). Keeping this idempotent means preload paths like
234        player_queues._load_item can fetch streamdetails without claiming the
235        source and blocking a subsequent cross-queue handoff.
236        """
237        if item_id != AUDIO_SOURCE_ID:
238            raise MediaNotFoundError(f"Unknown AudioSource: {item_id}")
239        return StreamDetails(
240            provider=self.instance_id,
241            item_id=item_id,
242            audio_format=self._audio_format,
243            media_type=MediaType.AUDIO_SOURCE,
244            stream_type=StreamType.CUSTOM,
245            stream_metadata=StreamMetadata(
246                title=self._vban_stream_name,
247                artist=self._sender_host,
248            ),
249        )
250
251    async def get_audio_stream(  # noqa: PLR0915
252        self, streamdetails: StreamDetails, seek_position: int = 0
253    ) -> AsyncGenerator[bytes]:
254        """Yield raw PCM chunks from the VBANIncomingStream queue."""
255        assert self._vban_stream  # for type checking
256        assert self._udp_socket_fut  # for type checking
257        _stream_id = str(uuid4())
258        self._active_stream_id = _stream_id
259        consumer_queue = self._in_use_by_player
260        # Snapshot the active session id so a same-queue reconnect (which
261        # refreshes _active_session_id but not _in_use_by_player) supersedes
262        # this stream: the loop exits and the finally release skips so it
263        # doesn't clobber the new session's claim.
264        captured_session_id = self._active_session_id
265        _stream_details = (
266            f"ID: {_stream_id}//Queue: {consumer_queue}//"
267            f"Stream: {self._vban_stream_name}//"
268            f"Config: {self._audio_format.output_format_str}"
269        )
270        _stream_acquired = False
271
272        # Drain any leftovers in the queue from previous use
273        while self._vban_stream.get_packet_nowait():
274            pass
275
276        if self._stats_reporter:
277            self._stats_reporter.start(stream_id=_stream_id, stream_details=_stream_details)
278
279        self.logger.debug("Ready to receive VBAN PCM audio stream: %s", _stream_details)
280
281        try:
282            while True:
283                if self._in_use_by_player != consumer_queue:
284                    self.logger.debug(
285                        "Stopping VBAN PCM audio stream receiver: %s - Reason: plugin is no "
286                        "longer in use by queue %s",
287                        _stream_details,
288                        consumer_queue,
289                    )
290                    break
291                if self._active_session_id != captured_session_id:
292                    self.logger.debug(
293                        "Stopping VBAN PCM audio stream receiver: %s - Reason: same-queue "
294                        "reconnect superseded this session",
295                        _stream_details,
296                    )
297                    break
298                if self._active_stream_id != _stream_id:
299                    self.logger.debug(
300                        "Stopping VBAN PCM audio stream receiver: %s - Reason: stream_id has "
301                        "changed from %s to %s meaning %s is a stale stream reader which was "
302                        "not cleanly closed",
303                        _stream_details,
304                        _stream_id,
305                        self._active_stream_id,
306                        _stream_id,
307                    )
308                    break
309                if self._udp_socket_fut.done():
310                    self.logger.debug(
311                        "Stopping VBAN PCM audio stream receiver: %s - Reason: UDP socket closed",
312                        _stream_details,
313                    )
314                    break
315
316                try:
317                    async with asyncio.timeout(1):
318                        packet = await self._vban_stream.get_packet()
319                        # Check if the stream_id has changed underneath us while waiting
320                        if self._active_stream_id != _stream_id:
321                            break
322                    if not _stream_acquired:
323                        _stream_acquired = True
324                        self.logger.debug("Acquired VBAN PCM audio stream: %s", _stream_details)
325                    if self._stats_reporter:
326                        self._stats_reporter.update(
327                            instance_id=_stream_id, vban_bytes_len=len(packet.body.data)
328                        )
329                    yield packet.body.data
330                except TimeoutError:
331                    # cold-start: fail fast if the configured sender never sends
332                    if not _stream_acquired:
333                        raise AudioError(
334                            f"VBAN sender {self._sender_host!r} did not send any packets "
335                            f"on stream {self._vban_stream_name!r}",
336                            translation_key="no_packets",
337                            translation_owner=self.translation_owner,
338                            translation_args=[self._sender_host, self._vban_stream_name],
339                        ) from None
340                    continue
341                except asyncio.QueueShutDown:
342                    self.logger.error(
343                        "Found VBANIncomingStream queue shut down when attempting to get VBAN "
344                        "packet for audio stream: %s",
345                        _stream_details,
346                    )
347                    break
348        finally:
349            self._cancel_stats_reporter(_stream_id)
350            # Guard release on BOTH queue id AND session id so a stale generator
351            # teardown after a same-queue reconnect doesn't clear the new
352            # session's claim.
353            if (
354                self._in_use_by_player == consumer_queue
355                and self._active_session_id == captured_session_id
356            ):
357                self._in_use_by_player = None
358            self.logger.debug("Stopped VBAN PCM audio stream receiver: %s", _stream_details)
359
360    async def on_source_selected(
361        self, source_id: str, player_id: str, owner_player_id: str, stream_session_id: str
362    ) -> None:
363        """Claim the source for this queue and let any prior stream wind down."""
364        if source_id != AUDIO_SOURCE_ID:
365            return
366        # Claim ownership for this queue. The lock lives here (not in
367        # get_stream_details) so preload paths can fetch streamdetails without
368        # accidentally blocking a subsequent cross-queue handoff at the actual
369        # stream request. There is no cmd_stop on a previous player like the
370        # Spotify Connect / AirPlay / Yandex receivers do: VBAN is a purely
371        # passive UDP receiver with no concept of an "active player" — the
372        # previous queue's get_audio_stream loop notices the queue change on
373        # its 1s timeout and exits cleanly on its own.
374        self._in_use_by_player = owner_player_id
375        # Record this request's session id so a later on_source_unselected can
376        # tell whether it is the live teardown or a stale callback from a
377        # superseded same-queue request.
378        self._active_session_id = stream_session_id
379
380    async def on_source_unselected(
381        self, source_id: str, owner_player_id: str, stream_session_id: str
382    ) -> None:
383        """Release the queue-scoped exclusive claim when MA tears down the stream."""
384        if source_id != AUDIO_SOURCE_ID:
385            return
386        # Reject stale callbacks: only release if this is still the active
387        # session. A owner_player_id check alone is not sufficient — same-queue
388        # reconnects (player drops + reopens the same stream URL before the
389        # original request's finally fires) would otherwise let the old
390        # request's late callback clear the live claim of the new stream.
391        if self._active_session_id != stream_session_id:
392            return
393        self._active_session_id = None
394        if self._in_use_by_player == owner_player_id:
395            self._in_use_by_player = None
396
397    def _cancel_stats_reporter(self, instance_id: str | None = None) -> None:
398        """Cancel a running stats reporter."""
399        if self._stats_reporter:
400            self.logger.debug("Cancelling stats reporter")
401            self._stats_reporter.cancel(instance_id)
402