music-assistant-server

1"""
2Controller to stream audio to players.
3
4The streams controller hosts a basic, unprotected HTTP-only webserver
5purely to stream audio packets to players.
6"""
7
8from __future__ import annotations
9
10import asyncio
11import logging
12import os
13import struct
14import time
15from collections.abc import AsyncGenerator
16from contextlib import aclosing
17from math import ceil
18from typing import TYPE_CHECKING, cast
19from uuid import uuid4
20
21from aiofiles.os import wrap
22from aiohttp import web
23from music_assistant_models.audio_processing import AudioQueueProcessing
24from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
25from music_assistant_models.enums import (
26    ConfigEntryType,
27    ContentType,
28    CrossfadeMode,
29    MediaType,
30    PlayerFeature,
31    ProviderType,
32    VolumeNormalizationMode,
33)
34from music_assistant_models.errors import (
35    AudioError,
36    InvalidDataError,
37    MediaNotFoundError,
38    ProviderUnavailableError,
39)
40from music_assistant_models.helpers import get_global_cache_value
41from music_assistant_models.media_items import AudioFormat
42
43from music_assistant.constants import (
44    CONF_BACKGROUND_SCAN_CONCURRENCY,
45    CONF_BIND_IP,
46    CONF_BIND_PORT,
47    CONF_CROSSFADE_DURATION,
48    CONF_CROSSFADE_MODE,
49    CONF_ENTRY_ENABLE_ICY_METADATA,
50    CONF_ENTRY_LOG_LEVEL,
51    CONF_ENTRY_VOLUME_NORMALIZATION_TARGET,
52    CONF_HTTP_PROFILE,
53    CONF_OUTPUT_CODEC,
54    CONF_PLAYER_QUEUES,
55    CONF_PUBLISH_IP,
56    CONF_VALUE_AUTO,
57    CONF_VOLUME_NORMALIZATION_FIXED_GAIN_RADIO,
58    CONF_VOLUME_NORMALIZATION_FIXED_GAIN_TRACKS,
59    CONF_VOLUME_NORMALIZATION_RADIO,
60    CONF_VOLUME_NORMALIZATION_TRACKS,
61    DEFAULT_BACKGROUND_SCAN_CONCURRENCY,
62    DEFAULT_HOST,
63    DEFAULT_STREAM_HEADERS,
64    DLNA_CONTENT_FEATURES,
65    DLNA_CONTENT_FEATURES_REALTIME,
66    ICY_HEADERS,
67    SILENCE_FILE,
68    VERBOSE_LOG_LEVEL,
69    WILDCARD_BIND_IPS,
70)
71from music_assistant.controllers.players.helpers import AnnounceData
72from music_assistant.controllers.streams.announcements import (
73    DEFAULT_RENDER_TIMEOUT,
74    AnnouncementRenderer,
75)
76from music_assistant.controllers.streams.audio import StreamsAudio, overlay_active
77from music_assistant.controllers.streams.audio_analysis import AudioAnalysisController
78from music_assistant.controllers.streams.audio_processing import (
79    AudioProcessingManager,
80)
81from music_assistant.controllers.streams.constants import (
82    CONF_ALLOW_CROSSFADE_SAME_ALBUM,
83    CONF_BUFFER_SIZE,
84    CONF_BUFFER_SIZE_DEFAULT,
85    CONF_SMART_FADES_LOG_LEVEL,
86    DEFAULT_PORT,
87    FLOW_STREAM_LEAD_OUT_SECONDS,
88    BufferSize,
89    get_available_buffer_sizes,
90)
91from music_assistant.helpers.audio import (
92    calculate_content_length,
93    get_content_length,
94    get_mime_type,
95    store_content_length_in_cache,
96)
97from music_assistant.helpers.buffered_generator import buffered
98from music_assistant.helpers.ffmpeg import (
99    CACHE_ATTR_FFMPEG_VERSION,
100    CACHE_ATTR_LIBSOXR_PRESENT,
101    check_ffmpeg_version,
102    get_ffmpeg_stream,
103)
104from music_assistant.helpers.ffmpeg import LOGGER as FFMPEG_LOGGER
105from music_assistant.helpers.util import (
106    format_ip_for_url,
107    get_ip_addresses,
108    get_publish_ip_candidates,
109    get_source_ip_for_target,
110    sanitize_http_header_value,
111)
112from music_assistant.helpers.webserver import Webserver, redact_sensitive_headers
113from music_assistant.models.core_controller import CoreController
114from music_assistant.models.music_provider import MusicProvider
115from music_assistant.models.plugin import PluginProvider
116from music_assistant.providers.universal_group.constants import UGP_PREFIX
117from music_assistant.providers.universal_group.player import UniversalGroupPlayer
118
119if TYPE_CHECKING:
120    from music_assistant_models.config_entries import CoreConfig
121    from music_assistant_models.player import PlayerMedia
122    from music_assistant_models.player_queue import PlayerQueue
123    from music_assistant_models.queue_item import QueueItem
124    from music_assistant_models.streamdetails import StreamMetadata
125
126    from music_assistant.helpers.json import SerializableType
127    from music_assistant.mass import MusicAssistant
128
129
130isfile = wrap(os.path.isfile)
131
132
133def _streaming_wav_header(output_format: AudioFormat) -> bytes:
134    """Build a WAV header with open-ended (0xFFFFFFFF) RIFF/data sizes for live streams."""
135    channels = output_format.channels
136    sample_rate = output_format.sample_rate
137    bits_per_sample = output_format.bit_depth
138    byte_rate = sample_rate * channels * (bits_per_sample // 8)
139    block_align = channels * (bits_per_sample // 8)
140    # RIFF size & data size both set to 0xFFFFFFFF so clients honoring the WAV
141    # length fields don't cut the stream off (default header hardcodes ~6.7h).
142    return (
143        b"RIFF"
144        + struct.pack("<L", 0xFFFFFFFF)
145        + b"WAVE"
146        + b"fmt "
147        + struct.pack(
148            "<LHHLLHH", 16, 1, channels, sample_rate, byte_rate, block_align, bits_per_sample
149        )
150        + b"data"
151        + struct.pack("<L", 0xFFFFFFFF)
152    )
153
154
155async def _wav_passthrough_stream(
156    audio_input: AsyncGenerator[bytes], output_format: AudioFormat
157) -> AsyncGenerator[bytes]:
158    """Yield a WAV header followed by raw PCM bytes from ``audio_input``."""
159    yield _streaming_wav_header(output_format)
160    async for chunk in audio_input:
161        yield chunk
162
163
164def _get_publish_addresses(
165    bind_ip: str, configured_publish_ip: str | None, publish_candidates: tuple[str, ...]
166) -> list[str]:
167    """
168    Return the addresses this host publishes on, best candidate first.
169
170    :param bind_ip: The configured bind IP (a wildcard means all interfaces).
171    :param configured_publish_ip: The explicitly configured publish IP, or None when auto.
172    :param publish_candidates: Host addresses reachable from the local network, ranked.
173    """
174    if configured_publish_ip:
175        # an explicitly configured address is the authoritative answer
176        return [configured_publish_ip]
177    if bind_ip and bind_ip not in WILDCARD_BIND_IPS:
178        # only one interface is served, so no other address can be reached
179        return [bind_ip]
180    # auto-detected: keep the whole ranked list - publish_ip takes the best of them and
181    # the network fingerprint watches all of them to spot an interface change
182    return list(publish_candidates)
183
184
185class StreamsController(CoreController):
186    """Controller to stream audio to players."""
187
188    domain: str = "streams"
189
190    def __init__(self, mass: MusicAssistant) -> None:
191        """Initialize instance."""
192        super().__init__(mass)
193        self._server = Webserver(self.logger, enable_dynamic_routes=True)
194        self.register_dynamic_route = self._server.register_dynamic_route
195        self.unregister_dynamic_route = self._server.unregister_dynamic_route
196        self.manifest.name = "Streamserver"
197        self.manifest.description = (
198            "Music Assistant's core controller that is responsible for "
199            "streaming audio to players on the local network."
200        )
201        self.manifest.icon = "cast-audio"
202        self.announcement_renderer = AnnouncementRenderer()
203        self._bind_ip: str = "0.0.0.0"
204        self._base_url: str = ""
205        self._configured_publish_ip: str | None = None
206        # every address players may reach this host on, best candidate first; publish_ip is
207        # the first of them and the network fingerprint watches the whole list for changes
208        self._publish_addresses: list[str] = []
209        # the network as it was at the previous setup, to spot a runtime change
210        self._network_fingerprint: tuple[str, str, int, tuple[str, ...]] | None = None
211        self.audio = StreamsAudio(mass)
212        self.audio_processing = AudioProcessingManager(mass)
213        self._audio_analysis = AudioAnalysisController(self)
214        # Number of queue streams (single item or flow) actively serving a player right now.
215        # Audio analysis reads this (via audio_analysis.playback_active) to yield CPU while a
216        # queue stream is live. Announcements are a separate path that never runs analysis.
217        self._active_output_streams = 0
218
219    @property
220    def audio_analysis(self) -> AudioAnalysisController:
221        """Return the AudioAnalysisController instance."""
222        return self._audio_analysis
223
224    def output_stream_active(self) -> bool:
225        """Return whether a queue stream (single item or flow) is actively serving a player."""
226        return self._active_output_streams > 0
227
228    async def get_diagnostics(self) -> dict[str, SerializableType]:
229        """Return diagnostics info for this controller to include in diagnostics reports."""
230        return {
231            "ffmpeg_version": get_global_cache_value(CACHE_ATTR_FFMPEG_VERSION),
232            "libsoxr_support": get_global_cache_value(CACHE_ATTR_LIBSOXR_PRESENT),
233            "active_output_streams": self._active_output_streams,
234            "active_announcements": self.announcement_renderer.active_announcements,
235            "active_announcement_renders": self.announcement_renderer.active_renders,
236            "publish_ip_configured": self._configured_publish_ip is not None,
237        }
238
239    @property
240    def base_url(self) -> str:
241        """Return the base_url for the streamserver."""
242        return self._base_url
243
244    @property
245    def bind_ip(self) -> str:
246        """Return the IP address this streamserver is bound to."""
247        return self._bind_ip
248
249    async def get_source_ip(self, target_ip: str | None = None) -> str | None:
250        """
251        Return a local, bindable source IP on the player-facing network.
252
253        For callers that bind a socket or hand a local interface address to a helper
254        process, so their traffic leaves on the network the players live on. The result
255        is always an address of this host, never the advertised address, which may not
256        exist here at all.
257
258        Returns None when no single interface should be pinned, which the caller must
259        read as "bind all interfaces and let the routing table decide".
260
261        :param target_ip: IP address of the device the traffic is meant for. Omit it for
262            a shared consumer that serves every player at once; such a caller can only be
263            pinned by an explicitly configured bind IP.
264        """
265        if self._bind_ip and self._bind_ip not in WILDCARD_BIND_IPS:
266            if target_ip and not _same_ip_family(self._bind_ip, target_ip):
267                return None
268            return self._bind_ip
269        if not target_ip:
270            return None
271        return await get_source_ip_for_target(target_ip) or None
272
273    def get_publish_ip(self, target_ip: str) -> str | None:
274        """
275        Return the address to advertise to the device at ``target_ip``, if one is configured.
276
277        Only an explicitly configured publish IP is returned. An auto-detected one is a
278        guess at this host's primary interface, which on a multi-homed host is not
279        necessarily the network the players live on, so callers that can derive the
280        address from the connection itself must prefer that over the guess.
281
282        Returns None when no publish IP was configured, or when the configured one cannot
283        apply to this device.
284
285        :param target_ip: IP address of the device that would receive the address, used to
286            reject an address of the wrong IP family.
287        """
288        if not self._configured_publish_ip:
289            return None
290        if not _same_ip_family(self._configured_publish_ip, target_ip):
291            return None
292        return self._configured_publish_ip
293
294    @property
295    def smart_fades_available(self) -> bool:
296        """
297        Return whether smart crossfade can be used on this server.
298
299        Requires a large-enough audio buffer (at least balanced) and a loaded
300        smart fades audio analysis provider.
301        """
302        buffer_size = BufferSize(
303            self.mass.config.get_raw_core_config_value(
304                self.domain, CONF_BUFFER_SIZE, CONF_BUFFER_SIZE_DEFAULT
305            )
306        )
307        return (
308            buffer_size != BufferSize.MINIMAL and self.audio_analysis.smart_fades_provider_available
309        )
310
311    def get_crossfade_mode(self, queue: PlayerQueue) -> CrossfadeMode:
312        """
313        Return the effective crossfade mode for a queue.
314
315        Combines the per-play on/off toggle with the crossfade_mode setting and smart fades
316        availability: smart when enabled, selected and available; standard when enabled but smart
317        is not selected/available; disabled otherwise.
318        """
319        if not queue.crossfade_enabled:
320            return CrossfadeMode.DISABLED
321        # default to smart when this server can use it, else standard
322        default_mode = (
323            CrossfadeMode.SMART_CROSSFADE
324            if self.smart_fades_available
325            else CrossfadeMode.STANDARD_CROSSFADE
326        )
327        mode = self.mass.config.get_effective_player_queue_config_value(
328            queue.queue_id, CONF_CROSSFADE_MODE, default_mode
329        )
330        if mode == CrossfadeMode.SMART_CROSSFADE and self.smart_fades_available:
331            return CrossfadeMode.SMART_CROSSFADE
332        return CrossfadeMode.STANDARD_CROSSFADE
333
334    def is_smart_fades_active(self, queue: PlayerQueue) -> bool:
335        """Return whether the queue's effective crossfade mode is smart crossfade."""
336        return self.get_crossfade_mode(queue) == CrossfadeMode.SMART_CROSSFADE
337
338    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
339        """Return all Config Entries for this core module (if any)."""
340        ip_addresses = await get_ip_addresses(include_ipv6=True)
341        return (
342            ConfigEntry(
343                key=CONF_BUFFER_SIZE,
344                type=ConfigEntryType.STRING,
345                default_value=CONF_BUFFER_SIZE_DEFAULT,
346                # Only offer presets the host's RAM can sustain (Balanced >= 4GB,
347                # Maximum >= 7GB); see get_available_buffer_sizes.
348                options=[ConfigValueOption(size.value) for size in get_available_buffer_sizes()],
349                required=False,
350                category="playback",
351            ),
352            ConfigEntry(
353                key=CONF_VOLUME_NORMALIZATION_RADIO,
354                type=ConfigEntryType.STRING,
355                default_value=VolumeNormalizationMode.FALLBACK_DYNAMIC,
356                options=[
357                    ConfigValueOption(x.value, title=x.value.replace("_", " ").title())
358                    for x in VolumeNormalizationMode
359                ],
360                category="playback",
361            ),
362            ConfigEntry(
363                key=CONF_VOLUME_NORMALIZATION_TRACKS,
364                type=ConfigEntryType.STRING,
365                default_value=VolumeNormalizationMode.FALLBACK_DYNAMIC,
366                options=[
367                    ConfigValueOption(x.value, title=x.value.replace("_", " ").title())
368                    for x in VolumeNormalizationMode
369                ],
370                category="playback",
371            ),
372            ConfigEntry(
373                key=CONF_VOLUME_NORMALIZATION_FIXED_GAIN_RADIO,
374                type=ConfigEntryType.FLOAT,
375                range=(-20, 10),
376                default_value=-6,
377                category="playback",
378            ),
379            ConfigEntry(
380                key=CONF_VOLUME_NORMALIZATION_FIXED_GAIN_TRACKS,
381                type=ConfigEntryType.FLOAT,
382                range=(-20, 10),
383                default_value=-6,
384                category="playback",
385            ),
386            CONF_ENTRY_VOLUME_NORMALIZATION_TARGET,
387            ConfigEntry(
388                key=CONF_ALLOW_CROSSFADE_SAME_ALBUM,
389                type=ConfigEntryType.BOOLEAN,
390                default_value=False,
391                category="playback",
392            ),
393            ConfigEntry(
394                key=CONF_PUBLISH_IP,
395                type=ConfigEntryType.STRING,
396                default_value=CONF_VALUE_AUTO,
397                required=False,
398                category="generic",
399                advanced=True,
400                requires_reload=True,
401            ),
402            ConfigEntry(
403                key=CONF_BIND_PORT,
404                type=ConfigEntryType.INTEGER,
405                default_value=DEFAULT_PORT,
406                category="generic",
407                advanced=True,
408                requires_reload=True,
409            ),
410            ConfigEntry(
411                key=CONF_BIND_IP,
412                type=ConfigEntryType.STRING,
413                default_value=DEFAULT_HOST,
414                options=[ConfigValueOption(x, title=x) for x in {DEFAULT_HOST, *ip_addresses}],
415                category="generic",
416                advanced=True,
417                required=False,
418                requires_reload=True,
419            ),
420            ConfigEntry(
421                key=CONF_SMART_FADES_LOG_LEVEL,
422                type=ConfigEntryType.STRING,
423                options=CONF_ENTRY_LOG_LEVEL.options,
424                default_value="GLOBAL",
425                category="audio_analysis",
426                advanced=True,
427            ),
428            ConfigEntry(
429                key=CONF_BACKGROUND_SCAN_CONCURRENCY,
430                type=ConfigEntryType.INTEGER,
431                range=(1, 16),
432                default_value=DEFAULT_BACKGROUND_SCAN_CONCURRENCY,
433                category="audio_analysis",
434            ),
435        )
436
437    async def setup(self, config: CoreConfig) -> None:
438        """Async initialize of module."""
439        # initialize the audio sub-controller (needs mass.streams to be set)
440        self.audio.setup()
441        self._audio_analysis.setup()
442        # copy log level to audio/ffmpeg loggers
443        self.audio.logger.setLevel(self.logger.level)
444        FFMPEG_LOGGER.setLevel(self.logger.level)
445        self._setup_smart_fades_logger(config)
446        # perform check for ffmpeg version
447        await check_ffmpeg_version()
448        # start the webserver
449        self.publish_port = config.get_value(CONF_BIND_PORT, DEFAULT_PORT)
450        configured_publish_ip = str(config.get_value(CONF_PUBLISH_IP) or CONF_VALUE_AUTO)
451        self._configured_publish_ip = (
452            None if configured_publish_ip == CONF_VALUE_AUTO else configured_publish_ip
453        )
454        publish_candidates = await get_publish_ip_candidates(include_ipv6=True)
455        bind_ip = str(config.get_value(CONF_BIND_IP))
456        self._resolve_publish_state(bind_ip, publish_candidates)
457        await self._server.setup(
458            bind_ip=bind_ip,
459            bind_port=cast("int", self.publish_port),
460            static_routes=[
461                (
462                    "*",
463                    "/flow/{session_id}/{queue_id}/{queue_item_id}/{player_id}.{fmt}",
464                    self.serve_queue_flow_stream,
465                ),
466                (
467                    "*",
468                    "/single/{session_id}/{queue_id}/{queue_item_id}/{player_id}.{fmt}",
469                    self.serve_queue_item_stream,
470                ),
471                ("*", "/command/{queue_id}/{command}.mp3", self.serve_command_request),
472                ("*", "/announcement/{player_id}.{fmt}", self.serve_announcement_stream),
473            ],
474        )
475        # adopt what the server actually bound to: a configured port of 0 is only resolved
476        # by the OS at bind time and an unavailable bind IP falls back to all interfaces
477        self.publish_port = cast("int", self._server.port)
478        self._resolve_publish_state(self._server.bind_ip or DEFAULT_HOST, publish_candidates)
479        # print a big fat message in the log where the streamserver is running
480        # because this is a common source of issues for people with more complex setups
481        self.logger.log(
482            logging.INFO if self.mass.config.onboard_done else logging.WARNING,
483            "\n\n################################################################################\n"
484            "Started streamserver on %s:%s\n"
485            "This is the IP address that is communicated to players.\n"
486            "If this is incorrect, audio will not play!\n"
487            "See the documentation for how to configure the publish IP for the Streamserver\n"
488            "in Settings --> System --> Streams\n"
489            "################################################################################\n",
490            self.publish_ip,
491            self.publish_port,
492        )
493        await self._reload_network_dependent_providers()
494
495    async def close(self) -> None:
496        """Cleanup on exit."""
497        await self._audio_analysis.close()
498        await self._server.close()
499
500    async def resolve_stream_url(self, player_id: str, media: PlayerMedia) -> str:
501        """
502        Resolve the stream URL for the given PlayerMedia.
503
504        :param player_id: The (protocol) player ID requesting the stream.
505        :param media: The PlayerMedia object for which to resolve the stream URL.
506        :return: The resolved stream URL as a string.
507        """
508        if media.media_type in (MediaType.ANNOUNCEMENT, MediaType.FLOW_STREAM):
509            return media.uri
510        protocol_player = self.mass.players.get_player(player_id)
511        # AudioSource is realtime: serve as WAV (PCM + header) so the encode
512        # step is a no-op passthrough — drops a whole ffmpeg from the
513        # consumer-side pipeline and the latency that comes with it.
514        if media.media_type == MediaType.AUDIO_SOURCE:
515            output_codec = ContentType.WAV
516        else:
517            conf_output_codec = cast(
518                "str",
519                protocol_player.config.get_value(CONF_OUTPUT_CODEC, default="flac")
520                if protocol_player
521                else "flac",
522            )
523            output_codec = ContentType.try_parse(conf_output_codec)
524        fmt = output_codec.value
525        # handle raw pcm without exact format specifiers
526        if output_codec.is_pcm() and ";" not in fmt:
527            fmt += f";codec=pcm;rate={44100};bitrate={16};channels={2}"
528        extra_data = media.custom_data or {}
529        session_id = extra_data.get("session_id")
530        queue_item_id = media.queue_item_id
531        if not session_id or not queue_item_id:
532            raise InvalidDataError("Can not resolve stream URL: Invalid PlayerMedia data")
533        queue_id = media.source_id
534        queue = self.mass.player_queues.get(queue_id) if queue_id else None
535        crossfade_needs_flow_mode = (
536            # crossfade only applies to tracks; if the queue has it enabled but the player(protocol)
537            # does not support gapless playback, we need to enforce flow mode
538            media.media_type == MediaType.TRACK
539            and queue is not None
540            and queue.crossfade_enabled
541            and protocol_player
542            and not protocol_player.supports_gapless
543        )
544        # the audio overlay is mixed into the queue's continuous (flow) stream;
545        # per-item requests would restart the overlay at every track boundary
546        overlay_needs_flow_mode = queue is not None and overlay_active(queue)
547        # Determine flow_mode based on the actual player's capabilities.
548        # This is done here (just-in-time) because the player's protocol determines this
549        flow_mode = (
550            protocol_player is not None
551            and (protocol_player.flow_mode or crossfade_needs_flow_mode or overlay_needs_flow_mode)
552            and media.media_type not in (MediaType.RADIO, MediaType.AUDIO_SOURCE)
553        )
554        base_path = "flow" if flow_mode else "single"
555        return (
556            f"{self.base_url}/{base_path}/{session_id}/{queue_id}/{queue_item_id}/{player_id}.{fmt}"
557        )
558
559    def update_stream_metadata(
560        self,
561        queue_id: str,
562        source_id: str,
563        provider: str,
564        stream_metadata: StreamMetadata,
565    ) -> None:
566        """
567        Push a live stream metadata update for an active AudioSource queue item.
568
569        Used by plugin providers exposing an AudioSource (e.g. AirPlay receiver,
570        Spotify Connect) to surface live track-change info without restarting
571        the stream. The radio ICY metadata path uses the same underlying field
572        via ``_update_radio_stream_metadata`` but goes through a separate path.
573
574        The update is rejected silently unless the queue's current item is an
575        AudioSource owned by ``provider`` with ``item_id == source_id``. This
576        guard prevents a late callback (e.g. the provider firing one more
577        metadata update after MA has already moved the queue on to a track or
578        a different AudioSource) from stamping unrelated metadata over the
579        wrong item.
580
581        :param queue_id: The queue whose active item should receive the update.
582        :param source_id: The AudioSource.item_id emitting this metadata.
583        :param provider: The provider instance id emitting this metadata.
584        :param stream_metadata: The new stream metadata to attach.
585        """
586        queue = self.mass.player_queues.get(queue_id)
587        if queue is None:
588            return
589        current_item = queue.current_item
590        if current_item is None or current_item.streamdetails is None:
591            return
592        sd = current_item.streamdetails
593        if (
594            sd.media_type != MediaType.AUDIO_SOURCE
595            or sd.provider != provider
596            or sd.item_id != source_id
597        ):
598            # Log at debug so misbehaving providers firing constantly are
599            # diagnosable (count alone is the signal) without spamming higher
600            # log levels for the legitimate transition cases.
601            self.logger.debug(
602                "Rejected metadata update for queue %s from provider %s source %s "
603                "(current item: %s)",
604                queue_id,
605                provider,
606                source_id,
607                sd.uri if sd else "none",
608            )
609            return
610        # Re-check identity *after* preparing the write so a queue advance that
611        # races with this callback can't slip in between the guard above and
612        # the mutation below. Plugins fire these from arbitrary executor /
613        # event-loop threads (AirPlay metadata reader, Spotify webservice
614        # handler, AriaCast WebSocket reader); the GIL keeps each attribute
615        # write atomic but not the read-then-write sequence.
616        if queue.current_item is not current_item or current_item.streamdetails is not sd:
617            return
618        sd.stream_metadata = stream_metadata
619        sd.stream_metadata_last_updated = time.time()
620        self.mass.player_queues.signal_update(queue_id)
621
622    async def serve_queue_item_stream(self, request: web.Request) -> web.StreamResponse:  # noqa: PLR0915
623        """Stream single queueitem audio to a player."""
624        self._log_request(request)
625        queue_id = request.match_info["queue_id"]
626        player_id = request.match_info["player_id"]
627        if not (queue := self.mass.player_queues.get(queue_id)):
628            raise web.HTTPNotFound(reason=f"Unknown Queue: {queue_id}")
629        session_id = request.match_info["session_id"]
630        pq_data = self.mass.player_queues.queue_data(queue.queue_id)
631        if pq_data.session_id is None or session_id != pq_data.session_id:
632            raise web.HTTPNotFound(reason=f"Unknown (or invalid) session: {session_id}")
633        if not (player := self.mass.players.get_player(player_id)):
634            raise web.HTTPNotFound(reason=f"Unknown Player: {player_id}")
635        queue_item_id = request.match_info["queue_item_id"]
636        queue_item = self.mass.player_queues.get_item(queue_id, queue_item_id)
637        if not queue_item:
638            raise web.HTTPNotFound(reason=f"Unknown Queue item: {queue_item_id}")
639
640        is_audio_source = (
641            queue_item.media_item is not None
642            and queue_item.media_item.media_type == MediaType.AUDIO_SOURCE
643        )
644
645        # HEAD probes for AudioSource items return a minimal response without
646        # touching the plugin. on_source_selected is the lifecycle hook that
647        # claims ownership and fires off transfer/handoff side effects (stop
648        # the previous player, redirect on disallowed switch, etc.), and a
649        # renderer probing with HEAD before GET should not trigger any of
650        # that. The actual GET request goes through the full hook chain.
651        if request.method != "GET" and is_audio_source:
652            # Validate the providing plugin still exists before advertising the
653            # source. Many DLNA renderers cache HEAD responses; returning 200
654            # for a URI whose plugin has been unloaded would lie to the
655            # renderer and the follow-up GET would fail unrecoverably.
656            assert queue_item.media_item is not None
657            if not isinstance(
658                self.mass.get_provider(queue_item.media_item.provider), PluginProvider
659            ):
660                raise web.HTTPNotFound(
661                    reason=f"AudioSource provider {queue_item.media_item.provider} unavailable"
662                )
663            # For PCM-fmt URLs, advertise audio/wav in HEAD: most DLNA renderers
664            # key off the HEAD Content-Type to pick a decoder and do not handle
665            # raw PCM (application/octet-stream). The actual GET response will
666            # still wrap the bytes into a WAV container if needed via the same
667            # mime-type translation downstream.
668            head_fmt = request.match_info["fmt"]
669            if ContentType.try_parse(head_fmt).is_pcm():
670                head_fmt = ContentType.WAV.value
671            headers = {
672                **DEFAULT_STREAM_HEADERS,
673                "icy-name": sanitize_http_header_value(queue_item.name),
674                "contentFeatures.dlna.org": DLNA_CONTENT_FEATURES_REALTIME,
675                "Content-Type": get_mime_type(head_fmt),
676            }
677            resp = web.StreamResponse(status=200, reason="OK", headers=headers)
678            await resp.prepare(request)
679            return resp
680
681        # Fire on_source_selected hook for every AudioSource GET — this is the
682        # single point where exclusive plugin sources claim ownership. Firing
683        # unconditionally (regardless of whether streamdetails are cached from
684        # a previous request) means a disconnect/reconnect for the same queue
685        # item re-claims the lock with a fresh session id, instead of streaming
686        # against the stale ownership of the prior request.
687        # Source identity comes from queue_item.media_item because streamdetails
688        # may not exist yet on the first request.
689        # stream_session_id is a fresh per-request token threaded through to
690        # on_source_unselected so the provider can distinguish a stale
691        # teardown (e.g. a same-queue reconnect's first request completing
692        # AFTER its replacement has already started streaming) from the
693        # currently active session's real teardown.
694        audio_source_provider: PluginProvider | None = None
695        audio_source_id: str | None = None
696        stream_session_id = uuid4().hex
697        if (
698            is_audio_source
699            and queue_item.media_item is not None
700            and (prov := self.mass.get_provider(queue_item.media_item.provider))
701            and isinstance(prov, PluginProvider)
702        ):
703            audio_source_id = queue_item.media_item.item_id
704            # Wire the provider into the finally block BEFORE awaiting the
705            # hook: if the provider partially mutates state (claims the lock,
706            # records the session id) and then raises a non-RuntimeError
707            # exception (buggy plugin, asyncio.CancelledError, etc.), the
708            # finally must still fire on_source_unselected so the lock gets
709            # released. The provider's session-id guard makes a spurious
710            # release a no-op if the lock was never actually claimed.
711            audio_source_provider = prov
712            try:
713                await prov.on_source_selected(
714                    audio_source_id, player_id, queue_id, stream_session_id
715                )
716            except RuntimeError as err:
717                # Provider intentionally aborts the original request (e.g.
718                # allow_player_switch=False has just redirected play_media to
719                # the configured target). Surface as 404 so the disallowed
720                # player drops the connection cleanly instead of treating an
721                # uncaught 500 as transient and retrying. The provider
722                # contract requires raising BEFORE claiming, so this is a
723                # clean abort — but we still let the finally run, where the
724                # session-id guard makes the unselect a no-op.
725                self.logger.info(
726                    "AudioSource %s aborted stream for player %s: %s",
727                    audio_source_id,
728                    player_id,
729                    err,
730                )
731                raise web.HTTPNotFound(reason=str(err))
732
733        try:
734            if not queue_item.streamdetails:
735                try:
736                    queue_item.streamdetails = await self.audio.get_stream_details(
737                        queue_item=queue_item
738                    )
739                except Exception as e:
740                    self.logger.error(
741                        "Failed to get streamdetails for QueueItem %s: %s", queue_item_id, e
742                    )
743                    queue_item.available = False
744                    raise web.HTTPNotFound(
745                        reason=f"No streamdetails for Queue item: {queue_item_id}"
746                    )
747
748            standard_crossfade_duration = self.mass.config.get_raw_core_config_value(
749                CONF_PLAYER_QUEUES, CONF_CROSSFADE_DURATION, 8
750            )
751            if queue_item.media_type != MediaType.TRACK:
752                crossfade_mode = CrossfadeMode.DISABLED
753            else:
754                crossfade_mode = self.get_crossfade_mode(queue)
755            if (
756                crossfade_mode != CrossfadeMode.DISABLED
757                and PlayerFeature.GAPLESS_PLAYBACK not in player.state.supported_features
758            ):
759                self.logger.warning(
760                    "Crossfade disabled: Player %s does not support gapless playback, "
761                    "consider enabling flow mode to enable crossfade on this player.",
762                    player.state.name,
763                )
764                crossfade_mode = CrossfadeMode.DISABLED
765
766            # pick output format based on the streamdetails and player capabilities
767            pcm_format = await self.audio.select_pcm_format(
768                player=player,
769                streamdetails=queue_item.streamdetails,
770                crossfade_enabled=crossfade_mode != CrossfadeMode.DISABLED,
771                overlay_active=(queue_item.media_type == MediaType.RADIO and overlay_active(queue)),
772            )
773            output_format = await self.audio.get_output_format(
774                output_format_str=request.match_info["fmt"],
775                player=player,
776                content_sample_rate=pcm_format.sample_rate,
777                content_bit_depth=pcm_format.bit_depth,
778                media_type=queue_item.media_type,
779            )
780
781            # prepare request, add some DLNA/UPNP compatible headers
782            # icy-name is sanitized (all control chars, not just newlines) to avoid a
783            # "Potential header injection attack" ValueError by aiohttp
784            # see https://github.com/music-assistant/support/issues/4913
785            # and https://github.com/music-assistant/support/issues/5791
786            # use realtime DLNA flags for radio (sender-paced) since the source delivers slowly
787            dlna_features = (
788                DLNA_CONTENT_FEATURES_REALTIME
789                if queue_item.media_type != MediaType.TRACK
790                else DLNA_CONTENT_FEATURES
791            )
792            headers = {
793                **DEFAULT_STREAM_HEADERS,
794                "icy-name": sanitize_http_header_value(queue_item.name),
795                "contentFeatures.dlna.org": dlna_features,
796                "Content-Type": get_mime_type(output_format.output_format_str),
797            }
798
799            resp = web.StreamResponse(status=200, reason="OK", headers=headers)
800            resp.content_type = get_mime_type(output_format.output_format_str)
801            http_profile = player.get_config_value(CONF_HTTP_PROFILE, "default")
802            if http_profile == "forced_content_length" and not queue_item.duration:
803                # just set an insane high content length to make sure the player keeps playing
804                resp.content_length = calculate_content_length(output_format, 12 * 3600)
805            elif http_profile == "forced_content_length" and queue_item.duration:
806                # estimate content length based on effective duration
807                # account for seek position (e.g., crossfade from previous track)
808                seek_pos = queue_item.streamdetails.seek_position if queue_item.streamdetails else 0
809                effective_duration = max(queue_item.duration - seek_pos, 1)
810                # use cached actual bytes-per-second if available (from a previous stream)
811                resp.content_length = await get_content_length(
812                    self.mass, queue_item.uri, output_format, effective_duration
813                )
814            elif http_profile == "chunked":
815                resp.enable_chunked_encoding()
816
817            await resp.prepare(request)
818
819            # return early if this is not a GET request
820            if request.method != "GET":
821                return resp
822
823            self._update_audio_processing_context(
824                queue=queue,
825                queue_item=queue_item,
826                pcm_format=pcm_format,
827                crossfade_mode=crossfade_mode,
828                overlay_enabled=(
829                    queue_item.media_type == MediaType.RADIO and overlay_active(queue)
830                ),
831                session_id=session_id,
832            )
833
834            if crossfade_mode != CrossfadeMode.DISABLED:
835                # crossfade is enabled, use special crossfaded single item stream
836                # where the crossfade of the next track is present in the stream of
837                # a single track. This only works if the player supports gapless playback!
838                audio_input = self.audio.get_queue_item_stream_with_smartfade(
839                    player=player,
840                    queue_item=queue_item,
841                    pcm_format=pcm_format,
842                    crossfade_mode=crossfade_mode,
843                    standard_crossfade_duration=standard_crossfade_duration,
844                    session_id=session_id,
845                )
846            else:
847                # no crossfade, just a regular single item stream
848                audio_input = self.audio.get_queue_item_stream(
849                    queue_item=queue_item,
850                    pcm_format=pcm_format,
851                    seek_position=int(queue_item.streamdetails.seek_position),
852                    playback_speed=cast(
853                        "float", queue_item.extra_attributes.get("playback_speed", 1.0)
854                    ),
855                    session_id=session_id,
856                )
857            if queue_item.media_type == MediaType.RADIO and overlay_active(queue):
858                # radio plays as a single long-lived stream (never in flow mode),
859                # so mix the audio overlay in here
860                audio_input = self.audio.get_overlay_mixed_stream(queue, audio_input, pcm_format)
861            # stream the audio
862            # this final ffmpeg process in the chain converts raw lossless PCM into
863            # the desired output format for the player including any player specific
864            # filter params such as channels mixing, DSP, resampling and, only if
865            # needed, encoding to lossy formats
866            output_plan = self.audio.get_player_output_plan(
867                player_id=player.player_id,
868                input_format=pcm_format,
869                output_format=output_format,
870                shared_player_ids=player.state.group_members,
871                queue_id=queue_id,
872                session_id=session_id,
873                queue_item_id=queue_item.queue_item_id,
874            )
875            filter_params = output_plan.filter_params
876            # Fast path for live AudioSource: when the player accepts WAV at the
877            # source's exact PCM rate/depth/channels and no filters apply, we
878            # skip the encode ffmpeg entirely and just stream a WAV header
879            # followed by the raw PCM bytes — saves an ffmpeg process and the
880            # latency of its internal buffer on every realtime stream.
881            audio_bytes: AsyncGenerator[bytes]
882            if (
883                queue_item.media_type == MediaType.AUDIO_SOURCE
884                and output_format.content_type == ContentType.WAV
885                and not filter_params
886                and output_format.sample_rate == pcm_format.sample_rate
887                and output_format.bit_depth == pcm_format.bit_depth
888                and output_format.channels == pcm_format.channels
889            ):
890                audio_bytes = _wav_passthrough_stream(audio_input, output_format)
891            else:
892                audio_bytes = get_ffmpeg_stream(
893                    audio_input=audio_input,
894                    input_format=pcm_format,
895                    output_format=output_format,
896                    filter_params=filter_params,
897                )
898            first_chunk_received = False
899            bytes_sent = 0
900            # Mark this player as actively streaming so audio analysis yields CPU to playback
901            # for the duration of the transfer (see audio_analysis.playback_active).
902            self._active_output_streams += 1
903            try:
904                # aclosing guarantees the generator (and thus the ffmpeg process chain
905                # behind it) is torn down immediately when the player disconnects
906                # mid-stream, instead of lingering until garbage collection finalizes
907                # the abandoned generator.
908                async with aclosing(audio_bytes):
909                    async for chunk in audio_bytes:
910                        try:
911                            await resp.write(chunk)
912                            bytes_sent += len(chunk)
913                            if not first_chunk_received:
914                                first_chunk_received = True
915                                # inform the queue that the track is now loaded in the buffer
916                                # so for example the next track can be enqueued
917                                self.mass.player_queues.track_loaded_in_buffer(
918                                    queue_item.queue_id, queue_item.queue_item_id
919                                )
920                        except (BrokenPipeError, ConnectionResetError, ConnectionError) as err:
921                            if (
922                                first_chunk_received
923                                and not player.stop_called
924                                and queue_item.streamdetails.duration  # ignore for radio streams
925                            ):
926                                # Player disconnected (unexpected) after receiving at least
927                                # some data. This could indicate buffering issues, network
928                                # problems, or player-specific issues.
929                                self.logger.warning(
930                                    "Player %s disconnected prematurely from stream for %s (%s) - "
931                                    "error: %s, sent %d bytes, content_length=%s",
932                                    queue.display_name,
933                                    queue_item.name,
934                                    queue_item.uri,
935                                    err.__class__.__name__,
936                                    bytes_sent,
937                                    resp.content_length,
938                                )
939                            break
940            finally:
941                self._active_output_streams -= 1
942            if queue_item.streamdetails.stream_error:
943                self.logger.error(
944                    "Error streaming QueueItem %s (%s) to %s",
945                    queue_item.name,
946                    queue_item.uri,
947                    queue.display_name,
948                )
949            elif (
950                bytes_sent > 0
951                and queue_item.streamdetails
952                and queue_item.streamdetails.seconds_streamed
953                and queue_item.duration
954            ):
955                # cache the actual encoded bytes-per-second for this URI + output format
956                # so future content_length estimates are near-exact
957                self.mass.create_task(
958                    store_content_length_in_cache(
959                        self.mass,
960                        queue_item.uri,
961                        output_format,
962                        bytes_sent,
963                        queue_item.streamdetails.seconds_streamed,
964                    )
965                )
966            return resp
967        finally:
968            # Paired with on_source_selected — fires regardless of how streaming
969            # ended (normal completion, client disconnect, exception). Lets
970            # NAMED_PIPE plugins release ownership without depending on an
971            # external session event. The stream_session_id is the same token
972            # passed to on_source_selected; the provider must reject the
973            # callback if it does not match the currently stored active
974            # session (otherwise a stale teardown from a superseded same-queue
975            # request would clear the live claim of its replacement).
976            if audio_source_provider is not None and audio_source_id is not None:
977                # Provider teardown failures must not break the response cycle
978                # (we're already in finally for a stream that ended one way or
979                # another), but they MUST surface in logs — otherwise a buggy
980                # plugin leaks _in_use_by_queue forever and there is no trail.
981                try:
982                    await audio_source_provider.on_source_unselected(
983                        audio_source_id, queue_id, stream_session_id
984                    )
985                except Exception:
986                    self.logger.warning(
987                        "on_source_unselected raised for provider %s source %s queue %s",
988                        audio_source_provider.instance_id,
989                        audio_source_id,
990                        queue_id,
991                        exc_info=True,
992                    )
993
994    async def serve_queue_flow_stream(self, request: web.Request) -> web.StreamResponse:  # noqa: PLR0915
995        """Stream Queue Flow audio to player."""
996        self._log_request(request)
997        queue_id = request.match_info["queue_id"]
998        player_id = request.match_info["player_id"]
999        if not (queue := self.mass.player_queues.get(queue_id)):
1000            raise web.HTTPNotFound(reason=f"Unknown Queue: {queue_id}")
1001        session_id = request.match_info["session_id"]
1002        queue_data = self.mass.player_queues.queue_data(queue_id)
1003        if queue_data.session_id is None or session_id != queue_data.session_id:
1004            raise web.HTTPNotFound(reason=f"Unknown (or invalid) session: {session_id}")
1005        if not (player := self.mass.players.get_player(player_id)):
1006            raise web.HTTPNotFound(reason=f"Unknown Player: {player_id}")
1007        start_queue_item_id = request.match_info["queue_item_id"]
1008        start_queue_item = self.mass.player_queues.get_item(queue_id, start_queue_item_id)
1009        if not start_queue_item:
1010            raise web.HTTPNotFound(reason=f"Unknown Queue item: {start_queue_item_id}")
1011
1012        # select the PCM format for the flow stream, anchored on the first track
1013        crossfade_mode = (
1014            self.get_crossfade_mode(queue)
1015            if start_queue_item.media_type == MediaType.TRACK
1016            else CrossfadeMode.DISABLED
1017        )
1018        flow_pcm_format = await self.audio.select_flow_pcm_format(
1019            player,
1020            start_streamdetails=start_queue_item.streamdetails,
1021            crossfade_enabled=crossfade_mode != CrossfadeMode.DISABLED,
1022            overlay_active=overlay_active(queue),
1023        )
1024
1025        # work out output format/details
1026        output_format = await self.audio.get_output_format(
1027            output_format_str=request.match_info["fmt"],
1028            player=player,
1029            content_sample_rate=flow_pcm_format.sample_rate,
1030            content_bit_depth=flow_pcm_format.bit_depth,
1031            media_type=start_queue_item.media_type,
1032        )
1033        # work out ICY metadata support
1034        icy_preference = self.mass.config.get_raw_player_config_value(
1035            player_id,
1036            CONF_ENTRY_ENABLE_ICY_METADATA.key,
1037            CONF_ENTRY_ENABLE_ICY_METADATA.default_value,
1038        )
1039        enable_icy = request.headers.get("Icy-MetaData", "") == "1" and icy_preference != "disabled"
1040        icy_meta_interval = 256000 if icy_preference == "full" else 16384
1041
1042        # prepare request, add some DLNA/UPNP compatible headers.
1043        # icy-name (in DEFAULT_STREAM_HEADERS) is always present so players have a
1044        # readable stream name; the rest of the ICY/shoutcast metadata headers are
1045        # only advertised when the client actually requested ICY metadata, rather
1046        # than on every flow response.
1047        headers = {
1048            **DEFAULT_STREAM_HEADERS,
1049            **(ICY_HEADERS if enable_icy else {}),
1050            "contentFeatures.dlna.org": DLNA_CONTENT_FEATURES_REALTIME,
1051            "Content-Type": get_mime_type(output_format.output_format_str),
1052        }
1053        if enable_icy:
1054            headers["icy-metaint"] = str(icy_meta_interval)
1055
1056        resp = web.StreamResponse(status=200, reason="OK", headers=headers)
1057        http_profile = player.get_config_value(CONF_HTTP_PROFILE, "default")
1058        if http_profile == "forced_content_length":
1059            # just set an insane high content length to make sure the player keeps playing
1060            resp.content_length = calculate_content_length(output_format, 12 * 3600)
1061        elif http_profile == "chunked":
1062            resp.enable_chunked_encoding()
1063
1064        await resp.prepare(request)
1065
1066        # return early if this is not a GET request
1067        if request.method != "GET":
1068            return resp
1069
1070        self._update_audio_processing_context(
1071            queue=queue,
1072            queue_item=start_queue_item,
1073            pcm_format=flow_pcm_format,
1074            crossfade_mode=crossfade_mode,
1075            overlay_enabled=overlay_active(queue),
1076            session_id=session_id,
1077        )
1078        output_plan = self.audio.get_player_output_plan(
1079            player.player_id,
1080            flow_pcm_format,
1081            output_format,
1082            shared_player_ids=player.state.group_members,
1083            queue_id=queue_id,
1084            session_id=session_id,
1085        )
1086
1087        # all checks passed, start streaming!
1088        # this final ffmpeg process in the chain will convert the raw, lossless PCM audio into
1089        # the desired output format for the player including any player specific filter params
1090        # such as channels mixing, DSP, resampling and, only if needed, encoding to lossy formats
1091        self.logger.debug("Start serving Queue flow audio stream for %s", queue.display_name)
1092
1093        # Mark this player as actively streaming so audio analysis yields CPU to playback
1094        # for the duration of the flow stream (see audio_analysis.playback_active).
1095        self._active_output_streams += 1
1096        flow_stream = self.audio.get_queue_flow_stream(
1097            queue=queue,
1098            start_queue_item=start_queue_item,
1099            pcm_format=flow_pcm_format,
1100            session_id=session_id,
1101            protocol_player=player,
1102        )
1103        if overlay_active(queue):
1104            flow_stream = self.audio.get_overlay_mixed_stream(queue, flow_stream, flow_pcm_format)
1105        audio_bytes = get_ffmpeg_stream(
1106            audio_input=flow_stream,
1107            input_format=flow_pcm_format,
1108            output_format=output_format,
1109            filter_params=output_plan.filter_params,
1110            # we need to slowly feed the music to avoid the player stopping and later
1111            # restarting (or completely failing) the audio stream by keeping the buffer short.
1112            # this is reported to be an issue especially with Chromecast players.
1113            # see for example: https://github.com/music-assistant/support/issues/3717
1114            # allow buffer ahead of a few seconds and read rest in (near) realtime
1115            extra_input_args=["-readrate", "1.05", "-readrate_initial_burst", "5"],
1116            chunk_size=icy_meta_interval if enable_icy else calculate_content_length(output_format),
1117        )
1118        client_disconnected = False
1119        try:
1120            # aclosing guarantees the flow stream (and thus the ffmpeg process chain
1121            # behind it) is torn down immediately when the player disconnects
1122            # mid-stream, instead of lingering until garbage collection finalizes
1123            # the abandoned generator.
1124            async with aclosing(audio_bytes):
1125                async for chunk in audio_bytes:
1126                    try:
1127                        await resp.write(chunk)
1128                    except BrokenPipeError, ConnectionResetError, ConnectionError:
1129                        # race condition
1130                        client_disconnected = True
1131                        break
1132
1133                    if not enable_icy:
1134                        continue
1135
1136                    # if icy metadata is enabled, send the icy metadata after the chunk
1137                    if (
1138                        # use current item here and not buffered item, otherwise
1139                        # the icy metadata will be too much ahead
1140                        (current_item := queue.current_item)
1141                        and current_item.streamdetails
1142                        and current_item.streamdetails.stream_title
1143                    ):
1144                        title = current_item.streamdetails.stream_title
1145                    elif queue and current_item and current_item.name:
1146                        title = current_item.name
1147                    else:
1148                        title = "Music Assistant"
1149                    metadata = f"StreamTitle='{title}';".encode()
1150                    if icy_preference == "full" and current_item and current_item.image:
1151                        metadata += f"StreamURL='{current_item.image.path}'".encode()
1152                    while len(metadata) % 16 != 0:
1153                        metadata += b"\x00"
1154                    length = len(metadata)
1155                    length_b = chr(int(length / 16)).encode()
1156                    await resp.write(length_b + metadata)
1157        finally:
1158            self._active_output_streams -= 1
1159
1160        if not client_disconnected and http_profile == "forced_content_length":
1161            await self._finish_flow_stream(resp, queue_id, session_id)
1162
1163        return resp
1164
1165    async def serve_command_request(self, request: web.Request) -> web.FileResponse:
1166        """Handle special 'command' request for a player."""
1167        self._log_request(request)
1168        queue_id = request.match_info["queue_id"]
1169        command = request.match_info["command"]
1170        if command == "next":
1171            self.mass.create_task(self.mass.player_queues.next(queue_id))
1172        return web.FileResponse(SILENCE_FILE, headers={"icy-name": "Music Assistant"})
1173
1174    async def serve_announcement_stream(self, request: web.Request) -> web.StreamResponse:
1175        """Stream announcement audio to a player."""
1176        self._log_request(request)
1177        player_id = request.match_info["player_id"]
1178        if not (player := self.mass.players.get_player(player_id)):
1179            raise web.HTTPNotFound(reason=f"Unknown Player: {player_id}")
1180        if not (announce_data := self.announcement_renderer.get_for_player(player_id)):
1181            raise web.HTTPNotFound(reason=f"No pending announcements for Player: {player_id}")
1182
1183        # work out output format/details
1184        fmt = request.match_info["fmt"]
1185        audio_format = AudioFormat(content_type=ContentType.try_parse(fmt))
1186
1187        http_profile = self._get_announcement_http_profile(player_id, announce_data)
1188
1189        # return early if this is not a GET request:
1190        # players often probe the url with a HEAD request before fetching it and
1191        # rendering the announcement for such a probe would run the entire (costly)
1192        # TTS/ffmpeg chain twice for a single announcement.
1193        if request.method != "GET":
1194            resp = web.StreamResponse(status=200, reason="OK", headers=DEFAULT_STREAM_HEADERS)
1195            resp.content_type = get_mime_type(audio_format.output_format_str)
1196            if http_profile == "chunked":
1197                resp.enable_chunked_encoding()
1198            await resp.prepare(request)
1199            return resp
1200
1201        if http_profile == "forced_content_length":
1202            # given the fact that an announcement is just a short audio clip,
1203            # just send it over completely at once so we have a fixed content length
1204            data = bytearray()
1205            announcement_stream = self.get_announcement_stream(announce_data, audio_format)
1206            # aclosing guarantees the stream (and thus the ffmpeg process chain behind
1207            # it) is torn down immediately when the request is cancelled, instead of
1208            # lingering until garbage collection finalizes the abandoned generator.
1209            async with aclosing(announcement_stream):
1210                async for chunk in announcement_stream:
1211                    data += chunk
1212            return web.Response(
1213                body=bytes(data),
1214                content_type=get_mime_type(audio_format.output_format_str),
1215                headers=DEFAULT_STREAM_HEADERS,
1216            )
1217
1218        resp = web.StreamResponse(status=200, reason="OK", headers=DEFAULT_STREAM_HEADERS)
1219        resp.content_type = get_mime_type(audio_format.output_format_str)
1220        if http_profile == "chunked":
1221            resp.enable_chunked_encoding()
1222
1223        await resp.prepare(request)
1224
1225        # all checks passed, start streaming!
1226        self.logger.debug(
1227            "Start serving audio stream for Announcement %s to %s",
1228            announce_data["announcement_url"],
1229            player.display_name,
1230        )
1231        announcement_stream = self.get_announcement_stream(announce_data, audio_format)
1232        # aclosing guarantees the stream (and thus the ffmpeg process chain behind
1233        # it) is torn down immediately when the player disconnects mid-stream,
1234        # instead of lingering until garbage collection finalizes the abandoned
1235        # generator.
1236        async with aclosing(announcement_stream):
1237            async for chunk in announcement_stream:
1238                try:
1239                    await resp.write(chunk)
1240                except BrokenPipeError, ConnectionResetError:
1241                    break
1242
1243        self.logger.debug(
1244            "Finished serving audio stream for Announcement %s to %s",
1245            announce_data["announcement_url"],
1246            player.display_name,
1247        )
1248
1249        return resp
1250
1251    def get_command_url(self, player_or_queue_id: str, command: str) -> str:
1252        """Get the url for the special command stream."""
1253        return f"{self.base_url}/command/{player_or_queue_id}/{command}.mp3"
1254
1255    def get_announcement_url(
1256        self,
1257        player_id: str,
1258        content_type: ContentType = ContentType.MP3,
1259    ) -> str:
1260        """
1261        Get the url that serves the announcement registered for the given player.
1262
1263        :param player_id: The player the announcement is played on.
1264        :param content_type: The format to serve the announcement in.
1265        """
1266        # use stream server to host announcement on local network
1267        # this ensures playback on all players, including ones that do not
1268        # like https hosts and it also offers the pre-announce 'bell'
1269        return f"{self.base_url}/announcement/{player_id}.{content_type.value}"
1270
1271    def get_stream(
1272        self,
1273        media: PlayerMedia,
1274        pcm_format: AudioFormat,
1275        player_id: str | None = None,
1276        force_flow_mode: bool = False,
1277        use_flow_stream_buffering: bool = False,
1278    ) -> AsyncGenerator[bytes]:
1279        """
1280        Get a stream of the given media as raw PCM audio.
1281
1282        This is used as helper for player providers that can consume the raw PCM
1283        audio stream directly (e.g. AirPlay) and not rely on HTTP transport.
1284
1285        :param media: The PlayerMedia to stream.
1286        :param pcm_format: The desired output PCM format.
1287        :param player_id: The player ID requesting the stream. Used to determine
1288            if flow mode should be used based on the player's capabilities.
1289        :param force_flow_mode: Force flow mode regardless of player capabilities.
1290            Used for multi-client streaming scenarios that require continuous streams.
1291        :param use_flow_stream_buffering: Buffer the flow stream to provide headroom
1292            during smart fades transitions. Use for consumers that read directly
1293            (e.g. AirPlay, Snapcast) and can't tolerate stalls.
1294        """
1295        # select audio source
1296        if media.media_type == MediaType.ANNOUNCEMENT:
1297            # special case: stream announcement
1298            assert media.custom_data
1299            return self.get_announcement_stream(cast("AnnounceData", media.custom_data), pcm_format)
1300        if (
1301            media.source_id
1302            and media.source_id.startswith(UGP_PREFIX)
1303            and media.uri
1304            and "/ugp/" in media.uri
1305        ):
1306            # special case: member player accessing UGP stream
1307            # Check URI to distinguish from the UGP accessing its own stream
1308            ugp_player = cast("UniversalGroupPlayer", self.mass.players.get_player(media.source_id))
1309            ugp_stream = ugp_player.stream
1310            assert ugp_stream is not None  # for type checker
1311            if ugp_stream.base_pcm_format == pcm_format:
1312                # no conversion needed
1313                return ugp_stream.subscribe_raw()
1314            return ugp_stream.get_stream(output_format=pcm_format)
1315        if media.source_id and media.queue_item_id:
1316            # Queue stream request - determine flow_mode based on player capabilities
1317            # or force it if explicitly requested (e.g., for multi-client streaming)
1318            protocol_player = self.mass.players.get_player(player_id) if player_id else None
1319            queue_id = media.source_id
1320            queue = self.mass.player_queues.get(queue_id)
1321            queue_session_id = cast(
1322                "str | None",
1323                (media.custom_data or {}).get("session_id"),
1324            )
1325            crossfade_needs_flow_mode = (
1326                # crossfade only applies to tracks; if the queue has it enabled but the
1327                # player(protocol) does not support gapless playback, we need to enforce flow mode
1328                media.media_type == MediaType.TRACK
1329                and queue is not None
1330                and queue.crossfade_enabled
1331                and protocol_player
1332                and not protocol_player.supports_gapless
1333            )
1334            # the audio overlay is mixed into the queue's continuous (flow) stream;
1335            # per-item requests would restart the overlay at every track boundary
1336            overlay_needs_flow_mode = queue is not None and overlay_active(queue)
1337            flow_mode = (
1338                force_flow_mode
1339                or (protocol_player is not None and protocol_player.flow_mode)
1340                or crossfade_needs_flow_mode
1341                or overlay_needs_flow_mode
1342            )
1343            if media.media_type in (MediaType.RADIO, MediaType.AUDIO_SOURCE):
1344                # flow_mode for live/infinite streams is pointless
1345                flow_mode = False
1346            if flow_mode:
1347                # flow stream request
1348                assert queue
1349                start_queue_item = self.mass.player_queues.get_item(
1350                    media.source_id, media.queue_item_id
1351                )
1352                assert start_queue_item
1353                crossfade_mode = (
1354                    self.get_crossfade_mode(queue)
1355                    if start_queue_item.media_type == MediaType.TRACK
1356                    else CrossfadeMode.DISABLED
1357                )
1358                self._update_audio_processing_context(
1359                    queue=queue,
1360                    queue_item=start_queue_item,
1361                    pcm_format=pcm_format,
1362                    crossfade_mode=crossfade_mode,
1363                    overlay_enabled=overlay_active(queue),
1364                    session_id=queue_session_id,
1365                )
1366                flow_stream = self.audio.get_queue_flow_stream(
1367                    queue=queue,
1368                    start_queue_item=start_queue_item,
1369                    pcm_format=pcm_format,
1370                    session_id=queue_session_id,
1371                    protocol_player=protocol_player,
1372                )
1373                if overlay_active(queue):
1374                    flow_stream = self.audio.get_overlay_mixed_stream(
1375                        queue, flow_stream, pcm_format
1376                    )
1377                if use_flow_stream_buffering:
1378                    return buffered(flow_stream, buffer_size=30, min_buffer_before_yield=1)
1379                return flow_stream
1380            # single item stream (e.g. radio or non-flow mode)
1381            queue_item = self.mass.player_queues.get_item(media.source_id, media.queue_item_id)
1382            assert queue_item
1383            if queue is not None:
1384                self._update_audio_processing_context(
1385                    queue=queue,
1386                    queue_item=queue_item,
1387                    pcm_format=pcm_format,
1388                    crossfade_mode=(
1389                        self.get_crossfade_mode(queue)
1390                        if queue_item.media_type == MediaType.TRACK
1391                        else CrossfadeMode.DISABLED
1392                    ),
1393                    overlay_enabled=(
1394                        queue_item.media_type == MediaType.RADIO and overlay_active(queue)
1395                    ),
1396                    session_id=queue_session_id,
1397                )
1398            inner_stream = self.audio.get_queue_item_stream(
1399                queue_item=queue_item,
1400                pcm_format=pcm_format,
1401                seek_position=(
1402                    int(queue_item.streamdetails.seek_position) if queue_item.streamdetails else 0
1403                ),
1404                playback_speed=cast(
1405                    "float", queue_item.extra_attributes.get("playback_speed", 1.0)
1406                ),
1407                session_id=queue_session_id,
1408            )
1409            if (
1410                queue is not None
1411                and queue_item.media_type == MediaType.RADIO
1412                and overlay_active(queue)
1413            ):
1414                # radio plays as a single long-lived stream, so mix the overlay in here
1415                inner_stream = self.audio.get_overlay_mixed_stream(queue, inner_stream, pcm_format)
1416            # mirror the on_source_selected/unselected lifecycle the HTTP route
1417            # fires, so direct-PCM consumers (AirPlay, Snapcast, UGP) honour the
1418            # plugin contract too
1419            if (
1420                queue_item.media_item is not None
1421                and queue_item.media_item.media_type == MediaType.AUDIO_SOURCE
1422            ):
1423                return self._wrap_with_audio_source_lifecycle(
1424                    inner=inner_stream,
1425                    queue_item=queue_item,
1426                    player_id=player_id or media.source_id,
1427                )
1428            return inner_stream
1429        # assume url or some other direct path
1430        # NOTE: this will fail if its an uri not playable by ffmpeg
1431        return get_ffmpeg_stream(
1432            audio_input=media.uri,
1433            input_format=AudioFormat(content_type=ContentType.try_parse(media.uri)),
1434            output_format=pcm_format,
1435        )
1436
1437    async def get_preview_stream(
1438        self,
1439        provider_instance_id_or_domain: str,
1440        item_id: str,
1441        media_type: MediaType = MediaType.TRACK,
1442    ) -> AsyncGenerator[bytes]:
1443        """Create a 30 seconds preview audioclip for the given media item."""
1444        if not (music_prov := self.mass.get_provider(provider_instance_id_or_domain)):
1445            raise ProviderUnavailableError
1446        if music_prov.type != ProviderType.MUSIC:
1447            msg = f"{provider_instance_id_or_domain} is not a music provider"
1448            raise InvalidDataError(msg)
1449        music_prov = cast("MusicProvider", music_prov)
1450
1451        try:
1452            await self.mass.music.get_item(
1453                media_type,
1454                item_id,
1455                provider_instance_id_or_domain,
1456                allow_update_metadata=False,
1457            )
1458        except MediaNotFoundError as err:
1459            msg = f"Item {item_id} not found in provider {provider_instance_id_or_domain}"
1460            raise InvalidDataError(msg) from err
1461
1462        streamdetails = await music_prov.get_stream_details(item_id, media_type)
1463        pcm_format = AudioFormat(
1464            content_type=ContentType.from_bit_depth(streamdetails.audio_format.bit_depth),
1465            sample_rate=streamdetails.audio_format.sample_rate,
1466            bit_depth=streamdetails.audio_format.bit_depth,
1467            channels=streamdetails.audio_format.channels,
1468        )
1469        async for chunk in get_ffmpeg_stream(
1470            audio_input=self.audio.get_media_stream(
1471                streamdetails=streamdetails, pcm_format=pcm_format
1472            ),
1473            input_format=pcm_format,
1474            output_format=AudioFormat(content_type=ContentType.AAC),
1475            extra_input_args=["-t", "30"],
1476        ):
1477            yield chunk
1478
1479    async def get_announcement_stream(
1480        self, announce_data: AnnounceData, output_format: AudioFormat
1481    ) -> AsyncGenerator[bytes]:
1482        """
1483        Get the audio of an announcement (pre-announce chime + announcement).
1484
1485        Any number of consumers may stream the same announcement at once; its source is
1486        fetched and decoded only once. The audio stays available while the stream is
1487        held open.
1488
1489        :param announce_data: The announcement to stream.
1490        :param output_format: The format to deliver the audio in.
1491        """
1492        render = self.announcement_renderer.acquire(announce_data)
1493        try:
1494            # aclosing guarantees this consumer's ffmpeg encoder is torn down
1495            # immediately when it goes away, instead of lingering until garbage
1496            # collection finalizes the abandoned generator.
1497            stream = render.get_stream(output_format)
1498            async with aclosing(stream):
1499                async for chunk in stream:
1500                    yield chunk
1501        finally:
1502            await self.announcement_renderer.release(render)
1503
1504    async def get_announcement_duration(
1505        self, announcement: PlayerMedia, timeout: float = DEFAULT_RENDER_TIMEOUT
1506    ) -> int | None:
1507        """
1508        Get the exact duration (in seconds) of an announcement, once it finished rendering.
1509
1510        Waits for the audio to be rendered in full, so call this while the announcement
1511        plays rather than before handing it to a player. Returns None when the length can
1512        not be determined, e.g. the announcement is no longer playing or its source did
1513        not deliver in time.
1514
1515        :param announcement: The announcement to return the duration for.
1516        :param timeout: Maximum time to wait for the audio to finish rendering.
1517        """
1518        if announcement.duration:
1519            return announcement.duration
1520        if not announcement.custom_data:
1521            return None
1522        render = self.announcement_renderer.get(cast("AnnounceData", announcement.custom_data))
1523        if render is None:
1524            return None
1525        duration = await render.wait_finished(timeout)
1526        return ceil(duration) if duration else None
1527
1528    async def _wrap_with_audio_source_lifecycle(
1529        self,
1530        inner: AsyncGenerator[bytes],
1531        queue_item: QueueItem,
1532        player_id: str,
1533    ) -> AsyncGenerator[bytes]:
1534        """
1535        Wrap an AudioSource queue item stream with on_source_selected/unselected hooks.
1536
1537        Direct-PCM consumers (AirPlay, Snapcast, UGP, ...) call ``get_stream`` instead
1538        of going through the HTTP route, but the plugin contract requires the
1539        lifecycle hooks to fire for every actual stream request — they're what
1540        claim/release the per-queue exclusive ownership and trigger acquisition
1541        side effects like the Spotify Connect Web API play kick. This wrapper
1542        gives those consumers the same lifecycle the HTTP route already provides.
1543
1544        :param inner: The underlying audio stream generator.
1545        :param queue_item: The AudioSource queue item being streamed.
1546        :param player_id: The protocol player consuming this stream.
1547        """
1548        media_item = queue_item.media_item
1549        assert media_item is not None  # caller checked media_type == AUDIO_SOURCE
1550        prov = self.mass.get_provider(media_item.provider)
1551        queue_id = queue_item.queue_id
1552        if not isinstance(prov, PluginProvider):
1553            async for chunk in inner:
1554                yield chunk
1555            return
1556        source_id = media_item.item_id
1557        stream_session_id = uuid4().hex
1558        # single try/finally so on_source_unselected fires even when
1559        # on_source_selected raises after partially claiming state; the
1560        # provider's session_id guard makes a no-op claim release safe.
1561        try:
1562            try:
1563                await prov.on_source_selected(source_id, player_id, queue_id, stream_session_id)
1564            except RuntimeError as err:
1565                # provider intentionally aborts the request — surface as AudioError
1566                raise AudioError(str(err)) from err
1567            async for chunk in inner:
1568                yield chunk
1569        finally:
1570            try:
1571                await prov.on_source_unselected(source_id, queue_id, stream_session_id)
1572            except Exception as err:
1573                self.logger.exception(
1574                    "on_source_unselected raised for provider %s source %s queue %s: %s",
1575                    prov.instance_id,
1576                    source_id,
1577                    queue_id,
1578                    err,
1579                )
1580
1581    def _update_audio_processing_context(
1582        self,
1583        queue: PlayerQueue,
1584        queue_item: QueueItem,
1585        pcm_format: AudioFormat,
1586        crossfade_mode: CrossfadeMode,
1587        overlay_enabled: bool,
1588        session_id: str | None = None,
1589    ) -> None:
1590        """
1591        Store the shared processing context selected for a queue item.
1592
1593        :param queue: Active player queue.
1594        :param queue_item: Queue item being prepared.
1595        :param pcm_format: Shared PCM format leaving queue processing.
1596        :param crossfade_mode: Effective crossfade mode for the item.
1597        :param overlay_enabled: Whether an overlay is mixed into this stream.
1598        :param session_id: Queue session that owns processing-detail updates.
1599        """
1600        if queue_item.streamdetails is None:
1601            return
1602        queue_data = self.mass.player_queues.queue_data_or_none(queue.queue_id)
1603        if (
1604            queue_data is None
1605            or (processing_session_id := session_id or queue_data.session_id) is None
1606            or queue_data.session_id != processing_session_id
1607        ):
1608            return
1609        self.audio_processing.start_session(queue.queue_id, processing_session_id)
1610        self.audio_processing.update_item_context(
1611            queue_id=queue.queue_id,
1612            session_id=processing_session_id,
1613            queue_item_id=queue_item.queue_item_id,
1614            queue_processing=AudioQueueProcessing(
1615                pcm_format=pcm_format,
1616                playback_speed=cast(
1617                    "float",
1618                    queue_item.extra_attributes.get("playback_speed", 1.0),
1619                ),
1620                crossfade_mode=crossfade_mode,
1621                overlay_active=overlay_enabled,
1622            ),
1623            alters_audio=queue_item.streamdetails.fade_in,
1624        )
1625
1626    def _get_announcement_http_profile(self, player_id: str, announce_data: AnnounceData) -> str:
1627        """
1628        Resolve the http profile for serving an announcement stream.
1629
1630        Announcement urls are registered under the visible player's id, but the
1631        stream may be fetched by a linked protocol player; the profile must come
1632        from the player that actually performs the fetch.
1633        """
1634        announce_player = None
1635        if announce_player_id := announce_data.get("announce_player_id"):
1636            announce_player = self.mass.players.get_player(announce_player_id)
1637        if announce_player is None:
1638            announce_player = self.mass.players.get_player(player_id)
1639        if announce_player is None:
1640            return "default"
1641        return announce_player.get_output_config_value(CONF_HTTP_PROFILE, "default")
1642
1643    async def _finish_flow_stream(
1644        self, resp: web.StreamResponse, queue_id: str, session_id: str
1645    ) -> None:
1646        """
1647        Close a fully served flow stream, giving the player time to drain when it ends the queue.
1648
1649        :param resp: The flow stream response, already fully written.
1650        :param queue_id: Id of the queue the flow stream belongs to.
1651        :param session_id: Stream session this response was opened for.
1652        """
1653        if self.mass.player_queues.flow_queue_exhausted(queue_id, session_id):
1654            # the player is still holding a few seconds of audio it has not rendered yet
1655            # and drops that as soon as the stream ends, so let it play out first.
1656            # a flow that ends to be restarted right away gets no such grace: there the
1657            # player should go idle as soon as possible so the next stream can start.
1658            self.logger.debug(
1659                "Flow stream for queue %s reached the end of the queue - holding the "
1660                "connection open for %ss so the player can play out its buffer",
1661                queue_id,
1662                FLOW_STREAM_LEAD_OUT_SECONDS,
1663            )
1664            await asyncio.sleep(FLOW_STREAM_LEAD_OUT_SECONDS)
1665        # aiohttp derives keep-alive from the request, so the 'Connection: close' we
1666        # advertise is relayed to the player but never applied to the response itself.
1667        # Without this the player is left waiting on a stream that already ended.
1668        resp.force_close()
1669
1670    def _log_request(self, request: web.Request) -> None:
1671        """Log request."""
1672        if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
1673            self.logger.log(
1674                VERBOSE_LOG_LEVEL,
1675                "Got %s request to %s from %s\nheaders: %s\n",
1676                request.method,
1677                request.path,
1678                request.remote,
1679                redact_sensitive_headers(request.headers),
1680            )
1681        else:
1682            self.logger.debug(
1683                "Got %s request to %s from %s (HTTP/%s.%s, connection: %s)",
1684                request.method,
1685                request.path,
1686                request.remote,
1687                request.version.major,
1688                request.version.minor,
1689                request.headers.get("Connection", "-"),
1690            )
1691
1692    async def _reload_network_dependent_providers(self) -> None:
1693        """Reload the providers that captured the streamserver network, if it changed."""
1694        previous = self._network_fingerprint
1695        current = (
1696            self._bind_ip,
1697            str(self.publish_ip),
1698            cast("int", self.publish_port),
1699            tuple(self._publish_addresses),
1700        )
1701        if previous is None or previous == current:
1702            self._network_fingerprint = current
1703            return
1704        # these providers bind or advertise the network while they load, so a plain
1705        # reload is what moves them over - they share no lighter rebind path
1706        instance_ids = [
1707            prov.instance_id
1708            for prov in self.mass.providers
1709            if prov.reload_on_streams_network_change
1710        ]
1711        for instance_id in instance_ids:
1712            try:
1713                config = await self.mass.config.get_provider_config(instance_id)
1714                self.logger.info(
1715                    "Streamserver network changed, reloading provider %s",
1716                    config.name or config.domain,
1717                )
1718                await self.mass.load_provider_config(config)
1719            except Exception as err:
1720                self.logger.warning(
1721                    "Error reloading provider %s: %s",
1722                    instance_id,
1723                    str(err) or err.__class__.__name__,
1724                    exc_info=err,
1725                )
1726        # only mark the new network as applied once the loop completed, so a run cut short
1727        # by a second config change runs again on the next reload
1728        self._network_fingerprint = current
1729
1730    def _setup_smart_fades_logger(self, config: CoreConfig) -> None:
1731        """Set up smart fades logger level."""
1732        log_level = str(config.get_value(CONF_SMART_FADES_LOG_LEVEL))
1733        if log_level == "GLOBAL":
1734            self.audio.smart_fades_mixer.logger.setLevel(self.logger.level)
1735        else:
1736            self.audio.smart_fades_mixer.logger.setLevel(log_level)
1737
1738    def _resolve_publish_state(self, bind_ip: str, publish_candidates: tuple[str, ...]) -> None:
1739        """
1740        Resolve the addresses and base URL to advertise for the given bind address.
1741
1742        Reads ``self.publish_port``, so set that first.
1743
1744        :param bind_ip: Address the streamserver binds to (a wildcard means all interfaces).
1745        :param publish_candidates: Host addresses reachable from the local network, ranked.
1746        """
1747        self._bind_ip = bind_ip
1748        self._publish_addresses = _get_publish_addresses(
1749            bind_ip, self._configured_publish_ip, publish_candidates
1750        )
1751        # the single address players are handed, taken from the top of the ranked list
1752        self.publish_ip = self._publish_addresses[0]
1753        self._base_url = f"http://{format_ip_for_url(self.publish_ip)}:{self.publish_port}"
1754
1755
1756def _same_ip_family(ip: str, other_ip: str) -> bool:
1757    """Return whether two addresses belong to the same IP family."""
1758    return (":" in ip) == (":" in other_ip)
1759