music-assistant-server

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