/
/
/
1"""FFMpeg related helpers."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import re
8import time
9from collections import deque
10from collections.abc import AsyncGenerator, Sequence
11from contextlib import suppress
12from copy import copy
13from dataclasses import dataclass
14from typing import TYPE_CHECKING, Final
15
16from music_assistant_models.enums import ContentType
17from music_assistant_models.errors import AudioError
18from music_assistant_models.helpers import get_global_cache_value, set_global_cache_values
19
20from music_assistant.constants import VERBOSE_LOG_LEVEL
21
22from .dsp import ComplexFilter, ComplexFilterInput
23from .process import AsyncProcess, check_output
24from .util import close_async_generator
25
26if TYPE_CHECKING:
27 from music_assistant_models.media_items import AudioFormat
28
29LOGGER = logging.getLogger("ffmpeg")
30MINIMAL_FFMPEG_VERSION = 7
31CACHE_ATTR_LIBSOXR_PRESENT: Final[str] = "libsoxr_present"
32CACHE_ATTR_FFMPEG_VERSION: Final[str] = "ffmpeg_version"
33CACHE_ATTR_HLS_CMAF_BLOCKED: Final[str] = "hls_cmaf_blocked"
34DEFAULT_MP3_BIT_RATE: Final[int] = 320
35
36# FFmpeg's mono->stereo rematrix spreads a source at 1/sqrt(2) per channel; this factor
37# restores its original level. _get_channel_conform_filter avoids the same loss on the
38# main decode path by duplicating the channel instead.
39_MONO_WIDEN_COMPENSATION: Final[float] = 2**0.5
40
41# FFmpeg applies these to the single input they precede, not to the command as a whole,
42# so every input we open has to bring its own copy.
43_INPUT_READ_ARGS: Final[list[str]] = [
44 "-protocol_whitelist",
45 "file,hls,http,https,tcp,tls,crypto,pipe,data,fd,rtp,udp,concat",
46 "-probesize",
47 "8096",
48 "-analyzeduration",
49 "500000", # 0.5 seconds should be enough to detect the format
50]
51
52# Regex patterns to extract audio format details from ffmpeg's stderr output.
53# Examples of the lines we parse:
54# Stream #0:0: Audio: mp3, 44100 Hz, stereo, fltp, 320 kb/s
55# Stream #0:0(eng): Audio: aac (LC) (mp4a / 0x6134706D), 44100 Hz, stereo, fltp, 254 kb/s
56# Stream #0:0: Audio: flac, 96000 Hz, stereo, s32 (24 bit)
57# Duration: 00:03:25.78, start: 0.000000, bitrate: 320 kb/s
58_FFMPEG_SAMPLE_RATE_RE: Final = re.compile(r"(\d+) Hz")
59_FFMPEG_BIT_RATE_RE: Final = re.compile(r"(\d+) kb/s")
60_FFMPEG_EXPLICIT_BIT_DEPTH_RE: Final = re.compile(r"\((\d+) bit\)")
61_FFMPEG_SAMPLE_FMT_RE: Final = re.compile(r"\b(u8p?|s16p?|s24p?|s32p?|fltp?|dblp?)\b")
62_FFMPEG_DURATION_RE: Final = re.compile(r"Duration: (\d+):(\d+):(\d+(?:\.\d+)?)")
63
64# Mapping from ffmpeg sample format token to bit depth.
65# Note: planar variants (suffix 'p') describe memory layout only.
66# Floating point formats (flt/fltp/dbl/dblp) are typically the decoder's internal
67# representation for lossy codecs and do not reflect source bit depth, so the
68# caller decides whether to apply them based on the codec.
69_SAMPLE_FMT_BIT_DEPTH: Final[dict[str, int]] = {
70 "u8": 8,
71 "u8p": 8,
72 "s16": 16,
73 "s16p": 16,
74 "s24": 24,
75 "s24p": 24,
76 "s32": 32,
77 "s32p": 32,
78 "flt": 32,
79 "fltp": 32,
80 "dbl": 64,
81 "dblp": 64,
82}
83
84
85@dataclass
86class FFMpegStreamInfo:
87 """Audio format details parsed from an ffmpeg 'Stream #' log line."""
88
89 codec: ContentType
90 sample_rate: int | None = None
91 bit_depth: int | None = None
92 bit_rate: int | None = None
93
94
95class FFMpeg(AsyncProcess):
96 """FFMpeg wrapped as AsyncProcess."""
97
98 def __init__(
99 self,
100 audio_input: AsyncGenerator[bytes] | str | int,
101 input_format: AudioFormat,
102 output_format: AudioFormat,
103 filter_params: Sequence[str | ComplexFilter] | None = None,
104 extra_input_args: list[str] | None = None,
105 extra_output_args: list[str] | None = None,
106 audio_output: str | int = "-",
107 collect_log_history: bool = False,
108 loglevel: str = "info",
109 ) -> None:
110 """Initialize AsyncProcess."""
111 ffmpeg_args = get_ffmpeg_args(
112 input_format=input_format,
113 output_format=output_format,
114 filter_params=filter_params or [],
115 input_path=audio_input if isinstance(audio_input, str) else "-",
116 output_path=audio_output if isinstance(audio_output, str) else "-",
117 extra_input_args=extra_input_args or [],
118 extra_output_args=extra_output_args or [],
119 loglevel=loglevel,
120 )
121 self.audio_input = audio_input
122 self.input_format = input_format
123 self.collect_log_history = collect_log_history
124 self.log_history: deque[str] = deque(maxlen=100)
125 self.concat_error = False # switch to True if concat demuxer fails on MultiPartFiles
126 # Audio format details for the input and output stream as detected from ffmpeg's
127 # own stderr probe output. input_stream_info is also mirrored onto self.input_format
128 # so callers that share the AudioFormat (e.g. streamdetails) pick up the corrected
129 # values; output_stream_info is informational (useful for logging / future UI use).
130 self.input_stream_info: FFMpegStreamInfo | None = None
131 self.output_stream_info: FFMpegStreamInfo | None = None
132 # Source duration in (whole) seconds as detected from the ffmpeg input log line,
133 # or None if not yet parsed / not reported (e.g. live radio streams).
134 self.parsed_duration: int | None = None
135 self._stdin_feeder_task: asyncio.Task[None] | None = None
136 self._stdin_feeder_exception: Exception | None = None
137 self._stderr_reader_task: asyncio.Task[None] | None = None
138 # holds the detached abort-on-corrupt-stream task from _log_reader_task so it
139 # isn't garbage collected mid-flight; not otherwise awaited
140 self._abort_task: asyncio.Task[None] | None = None
141 # ffmpeg emits 'Input #N, ...' and 'Output #N, ...' headers before each block of
142 # 'Stream #' lines; we track which block the next stream line belongs to.
143 # Defaults to "input" so a stray Stream # line before any header still routes there.
144 self._current_log_section: str = "input"
145 stdin: bool | int
146 if audio_input == "-" or isinstance(audio_input, AsyncGenerator):
147 stdin = True
148 else:
149 stdin = audio_input if isinstance(audio_input, int) else False
150 stdout = audio_output if isinstance(audio_output, int) else bool(audio_output == "-")
151 super().__init__(
152 ffmpeg_args,
153 stdin=stdin,
154 stdout=stdout,
155 stderr=True,
156 )
157 self.logger = LOGGER
158
159 @property
160 def stdin_feeder_exception(self) -> Exception | None:
161 """Return the exception raised by the stdin feeder task, if any."""
162 return self._stdin_feeder_exception
163
164 async def start(self) -> None:
165 """Perform Async init of process."""
166 await super().start()
167 if self.proc:
168 self.logger = LOGGER.getChild(str(self.proc.pid))
169 clean_args = []
170 for arg in self._args[1:]:
171 if arg.startswith("http"):
172 clean_args.append("<URL>")
173 elif "/" in arg and "." in arg:
174 clean_args.append("<FILE>")
175 elif arg.startswith("data:application/"):
176 clean_args.append("<DATA>")
177 else:
178 clean_args.append(arg)
179 args_str = " ".join(clean_args)
180 self.logger.log(VERBOSE_LOG_LEVEL, "started with args: %s", args_str)
181 self._stderr_reader_task = asyncio.create_task(self._log_reader_task())
182 if isinstance(self.audio_input, AsyncGenerator):
183 self._stdin_feeder_task = asyncio.create_task(self._feed_stdin())
184
185 async def communicate(
186 self,
187 input: bytes | None = None, # noqa: A002
188 timeout: float | None = None,
189 ) -> tuple[bytes, bytes]:
190 """Override communicate to avoid blocking."""
191 if self._stdin_feeder_task:
192 await self._cancel_and_await(self._stdin_feeder_task, "stdin feeder")
193 if self._stderr_reader_task:
194 await self._cancel_and_await(self._stderr_reader_task, "stderr reader")
195 return await super().communicate(input, timeout)
196
197 async def _log_reader_task(self) -> None:
198 """Read ffmpeg log from stderr."""
199 decode_errors = 0
200 decode_errors_reported = False
201 async for line in self.iter_stderr():
202 if self.collect_log_history:
203 self.log_history.append(line)
204 # ffmpeg logging can be quite verbose, so we only log critical errors
205 # unless verbose logging is enabled
206 if "critical" in line:
207 self.logger.error(line)
208 elif self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
209 self.logger.log(VERBOSE_LOG_LEVEL, line)
210
211 if "Invalid data found when processing input" in line:
212 decode_errors += 1
213 if decode_errors >= 50 and not decode_errors_reported:
214 # stream is too corrupted to bother decoding further: report once (instead
215 # of promoting every remaining line to ERROR) and abort. close() awaits
216 # this very stderr reader task, and a task awaiting itself raises
217 # RuntimeError, so the abort must run as a detached task rather than be
218 # awaited here.
219 decode_errors_reported = True
220 self.logger.error(
221 "Excessive decode errors (%d+) for this stream; aborting", decode_errors
222 )
223 self._abort_task = asyncio.create_task(self.close())
224
225 # Log reconnection events for radio streams
226 if "Opening" in line or "Reconnect" in line or "reconnect" in line:
227 self.logger.debug("FFmpeg: %s", line)
228
229 if "Error during demuxing" in line:
230 # this can occur if using the concat demuxer for multipart files
231 # and should raise an exception to prevent false progress logging
232 self.concat_error = True
233
234 # Track which ffmpeg block we're currently parsing so the next 'Stream #'
235 # audio line is routed to the correct slot (input vs output).
236 if line.startswith("Input #"):
237 self._current_log_section = "input"
238 elif line.startswith("Output #"):
239 self._current_log_section = "output"
240
241 # Capture the first audio stream line per section. Provider-supplied input
242 # details are often incomplete (e.g. defaults to 44.1/16) or missing for
243 # lossy codecs, so we mirror the parsed input values onto input_format too.
244 if self._current_log_section == "input" and self.input_stream_info is None:
245 if stream_info := parse_ffmpeg_stream_info(line):
246 self.input_stream_info = stream_info
247 self._log_stream_info("input", stream_info)
248 self._apply_input_stream_info(stream_info)
249 elif self._current_log_section == "output" and self.output_stream_info is None:
250 if stream_info := parse_ffmpeg_stream_info(line):
251 self.output_stream_info = stream_info
252 self._log_stream_info("output", stream_info)
253
254 # Source duration is reported separately from the stream info. Useful when
255 # the provider didn't supply one (some podcast feeds report total_time=0).
256 if self.parsed_duration is None:
257 duration = parse_ffmpeg_duration(line)
258 if duration is not None:
259 self.parsed_duration = duration
260 self.logger.debug("Detected input duration: %s seconds", duration)
261 del line
262
263 async def _feed_stdin(self) -> None:
264 """Feed stdin with audio chunks from an AsyncGenerator."""
265 assert not isinstance(self.audio_input, str | int)
266 generator_exhausted = False
267 cancelled = False
268 status = "running"
269 chunk_count = 0
270 self.logger.log(VERBOSE_LOG_LEVEL, "Start reading audio data from source...")
271 try:
272 start = time.time()
273 while True:
274 try:
275 chunk = await anext(self.audio_input)
276 except StopAsyncIteration:
277 generator_exhausted = True
278 break
279 except Exception as err:
280 self._stdin_feeder_exception = err
281 raise
282 chunk_count += 1
283 if self.closed:
284 return
285 await self.write(chunk)
286 except asyncio.CancelledError:
287 status = "cancelled"
288 raise
289 except Exception:
290 status = "aborted with error"
291 raise
292 finally:
293 LOGGER.log(
294 VERBOSE_LOG_LEVEL,
295 "fill_buffer_task: %s (%s chunks received) in in %.2fs",
296 status,
297 chunk_count,
298 time.time() - start,
299 )
300 if not cancelled:
301 await self.write_eof()
302 # we need to ensure that we close the async generator
303 # if we get cancelled otherwise it keeps lingering forever
304 if not generator_exhausted:
305 await close_async_generator(self.audio_input)
306
307 def _apply_input_stream_info(self, info: FFMpegStreamInfo) -> None:
308 """Mirror values from a parsed ffmpeg input stream line onto self.input_format."""
309 # content_type is the container format; only fill it in if the provider didn't
310 # specify one. codec_type is the audio codec ffmpeg detected; only override
311 # if we actually parsed a known codec (don't clobber a provider value with UNKNOWN).
312 if info.codec != ContentType.UNKNOWN:
313 if self.input_format.content_type == ContentType.UNKNOWN:
314 self.input_format.content_type = info.codec
315 self.input_format.codec_type = info.codec
316 if info.sample_rate:
317 self.input_format.sample_rate = info.sample_rate
318 if info.bit_depth:
319 self.input_format.bit_depth = info.bit_depth
320 if info.bit_rate:
321 self.input_format.bit_rate = info.bit_rate
322
323 def _log_stream_info(self, label: str, info: FFMpegStreamInfo) -> None:
324 """Log a parsed FFMpegStreamInfo object at debug level."""
325 self.logger.debug(
326 "Detected %s stream info: codec=%s sample_rate=%s bit_depth=%s bit_rate=%s kb/s",
327 label,
328 info.codec,
329 info.sample_rate,
330 info.bit_depth,
331 info.bit_rate,
332 )
333
334
335def parse_ffmpeg_stream_info(line: str) -> FFMpegStreamInfo | None:
336 """
337 Extract audio format details from an ffmpeg 'Stream #X: Audio: ...' log line.
338
339 :param line: A single ffmpeg stderr log line.
340 :returns: FFMpegStreamInfo when the line describes an audio stream,
341 otherwise None.
342 """
343 if not (line.startswith("Stream #") and ": Audio: " in line):
344 return None
345
346 # the codec name is the first token right after "Audio: ", stripping
347 # any trailing profile annotation like "(LC)" or container suffix
348 codec_part = line.split(": Audio: ", 1)[1].split(" ", 1)[0].split(",", maxsplit=1)[0]
349 codec = ContentType.try_parse(codec_part)
350
351 info = FFMpegStreamInfo(codec=codec)
352 if match := _FFMPEG_SAMPLE_RATE_RE.search(line):
353 info.sample_rate = int(match.group(1))
354 if match := _FFMPEG_BIT_RATE_RE.search(line):
355 info.bit_rate = int(match.group(1))
356 # Bit depth: an explicit "(N bit)" annotation wins (this is how ffmpeg reports
357 # 24-bit FLAC stored in an s32 sample format), otherwise infer from the sample
358 # format token. Lossy codecs report the decoder's internal precision (typically
359 # fltp), so we ignore the sample format token for those.
360 if match := _FFMPEG_EXPLICIT_BIT_DEPTH_RE.search(line):
361 info.bit_depth = int(match.group(1))
362 elif codec.is_lossless() and (match := _FFMPEG_SAMPLE_FMT_RE.search(line)):
363 info.bit_depth = _SAMPLE_FMT_BIT_DEPTH.get(match.group(1))
364
365 return info
366
367
368def parse_ffmpeg_duration(line: str) -> int | None:
369 """
370 Extract the source duration in seconds from an ffmpeg 'Duration: ...' log line.
371
372 :param line: A single ffmpeg stderr log line.
373 :returns: Duration in whole seconds, or None if the line does not contain
374 a parseable duration (e.g. 'Duration: N/A' on live streams).
375 """
376 match = _FFMPEG_DURATION_RE.search(line)
377 if not match:
378 return None
379 hours, minutes, seconds = match.groups()
380 return int(hours) * 3600 + int(minutes) * 60 + int(float(seconds))
381
382
383async def get_ffmpeg_stream(
384 audio_input: AsyncGenerator[bytes] | str,
385 input_format: AudioFormat,
386 output_format: AudioFormat,
387 filter_params: Sequence[str | ComplexFilter] | None = None,
388 chunk_size: int | None = None,
389 extra_input_args: list[str] | None = None,
390 extra_output_args: list[str] | None = None,
391) -> AsyncGenerator[bytes]:
392 """
393 Get the ffmpeg audio stream as async generator.
394
395 Takes care of resampling and/or recoding if needed,
396 according to player preferences.
397 """
398 async with FFMpeg(
399 audio_input=audio_input,
400 input_format=input_format,
401 output_format=output_format,
402 filter_params=filter_params,
403 extra_input_args=extra_input_args,
404 extra_output_args=extra_output_args,
405 collect_log_history=True,
406 ) as ffmpeg_proc:
407 # read final chunks from stdout
408 iterator = ffmpeg_proc.iter_chunked(chunk_size) if chunk_size else ffmpeg_proc.iter_any()
409 async for chunk in iterator:
410 yield chunk
411 # reap the process before trusting returncode: a stream aborted mid-decode (e.g.
412 # excessive decode errors) closes stdout early, which ends the loop above before
413 # the OS process has actually exited, leaving returncode as None if checked directly
414 with suppress(TimeoutError):
415 await ffmpeg_proc.wait_with_timeout(5)
416 if ffmpeg_proc.returncode not in (None, 0) or ffmpeg_proc.concat_error:
417 # unclean exit of ffmpeg - raise error with log tail
418 log_lines = -20 if ffmpeg_proc.concat_error else -5
419 log_tail = "\n" + "\n".join(list(ffmpeg_proc.log_history)[log_lines:])
420 raise AudioError(log_tail)
421 if feeder_exception := ffmpeg_proc.stdin_feeder_exception:
422 raise AudioError("Error while feeding audio to FFmpeg") from feeder_exception
423
424
425async def get_ffmpeg_overlay_stream(
426 audio_input: AsyncGenerator[bytes],
427 overlay_input: str,
428 pcm_format: AudioFormat,
429 overlay_volume: int = 100,
430 chunk_size: int | None = None,
431) -> AsyncGenerator[bytes]:
432 """
433 Mix a looping audio overlay into a PCM audio stream.
434
435 The overlay is looped for the full duration of the main stream and the mixed
436 output has the exact same PCM format and duration as the main input. For a stereo
437 output, a mono overlay mixes in at the same level as an equivalent stereo one. If
438 the overlay input fails mid-stream, the main audio continues unaffected.
439
440 :param audio_input: The main audio stream (raw PCM in ``pcm_format``).
441 :param overlay_input: File path or URL of the overlay audio.
442 :param overlay_volume: Overlay loudness relative to the main audio in
443 percent (100 = equally loud, max 200).
444 :param pcm_format: PCM format of both the main input and the mixed output.
445 :param chunk_size: Optional exact chunk size for the yielded audio.
446 """
447 async with FFMpeg(
448 audio_input=audio_input,
449 # ffmpeg mirrors the metadata it probes from the input onto input_format,
450 # so hand it a copy to keep that mutation off the caller's format.
451 input_format=copy(pcm_format),
452 output_format=pcm_format,
453 filter_params=[_build_overlay_mixer(overlay_input, pcm_format, overlay_volume)],
454 collect_log_history=True,
455 ) as ffmpeg_proc:
456 iterator = ffmpeg_proc.iter_chunked(chunk_size) if chunk_size else ffmpeg_proc.iter_any()
457 async for chunk in iterator:
458 yield chunk
459 # reap the process before trusting returncode: a stream aborted mid-decode (e.g.
460 # excessive decode errors) closes stdout early, which ends the loop above before
461 # the OS process has actually exited, leaving returncode as None if checked directly
462 with suppress(TimeoutError):
463 await ffmpeg_proc.wait_with_timeout(5)
464 if ffmpeg_proc.returncode not in (None, 0):
465 # unclean exit of ffmpeg - raise error with log tail
466 log_tail = "\n" + "\n".join(list(ffmpeg_proc.log_history)[-5:])
467 raise AudioError(log_tail)
468 if feeder_exception := ffmpeg_proc.stdin_feeder_exception:
469 raise AudioError("Error while feeding audio to FFmpeg") from feeder_exception
470
471
472def get_ffmpeg_resample_filter(
473 input_format: AudioFormat,
474 output_format: AudioFormat,
475 filter_params: Sequence[str | ComplexFilter],
476) -> str | None:
477 """
478 Return the resampling and dithering filter required for a format conversion.
479
480 :param input_format: Format entering FFmpeg.
481 :param output_format: Requested FFmpeg output format.
482 :param filter_params: Filters that run before resampling.
483 """
484 if input_format.sample_rate == output_format.sample_rate and not (
485 input_format.bit_depth > 16 and output_format.bit_depth == 16
486 ):
487 return None
488 libsoxr_support = get_global_cache_value(CACHE_ATTR_LIBSOXR_PRESENT)
489 # loudnorm and libsoxr cannot be combined due to https://trac.ffmpeg.org/ticket/11323
490 if libsoxr_support and not any(
491 "loudnorm" in value for value in filter_params if isinstance(value, str)
492 ):
493 resample_filter = "aresample=resampler=soxr:precision=30"
494 else:
495 resample_filter = "aresample=resampler=swr"
496 if input_format.sample_rate != output_format.sample_rate:
497 resample_filter += f":osr={output_format.sample_rate}"
498 if output_format.bit_depth == 16 and input_format.bit_depth > 16:
499 resample_filter += ":osf=s16:dither_method=triangular_hp"
500 return resample_filter
501
502
503def get_ffmpeg_args(
504 input_format: AudioFormat,
505 output_format: AudioFormat,
506 filter_params: Sequence[str | ComplexFilter],
507 input_path: str = "-",
508 output_path: str = "-",
509 extra_input_args: list[str] | None = None,
510 extra_output_args: list[str] | None = None,
511 loglevel: str = "error",
512) -> list[str]:
513 """Collect all args to send to the ffmpeg process."""
514 filter_params = list(filter_params)
515 if extra_input_args is None:
516 extra_input_args = []
517 if extra_output_args is None:
518 extra_output_args = []
519 # the binary plus the options that apply to the command as a whole
520 global_args = [
521 "ffmpeg",
522 "-hide_banner",
523 "-loglevel",
524 loglevel,
525 "-nostats",
526 "-ignore_unknown",
527 ]
528 # collect args for the main input, mirroring how _build_filtergraph_args opens the
529 # extra inputs: the read args lead the group so the caller can still override them
530 input_args = [*_INPUT_READ_ARGS, *extra_input_args]
531 if "-f" not in extra_input_args:
532 # without an input format of their own, the caller leaves the input spec to us
533 if input_path.startswith("http"):
534 # append reconnect options for direct stream from http
535 input_args += [
536 # Reconnect automatically when disconnected before EOF is hit.
537 "-reconnect",
538 "1",
539 # Set the maximum delay in seconds after which to give up reconnecting.
540 "-reconnect_delay_max",
541 "10",
542 # If set then even streamed/non seekable streams will be reconnected on errors.
543 "-reconnect_streamed",
544 "1",
545 # Reconnect automatically in case of TCP/TLS errors during connect.
546 "-reconnect_on_network_error",
547 "0",
548 # A comma separated list of HTTP status codes to reconnect on.
549 # The list can include specific status codes (e.g. 503) or the strings 4xx / 5xx.
550 "-reconnect_on_http_error",
551 "5xx,429",
552 ]
553 if "-post_data" in extra_input_args:
554 # ffmpeg does not include Range headers on POST reconnects, so byte-range
555 # seeking via reconnect is not available. Mark the stream non-seekable so
556 # demuxers do not attempt end-of-file probes (e.g. OGG duration detection)
557 # that would trigger Range-less restarts from byte 0. MA-initiated seeks
558 # still work via -ss decode-and-discard.
559 input_args += ["-seekable", "0"]
560 if input_format.content_type.is_pcm():
561 input_args += [
562 *get_ffmpeg_channel_args(input_format),
563 "-ar",
564 str(input_format.sample_rate),
565 "-acodec",
566 input_format.content_type.name.lower(),
567 "-f",
568 input_format.content_type.value,
569 ]
570 if input_format.codec_type != ContentType.UNKNOWN:
571 input_args += ["-acodec", input_format.codec_type.name.lower()]
572
573 # add input path at the end
574 input_args += ["-i", input_path]
575
576 # collect output args
577 output_args = get_ffmpeg_channel_args(output_format)
578 if output_path.upper() == "NULL":
579 # devnull stream: nothing is encoded here, so there is no channel count to declare
580 output_path = "-"
581 output_args = ["-f", "null"]
582 elif output_format.content_type.is_pcm():
583 # use explicit format identifier for pcm formats
584 output_args += [
585 "-ar",
586 str(output_format.sample_rate),
587 "-acodec",
588 output_format.content_type.name.lower(),
589 "-f",
590 output_format.content_type.value,
591 ]
592 elif output_format.content_type == ContentType.NUT:
593 # passthrough-mode (for creating the cache) using NUT container.
594 # -acodec copy leaves the source untouched, so there is no channel count to declare
595 output_args = [
596 "-vn",
597 "-dn",
598 "-sn",
599 "-acodec",
600 "copy",
601 "-f",
602 "nut",
603 ]
604 elif output_format.content_type == ContentType.AAC:
605 output_args += ["-f", "adts", "-c:a", "aac", "-b:a", "256k"]
606 elif output_format.content_type == ContentType.MP3:
607 output_args += ["-f", "mp3", "-b:a", f"{DEFAULT_MP3_BIT_RATE}k"]
608 elif output_format.content_type == ContentType.WAV:
609 pcm_format = ContentType.from_bit_depth(output_format.bit_depth)
610 output_args += [
611 "-ar",
612 str(output_format.sample_rate),
613 "-acodec",
614 pcm_format.name.lower(),
615 "-f",
616 "wav",
617 ]
618 elif output_format.content_type == ContentType.FLAC:
619 # use level 0 compression for fastest encoding
620 sample_fmt = "s32" if output_format.bit_depth > 16 else "s16"
621 output_args += [
622 "-sample_fmt",
623 sample_fmt,
624 "-ar",
625 str(output_format.sample_rate),
626 "-f",
627 "flac",
628 "-compression_level",
629 "0",
630 ]
631 else:
632 raise RuntimeError("Invalid/unsupported output format specified")
633
634 output_args += extra_output_args # append the extra output args
635 # append (final) output path at the end of the args
636 output_args.append(output_path)
637
638 # runs ahead of the caller's own filters, so channel-aware ones such as the
639 # per-channel preamp see the conformed layout instead of the source layout
640 if channel_filter := _get_channel_conform_filter(input_format.channels, output_format.channels):
641 filter_params = [channel_filter, *filter_params]
642
643 if resample_filter := get_ffmpeg_resample_filter(
644 input_format,
645 output_format,
646 filter_params,
647 ):
648 filter_params.append(resample_filter)
649
650 # a complex fragment brings its own inputs, which must follow the main input
651 filter_input_args, filter_args = (
652 _build_filtergraph_args(filter_params) if filter_params else ([], [])
653 )
654
655 return global_args + input_args + filter_input_args + filter_args + output_args
656
657
658def get_ffmpeg_channel_args(audio_format: AudioFormat) -> list[str]:
659 """
660 Return the FFmpeg channel count/layout arguments for the given audio format.
661
662 The layout is only named for channel counts that map to exactly one layout.
663
664 :param audio_format: Format to describe.
665 """
666 args = ["-ac", str(audio_format.channels)]
667 if layout := _get_channel_layout_name(audio_format.channels):
668 args += ["-channel_layout", layout]
669 return args
670
671
672def get_ffmpeg_hls_cmaf_input_args() -> list[str]:
673 """
674 Return HLS demuxer input arguments that let CMAF segments through, if any are needed.
675
676 Pass these only for a playlist from a source known to serve CMAF, never for a playlist
677 URL that a user supplied.
678 """
679 # The check this relaxes is hardening against hostile playlists, hence opt-in per caller.
680 # allowed_extensions cannot narrow it: the demuxer matches a segment URL against that
681 # option *and* against a hardcoded per-format extension list that no option reaches, so
682 # switching the check off is the only lever over the second one.
683 if get_global_cache_value(CACHE_ATTR_HLS_CMAF_BLOCKED):
684 return ["-extension_picky", "0"]
685 return []
686
687
688async def check_ffmpeg_version() -> None:
689 """Check that ffmpeg is present and usable, and cache the capabilities it reports."""
690 # check for FFmpeg presence
691 try:
692 returncode, output = await check_output("ffmpeg", "-version")
693 except FileNotFoundError:
694 raise AudioError(
695 "FFmpeg binary is missing from system. "
696 "Please install ffmpeg on your OS to enable playback."
697 )
698 if returncode != 0:
699 err_msg = "Error determining FFmpeg version on your system."
700 if returncode < 0:
701 # error below 0 is often illegal instruction
702 err_msg += " - Your CPU may be too old to run this version of FFmpeg."
703 err_msg += f" - Additional info: {returncode} {output.decode().strip()}"
704 raise AudioError(err_msg)
705 # parse version number from output
706 try:
707 version = output.decode().split("ffmpeg version ")[1].split(" ")[0].split("-")[0]
708 except IndexError:
709 raise AudioError(
710 "Error determining FFmpeg version on your system."
711 f"Additional info: {returncode} {output.decode().strip()}"
712 )
713 libsoxr_support = "enable-libsoxr" in output.decode()
714 # 7.1.1 backported a segment extension check without whitelisting CMAF, so it rejects the
715 # .cmfa segments some services serve; 7.1.2 whitelisted them, see
716 # https://trac.ffmpeg.org/ticket/11526. Probe the demuxer rather than compare versions,
717 # which builds from git report as e.g. "N-121037-g1234567". A probe that fails reads as
718 # "not blocked", so the check stays in place. Drop this once every supported build
719 # whitelists CMAF.
720 returncode, hls_options = await check_output("ffmpeg", "-hide_banner", "-h", "demuxer=hls")
721 cmaf_blocked = (
722 returncode == 0 and b"extension_picky" in hls_options and b"cmfa" not in hls_options
723 )
724 # use globals as in-memory cache
725 await set_global_cache_values(
726 {
727 CACHE_ATTR_LIBSOXR_PRESENT: libsoxr_support,
728 CACHE_ATTR_FFMPEG_VERSION: version,
729 CACHE_ATTR_HLS_CMAF_BLOCKED: cmaf_blocked,
730 }
731 )
732
733 major_version = int("".join(char for char in version.split(".")[0] if not char.isalpha()))
734 if major_version < MINIMAL_FFMPEG_VERSION:
735 raise AudioError(
736 f"FFmpeg version {version} is not supported. "
737 f"Minimal version required is {MINIMAL_FFMPEG_VERSION}."
738 )
739
740 LOGGER.info(
741 "Detected ffmpeg version %s %s",
742 version,
743 "with libsoxr support" if libsoxr_support else "",
744 )
745
746
747def _get_channel_layout_name(channels: int) -> str | None:
748 """
749 Return FFmpeg's layout name for a channel count, or None when it has no unambiguous one.
750
751 :param channels: Number of channels to name.
752 """
753 if channels == 1:
754 return "mono"
755 if channels == 2:
756 return "stereo"
757 # a wider count maps to several possible layouts (5.1 vs 5.1(side), 7.1 vs 7.1(wide), ...)
758 # and a named layout wins over -ac, so naming the wrong one would make FFmpeg misread the
759 # stream as that layout. Left unnamed, it derives the default for the count itself.
760 return None
761
762
763def _get_channel_conform_filter(input_channels: int, output_channels: int) -> str | None:
764 """
765 Return the filter that maps the source onto the output channel count, if one is needed.
766
767 :param input_channels: Channel count entering FFmpeg.
768 :param output_channels: Channel count the output is encoded at.
769 :return: The filter to run before any caller supplied ones, or None when the
770 source already carries the requested channel count.
771 """
772 if input_channels > 2 and output_channels <= 2:
773 # a single channel output needs this fold too, otherwise a mono/left/right pan
774 # would only see the front channels and silently drop the center and surround.
775 # aformat leaves the rematrix to ffmpeg, which picks the correct coefficients
776 # for whatever layout the input turns out to have (and, for an integer output,
777 # scales them to stay clip-safe). A fixed pan expression, naming channels that
778 # a given layout may not even have, can do neither.
779 return "aformat=channel_layouts=stereo"
780 if input_channels == 1 and output_channels > 1:
781 # duplicate rather than leaving the widening to ffmpeg, whose rematrix
782 # spreads the source at 1/sqrt(2) per channel and so costs 3 dB
783 return "pan=stereo|c0=c0|c1=c0"
784 return None
785
786
787def _get_overlay_volume_filter(overlay_volume: int, output_channels: int) -> str:
788 """
789 Return the filter that scales an overlay source to the requested loudness.
790
791 :param overlay_volume: Requested overlay loudness in percent.
792 :param output_channels: Channel count of the mixed output.
793 """
794 gain = overlay_volume / 100
795 if output_channels != 2:
796 # a mono source widened to more than two channels is routed to the centre at full
797 # level, so only a stereo output loses any. No overlay call site is non-stereo today.
798 return f"volume={gain}"
799 # nb_channels is evaluated where this filter sits, ahead of any layout conversion, so it
800 # still reports the source's own count: only a mono source is scaled up, leaving a stereo
801 # one and its image untouched. Comma-free, as a comma would end this filter in the graph.
802 return f"volume={gain}*{_MONO_WIDEN_COMPENSATION}^not(nb_channels-1)"
803
804
805def _build_overlay_mixer(
806 overlay_input: str, pcm_format: AudioFormat, overlay_volume: int
807) -> ComplexFilter:
808 """
809 Build the filter that mixes a looping audio overlay into the main audio.
810
811 :param overlay_input: File path or URL of the overlay audio.
812 :param pcm_format: PCM format of the main input and the mixed output.
813 :param overlay_volume: Overlay loudness relative to the main audio in percent.
814 """
815 input_args = []
816 if overlay_input.startswith("http"):
817 input_args += [
818 "-reconnect",
819 "1",
820 "-reconnect_delay_max",
821 "10",
822 "-reconnect_streamed",
823 "1",
824 ]
825 input_args += ["-stream_loop", "-1"]
826 # conform the overlay to the main stream's layout so amix sees two matching inputs;
827 # an unnameable count is left to FFmpeg's own negotiation
828 layout = _get_channel_layout_name(pcm_format.channels)
829 conform_filter = f",aformat=channel_layouts={layout}" if layout else ""
830 return ComplexFilter(
831 # the main audio is amix's first input, so duration=first follows its length;
832 # normalize=0 keeps the original levels (no averaging)
833 body="amix=inputs=2:duration=first:normalize=0",
834 inputs=[
835 ComplexFilterInput(
836 path=overlay_input,
837 # silenceremove strips a near-silent intro from the overlay source (e.g. a
838 # soft fade-in) so it becomes audible right away; it is a no-op for sources
839 # that already start at full level. It runs before volume so detection is
840 # based on the source's own levels rather than the scaled output. volume
841 # in turn has to stay ahead of the resample and conform steps, which
842 # replace the source's own channel count with the output's.
843 filters=(
844 f"silenceremove=start_periods=1:start_threshold=-40dB,"
845 f"{_get_overlay_volume_filter(overlay_volume, pcm_format.channels)},"
846 f"aresample={pcm_format.sample_rate}"
847 f"{conform_filter}"
848 ),
849 input_args=input_args,
850 )
851 ],
852 )
853
854
855def _build_filtergraph_args(
856 filter_params: list[str | ComplexFilter],
857) -> tuple[list[str], list[str]]:
858 """
859 Render a DSP filter chain to FFmpeg command-line arguments.
860
861 :param filter_params: Ordered chain of plain filter strings and/or complex
862 fragments that need extra audio inputs.
863 :return: Extra input arguments to append after the main input, and the
864 filter arguments themselves.
865 """
866 if not any(isinstance(item, ComplexFilter) for item in filter_params):
867 simple = [item for item in filter_params if isinstance(item, str) and item]
868 return [], (["-af", ",".join(simple)] if simple else [])
869
870 input_args: list[str] = []
871 parts: list[str] = []
872 pending: list[str] = []
873 current = "0:a"
874 counter = 0
875 # the main input is 0, so extra inputs are numbered from 1 in the order added
876 next_input = 1
877
878 def next_label() -> str:
879 nonlocal counter
880 counter += 1
881 return f"dsp{counter}"
882
883 def flush_pending() -> None:
884 nonlocal current
885 if not pending:
886 return
887 label = next_label()
888 parts.append(f"[{current}]{','.join(pending)}[{label}]")
889 current = label
890 pending.clear()
891
892 for item in filter_params:
893 if isinstance(item, str):
894 if item:
895 pending.append(item)
896 continue
897 # a complex fragment closes the current simple run, adds its own inputs to
898 # the command, then consumes the main pad plus those inputs
899 flush_pending()
900 source_labels: list[str] = []
901 for extra_input in item.inputs:
902 input_args += [*_INPUT_READ_ARGS, *extra_input.input_args, "-i", extra_input.path]
903 source = f"{next_input}:a"
904 next_input += 1
905 if extra_input.filters:
906 label = next_label()
907 parts.append(f"[{source}]{extra_input.filters}[{label}]")
908 source = label
909 source_labels.append(source)
910 label = next_label()
911 inputs = f"[{current}]" + "".join(f"[{sl}]" for sl in source_labels)
912 parts.append(f"{inputs}{item.body}[{label}]")
913 current = label
914 flush_pending()
915
916 return input_args, ["-filter_complex", ";".join(parts), "-map", f"[{current}]"]
917