music-assistant-server

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