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