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                if last_fadeout_part:
2421                    queue_track.streamdetails.seek_position = raw_seek_position
2422                continue
2423            if last_fadeout_part:
2424                # edge case: we did not get enough data to make the crossfade
2425                # attribute these bytes to the previous track (they are its tail)
2426                for pcm_slice in iter_pcm_slices(last_fadeout_part, pcm_format, 1000):
2427                    yield pcm_slice
2428                    await asyncio.sleep(0)
2429                # no crossfade happened — undo the eager seek_position
2430                queue_track.streamdetails.seek_position = raw_seek_position
2431                # full tail was pre-counted and is now yielded as-is
2432                last_fadeout_part = b""
2433            if self.crossfade_allowed(
2434                queue_track,
2435                crossfade_mode=crossfade_mode,
2436                player_id=queue.queue_id,
2437                flow_mode=True,
2438            ):
2439                last_fadeout_part = bytes(crossfade_buffer[-crossfade_buffer_size:])
2440                last_streamdetails = queue_track.streamdetails
2441                last_play_log_entry = play_log_entry
2442                remaining_bytes = bytes(crossfade_buffer[:-crossfade_buffer_size])
2443                if remaining_bytes:
2444                    for pcm_slice in iter_pcm_slices(remaining_bytes, pcm_format, 1000):
2445                        yield pcm_slice
2446                        await asyncio.sleep(0)
2447                    bytes_written += len(remaining_bytes)
2448                del remaining_bytes
2449            elif crossfade_mode != CrossfadeMode.DISABLED and crossfade_buffer:
2450                bytes_written += len(crossfade_buffer)
2451                for pcm_slice in iter_pcm_slices(bytes(crossfade_buffer), pcm_format, 1000):
2452                    yield pcm_slice
2453                    await asyncio.sleep(0)
2454            crossfade_buffer = bytearray()
2455
2456            # update duration details based on the actual pcm data we sent
2457            # this also accounts for crossfade and silence stripping
2458            seconds_streamed = bytes_written / pcm_sample_size
2459            queue_track.streamdetails.seconds_streamed = seconds_streamed
2460            # the held-back crossfade tail still counts as this track's media-time
2461            tail_seconds = len(last_fadeout_part) / pcm_sample_size
2462            # streamdetails.duration is in media-time; seconds_streamed is stream-time
2463            # (post-atempo), so we scale by the track's playback_speed to recover media-time.
2464            queue_track.streamdetails.duration = int(
2465                queue_track.streamdetails.seek_position
2466                + (seconds_streamed + tail_seconds) * track_playback_speed
2467            )
2468            # propagate accurate duration to queue_item so UI displays it
2469            queue_track.duration = queue_track.streamdetails.duration
2470            play_log_entry.seconds_streamed = seconds_streamed
2471            play_log_entry.duration = queue_track.streamdetails.duration
2472            if last_play_log_entry is play_log_entry and last_fadeout_part:
2473                # Pre-count the full crossfade tail so the queue index calculation
2474                # doesn't undercount while waiting for the next track's crossfade mix.
2475                # This will be corrected to crossfade_total/2 once the mix completes.
2476                assert play_log_entry.seconds_streamed is not None
2477                play_log_entry.seconds_streamed += len(last_fadeout_part) / pcm_sample_size
2478            self.logger.debug(
2479                "Finished Streaming queue track: %s (%s) on queue %s",
2480                queue_track.streamdetails.uri,
2481                queue_track.name,
2482                queue.display_name,
2483            )
2484        #### HANDLE END OF QUEUE FLOW STREAM
2485        # skip end-of-queue bookkeeping if a newer producer has superseded us;
2486        # the new producer owns queue_buffer_completed and the play log now
2487        if _superseded():
2488            self.logger.debug(
2489                "Flow stream for queue %s superseded - skipping end-of-queue handling",
2490                queue.display_name,
2491            )
2492            return
2493        # end of queue flow: make sure we yield the last_fadeout_part
2494        if last_fadeout_part:
2495            for pcm_slice in iter_pcm_slices(last_fadeout_part, pcm_format, 1000):
2496                yield pcm_slice
2497                await asyncio.sleep(0)
2498            # correct seconds streamed - the duration already includes the tail
2499            last_part_seconds = len(last_fadeout_part) / pcm_sample_size
2500            streamdetails = queue_track.streamdetails
2501            assert streamdetails is not None
2502            streamdetails.seconds_streamed = (
2503                streamdetails.seconds_streamed or 0
2504            ) + last_part_seconds
2505            # also update the play log entry so elapsed time tracking stays in sync
2506            if last_play_log_entry:
2507                assert last_play_log_entry.seconds_streamed is not None
2508                # full tail was pre-counted and is now yielded as-is
2509                last_play_log_entry.duration = streamdetails.duration
2510            last_fadeout_part = b""
2511        self.logger.info("Finished Queue Flow stream for Queue %s", queue.display_name)
2512        # only signal completion if we are still the active producer — a later
2513        # producer would (incorrectly) see this as its own completion otherwise
2514        if not _superseded():
2515            # inform the queue controller that all audio data has been generated
2516            # so it can handle the case where new items were added after the flow stream ended
2517            self.mass.player_queues.queue_buffer_completed(queue.queue_id, queue_exhausted)
2518
2519    async def get_overlay_mixed_stream(
2520        self,
2521        queue: PlayerQueue,
2522        audio_input: AsyncGenerator[bytes],
2523        pcm_format: AudioFormat,
2524    ) -> AsyncGenerator[bytes]:
2525        """
2526        Mix the queue's audio overlay (looping sound effect) into the given PCM stream.
2527
2528        The mixed output has the exact same PCM format, duration and chunking as the
2529        input stream. If the overlay source can not be resolved, the original stream
2530        is passed through unchanged so playback is never interrupted.
2531
2532        :param queue: The PlayerQueue holding the overlay source and volume.
2533        :param audio_input: The audio stream (raw PCM in ``pcm_format``) to mix into.
2534        :param pcm_format: PCM format of both the input and the mixed output.
2535        """
2536        overlay_input = await self._resolve_overlay_input(queue)
2537        if overlay_input is None:
2538            # overlay source unavailable: degrade gracefully to music-only
2539            async for chunk in audio_input:
2540                yield chunk
2541            return
2542        async for chunk in get_ffmpeg_overlay_stream(
2543            audio_input=audio_input,
2544            overlay_input=overlay_input,
2545            pcm_format=pcm_format,
2546            overlay_volume=queue.overlay_volume,
2547            chunk_size=pcm_format.pcm_sample_size,
2548        ):
2549            yield chunk
2550
2551    def crossfade_allowed(
2552        self,
2553        queue_item: QueueItem,
2554        crossfade_mode: CrossfadeMode,
2555        player_id: str,
2556        flow_mode: bool = False,
2557        next_queue_item: QueueItem | None = None,
2558        sample_rate: int | None = None,
2559        next_sample_rate: int | None = None,
2560    ) -> bool:
2561        """Get the crossfade config for a queue item."""
2562        if crossfade_mode == CrossfadeMode.DISABLED:
2563            return False
2564        if not (self.mass.player_queues.get(queue_item.queue_id)):
2565            return False  # just a guard
2566        if not (self.mass.players.get_player(player_id)):
2567            return False  # just a guard
2568        if queue_item.media_type != MediaType.TRACK:
2569            self.logger.debug("Skipping crossfade: current item is not a track")
2570            return False
2571        # check if the next item is part of the same album
2572        next_item = next_queue_item or self.mass.player_queues.get_next_item(
2573            queue_item.queue_id, queue_item.queue_item_id
2574        )
2575        if not next_item:
2576            # there is no next item!
2577            return False
2578        # check if next item is a track
2579        if next_item.media_type != MediaType.TRACK:
2580            self.logger.debug("Skipping crossfade: next item is not a track")
2581            return False
2582        if (
2583            isinstance(queue_item.media_item, Track)
2584            and isinstance(next_item.media_item, Track)
2585            and queue_item.media_item.album
2586            and next_item.media_item.album
2587            and queue_item.media_item.album == next_item.media_item.album
2588            and not self.mass.config.get_raw_core_config_value(
2589                "streams", CONF_ALLOW_CROSSFADE_SAME_ALBUM, False
2590            )
2591        ):
2592            # in general, crossfade is not desired for tracks of the same (gapless) album
2593            # because we have no accurate way to determine if the album is gapless or not,
2594            # for now we just never crossfade between tracks of the same album
2595            self.logger.debug("Skipping crossfade: next item is part of the same album")
2596            return False
2597
2598        # check if we're allowed to crossfade on different sample rates
2599        if (
2600            not flow_mode
2601            and sample_rate
2602            and next_sample_rate
2603            and sample_rate != next_sample_rate
2604            and not self.mass.config.get_raw_player_config_value(
2605                player_id,
2606                CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES.key,
2607                CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES.default_value,
2608            )
2609        ):
2610            self.logger.debug(
2611                "Skipping crossfade: player(protocol) does not support gapless playback "
2612                "with different sample rates (%s vs %s)",
2613                sample_rate,
2614                next_sample_rate,
2615            )
2616            return False
2617
2618        return True
2619
2620    def clear_crossfade_data(self, queue_id: str) -> None:
2621        """
2622        Clear any pending crossfade data for a queue.
2623
2624        :param queue_id: The queue ID to clear crossfade data for.
2625        """
2626        if queue_id in self._crossfade_data:
2627            self.logger.debug("Clearing crossfade data for queue %s", queue_id)
2628            del self._crossfade_data[queue_id]
2629
2630    async def get_shoutcast_stream(
2631        self, url: str, streamdetails: StreamDetails
2632    ) -> AsyncGenerator[bytes]:
2633        """
2634        Yield audio from a legacy Shoutcast server, with ICY metadata parsed inline.
2635
2636        :param url: Shoutcast stream URL.
2637        :param streamdetails: StreamDetails to update with ICY metadata as it arrives.
2638        """
2639        self.logger.debug("Start streaming from legacy Shoutcast server: %s", url)
2640
2641        parsed = urlparse(url)
2642        host = parsed.hostname
2643        port = parsed.port or 80
2644        path = parsed.path or "/"
2645        if parsed.query:
2646            path = f"{path}?{parsed.query}"
2647
2648        try:
2649            # Open raw socket connection
2650            reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=30)
2651        except TimeoutError as err:
2652            raise AudioError(f"Timeout connecting to Shoutcast stream {url}") from err
2653        except (OSError, ConnectionError) as err:
2654            raise AudioError(f"Failed to connect to Shoutcast stream {url}") from err
2655
2656        try:
2657            # Send HTTP request with ICY metadata header
2658            request = (
2659                f"GET {path} HTTP/1.1\r\n"
2660                f"Host: {host}\r\n"
2661                f"User-Agent: {HTTP_HEADERS['User-Agent']}\r\n"
2662                f"Icy-MetaData: 1\r\n\r\n"
2663            )
2664            writer.write(request.encode())
2665            await writer.drain()
2666
2667            # Read and parse response line
2668            try:
2669                response_line = await asyncio.wait_for(reader.readline(), timeout=10)
2670            except TimeoutError as err:
2671                raise AudioError("Timeout reading Shoutcast response") from err
2672
2673            if not response_line.startswith(b"ICY"):
2674                raise InvalidDataError("Invalid Shoutcast response")
2675
2676            # Read headers until empty line
2677            headers: dict[str, str] = {}
2678            while True:
2679                try:
2680                    line = await asyncio.wait_for(reader.readline(), timeout=5)
2681                except TimeoutError as err:
2682                    raise AudioError("Timeout reading Shoutcast headers") from err
2683
2684                if line in (b"\r\n", b"\n", b""):
2685                    break
2686
2687                if b":" in line:
2688                    try:
2689                        key, value = line.decode("latin-1", errors="ignore").split(":", 1)
2690                        headers[key.strip().lower()] = value.strip()
2691                    except UnicodeDecodeError, ValueError:
2692                        continue
2693
2694            # Get metadata interval
2695            meta_int_str = headers.get("icy-metaint")
2696            if not meta_int_str:
2697                raise InvalidDataError("No icy-metaint header in Shoutcast response")
2698
2699            try:
2700                meta_int = int(meta_int_str)
2701            except ValueError as err:
2702                raise InvalidDataError("Invalid icy-metaint value") from err
2703
2704            self.logger.debug("Connected to Shoutcast stream %s (icy-metaint: %s)", url, meta_int)
2705
2706            # Stream audio data with metadata parsing
2707            while True:
2708                try:
2709                    # Read audio chunk
2710                    audio_chunk = await reader.readexactly(meta_int)
2711                    yield audio_chunk
2712
2713                    # Read metadata length
2714                    meta_byte = await reader.readexactly(1)
2715                    if meta_byte == b"\x00":
2716                        continue
2717
2718                    meta_length = ord(meta_byte) * 16
2719                    meta_data = await reader.readexactly(meta_length)
2720                    self._parse_icy_metadata(meta_data, streamdetails)
2721
2722                except asyncio.exceptions.IncompleteReadError:
2723                    # End of stream
2724                    break
2725
2726        finally:
2727            writer.close()
2728            await writer.wait_closed()
2729
2730    # --- Private methods ---
2731
2732    def _notify_provider_streamed(
2733        self, streamdetails: StreamDetails, finished: bool, seconds_streamed: float
2734    ) -> None:
2735        """Report a (mostly) streamed item back to the provider that owns it."""
2736        if not finished and seconds_streamed < 90:
2737            return
2738        provider = self.mass.get_provider(streamdetails.provider)
2739        # plugin providers serve playable items too, but on_streamed is MusicProvider-only
2740        if provider is None or provider.type != ProviderType.MUSIC:
2741            return
2742        music_prov = cast("MusicProvider", provider)
2743        self.mass.create_task(music_prov.on_streamed(streamdetails))
2744
2745    def _get_volume_normalization_preference(
2746        self, streamdetails: StreamDetails
2747    ) -> VolumeNormalizationMode:
2748        """Return the configured normalization preference for the stream's media type."""
2749        conf_key = (
2750            CONF_VOLUME_NORMALIZATION_RADIO
2751            if streamdetails.media_type == MediaType.RADIO
2752            else CONF_VOLUME_NORMALIZATION_TRACKS
2753        )
2754        return VolumeNormalizationMode(
2755            self.mass.streams.get_config_value(conf_key, return_type=str)
2756        )
2757
2758    def _update_radio_stream_metadata(
2759        self,
2760        streamdetails: StreamDetails,
2761        artist: str | None,
2762        title: str,
2763        image_url: str | None = None,
2764        album: str | None = None,
2765    ) -> None:
2766        """
2767        Update radio stream metadata and trigger artwork lookup.
2768
2769        :param streamdetails: The stream details to update.
2770        :param artist: Artist name (will be normalized).
2771        :param title: Track title (will be cleaned for display).
2772        :param image_url: Optional image URL from stream metadata.
2773        :param album: Optional album name.
2774        """
2775        station_image_url = image_url or self.mass.metadata.get_radio_stream_station_image(
2776            streamdetails
2777        )
2778        artist_normalized = (
2779            self.mass.metadata.normalize_radio_artist_name(artist) if artist else None
2780        )
2781        display_title, _ = parse_title_and_version(title, strip_for_display=True)
2782
2783        streamdetails.stream_metadata = StreamMetadata(
2784            title=display_title,
2785            artist=artist_normalized,
2786            album=album,
2787            image_url=station_image_url,
2788        )
2789        streamdetails.stream_metadata_last_updated = time.time()
2790        if streamdetails.queue_id:
2791            self.mass.player_queues.signal_update(streamdetails.queue_id)
2792
2793        # Fetch artwork in background (track, album then artist)
2794        if artist and title and not image_url:
2795            self.mass.call_later(
2796                0.2,
2797                self.mass.metadata.update_radio_stream_artwork,
2798                streamdetails,
2799                task_id=f"update_radio_artwork_{streamdetails.queue_id}",
2800            )
2801
2802    async def _cache_radio_result(
2803        self,
2804        url: str,
2805        stream_type: StreamType,
2806        resolved_url: str | None = None,
2807    ) -> tuple[str, StreamType]:
2808        """Cache and return a radio stream resolution result."""
2809        result = (resolved_url or url, stream_type)
2810        await self.mass.cache.set(
2811            url,
2812            result,
2813            expiration=3600 * 3,
2814            provider=CACHE_PROVIDER,
2815            category=CACHE_CATEGORY_RESOLVED_RADIO_URL,
2816        )
2817        return result
2818
2819    async def _handle_client_error_for_radio_stream(
2820        self, url: str, err: aiohttp.ClientError, fallback_stream_type: StreamType
2821    ) -> tuple[str, StreamType]:
2822        """Handle aiohttp client errors during radio stream resolution."""
2823        # Prefer the final post-redirect URL: aiohttp follows redirects before raising,
2824        # but the original url may just point at a redirector rather than the ICY endpoint.
2825        request_info = getattr(err, "request_info", None)
2826        validate_url = str(request_info.url) if request_info is not None else url
2827
2828        # Check if this is a Shoutcast/ICY response that aiohttp can't parse
2829        if isinstance(err, aiohttp.ClientResponseError) and "ICY" in str(err).upper():
2830            self.logger.debug(
2831                "ICY response detected for %s, validating Shoutcast stream", validate_url
2832            )
2833            if await self._validate_shoutcast_stream(validate_url):
2834                return await self._cache_radio_result(
2835                    url, StreamType.SHOUTCAST, resolved_url=validate_url
2836                )
2837            self.logger.warning(
2838                "ICY response detected but Shoutcast validation failed for %s", validate_url
2839            )
2840            return await self._cache_radio_result(
2841                url, fallback_stream_type, resolved_url=validate_url
2842            )
2843
2844        # Other aiohttp errors - might still be Shoutcast, check it
2845        self.logger.debug("aiohttp error for %s, checking if legacy Shoutcast stream", validate_url)
2846        if await self._validate_shoutcast_stream(validate_url):
2847            return await self._cache_radio_result(
2848                url, StreamType.SHOUTCAST, resolved_url=validate_url
2849            )
2850
2851        # Unknown error - still try to stream
2852        self.logger.warning(
2853            "Failed to parse radio URL %s: %s - attempting direct stream", validate_url, str(err)
2854        )
2855        return await self._cache_radio_result(url, fallback_stream_type, resolved_url=validate_url)
2856
2857    async def _iter_audio_source_pcm(
2858        self,
2859        streamdetails: StreamDetails,
2860        pcm_format: AudioFormat,
2861    ) -> AsyncGenerator[bytes]:
2862        """Yield PCM for an AudioSource, bypassing ffmpeg when formats match."""
2863        if _pcm_formats_match(streamdetails.audio_format, pcm_format):
2864            source_gen = self._open_audio_source_generator(streamdetails)
2865            async for chunk in realtime_pcm_pacer(source_gen, pcm_format):
2866                yield chunk
2867            return
2868        # format mismatch → fall back to ffmpeg for resampling (still small chunks)
2869        async for chunk in self.get_media_stream(
2870            streamdetails=streamdetails,
2871            pcm_format=pcm_format,
2872            filter_params=None,
2873            chunk_seconds=AUDIO_SOURCE_CHUNK_SECONDS,
2874        ):
2875            yield chunk
2876
2877    def _open_audio_source_generator(self, streamdetails: StreamDetails) -> AsyncGenerator[bytes]:
2878        """Open the raw PCM generator for an AudioSource (CUSTOM or NAMED_PIPE)."""
2879        if streamdetails.stream_type == StreamType.CUSTOM:
2880            provider = self.mass.get_provider(streamdetails.provider)
2881            if provider is None:
2882                raise ProviderUnavailableError(
2883                    f"Provider {streamdetails.provider} for stream is no longer available"
2884                )
2885            provider = cast("MusicProvider | PluginProvider", provider)
2886            return provider.get_audio_stream(streamdetails)
2887        if streamdetails.stream_type == StreamType.NAMED_PIPE:
2888            assert isinstance(streamdetails.path, str)  # for type checking
2889            return read_named_pipe(streamdetails.path)
2890        raise AudioError(f"Unsupported stream_type {streamdetails.stream_type} for AudioSource")
2891
2892    def _parse_icy_metadata(self, meta_data: bytes, streamdetails: StreamDetails) -> None:
2893        """
2894        Parse ICY metadata and update streamdetails.
2895
2896        Sets the cleaned stream title and, when the title parses as "Artist - Track",
2897        triggers a radio-artwork metadata update.
2898
2899        :param meta_data: Raw metadata bytes from an ICY stream chunk.
2900        :param streamdetails: StreamDetails to update with parsed title and metadata.
2901        """
2902        if not meta_data:
2903            return
2904
2905        meta_data = meta_data.rstrip(b"\0")
2906        # Match StreamTitle, handling apostrophes in titles
2907        stream_title_re = re.search(rb"StreamTitle='(.*?)';", meta_data)
2908
2909        if not stream_title_re:
2910            self.logger.log(
2911                VERBOSE_LOG_LEVEL,
2912                "ICY metadata does not contain StreamTitle field. Raw: %s",
2913                meta_data.decode("utf-8", errors="replace")[:200],
2914            )
2915            return
2916
2917        try:
2918            # in 99% of the cases the stream title is utf-8 encoded
2919            stream_title = stream_title_re.group(1).decode("utf-8")
2920        except UnicodeDecodeError:
2921            # fallback to iso-8859-1
2922            stream_title = stream_title_re.group(1).decode("iso-8859-1", errors="replace")
2923
2924        cleaned_stream_title = clean_stream_title(stream_title)
2925
2926        if not cleaned_stream_title:
2927            return
2928
2929        if cleaned_stream_title == streamdetails.stream_title:
2930            return
2931
2932        self.logger.log(VERBOSE_LOG_LEVEL, "ICY Radio streamtitle original: %s", stream_title)
2933        self.logger.log(
2934            VERBOSE_LOG_LEVEL, "ICY Radio streamtitle cleaned: %s", cleaned_stream_title
2935        )
2936        streamdetails.stream_title = cleaned_stream_title
2937
2938        # Prefer station-provided cover art from the ICY 'StreamUrl' field (when it is
2939        # an image) over the MusicBrainz artwork lookup in _update_radio_stream_metadata.
2940        image_url = self._parse_icy_image_url(meta_data)
2941
2942        # Parse the original title for structured fields first so stations that announce
2943        # an album can refine the artwork lookup; fall back to the "Artist - Track" split.
2944        album: str | None = None
2945        if parsed := parse_quoted_stream_title(stream_title):
2946            track_name, artist_name_raw, album = parsed
2947        elif " - " in cleaned_stream_title:
2948            artist_name_raw, track_name = (
2949                part.strip() for part in cleaned_stream_title.split(" - ", 1)
2950            )
2951        else:
2952            return
2953
2954        if artist_name_raw and track_name:
2955            self.logger.debug(
2956                "ICY metadata: artist='%s', track='%s', album='%s'",
2957                artist_name_raw,
2958                track_name,
2959                album,
2960            )
2961            self._update_radio_stream_metadata(
2962                streamdetails,
2963                artist=artist_name_raw,
2964                title=track_name,
2965                album=album,
2966                image_url=image_url,
2967            )
2968
2969    def _parse_icy_image_url(self, meta_data: bytes) -> str | None:
2970        """
2971        Return a PNG or JPEG cover-art URL from the ICY 'StreamUrl' field, if present.
2972
2973        :param meta_data: Raw metadata bytes from an ICY stream chunk.
2974        """
2975        # The trailing semicolon is optional to match sources that omit it.
2976        stream_url_re = re.search(rb"StreamUrl='([^']*)'", meta_data)
2977        if not stream_url_re:
2978            return None
2979        try:
2980            image_url = stream_url_re.group(1).decode("utf-8").strip()
2981        except UnicodeDecodeError:
2982            return None
2983        if not image_url:
2984            return None
2985        # StreamUrl is not a standardized artwork field (reference clients such as VLC
2986        # ignore it and it conventionally holds a station website link), so only accept
2987        # values that point at a PNG or JPEG image.
2988        parsed = urlparse(image_url)
2989        if parsed.scheme not in ("http", "https"):
2990            return None
2991        if not parsed.path.lower().endswith((".png", ".jpg", ".jpeg")):
2992            return None
2993        self.logger.debug("ICY metadata: StreamUrl image='%s'", image_url)
2994        return image_url
2995
2996    async def _validate_shoutcast_stream(self, url: str) -> bool:
2997        """
2998        Return True if the URL responds with a legacy Shoutcast "ICY 200 OK" line.
2999
3000        :param url: The URL to validate.
3001        """
3002        try:
3003            parsed = urlparse(url)
3004            host = parsed.hostname
3005            port = parsed.port or 80
3006            path = parsed.path or "/"
3007            if parsed.query:
3008                path = f"{path}?{parsed.query}"
3009
3010            # Open raw socket connection with timeout
3011            reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=10)
3012            try:
3013                # Send minimal HTTP request with ICY metadata header
3014                request = f"GET {path} HTTP/1.1\r\nHost: {host}\r\nIcy-MetaData: 1\r\n\r\n"
3015                writer.write(request.encode())
3016                await writer.drain()
3017
3018                # Read just the response line
3019                response_line = await asyncio.wait_for(reader.readline(), timeout=5)
3020            finally:
3021                writer.close()
3022                await writer.wait_closed()
3023
3024            # Check if response starts with "ICY"
3025            decoded_line = response_line.decode("latin-1", errors="ignore").strip()
3026            return decoded_line.startswith("ICY")
3027
3028        except TimeoutError:
3029            self.logger.debug("Timeout during Shoutcast validation for %s", url)
3030            return False
3031        except OSError, ConnectionError:
3032            self.logger.debug("Connection failed during Shoutcast validation for %s", url)
3033            return False
3034        except UnicodeDecodeError:
3035            self.logger.debug("Invalid response encoding during Shoutcast validation for %s", url)
3036            return False
3037
3038    def _resolve_player_dsp_config(self, player: Player) -> DSPConfig:
3039        """
3040        Resolve the effective DSP config for a player.
3041
3042        Single source of truth shared by every code path that needs to know
3043        whether DSP will run for this player. Protocol wrappers defer to their
3044        parent player; single-leg ``player_group`` instances that don't expose
3045        ``MULTI_DEVICE_DSP`` defer to their first member; players whose grouping
3046        context prevents DSP get a disabled config back regardless.
3047
3048        :param player: The player to resolve DSP config for.
3049        """
3050        dsp_player_id = self._resolve_player_dsp_config_id(player)
3051        dsp = self.mass.config.get_player_dsp_config(dsp_player_id)
3052        if is_grouping_preventing_dsp(player):
3053            dsp.enabled = False
3054        elif player.provider.domain == "player_group" and (
3055            PlayerFeature.MULTI_DEVICE_DSP not in player.state.supported_features
3056        ):
3057            if not player.state.group_members:
3058                dsp.enabled = False
3059        return dsp
3060
3061    def _resolve_player_dsp_config_id(self, player: Player) -> str:
3062        """
3063        Return the player identifier that supplies the effective DSP config.
3064
3065        :param player: Player whose DSP config source should be resolved.
3066        """
3067        dsp_player_id = player.protocol_parent_id or player.player_id
3068        if (
3069            not is_grouping_preventing_dsp(player)
3070            and player.provider.domain == "player_group"
3071            and PlayerFeature.MULTI_DEVICE_DSP not in player.state.supported_features
3072            and player.state.group_members
3073        ):
3074            child_player = self.mass.players.get_player(player.state.group_members[0])
3075            assert child_player is not None
3076            dsp_player_id = child_player.player_id
3077        return dsp_player_id
3078
3079    def _get_output_channels(self, player: Player | None, player_id: str) -> str:
3080        """
3081        Return the configured output channels for the rendering player.
3082
3083        The value may be stored on the rendering player(protocol) itself (the
3084        protocol section of the config UI) or on its visible parent player (the
3085        native section); the rendering player's own stored value wins.
3086        """
3087        parent_id = player.protocol_parent_id if player and player.protocol_parent_id else player_id
3088        parent_value = self.mass.config.get_raw_player_config_value(
3089            parent_id, CONF_OUTPUT_CHANNELS, "stereo"
3090        )
3091        return self.mass.config.get_raw_player_config_value(
3092            player.player_id if player else player_id, CONF_OUTPUT_CHANNELS, parent_value
3093        )
3094
3095    def _pick_pcm_bit_depth(
3096        self,
3097        players: Iterable[Player],
3098        streamdetails: StreamDetails | None,
3099        crossfade_enabled: bool,
3100        overlay_active: bool = False,
3101    ) -> tuple[ContentType, int]:
3102        """
3103        Return ``(content_type, bit_depth)`` for an internal PCM stream.
3104
3105        F32 is chosen when audio processing (crossfade, audio overlay, volume
3106        normalization, DSP) will run on the stream — those need the extra
3107        headroom to avoid clipping and precision loss. Otherwise the source's
3108        native bit depth is reused so we don't waste memory upcasting a 16-bit
3109        stream to 32-bit just to pass it through. When the source is unknown
3110        (no streamdetails) we fall back to F32 conservatively.
3111        """
3112        if streamdetails is None:
3113            return INTERNAL_PCM_FORMAT.content_type, INTERNAL_PCM_FORMAT.bit_depth
3114        needs_headroom = (
3115            crossfade_enabled
3116            or overlay_active
3117            or streamdetails.volume_normalization_mode != VolumeNormalizationMode.DISABLED
3118            or any(self._resolve_player_dsp_config(player).enabled for player in players)
3119        )
3120        if needs_headroom:
3121            return INTERNAL_PCM_FORMAT.content_type, INTERNAL_PCM_FORMAT.bit_depth
3122        bit_depth = streamdetails.audio_format.bit_depth
3123        return ContentType.from_bit_depth(bit_depth), bit_depth
3124
3125    def _select_audio_source_pcm_format(
3126        self,
3127        player: Player,
3128        streamdetails: StreamDetails,
3129        supported_sample_rates: Iterable[int] | None = None,
3130    ) -> AudioFormat:
3131        """
3132        Return a passthrough PCM format for a realtime AudioSource item.
3133
3134        The format matches the source's native sample rate, bit depth and
3135        channel count whenever the player can accept them; if the player does
3136        not support the source's sample rate, it is snapped down to the
3137        closest supported rate. No F32 widening — realtime sources skip every
3138        processing stage that would otherwise need it. Surround sources are
3139        still folded down to stereo, which every output path requires anyway.
3140
3141        :param player: The player requesting the stream.
3142        :param streamdetails: Stream details for the AudioSource item.
3143        :param supported_sample_rates: Rates shared by every output player, if applicable.
3144        """
3145        resolved_sample_rates = (
3146            list(supported_sample_rates)
3147            if supported_sample_rates is not None
3148            else [sample_rate for sample_rate, _ in player.get_supported_sample_rates()]
3149        )
3150        source_rate = streamdetails.audio_format.sample_rate
3151        if source_rate in resolved_sample_rates:
3152            output_sample_rate = source_rate
3153        else:
3154            output_sample_rate = max(
3155                (rate for rate in resolved_sample_rates if rate <= source_rate),
3156                default=min(resolved_sample_rates),
3157            )
3158        bit_depth = streamdetails.audio_format.bit_depth
3159        return AudioFormat(
3160            content_type=ContentType.from_bit_depth(bit_depth),
3161            sample_rate=output_sample_rate,
3162            bit_depth=bit_depth,
3163            # a realtime source may announce more channels than anything downstream can
3164            # carry (a VBAN stream can be configured up to 8), and player handoff formats
3165            # copy this count straight through, so fold it here
3166            channels=min(streamdetails.audio_format.channels, 2),
3167        )
3168
3169    def _flow_restart_context(
3170        self, queue_id: str, protocol_player: Player | None
3171    ) -> tuple[str, list[int]]:
3172        """
3173        Resolve the flow mode config and supported sample rates for restart decisions.
3174
3175        Prefers the protocol player actually consuming the flow stream over the
3176        queue's (wrapper) player, whose config may lack the audio specific entries.
3177        """
3178        if protocol_player is None:
3179            protocol_player = self.mass.players.get_player(queue_id)
3180        if protocol_player is None:
3181            flow_mode_sample_rate_conf = self.mass.config.get_raw_player_config_value(
3182                queue_id, CONF_FLOW_MODE_SAMPLE_RATE, FLOW_MODE_SAMPLE_RATE_SMART
3183            )
3184            return flow_mode_sample_rate_conf, []
3185        flow_mode_sample_rate_conf = cast(
3186            "str",
3187            protocol_player.config.get_value(
3188                CONF_FLOW_MODE_SAMPLE_RATE, FLOW_MODE_SAMPLE_RATE_SMART
3189            ),
3190        )
3191        supported_sample_rates = sorted(
3192            {sr for sr, _ in protocol_player.get_supported_sample_rates()}
3193        )
3194        return flow_mode_sample_rate_conf, supported_sample_rates
3195
3196    def _flow_stream_needs_restart(
3197        self,
3198        queue_track: QueueItem,
3199        pcm_format: AudioFormat,
3200        supported_sample_rates: list[int],
3201        flow_mode_sample_rate_conf: str,
3202        is_first_track: bool,
3203    ) -> bool:
3204        """
3205        Return True if the upcoming queue track requires exiting the flow stream.
3206
3207        Covers every case where the flow loop should break and hand control back to
3208        the queue controller for restart:
3209
3210        - Live media (radio, audio sources): cannot be played inside a flow,
3211          the controller will fall back to a single-item stream.
3212        - Sample rate mismatch ('smart' / 'bit_perfect' modes only): the next
3213          track's sample rate (snapped up to the closest supported player rate,
3214          mirroring select_flow_pcm_format's anchoring logic) is incompatible with
3215          the current flow rate, so a new flow must be opened.
3216
3217        The first (anchor) track is always allowed to continue for the sample
3218        rate check; select_flow_pcm_format has already snapped the flow rate to it.
3219
3220        :param queue_track: The upcoming queue item.
3221        :param pcm_format: The current flow stream's PCM format.
3222        :param supported_sample_rates: Sorted list of the player's supported rates.
3223        :param flow_mode_sample_rate_conf: The flow mode sample rate config value.
3224        :param is_first_track: Whether this is the first track of the flow stream.
3225        """
3226        # live audio (radio, plugin or audio source) cannot be flowed; let the
3227        # queue controller fall back to single-item streaming for this item
3228        if queue_track.media_type in (MediaType.RADIO, MediaType.AUDIO_SOURCE):
3229            self.logger.info(
3230                "Live media item %s (%s, %s) encountered in flow stream "
3231                "- breaking out to single item stream",
3232                queue_track.queue_item_id,
3233                queue_track.name,
3234                queue_track.media_type,
3235            )
3236            return True
3237
3238        if is_first_track or queue_track.streamdetails is None:
3239            return False
3240        raw_next_rate = queue_track.streamdetails.audio_format.sample_rate
3241        if not raw_next_rate or not supported_sample_rates:
3242            return False
3243        effective_next_rate = _snap_supported_rate_up(raw_next_rate, supported_sample_rates)
3244
3245        # branch order mirrors select_flow_pcm_format: fixed-rate modes resample
3246        # everything to the chosen rate (no restart); bit_perfect restarts on any
3247        # mismatch; anything else falls through to smart-anchor behavior so
3248        # unknown/legacy config values don't silently pin the flow forever.
3249        if flow_mode_sample_rate_conf in (
3250            FLOW_MODE_SAMPLE_RATE_48000,
3251            FLOW_MODE_SAMPLE_RATE_96000,
3252            FLOW_MODE_SAMPLE_RATE_HIGHEST,
3253        ):
3254            needs_restart = False
3255        elif flow_mode_sample_rate_conf == FLOW_MODE_SAMPLE_RATE_BIT_PERFECT:
3256            needs_restart = effective_next_rate != pcm_format.sample_rate
3257        else:
3258            needs_restart = effective_next_rate > pcm_format.sample_rate
3259
3260        if needs_restart:
3261            self.logger.info(
3262                "Track %s (%s) sample rate %s (snapped to %s) incompatible with flow rate %s "
3263                "(mode: %s) - breaking out to restart flow stream",
3264                queue_track.queue_item_id,
3265                queue_track.name,
3266                raw_next_rate,
3267                effective_next_rate,
3268                pcm_format.sample_rate,
3269                flow_mode_sample_rate_conf,
3270            )
3271        return needs_restart
3272
3273    @asynccontextmanager
3274    async def _connect_radio_stream(self, url: str, **kwargs: Any) -> AsyncGenerator[Any]:
3275        """
3276        Connect to a radio stream URL with fallback for legacy SSL/TLS configurations.
3277
3278        Some radio servers use outdated TLS configurations that reject modern
3279        cipher suites. Since radio streams are public broadcast content,
3280        relaxing cipher requirements is acceptable.
3281
3282        :param url: The radio stream URL to connect to.
3283        :param kwargs: Additional keyword arguments passed to aiohttp get().
3284        """
3285        request_url = encoded_request_url(url)
3286        try:
3287            async with self.mass.http_session_no_ssl.get(request_url, **kwargs) as resp:
3288                yield resp
3289        except ClientConnectorSSLError:
3290            self.logger.info(
3291                "SSL handshake failed for %s, retrying with permissive cipher configuration", url
3292            )
3293            insecure_ssl_context = ssl_util.client_context_no_verify(
3294                ssl_util.SSLCipherList.INSECURE
3295            )
3296            async with self.mass.http_session_no_ssl.get(
3297                request_url, ssl=insecure_ssl_context, **kwargs
3298            ) as resp:
3299                yield resp
3300
3301    async def _update_hls_radio_metadata(
3302        self,
3303        streamdetails: StreamDetails,
3304        elapsed_time: int,
3305    ) -> None:
3306        """
3307        Update HLS radio stream metadata by fetching the playlist.
3308
3309        Fetches the HLS playlist and extracts metadata from EXTINF lines.
3310
3311        :param streamdetails: StreamDetails object to update with metadata
3312        :param elapsed_time: Current playback position in seconds (unused for live radio)
3313        """
3314        mass = self.mass
3315        try:
3316            # Get the actual media playlist URL from cache or resolve it
3317            # We cache the media_playlist_url in streamdetails.data to avoid re-resolving
3318            if streamdetails.data is None:
3319                streamdetails.data = {}
3320            media_playlist_url = streamdetails.data.get("hls_media_playlist_url")
3321            if not media_playlist_url:
3322                try:
3323                    assert isinstance(streamdetails.path, str)  # for type checking
3324                    substream = await self.get_hls_substream(streamdetails.path)
3325                    media_playlist_url = substream.path
3326                    streamdetails.data["hls_media_playlist_url"] = media_playlist_url
3327                except Exception as err:
3328                    self.logger.warning(
3329                        "Failed to resolve HLS substream for metadata monitoring: %s", err
3330                    )
3331                    return
3332
3333            # Fetch the media playlist
3334            timeout = ClientTimeout(total=0, connect=10, sock_read=30)
3335            try:
3336                async with mass.http_session_no_ssl.get(
3337                    encoded_request_url(media_playlist_url), timeout=timeout
3338                ) as resp:
3339                    resp.raise_for_status()
3340                    playlist_content = await resp.text()
3341            except ClientResponseError as err:
3342                # Session token likely expired (410/403) — drop cache so next poll re-resolves
3343                if err.status in (403, 410):
3344                    streamdetails.data.pop("hls_media_playlist_url", None)
3345                raise
3346
3347            # Parse the playlist and look for EXTINF metadata
3348            # The most recent segment usually has the current metadata
3349            lines = playlist_content.strip().split("\n")
3350            for line in reversed(lines):
3351                if line.startswith("#EXTINF:"):
3352                    # Extract metadata from EXTINF line
3353                    metadata = parse_extinf_metadata(line)
3354
3355                    # Build stream title from title and artist
3356                    title = metadata.get("title", "")
3357                    artist = metadata.get("artist", "")
3358                    image_url = (
3359                        metadata.get("image") or metadata.get("artwork") or metadata.get("cover")
3360                    )
3361                    if not artist and " - " in title:
3362                        artist, title = title.split(" - ", 1)
3363                    if title or artist:
3364                        # Format as "Artist - Title"
3365                        if artist and title:
3366                            stream_title = f"{artist} - {title}"
3367                        elif title:
3368                            stream_title = title
3369                        else:
3370                            stream_title = artist
3371
3372                        # Clean the stream title
3373                        cleaned_title = clean_stream_title(stream_title)
3374
3375                        # Only update if changed
3376                        if cleaned_title != streamdetails.stream_title and cleaned_title:
3377                            self.logger.log(
3378                                VERBOSE_LOG_LEVEL, "HLS Radio metadata updated: %s", cleaned_title
3379                            )
3380                            streamdetails.stream_title = cleaned_title
3381                            self._update_radio_stream_metadata(
3382                                streamdetails,
3383                                artist=artist or None,
3384                                title=title or cleaned_title,
3385                                image_url=image_url,
3386                            )
3387
3388                    # Only check the most recent EXTINF
3389                    break
3390
3391        except Exception as err:
3392            self.logger.debug("Error fetching HLS metadata: %s", err)
3393
3394    @staticmethod
3395    def _normalize_reconnecting_urls(url: str | list[MultiPartPath]) -> list[str]:
3396        """Normalize a single URL or a sequence into a non-empty list."""
3397        if isinstance(url, str):
3398            return [url]
3399        if not url:
3400            msg = "Radio stream requires at least one URL"
3401            raise InvalidDataError(msg)
3402        return [part.path for part in url]
3403
3404    async def _resolve_overlay_input(self, queue: PlayerQueue) -> str | None:
3405        """
3406        Resolve the queue's overlay source to a file path or URL for ffmpeg.
3407
3408        Returns None (with a warning logged) when the source can not be resolved,
3409        so the caller can degrade to music-only playback.
3410        """
3411        if not (mapping := queue.overlay_source):
3412            return None
3413        try:
3414            provider = self.mass.get_provider(mapping.provider)
3415            if provider is None:
3416                raise MediaNotFoundError(f"Provider {mapping.provider} is not available")
3417            stream_prov = cast("MusicProvider | PluginProvider", provider)
3418            streamdetails = await stream_prov.get_stream_details(
3419                mapping.item_id, MediaType.SOUND_EFFECT
3420            )
3421        except Exception as err:
3422            self.logger.warning(
3423                "Audio overlay source %s is unavailable (%s) - continuing without overlay",
3424                mapping.uri,
3425                str(err) or err.__class__.__name__,
3426            )
3427            return None
3428        if streamdetails.stream_type not in (StreamType.LOCAL_FILE, StreamType.HTTP) or not (
3429            isinstance(streamdetails.path, str)
3430        ):
3431            self.logger.warning(
3432                "Audio overlay source %s uses unsupported stream type %s "
3433                "- continuing without overlay",
3434                mapping.uri,
3435                streamdetails.stream_type,
3436            )
3437            return None
3438        if streamdetails.stream_type == StreamType.LOCAL_FILE and not await aiofiles.os.path.isfile(
3439            streamdetails.path
3440        ):
3441            # guard against stale sources: feeding a missing file to the mixer would
3442            # kill the whole (music) stream instead of just the overlay
3443            self.logger.warning(
3444                "Audio overlay source %s does not exist - continuing without overlay",
3445                streamdetails.path,
3446            )
3447            return None
3448        return streamdetails.path
3449