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