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