music-assistant-server

1"""Various helpers for audio streaming and manipulation."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import re
8import struct
9import urllib.parse
10from collections.abc import AsyncGenerator, Iterable, Iterator
11from contextlib import aclosing
12from io import BytesIO
13from typing import TYPE_CHECKING, Final
14
15from music_assistant_models.enums import (
16    ContentType,
17    MediaType,
18    PlayerFeature,
19    PlayerType,
20    VolumeNormalizationMode,
21)
22from music_assistant_models.errors import InvalidDataError
23from music_assistant_models.streamdetails import MultiPartPath
24
25from music_assistant.constants import (
26    MASS_LOGGER_NAME,
27    VERBOSE_LOG_LEVEL,
28)
29from music_assistant.helpers.json import JSON_DECODE_EXCEPTIONS, json_loads
30
31from .ffmpeg import DEFAULT_MP3_BIT_RATE, get_ffmpeg_stream
32from .process import AsyncProcess, communicate
33
34if TYPE_CHECKING:
35    from music_assistant_models.media_items import AudioFormat
36    from music_assistant_models.streamdetails import StreamDetails
37
38    from music_assistant.mass import MusicAssistant
39    from music_assistant.models.player import Player
40
41LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.helpers.audio")
42
43HTTP_HEADERS = {"User-Agent": "Lavf/60.16.100.MusicAssistant"}
44HTTP_HEADERS_ICY = {**HTTP_HEADERS, "Icy-MetaData": "1"}
45
46SLOW_PROVIDERS = ("tidal", "ytmusic", "apple_music")
47
48# Mapping of audio format identifiers to their correct IANA MIME types
49# where the format name differs from the MIME subtype.
50# Strict DLNA/UPnP devices reject non-standard MIME types (e.g. audio/mp3).
51_MIME_TYPE_OVERRIDES: Final[dict[str, str]] = {
52    "mp3": "audio/mpeg",
53}
54
55
56def get_mime_type(format_str: str) -> str:
57    """
58    Get the proper IANA MIME type for a given audio format string.
59
60    :param format_str: The audio format string (e.g. "mp3", "flac",
61        "pcm;codec=pcm;rate=44100;bitrate=16;channels=2").
62    """
63    base_format = format_str.split(";", maxsplit=1)[0]
64    if override := _MIME_TYPE_OVERRIDES.get(base_format):
65        return override
66    return f"audio/{format_str}"
67
68
69def parse_pcm_info(content_type: str) -> tuple[int, int, int]:
70    """
71    Parse PCM info from a codec/content_type string.
72
73    :param content_type: Content type string like "pcm;codec=pcm;rate=44100;bitrate=16;channels=2".
74    """
75    params = (
76        dict(urllib.parse.parse_qsl(content_type.replace(";", "&"))) if ";" in content_type else {}
77    )
78    sample_rate = int(params.get("rate", 44100))
79    sample_size = int(params.get("bitrate", 16))
80    channels = int(params.get("channels", 2))
81    return (sample_rate, sample_size, channels)
82
83
84CACHE_CATEGORY_RESOLVED_RADIO_URL: Final[int] = 100
85CACHE_PROVIDER: Final[str] = "audio"
86
87
88def iter_pcm_slices(
89    audio: bytes,
90    pcm_format: AudioFormat,
91    target_duration_ms: int = 100,
92) -> Iterator[bytes]:
93    """
94    Yield frame-aligned PCM slices of approximately ``target_duration_ms``.
95
96    Large PCM buffers (e.g. crossfade segments or full-track reads) are split
97    into fixed-size sub-chunks so that downstream consumers get predictable
98    chunk sizes for buffering, write-timeout management, and ring-buffer
99    bookkeeping.
100
101    :param audio: Raw PCM bytes to slice.
102    :param pcm_format: Format description (sample rate, bit depth, channels).
103    :param target_duration_ms: Desired slice length in milliseconds (default 100).
104    """
105    if not audio:
106        return
107    bytes_per_sample = max(1, pcm_format.bit_depth // 8)
108    frame_size = bytes_per_sample * pcm_format.channels
109    if frame_size <= 0:
110        yield audio
111        return
112    samples_per_slice = max(1, round((target_duration_ms / 1000) * pcm_format.sample_rate))
113    slice_size = max(frame_size, samples_per_slice * frame_size)
114    offset = 0
115    audio_len = len(audio)
116    while offset < audio_len:
117        end = min(audio_len, offset + slice_size)
118        # Align to frame boundary unless this is the tail of the buffer.
119        if end < audio_len:
120            aligned_end = end - (end % frame_size)
121            if aligned_end <= offset:
122                aligned_end = min(audio_len, offset + frame_size)
123            end = aligned_end
124        yield audio[offset:end]
125        offset = end
126
127
128def align_audio_to_frame_boundary(audio_data: bytes, pcm_format: AudioFormat) -> bytes:
129    """
130    Align audio data to frame boundaries by truncating incomplete frames.
131
132    :param audio_data: Raw PCM audio data to align.
133    :param pcm_format: AudioFormat of the audio data.
134    """
135    bytes_per_sample = pcm_format.bit_depth // 8
136    frame_size = bytes_per_sample * pcm_format.channels
137    valid_bytes = (len(audio_data) // frame_size) * frame_size
138    if valid_bytes != len(audio_data):
139        LOGGER.debug(
140            "Truncating %d bytes from audio buffer to align to frame boundary",
141            len(audio_data) - valid_bytes,
142        )
143        return audio_data[:valid_bytes]
144    return audio_data
145
146
147async def strip_silence(
148    audio_data: bytes,
149    pcm_format: AudioFormat,
150    reverse: bool = False,
151) -> bytes:
152    """
153    Strip silence from begin or end of pcm audio using ffmpeg.
154
155    :param audio_data: Raw PCM audio data.
156    :param pcm_format: AudioFormat of the audio data.
157    :param reverse: If True, strip from end instead of beginning.
158    """
159    args = ["ffmpeg", "-hide_banner", "-loglevel", "quiet"]
160    args += [
161        "-acodec",
162        pcm_format.content_type.name.lower(),
163        "-f",
164        pcm_format.content_type.value,
165        "-ac",
166        str(pcm_format.channels),
167        "-ar",
168        str(pcm_format.sample_rate),
169        "-i",
170        "-",
171    ]
172    if reverse:
173        args += [
174            "-af",
175            "areverse,atrim=start=0.2,silenceremove=start_periods=1"
176            ":start_silence=0.1:start_threshold=0.02,areverse",
177        ]
178    else:
179        args += [
180            "-af",
181            "atrim=start=0.2,silenceremove=start_periods=1:start_silence=0.1:start_threshold=0.02",
182        ]
183    args += ["-f", pcm_format.content_type.value, "-"]
184    _returncode, stripped_data, _stderr = await communicate(args, audio_data)
185
186    bytes_stripped = len(audio_data) - len(stripped_data)
187    if LOGGER.isEnabledFor(VERBOSE_LOG_LEVEL):
188        seconds_stripped = round(bytes_stripped / pcm_format.pcm_sample_size, 2)
189        location = "end" if reverse else "begin"
190        LOGGER.log(
191            VERBOSE_LOG_LEVEL,
192            "stripped %s seconds of silence from %s of pcm audio. bytes stripped: %s",
193            seconds_stripped,
194            location,
195            bytes_stripped,
196        )
197    return stripped_data
198
199
200def create_wave_header(
201    samplerate: int = 44100,
202    channels: int = 2,
203    bitspersample: int = 16,
204    duration: int | None = None,
205) -> bytes:
206    """Generate a wave header from given params."""
207    file = BytesIO()
208
209    # Generate format chunk
210    format_chunk_spec = b"<4sLHHLLHH"
211    format_chunk = struct.pack(
212        format_chunk_spec,
213        b"fmt ",  # Chunk id
214        16,  # Size of this chunk (excluding chunk id and this field)
215        1,  # Audio format, 1 for PCM
216        channels,  # Number of channels
217        int(samplerate),  # Samplerate, 44100, 48000, etc.
218        int(samplerate * channels * (bitspersample / 8)),  # Byterate
219        int(channels * (bitspersample / 8)),  # Blockalign
220        bitspersample,  # 16 bits for two byte samples, etc.
221    )
222    # Generate data chunk
223    # duration = 3600*6.7
224    data_chunk_spec = b"<4sL"
225    if duration is None:
226        # use max value possible
227        datasize = 4254768000  # = 6,7 hours at 44100/16
228    else:
229        # calculate from duration
230        numsamples = samplerate * duration
231        datasize = int(numsamples * channels * (bitspersample / 8))
232    data_chunk = struct.pack(
233        data_chunk_spec,
234        b"data",  # Chunk id
235        int(datasize),  # Chunk size (excluding chunk id and this field)
236    )
237    sum_items = [
238        # "WAVE" string following size field
239        4,
240        # "fmt " + chunk size field + chunk size
241        struct.calcsize(format_chunk_spec),
242        # Size of data chunk spec + data size
243        struct.calcsize(data_chunk_spec) + datasize,
244    ]
245    # Generate main header
246    all_chunks_size = int(sum(sum_items))
247    main_header_spec = b"<4sL4s"
248    main_header = struct.pack(main_header_spec, b"RIFF", all_chunks_size, b"WAVE")
249    # Write all the contents in
250    file.write(main_header)
251    file.write(format_chunk)
252    file.write(data_chunk)
253
254    # return file.getvalue(), all_chunks_size + 8
255    return file.getvalue()
256
257
258def parse_extinf_metadata(extinf_line: str) -> dict[str, str]:
259    """
260    Parse metadata from HLS EXTINF line.
261
262    Extracts structured metadata like title="...", artist="..." from EXTINF lines.
263    Common in iHeartRadio and other commercial radio HLS streams.
264
265    :param extinf_line: The EXTINF line containing metadata
266    """
267    metadata = {}
268
269    # Pattern to match key="value" pairs in the EXTINF line
270    # Handles nested quotes by matching everything until the closing quote
271    pattern = r'(\w+)="([^"]*)"'
272
273    matches = re.findall(pattern, extinf_line)
274    for key, value in matches:
275        metadata[key.lower()] = value
276
277    # Fallback: RFC 8216 plain title format `#EXTINF:<duration>,<title>`
278    if not metadata and "," in extinf_line:
279        title = extinf_line.split(",", 1)[1].strip()
280        if title:
281            metadata["title"] = title
282
283    return metadata
284
285
286def get_parts_from_position(
287    parts: list[MultiPartPath],
288    seek_position: int,
289) -> tuple[list[MultiPartPath], int]:
290    """
291    Get the remaining parts list from a timestamp.
292
293    Arguments:
294    parts: The list of  parts
295    seek_position: The seeking position in seconds of the tracklist
296
297    Returns:
298        In a tuple, A list of  parts, starting with the one at the requested
299        seek position and the position in seconds to seek to in the first
300        track.
301    """
302    skipped_duration = 0.0
303    for i, part in enumerate(parts):
304        if not isinstance(part, MultiPartPath):
305            raise InvalidDataError("Multi-file streamdetails requires a list of MultiPartPath")
306        if part.duration is None:
307            return parts, seek_position
308        if skipped_duration + part.duration < seek_position:
309            skipped_duration += part.duration
310            continue
311
312        position = seek_position - skipped_duration
313
314        # Seeking in some parts is inaccurate, making the seek to a chapter land on the end of
315        # the previous track. If we're within 2 second of the end, skip the current track
316        if position + 2 >= part.duration:
317            LOGGER.debug(
318                f"Skipping to the next part due to seek position being at the end: {position}",
319            )
320            if i + 1 < len(parts):
321                return parts[i + 1 :], 0
322            return parts[i:], int(position)  # last part, cannot skip
323
324        return parts[i:], int(position)
325
326    raise IndexError(f"Could not find any candidate part for position {seek_position}")
327
328
329def build_concat_filelist(paths: list[str]) -> str:
330    """
331    Build the file list content for ffmpeg's concat demuxer.
332
333    :param paths: The file paths to include, in playback order.
334    """
335    lines = []
336    for path in paths:
337        # The concat demuxer uses single quotes as delimiters, so a literal quote in the
338        # path must be written as '\'' to prevent the path being truncated at the quote.
339        escaped_path = path.replace("'", "'\\''")
340        lines.append(f"file '{escaped_path}'\n")
341    return "".join(lines)
342
343
344async def realtime_pcm_pacer(
345    inner: AsyncGenerator[bytes],
346    pcm_format: AudioFormat,
347) -> AsyncGenerator[bytes]:
348    """
349    Pace a PCM byte stream at the format's native rate.
350
351    Useful for live AudioSource streams whose producer is not realtime-paced
352    (e.g. librespot's pipe backend) — without rate-limiting the consumer would
353    buffer many seconds of audio ahead of playback, making skip/next laggy.
354
355    :param inner: Source generator yielding raw PCM bytes.
356    :param pcm_format: PCM format the inner generator emits.
357    """
358    bytes_per_second = pcm_format.sample_rate * pcm_format.channels * (pcm_format.bit_depth // 8)
359    if bytes_per_second <= 0 or not pcm_format.content_type.is_pcm():
360        # non-PCM or malformed format: pass through unchanged
361        async for chunk in inner:
362            yield chunk
363        return
364    loop = asyncio.get_running_loop()
365    start_time = loop.time()
366    total_bytes = 0
367    async for chunk in inner:
368        yield chunk
369        total_bytes += len(chunk)
370        expected_elapsed = total_bytes / bytes_per_second
371        actual_elapsed = loop.time() - start_time
372        if actual_elapsed < expected_elapsed:
373            await asyncio.sleep(expected_elapsed - actual_elapsed)
374
375
376async def audio_source_silence_keepalive(
377    inner: AsyncGenerator[bytes],
378    pcm_format: AudioFormat,
379    silence_chunk_ms: int = 100,
380    idle_threshold_s: float | None = None,
381) -> AsyncGenerator[bytes]:
382    """
383    Wrap a live AudioSource PCM stream and emit silence during idle gaps.
384
385    Plugin providers exposing an AudioSource may stop yielding bytes while the
386    upstream device is paused (e.g. user paused in the Spotify app). Without
387    bytes flowing the downstream consumer (ffmpeg / the player) may disconnect.
388    This wrapper inserts ``silence_chunk_ms`` worth of zero bytes whenever the
389    inner generator hasn't produced for ``idle_threshold_s`` seconds, while
390    relaying real bytes immediately when they arrive.
391
392    Only meaningful for PCM streams — injecting raw zero bytes into a compressed
393    stream (MP3/AAC/etc.) would corrupt the bitstream. For non-PCM ``pcm_format``
394    inputs the wrapper degrades to a transparent pass-through.
395
396    :param inner: The underlying async generator yielding raw PCM bytes.
397    :param pcm_format: PCM format the inner generator emits (used to size the
398        silence chunk so it lines up to a frame boundary).
399    :param silence_chunk_ms: Duration of each silence chunk in milliseconds.
400    :param idle_threshold_s: Seconds without input before silence is inserted.
401        Defaults to the chunk duration so silence flows at realtime — critical
402        for keeping HTTP consumers (Sonos, Chromecast) connected.
403    """
404    if idle_threshold_s is None:
405        idle_threshold_s = silence_chunk_ms / 1000
406    frame_size = pcm_format.channels * (pcm_format.bit_depth // 8)
407    bytes_per_second = (
408        pcm_format.sample_rate * frame_size if pcm_format.content_type.is_pcm() else 0
409    )
410    if bytes_per_second <= 0 or frame_size <= 0:
411        # non-PCM or malformed format: pass through unchanged, no silence injection
412        async for chunk in inner:
413            yield chunk
414        return
415
416    # Round the silence chunk size DOWN to a whole-frame multiple so emitted
417    # chunks line up to PCM frame boundaries for arbitrary silence_chunk_ms /
418    # sample-rate combinations.
419    raw_silence_bytes = bytes_per_second * silence_chunk_ms // 1000
420    silence_bytes = max(frame_size, (raw_silence_bytes // frame_size) * frame_size)
421    silence_chunk = b"\x00" * silence_bytes
422    # empty bytes is the end-of-stream sentinel; real PCM frames are never empty
423    queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=8)
424
425    async def _producer() -> None:
426        # aclosing ensures inner.aclose() runs on cancellation so the underlying
427        # generator's own finally (e.g. plugin lock release, fd cleanup) fires
428        # instead of leaking until GC.
429        try:
430            async with aclosing(inner) as managed_inner:
431                async for chunk in managed_inner:
432                    await queue.put(chunk)
433        finally:
434            await queue.put(b"")
435
436    producer_task = asyncio.create_task(_producer())
437    try:
438        while True:
439            try:
440                chunk = await asyncio.wait_for(queue.get(), timeout=idle_threshold_s)
441            except TimeoutError:
442                yield silence_chunk
443                continue
444            if not chunk:
445                break
446            yield chunk
447    finally:
448        producer_task.cancel()
449        try:
450            await producer_task
451        except asyncio.CancelledError:
452            pass
453        except Exception:
454            # log but don't re-raise: we're already in a finally and the
455            # downstream consumer has its own error handling for the outer stream.
456            LOGGER.exception("AudioSource producer task raised")
457
458
459async def get_silence(
460    duration: int,
461    output_format: AudioFormat,
462) -> AsyncGenerator[bytes]:
463    """Create stream of silence, encoded to format of choice."""
464    if output_format.content_type.is_pcm():
465        # pcm = just zeros
466        for _ in range(duration):
467            yield b"\0" * int(output_format.sample_rate * (output_format.bit_depth / 8) * 2)
468        return
469    if output_format.content_type == ContentType.WAV:
470        # wav silence = wave header + zero's
471        yield create_wave_header(
472            samplerate=output_format.sample_rate,
473            channels=2,
474            bitspersample=output_format.bit_depth,
475            duration=duration,
476        )
477        for _ in range(duration):
478            yield b"\0" * int(output_format.sample_rate * (output_format.bit_depth / 8) * 2)
479        return
480    # use ffmpeg for all other encodings
481    args = [
482        "ffmpeg",
483        "-hide_banner",
484        "-loglevel",
485        "quiet",
486        "-f",
487        "lavfi",
488        "-i",
489        f"anullsrc=r={output_format.sample_rate}:cl={'stereo'}",
490        "-t",
491        str(duration),
492        "-f",
493        output_format.output_format_str,
494        "-",
495    ]
496    async with AsyncProcess(args, stdout=True) as ffmpeg_proc:
497        async for chunk in ffmpeg_proc.iter_chunked():
498            yield chunk
499
500
501async def resample_pcm_audio(
502    input_audio: bytes | AsyncGenerator[bytes],
503    input_format: AudioFormat,
504    output_format: AudioFormat,
505    chunk_size: int | None = None,
506) -> AsyncGenerator[bytes]:
507    """
508    Resample PCM audio from input_format to output_format using ffmpeg.
509
510    Yields chunks of resampled audio as they become available.
511
512    :param input_audio: Raw PCM audio data or async generator of PCM chunks.
513    :param input_format: AudioFormat of the input audio.
514    :param output_format: Desired AudioFormat for the output audio.
515    :param chunk_size: Output chunk size in bytes. Defaults to 1 second of output PCM.
516    """
517    if chunk_size is None:
518        chunk_size = output_format.pcm_sample_size
519
520    async def _as_generator() -> AsyncGenerator[bytes]:
521        if isinstance(input_audio, bytes):
522            yield input_audio
523        else:
524            async for chunk in input_audio:
525                yield chunk
526
527    if input_format == output_format:
528        buffer = b""
529        async for chunk in _as_generator():
530            buffer += chunk
531            while len(buffer) >= chunk_size:
532                yield buffer[:chunk_size]
533                buffer = buffer[chunk_size:]
534        if buffer:
535            yield buffer
536        return
537
538    async for chunk in get_ffmpeg_stream(
539        audio_input=_as_generator(),
540        input_format=input_format,
541        output_format=output_format,
542        chunk_size=chunk_size,
543    ):
544        yield chunk
545
546
547def calculate_content_length(
548    fmt: AudioFormat,
549    seconds: float = 1,
550) -> int:
551    """
552    Calculate the estimated encoded size in bytes for a given format and duration.
553
554    For CBR lossy formats (MP3/AAC), the estimate is near-exact.
555    For lossless formats (FLAC), the estimate uses an empirical average
556    compression ratio and may differ from actual size by up to ~15%.
557    For uncompressed formats (PCM/WAV), the result is exact.
558
559    :param fmt: The audio format to estimate size for.
560    :param seconds: Duration in seconds.
561    """
562    pcm_size = int(fmt.sample_rate * (fmt.bit_depth / 8) * fmt.channels * seconds)
563    if fmt.content_type.is_pcm():
564        return pcm_size
565    if fmt.content_type in (ContentType.WAV, ContentType.AIFF, ContentType.DSF):
566        return pcm_size
567    if fmt.bit_rate and fmt.bit_rate < 10000:
568        return int(((fmt.bit_rate * 1000) / 8) * seconds)
569    if fmt.content_type in (ContentType.FLAC, ContentType.WAVPACK, ContentType.ALAC):
570        # FLAC compression_level 0: empirical ratio ~74.7% of PCM
571        # Source: https://z-issue.com/wp/flac-compression-level-comparison/
572        # Real-world variance: 65-85% depending on audio content.
573        return int(pcm_size * 0.747)
574    if fmt.content_type == ContentType.MP3:
575        return int(((DEFAULT_MP3_BIT_RATE * 1000) / 8) * seconds)
576    if fmt.content_type == ContentType.OGG:
577        return int((320000 / 8) * seconds)
578    if fmt.content_type in (ContentType.AAC, ContentType.M4A):
579        # CBR 256kbps as set in get_ffmpeg_args
580        return int((256000 / 8) * seconds)
581    return int((320000 / 8) * seconds)
582
583
584def get_output_format_key(fmt: AudioFormat) -> str:
585    """
586    Get a stable key representing the output encoding parameters.
587
588    :param fmt: The output audio format.
589    """
590    return f"{fmt.content_type.value}_{fmt.sample_rate}_{fmt.bit_depth}_{fmt.channels}"
591
592
593CONTENT_LENGTH_CACHE_CATEGORY = 50
594CONTENT_LENGTH_CACHE_PROVIDER = "audio"
595CONTENT_LENGTH_CACHE_EXPIRATION = 365 * 86400  # 1 year
596
597
598async def get_content_length(
599    mass: MusicAssistant,
600    uri: str,
601    output_format: AudioFormat,
602    seconds: float,
603) -> int:
604    """
605    Get the estimated encoded size, using cached actual measurement when available.
606
607    After a track has been fully streamed, its actual content size and duration
608    are cached. On subsequent plays this gives a near-exact content_length:
609    - Exact when the requested duration matches the cached duration.
610    - Very accurate when the duration differs (derived bytes-per-second).
611
612    Falls back to the static estimate from calculate_content_length() if no cache entry exists.
613
614    :param mass: The MusicAssistant instance (for cache access).
615    :param uri: The media URI (e.g. "qobuz://track/12345").
616    :param output_format: The output audio format.
617    :param seconds: Duration in seconds to estimate.
618    """
619    cache_key = f"{uri}/{get_output_format_key(output_format)}"
620    cached: dict[str, float] | None = await mass.cache.get(
621        cache_key,
622        provider=CONTENT_LENGTH_CACHE_PROVIDER,
623        category=CONTENT_LENGTH_CACHE_CATEGORY,
624    )
625    if cached is not None:
626        cached_size = cached["size"]
627        cached_duration = cached["duration"]
628        if abs(seconds - cached_duration) < 1:
629            # same duration: return the exact cached size
630            return int(cached_size)
631        # different duration: derive bytes-per-second from the cached measurement
632        return int((cached_size / cached_duration) * seconds)
633    return calculate_content_length(output_format, seconds)
634
635
636async def store_content_length_in_cache(
637    mass: MusicAssistant,
638    uri: str,
639    output_format: AudioFormat,
640    content_size: int,
641    seconds_streamed: float,
642) -> None:
643    """
644    Store the actual content size after a track has been fully streamed.
645
646    :param mass: The MusicAssistant instance (for cache access).
647    :param uri: The media URI (e.g. "qobuz://track/12345").
648    :param output_format: The output audio format used for encoding.
649    :param content_size: Total encoded bytes sent to the player.
650    :param seconds_streamed: Duration of audio streamed in seconds.
651    """
652    if seconds_streamed < 10 or content_size < 1000:
653        return
654    cache_key = f"{uri}/{get_output_format_key(output_format)}"
655    await mass.cache.set(
656        cache_key,
657        {"size": content_size, "duration": seconds_streamed},
658        expiration=CONTENT_LENGTH_CACHE_EXPIRATION,
659        provider=CONTENT_LENGTH_CACHE_PROVIDER,
660        category=CONTENT_LENGTH_CACHE_CATEGORY,
661        persistent=True,
662    )
663
664
665PROBED_DURATION_CACHE_CATEGORY = 51
666PROBED_DURATION_CACHE_PROVIDER = "audio"
667PROBED_DURATION_CACHE_EXPIRATION = 365 * 86400  # 1 year
668
669
670async def get_probed_duration(mass: MusicAssistant, uri: str) -> int | None:
671    """
672    Get the duration determined during an earlier playback of the given item, if any.
673
674    Use for items whose provider does not report a duration, such as podcast episodes
675    from a feed without itunes:duration.
676
677    :param mass: The MusicAssistant instance (for cache access).
678    :param uri: The media item URI (e.g. "overcast--1://podcast_episode/abc").
679    :return: The duration in seconds, or None if the item was never played.
680    """
681    duration: int | None = await mass.cache.get(
682        uri,
683        provider=PROBED_DURATION_CACHE_PROVIDER,
684        category=PROBED_DURATION_CACHE_CATEGORY,
685    )
686    return duration
687
688
689async def store_probed_duration(mass: MusicAssistant, uri: str, duration: int) -> None:
690    """
691    Store the duration of an item that was determined while streaming it.
692
693    A duration below a second is ignored.
694
695    :param mass: The MusicAssistant instance (for cache access).
696    :param uri: The media item URI (e.g. "overcast--1://podcast_episode/abc").
697    :param duration: The duration in seconds.
698    """
699    if duration < 1:
700        return
701    await mass.cache.set(
702        uri,
703        duration,
704        expiration=PROBED_DURATION_CACHE_EXPIRATION,
705        provider=PROBED_DURATION_CACHE_PROVIDER,
706        category=PROBED_DURATION_CACHE_CATEGORY,
707        persistent=True,
708    )
709
710
711def get_bit_rate(fmt: AudioFormat) -> int:
712    """Get the (estimated) bit rate for a given AudioFormat, if known."""
713    if fmt.bit_rate:
714        return int(fmt.bit_rate / 1000) if fmt.bit_rate >= 10000 else fmt.bit_rate
715    return int((calculate_content_length(fmt, seconds=1) / 1000) * 8)
716
717
718def resolve_output_player_ids(
719    mass: MusicAssistant,
720    player_ids: Iterable[str],
721) -> set[str]:
722    """
723    Resolve output destinations to their user-facing player identifiers.
724
725    :param mass: Music Assistant instance.
726    :param player_ids: Player or protocol-player identifiers to resolve.
727    :return: Deduplicated user-facing player identifiers.
728    """
729    resolved_ids: set[str] = set()
730    for player_id in player_ids:
731        player = mass.players.get_player(player_id)
732        resolved_ids.add(
733            player.protocol_parent_id if player and player.protocol_parent_id else player_id
734        )
735    return resolved_ids
736
737
738def is_grouping_preventing_dsp(player: Player) -> bool:
739    """
740    Check if grouping is preventing DSP from being applied to this leader/PlayerGroup.
741
742    If this returns True, no DSP should be applied to the player.
743    This function will not check if the Player is in a group, the caller should do that first.
744    """
745    # We require the caller to handle non-leader cases themselves since player.state.synced_to
746    # can be unreliable in some edge cases
747    multi_device_dsp_supported = PlayerFeature.MULTI_DEVICE_DSP in player.state.supported_features
748    child_count = len(player.state.group_members) if player.state.group_members else 0
749
750    is_multiple_devices: bool
751    if player.provider.domain == "player_group":
752        # PlayerGroups have no leader, so having a child count of 1 means
753        # the group actually contains only a single player.
754        is_multiple_devices = child_count > 1
755    elif player.state.type == PlayerType.GROUP:
756        # This is an group player external to Music Assistant.
757        is_multiple_devices = True
758    else:
759        is_multiple_devices = child_count > 0
760    return is_multiple_devices and not multi_device_dsp_supported
761
762
763def parse_loudnorm(raw_stderr: bytes | str) -> float | None:
764    """Parse Loudness measurement from ffmpeg stderr output."""
765    stderr_data = raw_stderr.decode() if isinstance(raw_stderr, bytes) else raw_stderr
766    if "[Parsed_loudnorm_0 @" not in stderr_data:
767        return None
768    for jsun_chunk in stderr_data.split(" { "):
769        try:
770            stderr_data = "{" + jsun_chunk.rsplit("}")[0].strip() + "}"
771            loudness_data = json_loads(stderr_data)
772            return float(loudness_data["input_i"])
773        except (*JSON_DECODE_EXCEPTIONS, KeyError, ValueError, IndexError):
774            continue
775    return None
776
777
778def get_normalization_mode(
779    preference: VolumeNormalizationMode,
780    volume_normalization_enabled: bool,
781    streamdetails: StreamDetails,
782) -> VolumeNormalizationMode:
783    """
784    Get the volume normalization mode for a given queue and stream.
785
786    :param preference: The configured normalization preference for the stream's media type
787        (tracks or radio), from the streams core config.
788    :param volume_normalization_enabled: Whether normalization is enabled for the queue, already
789        resolved from the per-queue setting and its global (queue controller) fallback.
790    :param streamdetails: The stream to evaluate.
791    """
792    if not volume_normalization_enabled:
793        # disabled for this queue
794        return VolumeNormalizationMode.DISABLED
795    if streamdetails.media_type == MediaType.AUDIO_SOURCE:
796        # live/realtime: upstream producer owns loudness, no measurement to converge on
797        return VolumeNormalizationMode.DISABLED
798    if streamdetails.media_type == MediaType.SOUND_EFFECT:
799        # never measured, and the dynamic fallback compresses short clips
800        return VolumeNormalizationMode.DISABLED
801    if streamdetails.target_loudness is None:
802        # no target loudness set, disable normalization
803        return VolumeNormalizationMode.DISABLED
804
805    # handle no measurement available but fallback to dynamic mode is allowed
806    if streamdetails.loudness is None and preference == VolumeNormalizationMode.FALLBACK_DYNAMIC:
807        return VolumeNormalizationMode.DYNAMIC
808
809    # handle no measurement available and no fallback allowed
810    if streamdetails.loudness is None and preference == VolumeNormalizationMode.MEASUREMENT_ONLY:
811        return VolumeNormalizationMode.DISABLED
812
813    # handle no measurement available and fallback to fixed gain is allowed
814    if streamdetails.loudness is None and preference == VolumeNormalizationMode.FALLBACK_FIXED_GAIN:
815        return VolumeNormalizationMode.FIXED_GAIN
816
817    # handle measurement available - chosen mode is measurement
818    if streamdetails.loudness is not None and preference not in (
819        VolumeNormalizationMode.DISABLED,
820        VolumeNormalizationMode.FIXED_GAIN,
821        VolumeNormalizationMode.DYNAMIC,
822    ):
823        return VolumeNormalizationMode.MEASUREMENT_ONLY
824
825    # simply return the preference
826    return preference
827