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