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                provider=prov,
1091            )
1092            serving = True
1093            self._active_output_streams += 1
1094            try:
1095                async with aclosing(audio_bytes):
1096                    async for chunk in audio_bytes:
1097                        if (
1098                            self.mass.players.get_audio_source_session(source_player_id)
1099                            is not session
1100                            or session.playback_session_id != playback_session_id
1101                            or session.stream_session_id != stream_session_id
1102                        ):
1103                            self.logger.debug(
1104                                "Ending stream for %s: a newer request took the source over",
1105                                session.source.name,
1106                            )
1107                            break
1108                        try:
1109                            await resp.write(chunk)
1110                        except BrokenPipeError, ConnectionResetError, ConnectionError:
1111                            break
1112            finally:
1113                self._active_output_streams -= 1
1114            return resp
1115        finally:
1116            if claimed:
1117                try:
1118                    await prov.on_source_unselected(
1119                        session.source_id, source_player_id, stream_session_id
1120                    )
1121                except Exception:
1122                    self.logger.warning(
1123                        "on_source_unselected raised for provider %s source %s player %s",
1124                        prov.instance_id,
1125                        session.source_id,
1126                        source_player_id,
1127                        exc_info=True,
1128                    )
1129            if not serving:
1130                await self._release_unstarted_audio_source(session, playback_session_id)
1131
1132    async def serve_queue_flow_stream(self, request: web.Request) -> web.StreamResponse:  # noqa: PLR0915
1133        """Stream Queue Flow audio to player."""
1134        self._log_request(request)
1135        queue_id = request.match_info["queue_id"]
1136        player_id = request.match_info["player_id"]
1137        if not (queue := self.mass.player_queues.get(queue_id)):
1138            raise web.HTTPNotFound(reason=f"Unknown Queue: {queue_id}")
1139        session_id = request.match_info["session_id"]
1140        queue_data = self.mass.player_queues.queue_data(queue_id)
1141        if queue_data.session_id is None or session_id != queue_data.session_id:
1142            raise web.HTTPNotFound(reason=f"Unknown (or invalid) session: {session_id}")
1143        if not (player := self.mass.players.get_player(player_id)):
1144            raise web.HTTPNotFound(reason=f"Unknown Player: {player_id}")
1145        start_queue_item_id = request.match_info["queue_item_id"]
1146        start_queue_item = self.mass.player_queues.get_item(queue_id, start_queue_item_id)
1147        if not start_queue_item:
1148            raise web.HTTPNotFound(reason=f"Unknown Queue item: {start_queue_item_id}")
1149
1150        # select the PCM format for the flow stream, anchored on the first track
1151        crossfade_mode = (
1152            self.get_crossfade_mode(queue)
1153            if start_queue_item.media_type == MediaType.TRACK
1154            else CrossfadeMode.DISABLED
1155        )
1156        flow_pcm_format = await self.audio.select_flow_pcm_format(
1157            player,
1158            start_streamdetails=start_queue_item.streamdetails,
1159            crossfade_enabled=crossfade_mode != CrossfadeMode.DISABLED,
1160            overlay_active=overlay_active(queue),
1161        )
1162
1163        # work out output format/details
1164        output_format = await self.audio.get_output_format(
1165            output_format_str=request.match_info["fmt"],
1166            player=player,
1167            content_sample_rate=flow_pcm_format.sample_rate,
1168            content_bit_depth=flow_pcm_format.bit_depth,
1169            media_type=start_queue_item.media_type,
1170        )
1171        # work out ICY metadata support
1172        icy_preference = self.mass.config.get_raw_player_config_value(
1173            player_id,
1174            CONF_ENTRY_ENABLE_ICY_METADATA.key,
1175            CONF_ENTRY_ENABLE_ICY_METADATA.default_value,
1176        )
1177        enable_icy = request.headers.get("Icy-MetaData", "") == "1" and icy_preference != "disabled"
1178        icy_meta_interval = 256000 if icy_preference == "full" else 16384
1179
1180        # prepare request, add some DLNA/UPNP compatible headers.
1181        # icy-name (in DEFAULT_STREAM_HEADERS) is always present so players have a
1182        # readable stream name; the rest of the ICY/shoutcast metadata headers are
1183        # only advertised when the client actually requested ICY metadata, rather
1184        # than on every flow response.
1185        headers = {
1186            **DEFAULT_STREAM_HEADERS,
1187            **(ICY_HEADERS if enable_icy else {}),
1188            "contentFeatures.dlna.org": DLNA_CONTENT_FEATURES_REALTIME,
1189            "Content-Type": get_mime_type(output_format.output_format_str),
1190        }
1191        if enable_icy:
1192            headers["icy-metaint"] = str(icy_meta_interval)
1193
1194        resp = web.StreamResponse(status=200, reason="OK", headers=headers)
1195        http_profile = player.get_config_value(CONF_HTTP_PROFILE, "default")
1196        if http_profile == "forced_content_length":
1197            # just set an insane high content length to make sure the player keeps playing
1198            resp.content_length = calculate_content_length(output_format, 12 * 3600)
1199        elif http_profile == "chunked":
1200            resp.enable_chunked_encoding()
1201
1202        await resp.prepare(request)
1203
1204        # return early if this is not a GET request
1205        if request.method != "GET":
1206            return resp
1207
1208        self._update_audio_processing_context(
1209            queue=queue,
1210            queue_item=start_queue_item,
1211            pcm_format=flow_pcm_format,
1212            overlay_enabled=overlay_active(queue),
1213            session_id=session_id,
1214        )
1215        output_plan = self.audio.get_player_output_plan(
1216            player.player_id,
1217            flow_pcm_format,
1218            output_format,
1219            shared_player_ids=player.state.group_members,
1220            queue_id=queue_id,
1221            session_id=session_id,
1222        )
1223
1224        # all checks passed, start streaming!
1225        # this final ffmpeg process in the chain will convert the raw, lossless PCM audio into
1226        # the desired output format for the player including any player specific filter params
1227        # such as channels mixing, DSP, resampling and, only if needed, encoding to lossy formats
1228        self.logger.debug("Start serving Queue flow audio stream for %s", queue.display_name)
1229
1230        # Mark this player as actively streaming so audio analysis yields CPU to playback
1231        # for the duration of the flow stream (see audio_analysis.playback_active).
1232        self._active_output_streams += 1
1233        flow_stream = self.audio.get_queue_flow_stream(
1234            queue=queue,
1235            start_queue_item=start_queue_item,
1236            pcm_format=flow_pcm_format,
1237            session_id=session_id,
1238            protocol_player=player,
1239        )
1240        if overlay_active(queue):
1241            flow_stream = self.audio.get_overlay_mixed_stream(queue, flow_stream, flow_pcm_format)
1242        audio_bytes = get_ffmpeg_stream(
1243            audio_input=flow_stream,
1244            input_format=flow_pcm_format,
1245            output_format=output_format,
1246            filter_params=output_plan.filter_params,
1247            # we need to slowly feed the music to avoid the player stopping and later
1248            # restarting (or completely failing) the audio stream by keeping the buffer short.
1249            # this is reported to be an issue especially with Chromecast players.
1250            # see for example: https://github.com/music-assistant/support/issues/3717
1251            # allow buffer ahead of a few seconds and read rest in (near) realtime
1252            extra_input_args=["-readrate", "1.05", "-readrate_initial_burst", "5"],
1253            chunk_size=icy_meta_interval if enable_icy else calculate_content_length(output_format),
1254        )
1255        client_disconnected = False
1256        try:
1257            # aclosing guarantees the flow stream (and thus the ffmpeg process chain
1258            # behind it) is torn down immediately when the player disconnects
1259            # mid-stream, instead of lingering until garbage collection finalizes
1260            # the abandoned generator.
1261            async with aclosing(audio_bytes):
1262                async for chunk in audio_bytes:
1263                    try:
1264                        await resp.write(chunk)
1265                    except BrokenPipeError, ConnectionResetError, ConnectionError:
1266                        # race condition
1267                        client_disconnected = True
1268                        break
1269
1270                    if not enable_icy:
1271                        continue
1272
1273                    # if icy metadata is enabled, send the icy metadata after the chunk
1274                    if (
1275                        # use current item here and not buffered item, otherwise
1276                        # the icy metadata will be too much ahead
1277                        (current_item := queue.current_item)
1278                        and current_item.streamdetails
1279                        and current_item.streamdetails.stream_title
1280                    ):
1281                        title = current_item.streamdetails.stream_title
1282                    elif queue and current_item and current_item.name:
1283                        title = current_item.name
1284                    else:
1285                        title = "Music Assistant"
1286                    metadata = f"StreamTitle='{title}';".encode()
1287                    if icy_preference == "full" and current_item and current_item.image:
1288                        metadata += f"StreamURL='{current_item.image.path}'".encode()
1289                    while len(metadata) % 16 != 0:
1290                        metadata += b"\x00"
1291                    length = len(metadata)
1292                    length_b = chr(int(length / 16)).encode()
1293                    await resp.write(length_b + metadata)
1294        finally:
1295            self._active_output_streams -= 1
1296
1297        if not client_disconnected and http_profile == "forced_content_length":
1298            await self._finish_flow_stream(resp, queue_id, session_id)
1299
1300        return resp
1301
1302    async def serve_command_request(self, request: web.Request) -> web.FileResponse:
1303        """Handle special 'command' request for a player."""
1304        self._log_request(request)
1305        queue_id = request.match_info["queue_id"]
1306        session_id = request.match_info["session_id"]
1307        queue_data = self.mass.player_queues.queue_data_or_none(queue_id)
1308        if queue_data is None or queue_data.session_id != session_id:
1309            raise web.HTTPNotFound(reason=f"Unknown (or invalid) session: {session_id}")
1310        command = request.match_info["command"]
1311        if command == "next":
1312            self.mass.create_task(self.mass.player_queues.next(queue_id))
1313        return web.FileResponse(SILENCE_FILE, headers={"icy-name": "Music Assistant"})
1314
1315    async def serve_announcement_stream(self, request: web.Request) -> web.StreamResponse:
1316        """Stream announcement audio to a player."""
1317        self._log_request(request)
1318        player_id = request.match_info["player_id"]
1319        if not (player := self.mass.players.get_player(player_id)):
1320            raise web.HTTPNotFound(reason=f"Unknown Player: {player_id}")
1321        if not (announce_data := self.announcement_renderer.get_for_player(player_id)):
1322            raise web.HTTPNotFound(reason=f"No pending announcements for Player: {player_id}")
1323
1324        # work out output format/details
1325        fmt = request.match_info["fmt"]
1326        audio_format = AudioFormat(content_type=ContentType.try_parse(fmt))
1327
1328        http_profile = self._get_announcement_http_profile(player_id, announce_data)
1329
1330        # return early if this is not a GET request:
1331        # players often probe the url with a HEAD request before fetching it and
1332        # rendering the announcement for such a probe would run the entire (costly)
1333        # TTS/ffmpeg chain twice for a single announcement.
1334        if request.method != "GET":
1335            resp = web.StreamResponse(status=200, reason="OK", headers=DEFAULT_STREAM_HEADERS)
1336            resp.content_type = get_mime_type(audio_format.output_format_str)
1337            if http_profile == "chunked":
1338                resp.enable_chunked_encoding()
1339            await resp.prepare(request)
1340            return resp
1341
1342        if http_profile == "forced_content_length":
1343            # given the fact that an announcement is just a short audio clip,
1344            # just send it over completely at once so we have a fixed content length
1345            data = bytearray()
1346            announcement_stream = self.get_announcement_stream(announce_data, audio_format)
1347            # aclosing guarantees the stream (and thus the ffmpeg process chain behind
1348            # it) is torn down immediately when the request is cancelled, instead of
1349            # lingering until garbage collection finalizes the abandoned generator.
1350            async with aclosing(announcement_stream):
1351                async for chunk in announcement_stream:
1352                    data += chunk
1353            return web.Response(
1354                body=bytes(data),
1355                content_type=get_mime_type(audio_format.output_format_str),
1356                headers=DEFAULT_STREAM_HEADERS,
1357            )
1358
1359        resp = web.StreamResponse(status=200, reason="OK", headers=DEFAULT_STREAM_HEADERS)
1360        resp.content_type = get_mime_type(audio_format.output_format_str)
1361        if http_profile == "chunked":
1362            resp.enable_chunked_encoding()
1363
1364        await resp.prepare(request)
1365
1366        # all checks passed, start streaming!
1367        self.logger.debug(
1368            "Start serving audio stream for Announcement %s to %s",
1369            announce_data["announcement_url"],
1370            player.display_name,
1371        )
1372        announcement_stream = self.get_announcement_stream(announce_data, audio_format)
1373        # aclosing guarantees the stream (and thus the ffmpeg process chain behind
1374        # it) is torn down immediately when the player disconnects mid-stream,
1375        # instead of lingering until garbage collection finalizes the abandoned
1376        # generator.
1377        async with aclosing(announcement_stream):
1378            async for chunk in announcement_stream:
1379                try:
1380                    await resp.write(chunk)
1381                except BrokenPipeError, ConnectionResetError:
1382                    break
1383
1384        self.logger.debug(
1385            "Finished serving audio stream for Announcement %s to %s",
1386            announce_data["announcement_url"],
1387            player.display_name,
1388        )
1389
1390        return resp
1391
1392    def get_command_url(self, player_or_queue_id: str, command: str) -> str | None:
1393        """
1394        Get the url for the special command stream, or None if the queue is not playing.
1395
1396        :param player_or_queue_id: Queue (or player) to send the command to.
1397        :param command: Command the url triggers when fetched.
1398        """
1399        # resolve to the active queue: a protocol player (e.g. the cast child of a
1400        # universal player) does not own the active queue, its parent player does
1401        if active_queue := self.mass.player_queues.get_active_queue(player_or_queue_id):
1402            queue_id = active_queue.queue_id
1403        else:
1404            queue_id = player_or_queue_id
1405        queue_data = self.mass.player_queues.queue_data_or_none(queue_id)
1406        if queue_data is None or (session_id := queue_data.session_id) is None:
1407            return None
1408        return f"{self.base_url}/command/{session_id}/{queue_id}/{command}.mp3"
1409
1410    def get_announcement_url(
1411        self,
1412        player_id: str,
1413        content_type: ContentType = ContentType.MP3,
1414    ) -> str:
1415        """
1416        Get the url that serves the announcement registered for the given player.
1417
1418        :param player_id: The player the announcement is played on.
1419        :param content_type: The format to serve the announcement in.
1420        """
1421        # use stream server to host announcement on local network
1422        # this ensures playback on all players, including ones that do not
1423        # like https hosts and it also offers the pre-announce 'bell'
1424        return f"{self.base_url}/announcement/{player_id}.{content_type.value}"
1425
1426    def get_stream(
1427        self,
1428        media: PlayerMedia,
1429        pcm_format: AudioFormat,
1430        player_id: str | None = None,
1431        force_flow_mode: bool = False,
1432    ) -> AsyncGenerator[bytes]:
1433        """
1434        Get a stream of the given media as raw PCM audio.
1435
1436        This is used as helper for player providers that can consume the raw PCM
1437        audio stream directly (e.g. AirPlay) and not rely on HTTP transport.
1438
1439        :param media: The PlayerMedia to stream.
1440        :param pcm_format: The desired output PCM format.
1441        :param player_id: The player ID requesting the stream. Used to determine
1442            if flow mode should be used based on the player's capabilities.
1443        :param force_flow_mode: Force flow mode regardless of player capabilities.
1444            Used for multi-client streaming scenarios that require continuous streams.
1445        """
1446        # select audio source
1447        if media.media_type == MediaType.ANNOUNCEMENT:
1448            # special case: stream announcement
1449            assert media.custom_data
1450            return self.get_announcement_stream(cast("AnnounceData", media.custom_data), pcm_format)
1451        if (
1452            media.source_id
1453            and media.source_id.startswith(UGP_PREFIX)
1454            and media.uri
1455            and "/ugp/" in media.uri
1456        ):
1457            # special case: member player accessing UGP stream
1458            # Check URI to distinguish from the UGP accessing its own stream
1459            ugp_player = cast("UniversalGroupPlayer", self.mass.players.get_player(media.source_id))
1460            ugp_stream = ugp_player.stream
1461            assert ugp_stream is not None  # for type checker
1462            if ugp_stream.base_pcm_format == pcm_format:
1463                # no conversion needed
1464                return ugp_stream.subscribe_raw()
1465            return ugp_stream.get_stream(output_format=pcm_format)
1466        if (
1467            media.media_type == MediaType.AUDIO_SOURCE
1468            and not media.queue_item_id
1469            and media.source_id
1470            and (session := self.mass.players.get_audio_source_session(media.source_id))
1471        ):
1472            # a live source playing on a player rather than an item in a queue
1473            if media.queue_session_id != session.playback_session_id:
1474                # a stale request from a superseded session must not attach to the one
1475                # playing now; the http route rejects the same mismatch with a 404
1476                raise AudioError(
1477                    f"Unknown (or invalid) audio source session: {media.queue_session_id}"
1478                )
1479            return self._count_as_output_stream(
1480                self._get_audio_source_session_stream(
1481                    session, pcm_format, player_id or media.source_id
1482                )
1483            )
1484        if media.source_id and media.queue_item_id:
1485            # Queue stream request - determine flow_mode based on player capabilities
1486            # or force it if explicitly requested (e.g., for multi-client streaming)
1487            protocol_player = self.mass.players.get_player(player_id) if player_id else None
1488            queue_id = media.source_id
1489            queue = self.mass.player_queues.get(queue_id)
1490            queue_session_id = media.queue_session_id
1491            crossfade_needs_flow_mode = (
1492                # crossfade only applies to tracks; if the queue has it enabled but the
1493                # player(protocol) does not support gapless playback, we need to enforce flow mode
1494                media.media_type == MediaType.TRACK
1495                and queue is not None
1496                and queue.crossfade_enabled
1497                and protocol_player
1498                and not protocol_player.supports_gapless
1499            )
1500            # the audio overlay is mixed into the queue's continuous (flow) stream;
1501            # per-item requests would restart the overlay at every track boundary
1502            overlay_needs_flow_mode = queue is not None and overlay_active(queue)
1503            flow_mode = (
1504                force_flow_mode
1505                or (protocol_player is not None and protocol_player.flow_mode)
1506                or crossfade_needs_flow_mode
1507                or overlay_needs_flow_mode
1508            )
1509            if media.media_type in (MediaType.RADIO, MediaType.AUDIO_SOURCE):
1510                # flow_mode for live/infinite streams is pointless
1511                flow_mode = False
1512            if flow_mode:
1513                # flow stream request
1514                assert queue
1515                start_queue_item = self.mass.player_queues.get_item(
1516                    media.source_id, media.queue_item_id
1517                )
1518                assert start_queue_item
1519                self._update_audio_processing_context(
1520                    queue=queue,
1521                    queue_item=start_queue_item,
1522                    pcm_format=pcm_format,
1523                    overlay_enabled=overlay_active(queue),
1524                    session_id=queue_session_id,
1525                )
1526                flow_stream = self.audio.get_queue_flow_stream(
1527                    queue=queue,
1528                    start_queue_item=start_queue_item,
1529                    pcm_format=pcm_format,
1530                    session_id=queue_session_id,
1531                    protocol_player=protocol_player,
1532                )
1533                if overlay_active(queue):
1534                    flow_stream = self.audio.get_overlay_mixed_stream(
1535                        queue, flow_stream, pcm_format
1536                    )
1537                return self._count_as_output_stream(flow_stream)
1538            # single item stream (e.g. radio or non-flow mode)
1539            queue_item = self.mass.player_queues.get_item(media.source_id, media.queue_item_id)
1540            assert queue_item
1541            if queue is not None:
1542                self._update_audio_processing_context(
1543                    queue=queue,
1544                    queue_item=queue_item,
1545                    pcm_format=pcm_format,
1546                    overlay_enabled=(
1547                        queue_item.media_type == MediaType.RADIO and overlay_active(queue)
1548                    ),
1549                    session_id=queue_session_id,
1550                )
1551            inner_stream = self.audio.get_queue_item_stream(
1552                queue_item=queue_item,
1553                pcm_format=pcm_format,
1554                seek_position=(
1555                    int(queue_item.streamdetails.seek_position) if queue_item.streamdetails else 0
1556                ),
1557                playback_speed=cast(
1558                    "float", queue_item.extra_attributes.get("playback_speed", 1.0)
1559                ),
1560                session_id=queue_session_id,
1561            )
1562            if (
1563                queue is not None
1564                and queue_item.media_type == MediaType.RADIO
1565                and overlay_active(queue)
1566            ):
1567                # radio plays as a single long-lived stream, so mix the overlay in here
1568                inner_stream = self.audio.get_overlay_mixed_stream(queue, inner_stream, pcm_format)
1569            # mirror the on_source_selected/unselected lifecycle the HTTP route
1570            # fires, so direct-PCM consumers (AirPlay, Snapcast, UGP) honour the
1571            # plugin contract too
1572            if (
1573                queue_item.media_item is not None
1574                and queue_item.media_item.media_type == MediaType.AUDIO_SOURCE
1575            ):
1576                inner_stream = self._wrap_with_audio_source_lifecycle(
1577                    inner=inner_stream,
1578                    queue_item=queue_item,
1579                    player_id=player_id or media.source_id,
1580                )
1581            return self._count_as_output_stream(inner_stream)
1582        # assume url or some other direct path
1583        # NOTE: this will fail if its an uri not playable by ffmpeg
1584        return get_ffmpeg_stream(
1585            audio_input=media.uri,
1586            input_format=AudioFormat(content_type=ContentType.try_parse(media.uri)),
1587            output_format=pcm_format,
1588        )
1589
1590    async def get_preview_stream(
1591        self,
1592        provider_instance_id_or_domain: str,
1593        item_id: str,
1594        media_type: MediaType = MediaType.TRACK,
1595    ) -> AsyncGenerator[bytes]:
1596        """Create a 30 seconds preview audioclip for the given media item."""
1597        if not (music_prov := self.mass.get_provider(provider_instance_id_or_domain)):
1598            raise ProviderUnavailableError
1599        if music_prov.type != ProviderType.MUSIC:
1600            msg = f"{provider_instance_id_or_domain} is not a music provider"
1601            raise InvalidDataError(msg)
1602        music_prov = cast("MusicProvider", music_prov)
1603
1604        try:
1605            await self.mass.music.get_item(
1606                media_type,
1607                item_id,
1608                provider_instance_id_or_domain,
1609                allow_update_metadata=False,
1610            )
1611        except MediaNotFoundError as err:
1612            msg = f"Item {item_id} not found in provider {provider_instance_id_or_domain}"
1613            raise InvalidDataError(msg) from err
1614
1615        streamdetails = await music_prov.get_stream_details(item_id, media_type)
1616        pcm_format = AudioFormat(
1617            content_type=ContentType.from_bit_depth(streamdetails.audio_format.bit_depth),
1618            sample_rate=streamdetails.audio_format.sample_rate,
1619            bit_depth=streamdetails.audio_format.bit_depth,
1620            channels=streamdetails.audio_format.channels,
1621        )
1622        async for chunk in get_ffmpeg_stream(
1623            audio_input=self.audio.get_media_stream(
1624                streamdetails=streamdetails, pcm_format=pcm_format
1625            ),
1626            input_format=pcm_format,
1627            output_format=AudioFormat(content_type=ContentType.AAC),
1628            extra_input_args=["-t", "30"],
1629        ):
1630            yield chunk
1631
1632    async def get_announcement_stream(
1633        self, announce_data: AnnounceData, output_format: AudioFormat
1634    ) -> AsyncGenerator[bytes]:
1635        """
1636        Get the audio of an announcement (pre-announce chime + announcement).
1637
1638        Any number of consumers may stream the same announcement at once; its source is
1639        fetched and decoded only once. The audio stays available while the stream is
1640        held open.
1641
1642        :param announce_data: The announcement to stream.
1643        :param output_format: The format to deliver the audio in.
1644        """
1645        render = self.announcement_renderer.acquire(announce_data)
1646        try:
1647            # aclosing guarantees this consumer's ffmpeg encoder is torn down
1648            # immediately when it goes away, instead of lingering until garbage
1649            # collection finalizes the abandoned generator.
1650            stream = render.get_stream(output_format)
1651            async with aclosing(stream):
1652                async for chunk in stream:
1653                    yield chunk
1654        finally:
1655            await self.announcement_renderer.release(render)
1656
1657    async def get_announcement_duration(
1658        self, announcement: PlayerMedia, timeout: float = DEFAULT_RENDER_TIMEOUT
1659    ) -> int | None:
1660        """
1661        Get the exact duration (in seconds) of an announcement, once it finished rendering.
1662
1663        Waits for the audio to be rendered in full, so call this while the announcement
1664        plays rather than before handing it to a player. Returns None when the length can
1665        not be determined, e.g. the announcement is no longer playing or its source did
1666        not deliver in time.
1667
1668        :param announcement: The announcement to return the duration for.
1669        :param timeout: Maximum time to wait for the audio to finish rendering.
1670        """
1671        if announcement.duration:
1672            return announcement.duration
1673        if not announcement.custom_data:
1674            return None
1675        render = self.announcement_renderer.get(cast("AnnounceData", announcement.custom_data))
1676        if render is None:
1677            return None
1678        duration = await render.wait_finished(timeout)
1679        return ceil(duration) if duration else None
1680
1681    def _resolve_audio_source_request(
1682        self, request: web.Request
1683    ) -> tuple[AudioSourceSession, Player, PluginProvider]:
1684        """
1685        Resolve a source stream request to its session, consuming player and plugin.
1686
1687        :param request: The stream request to resolve.
1688        :raises web.HTTPNotFound: When any of the three is gone, or the url names a
1689            session that is no longer the one playing.
1690        """
1691        source_player_id = request.match_info["source_player_id"]
1692        player_id = request.match_info["player_id"]
1693        session_id = request.match_info["session_id"]
1694        session = self.mass.players.get_audio_source_session(source_player_id)
1695        if session is None:
1696            raise web.HTTPNotFound(reason=f"No audio source playing on {source_player_id}")
1697        if session_id != session.playback_session_id:
1698            raise web.HTTPNotFound(reason=f"Unknown (or invalid) session: {session_id}")
1699        if not (player := self.mass.players.get_player(player_id)):
1700            raise web.HTTPNotFound(reason=f"Unknown Player: {player_id}")
1701        prov = self.mass.get_provider(session.provider_instance_id)
1702        if not isinstance(prov, PluginProvider):
1703            raise web.HTTPNotFound(
1704                reason=f"AudioSource provider {session.provider_instance_id} unavailable"
1705            )
1706        return session, player, prov
1707
1708    async def _release_unstarted_audio_source(
1709        self, session: AudioSourceSession, playback_session_id: str
1710    ) -> None:
1711        """
1712        Take a source that never started off the player holding it.
1713
1714        The command that pointed the renderer here has already returned, so nothing
1715        else will clear the session: without this the player goes on publishing a
1716        source that never played, with its own queue held inactive behind it.
1717
1718        :param session: The session whose stream failed before any audio flowed.
1719        :param playback_session_id: Playback session active when stream setup started.
1720        """
1721        current_session = self.mass.players.get_audio_source_session(session.player_id)
1722        if (
1723            current_session is not session
1724            or current_session.playback_session_id != playback_session_id
1725        ):
1726            # already superseded, so it is not ours to release
1727            return
1728        self.logger.debug(
1729            "AudioSource %s never started on player %s, releasing it",
1730            session.source_id,
1731            session.player_id,
1732        )
1733        try:
1734            await self.mass.players.deselect_source(
1735                session.player_id,
1736                provider_instance_id=session.provider_instance_id,
1737                source_id=session.source_id,
1738                playback_session_id=playback_session_id,
1739            )
1740        except Exception:
1741            # deselect_source already absorbs the expected stop failures, so anything
1742            # arriving here is a defect worth a trail rather than a silent half-cleanup
1743            self.logger.warning(
1744                "Failed to release AudioSource %s on player %s",
1745                session.source_id,
1746                session.player_id,
1747                exc_info=True,
1748            )
1749
1750    async def _serve_audio_source_head(
1751        self, request: web.Request, session: AudioSourceSession
1752    ) -> web.StreamResponse:
1753        """
1754        Answer a HEAD probe for a live audio source without touching the plugin.
1755
1756        :param request: The probe to answer.
1757        :param session: The session whose source is being probed.
1758        """
1759        head_fmt = request.match_info["fmt"]
1760        if ContentType.try_parse(head_fmt).is_pcm():
1761            # most DLNA renderers pick a decoder from the HEAD content type and
1762            # cannot handle raw PCM
1763            head_fmt = ContentType.WAV.value
1764        resp = web.StreamResponse(
1765            status=200, reason="OK", headers=_audio_source_headers(session, head_fmt)
1766        )
1767        await resp.prepare(request)
1768        return resp
1769
1770    async def _prepare_audio_source_stream(
1771        self,
1772        request: web.Request,
1773        player: Player,
1774        session: AudioSourceSession,
1775        streamdetails: StreamDetails,
1776        provider: PluginProvider,
1777    ) -> tuple[web.StreamResponse, AsyncGenerator[bytes]]:
1778        """
1779        Open the response for a live audio source and build the audio behind it.
1780
1781        :param request: The stream request being answered.
1782        :param player: The player consuming this stream.
1783        :param session: The session whose source is being streamed.
1784        :param streamdetails: The stream details resolved for that source.
1785        :param provider: Plugin delivering the live source.
1786        :return: The prepared response and the encoded audio to write to it.
1787        """
1788        pcm_format = await self.audio.select_pcm_format(
1789            player=player,
1790            streamdetails=streamdetails,
1791            crossfade_enabled=False,
1792            overlay_active=False,
1793        )
1794        output_format = await self.audio.get_output_format(
1795            output_format_str=request.match_info["fmt"],
1796            player=player,
1797            content_sample_rate=pcm_format.sample_rate,
1798            content_bit_depth=pcm_format.bit_depth,
1799            media_type=MediaType.AUDIO_SOURCE,
1800        )
1801        resp = web.StreamResponse(
1802            status=200,
1803            reason="OK",
1804            headers=_audio_source_headers(session, output_format.output_format_str),
1805        )
1806        resp.content_type = get_mime_type(output_format.output_format_str)
1807        http_profile = player.get_config_value(CONF_HTTP_PROFILE, "default")
1808        if http_profile == "forced_content_length":
1809            # a live source has no length, so advertise one it will never reach
1810            resp.content_length = calculate_content_length(output_format, 12 * 3600)
1811        elif http_profile == "chunked":
1812            resp.enable_chunked_encoding()
1813        await resp.prepare(request)
1814
1815        audio_input = self.audio.get_audio_source_stream(
1816            streamdetails=streamdetails,
1817            pcm_format=pcm_format,
1818            raise_on_error=False,
1819            display_name=session.source.name,
1820        )
1821        filter_params = self.audio.get_player_output_plan(
1822            player_id=player.player_id,
1823            input_format=pcm_format,
1824            output_format=output_format,
1825            shared_player_ids=player.state.group_members,
1826            queue_id=session.player_id,
1827            session_id=session.playback_session_id,
1828        ).filter_params
1829        self._update_audio_source_processing_context(session, provider)
1830        if (
1831            output_format.content_type == ContentType.WAV
1832            and not filter_params
1833            and output_format.sample_rate == pcm_format.sample_rate
1834            and output_format.bit_depth == pcm_format.bit_depth
1835            and output_format.channels == pcm_format.channels
1836        ):
1837            # the player takes the source's exact PCM, so skip the encode ffmpeg and
1838            # its buffer latency and send a WAV header with the raw bytes
1839            return resp, _wav_passthrough_stream(audio_input, output_format)
1840        return resp, get_ffmpeg_stream(
1841            audio_input=audio_input,
1842            input_format=pcm_format,
1843            output_format=output_format,
1844            filter_params=filter_params,
1845            # keep the encode stage from reading further ahead than it needs to: a live
1846            # source's latency is whatever is buffered between it and the player
1847            extra_input_args=[
1848                "-readrate",
1849                SINGLE_ITEM_READRATE,
1850                "-readrate_initial_burst",
1851                SINGLE_ITEM_READRATE_INITIAL_BURST,
1852            ],
1853        )
1854
1855    async def _get_audio_source_session_stream(
1856        self,
1857        session: AudioSourceSession,
1858        pcm_format: AudioFormat,
1859        consumer_player_id: str,
1860    ) -> AsyncGenerator[bytes]:
1861        """
1862        Stream a live source to a consumer that takes raw PCM rather than the http url.
1863
1864        AirPlay, Snapcast, squeezelite's multi-client path, universal groups and the
1865        MSX bridge all consume PCM directly, so they never reach the http route and
1866        need the plugin lifecycle fired here instead — those hooks are what claim and
1867        release the source and kick acquisition side effects into life.
1868
1869        :param session: The live source session playing on its owner.
1870        :param pcm_format: The PCM format the consumer wants.
1871        :param consumer_player_id: The player consuming this stream, which is not
1872            necessarily the one that owns the source.
1873        """
1874        prov = self.mass.get_provider(session.provider_instance_id)
1875        if not isinstance(prov, PluginProvider):
1876            raise AudioError(
1877                f"AudioSource provider {session.provider_instance_id} is not available"
1878            )
1879        playback_session_id = session.playback_session_id
1880        stream_session_id = uuid4().hex
1881        serving = False
1882        try:
1883            try:
1884                await prov.on_source_selected(
1885                    session.source_id,
1886                    consumer_player_id,
1887                    session.player_id,
1888                    stream_session_id,
1889                )
1890            except RuntimeError as err:
1891                # the plugin refuses this consumer, e.g. it just redirected playback
1892                raise AudioError(str(err)) from err
1893            if (
1894                self.mass.players.get_audio_source_session(session.player_id) is not session
1895                or session.playback_session_id != playback_session_id
1896            ):
1897                raise AudioError("AudioSource session was superseded")
1898            session.stream_session_id = stream_session_id
1899            if (streamdetails := session.streamdetails) is None:
1900                streamdetails = await prov.get_stream_details(
1901                    session.source_id, MediaType.AUDIO_SOURCE
1902                )
1903                session.attach_streamdetails(streamdetails)
1904            self._update_audio_source_processing_context(session, prov)
1905            serving = True
1906            async for chunk in self.audio.get_audio_source_stream(
1907                streamdetails=streamdetails,
1908                pcm_format=pcm_format,
1909                raise_on_error=False,
1910                display_name=session.source.name,
1911            ):
1912                if (
1913                    self.mass.players.get_audio_source_session(session.player_id) is not session
1914                    or session.playback_session_id != playback_session_id
1915                    or session.stream_session_id != stream_session_id
1916                ):
1917                    break
1918                yield chunk
1919        finally:
1920            try:
1921                await prov.on_source_unselected(
1922                    session.source_id, session.player_id, stream_session_id
1923                )
1924            except Exception:
1925                self.logger.warning(
1926                    "on_source_unselected raised for provider %s source %s player %s",
1927                    prov.instance_id,
1928                    session.source_id,
1929                    session.player_id,
1930                    exc_info=True,
1931                )
1932            if not serving:
1933                await self._release_unstarted_audio_source(session, playback_session_id)
1934
1935    async def _wrap_with_audio_source_lifecycle(
1936        self,
1937        inner: AsyncGenerator[bytes],
1938        queue_item: QueueItem,
1939        player_id: str,
1940    ) -> AsyncGenerator[bytes]:
1941        """
1942        Wrap an AudioSource queue item stream with on_source_selected/unselected hooks.
1943
1944        Direct-PCM consumers (AirPlay, Snapcast, UGP, ...) call ``get_stream`` instead
1945        of going through the HTTP route, but the plugin contract requires the
1946        lifecycle hooks to fire for every actual stream request — they're what
1947        claim/release the per-queue exclusive ownership and trigger acquisition
1948        side effects like the Spotify Connect Web API play kick. This wrapper
1949        gives those consumers the same lifecycle the HTTP route already provides.
1950
1951        :param inner: The underlying audio stream generator.
1952        :param queue_item: The AudioSource queue item being streamed.
1953        :param player_id: The protocol player consuming this stream.
1954        """
1955        media_item = queue_item.media_item
1956        assert media_item is not None  # caller checked media_type == AUDIO_SOURCE
1957        prov = self.mass.get_provider(media_item.provider)
1958        queue_id = queue_item.queue_id
1959        if not isinstance(prov, PluginProvider):
1960            async for chunk in inner:
1961                yield chunk
1962            return
1963        source_id = media_item.item_id
1964        stream_session_id = uuid4().hex
1965        # single try/finally so on_source_unselected fires even when
1966        # on_source_selected raises after partially claiming state; the
1967        # provider's session_id guard makes a no-op claim release safe.
1968        try:
1969            try:
1970                await prov.on_source_selected(source_id, player_id, queue_id, stream_session_id)
1971            except RuntimeError as err:
1972                # provider intentionally aborts the request — surface as AudioError
1973                raise AudioError(str(err)) from err
1974            async for chunk in inner:
1975                yield chunk
1976        finally:
1977            try:
1978                await prov.on_source_unselected(source_id, queue_id, stream_session_id)
1979            except Exception:
1980                self.logger.exception(
1981                    "on_source_unselected raised for provider %s source %s queue %s",
1982                    prov.instance_id,
1983                    source_id,
1984                    queue_id,
1985                )
1986
1987    async def _count_as_output_stream(self, inner: AsyncGenerator[bytes]) -> AsyncGenerator[bytes]:
1988        """
1989        Forward a queue stream while it counts towards the active-output-stream gauge.
1990
1991        Direct-PCM consumers (AirPlay, Snapcast, Sendspin, Squeezelite, UGP, ...) call
1992        ``get_stream`` instead of going through the HTTP route, so without this they never
1993        register as playing and audio analysis keeps its idle CPU budget while they stream.
1994
1995        :param inner: The queue (flow or single item) stream to forward.
1996        """
1997        self._active_output_streams += 1
1998        try:
1999            # aclosing guarantees the generator (and thus the ffmpeg process chain behind
2000            # it) is torn down when the consumer stops iterating; an async for does not
2001            # close its iterator on its own.
2002            async with aclosing(inner):
2003                async for chunk in inner:
2004                    yield chunk
2005        finally:
2006            self._active_output_streams -= 1
2007
2008    def _served_by(self, queue_item: QueueItem | None, provider_instance: str) -> bool:
2009        """
2010        Return whether a queue item is a track the given provider instance serves.
2011
2012        :param queue_item: Queue item to check, or None when there is none.
2013        :param provider_instance: Instance id of the provider to match.
2014        """
2015        if queue_item is None or queue_item.media_type != MediaType.TRACK:
2016            return False
2017        if (streamdetails := queue_item.streamdetails) is not None:
2018            # already resolved, so this is the provider that will really serve it
2019            return streamdetails.provider == provider_instance
2020        if (media_item := queue_item.media_item) is None:
2021            return False
2022        return media_item.provider == provider_instance or any(
2023            mapping.provider_instance == provider_instance
2024            for mapping in media_item.provider_mappings
2025        )
2026
2027    def _update_audio_processing_context(
2028        self,
2029        queue: PlayerQueue,
2030        queue_item: QueueItem,
2031        pcm_format: AudioFormat,
2032        overlay_enabled: bool,
2033        session_id: str | None = None,
2034    ) -> None:
2035        """
2036        Store the shared processing context selected for a queue item.
2037
2038        Our own crossfade is left out on purpose: only the audio layer knows whether
2039        one really happens, and it reports that itself once the boundary has decided.
2040        A crossfade the source performs is the exception - the audio layer never sees
2041        that one, so it is carried here.
2042
2043        :param queue: Active player queue.
2044        :param queue_item: Queue item being prepared.
2045        :param pcm_format: Shared PCM format leaving queue processing.
2046        :param overlay_enabled: Whether an overlay is mixed into this stream.
2047        :param session_id: Queue session that owns processing-detail updates.
2048        """
2049        if queue_item.streamdetails is None:
2050            return
2051        queue_data = self.mass.player_queues.queue_data_or_none(queue.queue_id)
2052        if (
2053            queue_data is None
2054            or (processing_session_id := session_id or queue_data.session_id) is None
2055            or queue_data.session_id != processing_session_id
2056        ):
2057            return
2058        self.audio_processing.start_session(queue.queue_id, processing_session_id)
2059        self.audio_processing.update_item_context(
2060            queue_id=queue.queue_id,
2061            session_id=processing_session_id,
2062            queue_item_id=queue_item.queue_item_id,
2063            queue_processing=AudioQueueProcessing(
2064                pcm_format=pcm_format,
2065                playback_speed=cast(
2066                    "float",
2067                    queue_item.extra_attributes.get("playback_speed", 1.0),
2068                ),
2069                crossfade_mode=CrossfadeMode.DISABLED,
2070                overlay_active=overlay_enabled,
2071            ),
2072            alters_audio=queue_item.streamdetails.fade_in,
2073        )
2074
2075    def _update_audio_source_processing_context(
2076        self,
2077        session: AudioSourceSession,
2078        provider: PluginProvider,
2079    ) -> None:
2080        """
2081        Publish source-owned processing for a live AudioSource.
2082
2083        :param session: Active source session to publish.
2084        :param provider: Plugin delivering the live source.
2085        """
2086        if session.streamdetails is None:
2087            return
2088        self.audio_processing.update_source_context(
2089            session.player_id,
2090            session.playback_session_id,
2091            crossfade_enabled=provider.delivers_crossfaded_audio(session.streamdetails),
2092            volume_normalization_enabled=provider.delivers_normalized_audio(session.streamdetails),
2093        )
2094
2095    def _get_announcement_http_profile(self, player_id: str, announce_data: AnnounceData) -> str:
2096        """
2097        Resolve the http profile for serving an announcement stream.
2098
2099        Announcement urls are registered under the visible player's id, but the
2100        stream may be fetched by a linked protocol player; the profile must come
2101        from the player that actually performs the fetch.
2102        """
2103        announce_player = None
2104        if announce_player_id := announce_data.get("announce_player_id"):
2105            announce_player = self.mass.players.get_player(announce_player_id)
2106        if announce_player is None:
2107            announce_player = self.mass.players.get_player(player_id)
2108        if announce_player is None:
2109            return "default"
2110        return announce_player.get_output_config_value(CONF_HTTP_PROFILE, "default")
2111
2112    async def _finish_flow_stream(
2113        self, resp: web.StreamResponse, queue_id: str, session_id: str
2114    ) -> None:
2115        """
2116        Close a fully served flow stream, giving the player time to drain when it ends the queue.
2117
2118        :param resp: The flow stream response, already fully written.
2119        :param queue_id: Id of the queue the flow stream belongs to.
2120        :param session_id: Stream session this response was opened for.
2121        """
2122        if self.mass.player_queues.flow_queue_exhausted(queue_id, session_id):
2123            # the player is still holding a few seconds of audio it has not rendered yet
2124            # and drops that as soon as the stream ends, so let it play out first.
2125            # a flow that ends to be restarted right away gets no such grace: there the
2126            # player should go idle as soon as possible so the next stream can start.
2127            self.logger.debug(
2128                "Flow stream for queue %s reached the end of the queue - holding the "
2129                "connection open for %ss so the player can play out its buffer",
2130                queue_id,
2131                FLOW_STREAM_LEAD_OUT_SECONDS,
2132            )
2133            await asyncio.sleep(FLOW_STREAM_LEAD_OUT_SECONDS)
2134        # aiohttp derives keep-alive from the request, so the 'Connection: close' we
2135        # advertise is relayed to the player but never applied to the response itself.
2136        # Without this the player is left waiting on a stream that already ended.
2137        resp.force_close()
2138
2139    def _log_request(self, request: web.Request) -> None:
2140        """Log request."""
2141        if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
2142            self.logger.log(
2143                VERBOSE_LOG_LEVEL,
2144                "Got %s request to %s from %s\nheaders: %s\n",
2145                request.method,
2146                request.path,
2147                request.remote,
2148                redact_sensitive_headers(request.headers),
2149            )
2150        else:
2151            self.logger.debug(
2152                "Got %s request to %s from %s (HTTP/%s.%s, connection: %s)",
2153                request.method,
2154                request.path,
2155                request.remote,
2156                request.version.major,
2157                request.version.minor,
2158                request.headers.get("Connection", "-"),
2159            )
2160
2161    async def _reload_network_dependent_providers(self) -> None:
2162        """Reload the providers that captured the streamserver network, if it changed."""
2163        previous = self._network_fingerprint
2164        current = (
2165            self._bind_ip,
2166            str(self.publish_ip),
2167            cast("int", self.publish_port),
2168            tuple(self._publish_addresses),
2169        )
2170        if previous is None or previous == current:
2171            self._network_fingerprint = current
2172            return
2173        # these providers bind or advertise the network while they load, so a plain
2174        # reload is what moves them over - they share no lighter rebind path
2175        instance_ids = [
2176            prov.instance_id
2177            for prov in self.mass.providers
2178            if prov.reload_on_streams_network_change
2179        ]
2180        for instance_id in instance_ids:
2181            try:
2182                config = await self.mass.config.get_provider_config(instance_id)
2183                self.logger.info(
2184                    "Streamserver network changed, reloading provider %s",
2185                    config.name or config.domain,
2186                )
2187                await self.mass.load_provider_config(config)
2188            except Exception as err:
2189                self.logger.warning(
2190                    "Error reloading provider %s: %s",
2191                    instance_id,
2192                    str(err) or err.__class__.__name__,
2193                    exc_info=err,
2194                )
2195        # only mark the new network as applied once the loop completed, so a run cut short
2196        # by a second config change runs again on the next reload
2197        self._network_fingerprint = current
2198
2199    def _setup_smart_fades_logger(self, config: CoreConfig) -> None:
2200        """Set up smart fades logger level."""
2201        log_level = str(config.get_value(CONF_SMART_FADES_LOG_LEVEL))
2202        if log_level == "GLOBAL":
2203            self.audio.smart_fades_mixer.logger.setLevel(self.logger.level)
2204        else:
2205            self.audio.smart_fades_mixer.logger.setLevel(log_level)
2206
2207    def _resolve_publish_state(self, bind_ip: str, publish_candidates: tuple[str, ...]) -> None:
2208        """
2209        Resolve the addresses and base URL to advertise for the given bind address.
2210
2211        Reads ``self.publish_port``, so set that first.
2212
2213        :param bind_ip: Address the streamserver binds to (a wildcard means all interfaces).
2214        :param publish_candidates: Host addresses reachable from the local network, ranked.
2215        """
2216        self._bind_ip = bind_ip
2217        self._publish_addresses = _get_publish_addresses(
2218            bind_ip, self._configured_publish_ip, publish_candidates
2219        )
2220        # the single address players are handed, taken from the top of the ranked list
2221        self.publish_ip = self._publish_addresses[0]
2222        self._base_url = f"http://{format_ip_for_url(self.publish_ip)}:{self.publish_port}"
2223
2224
2225def _same_ip_family(ip: str, other_ip: str) -> bool:
2226    """Return whether two addresses belong to the same IP family."""
2227    return (":" in ip) == (":" in other_ip)
2228