music-assistant-server

1"""
2Audio streaming helpers that interact with core controllers and providers.
3
4This module contains all audio stream acquisition and processing functions
5that need access to the MusicAssistant instance. Generic audio utilities
6that do not need controller interaction live in helpers/audio.py.
7"""
8
9from __future__ import annotations
10
11import asyncio
12import logging
13import os
14import re
15import time
16from collections.abc import AsyncGenerator, Iterable
17from contextlib import asynccontextmanager
18from dataclasses import dataclass
19from functools import partial
20from typing import TYPE_CHECKING, Any, cast
21from urllib.parse import urlparse
22
23import aiofiles
24import aiofiles.os
25import aiohttp
26import shortuuid
27from aiohttp import ClientConnectorSSLError, ClientResponseError, ClientTimeout
28from music_assistant_models.audio_processing import (
29    AudioDSPDetails,
30    AudioOutputDetails,
31    AudioQueueProcessing,
32)
33from music_assistant_models.dsp import (
34    AudioChannel,
35    ConvolutionFilter,
36    DSPConfig,
37    DSPFilter,
38    DSPState,
39)
40from music_assistant_models.enums import (
41    ContentType,
42    CrossfadeMode,
43    MediaType,
44    PlayerFeature,
45    ProviderType,
46    StreamType,
47    VolumeNormalizationMode,
48)
49from music_assistant_models.errors import (
50    AudioError,
51    InvalidDataError,
52    MediaNotFoundError,
53    MusicAssistantError,
54    ProviderPermissionDenied,
55    ProviderUnavailableError,
56    QueueEmpty,
57    RetriesExhausted,
58)
59from music_assistant_models.media_items import AudioFormat, Track
60from music_assistant_models.player_queue import PlayLogEntry
61from music_assistant_models.streamdetails import MultiPartPath, StreamMetadata
62
63from music_assistant.constants import (
64    CONF_CROSSFADE_DURATION,
65    CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES,
66    CONF_ENTRY_VOLUME_NORMALIZATION_TARGET,
67    CONF_FLOW_MODE_SAMPLE_RATE,
68    CONF_OUTPUT_CHANNELS,
69    CONF_PLAYER_QUEUES,
70    CONF_VALUE_DISABLED,
71    CONF_VALUE_ENABLED,
72    CONF_VOLUME_NORMALIZATION,
73    CONF_VOLUME_NORMALIZATION_FIXED_GAIN_RADIO,
74    CONF_VOLUME_NORMALIZATION_FIXED_GAIN_TRACKS,
75    CONF_VOLUME_NORMALIZATION_RADIO,
76    CONF_VOLUME_NORMALIZATION_TARGET,
77    CONF_VOLUME_NORMALIZATION_TRACKS,
78    DSP_IRS_DIRNAME,
79    FLOW_MODE_SAMPLE_RATE_48000,
80    FLOW_MODE_SAMPLE_RATE_96000,
81    FLOW_MODE_SAMPLE_RATE_BIT_PERFECT,
82    FLOW_MODE_SAMPLE_RATE_HIGHEST,
83    FLOW_MODE_SAMPLE_RATE_SMART,
84    INTERNAL_PCM_FORMAT,
85    MASS_LOGGER_NAME,
86    STREAM_STALL_TIMEOUT,
87    STREAM_START_TIMEOUT,
88    VERBOSE_LOG_LEVEL,
89)
90from music_assistant.controllers.streams.audio_analysis import (
91    LOUDNESS_ANALYSIS_DOMAIN,
92)
93from music_assistant.controllers.streams.audio_buffer import AudioBuffer
94from music_assistant.controllers.streams.audio_processing import (
95    AudioOutputPlan,
96    get_normalization_details,
97)
98from music_assistant.controllers.streams.constants import (
99    CACHE_CATEGORY_RESOLVED_RADIO_URL,
100    CACHE_PROVIDER,
101    CONF_ALLOW_CROSSFADE_SAME_ALBUM,
102)
103from music_assistant.controllers.streams.ogg_handler import get_chained_ogg_stream
104from music_assistant.controllers.streams.smart_fades import SmartFadesMixer
105from music_assistant.controllers.streams.smart_fades.fades import SmartFade
106from music_assistant.controllers.streams.smart_fades.helpers import SMART_CROSSFADE_DURATION
107from music_assistant.helpers import ssl as ssl_util
108from music_assistant.helpers.aiohttp_client import encoded_request_url
109from music_assistant.helpers.audio import (
110    HTTP_HEADERS,
111    HTTP_HEADERS_ICY,
112    build_concat_filelist,
113    calculate_content_length,
114    get_bit_rate,
115    get_normalization_mode,
116    get_parts_from_position,
117    is_grouping_preventing_dsp,
118    iter_pcm_slices,
119    parse_extinf_metadata,
120    realtime_pcm_pacer,
121    resample_pcm_audio,
122    resolve_output_player_ids,
123)
124from music_assistant.helpers.dsp import ComplexFilter, filter_to_ffmpeg_params
125from music_assistant.helpers.ffmpeg import (
126    FFMpeg,
127    get_ffmpeg_overlay_stream,
128    get_ffmpeg_stream,
129)
130from music_assistant.helpers.named_pipe import read_named_pipe
131from music_assistant.helpers.playlists import IsHLSPlaylist, PlaylistItem, fetch_playlist, parse_m3u
132from music_assistant.helpers.throttle_retry import BYPASS_THROTTLER
133from music_assistant.helpers.util import (
134    clean_stream_title,
135    detect_charset,
136    parse_quoted_stream_title,
137    parse_title_and_version,
138    remove_file,
139)
140
141if TYPE_CHECKING:
142    from music_assistant_models.player_queue import PlayerQueue
143    from music_assistant_models.queue_item import QueueItem
144    from music_assistant_models.streamdetails import StreamDetails
145
146    from music_assistant.mass import MusicAssistant
147    from music_assistant.models.music_provider import MusicProvider
148    from music_assistant.models.player import Player
149    from music_assistant.models.plugin import PluginProvider
150
151# ruff: noqa: PLR0915
152
153# Seconds of PCM yielded directly to the player before the crossfade holdback starts buffering.
154WARMUP_DURATION = 8
155
156# Chunk size for the realtime AudioSource path; small enough to keep ffmpeg→consumer
157# latency below ~50 ms while still amortising per-chunk overhead.
158AUDIO_SOURCE_CHUNK_SECONDS = 0.02
159
160# Terminal errors get_icy_radio_stream raises once a single mirror is exhausted; the
161# multi-mirror reader treats these as the signal to fail over to the next URL.
162RADIO_MIRROR_FAILOVER_ERRORS = (
163    MediaNotFoundError,
164    ProviderPermissionDenied,
165    ProviderUnavailableError,
166    RetriesExhausted,
167    InvalidDataError,
168)
169
170
171@dataclass
172class CrossfadeData:
173    """Data class to hold crossfade data."""
174
175    data: bytes
176    fade_in_size: int
177    pcm_format: AudioFormat  # Format of the 'data' bytes (current/previous track's format)
178    fade_in_pcm_format: AudioFormat  # Format for 'fade_in_size' (next track's format)
179    queue_item_id: str
180    # Offset for the fade_in track's elapsed time calculation, to account for crossfade duration and trim
181    elapsed_time_offset: float = 0.0
182    # Normalization mode the intro PCM was baked with, used to pin the next track's body to the same mode
183    normalization_mode: VolumeNormalizationMode | None = None
184
185
186def _snap_supported_rate_up(target: int, supported_sample_rates: list[int]) -> int:
187    """Snap target up, falling back to its highest supported divisor or the maximum."""
188    if target in supported_sample_rates:
189        return target
190    higher = [r for r in supported_sample_rates if r > target]
191    if higher:
192        return min(higher)
193    same_family = [r for r in supported_sample_rates if target % r == 0]
194    return max(same_family) if same_family else max(supported_sample_rates)
195
196
197def _snap_supported_rate_down(target: int, supported_sample_rates: list[int]) -> int:
198    """Snap target down to the highest supported rate <= target, falling back to min."""
199    if target in supported_sample_rates:
200        return target
201    lower = [r for r in supported_sample_rates if r < target]
202    return max(lower) if lower else min(supported_sample_rates)
203
204
205def _pcm_formats_match(a: AudioFormat, b: AudioFormat) -> bool:
206    """Return True if two PCM formats describe identical raw bytes."""
207    return (
208        a.content_type == b.content_type
209        and a.sample_rate == b.sample_rate
210        and a.bit_depth == b.bit_depth
211        and a.channels == b.channels
212    )
213
214
215def overlay_active(queue: PlayerQueue) -> bool:
216    """Return True if the given queue has an audio overlay enabled and a source selected."""
217    return queue.overlay_enabled and queue.overlay_source is not None
218
219
220class StreamsAudio:
221    """Audio stream acquisition and processing for the streams controller."""
222
223    def __init__(self, mass: MusicAssistant) -> None:
224        """
225        Initialize StreamsAudio.
226
227        :param mass: The MusicAssistant instance.
228        """
229        self.mass = mass
230        self.logger = logging.getLogger(f"{MASS_LOGGER_NAME}.streams.audio")
231        self._crossfade_data: dict[str, CrossfadeData] = {}
232        self._smart_fades_mixer: SmartFadesMixer | None = None
233
234    def setup(self) -> None:
235        """Set up the audio sub-controller (called after all core controllers are created)."""
236        self._smart_fades_mixer = SmartFadesMixer(self.mass.streams)
237
238    @property
239    def smart_fades_mixer(self) -> SmartFadesMixer:
240        """Return the smart fades mixer."""
241        assert self._smart_fades_mixer is not None, "StreamsAudio.setup() not called"
242        return self._smart_fades_mixer
243
244    # --- Public methods ---
245
246    async def get_stream_details(
247        self,
248        queue_item: QueueItem,
249        seek_position: int = 0,
250        fade_in: bool = False,
251        prefer_album_loudness: bool = False,
252    ) -> StreamDetails:
253        """
254        Get streamdetails for the given QueueItem.
255
256        This is called just-in-time when a PlayerQueue wants a MediaItem to be played.
257        Do not try to request streamdetails too much in advance as this is expiring data.
258        """
259        mass = self.mass
260        streamdetails: StreamDetails | None = None
261        time_start = time.time()
262        self.logger.debug("Getting streamdetails for %s", queue_item.uri)
263
264        if not queue_item.media_item and not queue_item.streamdetails:
265            # in case of a non-media item queue item, the streamdetails should already be provided
266            # this should not happen, but guard it just in case
267            raise MediaNotFoundError(
268                f"Unable to retrieve streamdetails for {queue_item.name} ({queue_item.uri})"
269            )
270
271        if queue_item.streamdetails and (
272            # reuse if the buffer can serve this seek position (fast seek path)
273            (
274                queue_item.streamdetails.buffer
275                and queue_item.streamdetails.buffer.is_valid(int(seek_position * 1000))
276            )
277            # or reuse if streamdetails hasn't expired yet (new buffer will be created)
278            or (queue_item.streamdetails.created_at + queue_item.streamdetails.expiration)
279            > time.time()
280        ):
281            streamdetails = queue_item.streamdetails
282        else:
283            # need to (re)create streamdetails
284            # retrieve streamdetails from provider
285
286            media_item = queue_item.media_item
287            assert media_item is not None  # for type checking
288            preferred_providers: list[str] = []
289            if (
290                (pq_data := mass.player_queues.queue_data_or_none(queue_item.queue_id))
291                and pq_data.userid
292                and (playback_user := await mass.webserver.auth.get_user(pq_data.userid))
293                and playback_user.provider_filter
294            ):
295                # handle steering into user preferred providerinstance
296                preferred_providers = playback_user.provider_filter
297            else:
298                preferred_providers = [x.provider_instance for x in media_item.provider_mappings]
299            # Remember the last AudioError so we can re-raise its (actionable)
300            # message instead of the generic MediaNotFoundError below.
301            last_audio_error: AudioError | None = None
302            attempted: set[tuple[str, str]] = set()
303            for allow_other_provider in (False, True):
304                if streamdetails:
305                    break
306                # sort by quality and check item's availability
307                for prov_media in sorted(
308                    media_item.provider_mappings, key=lambda x: x.quality or 0, reverse=True
309                ):
310                    if not prov_media.available:
311                        self.logger.debug(f"Skipping unavailable {prov_media}")
312                        continue
313                    if (
314                        not allow_other_provider
315                        and prov_media.provider_instance not in preferred_providers
316                    ):
317                        continue
318                    # the second pass is there to widen to providers the steering held back,
319                    # not to give a mapping a second chance: without a user provider filter
320                    # the first pass already tried them all, so re-attempting one just buys
321                    # the same failure at the cost of another provider round-trip
322                    attempt = (prov_media.provider_instance, prov_media.item_id)
323                    if attempt in attempted:
324                        continue
325                    # guard that provider is available
326                    provider = mass.get_provider(prov_media.provider_instance)
327                    if not provider:
328                        self.logger.debug(f"Skipping {prov_media} - provider not available")
329                        continue  # provider not available ?
330                    attempted.add(attempt)
331                    # get streamdetails from provider; music and plugin providers
332                    # share this signature, so either type can own the item.
333                    try:
334                        BYPASS_THROTTLER.set(True)
335                        stream_prov = cast("MusicProvider | PluginProvider", provider)
336                        streamdetails = await stream_prov.get_stream_details(
337                            prov_media.item_id, media_item.media_type
338                        )
339                    except AudioError as err:
340                        last_audio_error = err
341                        self.logger.warning(str(err))
342                    except MusicAssistantError as err:
343                        self.logger.warning(str(err))
344                    else:
345                        break
346                    finally:
347                        BYPASS_THROTTLER.set(False)
348
349            if not streamdetails:
350                if last_audio_error is not None:
351                    raise last_audio_error
352                msg = f"Unable to retrieve streamdetails for {queue_item.name} ({queue_item.uri})"
353                raise MediaNotFoundError(msg)
354
355            # work out how to handle radio stream
356            if (
357                streamdetails.stream_type in (StreamType.ICY, StreamType.HLS, StreamType.HTTP)
358                and streamdetails.media_type == MediaType.RADIO
359                and isinstance(streamdetails.path, str)
360            ):
361                resolved_url, stream_type = await self.resolve_radio_stream(streamdetails.path)
362                streamdetails.path = resolved_url
363                streamdetails.stream_type = stream_type
364                # Set up metadata monitoring callback for HLS radio streams, if not already set
365                if (
366                    stream_type == StreamType.HLS
367                    and not streamdetails.stream_metadata_update_callback
368                ):
369                    streamdetails.stream_metadata_update_callback = partial(
370                        self._update_hls_radio_metadata
371                    )
372                    streamdetails.stream_metadata_update_interval = 5
373
374        # providers report an unknown duration as either None or 0
375        if not streamdetails.duration:
376            if queue_item.media_item and queue_item.media_item.duration:
377                streamdetails.duration = queue_item.media_item.duration
378            elif queue_item.duration:
379                streamdetails.duration = queue_item.duration
380        if seek_position and not streamdetails.allow_seek:
381            self.logger.warning("seeking is not possible on this stream!")
382            seek_position = 0
383        elif seek_position and not streamdetails.duration:
384            self.logger.warning("seeking is not possible on duration-less streams!")
385            seek_position = 0
386
387        # set queue_id on the streamdetails so we know what is being streamed
388        streamdetails.queue_id = queue_item.queue_id
389        # handle skip/fade_in details
390        streamdetails.seek_position = seek_position
391        streamdetails.fade_in = fade_in
392
393        streamdetails.prefer_album_loudness = prefer_album_loudness
394        conf_volume_normalization_target = float(
395            mass.streams.get_config_value(CONF_VOLUME_NORMALIZATION_TARGET, return_type=int)
396        )
397        # guard against invalid volume normalization values
398        # range and default_value are guaranteed to be set for this constant
399        volume_range = CONF_ENTRY_VOLUME_NORMALIZATION_TARGET.range
400        assert volume_range is not None
401        if (
402            conf_volume_normalization_target < volume_range[0]
403            or conf_volume_normalization_target >= volume_range[1]
404        ):
405            default_val = CONF_ENTRY_VOLUME_NORMALIZATION_TARGET.default_value
406            assert isinstance(default_val, (int, float))
407            conf_volume_normalization_target = float(default_val)
408            self.logger.warning(
409                "Invalid volume normalization target configured, resetting to default of %s LUFS",
410                CONF_ENTRY_VOLUME_NORMALIZATION_TARGET.default_value,
411            )
412        streamdetails.target_loudness = conf_volume_normalization_target
413        volume_normalization_enabled = (
414            mass.config.get_effective_player_queue_config_value(
415                streamdetails.queue_id, CONF_VOLUME_NORMALIZATION, CONF_VALUE_ENABLED
416            )
417            != CONF_VALUE_DISABLED
418        )
419        streamdetails.volume_normalization_mode = get_normalization_mode(
420            self._get_volume_normalization_preference(streamdetails),
421            volume_normalization_enabled,
422            streamdetails,
423        )
424
425        self.logger.debug(
426            "Retrieved streamdetails for %s in %s milliseconds",
427            queue_item.uri,
428            int((time.time() - time_start) * 1000),
429        )
430        return streamdetails
431
432    async def get_media_stream(
433        self,
434        streamdetails: StreamDetails,
435        pcm_format: AudioFormat,
436        seek_position: int = 0,
437        filter_params: list[str] | None = None,
438        chunk_seconds: float = 1.0,
439    ) -> AsyncGenerator[bytes]:
440        """
441        Get audio stream for given media details as raw PCM.
442
443        :param streamdetails: Details of the stream to fetch.
444        :param pcm_format: Target PCM format the consumer expects.
445        :param seek_position: Seek offset in seconds (only honoured when the
446            source allows seeking; ignored for live AudioSources).
447        :param filter_params: Optional ffmpeg filter expressions.
448        :param chunk_seconds: Size of each yielded chunk in seconds of audio.
449            Defaults to 1 s for track-like sources; callers streaming live
450            AudioSources should pass a much smaller value (e.g. 0.02) to keep
451            end-to-end latency low.
452        """
453        mass = self.mass
454        logger = self.logger.getChild("media_stream")
455        logger.log(VERBOSE_LOG_LEVEL, "Starting media stream for %s", streamdetails.uri)
456        # copy: the args below are appended per call, while the StreamDetails is cached on
457        # the queue item and reused across calls (retry, seek, background analysis)
458        extra_input_args = list(streamdetails.extra_input_args or [])
459        # the branches below zero out seek_position where the seek is delegated to the
460        # source itself, so keep the requested position for the duration writeback
461        requested_seek_position = seek_position
462
463        # work out audio source for these streamdetails
464        audio_source: str | AsyncGenerator[bytes]
465        stream_type = streamdetails.stream_type
466        if stream_type == StreamType.CUSTOM:
467            # MusicProvider and PluginProvider both expose get_audio_stream with the same shape
468            provider = mass.get_provider(streamdetails.provider)
469            if provider is None:
470                raise ProviderUnavailableError(
471                    f"Provider {streamdetails.provider} for stream is no longer available"
472                )
473            provider = cast("MusicProvider | PluginProvider", provider)
474            audio_source = provider.get_audio_stream(
475                streamdetails, seek_position=seek_position if streamdetails.can_seek else 0
476            )
477            seek_position = 0 if streamdetails.can_seek else seek_position
478        elif stream_type == StreamType.ICY:
479            assert streamdetails.path is not None
480            assert isinstance(streamdetails.path, (str, list))
481            audio_source = self.get_reconnecting_icy_radio_stream(streamdetails.path, streamdetails)
482            seek_position = 0
483        elif stream_type == StreamType.SHOUTCAST:
484            assert isinstance(streamdetails.path, str)
485            audio_source = self.get_shoutcast_stream(streamdetails.path, streamdetails)
486            seek_position = 0
487        elif stream_type == StreamType.IN_BAND:
488            assert isinstance(streamdetails.path, str)  # for type checking
489
490            # For IN_BAND (OGG/Opus) radio streams, use chained OGG handler.
491            # This handles the chained OGG format by stitching logical bitstreams together
492            # so FFmpeg sees a single continuous stream. Metadata is extracted in-band.
493            def _on_inband_metadata(metadata: dict[str, str]) -> None:
494                """Handle metadata extracted from the OGG stream."""
495                title = metadata.get("title", "")
496                artist = metadata.get("artist", "")
497                album = metadata.get("album", "")
498                if not artist and " - " in title:
499                    artist, title = title.split(" - ", 1)
500                if title or artist:
501                    if artist and title:
502                        stream_title = f"{artist} - {title}"
503                    elif title:
504                        stream_title = title
505                    else:
506                        stream_title = artist
507                    cleaned_title = clean_stream_title(stream_title)
508                    if cleaned_title and cleaned_title != streamdetails.stream_title:
509                        self.logger.log(VERBOSE_LOG_LEVEL, "In-band metadata: %s", cleaned_title)
510                        streamdetails.stream_title = cleaned_title
511                        self._update_radio_stream_metadata(
512                            streamdetails,
513                            artist=artist or None,
514                            title=title or cleaned_title,
515                            album=album or None,
516                        )
517
518            audio_source = get_chained_ogg_stream(
519                mass, streamdetails.path, metadata_callback=_on_inband_metadata
520            )
521            seek_position = 0  # seeking not possible on radio streams
522        elif stream_type == StreamType.HLS:
523            assert isinstance(streamdetails.path, str)  # for type checking
524            substream = await self.get_hls_substream(streamdetails.path)
525            audio_source = substream.path
526            if streamdetails.media_type == MediaType.RADIO:
527                # HLS streams (especially the BBC) struggle when they're played directly
528                # with ffmpeg, where they just stop after some minutes,
529                # so we tell ffmpeg to loop around in this case.
530                extra_input_args += ["-stream_loop", "-1", "-re"]
531        else:
532            # all other stream types (HTTP, FILE, etc)
533            if stream_type == StreamType.ENCRYPTED_HTTP:
534                assert streamdetails.decryption_key is not None  # for type checking
535                extra_input_args += ["-decryption_key", streamdetails.decryption_key]
536            if isinstance(streamdetails.path, list):
537                # multi part stream
538                audio_source = self.get_multi_file_stream(streamdetails, seek_position)
539                seek_position = 0  # handled by get_multi_file_stream
540            else:
541                # regular single file/url stream
542                assert isinstance(streamdetails.path, str)  # for type checking
543                audio_source = streamdetails.path
544
545        # pace ffmpeg at native rate for live sources; the producer (e.g.
546        # librespot's pipe backend) may otherwise write faster than realtime
547        if streamdetails.media_type == MediaType.AUDIO_SOURCE and "-re" not in extra_input_args:
548            extra_input_args += ["-re"]
549
550        # handle seek support
551        if seek_position and streamdetails.duration and streamdetails.allow_seek:
552            extra_input_args += ["-ss", str(int(seek_position))]
553
554        bytes_sent = 0
555        finished = False
556        cancelled = False
557        first_chunk_received = False
558        ffmpeg_loglevel = "debug" if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL) else "info"
559        # When a provider hands us already-decoded audio (e.g. Spotify Connect /
560        # AirPlay receivers piping PCM after their own decode), audio_format is
561        # the original source format meant for display while decoded_audio_format
562        # is what ffmpeg actually needs to read off the wire.
563        ffmpeg_input_format = streamdetails.decoded_audio_format or streamdetails.audio_format
564        ffmpeg_proc = FFMpeg(
565            audio_input=audio_source,
566            input_format=ffmpeg_input_format,
567            output_format=pcm_format,
568            filter_params=filter_params,
569            extra_input_args=extra_input_args,
570            collect_log_history=True,
571            loglevel=ffmpeg_loglevel,
572        )
573
574        try:
575            await ffmpeg_proc.start()
576            assert ffmpeg_proc.proc is not None  # for type checking
577            if logger.isEnabledFor(VERBOSE_LOG_LEVEL):
578                logger.log(
579                    VERBOSE_LOG_LEVEL,
580                    "Started media stream for %s - using streamtype: %s "
581                    "- pcm format: %s - ffmpeg PID: %s",
582                    streamdetails.uri,
583                    streamdetails.stream_type,
584                    pcm_format.content_type.value,
585                    ffmpeg_proc.proc.pid,
586                )
587            else:
588                logger.debug(
589                    "Started media stream for %s - using streamtype: %s",
590                    streamdetails.uri,
591                    streamdetails.stream_type,
592                )
593            stream_start = mass.loop.time()
594            chunk_size = calculate_content_length(pcm_format, chunk_seconds)
595            chunk_iter = ffmpeg_proc.iter_chunked(chunk_size)
596            while True:
597                # Time the read, not the yield: catches a stalled source, ignores backpressure.
598                read_timeout = (
599                    STREAM_START_TIMEOUT if not first_chunk_received else STREAM_STALL_TIMEOUT
600                )
601                try:
602                    async with asyncio.timeout(read_timeout):
603                        chunk = await anext(chunk_iter)
604                except StopAsyncIteration:
605                    break
606                except TimeoutError as err:
607                    raise AudioError(f"Source stalled: no audio for {read_timeout}s") from err
608                if not first_chunk_received:
609                    # At this point ffmpeg has started and should now know the codec used
610                    # for encoding the audio.
611                    # Note: ffmpeg_proc.input_format is the same object as
612                    # ffmpeg_input_format, so sample_rate / bit_depth / bit_rate
613                    # parsed from the ffmpeg log already live on streamdetails too.
614                    first_chunk_received = True
615                    # Skip the codec_type writeback when the provider declared a
616                    # decoded format: audio_format already holds the authoritative
617                    # source codec and the probed value would just be the
618                    # post-decode wire format (e.g. PCM for Spotify Connect).
619                    if streamdetails.decoded_audio_format is None:
620                        streamdetails.audio_format.codec_type = ffmpeg_proc.input_format.codec_type
621                    # Some providers omit (or report 0 for) the item duration; ffmpeg can
622                    # usually probe it from the source. Only apply when missing so we
623                    # don't clobber an accurate provider value with a rounded one.
624                    if ffmpeg_proc.parsed_duration is not None and not streamdetails.duration:
625                        streamdetails.duration = ffmpeg_proc.parsed_duration
626                    logger.debug(
627                        "First chunk received after %.2f seconds (codec detected: %s)",
628                        mass.loop.time() - stream_start,
629                        ffmpeg_proc.input_format.codec_type,
630                    )
631                yield chunk
632                bytes_sent += len(chunk)
633
634            # end of audio/track reached
635            logger.debug("End of media stream reached for %s", streamdetails.uri)
636            # wait until stderr also completed reading
637            await ffmpeg_proc.wait_with_timeout(5)
638            logger.log(
639                VERBOSE_LOG_LEVEL,
640                "FFmpeg process ended with return code %s for %s",
641                ffmpeg_proc.returncode,
642                streamdetails.uri,
643            )
644            if ffmpeg_proc.returncode not in (0, None):
645                log_trail = "\n".join(list(ffmpeg_proc.log_history)[-5:])
646                raise AudioError(f"FFMpeg exited with code {ffmpeg_proc.returncode}: {log_trail}")
647            if bytes_sent == 0:
648                # edge case: no audio data was received at all
649                raise AudioError("No audio was received")
650            finished = True
651        except (Exception, GeneratorExit, asyncio.CancelledError) as err:
652            if isinstance(err, asyncio.CancelledError | GeneratorExit):
653                # we were cancelled, just raise
654                cancelled = True
655                raise
656            # dump the last 10 lines of the log in case of an unclean exit
657            logger.warning("\n".join(list(ffmpeg_proc.log_history)[-10:]))
658            raise AudioError(f"Error while streaming: {err}") from err
659        finally:
660            # always ensure close is called which also handles all cleanup
661            await ffmpeg_proc.close()
662            # determine how many seconds we've received
663            # for pcm output we can calculate this easily
664            seconds_received = bytes_sent / pcm_format.pcm_sample_size if bytes_sent else 0
665            # store accurate duration, but only for a playthrough from the very start:
666            # a seeked stream yields the remaining audio, not the item's full length
667            if finished and not requested_seek_position and seconds_received:
668                streamdetails.duration = int(seconds_received)
669
670            logger.log(
671                VERBOSE_LOG_LEVEL,
672                "stream %s (with code %s) for %s",
673                "cancelled" if cancelled else "finished" if finished else "aborted",
674                ffmpeg_proc.returncode,
675                streamdetails.uri,
676            )
677
678    async def resolve_radio_stream(self, url: str) -> tuple[str, StreamType]:
679        """
680        Resolve a streaming radio URL.
681
682        Unwraps playlists and determines stream type (ICY, HLS, SHOUTCAST, IN_BAND, HTTP).
683
684        :param url: Radio stream URL to resolve
685        """
686        mass = self.mass
687        if cache := await mass.cache.get(
688            key=url, provider=CACHE_PROVIDER, category=CACHE_CATEGORY_RESOLVED_RADIO_URL
689        ):
690            if TYPE_CHECKING:
691                cache = cast("tuple[str, str]", cache)
692            return (cache[0], StreamType(cache[1]))
693
694        stream_type = StreamType.HTTP
695        timeout = ClientTimeout(total=None, connect=10, sock_read=5)
696
697        try:
698            async with self._connect_radio_stream(
699                url, headers=HTTP_HEADERS_ICY, allow_redirects=True, timeout=timeout
700            ) as resp:
701                headers = resp.headers
702                resp.raise_for_status()
703                if not resp.headers:
704                    raise InvalidDataError("no headers found")
705
706            if headers.get("icy-metaint") is not None:
707                stream_type = StreamType.ICY
708            elif headers.get("content-type", "") in ("application/ogg", "audio/ogg"):
709                # Ogg streams (Opus/Vorbis) have in-band metadata via Vorbis comments
710                stream_type = StreamType.IN_BAND
711
712            if (
713                url.endswith((".m3u", ".m3u8", ".pls"))
714                or ".m3u?" in url
715                or ".m3u8?" in url
716                or ".pls?" in url
717                or "audio/x-mpegurl" in headers.get("content-type", "")
718                or "audio/x-scpls" in headers.get("content-type", "")
719            ):
720                try:
721                    substreams = await fetch_playlist(mass, url)
722                    if not any(x for x in substreams if x.length):
723                        for line in substreams:
724                            if not line.is_url:
725                                continue
726                            return await self.resolve_radio_stream(line.path)
727                        raise InvalidDataError("No content found in playlist")
728                except IsHLSPlaylist:
729                    stream_type = StreamType.HLS
730
731        except TimeoutError as err:
732            self.logger.warning("Timeout while parsing radio URL %s", url)
733            raise InvalidDataError(f"Timeout connecting to {url}") from err
734
735        except aiohttp.ClientResponseError as err:
736            if err.status == 404:
737                raise MediaNotFoundError(f"Radio stream not found: {url}") from err
738            if err.status == 403:
739                raise InvalidDataError(f"Access denied to radio stream: {url}") from err
740            if err.status >= 500:
741                raise InvalidDataError(
742                    f"Radio stream server error (HTTP {err.status}): {url}"
743                ) from err
744            if err.status == 400:
745                # 400 errors might be from legacy Shoutcast servers
746                return await self._handle_client_error_for_radio_stream(url, err, stream_type)
747            raise InvalidDataError(f"HTTP error {err.status} from {url}") from err
748
749        except aiohttp.ClientError as err:
750            return await self._handle_client_error_for_radio_stream(url, err, stream_type)
751
752        return await self._cache_radio_result(url, stream_type)
753
754    async def get_icy_radio_stream(
755        self, url: str, streamdetails: StreamDetails
756    ) -> AsyncGenerator[bytes]:
757        """
758        Stream radio audio with ICY metadata support, reconnecting on disconnect.
759
760        Requires icy-metaint header support. Stream type should be validated
761        by resolve_radio_stream() before calling this function.
762
763        :param url: Radio stream URL
764        :param streamdetails: StreamDetails to update with metadata
765        """
766        self.logger.debug("Start streaming radio with ICY metadata from url %s", url)
767        timeout = ClientTimeout(total=0, connect=30, sock_read=5 * 60)
768        # Budget for *consecutive* reconnects that delivered no audio. A connection
769        # that actually streamed data resets it, so a healthy long-running stream can
770        # reconnect indefinitely while a dead/looping one bails out instead of spinning.
771        failed_reconnects = 0
772        max_failed_reconnects = 25
773
774        while True:
775            streamed_data = False
776            try:
777                async with self._connect_radio_stream(
778                    url, allow_redirects=True, headers=HTTP_HEADERS_ICY, timeout=timeout
779                ) as resp:
780                    # surface a non-200 (e.g. on reconnect) as a ClientResponseError so the
781                    # terminal/HTTP handling below applies instead of failing on the header
782                    resp.raise_for_status()
783                    meta_int_str = resp.headers.get("icy-metaint")
784                    if not meta_int_str:
785                        raise InvalidDataError(f"No icy-metaint header for radio stream: {url}")
786                    try:
787                        meta_int = int(meta_int_str)
788                    except ValueError as err:
789                        raise InvalidDataError(
790                            f"Invalid icy-metaint value for radio stream: {url}"
791                        ) from err
792                    if meta_int <= 0:
793                        raise InvalidDataError(f"Invalid icy-metaint value for radio stream: {url}")
794                    # readexactly raises IncompleteReadError when the server closes the
795                    # connection mid-frame; that (and the network errors below) drops us
796                    # out to the reconnect handler so a live stream survives the blip.
797                    while True:
798                        chunk = await resp.content.readexactly(meta_int)
799                        streamed_data = True
800                        yield chunk
801                        meta_byte = await resp.content.readexactly(1)
802                        if meta_byte == b"\x00":
803                            continue
804                        meta_length = ord(meta_byte) * 16
805                        meta_data = await resp.content.readexactly(meta_length)
806                        self._parse_icy_metadata(meta_data, streamdetails)
807            except asyncio.CancelledError:
808                self.logger.debug("ICY radio stream cancelled for %s", url)
809                raise
810            except aiohttp.ClientResponseError as err:
811                if err.status == 404:
812                    raise MediaNotFoundError(f"Radio stream not found: {url}") from err
813                if err.status == 403:
814                    raise ProviderPermissionDenied(f"Radio stream access denied: {url}") from err
815                raise ProviderUnavailableError(
816                    f"Radio stream returned HTTP {err.status}: {err}"
817                ) from err
818            except (
819                asyncio.IncompleteReadError,
820                aiohttp.ClientConnectionError,
821                aiohttp.ClientPayloadError,
822                aiohttp.ServerDisconnectedError,
823            ) as err:
824                if streamed_data:
825                    # a healthy session that dropped - reconnect without spending budget
826                    failed_reconnects = 0
827                    self.logger.debug("ICY radio stream dropped, reconnecting: %s", err)
828                else:
829                    failed_reconnects += 1
830                    if failed_reconnects > max_failed_reconnects:
831                        raise RetriesExhausted(
832                            f"ICY radio stream failed after {max_failed_reconnects} "
833                            f"reconnects without data: {err}"
834                        ) from err
835                    self.logger.warning(
836                        "ICY radio stream reconnect produced no data (%d/%d): %s",
837                        failed_reconnects,
838                        max_failed_reconnects,
839                        err,
840                    )
841                await asyncio.sleep(0.5)
842
843    async def get_reconnecting_icy_radio_stream(
844        self, url: str | list[MultiPartPath], streamdetails: StreamDetails
845    ) -> AsyncGenerator[bytes]:
846        """
847        Yield ICY radio audio with metadata, failing over across mirror URLs.
848
849        A single URL is delegated to :meth:`get_icy_radio_stream`, which already reconnects
850        on disconnect. Multiple URLs are treated as interchangeable mirrors and tried in turn;
851        a mirror that delivers audio resets the failover budget, so a healthy mirror keeps
852        streaming while a set of unreachable mirrors raises the last error instead of spinning.
853
854        :param url: One stream URL, or a list of mirror URLs to fail over between.
855        :param streamdetails: StreamDetails to update with metadata.
856        """
857        urls = self._normalize_reconnecting_urls(url)
858        if len(urls) == 1:
859            async for chunk in self.get_icy_radio_stream(urls[0], streamdetails):
860                yield chunk
861            return
862
863        url_index = 0
864        failed_rotations = 0
865        max_failed_rotations = len(urls) * 2
866        last_err: MusicAssistantError | None = None
867        while failed_rotations <= max_failed_rotations:
868            current_url = urls[url_index % len(urls)]
869            url_index += 1
870            delivered_audio = False
871            try:
872                async for chunk in self.get_icy_radio_stream(current_url, streamdetails):
873                    delivered_audio = True
874                    failed_rotations = 0
875                    # release the previous failure while healthy: it pins the full
876                    # exception traceback (with frames) for the lifetime of the stream
877                    last_err = None
878                    yield chunk
879                return
880            except RADIO_MIRROR_FAILOVER_ERRORS as err:
881                last_err = err
882                if not delivered_audio:
883                    failed_rotations += 1
884                self.logger.warning(
885                    "ICY radio mirror %s failed, trying next url (%d/%d): %s",
886                    current_url,
887                    failed_rotations,
888                    max_failed_rotations,
889                    err,
890                )
891        if last_err is not None:
892            raise last_err
893
894    async def get_reconnecting_radio_stream(self, url: str) -> AsyncGenerator[bytes]:
895        """
896        Yield continuous radio stream data, automatically reconnecting on disconnect.
897
898        :param url: URL of the radio stream.
899        """
900        timeout = ClientTimeout(total=None, connect=30, sock_read=5 * 60)
901        reconnect_count = 0
902        max_reconnects = 1000  # Allow many reconnects for long-running radio
903
904        while reconnect_count <= max_reconnects:
905            try:
906                async with self._connect_radio_stream(
907                    url, allow_redirects=True, headers=HTTP_HEADERS, timeout=timeout
908                ) as resp:
909                    chunk_count = 0
910                    async for chunk in resp.content.iter_any():
911                        chunk_count += 1
912                        yield chunk
913
914                    # Connection closed normally - reconnect
915                    self.logger.debug(
916                        "Radio stream connection closed after %d chunks, reconnecting... "
917                        "(reconnect #%d)",
918                        chunk_count,
919                        reconnect_count,
920                    )
921                    reconnect_count += 1
922                    await asyncio.sleep(0.1)  # Brief delay before reconnect
923
924            except asyncio.CancelledError:
925                self.logger.debug("Radio stream cancelled for %s", url)
926                raise
927            except (
928                aiohttp.ClientConnectionError,
929                aiohttp.ClientPayloadError,
930                aiohttp.ServerDisconnectedError,
931            ) as err:
932                # Transient network errors - retry
933                self.logger.warning("Radio stream error (reconnect #%d): %s", reconnect_count, err)
934                reconnect_count += 1
935                if reconnect_count > max_reconnects:
936                    raise RetriesExhausted(
937                        f"Radio stream failed after {max_reconnects} reconnects: {err}"
938                    ) from err
939                await asyncio.sleep(0.5)
940            except aiohttp.ClientResponseError as err:
941                if err.status == 404:
942                    raise MediaNotFoundError(f"Radio stream not found: {url}") from err
943                if err.status == 403:
944                    raise ProviderPermissionDenied(f"Radio stream access denied: {url}") from err
945                # Other HTTP errors (5xx etc) - could be temporary
946                raise ProviderUnavailableError(
947                    f"Radio stream returned HTTP {err.status}: {err}"
948                ) from err
949
950        self.logger.warning("Radio stream reached max reconnects (%d) for %s", max_reconnects, url)
951
952    async def get_hls_substream(self, url: str) -> PlaylistItem:
953        """Select the (highest quality) HLS substream for given HLS playlist/URL."""
954        mass = self.mass
955        timeout = ClientTimeout(total=None, connect=30, sock_read=5 * 60)
956        # fetch master playlist and select (best) child playlist
957        # https://datatracker.ietf.org/doc/html/draft-pantos-http-live-streaming-19#section-10
958        async with mass.http_session_no_ssl.get(
959            encoded_request_url(url), allow_redirects=True, headers=HTTP_HEADERS, timeout=timeout
960        ) as resp:
961            resp.raise_for_status()
962            raw_data = await resp.read()
963            encoding = await detect_charset(raw_data, preferred=resp.charset)
964            master_m3u_data = raw_data.decode(encoding, errors="replace")
965        substreams = parse_m3u(master_m3u_data)
966        # There is a chance that we did not get a master playlist with subplaylists
967        # but just a single master/sub playlist with the actual audio stream(s)
968        # so we need to detect if the playlist child's contain audio streams or
969        # sub-playlists.
970        if any(
971            x
972            for x in substreams
973            if (x.length or x.path.endswith((".mp4", ".aac")))
974            and not x.path.endswith((".m3u", ".m3u8"))
975        ):
976            return PlaylistItem(path=url, key=substreams[0].key)
977        # sort substreams on best quality (highest bandwidth) when available
978        if any(x for x in substreams if x.stream_info):
979            substreams.sort(
980                key=lambda x: int(
981                    x.stream_info.get("BANDWIDTH", "0") if x.stream_info is not None else 0
982                ),
983                reverse=True,
984            )
985        substream = substreams[0]
986        if not substream.path.startswith("http"):
987            # path is relative, stitch it together
988            base_path = url.rsplit("/", 1)[0]
989            substream.path = base_path + "/" + substream.path
990        return substream
991
992    async def get_multi_file_stream(
993        self,
994        streamdetails: StreamDetails,
995        seek_position: int = 0,
996    ) -> AsyncGenerator[bytes]:
997        """
998        Return audio stream for a concatenation of multiple files.
999
1000        Arguments:
1001        seek_position: The position to seek to in seconds
1002        """
1003        if not isinstance(streamdetails.path, list):
1004            raise InvalidDataError("Multi-file streamdetails requires a list of MultiPartPath")
1005        parts, seek_position = get_parts_from_position(streamdetails.path, seek_position)
1006        files_list = [part.path for part in parts]
1007
1008        # concat input files
1009        temp_file = f"/tmp/{shortuuid.random(20)}.txt"  # noqa: S108
1010        async with aiofiles.open(temp_file, "w") as f:
1011            await f.write(build_concat_filelist(files_list))
1012
1013        try:
1014            async for chunk in get_ffmpeg_stream(
1015                audio_input=temp_file,
1016                input_format=streamdetails.audio_format,
1017                output_format=AudioFormat(
1018                    content_type=ContentType.NUT,
1019                    sample_rate=streamdetails.audio_format.sample_rate,
1020                    bit_depth=streamdetails.audio_format.bit_depth,
1021                    channels=streamdetails.audio_format.channels,
1022                ),
1023                extra_input_args=[
1024                    "-safe",
1025                    "0",
1026                    "-f",
1027                    "concat",
1028                    "-i",
1029                    temp_file,
1030                    "-ss",
1031                    str(seek_position),
1032                ],
1033            ):
1034                yield chunk
1035        finally:
1036            await remove_file(temp_file)
1037
1038    def get_player_output_plan(
1039        self,
1040        player_id: str,
1041        input_format: AudioFormat,
1042        output_format: AudioFormat,
1043        *,
1044        shared_player_ids: Iterable[str] | None = None,
1045        handoff_format: AudioFormat | None = None,
1046        queue_id: str | None = None,
1047        session_id: str | None = None,
1048        queue_item_id: str | None = None,
1049    ) -> AudioOutputPlan:
1050        """
1051        Return executable filters and matching output details for a player.
1052
1053        :param player_id: Destination player identifier.
1054        :param input_format: PCM format entering player-specific processing.
1055        :param output_format: Furthest downstream output format known to the server.
1056        :param shared_player_ids: Additional players receiving this identical output path.
1057            An empty iterable marks a path that can gain shared destinations later.
1058        :param handoff_format: Earlier provider handoff format when it differs.
1059        :param queue_id: Explicit queue identifier for the processing snapshot.
1060        :param session_id: Explicit queue session identifier for the processing snapshot.
1061        :param queue_item_id: Queue item for a single-item output path.
1062        """
1063        filter_params: list[str | ComplexFilter] = []
1064        player = self.mass.players.get_player(player_id)
1065        destination_player_id = (
1066            player.protocol_parent_id if player and player.protocol_parent_id else player_id
1067        )
1068        resolved_shared_player_ids = (
1069            resolve_output_player_ids(self.mass, shared_player_ids) - {destination_player_id}
1070            if shared_player_ids is not None
1071            else None
1072        )
1073        destination_player_ids = {destination_player_id, *(resolved_shared_player_ids or ())}
1074        if player:
1075            dsp_config_id = self._resolve_player_dsp_config_id(player)
1076            dsp = self._resolve_player_dsp_config(player)
1077            configured_dsp = self.mass.config.get_player_dsp_config(dsp_config_id)
1078            if configured_dsp.enabled and not dsp.enabled and is_grouping_preventing_dsp(player):
1079                dsp_state = DSPState.DISABLED_BY_UNSUPPORTED_GROUP
1080            else:
1081                dsp_state = DSPState.ENABLED if dsp.enabled else DSPState.DISABLED
1082        else:
1083            dsp_config_id = player_id
1084            dsp = self.mass.config.get_player_dsp_config(player_id)
1085            dsp_state = DSPState.ENABLED if dsp.enabled else DSPState.DISABLED
1086
1087        enabled_filters = [dsp_filter for dsp_filter in dsp.filters if dsp_filter.enabled]
1088        # a neutral filter (0 dB gain, centered balance) emits no params; exclude
1089        # it so it is not reported as an active, non-bit-perfect stage
1090        effective_filters: list[DSPFilter] = []
1091        if dsp.enabled:
1092            if dsp.input_gain != 0:
1093                filter_params.append(f"volume={dsp.input_gain}dB")
1094            ir_dir = os.path.join(self.mass.storage_path, DSP_IRS_DIRNAME)
1095            known_ir_ids = {record["ir_id"] for record in self.mass.config.get_dsp_irs()}
1096            for dsp_filter in enabled_filters:
1097                if isinstance(dsp_filter, ConvolutionFilter) and dsp_filter.ir_id:
1098                    # ffmpeg fails to open the graph if the impulse response file is gone,
1099                    # which costs the player all audio, so drop the filter instead
1100                    if dsp_filter.ir_id not in known_ir_ids:
1101                        self.logger.warning(
1102                            "Skipping the convolution filter of player %s: "
1103                            "impulse response %s is not stored",
1104                            player_id,
1105                            dsp_filter.ir_id,
1106                        )
1107                        continue
1108                params = filter_to_ffmpeg_params(dsp_filter, input_format, ir_dir=ir_dir)
1109                if not params:
1110                    continue
1111                filter_params.extend(params)
1112                effective_filters.append(dsp_filter)
1113            if dsp.output_gain != 0:
1114                filter_params.append(f"volume={dsp.output_gain}dB")
1115
1116        channel_value = self._get_output_channels(player, player_id)
1117        source_channel = None
1118        channel_mix = ""
1119        # a single channel source is already the downmix and holds no FL/FR to select
1120        # from, where a pan would silently resolve every gain to zero
1121        if input_format.channels > 1:
1122            if channel_value == "left":
1123                source_channel = AudioChannel.FL
1124                channel_mix = "FL"
1125            elif channel_value == "right":
1126                source_channel = AudioChannel.FR
1127                channel_mix = "FR"
1128            elif channel_value == "mono":
1129                # both source channels feed the downmix, so report ALL to keep the
1130                # output from ever being presented as bit perfect
1131                source_channel = AudioChannel.ALL
1132                channel_mix = "0.5*FL+0.5*FR"
1133        if channel_mix:
1134            # the pan runs in the command that emits the handoff format, and it feeds
1135            # every channel of it explicitly: leaving ffmpeg to upmix from a single
1136            # channel costs 3 dB through its rematrix
1137            if (handoff_format or output_format).channels == 1:
1138                filter_params.append(f"pan=mono|c0={channel_mix}")
1139            else:
1140                filter_params.append(f"pan=stereo|c0={channel_mix}|c1={channel_mix}")
1141
1142        output_details = AudioOutputDetails(
1143            player_ids=sorted(destination_player_ids),
1144            dsp=AudioDSPDetails(
1145                state=dsp_state,
1146                input_gain=dsp.input_gain if dsp.enabled else 0.0,
1147                filters=effective_filters,
1148                output_gain=dsp.output_gain if dsp.enabled else 0.0,
1149                preset_id=dsp.preset_id,
1150            ),
1151            source_channel=source_channel,
1152            output_format=output_format,
1153        )
1154        output_plan = AudioOutputPlan(
1155            filter_params=filter_params,
1156            output_details=output_details,
1157            input_format=input_format,
1158            handoff_format=handoff_format,
1159            dsp_config_id=dsp_config_id,
1160        )
1161        if queue_id is not None and session_id is not None:
1162            self.mass.streams.audio_processing.update_output(
1163                destination_player_id,
1164                output_plan,
1165                shared_player_ids=resolved_shared_player_ids,
1166                queue_id=queue_id,
1167                session_id=session_id,
1168                queue_item_id=queue_item_id,
1169            )
1170        self.logger.log(
1171            VERBOSE_LOG_LEVEL,
1172            "Generated ffmpeg params for player %s: %s",
1173            player_id,
1174            filter_params,
1175        )
1176        return output_plan
1177
1178    async def get_output_format(
1179        self,
1180        output_format_str: str,
1181        player: Player,
1182        content_sample_rate: int,
1183        content_bit_depth: int,
1184        media_type: MediaType = MediaType.UNKNOWN,
1185    ) -> AudioFormat:
1186        """Parse (player specific) output format details for given format string."""
1187        content_type: ContentType = ContentType.try_parse(output_format_str)
1188        player_supported_rates = player.get_supported_sample_rates()
1189        supported_sample_rates = [sr for sr, _ in player_supported_rates]
1190        if content_sample_rate in supported_sample_rates:
1191            output_sample_rate = content_sample_rate
1192        else:
1193            output_sample_rate = max(supported_sample_rates)
1194        # only consider bit depths that are actually paired with the chosen sample rate
1195        bit_depths_for_rate = [
1196            bd for (sr, bd) in player_supported_rates if sr == output_sample_rate
1197        ]
1198        output_bit_depth = min(content_bit_depth, max(bit_depths_for_rate, default=16))
1199
1200        if not content_type.is_lossless():
1201            # no point in having a higher bit depth for lossy formats
1202            output_bit_depth = 16
1203            output_sample_rate = min(48000, output_sample_rate)
1204        if media_type not in (MediaType.TRACK, MediaType.AUDIO_SOURCE, MediaType.FLOW_STREAM):
1205            # no point in having a higher bit depth for non-track media types (e.g. TTS, radio)
1206            output_bit_depth = min(output_bit_depth, 16)
1207        if output_format_str == "pcm":
1208            content_type = ContentType.from_bit_depth(output_bit_depth)
1209
1210        output_channels_str = self._get_output_channels(player, player.player_id)
1211        fmt = AudioFormat(
1212            content_type=content_type,
1213            sample_rate=output_sample_rate,
1214            bit_depth=output_bit_depth,
1215            channels=1 if output_channels_str != "stereo" else 2,
1216        )
1217        fmt.bit_rate = get_bit_rate(fmt)
1218        return fmt
1219
1220    async def select_pcm_format(
1221        self,
1222        player: Player,
1223        streamdetails: StreamDetails,
1224        crossfade_enabled: bool,
1225        overlay_active: bool = False,
1226    ) -> AudioFormat:
1227        """
1228        Select the internal PCM format for streaming a single queue item.
1229
1230        Used by the per-item (non-flow) stream path. The sample rate is the highest
1231        rate the player supports that is <= the source rate, so the source is never
1232        upsampled. The bit depth follows the source unless audio processing
1233        (crossfade, volume normalization, DSP) is active — those need F32 headroom
1234        to avoid clipping/precision loss. Surround sources are folded down to stereo.
1235        Realtime AudioSource items skip all processing and get a pure passthrough
1236        format (source rate/bit depth when the player supports them).
1237
1238        :param player: The player requesting the stream.
1239        :param streamdetails: Stream details for the current item.
1240        :param crossfade_enabled: Whether crossfade is enabled for this stream.
1241        :param overlay_active: Whether an audio overlay will be mixed into this stream.
1242        """
1243        if streamdetails.media_type == MediaType.AUDIO_SOURCE:
1244            return self._select_audio_source_pcm_format(player, streamdetails)
1245        supported_sample_rates = [sr for sr, _ in player.get_supported_sample_rates()]
1246        # snap-down: pick the highest supported rate <= source. when the source rate
1247        # is below every supported rate (e.g. 22 kHz content on a 44.1k-only player),
1248        # fall back to the lowest supported rate instead of a hardcoded 48 kHz that
1249        # the player may not actually support.
1250        output_sample_rate = max(
1251            (r for r in supported_sample_rates if r <= streamdetails.audio_format.sample_rate),
1252            default=min(supported_sample_rates),
1253        )
1254        content_type, bit_depth = self._pick_pcm_bit_depth(
1255            (player,),
1256            streamdetails,
1257            crossfade_enabled,
1258            overlay_active,
1259        )
1260        pcm_format = AudioFormat(
1261            sample_rate=output_sample_rate,
1262            content_type=content_type,
1263            bit_depth=bit_depth,
1264            # fold surround sources down to stereo right at the decode step: no
1265            # output format carries more than two channels, so a wider PCM format
1266            # only makes every bytes-to-seconds sum on the stream come out short
1267            channels=min(streamdetails.audio_format.channels, 2),
1268        )
1269        if crossfade_enabled or overlay_active:
1270            pcm_format.channels = 2
1271        return pcm_format
1272
1273    async def select_flow_pcm_format(
1274        self,
1275        player: Player,
1276        start_streamdetails: StreamDetails | None = None,
1277        crossfade_enabled: bool = False,
1278        overlay_active: bool = False,
1279        fallback_sample_rate: int | None = None,
1280        output_players: Iterable[Player] | None = None,
1281    ) -> AudioFormat:
1282        """
1283        Select the internal PCM format for a Queue Flow Mode stream.
1284
1285        Used by the gapless flow path that stitches multiple queue items into one
1286        continuous PCM stream. The sample rate is driven by the player's
1287        ``CONF_FLOW_MODE_SAMPLE_RATE`` setting (smart/bit_perfect/48k/96k/highest)
1288        — for the anchored modes it follows the first track's rate, for the fixed
1289        modes it snaps to the configured rate. The bit depth follows the first
1290        track's source unless audio processing is active (then F32 for headroom),
1291        avoiding an unnecessary up-convert to 32-bit when none of the consumers
1292        will benefit from it. When the first item is a realtime AudioSource, the
1293        flow mode config is ignored and a pure passthrough format is used so the
1294        source audio is delivered with minimum overhead and latency.
1295
1296        :param player: The player the flow stream is being prepared for.
1297        :param start_streamdetails: Stream details of the first track in the flow.
1298            Required for the anchored modes ('smart' / 'bit_perfect') and for the
1299            bit-depth optimization. May be omitted for the fixed-rate modes — when
1300            omitted the bit depth defaults to F32.
1301        :param crossfade_enabled: Whether the queue will use crossfade transitions.
1302        :param overlay_active: Whether an audio overlay will be mixed into the stream.
1303        :param fallback_sample_rate: Preferred rate when the first item format is unknown.
1304        :param output_players: All players consuming the shared PCM stream. Their common
1305            sample rates and processing requirements determine the session format.
1306        """
1307        players = tuple(output_players) if output_players is not None else (player,)
1308        if not players:
1309            raise AudioError("At least one output player is required")
1310        supported_sample_rates = sorted(
1311            set.intersection(
1312                *(
1313                    {sample_rate for sample_rate, _ in item.get_supported_sample_rates()}
1314                    for item in players
1315                )
1316            )
1317        )
1318        if not supported_sample_rates:
1319            raise AudioError("Output players do not share a supported sample rate")
1320        if start_streamdetails is not None and (
1321            start_streamdetails.media_type == MediaType.AUDIO_SOURCE
1322        ):
1323            return self._select_audio_source_pcm_format(
1324                player,
1325                start_streamdetails,
1326                supported_sample_rates=supported_sample_rates,
1327            )
1328        flow_mode_conf = cast(
1329            "str",
1330            player.config.get_value(CONF_FLOW_MODE_SAMPLE_RATE, FLOW_MODE_SAMPLE_RATE_SMART),
1331        )
1332
1333        if flow_mode_conf == FLOW_MODE_SAMPLE_RATE_HIGHEST:
1334            output_sample_rate = max(supported_sample_rates)
1335        elif flow_mode_conf == FLOW_MODE_SAMPLE_RATE_48000:
1336            # for the fixed-rate modes, the user picked a specific bandwidth/quality
1337            # ceiling; prefer the highest supported rate <= target
1338            output_sample_rate = _snap_supported_rate_down(48000, supported_sample_rates)
1339        elif flow_mode_conf == FLOW_MODE_SAMPLE_RATE_96000:
1340            output_sample_rate = _snap_supported_rate_down(96000, supported_sample_rates)
1341        else:
1342            # smart or bit_perfect (default): anchor the flow at the starting track's
1343            # sample rate; if the player doesn't natively support it, upsample to the
1344            # closest higher supported rate
1345            target_rate = (
1346                start_streamdetails.audio_format.sample_rate
1347                if start_streamdetails
1348                else (
1349                    fallback_sample_rate
1350                    if fallback_sample_rate is not None
1351                    else max(supported_sample_rates)
1352                )
1353            )
1354            output_sample_rate = _snap_supported_rate_up(target_rate, supported_sample_rates)
1355
1356        content_type, bit_depth = self._pick_pcm_bit_depth(
1357            players, start_streamdetails, crossfade_enabled, overlay_active
1358        )
1359        return AudioFormat(
1360            content_type=content_type,
1361            sample_rate=output_sample_rate,
1362            bit_depth=bit_depth,
1363            channels=2,
1364        )
1365
1366    async def get_audio_source_stream(
1367        self,
1368        queue_item: QueueItem,
1369        pcm_format: AudioFormat,
1370        raise_on_error: bool = True,
1371    ) -> AsyncGenerator[bytes]:
1372        """
1373        Get the realtime PCM stream for an AudioSource queue item.
1374
1375        AudioSources are live/realtime: bytes flow at the producer's pace, with
1376        no pre-buffering, no loudness hydration, no volume normalization, no
1377        crossfade/fade-in, no playback-speed shift, no next-track preload. The
1378        path stays as small as possible to keep end-to-end latency low.
1379
1380        Fast path: when the source PCM format already matches the consumer's
1381        ``pcm_format``, the provider's bytes are paced in Python and forwarded
1382        directly — no ffmpeg in the data path.
1383
1384        Slow path: when formats differ, ffmpeg resamples/recodes the stream
1385        (with ``-re`` for rate pacing) via ``get_media_stream``.
1386
1387        :param queue_item: The AudioSource queue item to stream.
1388        :param pcm_format: Output PCM format the consumer wants.
1389        :param raise_on_error: Re-raise stream errors instead of swallowing them.
1390        """
1391        streamdetails = queue_item.streamdetails
1392        assert streamdetails
1393        logger = self.logger.getChild("audio_source_stream")
1394        bytes_received = 0
1395        try:
1396            async for chunk in self._iter_audio_source_pcm(streamdetails, pcm_format):
1397                bytes_received += len(chunk)
1398                yield chunk
1399        except AudioError as err:
1400            streamdetails.stream_error = True
1401            # revoke availability when the stream never produced any audio
1402            if bytes_received == 0:
1403                queue_item.available = False
1404            if raise_on_error:
1405                raise
1406            logger.error(
1407                "AudioError while streaming AudioSource %s (%s): %s",
1408                queue_item.name,
1409                streamdetails.uri,
1410                err,
1411            )
1412        except asyncio.CancelledError:
1413            raise
1414        except Exception as err:
1415            streamdetails.stream_error = True
1416            if raise_on_error:
1417                raise
1418            logger.exception(
1419                "Unexpected error while streaming AudioSource %s (%s): %s",
1420                queue_item.name,
1421                streamdetails.uri,
1422                err,
1423            )
1424        finally:
1425            streamdetails.seconds_streamed = bytes_received / pcm_format.pcm_sample_size
1426
1427    async def get_queue_item_stream(
1428        self,
1429        queue_item: QueueItem,
1430        pcm_format: AudioFormat,
1431        seek_position: int = 0,
1432        playback_speed: float = 1.0,
1433        raise_on_error: bool = True,
1434        normalization_override: VolumeNormalizationMode | None = None,
1435        session_id: str | None = None,
1436    ) -> AsyncGenerator[bytes]:
1437        """
1438        Get the (PCM) audio stream for a single queue item.
1439
1440        Audio is always served from the AudioBuffer which stores raw decoded PCM.
1441        Volume normalization and other filters are applied on-the-fly when reading
1442        from the buffer.
1443
1444        AudioSource items dispatch to ``get_audio_source_stream`` instead: they
1445        are realtime and bypass the buffering/normalization/filter machinery.
1446
1447        :param normalization_override: Force this volume normalization mode instead of
1448            re-evaluating it from the (possibly just-updated) loudness measurement. Used by
1449            the crossfade path to keep a track's replayed intro and its body on the same mode.
1450        :param session_id: Queue session that owns processing-detail updates.
1451        """
1452        streamdetails = queue_item.streamdetails
1453        assert streamdetails
1454
1455        # streamdetails are cached and reused for retries; reset this before any
1456        # media-type-specific dispatch so AudioSource failures do not stick.
1457        streamdetails.stream_error = False
1458
1459        if queue_item.media_type == MediaType.AUDIO_SOURCE:
1460            async for chunk in self.get_audio_source_stream(
1461                queue_item=queue_item,
1462                pcm_format=pcm_format,
1463                raise_on_error=raise_on_error,
1464            ):
1465                yield chunk
1466            return
1467        filter_params: list[str] = []
1468
1469        logger = self.logger.getChild("queue_item_stream")
1470
1471        if normalization_override is not None:
1472            # crossfade path pins the body to the intro's mode; skip hydration/re-eval that could flip it
1473            streamdetails.volume_normalization_mode = normalization_override
1474        else:
1475            # hydrate loudness from audio analysis (just-in-time, so that a measurement
1476            # completed during a previous play is picked up here). A live analyzer run
1477            # may have already populated streamdetails.loudness in memory — don't clobber
1478            # that, and don't clobber a value set upstream by the music provider.
1479            if streamdetails.loudness is None:
1480                if analysis := await self.mass.streams.audio_analysis.get_audio_analysis(
1481                    streamdetails.item_id,
1482                    streamdetails.provider,
1483                    media_type=streamdetails.media_type,
1484                    # use the authoritative EBU R128 value, not another provider's loudness proxy
1485                    priority=(LOUDNESS_ANALYSIS_DOMAIN,),
1486                ):
1487                    if analysis.loudness_integrated is not None:
1488                        streamdetails.loudness = round(analysis.loudness_integrated, 2)
1489                    if analysis.loudness_album is not None and streamdetails.loudness_album is None:
1490                        streamdetails.loudness_album = round(analysis.loudness_album, 2)
1491
1492            # re-evaluate normalization mode: the background loudness analyzer may have
1493            # updated streamdetails.loudness since get_stream_details was called
1494            if streamdetails.queue_id:
1495                volume_normalization_enabled = (
1496                    self.mass.config.get_effective_player_queue_config_value(
1497                        streamdetails.queue_id, CONF_VOLUME_NORMALIZATION, CONF_VALUE_ENABLED
1498                    )
1499                    != CONF_VALUE_DISABLED
1500                )
1501                streamdetails.volume_normalization_mode = get_normalization_mode(
1502                    self._get_volume_normalization_preference(streamdetails),
1503                    volume_normalization_enabled,
1504                    streamdetails,
1505                )
1506
1507        # handle volume normalization
1508        gain_correct: float | None = None
1509        if streamdetails.volume_normalization_mode == VolumeNormalizationMode.DYNAMIC:
1510            filter_rule = (
1511                f"loudnorm=I={streamdetails.target_loudness}"
1512                ":TP=-2.0:LRA=10.0:offset=0.0:print_format=json"
1513            )
1514            filter_params.append(filter_rule)
1515        elif streamdetails.volume_normalization_mode == VolumeNormalizationMode.FIXED_GAIN:
1516            config_key = (
1517                CONF_VOLUME_NORMALIZATION_FIXED_GAIN_TRACKS
1518                if streamdetails.media_type == MediaType.TRACK
1519                else CONF_VOLUME_NORMALIZATION_FIXED_GAIN_RADIO
1520            )
1521            gain_value = self.mass.streams.get_config_value(config_key, return_type=float)
1522            gain_correct = round(gain_value, 2)
1523            filter_params.append(f"volume={gain_correct}dB")
1524        elif streamdetails.volume_normalization_mode == VolumeNormalizationMode.MEASUREMENT_ONLY:
1525            target_loudness = (
1526                float(streamdetails.target_loudness)
1527                if streamdetails.target_loudness is not None
1528                else 0.0
1529            )
1530            if streamdetails.prefer_album_loudness and streamdetails.loudness_album is not None:
1531                gain_correct = target_loudness - float(streamdetails.loudness_album)
1532            elif streamdetails.loudness is not None:
1533                gain_correct = target_loudness - float(streamdetails.loudness)
1534            else:
1535                gain_correct = 0.0
1536            gain_correct = round(gain_correct, 2)
1537            filter_params.append(f"volume={gain_correct}dB")
1538        streamdetails.volume_normalization_gain_correct = gain_correct
1539
1540        # handle playback speed
1541        if playback_speed != 1.0:
1542            filter_params.append(f"atempo={playback_speed}")
1543
1544        # handle optional fade-in
1545        if streamdetails.fade_in:
1546            filter_params.insert(0, "afade=type=in:start_time=0:duration=3")
1547
1548        logger.log(
1549            VERBOSE_LOG_LEVEL,
1550            "Starting queue item stream for %s (%s)"
1551            " - using fade-in: %s"
1552            " - using volume normalization: %s"
1553            " - using playback speed: %s",
1554            queue_item.name,
1555            streamdetails.uri,
1556            streamdetails.fade_in,
1557            streamdetails.volume_normalization_mode,
1558            playback_speed,
1559        )
1560
1561        # get or create the AudioBuffer (stores raw decoded PCM)
1562        seek_position_ms = int(seek_position * 1000)
1563        audio_buffer = await AudioBuffer.get_buffer(
1564            mass=self.mass,
1565            streamdetails=streamdetails,
1566            seek_position_ms=seek_position_ms,
1567            reason="streaming",
1568        )
1569        if (
1570            streamdetails.queue_id
1571            and (queue_data := self.mass.player_queues.queue_data_or_none(streamdetails.queue_id))
1572            and (processing_session_id := session_id or queue_data.session_id)
1573        ):
1574            self.mass.streams.audio_processing.update_item_runtime(
1575                queue_id=streamdetails.queue_id,
1576                session_id=processing_session_id,
1577                queue_item_id=queue_item.queue_item_id,
1578                input_format=audio_buffer.pcm_format,
1579                pcm_format=pcm_format,
1580                normalization=get_normalization_details(streamdetails, gain_correct),
1581                playback_speed=playback_speed,
1582                alters_audio=streamdetails.fade_in,
1583            )
1584        # read from buffer with filters applied (volume normalization, speed, fade-in, etc.)
1585        # if no processing needed, this yields directly from the buffer
1586        media_stream_gen = audio_buffer.get_stream(
1587            output_format=pcm_format,
1588            seek_position_ms=seek_position_ms,
1589            filter_params=filter_params or None,
1590        )
1591
1592        first_chunk_received = False
1593        bytes_received = 0
1594        finished = False
1595        next_buffer_triggered = False
1596        stream_started_at = asyncio.get_event_loop().time()
1597        try:
1598            async for chunk in media_stream_gen:
1599                bytes_received += len(chunk)
1600                if not first_chunk_received:
1601                    first_chunk_received = True
1602                    logger.log(
1603                        VERBOSE_LOG_LEVEL,
1604                        "First audio chunk received for %s (%s) after %.2f seconds",
1605                        queue_item.name,
1606                        streamdetails.uri,
1607                        asyncio.get_event_loop().time() - stream_started_at,
1608                    )
1609                # trigger pre-buffering of the next item well before end
1610                # to ensure the raw PCM is ready when the next item needs to be streamed.
1611                # tracks and sound effects are finite files that fill and close immediately;
1612                # live sources (radio, audio_source) open an upstream connection that would
1613                # sit idle and likely time out before the player actually consumes it.
1614                if (
1615                    not next_buffer_triggered
1616                    and streamdetails.duration
1617                    and (queue := self.mass.player_queues.get_active_queue(queue_item.queue_id))
1618                    and queue.next_item
1619                    and queue.next_item.queue_item_id != queue_item.queue_item_id
1620                    and queue.next_item.media_type in (MediaType.TRACK, MediaType.SOUND_EFFECT)
1621                    and (bytes_received / pcm_format.pcm_sample_size + seek_position)
1622                    >= streamdetails.duration - 60
1623                ):
1624                    next_buffer_triggered = True
1625                    self.mass.player_queues.prepare_next_audio_buffer(queue_item.queue_id)
1626                yield chunk
1627                del chunk
1628            # The FFmpeg stdin feeder swallows producer errors; re-raise even after
1629            # partial audio so a failed source is not mistaken for a completed stream.
1630            if audio_buffer.has_error:
1631                raise AudioError("Failed to stream audio") from audio_buffer._producer_error
1632            finished = True
1633        except AudioError as err:
1634            streamdetails.stream_error = True
1635            # revoke availability when the stream never produced any audio
1636            if bytes_received == 0:
1637                queue_item.available = False
1638            if raise_on_error:
1639                raise
1640            logger.error(
1641                "AudioError while streaming queue item %s (%s): %s",
1642                queue_item.name,
1643                streamdetails.uri,
1644                err,
1645            )
1646        except asyncio.CancelledError:
1647            raise
1648        except Exception as err:
1649            streamdetails.stream_error = True
1650            if raise_on_error:
1651                raise
1652            logger.exception(
1653                "Unexpected error while streaming queue item %s (%s): %s",
1654                queue_item.name,
1655                streamdetails.uri,
1656                err,
1657            )
1658        finally:
1659            seconds_streamed = bytes_received / pcm_format.pcm_sample_size
1660            streamdetails.seconds_streamed = seconds_streamed
1661            logger.log(
1662                VERBOSE_LOG_LEVEL,
1663                "stream %s for %s in %.2f seconds - seconds streamed/buffered: %.2f",
1664                "aborted" if not finished else "finished",
1665                streamdetails.uri,
1666                asyncio.get_event_loop().time() - stream_started_at,
1667                seconds_streamed,
1668            )
1669            self._notify_provider_streamed(streamdetails, finished, seconds_streamed)
1670
1671    async def get_queue_item_stream_with_smartfade(
1672        self,
1673        player: Player,
1674        queue_item: QueueItem,
1675        pcm_format: AudioFormat,
1676        crossfade_mode: CrossfadeMode = CrossfadeMode.SMART_CROSSFADE,
1677        standard_crossfade_duration: int = 10,
1678        session_id: str | None = None,
1679    ) -> AsyncGenerator[bytes]:
1680        """
1681        Return one queue item with a crossfade into the next item.
1682
1683        :param player: Player consuming the stream.
1684        :param queue_item: Queue item to stream.
1685        :param pcm_format: Shared PCM format.
1686        :param crossfade_mode: Effective crossfade mode.
1687        :param standard_crossfade_duration: Configured standard crossfade duration.
1688        :param session_id: Queue session that owns processing-detail updates.
1689        """
1690        queue = self.mass.player_queues.get(queue_item.queue_id)
1691        if not queue:
1692            raise RuntimeError(f"Queue {queue_item.queue_id} not found")
1693
1694        streamdetails = queue_item.streamdetails
1695        assert streamdetails
1696        crossfade_data = self._crossfade_data.get(queue.queue_id)
1697
1698        if crossfade_data and streamdetails.seek_position > 0:
1699            # don't do crossfade when seeking into track
1700            self.logger.debug(
1701                "Discarding crossfade data for queue %s - seeking into track (pos=%s)",
1702                queue.display_name,
1703                streamdetails.seek_position,
1704            )
1705            crossfade_data = None
1706        if crossfade_data and (crossfade_data.queue_item_id != queue_item.queue_item_id):
1707            # edge case alert: the next item changed just while we were preloading/crossfading
1708            self.logger.warning(
1709                "Skipping crossfade data for queue %s - next item changed!"
1710                " (expected queue_item_id=%s, got=%s)",
1711                queue.display_name,
1712                crossfade_data.queue_item_id,
1713                queue_item.queue_item_id,
1714            )
1715            crossfade_data = None
1716            self._crossfade_data.pop(queue.queue_id, None)
1717        elif not crossfade_data:
1718            self.logger.debug(
1719                "No crossfade data available for queue %s (queue_item_id=%s)",
1720                queue.display_name,
1721                queue_item.queue_item_id,
1722            )
1723
1724        self.logger.debug(
1725            "Start Streaming queue track: %s (%s) for queue %s on player %s"
1726            "- crossfade mode: %s "
1727            "- crossfading from previous track: %s ",
1728            queue_item.streamdetails.uri if queue_item.streamdetails else "Unknown URI",
1729            queue_item.name,
1730            queue.display_name,
1731            player.name,
1732            crossfade_mode,
1733            "true" if crossfade_data else "false",
1734        )
1735
1736        buffer = bytearray()
1737        bytes_written = 0
1738        # calculate crossfade buffer size
1739        crossfade_buffer_duration = (
1740            SMART_CROSSFADE_DURATION
1741            if crossfade_mode == CrossfadeMode.SMART_CROSSFADE
1742            else standard_crossfade_duration
1743        )
1744        crossfade_buffer_duration = min(
1745            crossfade_buffer_duration,
1746            int(streamdetails.duration / 2)
1747            if streamdetails.duration
1748            else crossfade_buffer_duration,
1749        )
1750        # skip crossfade if buffer would be too small to be meaningful
1751        if crossfade_buffer_duration < 5:
1752            crossfade_buffer_duration = 0
1753        # Ensure crossfade buffer size is aligned to frame boundaries
1754        # Frame size = bytes_per_sample * channels
1755        bytes_per_sample = pcm_format.bit_depth // 8
1756        frame_size = bytes_per_sample * pcm_format.channels
1757        crossfade_buffer_size = int(pcm_format.pcm_sample_size * crossfade_buffer_duration)
1758        # Round down to nearest frame boundary
1759        crossfade_buffer_size = (crossfade_buffer_size // frame_size) * frame_size
1760        fade_out_data: bytes | None = None
1761        uncredited_tail_bytes = 0
1762
1763        # pin the body to DYNAMIC when the intro was baked DYNAMIC,
1764        # else a late measurement flips it and causes a volume jump
1765        norm_override: VolumeNormalizationMode | None = None
1766        if crossfade_data and crossfade_data.normalization_mode == VolumeNormalizationMode.DYNAMIC:
1767            norm_override = VolumeNormalizationMode.DYNAMIC
1768
1769        if crossfade_data:
1770            # reported media-time (TRIM + CF) is decoupled from the raw buffer seek below (X)
1771            streamdetails.seek_position = crossfade_data.elapsed_time_offset
1772            # yield the POST portion (resample if previous track's format differs)
1773            if crossfade_data.pcm_format != pcm_format:
1774                async for _chunk in resample_pcm_audio(
1775                    crossfade_data.data, crossfade_data.pcm_format, pcm_format
1776                ):
1777                    yield _chunk
1778                    bytes_written += len(_chunk)
1779            else:
1780                for pcm_slice in iter_pcm_slices(crossfade_data.data, pcm_format, 1000):
1781                    yield pcm_slice
1782                    await asyncio.sleep(0)
1783                bytes_written += len(crossfade_data.data)
1784            # skip past the audio already consumed by the crossfade
1785            fade_in_duration_seconds = (
1786                crossfade_data.fade_in_size / crossfade_data.fade_in_pcm_format.pcm_sample_size
1787            )
1788            discard_seconds = int(fade_in_duration_seconds)
1789            fractional_seconds = fade_in_duration_seconds - discard_seconds
1790            discard_leftover = int(fractional_seconds * pcm_format.pcm_sample_size)
1791            discard_leftover = (discard_leftover // frame_size) * frame_size
1792            crossfade_data = None
1793            self._crossfade_data.pop(queue.queue_id, None)
1794        else:
1795            discard_seconds = int(streamdetails.seek_position)
1796            discard_leftover = 0
1797
1798        # Yield the first WARMUP_DURATION worth of audio immediately so playback starts
1799        # right away. After that, start accumulating the crossfade holdback buffer.
1800        warmup_size = int(pcm_format.pcm_sample_size * WARMUP_DURATION)
1801        warmup_bytes = 0
1802        total_chunks_received = 0
1803        playback_speed = cast("float", queue_item.extra_attributes.get("playback_speed", 1.0))
1804        async for chunk in self.get_queue_item_stream(
1805            queue_item,
1806            pcm_format,
1807            seek_position=discard_seconds,
1808            playback_speed=playback_speed,
1809            normalization_override=norm_override,
1810            session_id=session_id,
1811        ):
1812            total_chunks_received += 1
1813            if discard_leftover:
1814                chunk = chunk[discard_leftover:]  # noqa: PLW2901
1815                discard_leftover = 0
1816
1817            if warmup_bytes < warmup_size:
1818                # warmup: yield directly, don't buffer
1819                yield chunk
1820                warmup_bytes += len(chunk)
1821                bytes_written += len(chunk)
1822                del chunk
1823                continue
1824
1825            buffer.extend(chunk)
1826            del chunk
1827            if len(buffer) < crossfade_buffer_size:
1828                await asyncio.sleep(0)
1829                continue
1830            # yield everything above the crossfade buffer
1831            while len(buffer) > crossfade_buffer_size:
1832                yield bytes(buffer[: pcm_format.pcm_sample_size])
1833                bytes_written += pcm_format.pcm_sample_size
1834                del buffer[: pcm_format.pcm_sample_size]
1835                await asyncio.sleep(0)
1836
1837        #### HANDLE END OF TRACK
1838
1839        # get next track for crossfade
1840        crossfade_start_time = asyncio.get_event_loop().time()
1841        next_queue_item: QueueItem | None
1842        try:
1843            self.logger.debug(
1844                "Preloading NEXT track for crossfade for queue %s", queue.display_name
1845            )
1846            next_queue_item = await self.mass.player_queues.load_next_queue_item(
1847                queue.queue_id, queue_item.queue_item_id
1848            )
1849            # set index_in_buffer to prevent our next track is overwritten while preloading
1850            if next_queue_item.streamdetails is None:
1851                raise InvalidDataError(
1852                    f"No streamdetails for next queue item {next_queue_item.queue_item_id}"
1853                )
1854            queue.index_in_buffer = self.mass.player_queues.index_by_id(
1855                queue.queue_id, next_queue_item.queue_item_id
1856            )
1857        except QueueEmpty:
1858            # end of queue reached, no next item
1859            next_queue_item = None
1860
1861        crossfade_allowed = False
1862        if next_queue_item and next_queue_item.streamdetails:
1863            next_pcm = await self.select_pcm_format(
1864                player=player,
1865                streamdetails=next_queue_item.streamdetails,
1866                crossfade_enabled=True,
1867            )
1868            crossfade_allowed = self.crossfade_allowed(
1869                queue_item,
1870                crossfade_mode=crossfade_mode,
1871                player_id=player.player_id,
1872                flow_mode=False,
1873                next_queue_item=next_queue_item,
1874                sample_rate=pcm_format.sample_rate,
1875                next_sample_rate=next_pcm.sample_rate,
1876            )
1877        if not crossfade_allowed:
1878            # no crossfade enabled/allowed, just yield the buffer last part
1879            bytes_written += len(buffer)
1880            for pcm_slice in iter_pcm_slices(bytes(buffer), pcm_format, 1000):
1881                yield pcm_slice
1882                await asyncio.sleep(0)
1883        else:
1884            assert next_queue_item is not None
1885            # the remaining buffer is the fade-out tail of the current track
1886            fade_out_data = bytes(buffer)
1887            buffer = bytearray()
1888            # initialized before the try block — the except handler reads these
1889            first_part_written = 0
1890            second_part_buf = bytearray()
1891            try:
1892                # wrap the next track's stream in a counting generator that caps
1893                # at crossfade_buffer_size and tracks how many bytes were consumed
1894                fade_in_bytes_consumed = 0
1895
1896                _next_item = next_queue_item
1897
1898                async def _limited_fade_in() -> AsyncGenerator[bytes]:
1899                    nonlocal fade_in_bytes_consumed
1900                    async for chunk in self.get_queue_item_stream(
1901                        _next_item,
1902                        pcm_format,
1903                        playback_speed=cast(
1904                            "float", _next_item.extra_attributes.get("playback_speed", 1.0)
1905                        ),
1906                        session_id=session_id,
1907                    ):
1908                        remaining = crossfade_buffer_size - fade_in_bytes_consumed
1909                        if remaining <= 0:
1910                            break
1911                        if len(chunk) > remaining:
1912                            fade_in_bytes_consumed += remaining
1913                            yield chunk[:remaining]
1914                            break
1915                        fade_in_bytes_consumed += len(chunk)
1916                        yield chunk
1917
1918                smart_fade = await self.smart_fades_mixer.build(
1919                    fade_in_streamdetails=cast("StreamDetails", next_queue_item.streamdetails),
1920                    fade_out_streamdetails=streamdetails,
1921                    pcm_format=pcm_format,
1922                    standard_crossfade_duration=standard_crossfade_duration,
1923                    mode=crossfade_mode,
1924                    fade_out_data=fade_out_data,
1925                    fade_in_bytes_len=crossfade_buffer_size,
1926                )
1927                crossfade_timing = smart_fade.timing_info
1928                # Split mix output at end-of-overlap: PRE+CF to A, POST to B's intro.
1929                fadeout_share_bytes = int(
1930                    (crossfade_timing.pre_crossfade_duration + crossfade_timing.crossfade_duration)
1931                    * pcm_format.pcm_sample_size
1932                )
1933                fadeout_share_bytes = (fadeout_share_bytes // frame_size) * frame_size
1934                async for mix_chunk in self.smart_fades_mixer.mix(
1935                    smart_fade,
1936                    fade_in_part=_limited_fade_in(),
1937                    fade_out_part=fade_out_data,
1938                    pcm_format=pcm_format,
1939                ):
1940                    if first_part_written < fadeout_share_bytes:
1941                        # split this chunk so A gets exactly fadeout_share_bytes
1942                        remaining = fadeout_share_bytes - first_part_written
1943                        if len(mix_chunk) > remaining:
1944                            yield mix_chunk[:remaining]
1945                            first_part_written += remaining
1946                            bytes_written += remaining
1947                            second_part_buf.extend(mix_chunk[remaining:])
1948                        else:
1949                            yield mix_chunk
1950                            first_part_written += len(mix_chunk)
1951                            bytes_written += len(mix_chunk)
1952                    else:
1953                        second_part_buf.extend(mix_chunk)
1954                # tail consumed by the mix but not credited to bytes_written
1955                uncredited_tail_bytes = len(fade_out_data) - first_part_written
1956                self._crossfade_data[queue_item.queue_id] = CrossfadeData(
1957                    data=bytes(second_part_buf),
1958                    fade_in_size=fade_in_bytes_consumed,
1959                    pcm_format=pcm_format,
1960                    fade_in_pcm_format=pcm_format,
1961                    queue_item_id=next_queue_item.queue_item_id,
1962                    elapsed_time_offset=(
1963                        crossfade_timing.fadein_trimmed_duration
1964                        + crossfade_timing.crossfade_duration
1965                    ),
1966                    normalization_mode=cast(
1967                        "StreamDetails", next_queue_item.streamdetails
1968                    ).volume_normalization_mode,
1969                )
1970                crossfade_elapsed = asyncio.get_event_loop().time() - crossfade_start_time
1971                self.logger.debug(
1972                    "Stored crossfade data for queue %s"
1973                    " - next queue_item_id: %s (preparation took %.1fs)",
1974                    queue.display_name,
1975                    next_queue_item.queue_item_id,
1976                    crossfade_elapsed,
1977                )
1978            except Exception as err:
1979                if first_part_written or second_part_buf:
1980                    # partial mix already played — concat'd fade_out_data would duplicate audio
1981                    raise
1982                # crossfade failed, fall back to just yielding the fade_out_data
1983                self.logger.warning(
1984                    "Crossfade failed for queue %s: %s",
1985                    queue.display_name,
1986                    err,
1987                )
1988                next_queue_item = None
1989                for pcm_slice in iter_pcm_slices(fade_out_data, pcm_format, 1000):
1990                    yield pcm_slice
1991                    await asyncio.sleep(0)
1992                bytes_written += len(fade_out_data)
1993                del fade_out_data
1994        # make sure the buffer gets cleaned up
1995        del buffer
1996        # update duration details based on the actual pcm data we sent
1997        # this also accounts for crossfade and silence stripping
1998        seconds_streamed = bytes_written / pcm_format.pcm_sample_size
1999        streamdetails.seconds_streamed = seconds_streamed
2000        uncredited_tail_seconds = uncredited_tail_bytes / pcm_format.pcm_sample_size
2001        # streamdetails.duration is in media-time; seconds_streamed is stream-time
2002        # (post-atempo), so we scale by playback_speed to recover media-time.
2003        streamdetails.duration = int(
2004            streamdetails.seek_position
2005            + (seconds_streamed + uncredited_tail_seconds) * playback_speed
2006        )
2007        # propagate accurate duration to queue_item so UI displays it
2008        queue_item.duration = streamdetails.duration
2009        self.logger.debug(
2010            "Finished Streaming queue track: %s (%s) on queue %s "
2011            "- crossfade data prepared for next track: %s",
2012            streamdetails.uri,
2013            queue_item.name,
2014            queue.display_name,
2015            next_queue_item.name if next_queue_item else "N/A",
2016        )
2017
2018    async def get_queue_flow_stream(
2019        self,
2020        queue: PlayerQueue,
2021        start_queue_item: QueueItem,
2022        pcm_format: AudioFormat,
2023        session_id: str | None = None,
2024        protocol_player: Player | None = None,
2025    ) -> AsyncGenerator[bytes]:
2026        """
2027        Get a flow stream of all tracks in the queue as raw PCM audio.
2028
2029        yields chunks of exactly 1 second of audio in the given pcm_format.
2030
2031        :param queue: Queue being streamed.
2032        :param start_queue_item: First queue item in the flow stream.
2033        :param pcm_format: Shared PCM format for the complete flow stream.
2034        :param session_id: Queue session that owns processing-detail updates.
2035        :param protocol_player: The protocol player actually consuming the flow stream.
2036            Must be the same player that was used to select ``pcm_format`` so
2037            restart decisions are made against the correct supported sample rates
2038            and flow mode configuration. Falls back to the queue's player when omitted.
2039        """
2040        # ruff: noqa: PLR0915
2041        assert pcm_format.content_type.is_pcm()
2042        queue_track = None
2043        last_fadeout_part: bytes = b""
2044        last_streamdetails: StreamDetails | None = None
2045        last_play_log_entry: PlayLogEntry | None = None
2046        # Snapshot the queue's current session_id. PlayerQueues rotates this on
2047        # every new stream session, so if a newer producer takes over the queue
2048        # (rapid track switch, sync-group reform, dynamic leader handoff) the
2049        # snapshot will no longer match and we exit cleanly on the next yield or
2050        # playlog append — preventing two producers from writing to the same
2051        # pq_data.flow_mode_stream_log.
2052        pq_data = self.mass.player_queues.queue_data(queue.queue_id)
2053        flow_session_id = session_id or pq_data.session_id
2054        if flow_session_id is None or pq_data.session_id != flow_session_id:
2055            self.logger.debug(
2056                "Ignoring stale flow stream for queue %s (session %s, active %s)",
2057                queue.display_name,
2058                flow_session_id,
2059                pq_data.session_id,
2060            )
2061            return
2062        queue.flow_mode = True
2063        pq_data.flow_mode_stream_log = []
2064        if not start_queue_item:
2065            # this can happen in some (edge case) race conditions
2066            return
2067        pcm_sample_size = pcm_format.pcm_sample_size
2068        if start_queue_item.media_type != MediaType.TRACK:
2069            # no crossfade on non-tracks
2070            crossfade_mode = CrossfadeMode.DISABLED
2071            standard_crossfade_duration = 0
2072        else:
2073            crossfade_mode = self.mass.streams.get_crossfade_mode(queue)
2074            # crossfade duration is a global (queue controller) setting; fallback matches
2075            # CONF_ENTRY_CROSSFADE_DURATION's default
2076            standard_crossfade_duration = self.mass.config.get_raw_core_config_value(
2077                CONF_PLAYER_QUEUES, CONF_CROSSFADE_DURATION, 8
2078            )
2079        flow_mode_sample_rate_conf, flow_supported_sample_rates = self._flow_restart_context(
2080            queue.queue_id, protocol_player
2081        )
2082        # note: get_crossfade_mode() already falls back to standard when smart fades aren't
2083        # available (no analysis provider / minimal buffer), so crossfade_mode is safe to use.
2084        self.logger.info(
2085            "Start Queue Flow stream for Queue %s - crossfade: %s %s",
2086            queue.display_name,
2087            crossfade_mode,
2088            f"({standard_crossfade_duration}s)"
2089            if crossfade_mode == CrossfadeMode.STANDARD_CROSSFADE
2090            else "",
2091        )
2092        total_chunks_received = 0
2093
2094        def _superseded() -> bool:
2095            """Return True if a newer stream session has taken over this queue."""
2096            return pq_data.session_id != flow_session_id
2097
2098        queue_exhausted = False
2099        while True:
2100            # bail out early if a newer producer has taken over this queue,
2101            # so we don't append another entry to a stream log we no longer own
2102            if _superseded():
2103                self.logger.debug(
2104                    "Flow stream for queue %s superseded (session %s -> %s) "
2105                    "- exiting before next track",
2106                    queue.display_name,
2107                    flow_session_id,
2108                    pq_data.session_id,
2109                )
2110                return
2111            # get (next) queue item to stream
2112            if queue_track is None:
2113                queue_track = start_queue_item
2114            else:
2115                try:
2116                    queue_track = await self.mass.player_queues.load_next_queue_item(
2117                        queue.queue_id, queue_track.queue_item_id
2118                    )
2119                except QueueEmpty:
2120                    queue_exhausted = True
2121                    break
2122
2123            if self._flow_stream_needs_restart(
2124                queue_track,
2125                pcm_format,
2126                flow_supported_sample_rates,
2127                flow_mode_sample_rate_conf,
2128                is_first_track=queue_track is start_queue_item,
2129            ):
2130                break
2131
2132            if queue_track.streamdetails is None:
2133                self.logger.error(
2134                    "No StreamDetails for queue item %s (%s) on queue %s - skipping track",
2135                    queue_track.queue_item_id,
2136                    queue_track.name,
2137                    queue.display_name,
2138                )
2139                continue
2140            if flow_session_id is not None:
2141                self.mass.streams.audio_processing.update_item_context(
2142                    queue_id=queue.queue_id,
2143                    session_id=flow_session_id,
2144                    queue_item_id=queue_track.queue_item_id,
2145                    queue_processing=AudioQueueProcessing(
2146                        pcm_format=pcm_format,
2147                        playback_speed=cast(
2148                            "float",
2149                            queue_track.extra_attributes.get("playback_speed", 1.0),
2150                        ),
2151                        crossfade_mode=crossfade_mode,
2152                        overlay_active=overlay_active(queue),
2153                    ),
2154                    alters_audio=queue_track.streamdetails.fade_in,
2155                )
2156
2157            self.logger.debug(
2158                "Start Streaming queue track: %s (%s) for queue %s",
2159                queue_track.streamdetails.uri,
2160                queue_track.name,
2161                queue.display_name,
2162            )
2163            # last chance to bail before mutating the stream log: a newer producer
2164            # may have taken over while we were awaiting load_next_queue_item
2165            if _superseded():
2166                self.logger.debug(
2167                    "Flow stream for queue %s superseded - exiting before playlog append",
2168                    queue.display_name,
2169                )
2170                return
2171            track_playback_speed = cast(
2172                "float", queue_track.extra_attributes.get("playback_speed", 1.0)
2173            )
2174            # calculate crossfade buffer size
2175            crossfade_buffer_duration = (
2176                SMART_CROSSFADE_DURATION
2177                if crossfade_mode == CrossfadeMode.SMART_CROSSFADE
2178                else standard_crossfade_duration
2179            )
2180            crossfade_buffer_duration = min(
2181                crossfade_buffer_duration,
2182                int(queue_track.streamdetails.duration / 2)
2183                if queue_track.streamdetails.duration
2184                else crossfade_buffer_duration,
2185            )
2186            # skip crossfade if buffer would be too small to be meaningful
2187            if crossfade_buffer_duration < 5:
2188                crossfade_buffer_duration = 0
2189            # Ensure crossfade buffer size is aligned to frame boundaries
2190            # Frame size = bytes_per_sample * channels
2191            bytes_per_sample = pcm_format.bit_depth // 8
2192            frame_size = bytes_per_sample * pcm_format.channels
2193            crossfade_buffer_size = int(pcm_format.pcm_sample_size * crossfade_buffer_duration)
2194            # Round down to nearest frame boundary
2195            crossfade_buffer_size = (crossfade_buffer_size // frame_size) * frame_size
2196            warmup_size = int(pcm_format.pcm_sample_size * WARMUP_DURATION)
2197
2198            # raw_seek_position feeds the PCM buffer; streamdetails.seek_position
2199            # (overwritten below) only drives reported elapsed time.
2200            raw_seek_position = queue_track.streamdetails.seek_position
2201            # Build eagerly so seek_position is set before PlayLogEntry is appended —
2202            # consumer-paced mix() would otherwise let the queue briefly report 0.
2203            crossfade_smart_fade: SmartFade | None = None
2204            if (
2205                last_fadeout_part
2206                and last_streamdetails
2207                and crossfade_buffer_size > 0
2208                and crossfade_mode != CrossfadeMode.DISABLED
2209            ):
2210                crossfade_smart_fade = await self.smart_fades_mixer.build(
2211                    fade_in_streamdetails=queue_track.streamdetails,
2212                    fade_out_streamdetails=last_streamdetails,
2213                    pcm_format=pcm_format,
2214                    standard_crossfade_duration=standard_crossfade_duration,
2215                    mode=crossfade_mode,
2216                    fade_out_data=last_fadeout_part,
2217                    fade_in_bytes_len=crossfade_buffer_size,
2218                )
2219                timing_info = crossfade_smart_fade.timing_info
2220                queue_track.streamdetails.seek_position = (
2221                    raw_seek_position
2222                    + timing_info.fadein_trimmed_duration
2223                    + timing_info.crossfade_duration
2224                )
2225            # append to play log so the queue controller can work out which track is playing
2226            play_log_entry = PlayLogEntry(queue_track.queue_item_id)
2227            pq_data.flow_mode_stream_log.append(play_log_entry)
2228
2229            bytes_written = 0
2230            crossfade_buffer = bytearray()
2231            warmup_bytes = 0
2232            first_chunk_received = False
2233
2234            async for chunk in self.get_queue_item_stream(
2235                queue_track,
2236                pcm_format=pcm_format,
2237                seek_position=int(raw_seek_position),
2238                playback_speed=cast(
2239                    "float", queue_track.extra_attributes.get("playback_speed", 1.0)
2240                ),
2241                raise_on_error=False,
2242                session_id=flow_session_id,
2243            ):
2244                # if a newer producer has taken over this queue, stop sending
2245                # audio and exit cleanly before the outer-loop end-of-track
2246                # bookkeeping mutates seconds_streamed / duration on the log
2247                if _superseded():
2248                    self.logger.debug(
2249                        "Flow stream for queue %s superseded - stopping chunk yield",
2250                        queue.display_name,
2251                    )
2252                    return
2253                total_chunks_received += 1
2254                if not first_chunk_received:
2255                    first_chunk_received = True
2256                    # inform the queue that the track is now loaded in the buffer
2257                    # so the next track can be preloaded
2258                    self.mass.player_queues.track_loaded_in_buffer(
2259                        queue.queue_id, queue_track.queue_item_id
2260                    )
2261
2262                if crossfade_mode == CrossfadeMode.DISABLED:
2263                    # no cross/smart fade: yield chunks directly without intermediate buffer
2264                    yield chunk
2265                    bytes_written += len(chunk)
2266                    del chunk
2267                    continue
2268
2269                # Warmup: yield chunks directly until we have streamed WARMUP_DURATION
2270                # worth of audio, so playback starts immediately. Skip warmup when
2271                # crossfade data from the previous track is pending — we need a full
2272                # buffer for the mix.
2273                if warmup_bytes < warmup_size and not last_fadeout_part:
2274                    yield chunk
2275                    warmup_bytes += len(chunk)
2276                    bytes_written += len(chunk)
2277                    del chunk
2278                    continue
2279
2280                # smart fades enabled: accumulate chunks in crossfade buffer
2281                crossfade_buffer.extend(chunk)
2282                del chunk
2283                if len(crossfade_buffer) < crossfade_buffer_size:
2284                    await asyncio.sleep(0)
2285                    continue
2286
2287                # handle crossfade of previous track and new track
2288                if (
2289                    last_fadeout_part
2290                    and last_streamdetails
2291                    and crossfade_smart_fade is not None
2292                    and last_play_log_entry is not None
2293                ):
2294                    fadein_part = bytes(crossfade_buffer[:crossfade_buffer_size])
2295                    remaining_bytes = bytes(crossfade_buffer[crossfade_buffer_size:])
2296                    try:
2297                        crossfade_bytes_written = 0
2298                        async for mix_chunk in self.smart_fades_mixer.mix(
2299                            crossfade_smart_fade,
2300                            fade_in_part=fadein_part,
2301                            fade_out_part=last_fadeout_part,
2302                            pcm_format=pcm_format,
2303                        ):
2304                            yield mix_chunk
2305                            crossfade_bytes_written += len(mix_chunk)
2306                    except Exception as mix_err:
2307                        if crossfade_bytes_written:
2308                            # partial mix already played — concat'd tail would duplicate audio
2309                            raise
2310                        self.logger.warning(
2311                            "Crossfade mixer failed for %s, falling back to simple concat: %s",
2312                            queue_track.name,
2313                            mix_err,
2314                        )
2315                        for pcm_slice in iter_pcm_slices(last_fadeout_part, pcm_format, 1000):
2316                            yield pcm_slice
2317                            await asyncio.sleep(0)
2318                        # full tail was pre-counted and is now yielded as-is
2319                        crossfade_bytes_written = 0
2320                        remaining_bytes = bytes(crossfade_buffer)
2321                        # mix failed — undo the eager seek_position
2322                        queue_track.streamdetails.seek_position = raw_seek_position
2323                    if crossfade_bytes_written:
2324                        # Split mix output at end-of-overlap: PRE+CF to A, POST to B.
2325                        fadeout_share_seconds = (
2326                            timing_info.pre_crossfade_duration + timing_info.crossfade_duration
2327                        )
2328                        fadeout_share = int(fadeout_share_seconds * pcm_sample_size)
2329                        fadeout_share = (fadeout_share // frame_size) * frame_size
2330                        fadeout_share = min(fadeout_share, crossfade_bytes_written)
2331                        fadein_share = crossfade_bytes_written - fadeout_share
2332                        bytes_written += fadein_share
2333                        if last_play_log_entry:
2334                            assert last_play_log_entry.seconds_streamed is not None
2335                            # correct pre-counted full tail to the timing-based share
2336                            last_play_log_entry.seconds_streamed += (
2337                                fadeout_share - len(last_fadeout_part)
2338                            ) / pcm_sample_size
2339                    if remaining_bytes:
2340                        for pcm_slice in iter_pcm_slices(remaining_bytes, pcm_format, 1000):
2341                            yield pcm_slice
2342                            await asyncio.sleep(0)
2343                        bytes_written += len(remaining_bytes)
2344                        del remaining_bytes
2345                    last_fadeout_part = b""
2346                    last_streamdetails = None
2347                    crossfade_buffer = bytearray()
2348                    warmup_bytes = 0
2349
2350                # yield everything above the crossfade buffer size
2351                while len(crossfade_buffer) > crossfade_buffer_size:
2352                    yield bytes(crossfade_buffer[:pcm_sample_size])
2353                    bytes_written += pcm_sample_size
2354                    del crossfade_buffer[:pcm_sample_size]
2355                    await asyncio.sleep(0)
2356
2357            # A source error after partial audio must not look like a completed item.
2358            # Progress reporting skips items with stream_error, so the item is not
2359            # marked played; move on to the next queue item like the zero-audio path.
2360            if first_chunk_received and queue_track.streamdetails.stream_error:
2361                if _superseded():
2362                    return
2363                self.logger.warning(
2364                    "Track %s (%s) on queue %s aborted by a stream error - skipping",
2365                    queue_track.name,
2366                    queue_track.streamdetails.uri,
2367                    queue.display_name,
2368                )
2369                # the audio sent so far will still play out; keep the play log entry
2370                # honest about how much of this item was actually streamed
2371                play_log_entry.seconds_streamed = bytes_written / pcm_sample_size
2372                if last_fadeout_part:
2373                    # crossfade into this item never happened — undo the eager seek_position
2374                    queue_track.streamdetails.seek_position = raw_seek_position
2375                continue
2376
2377            #### HANDLE END OF TRACK
2378            if not first_chunk_received:
2379                self.logger.warning(
2380                    "Track %s (%s) on queue %s produced no audio data - skipping",
2381                    queue_track.name,
2382                    queue_track.streamdetails.uri if queue_track.streamdetails else "unknown",
2383                    queue.display_name,
2384                )
2385                queue_track.streamdetails.stream_error = True
2386                play_log_entry.seconds_streamed = 0
2387                continue
2388            if last_fadeout_part:
2389                # edge case: we did not get enough data to make the crossfade
2390                # attribute these bytes to the previous track (they are its tail)
2391                for pcm_slice in iter_pcm_slices(last_fadeout_part, pcm_format, 1000):
2392                    yield pcm_slice
2393                    await asyncio.sleep(0)
2394                # no crossfade happened — undo the eager seek_position
2395                queue_track.streamdetails.seek_position = raw_seek_position
2396                # full tail was pre-counted and is now yielded as-is
2397                last_fadeout_part = b""
2398            if self.crossfade_allowed(
2399                queue_track,
2400                crossfade_mode=crossfade_mode,
2401                player_id=queue.queue_id,
2402                flow_mode=True,
2403            ):
2404                last_fadeout_part = bytes(crossfade_buffer[-crossfade_buffer_size:])
2405                last_streamdetails = queue_track.streamdetails
2406                last_play_log_entry = play_log_entry
2407                remaining_bytes = bytes(crossfade_buffer[:-crossfade_buffer_size])
2408                if remaining_bytes:
2409                    for pcm_slice in iter_pcm_slices(remaining_bytes, pcm_format, 1000):
2410                        yield pcm_slice
2411                        await asyncio.sleep(0)
2412                    bytes_written += len(remaining_bytes)
2413                del remaining_bytes
2414            elif crossfade_mode != CrossfadeMode.DISABLED and crossfade_buffer:
2415                bytes_written += len(crossfade_buffer)
2416                for pcm_slice in iter_pcm_slices(bytes(crossfade_buffer), pcm_format, 1000):
2417                    yield pcm_slice
2418                    await asyncio.sleep(0)
2419            crossfade_buffer = bytearray()
2420
2421            # update duration details based on the actual pcm data we sent
2422            # this also accounts for crossfade and silence stripping
2423            seconds_streamed = bytes_written / pcm_sample_size
2424            queue_track.streamdetails.seconds_streamed = seconds_streamed
2425            # the held-back crossfade tail still counts as this track's media-time
2426            tail_seconds = len(last_fadeout_part) / pcm_sample_size
2427            # streamdetails.duration is in media-time; seconds_streamed is stream-time
2428            # (post-atempo), so we scale by the track's playback_speed to recover media-time.
2429            queue_track.streamdetails.duration = int(
2430                queue_track.streamdetails.seek_position
2431                + (seconds_streamed + tail_seconds) * track_playback_speed
2432            )
2433            # propagate accurate duration to queue_item so UI displays it
2434            queue_track.duration = queue_track.streamdetails.duration
2435            play_log_entry.seconds_streamed = seconds_streamed
2436            play_log_entry.duration = queue_track.streamdetails.duration
2437            if last_play_log_entry is play_log_entry and last_fadeout_part:
2438                # Pre-count the full crossfade tail so the queue index calculation
2439                # doesn't undercount while waiting for the next track's crossfade mix.
2440                # This will be corrected to crossfade_total/2 once the mix completes.
2441                assert play_log_entry.seconds_streamed is not None
2442                play_log_entry.seconds_streamed += len(last_fadeout_part) / pcm_sample_size
2443            self.logger.debug(
2444                "Finished Streaming queue track: %s (%s) on queue %s",
2445                queue_track.streamdetails.uri,
2446                queue_track.name,
2447                queue.display_name,
2448            )
2449        #### HANDLE END OF QUEUE FLOW STREAM
2450        # skip end-of-queue bookkeeping if a newer producer has superseded us;
2451        # the new producer owns queue_buffer_completed and the play log now
2452        if _superseded():
2453            self.logger.debug(
2454                "Flow stream for queue %s superseded - skipping end-of-queue handling",
2455                queue.display_name,
2456            )
2457            return
2458        # end of queue flow: make sure we yield the last_fadeout_part
2459        if last_fadeout_part:
2460            for pcm_slice in iter_pcm_slices(last_fadeout_part, pcm_format, 1000):
2461                yield pcm_slice
2462                await asyncio.sleep(0)
2463            # correct seconds streamed - the duration already includes the tail
2464            last_part_seconds = len(last_fadeout_part) / pcm_sample_size
2465            streamdetails = queue_track.streamdetails
2466            assert streamdetails is not None
2467            streamdetails.seconds_streamed = (
2468                streamdetails.seconds_streamed or 0
2469            ) + last_part_seconds
2470            # also update the play log entry so elapsed time tracking stays in sync
2471            if last_play_log_entry:
2472                assert last_play_log_entry.seconds_streamed is not None
2473                # full tail was pre-counted and is now yielded as-is
2474                last_play_log_entry.duration = streamdetails.duration
2475            last_fadeout_part = b""
2476        self.logger.info("Finished Queue Flow stream for Queue %s", queue.display_name)
2477        # only signal completion if we are still the active producer — a later
2478        # producer would (incorrectly) see this as its own completion otherwise
2479        if not _superseded():
2480            # inform the queue controller that all audio data has been generated
2481            # so it can handle the case where new items were added after the flow stream ended
2482            self.mass.player_queues.queue_buffer_completed(queue.queue_id, queue_exhausted)
2483
2484    async def get_overlay_mixed_stream(
2485        self,
2486        queue: PlayerQueue,
2487        audio_input: AsyncGenerator[bytes],
2488        pcm_format: AudioFormat,
2489    ) -> AsyncGenerator[bytes]:
2490        """
2491        Mix the queue's audio overlay (looping sound effect) into the given PCM stream.
2492
2493        The mixed output has the exact same PCM format, duration and chunking as the
2494        input stream. If the overlay source can not be resolved, the original stream
2495        is passed through unchanged so playback is never interrupted.
2496
2497        :param queue: The PlayerQueue holding the overlay source and volume.
2498        :param audio_input: The audio stream (raw PCM in ``pcm_format``) to mix into.
2499        :param pcm_format: PCM format of both the input and the mixed output.
2500        """
2501        overlay_input = await self._resolve_overlay_input(queue)
2502        if overlay_input is None:
2503            # overlay source unavailable: degrade gracefully to music-only
2504            async for chunk in audio_input:
2505                yield chunk
2506            return
2507        async for chunk in get_ffmpeg_overlay_stream(
2508            audio_input=audio_input,
2509            overlay_input=overlay_input,
2510            pcm_format=pcm_format,
2511            overlay_volume=queue.overlay_volume,
2512            chunk_size=pcm_format.pcm_sample_size,
2513        ):
2514            yield chunk
2515
2516    def crossfade_allowed(
2517        self,
2518        queue_item: QueueItem,
2519        crossfade_mode: CrossfadeMode,
2520        player_id: str,
2521        flow_mode: bool = False,
2522        next_queue_item: QueueItem | None = None,
2523        sample_rate: int | None = None,
2524        next_sample_rate: int | None = None,
2525    ) -> bool:
2526        """Get the crossfade config for a queue item."""
2527        if crossfade_mode == CrossfadeMode.DISABLED:
2528            return False
2529        if not (self.mass.player_queues.get(queue_item.queue_id)):
2530            return False  # just a guard
2531        if not (self.mass.players.get_player(player_id)):
2532            return False  # just a guard
2533        if queue_item.media_type != MediaType.TRACK:
2534            self.logger.debug("Skipping crossfade: current item is not a track")
2535            return False
2536        # check if the next item is part of the same album
2537        next_item = next_queue_item or self.mass.player_queues.get_next_item(
2538            queue_item.queue_id, queue_item.queue_item_id
2539        )
2540        if not next_item:
2541            # there is no next item!
2542            return False
2543        # check if next item is a track
2544        if next_item.media_type != MediaType.TRACK:
2545            self.logger.debug("Skipping crossfade: next item is not a track")
2546            return False
2547        if (
2548            isinstance(queue_item.media_item, Track)
2549            and isinstance(next_item.media_item, Track)
2550            and queue_item.media_item.album
2551            and next_item.media_item.album
2552            and queue_item.media_item.album == next_item.media_item.album
2553            and not self.mass.config.get_raw_core_config_value(
2554                "streams", CONF_ALLOW_CROSSFADE_SAME_ALBUM, False
2555            )
2556        ):
2557            # in general, crossfade is not desired for tracks of the same (gapless) album
2558            # because we have no accurate way to determine if the album is gapless or not,
2559            # for now we just never crossfade between tracks of the same album
2560            self.logger.debug("Skipping crossfade: next item is part of the same album")
2561            return False
2562
2563        # check if we're allowed to crossfade on different sample rates
2564        if (
2565            not flow_mode
2566            and sample_rate
2567            and next_sample_rate
2568            and sample_rate != next_sample_rate
2569            and not self.mass.config.get_raw_player_config_value(
2570                player_id,
2571                CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES.key,
2572                CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES.default_value,
2573            )
2574        ):
2575            self.logger.debug(
2576                "Skipping crossfade: player(protocol) does not support gapless playback "
2577                "with different sample rates (%s vs %s)",
2578                sample_rate,
2579                next_sample_rate,
2580            )
2581            return False
2582
2583        return True
2584
2585    def clear_crossfade_data(self, queue_id: str) -> None:
2586        """
2587        Clear any pending crossfade data for a queue.
2588
2589        :param queue_id: The queue ID to clear crossfade data for.
2590        """
2591        if queue_id in self._crossfade_data:
2592            self.logger.debug("Clearing crossfade data for queue %s", queue_id)
2593            del self._crossfade_data[queue_id]
2594
2595    async def get_shoutcast_stream(
2596        self, url: str, streamdetails: StreamDetails
2597    ) -> AsyncGenerator[bytes]:
2598        """
2599        Yield audio from a legacy Shoutcast server, with ICY metadata parsed inline.
2600
2601        :param url: Shoutcast stream URL.
2602        :param streamdetails: StreamDetails to update with ICY metadata as it arrives.
2603        """
2604        self.logger.debug("Start streaming from legacy Shoutcast server: %s", url)
2605
2606        parsed = urlparse(url)
2607        host = parsed.hostname
2608        port = parsed.port or 80
2609        path = parsed.path or "/"
2610        if parsed.query:
2611            path = f"{path}?{parsed.query}"
2612
2613        try:
2614            # Open raw socket connection
2615            reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=30)
2616        except TimeoutError as err:
2617            raise AudioError(f"Timeout connecting to Shoutcast stream {url}") from err
2618        except (OSError, ConnectionError) as err:
2619            raise AudioError(f"Failed to connect to Shoutcast stream {url}") from err
2620
2621        try:
2622            # Send HTTP request with ICY metadata header
2623            request = (
2624                f"GET {path} HTTP/1.1\r\n"
2625                f"Host: {host}\r\n"
2626                f"User-Agent: {HTTP_HEADERS['User-Agent']}\r\n"
2627                f"Icy-MetaData: 1\r\n\r\n"
2628            )
2629            writer.write(request.encode())
2630            await writer.drain()
2631
2632            # Read and parse response line
2633            try:
2634                response_line = await asyncio.wait_for(reader.readline(), timeout=10)
2635            except TimeoutError as err:
2636                raise AudioError("Timeout reading Shoutcast response") from err
2637
2638            if not response_line.startswith(b"ICY"):
2639                raise InvalidDataError("Invalid Shoutcast response")
2640
2641            # Read headers until empty line
2642            headers: dict[str, str] = {}
2643            while True:
2644                try:
2645                    line = await asyncio.wait_for(reader.readline(), timeout=5)
2646                except TimeoutError as err:
2647                    raise AudioError("Timeout reading Shoutcast headers") from err
2648
2649                if line in (b"\r\n", b"\n", b""):
2650                    break
2651
2652                if b":" in line:
2653                    try:
2654                        key, value = line.decode("latin-1", errors="ignore").split(":", 1)
2655                        headers[key.strip().lower()] = value.strip()
2656                    except UnicodeDecodeError, ValueError:
2657                        continue
2658
2659            # Get metadata interval
2660            meta_int_str = headers.get("icy-metaint")
2661            if not meta_int_str:
2662                raise InvalidDataError("No icy-metaint header in Shoutcast response")
2663
2664            try:
2665                meta_int = int(meta_int_str)
2666            except ValueError as err:
2667                raise InvalidDataError("Invalid icy-metaint value") from err
2668
2669            self.logger.debug("Connected to Shoutcast stream %s (icy-metaint: %s)", url, meta_int)
2670
2671            # Stream audio data with metadata parsing
2672            while True:
2673                try:
2674                    # Read audio chunk
2675                    audio_chunk = await reader.readexactly(meta_int)
2676                    yield audio_chunk
2677
2678                    # Read metadata length
2679                    meta_byte = await reader.readexactly(1)
2680                    if meta_byte == b"\x00":
2681                        continue
2682
2683                    meta_length = ord(meta_byte) * 16
2684                    meta_data = await reader.readexactly(meta_length)
2685                    self._parse_icy_metadata(meta_data, streamdetails)
2686
2687                except asyncio.exceptions.IncompleteReadError:
2688                    # End of stream
2689                    break
2690
2691        finally:
2692            writer.close()
2693            await writer.wait_closed()
2694
2695    # --- Private methods ---
2696
2697    def _notify_provider_streamed(
2698        self, streamdetails: StreamDetails, finished: bool, seconds_streamed: float
2699    ) -> None:
2700        """Report a (mostly) streamed item back to the provider that owns it."""
2701        if not finished and seconds_streamed < 90:
2702            return
2703        provider = self.mass.get_provider(streamdetails.provider)
2704        # plugin providers serve playable items too, but on_streamed is MusicProvider-only
2705        if provider is None or provider.type != ProviderType.MUSIC:
2706            return
2707        music_prov = cast("MusicProvider", provider)
2708        self.mass.create_task(music_prov.on_streamed(streamdetails))
2709
2710    def _get_volume_normalization_preference(
2711        self, streamdetails: StreamDetails
2712    ) -> VolumeNormalizationMode:
2713        """Return the configured normalization preference for the stream's media type."""
2714        conf_key = (
2715            CONF_VOLUME_NORMALIZATION_RADIO
2716            if streamdetails.media_type == MediaType.RADIO
2717            else CONF_VOLUME_NORMALIZATION_TRACKS
2718        )
2719        return VolumeNormalizationMode(
2720            self.mass.streams.get_config_value(conf_key, return_type=str)
2721        )
2722
2723    def _update_radio_stream_metadata(
2724        self,
2725        streamdetails: StreamDetails,
2726        artist: str | None,
2727        title: str,
2728        image_url: str | None = None,
2729        album: str | None = None,
2730    ) -> None:
2731        """
2732        Update radio stream metadata and trigger artwork lookup.
2733
2734        :param streamdetails: The stream details to update.
2735        :param artist: Artist name (will be normalized).
2736        :param title: Track title (will be cleaned for display).
2737        :param image_url: Optional image URL from stream metadata.
2738        :param album: Optional album name.
2739        """
2740        station_image_url = image_url or self.mass.metadata.get_radio_stream_station_image(
2741            streamdetails
2742        )
2743        artist_normalized = (
2744            self.mass.metadata.normalize_radio_artist_name(artist) if artist else None
2745        )
2746        display_title, _ = parse_title_and_version(title, strip_for_display=True)
2747
2748        streamdetails.stream_metadata = StreamMetadata(
2749            title=display_title,
2750            artist=artist_normalized,
2751            album=album,
2752            image_url=station_image_url,
2753        )
2754        streamdetails.stream_metadata_last_updated = time.time()
2755        if streamdetails.queue_id:
2756            self.mass.player_queues.signal_update(streamdetails.queue_id)
2757
2758        # Fetch artwork in background (track, album then artist)
2759        if artist and title and not image_url:
2760            self.mass.call_later(
2761                0.2,
2762                self.mass.metadata.update_radio_stream_artwork,
2763                streamdetails,
2764                task_id=f"update_radio_artwork_{streamdetails.queue_id}",
2765            )
2766
2767    async def _cache_radio_result(
2768        self,
2769        url: str,
2770        stream_type: StreamType,
2771        resolved_url: str | None = None,
2772    ) -> tuple[str, StreamType]:
2773        """Cache and return a radio stream resolution result."""
2774        result = (resolved_url or url, stream_type)
2775        await self.mass.cache.set(
2776            url,
2777            result,
2778            expiration=3600 * 3,
2779            provider=CACHE_PROVIDER,
2780            category=CACHE_CATEGORY_RESOLVED_RADIO_URL,
2781        )
2782        return result
2783
2784    async def _handle_client_error_for_radio_stream(
2785        self, url: str, err: aiohttp.ClientError, fallback_stream_type: StreamType
2786    ) -> tuple[str, StreamType]:
2787        """Handle aiohttp client errors during radio stream resolution."""
2788        # Prefer the final post-redirect URL: aiohttp follows redirects before raising,
2789        # but the original url may just point at a redirector rather than the ICY endpoint.
2790        request_info = getattr(err, "request_info", None)
2791        validate_url = str(request_info.url) if request_info is not None else url
2792
2793        # Check if this is a Shoutcast/ICY response that aiohttp can't parse
2794        if isinstance(err, aiohttp.ClientResponseError) and "ICY" in str(err).upper():
2795            self.logger.debug(
2796                "ICY response detected for %s, validating Shoutcast stream", validate_url
2797            )
2798            if await self._validate_shoutcast_stream(validate_url):
2799                return await self._cache_radio_result(
2800                    url, StreamType.SHOUTCAST, resolved_url=validate_url
2801                )
2802            self.logger.warning(
2803                "ICY response detected but Shoutcast validation failed for %s", validate_url
2804            )
2805            return await self._cache_radio_result(
2806                url, fallback_stream_type, resolved_url=validate_url
2807            )
2808
2809        # Other aiohttp errors - might still be Shoutcast, check it
2810        self.logger.debug("aiohttp error for %s, checking if legacy Shoutcast stream", validate_url)
2811        if await self._validate_shoutcast_stream(validate_url):
2812            return await self._cache_radio_result(
2813                url, StreamType.SHOUTCAST, resolved_url=validate_url
2814            )
2815
2816        # Unknown error - still try to stream
2817        self.logger.warning(
2818            "Failed to parse radio URL %s: %s - attempting direct stream", validate_url, str(err)
2819        )
2820        return await self._cache_radio_result(url, fallback_stream_type, resolved_url=validate_url)
2821
2822    async def _iter_audio_source_pcm(
2823        self,
2824        streamdetails: StreamDetails,
2825        pcm_format: AudioFormat,
2826    ) -> AsyncGenerator[bytes]:
2827        """Yield PCM for an AudioSource, bypassing ffmpeg when formats match."""
2828        if _pcm_formats_match(streamdetails.audio_format, pcm_format):
2829            source_gen = self._open_audio_source_generator(streamdetails)
2830            async for chunk in realtime_pcm_pacer(source_gen, pcm_format):
2831                yield chunk
2832            return
2833        # format mismatch → fall back to ffmpeg for resampling (still small chunks)
2834        async for chunk in self.get_media_stream(
2835            streamdetails=streamdetails,
2836            pcm_format=pcm_format,
2837            filter_params=None,
2838            chunk_seconds=AUDIO_SOURCE_CHUNK_SECONDS,
2839        ):
2840            yield chunk
2841
2842    def _open_audio_source_generator(self, streamdetails: StreamDetails) -> AsyncGenerator[bytes]:
2843        """Open the raw PCM generator for an AudioSource (CUSTOM or NAMED_PIPE)."""
2844        if streamdetails.stream_type == StreamType.CUSTOM:
2845            provider = self.mass.get_provider(streamdetails.provider)
2846            if provider is None:
2847                raise ProviderUnavailableError(
2848                    f"Provider {streamdetails.provider} for stream is no longer available"
2849                )
2850            provider = cast("MusicProvider | PluginProvider", provider)
2851            return provider.get_audio_stream(streamdetails)
2852        if streamdetails.stream_type == StreamType.NAMED_PIPE:
2853            assert isinstance(streamdetails.path, str)  # for type checking
2854            return read_named_pipe(streamdetails.path)
2855        raise AudioError(f"Unsupported stream_type {streamdetails.stream_type} for AudioSource")
2856
2857    def _parse_icy_metadata(self, meta_data: bytes, streamdetails: StreamDetails) -> None:
2858        """
2859        Parse ICY metadata and update streamdetails.
2860
2861        Sets the cleaned stream title and, when the title parses as "Artist - Track",
2862        triggers a radio-artwork metadata update.
2863
2864        :param meta_data: Raw metadata bytes from an ICY stream chunk.
2865        :param streamdetails: StreamDetails to update with parsed title and metadata.
2866        """
2867        if not meta_data:
2868            return
2869
2870        meta_data = meta_data.rstrip(b"\0")
2871        # Match StreamTitle, handling apostrophes in titles
2872        stream_title_re = re.search(rb"StreamTitle='(.*?)';", meta_data)
2873
2874        if not stream_title_re:
2875            self.logger.log(
2876                VERBOSE_LOG_LEVEL,
2877                "ICY metadata does not contain StreamTitle field. Raw: %s",
2878                meta_data.decode("utf-8", errors="replace")[:200],
2879            )
2880            return
2881
2882        try:
2883            # in 99% of the cases the stream title is utf-8 encoded
2884            stream_title = stream_title_re.group(1).decode("utf-8")
2885        except UnicodeDecodeError:
2886            # fallback to iso-8859-1
2887            stream_title = stream_title_re.group(1).decode("iso-8859-1", errors="replace")
2888
2889        cleaned_stream_title = clean_stream_title(stream_title)
2890
2891        if not cleaned_stream_title:
2892            return
2893
2894        if cleaned_stream_title == streamdetails.stream_title:
2895            return
2896
2897        self.logger.log(VERBOSE_LOG_LEVEL, "ICY Radio streamtitle original: %s", stream_title)
2898        self.logger.log(
2899            VERBOSE_LOG_LEVEL, "ICY Radio streamtitle cleaned: %s", cleaned_stream_title
2900        )
2901        streamdetails.stream_title = cleaned_stream_title
2902
2903        # Prefer station-provided cover art from the ICY 'StreamUrl' field (when it is
2904        # an image) over the MusicBrainz artwork lookup in _update_radio_stream_metadata.
2905        image_url = self._parse_icy_image_url(meta_data)
2906
2907        # Parse the original title for structured fields first so stations that announce
2908        # an album can refine the artwork lookup; fall back to the "Artist - Track" split.
2909        album: str | None = None
2910        if parsed := parse_quoted_stream_title(stream_title):
2911            track_name, artist_name_raw, album = parsed
2912        elif " - " in cleaned_stream_title:
2913            artist_name_raw, track_name = (
2914                part.strip() for part in cleaned_stream_title.split(" - ", 1)
2915            )
2916        else:
2917            return
2918
2919        if artist_name_raw and track_name:
2920            self.logger.debug(
2921                "ICY metadata: artist='%s', track='%s', album='%s'",
2922                artist_name_raw,
2923                track_name,
2924                album,
2925            )
2926            self._update_radio_stream_metadata(
2927                streamdetails,
2928                artist=artist_name_raw,
2929                title=track_name,
2930                album=album,
2931                image_url=image_url,
2932            )
2933
2934    def _parse_icy_image_url(self, meta_data: bytes) -> str | None:
2935        """
2936        Return a PNG or JPEG cover-art URL from the ICY 'StreamUrl' field, if present.
2937
2938        :param meta_data: Raw metadata bytes from an ICY stream chunk.
2939        """
2940        # The trailing semicolon is optional to match sources that omit it.
2941        stream_url_re = re.search(rb"StreamUrl='([^']*)'", meta_data)
2942        if not stream_url_re:
2943            return None
2944        try:
2945            image_url = stream_url_re.group(1).decode("utf-8").strip()
2946        except UnicodeDecodeError:
2947            return None
2948        if not image_url:
2949            return None
2950        # StreamUrl is not a standardized artwork field (reference clients such as VLC
2951        # ignore it and it conventionally holds a station website link), so only accept
2952        # values that point at a PNG or JPEG image.
2953        parsed = urlparse(image_url)
2954        if parsed.scheme not in ("http", "https"):
2955            return None
2956        if not parsed.path.lower().endswith((".png", ".jpg", ".jpeg")):
2957            return None
2958        self.logger.debug("ICY metadata: StreamUrl image='%s'", image_url)
2959        return image_url
2960
2961    async def _validate_shoutcast_stream(self, url: str) -> bool:
2962        """
2963        Return True if the URL responds with a legacy Shoutcast "ICY 200 OK" line.
2964
2965        :param url: The URL to validate.
2966        """
2967        try:
2968            parsed = urlparse(url)
2969            host = parsed.hostname
2970            port = parsed.port or 80
2971            path = parsed.path or "/"
2972            if parsed.query:
2973                path = f"{path}?{parsed.query}"
2974
2975            # Open raw socket connection with timeout
2976            reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=10)
2977            try:
2978                # Send minimal HTTP request with ICY metadata header
2979                request = f"GET {path} HTTP/1.1\r\nHost: {host}\r\nIcy-MetaData: 1\r\n\r\n"
2980                writer.write(request.encode())
2981                await writer.drain()
2982
2983                # Read just the response line
2984                response_line = await asyncio.wait_for(reader.readline(), timeout=5)
2985            finally:
2986                writer.close()
2987                await writer.wait_closed()
2988
2989            # Check if response starts with "ICY"
2990            decoded_line = response_line.decode("latin-1", errors="ignore").strip()
2991            return decoded_line.startswith("ICY")
2992
2993        except TimeoutError:
2994            self.logger.debug("Timeout during Shoutcast validation for %s", url)
2995            return False
2996        except OSError, ConnectionError:
2997            self.logger.debug("Connection failed during Shoutcast validation for %s", url)
2998            return False
2999        except UnicodeDecodeError:
3000            self.logger.debug("Invalid response encoding during Shoutcast validation for %s", url)
3001            return False
3002
3003    def _resolve_player_dsp_config(self, player: Player) -> DSPConfig:
3004        """
3005        Resolve the effective DSP config for a player.
3006
3007        Single source of truth shared by every code path that needs to know
3008        whether DSP will run for this player. Protocol wrappers defer to their
3009        parent player; single-leg ``player_group`` instances that don't expose
3010        ``MULTI_DEVICE_DSP`` defer to their first member; players whose grouping
3011        context prevents DSP get a disabled config back regardless.
3012
3013        :param player: The player to resolve DSP config for.
3014        """
3015        dsp_player_id = self._resolve_player_dsp_config_id(player)
3016        dsp = self.mass.config.get_player_dsp_config(dsp_player_id)
3017        if is_grouping_preventing_dsp(player):
3018            dsp.enabled = False
3019        elif player.provider.domain == "player_group" and (
3020            PlayerFeature.MULTI_DEVICE_DSP not in player.state.supported_features
3021        ):
3022            if not player.state.group_members:
3023                dsp.enabled = False
3024        return dsp
3025
3026    def _resolve_player_dsp_config_id(self, player: Player) -> str:
3027        """
3028        Return the player identifier that supplies the effective DSP config.
3029
3030        :param player: Player whose DSP config source should be resolved.
3031        """
3032        dsp_player_id = player.protocol_parent_id or player.player_id
3033        if (
3034            not is_grouping_preventing_dsp(player)
3035            and player.provider.domain == "player_group"
3036            and PlayerFeature.MULTI_DEVICE_DSP not in player.state.supported_features
3037            and player.state.group_members
3038        ):
3039            child_player = self.mass.players.get_player(player.state.group_members[0])
3040            assert child_player is not None
3041            dsp_player_id = child_player.player_id
3042        return dsp_player_id
3043
3044    def _get_output_channels(self, player: Player | None, player_id: str) -> str:
3045        """
3046        Return the configured output channels for the rendering player.
3047
3048        The value may be stored on the rendering player(protocol) itself (the
3049        protocol section of the config UI) or on its visible parent player (the
3050        native section); the rendering player's own stored value wins.
3051        """
3052        parent_id = player.protocol_parent_id if player and player.protocol_parent_id else player_id
3053        parent_value = self.mass.config.get_raw_player_config_value(
3054            parent_id, CONF_OUTPUT_CHANNELS, "stereo"
3055        )
3056        return self.mass.config.get_raw_player_config_value(
3057            player.player_id if player else player_id, CONF_OUTPUT_CHANNELS, parent_value
3058        )
3059
3060    def _pick_pcm_bit_depth(
3061        self,
3062        players: Iterable[Player],
3063        streamdetails: StreamDetails | None,
3064        crossfade_enabled: bool,
3065        overlay_active: bool = False,
3066    ) -> tuple[ContentType, int]:
3067        """
3068        Return ``(content_type, bit_depth)`` for an internal PCM stream.
3069
3070        F32 is chosen when audio processing (crossfade, audio overlay, volume
3071        normalization, DSP) will run on the stream — those need the extra
3072        headroom to avoid clipping and precision loss. Otherwise the source's
3073        native bit depth is reused so we don't waste memory upcasting a 16-bit
3074        stream to 32-bit just to pass it through. When the source is unknown
3075        (no streamdetails) we fall back to F32 conservatively.
3076        """
3077        if streamdetails is None:
3078            return INTERNAL_PCM_FORMAT.content_type, INTERNAL_PCM_FORMAT.bit_depth
3079        needs_headroom = (
3080            crossfade_enabled
3081            or overlay_active
3082            or streamdetails.volume_normalization_mode != VolumeNormalizationMode.DISABLED
3083            or any(self._resolve_player_dsp_config(player).enabled for player in players)
3084        )
3085        if needs_headroom:
3086            return INTERNAL_PCM_FORMAT.content_type, INTERNAL_PCM_FORMAT.bit_depth
3087        bit_depth = streamdetails.audio_format.bit_depth
3088        return ContentType.from_bit_depth(bit_depth), bit_depth
3089
3090    def _select_audio_source_pcm_format(
3091        self,
3092        player: Player,
3093        streamdetails: StreamDetails,
3094        supported_sample_rates: Iterable[int] | None = None,
3095    ) -> AudioFormat:
3096        """
3097        Return a passthrough PCM format for a realtime AudioSource item.
3098
3099        The format matches the source's native sample rate, bit depth and
3100        channel count whenever the player can accept them; if the player does
3101        not support the source's sample rate, it is snapped down to the
3102        closest supported rate. No F32 widening — realtime sources skip every
3103        processing stage that would otherwise need it. Surround sources are
3104        still folded down to stereo, which every output path requires anyway.
3105
3106        :param player: The player requesting the stream.
3107        :param streamdetails: Stream details for the AudioSource item.
3108        :param supported_sample_rates: Rates shared by every output player, if applicable.
3109        """
3110        resolved_sample_rates = (
3111            list(supported_sample_rates)
3112            if supported_sample_rates is not None
3113            else [sample_rate for sample_rate, _ in player.get_supported_sample_rates()]
3114        )
3115        source_rate = streamdetails.audio_format.sample_rate
3116        if source_rate in resolved_sample_rates:
3117            output_sample_rate = source_rate
3118        else:
3119            output_sample_rate = max(
3120                (rate for rate in resolved_sample_rates if rate <= source_rate),
3121                default=min(resolved_sample_rates),
3122            )
3123        bit_depth = streamdetails.audio_format.bit_depth
3124        return AudioFormat(
3125            content_type=ContentType.from_bit_depth(bit_depth),
3126            sample_rate=output_sample_rate,
3127            bit_depth=bit_depth,
3128            # a realtime source may announce more channels than anything downstream can
3129            # carry (a VBAN stream can be configured up to 8), and player handoff formats
3130            # copy this count straight through, so fold it here
3131            channels=min(streamdetails.audio_format.channels, 2),
3132        )
3133
3134    def _flow_restart_context(
3135        self, queue_id: str, protocol_player: Player | None
3136    ) -> tuple[str, list[int]]:
3137        """
3138        Resolve the flow mode config and supported sample rates for restart decisions.
3139
3140        Prefers the protocol player actually consuming the flow stream over the
3141        queue's (wrapper) player, whose config may lack the audio specific entries.
3142        """
3143        if protocol_player is None:
3144            protocol_player = self.mass.players.get_player(queue_id)
3145        if protocol_player is None:
3146            flow_mode_sample_rate_conf = self.mass.config.get_raw_player_config_value(
3147                queue_id, CONF_FLOW_MODE_SAMPLE_RATE, FLOW_MODE_SAMPLE_RATE_SMART
3148            )
3149            return flow_mode_sample_rate_conf, []
3150        flow_mode_sample_rate_conf = cast(
3151            "str",
3152            protocol_player.config.get_value(
3153                CONF_FLOW_MODE_SAMPLE_RATE, FLOW_MODE_SAMPLE_RATE_SMART
3154            ),
3155        )
3156        supported_sample_rates = sorted(
3157            {sr for sr, _ in protocol_player.get_supported_sample_rates()}
3158        )
3159        return flow_mode_sample_rate_conf, supported_sample_rates
3160
3161    def _flow_stream_needs_restart(
3162        self,
3163        queue_track: QueueItem,
3164        pcm_format: AudioFormat,
3165        supported_sample_rates: list[int],
3166        flow_mode_sample_rate_conf: str,
3167        is_first_track: bool,
3168    ) -> bool:
3169        """
3170        Return True if the upcoming queue track requires exiting the flow stream.
3171
3172        Covers every case where the flow loop should break and hand control back to
3173        the queue controller for restart:
3174
3175        - Live media (radio, audio sources): cannot be played inside a flow,
3176          the controller will fall back to a single-item stream.
3177        - Sample rate mismatch ('smart' / 'bit_perfect' modes only): the next
3178          track's sample rate (snapped up to the closest supported player rate,
3179          mirroring select_flow_pcm_format's anchoring logic) is incompatible with
3180          the current flow rate, so a new flow must be opened.
3181
3182        The first (anchor) track is always allowed to continue for the sample
3183        rate check; select_flow_pcm_format has already snapped the flow rate to it.
3184
3185        :param queue_track: The upcoming queue item.
3186        :param pcm_format: The current flow stream's PCM format.
3187        :param supported_sample_rates: Sorted list of the player's supported rates.
3188        :param flow_mode_sample_rate_conf: The flow mode sample rate config value.
3189        :param is_first_track: Whether this is the first track of the flow stream.
3190        """
3191        # live audio (radio, plugin or audio source) cannot be flowed; let the
3192        # queue controller fall back to single-item streaming for this item
3193        if queue_track.media_type in (MediaType.RADIO, MediaType.AUDIO_SOURCE):
3194            self.logger.info(
3195                "Live media item %s (%s, %s) encountered in flow stream "
3196                "- breaking out to single item stream",
3197                queue_track.queue_item_id,
3198                queue_track.name,
3199                queue_track.media_type,
3200            )
3201            return True
3202
3203        if is_first_track or queue_track.streamdetails is None:
3204            return False
3205        raw_next_rate = queue_track.streamdetails.audio_format.sample_rate
3206        if not raw_next_rate or not supported_sample_rates:
3207            return False
3208        effective_next_rate = _snap_supported_rate_up(raw_next_rate, supported_sample_rates)
3209
3210        # branch order mirrors select_flow_pcm_format: fixed-rate modes resample
3211        # everything to the chosen rate (no restart); bit_perfect restarts on any
3212        # mismatch; anything else falls through to smart-anchor behavior so
3213        # unknown/legacy config values don't silently pin the flow forever.
3214        if flow_mode_sample_rate_conf in (
3215            FLOW_MODE_SAMPLE_RATE_48000,
3216            FLOW_MODE_SAMPLE_RATE_96000,
3217            FLOW_MODE_SAMPLE_RATE_HIGHEST,
3218        ):
3219            needs_restart = False
3220        elif flow_mode_sample_rate_conf == FLOW_MODE_SAMPLE_RATE_BIT_PERFECT:
3221            needs_restart = effective_next_rate != pcm_format.sample_rate
3222        else:
3223            needs_restart = effective_next_rate > pcm_format.sample_rate
3224
3225        if needs_restart:
3226            self.logger.info(
3227                "Track %s (%s) sample rate %s (snapped to %s) incompatible with flow rate %s "
3228                "(mode: %s) - breaking out to restart flow stream",
3229                queue_track.queue_item_id,
3230                queue_track.name,
3231                raw_next_rate,
3232                effective_next_rate,
3233                pcm_format.sample_rate,
3234                flow_mode_sample_rate_conf,
3235            )
3236        return needs_restart
3237
3238    @asynccontextmanager
3239    async def _connect_radio_stream(self, url: str, **kwargs: Any) -> AsyncGenerator[Any]:
3240        """
3241        Connect to a radio stream URL with fallback for legacy SSL/TLS configurations.
3242
3243        Some radio servers use outdated TLS configurations that reject modern
3244        cipher suites. Since radio streams are public broadcast content,
3245        relaxing cipher requirements is acceptable.
3246
3247        :param url: The radio stream URL to connect to.
3248        :param kwargs: Additional keyword arguments passed to aiohttp get().
3249        """
3250        request_url = encoded_request_url(url)
3251        try:
3252            async with self.mass.http_session_no_ssl.get(request_url, **kwargs) as resp:
3253                yield resp
3254        except ClientConnectorSSLError:
3255            self.logger.info(
3256                "SSL handshake failed for %s, retrying with permissive cipher configuration", url
3257            )
3258            insecure_ssl_context = ssl_util.client_context_no_verify(
3259                ssl_util.SSLCipherList.INSECURE
3260            )
3261            async with self.mass.http_session_no_ssl.get(
3262                request_url, ssl=insecure_ssl_context, **kwargs
3263            ) as resp:
3264                yield resp
3265
3266    async def _update_hls_radio_metadata(
3267        self,
3268        streamdetails: StreamDetails,
3269        elapsed_time: int,
3270    ) -> None:
3271        """
3272        Update HLS radio stream metadata by fetching the playlist.
3273
3274        Fetches the HLS playlist and extracts metadata from EXTINF lines.
3275
3276        :param streamdetails: StreamDetails object to update with metadata
3277        :param elapsed_time: Current playback position in seconds (unused for live radio)
3278        """
3279        mass = self.mass
3280        try:
3281            # Get the actual media playlist URL from cache or resolve it
3282            # We cache the media_playlist_url in streamdetails.data to avoid re-resolving
3283            if streamdetails.data is None:
3284                streamdetails.data = {}
3285            media_playlist_url = streamdetails.data.get("hls_media_playlist_url")
3286            if not media_playlist_url:
3287                try:
3288                    assert isinstance(streamdetails.path, str)  # for type checking
3289                    substream = await self.get_hls_substream(streamdetails.path)
3290                    media_playlist_url = substream.path
3291                    streamdetails.data["hls_media_playlist_url"] = media_playlist_url
3292                except Exception as err:
3293                    self.logger.warning(
3294                        "Failed to resolve HLS substream for metadata monitoring: %s", err
3295                    )
3296                    return
3297
3298            # Fetch the media playlist
3299            timeout = ClientTimeout(total=0, connect=10, sock_read=30)
3300            try:
3301                async with mass.http_session_no_ssl.get(
3302                    encoded_request_url(media_playlist_url), timeout=timeout
3303                ) as resp:
3304                    resp.raise_for_status()
3305                    playlist_content = await resp.text()
3306            except ClientResponseError as err:
3307                # Session token likely expired (410/403) — drop cache so next poll re-resolves
3308                if err.status in (403, 410):
3309                    streamdetails.data.pop("hls_media_playlist_url", None)
3310                raise
3311
3312            # Parse the playlist and look for EXTINF metadata
3313            # The most recent segment usually has the current metadata
3314            lines = playlist_content.strip().split("\n")
3315            for line in reversed(lines):
3316                if line.startswith("#EXTINF:"):
3317                    # Extract metadata from EXTINF line
3318                    metadata = parse_extinf_metadata(line)
3319
3320                    # Build stream title from title and artist
3321                    title = metadata.get("title", "")
3322                    artist = metadata.get("artist", "")
3323                    image_url = (
3324                        metadata.get("image") or metadata.get("artwork") or metadata.get("cover")
3325                    )
3326                    if not artist and " - " in title:
3327                        artist, title = title.split(" - ", 1)
3328                    if title or artist:
3329                        # Format as "Artist - Title"
3330                        if artist and title:
3331                            stream_title = f"{artist} - {title}"
3332                        elif title:
3333                            stream_title = title
3334                        else:
3335                            stream_title = artist
3336
3337                        # Clean the stream title
3338                        cleaned_title = clean_stream_title(stream_title)
3339
3340                        # Only update if changed
3341                        if cleaned_title != streamdetails.stream_title and cleaned_title:
3342                            self.logger.log(
3343                                VERBOSE_LOG_LEVEL, "HLS Radio metadata updated: %s", cleaned_title
3344                            )
3345                            streamdetails.stream_title = cleaned_title
3346                            self._update_radio_stream_metadata(
3347                                streamdetails,
3348                                artist=artist or None,
3349                                title=title or cleaned_title,
3350                                image_url=image_url,
3351                            )
3352
3353                    # Only check the most recent EXTINF
3354                    break
3355
3356        except Exception as err:
3357            self.logger.debug("Error fetching HLS metadata: %s", err)
3358
3359    @staticmethod
3360    def _normalize_reconnecting_urls(url: str | list[MultiPartPath]) -> list[str]:
3361        """Normalize a single URL or a sequence into a non-empty list."""
3362        if isinstance(url, str):
3363            return [url]
3364        if not url:
3365            msg = "Radio stream requires at least one URL"
3366            raise InvalidDataError(msg)
3367        return [part.path for part in url]
3368
3369    async def _resolve_overlay_input(self, queue: PlayerQueue) -> str | None:
3370        """
3371        Resolve the queue's overlay source to a file path or URL for ffmpeg.
3372
3373        Returns None (with a warning logged) when the source can not be resolved,
3374        so the caller can degrade to music-only playback.
3375        """
3376        if not (mapping := queue.overlay_source):
3377            return None
3378        try:
3379            provider = self.mass.get_provider(mapping.provider)
3380            if provider is None:
3381                raise MediaNotFoundError(f"Provider {mapping.provider} is not available")
3382            stream_prov = cast("MusicProvider | PluginProvider", provider)
3383            streamdetails = await stream_prov.get_stream_details(
3384                mapping.item_id, MediaType.SOUND_EFFECT
3385            )
3386        except Exception as err:
3387            self.logger.warning(
3388                "Audio overlay source %s is unavailable (%s) - continuing without overlay",
3389                mapping.uri,
3390                str(err) or err.__class__.__name__,
3391            )
3392            return None
3393        if streamdetails.stream_type not in (StreamType.LOCAL_FILE, StreamType.HTTP) or not (
3394            isinstance(streamdetails.path, str)
3395        ):
3396            self.logger.warning(
3397                "Audio overlay source %s uses unsupported stream type %s "
3398                "- continuing without overlay",
3399                mapping.uri,
3400                streamdetails.stream_type,
3401            )
3402            return None
3403        if streamdetails.stream_type == StreamType.LOCAL_FILE and not await aiofiles.os.path.isfile(
3404            streamdetails.path
3405        ):
3406            # guard against stale sources: feeding a missing file to the mixer would
3407            # kill the whole (music) stream instead of just the overlay
3408            self.logger.warning(
3409                "Audio overlay source %s does not exist - continuing without overlay",
3410                streamdetails.path,
3411            )
3412            return None
3413        return streamdetails.path
3414