/
/
/
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 elif input_format.codec_type != ContentType.UNKNOWN:
571 # ffmpeg honours the last -acodec it is given, so this must not follow the
572 # raw PCM decoder declared above
573 input_args += ["-acodec", input_format.codec_type.name.lower()]
574
575 # add input path at the end
576 input_args += ["-i", input_path]
577
578 # collect output args
579 output_args = get_ffmpeg_channel_args(output_format)
580 if output_path.upper() == "NULL":
581 # devnull stream: nothing is encoded here, so there is no channel count to declare
582 output_path = "-"
583 output_args = ["-f", "null"]
584 elif output_format.content_type.is_pcm():
585 # use explicit format identifier for pcm formats
586 output_args += [
587 "-ar",
588 str(output_format.sample_rate),
589 "-acodec",
590 output_format.content_type.name.lower(),
591 "-f",
592 output_format.content_type.value,
593 ]
594 elif output_format.content_type == ContentType.NUT:
595 # passthrough-mode (for creating the cache) using NUT container.
596 # -acodec copy leaves the source untouched, so there is no channel count to declare
597 output_args = [
598 "-vn",
599 "-dn",
600 "-sn",
601 "-acodec",
602 "copy",
603 "-f",
604 "nut",
605 ]
606 elif output_format.content_type == ContentType.AAC:
607 output_args += ["-f", "adts", "-c:a", "aac", "-b:a", "256k"]
608 elif output_format.content_type == ContentType.MP3:
609 output_args += ["-f", "mp3", "-b:a", f"{DEFAULT_MP3_BIT_RATE}k"]
610 elif output_format.content_type == ContentType.WAV:
611 pcm_format = ContentType.from_bit_depth(output_format.bit_depth)
612 output_args += [
613 "-ar",
614 str(output_format.sample_rate),
615 "-acodec",
616 pcm_format.name.lower(),
617 "-f",
618 "wav",
619 ]
620 elif output_format.content_type == ContentType.FLAC:
621 # use level 0 compression for fastest encoding
622 sample_fmt = "s32" if output_format.bit_depth > 16 else "s16"
623 output_args += [
624 "-sample_fmt",
625 sample_fmt,
626 "-ar",
627 str(output_format.sample_rate),
628 "-f",
629 "flac",
630 "-compression_level",
631 "0",
632 ]
633 else:
634 raise RuntimeError("Invalid/unsupported output format specified")
635
636 output_args += extra_output_args # append the extra output args
637 # append (final) output path at the end of the args
638 output_args.append(output_path)
639
640 # runs ahead of the caller's own filters, so channel-aware ones such as the
641 # per-channel preamp see the conformed layout instead of the source layout
642 if channel_filter := _get_channel_conform_filter(input_format.channels, output_format.channels):
643 filter_params = [channel_filter, *filter_params]
644
645 if resample_filter := get_ffmpeg_resample_filter(
646 input_format,
647 output_format,
648 filter_params,
649 ):
650 filter_params.append(resample_filter)
651
652 # a complex fragment brings its own inputs, which must follow the main input
653 filter_input_args, filter_args = (
654 _build_filtergraph_args(filter_params) if filter_params else ([], [])
655 )
656
657 return global_args + input_args + filter_input_args + filter_args + output_args
658
659
660def get_ffmpeg_channel_args(audio_format: AudioFormat) -> list[str]:
661 """
662 Return the FFmpeg channel count/layout arguments for the given audio format.
663
664 The layout is only named for channel counts that map to exactly one layout.
665
666 :param audio_format: Format to describe.
667 """
668 args = ["-ac", str(audio_format.channels)]
669 if layout := _get_channel_layout_name(audio_format.channels):
670 args += ["-channel_layout", layout]
671 return args
672
673
674def get_ffmpeg_hls_cmaf_input_args() -> list[str]:
675 """
676 Return HLS demuxer input arguments that let CMAF segments through, if any are needed.
677
678 Pass these only for a playlist from a source known to serve CMAF, never for a playlist
679 URL that a user supplied.
680 """
681 # The check this relaxes is hardening against hostile playlists, hence opt-in per caller.
682 # allowed_extensions cannot narrow it: the demuxer matches a segment URL against that
683 # option *and* against a hardcoded per-format extension list that no option reaches, so
684 # switching the check off is the only lever over the second one.
685 if get_global_cache_value(CACHE_ATTR_HLS_CMAF_BLOCKED):
686 return ["-extension_picky", "0"]
687 return []
688
689
690async def check_ffmpeg_version() -> None:
691 """Check that ffmpeg is present and usable, and cache the capabilities it reports."""
692 # check for FFmpeg presence
693 try:
694 returncode, output = await check_output("ffmpeg", "-version")
695 except FileNotFoundError:
696 raise AudioError(
697 "FFmpeg binary is missing from system. "
698 "Please install ffmpeg on your OS to enable playback."
699 )
700 if returncode != 0:
701 err_msg = "Error determining FFmpeg version on your system."
702 if returncode < 0:
703 # error below 0 is often illegal instruction
704 err_msg += " - Your CPU may be too old to run this version of FFmpeg."
705 err_msg += f" - Additional info: {returncode} {output.decode().strip()}"
706 raise AudioError(err_msg)
707 # parse version number from output
708 try:
709 version = output.decode().split("ffmpeg version ")[1].split(" ")[0].split("-")[0]
710 except IndexError:
711 raise AudioError(
712 "Error determining FFmpeg version on your system."
713 f"Additional info: {returncode} {output.decode().strip()}"
714 )
715 libsoxr_support = "enable-libsoxr" in output.decode()
716 # 7.1.1 backported a segment extension check without whitelisting CMAF, so it rejects the
717 # .cmfa segments some services serve; 7.1.2 whitelisted them, see
718 # https://trac.ffmpeg.org/ticket/11526. Probe the demuxer rather than compare versions,
719 # which builds from git report as e.g. "N-121037-g1234567". A probe that fails reads as
720 # "not blocked", so the check stays in place. Drop this once every supported build
721 # whitelists CMAF.
722 returncode, hls_options = await check_output("ffmpeg", "-hide_banner", "-h", "demuxer=hls")
723 cmaf_blocked = (
724 returncode == 0 and b"extension_picky" in hls_options and b"cmfa" not in hls_options
725 )
726 # use globals as in-memory cache
727 await set_global_cache_values(
728 {
729 CACHE_ATTR_LIBSOXR_PRESENT: libsoxr_support,
730 CACHE_ATTR_FFMPEG_VERSION: version,
731 CACHE_ATTR_HLS_CMAF_BLOCKED: cmaf_blocked,
732 }
733 )
734
735 major_version = int("".join(char for char in version.split(".")[0] if not char.isalpha()))
736 if major_version < MINIMAL_FFMPEG_VERSION:
737 raise AudioError(
738 f"FFmpeg version {version} is not supported. "
739 f"Minimal version required is {MINIMAL_FFMPEG_VERSION}."
740 )
741
742 LOGGER.info(
743 "Detected ffmpeg version %s %s",
744 version,
745 "with libsoxr support" if libsoxr_support else "",
746 )
747
748
749def _get_channel_layout_name(channels: int) -> str | None:
750 """
751 Return FFmpeg's layout name for a channel count, or None when it has no unambiguous one.
752
753 :param channels: Number of channels to name.
754 """
755 if channels == 1:
756 return "mono"
757 if channels == 2:
758 return "stereo"
759 # a wider count maps to several possible layouts (5.1 vs 5.1(side), 7.1 vs 7.1(wide), ...)
760 # and a named layout wins over -ac, so naming the wrong one would make FFmpeg misread the
761 # stream as that layout. Left unnamed, it derives the default for the count itself.
762 return None
763
764
765def _get_channel_conform_filter(input_channels: int, output_channels: int) -> str | None:
766 """
767 Return the filter that maps the source onto the output channel count, if one is needed.
768
769 :param input_channels: Channel count entering FFmpeg.
770 :param output_channels: Channel count the output is encoded at.
771 :return: The filter to run before any caller supplied ones, or None when the
772 source already carries the requested channel count.
773 """
774 if input_channels > 2 and output_channels <= 2:
775 # a single channel output needs this fold too, otherwise a mono/left/right pan
776 # would only see the front channels and silently drop the center and surround.
777 # aformat leaves the rematrix to ffmpeg, which picks the correct coefficients
778 # for whatever layout the input turns out to have (and, for an integer output,
779 # scales them to stay clip-safe). A fixed pan expression, naming channels that
780 # a given layout may not even have, can do neither.
781 return "aformat=channel_layouts=stereo"
782 if input_channels == 1 and output_channels > 1:
783 # duplicate rather than leaving the widening to ffmpeg, whose rematrix
784 # spreads the source at 1/sqrt(2) per channel and so costs 3 dB
785 return "pan=stereo|c0=c0|c1=c0"
786 return None
787
788
789def _get_overlay_volume_filter(overlay_volume: int, output_channels: int) -> str:
790 """
791 Return the filter that scales an overlay source to the requested loudness.
792
793 :param overlay_volume: Requested overlay loudness in percent.
794 :param output_channels: Channel count of the mixed output.
795 """
796 gain = overlay_volume / 100
797 if output_channels != 2:
798 # a mono source widened to more than two channels is routed to the centre at full
799 # level, so only a stereo output loses any. No overlay call site is non-stereo today.
800 return f"volume={gain}"
801 # nb_channels is evaluated where this filter sits, ahead of any layout conversion, so it
802 # still reports the source's own count: only a mono source is scaled up, leaving a stereo
803 # one and its image untouched. Comma-free, as a comma would end this filter in the graph.
804 return f"volume={gain}*{_MONO_WIDEN_COMPENSATION}^not(nb_channels-1)"
805
806
807def _build_overlay_mixer(
808 overlay_input: str, pcm_format: AudioFormat, overlay_volume: int
809) -> ComplexFilter:
810 """
811 Build the filter that mixes a looping audio overlay into the main audio.
812
813 :param overlay_input: File path or URL of the overlay audio.
814 :param pcm_format: PCM format of the main input and the mixed output.
815 :param overlay_volume: Overlay loudness relative to the main audio in percent.
816 """
817 input_args = []
818 if overlay_input.startswith("http"):
819 input_args += [
820 "-reconnect",
821 "1",
822 "-reconnect_delay_max",
823 "10",
824 "-reconnect_streamed",
825 "1",
826 ]
827 input_args += ["-stream_loop", "-1"]
828 # conform the overlay to the main stream's layout so amix sees two matching inputs;
829 # an unnameable count is left to FFmpeg's own negotiation
830 layout = _get_channel_layout_name(pcm_format.channels)
831 conform_filter = f",aformat=channel_layouts={layout}" if layout else ""
832 return ComplexFilter(
833 # the main audio is amix's first input, so duration=first follows its length;
834 # normalize=0 keeps the original levels (no averaging)
835 body="amix=inputs=2:duration=first:normalize=0",
836 inputs=[
837 ComplexFilterInput(
838 path=overlay_input,
839 # silenceremove strips a near-silent intro from the overlay source (e.g. a
840 # soft fade-in) so it becomes audible right away; it is a no-op for sources
841 # that already start at full level. It runs before volume so detection is
842 # based on the source's own levels rather than the scaled output. volume
843 # in turn has to stay ahead of the resample and conform steps, which
844 # replace the source's own channel count with the output's.
845 filters=(
846 f"silenceremove=start_periods=1:start_threshold=-40dB,"
847 f"{_get_overlay_volume_filter(overlay_volume, pcm_format.channels)},"
848 f"aresample={pcm_format.sample_rate}"
849 f"{conform_filter}"
850 ),
851 input_args=input_args,
852 )
853 ],
854 )
855
856
857def _build_filtergraph_args(
858 filter_params: list[str | ComplexFilter],
859) -> tuple[list[str], list[str]]:
860 """
861 Render a DSP filter chain to FFmpeg command-line arguments.
862
863 :param filter_params: Ordered chain of plain filter strings and/or complex
864 fragments that need extra audio inputs.
865 :return: Extra input arguments to append after the main input, and the
866 filter arguments themselves.
867 """
868 if not any(isinstance(item, ComplexFilter) for item in filter_params):
869 simple = [item for item in filter_params if isinstance(item, str) and item]
870 return [], (["-af", ",".join(simple)] if simple else [])
871
872 input_args: list[str] = []
873 parts: list[str] = []
874 pending: list[str] = []
875 current = "0:a"
876 counter = 0
877 # the main input is 0, so extra inputs are numbered from 1 in the order added
878 next_input = 1
879
880 def next_label() -> str:
881 nonlocal counter
882 counter += 1
883 return f"dsp{counter}"
884
885 def flush_pending() -> None:
886 nonlocal current
887 if not pending:
888 return
889 label = next_label()
890 parts.append(f"[{current}]{','.join(pending)}[{label}]")
891 current = label
892 pending.clear()
893
894 for item in filter_params:
895 if isinstance(item, str):
896 if item:
897 pending.append(item)
898 continue
899 # a complex fragment closes the current simple run, adds its own inputs to
900 # the command, then consumes the main pad plus those inputs
901 flush_pending()
902 source_labels: list[str] = []
903 for extra_input in item.inputs:
904 input_args += [*_INPUT_READ_ARGS, *extra_input.input_args, "-i", extra_input.path]
905 source = f"{next_input}:a"
906 next_input += 1
907 if extra_input.filters:
908 label = next_label()
909 parts.append(f"[{source}]{extra_input.filters}[{label}]")
910 source = label
911 source_labels.append(source)
912 label = next_label()
913 inputs = f"[{current}]" + "".join(f"[{sl}]" for sl in source_labels)
914 parts.append(f"{inputs}{item.body}[{label}]")
915 current = label
916 flush_pending()
917
918 return input_args, ["-filter_complex", ";".join(parts), "-map", f"[{current}]"]
919