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