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