music-assistant-server

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