music-assistant-server

1"""
2Controller to stream audio to players.
3
4The streams controller hosts a basic, unprotected HTTP-only webserver
5purely to stream audio packets to players and some control endpoints such as
6the upnp callbacks and json rpc api for slimproto clients.
7"""
8
9from __future__ import annotations
10
11import asyncio
12import gc
13import logging
14import os
15import urllib.parse
16from collections.abc import AsyncGenerator
17from dataclasses import dataclass
18from typing import TYPE_CHECKING, Final, cast
19
20from aiofiles.os import wrap
21from aiohttp import web
22from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
23from music_assistant_models.enums import (
24    ConfigEntryType,
25    ContentType,
26    MediaType,
27    PlayerFeature,
28    StreamType,
29    VolumeNormalizationMode,
30)
31from music_assistant_models.errors import (
32    AudioError,
33    InvalidDataError,
34    ProviderUnavailableError,
35    QueueEmpty,
36)
37from music_assistant_models.media_items import AudioFormat, Track
38from music_assistant_models.player_queue import PlayLogEntry
39
40from music_assistant.constants import (
41    ANNOUNCE_ALERT_FILE,
42    CONF_BIND_IP,
43    CONF_BIND_PORT,
44    CONF_CROSSFADE_DURATION,
45    CONF_ENTRY_ENABLE_ICY_METADATA,
46    CONF_ENTRY_LOG_LEVEL,
47    CONF_ENTRY_SUPPORT_GAPLESS_DIFFERENT_SAMPLE_RATES,
48    CONF_ENTRY_ZEROCONF_INTERFACES,
49    CONF_HTTP_PROFILE,
50    CONF_OUTPUT_CHANNELS,
51    CONF_OUTPUT_CODEC,
52    CONF_PUBLISH_IP,
53    CONF_SAMPLE_RATES,
54    CONF_SMART_FADES_MODE,
55    CONF_VOLUME_NORMALIZATION_FIXED_GAIN_RADIO,
56    CONF_VOLUME_NORMALIZATION_FIXED_GAIN_TRACKS,
57    CONF_VOLUME_NORMALIZATION_RADIO,
58    CONF_VOLUME_NORMALIZATION_TRACKS,
59    DEFAULT_STREAM_HEADERS,
60    ICY_HEADERS,
61    INTERNAL_PCM_FORMAT,
62    SILENCE_FILE,
63    VERBOSE_LOG_LEVEL,
64)
65from music_assistant.controllers.players.player_controller import AnnounceData
66from music_assistant.controllers.streams.smart_fades import SmartFadesMixer
67from music_assistant.controllers.streams.smart_fades.analyzer import SmartFadesAnalyzer
68from music_assistant.controllers.streams.smart_fades.fades import SMART_CROSSFADE_DURATION
69from music_assistant.helpers.audio import LOGGER as AUDIO_LOGGER
70from music_assistant.helpers.audio import (
71    get_buffered_media_stream,
72    get_chunksize,
73    get_media_stream,
74    get_player_filter_params,
75    get_stream_details,
76    resample_pcm_audio,
77)
78from music_assistant.helpers.buffered_generator import buffered, use_buffer
79from music_assistant.helpers.ffmpeg import LOGGER as FFMPEG_LOGGER
80from music_assistant.helpers.ffmpeg import check_ffmpeg_version, get_ffmpeg_stream
81from music_assistant.helpers.util import (
82    divide_chunks,
83    get_ip_addresses,
84    get_total_system_memory,
85    select_free_port,
86)
87from music_assistant.helpers.webserver import Webserver
88from music_assistant.models.core_controller import CoreController
89from music_assistant.models.music_provider import MusicProvider
90from music_assistant.models.plugin import PluginProvider, PluginSource
91from music_assistant.models.smart_fades import SmartFadesMode
92from music_assistant.providers.universal_group.constants import UGP_PREFIX
93from music_assistant.providers.universal_group.player import UniversalGroupPlayer
94
95if TYPE_CHECKING:
96    from music_assistant_models.config_entries import CoreConfig
97    from music_assistant_models.player import PlayerMedia
98    from music_assistant_models.player_queue import PlayerQueue
99    from music_assistant_models.queue_item import QueueItem
100    from music_assistant_models.streamdetails import StreamDetails
101
102    from music_assistant.mass import MusicAssistant
103    from music_assistant.models.player import Player
104
105
106isfile = wrap(os.path.isfile)
107
108CONF_ALLOW_BUFFER: Final[str] = "allow_buffering"
109CONF_ALLOW_CROSSFADE_SAME_ALBUM: Final[str] = "allow_crossfade_same_album"
110CONF_SMART_FADES_LOG_LEVEL: Final[str] = "smart_fades_log_level"
111
112# Calculate total system memory once at module load time
113TOTAL_SYSTEM_MEMORY_GB: Final[float] = get_total_system_memory()
114CONF_ALLOW_BUFFER_DEFAULT = TOTAL_SYSTEM_MEMORY_GB >= 8.0
115
116
117def parse_pcm_info(content_type: str) -> tuple[int, int, int]:
118    """Parse PCM info from a codec/content_type string."""
119    params = (
120        dict(urllib.parse.parse_qsl(content_type.replace(";", "&"))) if ";" in content_type else {}
121    )
122    sample_rate = int(params.get("rate", 44100))
123    sample_size = int(params.get("bitrate", 16))
124    channels = int(params.get("channels", 2))
125    return (sample_rate, sample_size, channels)
126
127
128@dataclass
129class CrossfadeData:
130    """Data class to hold crossfade data."""
131
132    data: bytes
133    fade_in_size: int
134    pcm_format: AudioFormat  # Format of the 'data' bytes (current/previous track's format)
135    fade_in_pcm_format: AudioFormat  # Format for 'fade_in_size' (next track's format)
136    queue_item_id: str
137
138
139class StreamsController(CoreController):
140    """Webserver Controller to stream audio to players."""
141
142    domain: str = "streams"
143
144    def __init__(self, mass: MusicAssistant) -> None:
145        """Initialize instance."""
146        super().__init__(mass)
147        self._server = Webserver(self.logger, enable_dynamic_routes=True)
148        self.register_dynamic_route = self._server.register_dynamic_route
149        self.unregister_dynamic_route = self._server.unregister_dynamic_route
150        self.manifest.name = "Streamserver"
151        self.manifest.description = (
152            "Music Assistant's core controller that is responsible for "
153            "streaming audio to players on the local network."
154        )
155        self.manifest.icon = "cast-audio"
156        self.announcements: dict[str, AnnounceData] = {}
157        self._crossfade_data: dict[str, CrossfadeData] = {}
158        self._bind_ip: str = "0.0.0.0"
159        self._smart_fades_mixer = SmartFadesMixer(self)
160        self._smart_fades_analyzer = SmartFadesAnalyzer(self)
161
162    @property
163    def base_url(self) -> str:
164        """Return the base_url for the streamserver."""
165        return self._server.base_url
166
167    @property
168    def bind_ip(self) -> str:
169        """Return the IP address this streamserver is bound to."""
170        return self._bind_ip
171
172    @property
173    def smart_fades_mixer(self) -> SmartFadesMixer:
174        """Return the SmartFadesMixer instance."""
175        return self._smart_fades_mixer
176
177    @property
178    def smart_fades_analyzer(self) -> SmartFadesAnalyzer:
179        """Return the SmartFadesAnalyzer instance."""
180        return self._smart_fades_analyzer
181
182    async def get_config_entries(
183        self,
184        action: str | None = None,
185        values: dict[str, ConfigValueType] | None = None,
186    ) -> tuple[ConfigEntry, ...]:
187        """Return all Config Entries for this core module (if any)."""
188        ip_addresses = await get_ip_addresses()
189        default_port = await select_free_port(8097, 9200)
190        return (
191            ConfigEntry(
192                key=CONF_ALLOW_BUFFER,
193                type=ConfigEntryType.BOOLEAN,
194                default_value=CONF_ALLOW_BUFFER_DEFAULT,
195                label="Allow (in-memory) buffering of (track) audio",
196                description="By default, Music Assistant tries to be as resource "
197                "efficient as possible when streaming audio, especially considering "
198                "low-end devices such as Raspberry Pi's. This means that audio "
199                "buffering is disabled by default to reduce memory usage. \n\n"
200                "Enabling this option allows for in-memory buffering of audio, "
201                "which (massively) improves playback (and seeking) performance but it comes "
202                "at the cost of increased memory usage. "
203                "If you run Music Assistant on a capable device with enough memory, "
204                "enabling this option is strongly recommended.",
205                required=False,
206                category="audio",
207            ),
208            ConfigEntry(
209                key=CONF_VOLUME_NORMALIZATION_RADIO,
210                type=ConfigEntryType.STRING,
211                default_value=VolumeNormalizationMode.FALLBACK_FIXED_GAIN,
212                label="Volume normalization method for radio streams",
213                options=[
214                    ConfigValueOption(x.value.replace("_", " ").title(), x.value)
215                    for x in VolumeNormalizationMode
216                ],
217                category="audio",
218            ),
219            ConfigEntry(
220                key=CONF_VOLUME_NORMALIZATION_TRACKS,
221                type=ConfigEntryType.STRING,
222                default_value=VolumeNormalizationMode.FALLBACK_DYNAMIC,
223                label="Volume normalization method for tracks",
224                options=[
225                    ConfigValueOption(x.value.replace("_", " ").title(), x.value)
226                    for x in VolumeNormalizationMode
227                ],
228                category="audio",
229            ),
230            ConfigEntry(
231                key=CONF_VOLUME_NORMALIZATION_FIXED_GAIN_RADIO,
232                type=ConfigEntryType.FLOAT,
233                range=(-20, 10),
234                default_value=-6,
235                label="Fixed/fallback gain adjustment for radio streams",
236                category="audio",
237            ),
238            ConfigEntry(
239                key=CONF_VOLUME_NORMALIZATION_FIXED_GAIN_TRACKS,
240                type=ConfigEntryType.FLOAT,
241                range=(-20, 10),
242                default_value=-6,
243                label="Fixed/fallback gain adjustment for tracks",
244                category="audio",
245            ),
246            ConfigEntry(
247                key=CONF_ALLOW_CROSSFADE_SAME_ALBUM,
248                type=ConfigEntryType.BOOLEAN,
249                default_value=False,
250                label="Allow crossfade between tracks from the same album",
251                description="Enabling this option allows for crossfading between tracks "
252                "that are part of the same album.",
253                category="audio",
254            ),
255            ConfigEntry(
256                key=CONF_PUBLISH_IP,
257                type=ConfigEntryType.STRING,
258                default_value=ip_addresses[0],
259                label="Published IP address",
260                description="This IP address is communicated to players where to find this server."
261                "\nMake sure that this IP can be reached by players on the local network, "
262                "otherwise audio streaming will not work.",
263                required=False,
264                category="advanced",
265                requires_reload=True,
266            ),
267            ConfigEntry(
268                key=CONF_BIND_PORT,
269                type=ConfigEntryType.INTEGER,
270                default_value=default_port,
271                label="TCP Port",
272                description="The TCP port to run the server. "
273                "Make sure that this server can be reached "
274                "on the given IP and TCP port by players on the local network.",
275                category="advanced",
276                requires_reload=True,
277            ),
278            ConfigEntry(
279                key=CONF_BIND_IP,
280                type=ConfigEntryType.STRING,
281                default_value="0.0.0.0",
282                options=[ConfigValueOption(x, x) for x in {"0.0.0.0", *ip_addresses}],
283                label="Bind to IP/interface",
284                description="Start the stream server on this specific interface. \n"
285                "Use 0.0.0.0 to bind to all interfaces, which is the default. \n"
286                "This is an advanced setting that should normally "
287                "not be adjusted in regular setups.",
288                category="advanced",
289                required=False,
290                requires_reload=True,
291            ),
292            ConfigEntry(
293                key=CONF_SMART_FADES_LOG_LEVEL,
294                type=ConfigEntryType.STRING,
295                label="Smart Fades Log level",
296                description="Log level for the Smart Fades mixer and analyzer.",
297                options=CONF_ENTRY_LOG_LEVEL.options,
298                default_value="GLOBAL",
299                category="advanced",
300            ),
301            CONF_ENTRY_ZEROCONF_INTERFACES,
302        )
303
304    async def setup(self, config: CoreConfig) -> None:
305        """Async initialize of module."""
306        # copy log level to audio/ffmpeg loggers
307        AUDIO_LOGGER.setLevel(self.logger.level)
308        FFMPEG_LOGGER.setLevel(self.logger.level)
309        self._setup_smart_fades_logger(config)
310        # perform check for ffmpeg version
311        await check_ffmpeg_version()
312        # start the webserver
313        self.publish_port = config.get_value(CONF_BIND_PORT)
314        self.publish_ip = config.get_value(CONF_PUBLISH_IP)
315        self._bind_ip = bind_ip = str(config.get_value(CONF_BIND_IP))
316        # print a big fat message in the log where the streamserver is running
317        # because this is a common source of issues for people with more complex setups
318        self.logger.log(
319            logging.INFO if self.mass.config.onboard_done else logging.WARNING,
320            "\n\n################################################################################\n"
321            "Starting streamserver on  %s:%s\n"
322            "This is the IP address that is communicated to players.\n"
323            "If this is incorrect, audio will not play!\n"
324            "See the documentation how to configure the publish IP for the Streamserver\n"
325            "in Settings --> Core modules --> Streamserver\n"
326            "################################################################################\n",
327            self.publish_ip,
328            self.publish_port,
329        )
330        await self._server.setup(
331            bind_ip=bind_ip,
332            bind_port=cast("int", self.publish_port),
333            base_url=f"http://{self.publish_ip}:{self.publish_port}",
334            static_routes=[
335                (
336                    "*",
337                    "/flow/{session_id}/{queue_id}/{queue_item_id}.{fmt}",
338                    self.serve_queue_flow_stream,
339                ),
340                (
341                    "*",
342                    "/single/{session_id}/{queue_id}/{queue_item_id}.{fmt}",
343                    self.serve_queue_item_stream,
344                ),
345                (
346                    "*",
347                    "/command/{queue_id}/{command}.mp3",
348                    self.serve_command_request,
349                ),
350                (
351                    "*",
352                    "/announcement/{player_id}.{fmt}",
353                    self.serve_announcement_stream,
354                ),
355                (
356                    "*",
357                    "/pluginsource/{plugin_source}/{player_id}.{fmt}",
358                    self.serve_plugin_source_stream,
359                ),
360            ],
361        )
362        # Start periodic garbage collection task
363        # This ensures memory from audio buffers and streams is cleaned up regularly
364        self.mass.call_later(900, self._periodic_garbage_collection)  # 15 minutes
365
366    async def close(self) -> None:
367        """Cleanup on exit."""
368        await self._server.close()
369
370    async def resolve_stream_url(
371        self,
372        session_id: str,
373        queue_item: QueueItem,
374        flow_mode: bool = False,
375        player_id: str | None = None,
376    ) -> str:
377        """Resolve the stream URL for the given QueueItem."""
378        if not player_id:
379            player_id = queue_item.queue_id
380        conf_output_codec = await self.mass.config.get_player_config_value(
381            player_id, CONF_OUTPUT_CODEC, default="flac", return_type=str
382        )
383        output_codec = ContentType.try_parse(conf_output_codec or "flac")
384        fmt = output_codec.value
385        # handle raw pcm without exact format specifiers
386        if output_codec.is_pcm() and ";" not in fmt:
387            fmt += f";codec=pcm;rate={44100};bitrate={16};channels={2}"
388        base_path = "flow" if flow_mode else "single"
389        return f"{self._server.base_url}/{base_path}/{session_id}/{queue_item.queue_id}/{queue_item.queue_item_id}.{fmt}"  # noqa: E501
390
391    async def get_plugin_source_url(
392        self,
393        plugin_source: PluginSource,
394        player_id: str,
395    ) -> str:
396        """Get the url for the Plugin Source stream/proxy."""
397        if plugin_source.audio_format.content_type.is_pcm():
398            fmt = ContentType.WAV.value
399        else:
400            fmt = plugin_source.audio_format.content_type.value
401        return f"{self._server.base_url}/pluginsource/{plugin_source.id}/{player_id}.{fmt}"
402
403    async def serve_queue_item_stream(self, request: web.Request) -> web.StreamResponse:
404        """Stream single queueitem audio to a player."""
405        self._log_request(request)
406        queue_id = request.match_info["queue_id"]
407        queue = self.mass.player_queues.get(queue_id)
408        if not queue:
409            raise web.HTTPNotFound(reason=f"Unknown Queue: {queue_id}")
410        session_id = request.match_info["session_id"]
411        if queue.session_id and session_id != queue.session_id:
412            raise web.HTTPNotFound(reason=f"Unknown (or invalid) session: {session_id}")
413        queue_player = self.mass.players.get(queue_id)
414        queue_item_id = request.match_info["queue_item_id"]
415        queue_item = self.mass.player_queues.get_item(queue_id, queue_item_id)
416        if not queue_item:
417            raise web.HTTPNotFound(reason=f"Unknown Queue item: {queue_item_id}")
418        if not queue_item.streamdetails:
419            try:
420                queue_item.streamdetails = await get_stream_details(
421                    mass=self.mass, queue_item=queue_item
422                )
423            except Exception as e:
424                self.logger.error(
425                    "Failed to get streamdetails for QueueItem %s: %s", queue_item_id, e
426                )
427                queue_item.available = False
428                raise web.HTTPNotFound(reason=f"No streamdetails for Queue item: {queue_item_id}")
429
430        # pick output format based on the streamdetails and player capabilities
431        if not queue_player:
432            raise web.HTTPNotFound(reason=f"Unknown Player: {queue_id}")
433
434        # work out pcm format based on streamdetails
435        pcm_format = await self._select_pcm_format(
436            player=queue_player,
437            streamdetails=queue_item.streamdetails,
438            smartfades_enabled=True,
439        )
440        output_format = await self.get_output_format(
441            output_format_str=request.match_info["fmt"],
442            player=queue_player,
443            content_sample_rate=pcm_format.sample_rate,
444            content_bit_depth=pcm_format.bit_depth,
445        )
446
447        # prepare request, add some DLNA/UPNP compatible headers
448        headers = {
449            **DEFAULT_STREAM_HEADERS,
450            "icy-name": queue_item.name,
451            "contentFeatures.dlna.org": "DLNA.ORG_OP=01;DLNA.ORG_FLAGS=01500000000000000000000000000000",  # noqa: E501
452            "Accept-Ranges": "none",
453            "Content-Type": f"audio/{output_format.output_format_str}",
454        }
455        resp = web.StreamResponse(
456            status=200,
457            reason="OK",
458            headers=headers,
459        )
460        resp.content_type = f"audio/{output_format.output_format_str}"
461        http_profile = await self.mass.config.get_player_config_value(
462            queue_id, CONF_HTTP_PROFILE, default="default", return_type=str
463        )
464        if http_profile == "forced_content_length" and not queue_item.duration:
465            # just set an insane high content length to make sure the player keeps playing
466            resp.content_length = get_chunksize(output_format, 12 * 3600)
467        elif http_profile == "forced_content_length" and queue_item.duration:
468            # guess content length based on duration
469            resp.content_length = get_chunksize(output_format, queue_item.duration)
470        elif http_profile == "chunked":
471            resp.enable_chunked_encoding()
472
473        await resp.prepare(request)
474
475        # return early if this is not a GET request
476        if request.method != "GET":
477            return resp
478
479        if queue_item.media_type != MediaType.TRACK:
480            # no crossfade on non-tracks
481            smart_fades_mode = SmartFadesMode.DISABLED
482        else:
483            smart_fades_mode = await self.mass.config.get_player_config_value(
484                queue.queue_id, CONF_SMART_FADES_MODE, return_type=SmartFadesMode
485            )
486            standard_crossfade_duration = self.mass.config.get_raw_player_config_value(
487                queue.queue_id, CONF_CROSSFADE_DURATION, 10
488            )
489        if (
490            smart_fades_mode != SmartFadesMode.DISABLED
491            and PlayerFeature.GAPLESS_PLAYBACK not in queue_player.supported_features
492        ):
493            # crossfade is not supported on this player due to missing gapless playback
494            self.logger.warning(
495                "Crossfade disabled: Player %s does not support gapless playback, "
496                "consider enabling flow mode to enable crossfade on this player.",
497                queue_player.display_name if queue_player else "Unknown Player",
498            )
499            smart_fades_mode = SmartFadesMode.DISABLED
500
501        if smart_fades_mode != SmartFadesMode.DISABLED:
502            # crossfade is enabled, use special crossfaded single item stream
503            # where the crossfade of the next track is present in the stream of
504            # a single track. This only works if the player supports gapless playback!
505            audio_input = self.get_queue_item_stream_with_smartfade(
506                queue_item=queue_item,
507                pcm_format=pcm_format,
508                smart_fades_mode=smart_fades_mode,
509                standard_crossfade_duration=standard_crossfade_duration,
510            )
511        else:
512            # no crossfade, just a regular single item stream
513            audio_input = self.get_queue_item_stream(
514                queue_item=queue_item,
515                pcm_format=pcm_format,
516                seek_position=queue_item.streamdetails.seek_position,
517            )
518        # stream the audio
519        # this final ffmpeg process in the chain will convert the raw, lossless PCM audio into
520        # the desired output format for the player including any player specific filter params
521        # such as channels mixing, DSP, resampling and, only if needed, encoding to lossy formats
522        if queue_item.media_type == MediaType.RADIO:
523            # keep very short buffer for radio streams
524            # to keep them (more or less) realtime and prevent time outs
525            read_rate_input_args = ["-readrate", "1.0", "-readrate_initial_burst", "2"]
526        else:
527            # just allow the player to buffer whatever it wants for single item streams
528            read_rate_input_args = None
529
530        first_chunk_received = False
531        bytes_sent = 0
532        async for chunk in get_ffmpeg_stream(
533            audio_input=audio_input,
534            input_format=pcm_format,
535            output_format=output_format,
536            filter_params=get_player_filter_params(
537                self.mass,
538                player_id=queue_player.player_id,
539                input_format=pcm_format,
540                output_format=output_format,
541            ),
542            extra_input_args=read_rate_input_args,
543        ):
544            try:
545                await resp.write(chunk)
546                bytes_sent += len(chunk)
547                if not first_chunk_received:
548                    first_chunk_received = True
549                    # inform the queue that the track is now loaded in the buffer
550                    # so for example the next track can be enqueued
551                    self.mass.player_queues.track_loaded_in_buffer(
552                        queue_item.queue_id, queue_item.queue_item_id
553                    )
554            except (BrokenPipeError, ConnectionResetError, ConnectionError) as err:
555                if first_chunk_received and not queue_player.stop_called:
556                    # Player disconnected (unexpected) after receiving at least some data
557                    # This could indicate buffering issues, network problems,
558                    # or player-specific issues
559                    bytes_expected = get_chunksize(output_format, queue_item.duration or 3600)
560                    self.logger.warning(
561                        "Player %s disconnected prematurely from stream for %s (%s) - "
562                        "error: %s, sent %d bytes, expected (approx) bytes=%d",
563                        queue.display_name,
564                        queue_item.name,
565                        queue_item.uri,
566                        err.__class__.__name__,
567                        bytes_sent,
568                        bytes_expected,
569                    )
570                break
571        if queue_item.streamdetails.stream_error:
572            self.logger.error(
573                "Error streaming QueueItem %s (%s) to %s - will try to skip to next item",
574                queue_item.name,
575                queue_item.uri,
576                queue.display_name,
577            )
578            # try to skip to the next item in the queue after a short delay
579            self.mass.call_later(5, self.mass.player_queues.next(queue_id))
580        return resp
581
582    async def serve_queue_flow_stream(self, request: web.Request) -> web.StreamResponse:
583        """Stream Queue Flow audio to player."""
584        self._log_request(request)
585        queue_id = request.match_info["queue_id"]
586        queue = self.mass.player_queues.get(queue_id)
587        if not queue:
588            raise web.HTTPNotFound(reason=f"Unknown Queue: {queue_id}")
589        if not (queue_player := self.mass.players.get(queue_id)):
590            raise web.HTTPNotFound(reason=f"Unknown Player: {queue_id}")
591        start_queue_item_id = request.match_info["queue_item_id"]
592        start_queue_item = self.mass.player_queues.get_item(queue_id, start_queue_item_id)
593        if not start_queue_item:
594            raise web.HTTPNotFound(reason=f"Unknown Queue item: {start_queue_item_id}")
595
596        queue.flow_mode_stream_log = []
597
598        # select the highest possible PCM settings for this player
599        flow_pcm_format = await self._select_flow_format(queue_player)
600
601        # work out output format/details
602        output_format = await self.get_output_format(
603            output_format_str=request.match_info["fmt"],
604            player=queue_player,
605            content_sample_rate=flow_pcm_format.sample_rate,
606            content_bit_depth=flow_pcm_format.bit_depth,
607        )
608        # work out ICY metadata support
609        icy_preference = self.mass.config.get_raw_player_config_value(
610            queue_id,
611            CONF_ENTRY_ENABLE_ICY_METADATA.key,
612            CONF_ENTRY_ENABLE_ICY_METADATA.default_value,
613        )
614        enable_icy = request.headers.get("Icy-MetaData", "") == "1" and icy_preference != "disabled"
615        icy_meta_interval = 256000 if icy_preference == "full" else 16384
616
617        # prepare request, add some DLNA/UPNP compatible headers
618        headers = {
619            **DEFAULT_STREAM_HEADERS,
620            **ICY_HEADERS,
621            "contentFeatures.dlna.org": "DLNA.ORG_OP=01;DLNA.ORG_FLAGS=01700000000000000000000000000000",  # noqa: E501
622            "Accept-Ranges": "none",
623            "Content-Type": f"audio/{output_format.output_format_str}",
624        }
625        if enable_icy:
626            headers["icy-metaint"] = str(icy_meta_interval)
627
628        resp = web.StreamResponse(
629            status=200,
630            reason="OK",
631            headers=headers,
632        )
633        http_profile = await self.mass.config.get_player_config_value(
634            queue_id, CONF_HTTP_PROFILE, default="default", return_type=str
635        )
636        if http_profile == "forced_content_length":
637            # just set an insane high content length to make sure the player keeps playing
638            resp.content_length = get_chunksize(output_format, 12 * 3600)
639        elif http_profile == "chunked":
640            resp.enable_chunked_encoding()
641
642        await resp.prepare(request)
643
644        # return early if this is not a GET request
645        if request.method != "GET":
646            return resp
647
648        # all checks passed, start streaming!
649        # this final ffmpeg process in the chain will convert the raw, lossless PCM audio into
650        # the desired output format for the player including any player specific filter params
651        # such as channels mixing, DSP, resampling and, only if needed, encoding to lossy formats
652        self.logger.debug("Start serving Queue flow audio stream for %s", queue.display_name)
653
654        async for chunk in get_ffmpeg_stream(
655            audio_input=self.get_queue_flow_stream(
656                queue=queue,
657                start_queue_item=start_queue_item,
658                pcm_format=flow_pcm_format,
659            ),
660            input_format=flow_pcm_format,
661            output_format=output_format,
662            filter_params=get_player_filter_params(
663                self.mass, queue_player.player_id, flow_pcm_format, output_format
664            ),
665            # we need to slowly feed the music to avoid the player stopping and later
666            # restarting (or completely failing) the audio stream by keeping the buffer short.
667            # this is reported to be an issue especially with Chromecast players.
668            # see for example: https://github.com/music-assistant/support/issues/3717
669            # allow buffer ahead of 6 seconds and read rest in realtime
670            extra_input_args=["-readrate", "1.0", "-readrate_initial_burst", "6"],
671            chunk_size=icy_meta_interval if enable_icy else get_chunksize(output_format),
672        ):
673            try:
674                await resp.write(chunk)
675            except (BrokenPipeError, ConnectionResetError, ConnectionError):
676                # race condition
677                break
678
679            if not enable_icy:
680                continue
681
682            # if icy metadata is enabled, send the icy metadata after the chunk
683            if (
684                # use current item here and not buffered item, otherwise
685                # the icy metadata will be too much ahead
686                (current_item := queue.current_item)
687                and current_item.streamdetails
688                and current_item.streamdetails.stream_title
689            ):
690                title = current_item.streamdetails.stream_title
691            elif queue and current_item and current_item.name:
692                title = current_item.name
693            else:
694                title = "Music Assistant"
695            metadata = f"StreamTitle='{title}';".encode()
696            if icy_preference == "full" and current_item and current_item.image:
697                metadata += f"StreamURL='{current_item.image.path}'".encode()
698            while len(metadata) % 16 != 0:
699                metadata += b"\x00"
700            length = len(metadata)
701            length_b = chr(int(length / 16)).encode()
702            await resp.write(length_b + metadata)
703
704        return resp
705
706    async def serve_command_request(self, request: web.Request) -> web.FileResponse:
707        """Handle special 'command' request for a player."""
708        self._log_request(request)
709        queue_id = request.match_info["queue_id"]
710        command = request.match_info["command"]
711        if command == "next":
712            self.mass.create_task(self.mass.player_queues.next(queue_id))
713        return web.FileResponse(SILENCE_FILE, headers={"icy-name": "Music Assistant"})
714
715    async def serve_announcement_stream(self, request: web.Request) -> web.StreamResponse:
716        """Stream announcement audio to a player."""
717        self._log_request(request)
718        player_id = request.match_info["player_id"]
719        player = self.mass.player_queues.get(player_id)
720        if not player:
721            raise web.HTTPNotFound(reason=f"Unknown Player: {player_id}")
722        if not (announce_data := self.announcements.get(player_id)):
723            raise web.HTTPNotFound(reason=f"No pending announcements for Player: {player_id}")
724
725        # work out output format/details
726        fmt = request.match_info["fmt"]
727        audio_format = AudioFormat(content_type=ContentType.try_parse(fmt))
728
729        http_profile = await self.mass.config.get_player_config_value(
730            player_id, CONF_HTTP_PROFILE, default="default", return_type=str
731        )
732        if http_profile == "forced_content_length":
733            # given the fact that an announcement is just a short audio clip,
734            # just send it over completely at once so we have a fixed content length
735            data = b""
736            async for chunk in self.get_announcement_stream(
737                announcement_url=announce_data["announcement_url"],
738                output_format=audio_format,
739                pre_announce=announce_data["pre_announce"],
740                pre_announce_url=announce_data["pre_announce_url"],
741            ):
742                data += chunk
743            return web.Response(
744                body=data,
745                content_type=f"audio/{audio_format.output_format_str}",
746                headers=DEFAULT_STREAM_HEADERS,
747            )
748
749        resp = web.StreamResponse(
750            status=200,
751            reason="OK",
752            headers=DEFAULT_STREAM_HEADERS,
753        )
754        resp.content_type = f"audio/{audio_format.output_format_str}"
755        if http_profile == "chunked":
756            resp.enable_chunked_encoding()
757
758        await resp.prepare(request)
759
760        # return early if this is not a GET request
761        if request.method != "GET":
762            return resp
763
764        # all checks passed, start streaming!
765        self.logger.debug(
766            "Start serving audio stream for Announcement %s to %s",
767            announce_data["announcement_url"],
768            player.display_name,
769        )
770        async for chunk in self.get_announcement_stream(
771            announcement_url=announce_data["announcement_url"],
772            output_format=audio_format,
773            pre_announce=announce_data["pre_announce"],
774            pre_announce_url=announce_data["pre_announce_url"],
775        ):
776            try:
777                await resp.write(chunk)
778            except (BrokenPipeError, ConnectionResetError):
779                break
780
781        self.logger.debug(
782            "Finished serving audio stream for Announcement %s to %s",
783            announce_data["announcement_url"],
784            player.display_name,
785        )
786
787        return resp
788
789    async def serve_plugin_source_stream(self, request: web.Request) -> web.StreamResponse:
790        """Stream PluginSource audio to a player."""
791        self._log_request(request)
792        plugin_source_id = request.match_info["plugin_source"]
793        provider = cast("PluginProvider", self.mass.get_provider(plugin_source_id))
794        if not provider:
795            raise ProviderUnavailableError(f"Unknown PluginSource: {plugin_source_id}")
796        # work out output format/details
797        player_id = request.match_info["player_id"]
798        player = self.mass.players.get(player_id)
799        if not player:
800            raise web.HTTPNotFound(reason=f"Unknown Player: {player_id}")
801        plugin_source = provider.get_source()
802        output_format = await self.get_output_format(
803            output_format_str=request.match_info["fmt"],
804            player=player,
805            content_sample_rate=plugin_source.audio_format.sample_rate,
806            content_bit_depth=plugin_source.audio_format.bit_depth,
807        )
808        headers = {
809            **DEFAULT_STREAM_HEADERS,
810            "contentFeatures.dlna.org": "DLNA.ORG_OP=01;DLNA.ORG_FLAGS=01700000000000000000000000000000",  # noqa: E501
811            "icy-name": plugin_source.name,
812            "Accept-Ranges": "none",
813            "Content-Type": f"audio/{output_format.output_format_str}",
814        }
815
816        resp = web.StreamResponse(
817            status=200,
818            reason="OK",
819            headers=headers,
820        )
821        resp.content_type = f"audio/{output_format.output_format_str}"
822        http_profile = await self.mass.config.get_player_config_value(
823            player_id, CONF_HTTP_PROFILE, default="default", return_type=str
824        )
825        if http_profile == "forced_content_length":
826            # just set an insanely high content length to make sure the player keeps playing
827            resp.content_length = get_chunksize(output_format, 12 * 3600)
828        elif http_profile == "chunked":
829            resp.enable_chunked_encoding()
830
831        await resp.prepare(request)
832
833        # return early if this is not a GET request
834        if request.method != "GET":
835            return resp
836
837        # all checks passed, start streaming!
838        if not plugin_source.audio_format:
839            raise InvalidDataError(f"No audio format for plugin source {plugin_source_id}")
840        async for chunk in self.get_plugin_source_stream(
841            plugin_source_id=plugin_source_id,
842            output_format=output_format,
843            player_id=player_id,
844            player_filter_params=get_player_filter_params(
845                self.mass, player_id, plugin_source.audio_format, output_format
846            ),
847        ):
848            try:
849                await resp.write(chunk)
850            except (BrokenPipeError, ConnectionResetError, ConnectionError):
851                break
852        return resp
853
854    def get_command_url(self, player_or_queue_id: str, command: str) -> str:
855        """Get the url for the special command stream."""
856        return f"{self.base_url}/command/{player_or_queue_id}/{command}.mp3"
857
858    def get_announcement_url(
859        self,
860        player_id: str,
861        announce_data: AnnounceData,
862        content_type: ContentType = ContentType.MP3,
863    ) -> str:
864        """Get the url for the special announcement stream."""
865        self.announcements[player_id] = announce_data
866        # use stream server to host announcement on local network
867        # this ensures playback on all players, including ones that do not
868        # like https hosts and it also offers the pre-announce 'bell'
869        return f"{self.base_url}/announcement/{player_id}.{content_type.value}"
870
871    def get_stream(
872        self, media: PlayerMedia, pcm_format: AudioFormat, force_flow_mode: bool = False
873    ) -> AsyncGenerator[bytes, None]:
874        """
875        Get a stream of the given media as raw PCM audio.
876
877        This is used as helper for player providers that can consume the raw PCM
878        audio stream directly (e.g. AirPlay) and not rely on HTTP transport.
879        """
880        # select audio source
881        if media.media_type == MediaType.ANNOUNCEMENT:
882            # special case: stream announcement
883            assert media.custom_data
884            audio_source = self.get_announcement_stream(
885                media.custom_data["announcement_url"],
886                output_format=pcm_format,
887                pre_announce=media.custom_data["pre_announce"],
888                pre_announce_url=media.custom_data["pre_announce_url"],
889            )
890        elif media.media_type == MediaType.PLUGIN_SOURCE:
891            # special case: plugin source stream
892            assert media.custom_data
893            audio_source = self.get_plugin_source_stream(
894                plugin_source_id=media.custom_data["source_id"],
895                output_format=pcm_format,
896                # need to pass player_id from the PlayerMedia object
897                # because this could have been a group
898                player_id=media.custom_data["player_id"],
899            )
900        elif (
901            media.media_type == MediaType.FLOW_STREAM
902            and media.source_id
903            and media.source_id.startswith(UGP_PREFIX)
904            and media.uri
905            and "/ugp/" in media.uri
906        ):
907            # special case: member player accessing UGP stream
908            # Check URI to distinguish from the UGP accessing its own stream
909            ugp_player = cast("UniversalGroupPlayer", self.mass.players.get(media.source_id))
910            ugp_stream = ugp_player.stream
911            assert ugp_stream is not None  # for type checker
912            if ugp_stream.base_pcm_format == pcm_format:
913                # no conversion needed
914                audio_source = ugp_stream.subscribe_raw()
915            else:
916                audio_source = ugp_stream.get_stream(output_format=pcm_format)
917        elif (
918            media.source_id
919            and media.queue_item_id
920            and (media.media_type == MediaType.FLOW_STREAM or force_flow_mode)
921        ):
922            # regular queue (flow) stream request
923            queue = self.mass.player_queues.get(media.source_id)
924            assert queue
925            start_queue_item = self.mass.player_queues.get_item(
926                media.source_id, media.queue_item_id
927            )
928            assert start_queue_item
929            audio_source = self.mass.streams.get_queue_flow_stream(
930                queue=queue,
931                start_queue_item=start_queue_item,
932                pcm_format=pcm_format,
933            )
934        elif media.source_id and media.queue_item_id:
935            # single item stream (e.g. radio)
936            queue_item = self.mass.player_queues.get_item(media.source_id, media.queue_item_id)
937            assert queue_item
938            audio_source = buffered(
939                self.get_queue_item_stream(
940                    queue_item=queue_item,
941                    pcm_format=pcm_format,
942                ),
943                buffer_size=10,
944                min_buffer_before_yield=2,
945            )
946        else:
947            # assume url or some other direct path
948            # NOTE: this will fail if its an uri not playable by ffmpeg
949            audio_source = get_ffmpeg_stream(
950                audio_input=media.uri,
951                input_format=AudioFormat(content_type=ContentType.try_parse(media.uri)),
952                output_format=pcm_format,
953            )
954        return audio_source
955
956    @use_buffer(buffer_size=30, min_buffer_before_yield=2)
957    async def get_queue_flow_stream(
958        self,
959        queue: PlayerQueue,
960        start_queue_item: QueueItem,
961        pcm_format: AudioFormat,
962    ) -> AsyncGenerator[bytes, None]:
963        """
964        Get a flow stream of all tracks in the queue as raw PCM audio.
965
966        yields chunks of exactly 1 second of audio in the given pcm_format.
967        """
968        # ruff: noqa: PLR0915
969        assert pcm_format.content_type.is_pcm()
970        queue_track = None
971        last_fadeout_part: bytes = b""
972        last_streamdetails: StreamDetails | None = None
973        last_play_log_entry: PlayLogEntry | None = None
974        queue.flow_mode = True
975        if not start_queue_item:
976            # this can happen in some (edge case) race conditions
977            return
978        pcm_sample_size = pcm_format.pcm_sample_size
979        if start_queue_item.media_type != MediaType.TRACK:
980            # no crossfade on non-tracks
981            smart_fades_mode = SmartFadesMode.DISABLED
982            standard_crossfade_duration = 0
983        else:
984            smart_fades_mode = await self.mass.config.get_player_config_value(
985                queue.queue_id, CONF_SMART_FADES_MODE, return_type=SmartFadesMode
986            )
987            standard_crossfade_duration = self.mass.config.get_raw_player_config_value(
988                queue.queue_id, CONF_CROSSFADE_DURATION, 10
989            )
990        self.logger.info(
991            "Start Queue Flow stream for Queue %s - crossfade: %s %s",
992            queue.display_name,
993            smart_fades_mode,
994            f"({standard_crossfade_duration}s)"
995            if smart_fades_mode == SmartFadesMode.STANDARD_CROSSFADE
996            else "",
997        )
998        total_bytes_sent = 0
999        total_chunks_received = 0
1000
1001        while True:
1002            # get (next) queue item to stream
1003            if queue_track is None:
1004                queue_track = start_queue_item
1005            else:
1006                try:
1007                    queue_track = await self.mass.player_queues.load_next_queue_item(
1008                        queue.queue_id, queue_track.queue_item_id
1009                    )
1010                except QueueEmpty:
1011                    break
1012
1013            if queue_track.streamdetails is None:
1014                raise InvalidDataError(
1015                    "No Streamdetails known for queue item %s",
1016                    queue_track.queue_item_id,
1017                )
1018
1019            self.logger.debug(
1020                "Start Streaming queue track: %s (%s) for queue %s",
1021                queue_track.streamdetails.uri,
1022                queue_track.name,
1023                queue.display_name,
1024            )
1025            # append to play log so the queue controller can work out which track is playing
1026            play_log_entry = PlayLogEntry(queue_track.queue_item_id)
1027            queue.flow_mode_stream_log.append(play_log_entry)
1028            # calculate crossfade buffer size
1029            crossfade_buffer_duration = (
1030                SMART_CROSSFADE_DURATION
1031                if smart_fades_mode == SmartFadesMode.SMART_CROSSFADE
1032                else standard_crossfade_duration
1033            )
1034            crossfade_buffer_duration = min(
1035                crossfade_buffer_duration,
1036                int(queue_track.streamdetails.duration / 2)
1037                if queue_track.streamdetails.duration
1038                else crossfade_buffer_duration,
1039            )
1040            # Ensure crossfade buffer size is aligned to frame boundaries
1041            # Frame size = bytes_per_sample * channels
1042            bytes_per_sample = pcm_format.bit_depth // 8
1043            frame_size = bytes_per_sample * pcm_format.channels
1044            crossfade_buffer_size = int(pcm_format.pcm_sample_size * crossfade_buffer_duration)
1045            # Round down to nearest frame boundary
1046            crossfade_buffer_size = (crossfade_buffer_size // frame_size) * frame_size
1047
1048            bytes_written = 0
1049            buffer = b""
1050            # handle incoming audio chunks
1051            first_chunk_received = False
1052            # buffer size needs to be big enough to include the crossfade part
1053
1054            async for chunk in self.get_queue_item_stream(
1055                queue_track,
1056                pcm_format=pcm_format,
1057                seek_position=queue_track.streamdetails.seek_position,
1058                raise_on_error=False,
1059            ):
1060                total_chunks_received += 1
1061                if not first_chunk_received:
1062                    first_chunk_received = True
1063                    # inform the queue that the track is now loaded in the buffer
1064                    # so the next track can be preloaded
1065                    self.mass.player_queues.track_loaded_in_buffer(
1066                        queue.queue_id, queue_track.queue_item_id
1067                    )
1068                if total_chunks_received < 10 and smart_fades_mode != SmartFadesMode.DISABLED:
1069                    # we want a stream to start as quickly as possible
1070                    # so for the first 10 chunks we keep a very short buffer
1071                    req_buffer_size = pcm_format.pcm_sample_size
1072                else:
1073                    req_buffer_size = (
1074                        pcm_sample_size
1075                        if smart_fades_mode == SmartFadesMode.DISABLED
1076                        else crossfade_buffer_size
1077                    )
1078
1079                # ALWAYS APPEND CHUNK TO BUFFER
1080                buffer += chunk
1081                del chunk
1082                if len(buffer) < req_buffer_size:
1083                    # buffer is not full enough, move on
1084                    # yield control to event loop with 10ms delay
1085                    await asyncio.sleep(0.01)
1086                    continue
1087
1088                ####  HANDLE CROSSFADE OF PREVIOUS TRACK AND NEW TRACK
1089                if last_fadeout_part and last_streamdetails:
1090                    # perform crossfade
1091                    fadein_part = buffer[:crossfade_buffer_size]
1092                    remaining_bytes = buffer[crossfade_buffer_size:]
1093                    # Use the mixer to handle all crossfade logic
1094                    crossfade_part = await self._smart_fades_mixer.mix(
1095                        fade_in_part=fadein_part,
1096                        fade_out_part=last_fadeout_part,
1097                        fade_in_streamdetails=queue_track.streamdetails,
1098                        fade_out_streamdetails=last_streamdetails,
1099                        pcm_format=pcm_format,
1100                        standard_crossfade_duration=standard_crossfade_duration,
1101                        mode=smart_fades_mode,
1102                    )
1103                    # because the crossfade exists of both the fadein and fadeout part
1104                    # we need to correct the bytes_written accordingly so the duration
1105                    # calculations at the end of the track are correct
1106                    crossfade_part_len = len(crossfade_part)
1107                    bytes_written += int(crossfade_part_len / 2)
1108                    if last_play_log_entry:
1109                        assert last_play_log_entry.seconds_streamed is not None
1110                        last_play_log_entry.seconds_streamed += (
1111                            crossfade_part_len / 2 / pcm_sample_size
1112                        )
1113                    # yield crossfade_part (in pcm_sample_size chunks)
1114                    for _chunk in divide_chunks(crossfade_part, pcm_sample_size):
1115                        yield _chunk
1116                        del _chunk
1117                    del crossfade_part
1118                    # also write the leftover bytes from the crossfade action
1119                    if remaining_bytes:
1120                        yield remaining_bytes
1121                        bytes_written += len(remaining_bytes)
1122                        del remaining_bytes
1123                    # clear vars
1124                    last_fadeout_part = b""
1125                    last_streamdetails = None
1126                    buffer = b""
1127
1128                #### OTHER: enough data in buffer, feed to output
1129                while len(buffer) > req_buffer_size:
1130                    yield buffer[:pcm_sample_size]
1131                    bytes_written += pcm_sample_size
1132                    buffer = buffer[pcm_sample_size:]
1133
1134            #### HANDLE END OF TRACK
1135            if last_fadeout_part:
1136                # edge case: we did not get enough data to make the crossfade
1137                for _chunk in divide_chunks(last_fadeout_part, pcm_sample_size):
1138                    yield _chunk
1139                    del _chunk
1140                bytes_written += len(last_fadeout_part)
1141                last_fadeout_part = b""
1142            if self._crossfade_allowed(
1143                queue_track, smart_fades_mode=smart_fades_mode, flow_mode=True
1144            ):
1145                # if crossfade is enabled, save fadeout part to pickup for next track
1146                last_fadeout_part = buffer[-crossfade_buffer_size:]
1147                last_streamdetails = queue_track.streamdetails
1148                last_play_log_entry = play_log_entry
1149                remaining_bytes = buffer[:-crossfade_buffer_size]
1150                if remaining_bytes:
1151                    yield remaining_bytes
1152                    bytes_written += len(remaining_bytes)
1153                del remaining_bytes
1154            elif buffer:
1155                # no crossfade enabled, just yield the buffer last part
1156                bytes_written += len(buffer)
1157                for _chunk in divide_chunks(buffer, pcm_sample_size):
1158                    yield _chunk
1159                    del _chunk
1160            # make sure the buffer gets cleaned up
1161            del buffer
1162
1163            # update duration details based on the actual pcm data we sent
1164            # this also accounts for crossfade and silence stripping
1165            seconds_streamed = bytes_written / pcm_sample_size
1166            queue_track.streamdetails.seconds_streamed = seconds_streamed
1167            queue_track.streamdetails.duration = int(
1168                queue_track.streamdetails.seek_position + seconds_streamed
1169            )
1170            play_log_entry.seconds_streamed = seconds_streamed
1171            play_log_entry.duration = queue_track.streamdetails.duration
1172            total_bytes_sent += bytes_written
1173            self.logger.debug(
1174                "Finished Streaming queue track: %s (%s) on queue %s",
1175                queue_track.streamdetails.uri,
1176                queue_track.name,
1177                queue.display_name,
1178            )
1179        #### HANDLE END OF QUEUE FLOW STREAM
1180        # end of queue flow: make sure we yield the last_fadeout_part
1181        if last_fadeout_part:
1182            for _chunk in divide_chunks(last_fadeout_part, pcm_sample_size):
1183                yield _chunk
1184                del _chunk
1185            # correct seconds streamed/duration
1186            last_part_seconds = len(last_fadeout_part) / pcm_sample_size
1187            streamdetails = queue_track.streamdetails
1188            assert streamdetails is not None
1189            streamdetails.seconds_streamed = (
1190                streamdetails.seconds_streamed or 0
1191            ) + last_part_seconds
1192            streamdetails.duration = int((streamdetails.duration or 0) + last_part_seconds)
1193            last_fadeout_part = b""
1194        total_bytes_sent += bytes_written
1195        self.logger.info("Finished Queue Flow stream for Queue %s", queue.display_name)
1196
1197    async def get_announcement_stream(
1198        self,
1199        announcement_url: str,
1200        output_format: AudioFormat,
1201        pre_announce: bool | str = False,
1202        pre_announce_url: str = ANNOUNCE_ALERT_FILE,
1203    ) -> AsyncGenerator[bytes, None]:
1204        """Get the special announcement stream."""
1205        announcement_data: asyncio.Queue[bytes | None] = asyncio.Queue(10)
1206        # we are doing announcement in PCM first to avoid multiple encodings
1207        # when mixing pre-announce and announcement
1208        # also we have to deal with some TTS sources being super slow in delivering audio
1209        # so we take an approach where we start fetching the announcement in the background
1210        # while we can already start playing the pre-announce sound (if any)
1211
1212        pcm_format = (
1213            output_format
1214            if output_format.content_type.is_pcm()
1215            else AudioFormat(
1216                sample_rate=output_format.sample_rate,
1217                content_type=ContentType.PCM_S16LE,
1218                bit_depth=16,
1219                channels=output_format.channels,
1220            )
1221        )
1222
1223        async def fetch_announcement() -> None:
1224            fmt = announcement_url.rsplit(".")[-1]
1225            async for chunk in get_ffmpeg_stream(
1226                audio_input=announcement_url,
1227                input_format=AudioFormat(content_type=ContentType.try_parse(fmt)),
1228                output_format=pcm_format,
1229                chunk_size=get_chunksize(pcm_format, 1),
1230            ):
1231                await announcement_data.put(chunk)
1232            await announcement_data.put(None)  # signal end of stream
1233
1234        self.mass.create_task(fetch_announcement())
1235
1236        async def _announcement_stream() -> AsyncGenerator[bytes, None]:
1237            """Generate the PCM audio stream for the announcement + optional pre-announce."""
1238            if pre_announce:
1239                async for chunk in get_ffmpeg_stream(
1240                    audio_input=pre_announce_url,
1241                    input_format=AudioFormat(content_type=ContentType.try_parse(pre_announce_url)),
1242                    output_format=pcm_format,
1243                    chunk_size=get_chunksize(pcm_format, 1),
1244                ):
1245                    yield chunk
1246            # pad silence while we're waiting for the announcement to be ready
1247            while announcement_data.empty():
1248                yield b"\0" * int(
1249                    pcm_format.sample_rate * (pcm_format.bit_depth / 8) * pcm_format.channels * 0.1
1250                )
1251                await asyncio.sleep(0.1)
1252            # stream announcement
1253            while True:
1254                announcement_chunk = await announcement_data.get()
1255                if announcement_chunk is None:
1256                    break
1257                yield announcement_chunk
1258
1259        if output_format == pcm_format:
1260            # no need to re-encode, just yield the raw PCM stream
1261            async for chunk in _announcement_stream():
1262                yield chunk
1263            return
1264
1265        # stream final announcement in requested output format
1266        async for chunk in get_ffmpeg_stream(
1267            audio_input=_announcement_stream(),
1268            input_format=pcm_format,
1269            output_format=output_format,
1270        ):
1271            yield chunk
1272
1273    async def get_plugin_source_stream(
1274        self,
1275        plugin_source_id: str,
1276        output_format: AudioFormat,
1277        player_id: str,
1278        player_filter_params: list[str] | None = None,
1279    ) -> AsyncGenerator[bytes, None]:
1280        """Get the special plugin source stream."""
1281        plugin_prov = cast("PluginProvider", self.mass.get_provider(plugin_source_id))
1282        if not plugin_prov:
1283            raise ProviderUnavailableError(f"Unknown PluginSource: {plugin_source_id}")
1284
1285        plugin_source = plugin_prov.get_source()
1286        self.logger.debug(
1287            "Start streaming PluginSource %s to %s using output format %s",
1288            plugin_source_id,
1289            player_id,
1290            output_format,
1291        )
1292        # this should already be set by the player controller, but just to be sure
1293        plugin_source.in_use_by = player_id
1294
1295        try:
1296            async for chunk in get_ffmpeg_stream(
1297                audio_input=cast(
1298                    "str | AsyncGenerator[bytes, None]",
1299                    plugin_prov.get_audio_stream(player_id)
1300                    if plugin_source.stream_type == StreamType.CUSTOM
1301                    else plugin_source.path,
1302                ),
1303                input_format=plugin_source.audio_format,
1304                output_format=output_format,
1305                filter_params=player_filter_params,
1306                extra_input_args=["-y", "-re"],
1307            ):
1308                if plugin_source.in_use_by != player_id:
1309                    # another player took over or the stream ended, stop streaming
1310                    break
1311                yield chunk
1312        finally:
1313            self.logger.debug(
1314                "Finished streaming PluginSource %s to %s", plugin_source_id, player_id
1315            )
1316            await asyncio.sleep(1)  # prevent race conditions when selecting source
1317            if plugin_source.in_use_by == player_id:
1318                # release control
1319                plugin_source.in_use_by = None
1320
1321    async def get_queue_item_stream(
1322        self,
1323        queue_item: QueueItem,
1324        pcm_format: AudioFormat,
1325        seek_position: int = 0,
1326        raise_on_error: bool = True,
1327    ) -> AsyncGenerator[bytes, None]:
1328        """Get the (PCM) audio stream for a single queue item."""
1329        # collect all arguments for ffmpeg
1330        streamdetails = queue_item.streamdetails
1331        assert streamdetails
1332        filter_params: list[str] = []
1333
1334        # handle volume normalization
1335        gain_correct: float | None = None
1336        if streamdetails.volume_normalization_mode == VolumeNormalizationMode.DYNAMIC:
1337            # volume normalization using loudnorm filter (in dynamic mode)
1338            # which also collects the measurement on the fly during playback
1339            # more info: https://k.ylo.ph/2016/04/04/loudnorm.html
1340            filter_rule = f"loudnorm=I={streamdetails.target_loudness}:TP=-2.0:LRA=10.0:offset=0.0"
1341            filter_rule += ":print_format=json"
1342            filter_params.append(filter_rule)
1343        elif streamdetails.volume_normalization_mode == VolumeNormalizationMode.FIXED_GAIN:
1344            # apply user defined fixed volume/gain correction
1345            config_key = (
1346                CONF_VOLUME_NORMALIZATION_FIXED_GAIN_TRACKS
1347                if streamdetails.media_type == MediaType.TRACK
1348                else CONF_VOLUME_NORMALIZATION_FIXED_GAIN_RADIO
1349            )
1350            gain_value = await self.mass.config.get_core_config_value(
1351                self.domain, config_key, default=0.0, return_type=float
1352            )
1353            gain_correct = round(gain_value, 2)
1354            filter_params.append(f"volume={gain_correct}dB")
1355        elif streamdetails.volume_normalization_mode == VolumeNormalizationMode.MEASUREMENT_ONLY:
1356            # volume normalization with known loudness measurement
1357            # apply volume/gain correction
1358            target_loudness = (
1359                float(streamdetails.target_loudness)
1360                if streamdetails.target_loudness is not None
1361                else 0.0
1362            )
1363            if streamdetails.prefer_album_loudness and streamdetails.loudness_album is not None:
1364                gain_correct = target_loudness - float(streamdetails.loudness_album)
1365            elif streamdetails.loudness is not None:
1366                gain_correct = target_loudness - float(streamdetails.loudness)
1367            else:
1368                gain_correct = 0.0
1369            gain_correct = round(gain_correct, 2)
1370            filter_params.append(f"volume={gain_correct}dB")
1371        streamdetails.volume_normalization_gain_correct = gain_correct
1372
1373        allow_buffer = bool(
1374            self.mass.config.get_raw_core_config_value(
1375                self.domain, CONF_ALLOW_BUFFER, CONF_ALLOW_BUFFER_DEFAULT
1376            )
1377            and streamdetails.duration
1378        )
1379
1380        self.logger.debug(
1381            "Starting queue item stream for %s (%s)"
1382            " - using buffer: %s"
1383            " - using fade-in: %s"
1384            " - using volume normalization: %s",
1385            queue_item.name,
1386            streamdetails.uri,
1387            allow_buffer,
1388            streamdetails.fade_in,
1389            streamdetails.volume_normalization_mode,
1390        )
1391        if allow_buffer:
1392            media_stream_gen = get_buffered_media_stream(
1393                self.mass,
1394                streamdetails=streamdetails,
1395                pcm_format=pcm_format,
1396                seek_position=int(seek_position),
1397                filter_params=filter_params,
1398            )
1399        else:
1400            media_stream_gen = get_media_stream(
1401                self.mass,
1402                streamdetails=streamdetails,
1403                pcm_format=pcm_format,
1404                seek_position=int(seek_position),
1405                filter_params=filter_params,
1406            )
1407
1408        first_chunk_received = False
1409        fade_in_buffer = b""
1410        bytes_received = 0
1411        finished = False
1412        stream_started_at = asyncio.get_event_loop().time()
1413        try:
1414            async for chunk in media_stream_gen:
1415                bytes_received += len(chunk)
1416                if not first_chunk_received:
1417                    first_chunk_received = True
1418                    self.logger.debug(
1419                        "First audio chunk received for %s (%s) after %.2f seconds",
1420                        queue_item.name,
1421                        streamdetails.uri,
1422                        asyncio.get_event_loop().time() - stream_started_at,
1423                    )
1424                # handle optional fade-in
1425                if streamdetails.fade_in:
1426                    if len(fade_in_buffer) < pcm_format.pcm_sample_size * 4:
1427                        fade_in_buffer += chunk
1428                    elif fade_in_buffer:
1429                        async for fade_chunk in get_ffmpeg_stream(
1430                            # NOTE: get_ffmpeg_stream signature says str | AsyncGenerator
1431                            # but FFMpeg class actually accepts bytes too. This works at
1432                            # runtime but needs type: ignore for mypy.
1433                            audio_input=fade_in_buffer + chunk,  # type: ignore[arg-type]
1434                            input_format=pcm_format,
1435                            output_format=pcm_format,
1436                            filter_params=["afade=type=in:start_time=0:duration=3"],
1437                        ):
1438                            yield fade_chunk
1439                    fade_in_buffer = b""
1440                    streamdetails.fade_in = False
1441                else:
1442                    yield chunk
1443                # help garbage collection by explicitly deleting chunk
1444                del chunk
1445            finished = True
1446        except AudioError as err:
1447            streamdetails.stream_error = True
1448            queue_item.available = False
1449            if raise_on_error:
1450                raise
1451            # yes, we swallow the error here after logging it
1452            # so the outer stream can handle it gracefully
1453            self.logger.error(
1454                "AudioError while streaming queue item %s (%s): %s",
1455                queue_item.name,
1456                streamdetails.uri,
1457                err,
1458            )
1459        finally:
1460            # determine how many seconds we've streamed
1461            # for pcm output we can calculate this easily
1462            seconds_streamed = bytes_received / pcm_format.pcm_sample_size
1463            streamdetails.seconds_streamed = seconds_streamed
1464            self.logger.debug(
1465                "stream %s for %s in %.2f seconds - seconds streamed/buffered: %.2f",
1466                "aborted" if not finished else "finished",
1467                streamdetails.uri,
1468                asyncio.get_event_loop().time() - stream_started_at,
1469                seconds_streamed,
1470            )
1471            # report stream to provider
1472            if (finished or seconds_streamed >= 90) and (
1473                music_prov := self.mass.get_provider(streamdetails.provider)
1474            ):
1475                if TYPE_CHECKING:  # avoid circular import
1476                    assert isinstance(music_prov, MusicProvider)
1477                self.mass.create_task(music_prov.on_streamed(streamdetails))
1478
1479    @use_buffer(buffer_size=30, min_buffer_before_yield=2)
1480    async def get_queue_item_stream_with_smartfade(
1481        self,
1482        queue_item: QueueItem,
1483        pcm_format: AudioFormat,
1484        smart_fades_mode: SmartFadesMode = SmartFadesMode.SMART_CROSSFADE,
1485        standard_crossfade_duration: int = 10,
1486    ) -> AsyncGenerator[bytes, None]:
1487        """Get the audio stream for a single queue item with (smart) crossfade to the next item."""
1488        queue = self.mass.player_queues.get(queue_item.queue_id)
1489        if not queue:
1490            raise RuntimeError(f"Queue {queue_item.queue_id} not found")
1491
1492        streamdetails = queue_item.streamdetails
1493        assert streamdetails
1494        crossfade_data = self._crossfade_data.pop(queue.queue_id, None)
1495
1496        if crossfade_data and streamdetails.seek_position > 0:
1497            # don't do crossfade when seeking into track
1498            crossfade_data = None
1499        if crossfade_data and (crossfade_data.queue_item_id != queue_item.queue_item_id):
1500            # edge case alert: the next item changed just while we were preloading/crossfading
1501            self.logger.warning(
1502                "Skipping crossfade data for queue %s - next item changed!", queue.display_name
1503            )
1504            crossfade_data = None
1505
1506        self.logger.debug(
1507            "Start Streaming queue track: %s (%s) for queue %s "
1508            "- crossfade mode: %s "
1509            "- crossfading from previous track: %s ",
1510            queue_item.streamdetails.uri if queue_item.streamdetails else "Unknown URI",
1511            queue_item.name,
1512            queue.display_name,
1513            smart_fades_mode,
1514            "true" if crossfade_data else "false",
1515        )
1516
1517        buffer = b""
1518        bytes_written = 0
1519        # calculate crossfade buffer size
1520        crossfade_buffer_duration = (
1521            SMART_CROSSFADE_DURATION
1522            if smart_fades_mode == SmartFadesMode.SMART_CROSSFADE
1523            else standard_crossfade_duration
1524        )
1525        crossfade_buffer_duration = min(
1526            crossfade_buffer_duration,
1527            int(streamdetails.duration / 2)
1528            if streamdetails.duration
1529            else crossfade_buffer_duration,
1530        )
1531        # Ensure crossfade buffer size is aligned to frame boundaries
1532        # Frame size = bytes_per_sample * channels
1533        bytes_per_sample = pcm_format.bit_depth // 8
1534        frame_size = bytes_per_sample * pcm_format.channels
1535        crossfade_buffer_size = int(pcm_format.pcm_sample_size * crossfade_buffer_duration)
1536        # Round down to nearest frame boundary
1537        crossfade_buffer_size = (crossfade_buffer_size // frame_size) * frame_size
1538        fade_out_data: bytes | None = None
1539
1540        if crossfade_data:
1541            # Calculate discard amount in seconds (format-independent)
1542            # Use fade_in_pcm_format because fade_in_size is in the next track's original format
1543            fade_in_duration_seconds = (
1544                crossfade_data.fade_in_size / crossfade_data.fade_in_pcm_format.pcm_sample_size
1545            )
1546            discard_seconds = int(fade_in_duration_seconds) - 1
1547            # Calculate discard amounts in CURRENT track's format
1548            discard_bytes = int(discard_seconds * pcm_format.pcm_sample_size)
1549            # Convert fade_in_size to current track's format for correct leftover calculation
1550            fade_in_size_in_current_format = int(
1551                fade_in_duration_seconds * pcm_format.pcm_sample_size
1552            )
1553            discard_leftover = fade_in_size_in_current_format - discard_bytes
1554        else:
1555            discard_seconds = streamdetails.seek_position
1556            discard_leftover = 0
1557        total_chunks_received = 0
1558        req_buffer_size = crossfade_buffer_size
1559        async for chunk in self.get_queue_item_stream(
1560            queue_item, pcm_format, seek_position=discard_seconds
1561        ):
1562            total_chunks_received += 1
1563            if discard_leftover:
1564                # discard leftover bytes from crossfade data
1565                chunk = chunk[discard_leftover:]  # noqa: PLW2901
1566                discard_leftover = 0
1567
1568            if total_chunks_received < 10:
1569                # we want a stream to start as quickly as possible
1570                # so for the first 10 chunks we keep a very short buffer
1571                req_buffer_size = pcm_format.pcm_sample_size
1572            else:
1573                req_buffer_size = crossfade_buffer_size
1574
1575            # ALWAYS APPEND CHUNK TO BUFFER
1576            buffer += chunk
1577            del chunk
1578            if len(buffer) < req_buffer_size:
1579                # buffer is not full enough, move on
1580                continue
1581
1582            ####  HANDLE CROSSFADE DATA FROM PREVIOUS TRACK
1583            if crossfade_data:
1584                # send the (second half of the) crossfade data
1585                if crossfade_data.pcm_format != pcm_format:
1586                    # edge case: pcm format mismatch, we need to resample
1587                    self.logger.debug(
1588                        "Resampling crossfade data from %s to %s for queue %s",
1589                        crossfade_data.pcm_format.sample_rate,
1590                        pcm_format.sample_rate,
1591                        queue.display_name,
1592                    )
1593                    resampled_data = await resample_pcm_audio(
1594                        crossfade_data.data,
1595                        crossfade_data.pcm_format,
1596                        pcm_format,
1597                    )
1598                    if resampled_data:
1599                        for _chunk in divide_chunks(resampled_data, pcm_format.pcm_sample_size):
1600                            yield _chunk
1601                        bytes_written += len(resampled_data)
1602                    else:
1603                        # Resampling failed, error already logged in resample_pcm_audio
1604                        # Skip crossfade data entirely - stream continues without it
1605                        self.logger.warning(
1606                            "Skipping crossfade data for queue %s due to resampling failure",
1607                            queue.display_name,
1608                        )
1609                else:
1610                    for _chunk in divide_chunks(crossfade_data.data, pcm_format.pcm_sample_size):
1611                        yield _chunk
1612                    bytes_written += len(crossfade_data.data)
1613                # clear vars
1614                crossfade_data = None
1615
1616            #### OTHER: enough data in buffer, feed to output
1617            while len(buffer) > req_buffer_size:
1618                yield buffer[: pcm_format.pcm_sample_size]
1619                bytes_written += pcm_format.pcm_sample_size
1620                buffer = buffer[pcm_format.pcm_sample_size :]
1621
1622        #### HANDLE END OF TRACK
1623
1624        if crossfade_data:
1625            # edge case: we did not get enough data to send the crossfade data
1626            # send the (second half of the) crossfade data
1627            if crossfade_data.pcm_format != pcm_format:
1628                # (yet another) edge case: pcm format mismatch, we need to resample
1629                self.logger.debug(
1630                    "Resampling remaining crossfade data from %s to %s for queue %s",
1631                    crossfade_data.pcm_format.sample_rate,
1632                    pcm_format.sample_rate,
1633                    queue.display_name,
1634                )
1635                resampled_crossfade_data = await resample_pcm_audio(
1636                    crossfade_data.data,
1637                    crossfade_data.pcm_format,
1638                    pcm_format,
1639                )
1640                if resampled_crossfade_data:
1641                    crossfade_data.data = resampled_crossfade_data
1642                else:
1643                    # Resampling failed, error already logged in resample_pcm_audio
1644                    # Skip the crossfade data entirely
1645                    self.logger.warning(
1646                        "Skipping remaining crossfade data for queue %s due to resampling failure",
1647                        queue.display_name,
1648                    )
1649                    crossfade_data = None
1650            if crossfade_data:
1651                for _chunk in divide_chunks(crossfade_data.data, pcm_format.pcm_sample_size):
1652                    yield _chunk
1653                bytes_written += len(crossfade_data.data)
1654                crossfade_data = None
1655
1656        # get next track for crossfade
1657        next_queue_item: QueueItem | None
1658        try:
1659            self.logger.debug(
1660                "Preloading NEXT track for crossfade for queue %s",
1661                queue.display_name,
1662            )
1663            next_queue_item = await self.mass.player_queues.load_next_queue_item(
1664                queue.queue_id, queue_item.queue_item_id
1665            )
1666            # set index_in_buffer to prevent our next track is overwritten while preloading
1667            if next_queue_item.streamdetails is None:
1668                raise InvalidDataError(
1669                    f"No streamdetails for next queue item {next_queue_item.queue_item_id}"
1670                )
1671            queue.index_in_buffer = self.mass.player_queues.index_by_id(
1672                queue.queue_id, next_queue_item.queue_item_id
1673            )
1674            queue_player = self.mass.players.get(queue.queue_id)
1675            assert queue_player is not None
1676            next_queue_item_pcm_format = await self._select_pcm_format(
1677                player=queue_player,
1678                streamdetails=next_queue_item.streamdetails,
1679                smartfades_enabled=True,
1680            )
1681        except QueueEmpty:
1682            # end of queue reached, no next item
1683            next_queue_item = None
1684
1685        if not next_queue_item or not self._crossfade_allowed(
1686            queue_item,
1687            smart_fades_mode=smart_fades_mode,
1688            flow_mode=False,
1689            next_queue_item=next_queue_item,
1690            sample_rate=pcm_format.sample_rate,
1691            next_sample_rate=next_queue_item_pcm_format.sample_rate,
1692        ):
1693            # no crossfade enabled/allowed, just yield the buffer last part
1694            bytes_written += len(buffer)
1695            for _chunk in divide_chunks(buffer, pcm_format.pcm_sample_size):
1696                yield _chunk
1697        else:
1698            # if crossfade is enabled, save fadeout part in buffer to pickup for next track
1699            fade_out_data = buffer
1700            buffer = b""
1701            try:
1702                async for chunk in self.get_queue_item_stream(
1703                    next_queue_item, next_queue_item_pcm_format
1704                ):
1705                    # append to buffer until we reach crossfade size
1706                    # we only need the first X seconds of the NEXT track so we can
1707                    # perform the crossfade.
1708                    # the crossfaded audio of the previous and next track will be
1709                    # sent in two equal parts: first half now, second half
1710                    # when the next track starts. We use CrossfadeData to store
1711                    # the second half to be picked up by the next track's stream generator.
1712                    # Note that we more or less expect the user to have enabled the in-memory
1713                    # buffer so we can keep the next track's audio data in memory.
1714                    buffer += chunk
1715                    del chunk
1716                    if len(buffer) >= crossfade_buffer_size:
1717                        break
1718                ####  HANDLE CROSSFADE OF PREVIOUS TRACK AND NEW TRACK
1719                # Store original buffer size before any resampling for fade_in_size calculation
1720                # This size is in the next track's original format which is what we need
1721                original_buffer_size = len(buffer)
1722                if next_queue_item_pcm_format != pcm_format:
1723                    # edge case: pcm format mismatch, we need to resample the next track's
1724                    # beginning part before crossfading
1725                    self.logger.debug(
1726                        "Resampling next track's crossfade from %s to %s for queue %s",
1727                        next_queue_item_pcm_format.sample_rate,
1728                        pcm_format.sample_rate,
1729                        queue.display_name,
1730                    )
1731                    buffer = await resample_pcm_audio(
1732                        buffer,
1733                        next_queue_item_pcm_format,
1734                        pcm_format,
1735                    )
1736                # perform actual (smart fades) crossfade using mixer
1737                crossfade_bytes = await self._smart_fades_mixer.mix(
1738                    fade_in_part=buffer,
1739                    fade_out_part=fade_out_data,
1740                    fade_in_streamdetails=cast("StreamDetails", next_queue_item.streamdetails),
1741                    fade_out_streamdetails=streamdetails,
1742                    pcm_format=pcm_format,
1743                    standard_crossfade_duration=standard_crossfade_duration,
1744                    mode=smart_fades_mode,
1745                )
1746                # send half of the crossfade_part (= approx the fadeout part)
1747                split_point = (len(crossfade_bytes) + 1) // 2
1748                crossfade_first = crossfade_bytes[:split_point]
1749                crossfade_second = crossfade_bytes[split_point:]
1750                del crossfade_bytes
1751                bytes_written += len(crossfade_first)
1752                for _chunk in divide_chunks(crossfade_first, pcm_format.pcm_sample_size):
1753                    yield _chunk
1754                # store the other half for the next track
1755                # IMPORTANT: crossfade_second data is in CURRENT track's format (pcm_format)
1756                # because it was created from the resampled buffer used for mixing.
1757                # BUT fade_in_size represents bytes in NEXT track's original format
1758                # (next_queue_item_pcm_format) because that's how much of the next track
1759                # was consumed during the crossfade. We need both formats to correctly
1760                # handle the crossfade data when the next track starts.
1761                self._crossfade_data[queue_item.queue_id] = CrossfadeData(
1762                    data=crossfade_second,
1763                    fade_in_size=original_buffer_size,
1764                    pcm_format=pcm_format,  # Format of the data (current track)
1765                    fade_in_pcm_format=next_queue_item_pcm_format,  # Format for fade_in_size
1766                    queue_item_id=next_queue_item.queue_item_id,
1767                )
1768            except AudioError:
1769                # no crossfade possible, just yield the fade_out_data
1770                next_queue_item = None
1771                yield fade_out_data
1772                bytes_written += len(fade_out_data)
1773                del fade_out_data
1774        # make sure the buffer gets cleaned up
1775        del buffer
1776        # update duration details based on the actual pcm data we sent
1777        # this also accounts for crossfade and silence stripping
1778        seconds_streamed = bytes_written / pcm_format.pcm_sample_size
1779        streamdetails.seconds_streamed = seconds_streamed
1780        streamdetails.duration = int(streamdetails.seek_position + seconds_streamed)
1781        self.logger.debug(
1782            "Finished Streaming queue track: %s (%s) on queue %s "
1783            "- crossfade data prepared for next track: %s",
1784            streamdetails.uri,
1785            queue_item.name,
1786            queue.display_name,
1787            next_queue_item.name if next_queue_item else "N/A",
1788        )
1789
1790    def _log_request(self, request: web.Request) -> None:
1791        """Log request."""
1792        if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
1793            self.logger.log(
1794                VERBOSE_LOG_LEVEL,
1795                "Got %s request to %s from %s\nheaders: %s\n",
1796                request.method,
1797                request.path,
1798                request.remote,
1799                request.headers,
1800            )
1801        else:
1802            self.logger.debug(
1803                "Got %s request to %s from %s",
1804                request.method,
1805                request.path,
1806                request.remote,
1807            )
1808
1809    async def get_output_format(
1810        self,
1811        output_format_str: str,
1812        player: Player,
1813        content_sample_rate: int,
1814        content_bit_depth: int,
1815    ) -> AudioFormat:
1816        """Parse (player specific) output format details for given format string."""
1817        content_type: ContentType = ContentType.try_parse(output_format_str)
1818        supported_rates_conf = cast(
1819            "list[tuple[str, str]]",
1820            await self.mass.config.get_player_config_value(
1821                player.player_id, CONF_SAMPLE_RATES, unpack_splitted_values=True
1822            ),
1823        )
1824        output_channels_str = self.mass.config.get_raw_player_config_value(
1825            player.player_id, CONF_OUTPUT_CHANNELS, "stereo"
1826        )
1827        supported_sample_rates = tuple(int(x[0]) for x in supported_rates_conf)
1828        supported_bit_depths = tuple(int(x[1]) for x in supported_rates_conf)
1829
1830        player_max_bit_depth = max(supported_bit_depths)
1831        output_bit_depth = min(content_bit_depth, player_max_bit_depth)
1832        if content_sample_rate in supported_sample_rates:
1833            output_sample_rate = content_sample_rate
1834        else:
1835            output_sample_rate = max(supported_sample_rates)
1836
1837        if not content_type.is_lossless():
1838            # no point in having a higher bit depth for lossy formats
1839            output_bit_depth = 16
1840            output_sample_rate = min(48000, output_sample_rate)
1841        if output_format_str == "pcm":
1842            content_type = ContentType.from_bit_depth(output_bit_depth)
1843        return AudioFormat(
1844            content_type=content_type,
1845            sample_rate=output_sample_rate,
1846            bit_depth=output_bit_depth,
1847            channels=1 if output_channels_str != "stereo" else 2,
1848        )
1849
1850    async def _select_flow_format(
1851        self,
1852        player: Player,
1853    ) -> AudioFormat:
1854        """Parse (player specific) flow stream PCM format."""
1855        supported_rates_conf = cast(
1856            "list[tuple[str, str]]",
1857            await self.mass.config.get_player_config_value(
1858                player.player_id, CONF_SAMPLE_RATES, unpack_splitted_values=True
1859            ),
1860        )
1861        supported_sample_rates = tuple(int(x[0]) for x in supported_rates_conf)
1862        output_sample_rate = INTERNAL_PCM_FORMAT.sample_rate
1863        for sample_rate in (192000, 96000, 48000, 44100):
1864            if sample_rate in supported_sample_rates:
1865                output_sample_rate = sample_rate
1866                break
1867        return AudioFormat(
1868            content_type=INTERNAL_PCM_FORMAT.content_type,
1869            sample_rate=output_sample_rate,
1870            bit_depth=INTERNAL_PCM_FORMAT.bit_depth,
1871            channels=2,
1872        )
1873
1874    async def _select_pcm_format(
1875        self,
1876        player: Player,
1877        streamdetails: StreamDetails,
1878        smartfades_enabled: bool,
1879    ) -> AudioFormat:
1880        """Parse (player specific) stream internal PCM format."""
1881        supported_rates_conf = cast(
1882            "list[tuple[str, str]]",
1883            await self.mass.config.get_player_config_value(
1884                player.player_id, CONF_SAMPLE_RATES, unpack_splitted_values=True
1885            ),
1886        )
1887        supported_sample_rates = tuple(int(x[0]) for x in supported_rates_conf)
1888        # use highest supported rate within content rate
1889        output_sample_rate = max(
1890            (r for r in supported_sample_rates if r <= streamdetails.audio_format.sample_rate),
1891            default=48000,  # sane/safe default
1892        )
1893        # work out pcm format based on streamdetails
1894        pcm_format = AudioFormat(
1895            sample_rate=output_sample_rate,
1896            # always use f32 internally for extra headroom for filters etc
1897            content_type=INTERNAL_PCM_FORMAT.content_type,
1898            bit_depth=INTERNAL_PCM_FORMAT.bit_depth,
1899            channels=streamdetails.audio_format.channels,
1900        )
1901        if smartfades_enabled:
1902            pcm_format.channels = 2  # force stereo for crossfading
1903
1904        return pcm_format
1905
1906    def _crossfade_allowed(
1907        self,
1908        queue_item: QueueItem,
1909        smart_fades_mode: SmartFadesMode,
1910        flow_mode: bool = False,
1911        next_queue_item: QueueItem | None = None,
1912        sample_rate: int | None = None,
1913        next_sample_rate: int | None = None,
1914    ) -> bool:
1915        """Get the crossfade config for a queue item."""
1916        if smart_fades_mode == SmartFadesMode.DISABLED:
1917            return False
1918        if not (self.mass.players.get(queue_item.queue_id)):
1919            return False  # just a guard
1920        if queue_item.media_type != MediaType.TRACK:
1921            self.logger.debug("Skipping crossfade: current item is not a track")
1922            return False
1923        # check if the next item is part of the same album
1924        next_item = next_queue_item or self.mass.player_queues.get_next_item(
1925            queue_item.queue_id, queue_item.queue_item_id
1926        )
1927        if not next_item:
1928            # there is no next item!
1929            return False
1930        # check if next item is a track
1931        if next_item.media_type != MediaType.TRACK:
1932            self.logger.debug("Skipping crossfade: next item is not a track")
1933            return False
1934        if (
1935            isinstance(queue_item.media_item, Track)
1936            and isinstance(next_item.media_item, Track)
1937            and queue_item.media_item.album
1938            and next_item.media_item.album
1939            and queue_item.media_item.album == next_item.media_item.album
1940            and not self.mass.config.get_raw_core_config_value(
1941                self.domain, CONF_ALLOW_CROSSFADE_SAME_ALBUM, False
1942            )
1943        ):
1944            # in general, crossfade is not desired for tracks of the same (gapless) album
1945            # because we have no accurate way to determine if the album is gapless or not,
1946            # for now we just never crossfade between tracks of the same album
1947            self.logger.debug("Skipping crossfade: next item is part of the same album")
1948            return False
1949
1950        # check if we're allowed to crossfade on different sample rates
1951        if (
1952            not flow_mode
1953            and sample_rate
1954            and next_sample_rate
1955            and sample_rate != next_sample_rate
1956            and not self.mass.config.get_raw_player_config_value(
1957                queue_item.queue_id,
1958                CONF_ENTRY_SUPPORT_GAPLESS_DIFFERENT_SAMPLE_RATES.key,
1959                CONF_ENTRY_SUPPORT_GAPLESS_DIFFERENT_SAMPLE_RATES.default_value,
1960            )
1961        ):
1962            self.logger.debug(
1963                "Skipping crossfade: player does not support gapless playback "
1964                "with different sample rates (%s vs %s)",
1965                sample_rate,
1966                next_sample_rate,
1967            )
1968            return False
1969
1970        return True
1971
1972    async def _periodic_garbage_collection(self) -> None:
1973        """Periodic garbage collection to free up memory from audio buffers and streams."""
1974        self.logger.log(
1975            VERBOSE_LOG_LEVEL,
1976            "Running periodic garbage collection...",
1977        )
1978        # Run garbage collection in executor to avoid blocking the event loop
1979        # Since this runs periodically (not in response to subprocess cleanup),
1980        # it's safe to run in a thread without causing thread-safety issues
1981        loop = asyncio.get_running_loop()
1982        collected = await loop.run_in_executor(None, gc.collect)
1983        self.logger.log(
1984            VERBOSE_LOG_LEVEL,
1985            "Garbage collection completed, collected %d objects",
1986            collected,
1987        )
1988        # Schedule next run in 15 minutes
1989        self.mass.call_later(900, self._periodic_garbage_collection)
1990
1991    def _setup_smart_fades_logger(self, config: CoreConfig) -> None:
1992        """Set up smart fades logger level."""
1993        log_level = str(config.get_value(CONF_SMART_FADES_LOG_LEVEL))
1994        if log_level == "GLOBAL":
1995            self.smart_fades_analyzer.logger.setLevel(self.logger.level)
1996            self.smart_fades_mixer.logger.setLevel(self.logger.level)
1997        else:
1998            self.smart_fades_analyzer.logger.setLevel(log_level)
1999            self.smart_fades_mixer.logger.setLevel(log_level)
2000