music-assistant-server

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