music-assistant-server
39.5 KB•PY
ffmpeg.py
39.5 KB • 929 lines • python
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 def _is_expected_task_error(self, err: BaseException) -> bool:
335 """Return whether a helper task error is an expected outcome rather than a failure."""
336 # deferred import: the provider models pull the controller graph in at
337 # import time, which this low-level helper must stay clear of
338 from music_assistant.models.music_provider import ProviderStreamLimitError # noqa: PLC0415
339
340 # a provider with no free source-stream slot is a normal outcome of a
341 # speculative prefetch: the caller retries once the current stream releases it
342 return isinstance(err, ProviderStreamLimitError)
343
344
345def parse_ffmpeg_stream_info(line: str) -> FFMpegStreamInfo | None:
346 """
347 Extract audio format details from an ffmpeg 'Stream #X: Audio: ...' log line.
348
349 :param line: A single ffmpeg stderr log line.
350 :returns: FFMpegStreamInfo when the line describes an audio stream,
351 otherwise None.
352 """
353 if not (line.startswith("Stream #") and ": Audio: " in line):
354 return None
355
356 # the codec name is the first token right after "Audio: ", stripping
357 # any trailing profile annotation like "(LC)" or container suffix
358 codec_part = line.split(": Audio: ", 1)[1].split(" ", 1)[0].split(",", maxsplit=1)[0]
359 codec = ContentType.try_parse(codec_part)
360
361 info = FFMpegStreamInfo(codec=codec)
362 if match := _FFMPEG_SAMPLE_RATE_RE.search(line):
363 info.sample_rate = int(match.group(1))
364 if match := _FFMPEG_BIT_RATE_RE.search(line):
365 info.bit_rate = int(match.group(1))
366 # Bit depth: an explicit "(N bit)" annotation wins (this is how ffmpeg reports
367 # 24-bit FLAC stored in an s32 sample format), otherwise infer from the sample
368 # format token. Lossy codecs report the decoder's internal precision (typically
369 # fltp), so we ignore the sample format token for those.
370 if match := _FFMPEG_EXPLICIT_BIT_DEPTH_RE.search(line):
371 info.bit_depth = int(match.group(1))
372 elif codec.is_lossless() and (match := _FFMPEG_SAMPLE_FMT_RE.search(line)):
373 info.bit_depth = _SAMPLE_FMT_BIT_DEPTH.get(match.group(1))
374
375 return info
376
377
378def parse_ffmpeg_duration(line: str) -> int | None:
379 """
380 Extract the source duration in seconds from an ffmpeg 'Duration: ...' log line.
381
382 :param line: A single ffmpeg stderr log line.
383 :returns: Duration in whole seconds, or None if the line does not contain
384 a parseable duration (e.g. 'Duration: N/A' on live streams).
385 """
386 match = _FFMPEG_DURATION_RE.search(line)
387 if not match:
388 return None
389 hours, minutes, seconds = match.groups()
390 return int(hours) * 3600 + int(minutes) * 60 + int(float(seconds))
391
392
393async def get_ffmpeg_stream(
394 audio_input: AsyncGenerator[bytes] | str,
395 input_format: AudioFormat,
396 output_format: AudioFormat,
397 filter_params: Sequence[str | ComplexFilter] | None = None,
398 chunk_size: int | None = None,
399 extra_input_args: list[str] | None = None,
400 extra_output_args: list[str] | None = None,
401) -> AsyncGenerator[bytes]:
402 """
403 Get the ffmpeg audio stream as async generator.
404
405 Takes care of resampling and/or recoding if needed,
406 according to player preferences.
407 """
408 async with FFMpeg(
409 audio_input=audio_input,
410 input_format=input_format,
411 output_format=output_format,
412 filter_params=filter_params,
413 extra_input_args=extra_input_args,
414 extra_output_args=extra_output_args,
415 collect_log_history=True,
416 ) as ffmpeg_proc:
417 # read final chunks from stdout
418 iterator = ffmpeg_proc.iter_chunked(chunk_size) if chunk_size else ffmpeg_proc.iter_any()
419 async for chunk in iterator:
420 yield chunk
421 # reap the process before trusting returncode: a stream aborted mid-decode (e.g.
422 # excessive decode errors) closes stdout early, which ends the loop above before
423 # the OS process has actually exited, leaving returncode as None if checked directly
424 with suppress(TimeoutError):
425 await ffmpeg_proc.wait_with_timeout(5)
426 if ffmpeg_proc.returncode not in (None, 0) or ffmpeg_proc.concat_error:
427 # unclean exit of ffmpeg - raise error with log tail
428 log_lines = -20 if ffmpeg_proc.concat_error else -5
429 log_tail = "\n" + "\n".join(list(ffmpeg_proc.log_history)[log_lines:])
430 raise AudioError(log_tail)
431 if feeder_exception := ffmpeg_proc.stdin_feeder_exception:
432 raise AudioError("Error while feeding audio to FFmpeg") from feeder_exception
433
434
435async def get_ffmpeg_overlay_stream(
436 audio_input: AsyncGenerator[bytes],
437 overlay_input: str,
438 pcm_format: AudioFormat,
439 overlay_volume: int = 100,
440 chunk_size: int | None = None,
441) -> AsyncGenerator[bytes]:
442 """
443 Mix a looping audio overlay into a PCM audio stream.
444
445 The overlay is looped for the full duration of the main stream and the mixed
446 output has the exact same PCM format and duration as the main input. For a stereo
447 output, a mono overlay mixes in at the same level as an equivalent stereo one. If
448 the overlay input fails mid-stream, the main audio continues unaffected.
449
450 :param audio_input: The main audio stream (raw PCM in ``pcm_format``).
451 :param overlay_input: File path or URL of the overlay audio.
452 :param overlay_volume: Overlay loudness relative to the main audio in
453 percent (100 = equally loud, max 200).
454 :param pcm_format: PCM format of both the main input and the mixed output.
455 :param chunk_size: Optional exact chunk size for the yielded audio.
456 """
457 async with FFMpeg(
458 audio_input=audio_input,
459 # ffmpeg mirrors the metadata it probes from the input onto input_format,
460 # so hand it a copy to keep that mutation off the caller's format.
461 input_format=copy(pcm_format),
462 output_format=pcm_format,
463 filter_params=[_build_overlay_mixer(overlay_input, pcm_format, overlay_volume)],
464 collect_log_history=True,
465 ) as ffmpeg_proc:
466 iterator = ffmpeg_proc.iter_chunked(chunk_size) if chunk_size else ffmpeg_proc.iter_any()
467 async for chunk in iterator:
468 yield chunk
469 # reap the process before trusting returncode: a stream aborted mid-decode (e.g.
470 # excessive decode errors) closes stdout early, which ends the loop above before
471 # the OS process has actually exited, leaving returncode as None if checked directly
472 with suppress(TimeoutError):
473 await ffmpeg_proc.wait_with_timeout(5)
474 if ffmpeg_proc.returncode not in (None, 0):
475 # unclean exit of ffmpeg - raise error with log tail
476 log_tail = "\n" + "\n".join(list(ffmpeg_proc.log_history)[-5:])
477 raise AudioError(log_tail)
478 if feeder_exception := ffmpeg_proc.stdin_feeder_exception:
479 raise AudioError("Error while feeding audio to FFmpeg") from feeder_exception
480
481
482def get_ffmpeg_resample_filter(
483 input_format: AudioFormat,
484 output_format: AudioFormat,
485 filter_params: Sequence[str | ComplexFilter],
486) -> str | None:
487 """
488 Return the resampling and dithering filter required for a format conversion.
489
490 :param input_format: Format entering FFmpeg.
491 :param output_format: Requested FFmpeg output format.
492 :param filter_params: Filters that run before resampling.
493 """
494 if input_format.sample_rate == output_format.sample_rate and not (
495 input_format.bit_depth > 16 and output_format.bit_depth == 16
496 ):
497 return None
498 libsoxr_support = get_global_cache_value(CACHE_ATTR_LIBSOXR_PRESENT)
499 # loudnorm and libsoxr cannot be combined due to https://trac.ffmpeg.org/ticket/11323
500 if libsoxr_support and not any(
501 "loudnorm" in value for value in filter_params if isinstance(value, str)
502 ):
503 resample_filter = "aresample=resampler=soxr:precision=30"
504 else:
505 resample_filter = "aresample=resampler=swr"
506 if input_format.sample_rate != output_format.sample_rate:
507 resample_filter += f":osr={output_format.sample_rate}"
508 if output_format.bit_depth == 16 and input_format.bit_depth > 16:
509 resample_filter += ":osf=s16:dither_method=triangular_hp"
510 return resample_filter
511
512
513def get_ffmpeg_args(
514 input_format: AudioFormat,
515 output_format: AudioFormat,
516 filter_params: Sequence[str | ComplexFilter],
517 input_path: str = "-",
518 output_path: str = "-",
519 extra_input_args: list[str] | None = None,
520 extra_output_args: list[str] | None = None,
521 loglevel: str = "error",
522) -> list[str]:
523 """Collect all args to send to the ffmpeg process."""
524 filter_params = list(filter_params)
525 if extra_input_args is None:
526 extra_input_args = []
527 if extra_output_args is None:
528 extra_output_args = []
529 # the binary plus the options that apply to the command as a whole
530 global_args = [
531 "ffmpeg",
532 "-hide_banner",
533 "-loglevel",
534 loglevel,
535 "-nostats",
536 "-ignore_unknown",
537 ]
538 # collect args for the main input, mirroring how _build_filtergraph_args opens the
539 # extra inputs: the read args lead the group so the caller can still override them
540 input_args = [*_INPUT_READ_ARGS, *extra_input_args]
541 if "-f" not in extra_input_args:
542 # without an input format of their own, the caller leaves the input spec to us
543 if input_path.startswith("http"):
544 # append reconnect options for direct stream from http
545 input_args += [
546 # Reconnect automatically when disconnected before EOF is hit.
547 "-reconnect",
548 "1",
549 # Set the maximum delay in seconds after which to give up reconnecting.
550 "-reconnect_delay_max",
551 "10",
552 # If set then even streamed/non seekable streams will be reconnected on errors.
553 "-reconnect_streamed",
554 "1",
555 # Reconnect automatically in case of TCP/TLS errors during connect.
556 "-reconnect_on_network_error",
557 "0",
558 # A comma separated list of HTTP status codes to reconnect on.
559 # The list can include specific status codes (e.g. 503) or the strings 4xx / 5xx.
560 "-reconnect_on_http_error",
561 "5xx,429",
562 ]
563 if "-post_data" in extra_input_args:
564 # ffmpeg does not include Range headers on POST reconnects, so byte-range
565 # seeking via reconnect is not available. Mark the stream non-seekable so
566 # demuxers do not attempt end-of-file probes (e.g. OGG duration detection)
567 # that would trigger Range-less restarts from byte 0. MA-initiated seeks
568 # still work via -ss decode-and-discard.
569 input_args += ["-seekable", "0"]
570 if input_format.content_type.is_pcm():
571 input_args += [
572 *get_ffmpeg_channel_args(input_format),
573 "-ar",
574 str(input_format.sample_rate),
575 "-acodec",
576 input_format.content_type.name.lower(),
577 "-f",
578 input_format.content_type.value,
579 ]
580 elif input_format.codec_type != ContentType.UNKNOWN:
581 # ffmpeg honours the last -acodec it is given, so this must not follow the
582 # raw PCM decoder declared above
583 input_args += ["-acodec", input_format.codec_type.name.lower()]
584
585 # add input path at the end
586 input_args += ["-i", input_path]
587
588 # collect output args
589 output_args = get_ffmpeg_channel_args(output_format)
590 if output_path.upper() == "NULL":
591 # devnull stream: nothing is encoded here, so there is no channel count to declare
592 output_path = "-"
593 output_args = ["-f", "null"]
594 elif output_format.content_type.is_pcm():
595 # use explicit format identifier for pcm formats
596 output_args += [
597 "-ar",
598 str(output_format.sample_rate),
599 "-acodec",
600 output_format.content_type.name.lower(),
601 "-f",
602 output_format.content_type.value,
603 ]
604 elif output_format.content_type == ContentType.NUT:
605 # passthrough-mode (for creating the cache) using NUT container.
606 # -acodec copy leaves the source untouched, so there is no channel count to declare
607 output_args = [
608 "-vn",
609 "-dn",
610 "-sn",
611 "-acodec",
612 "copy",
613 "-f",
614 "nut",
615 ]
616 elif output_format.content_type == ContentType.AAC:
617 output_args += ["-f", "adts", "-c:a", "aac", "-b:a", "256k"]
618 elif output_format.content_type == ContentType.MP3:
619 output_args += ["-f", "mp3", "-b:a", f"{DEFAULT_MP3_BIT_RATE}k"]
620 elif output_format.content_type == ContentType.WAV:
621 pcm_format = ContentType.from_bit_depth(output_format.bit_depth)
622 output_args += [
623 "-ar",
624 str(output_format.sample_rate),
625 "-acodec",
626 pcm_format.name.lower(),
627 "-f",
628 "wav",
629 ]
630 elif output_format.content_type == ContentType.FLAC:
631 # use level 0 compression for fastest encoding
632 sample_fmt = "s32" if output_format.bit_depth > 16 else "s16"
633 output_args += [
634 "-sample_fmt",
635 sample_fmt,
636 "-ar",
637 str(output_format.sample_rate),
638 "-f",
639 "flac",
640 "-compression_level",
641 "0",
642 ]
643 else:
644 raise RuntimeError("Invalid/unsupported output format specified")
645
646 output_args += extra_output_args # append the extra output args
647 # append (final) output path at the end of the args
648 output_args.append(output_path)
649
650 # runs ahead of the caller's own filters, so channel-aware ones such as the
651 # per-channel preamp see the conformed layout instead of the source layout
652 if channel_filter := _get_channel_conform_filter(input_format.channels, output_format.channels):
653 filter_params = [channel_filter, *filter_params]
654
655 if resample_filter := get_ffmpeg_resample_filter(
656 input_format,
657 output_format,
658 filter_params,
659 ):
660 filter_params.append(resample_filter)
661
662 # a complex fragment brings its own inputs, which must follow the main input
663 filter_input_args, filter_args = (
664 _build_filtergraph_args(filter_params) if filter_params else ([], [])
665 )
666
667 return global_args + input_args + filter_input_args + filter_args + output_args
668
669
670def get_ffmpeg_channel_args(audio_format: AudioFormat) -> list[str]:
671 """
672 Return the FFmpeg channel count/layout arguments for the given audio format.
673
674 The layout is only named for channel counts that map to exactly one layout.
675
676 :param audio_format: Format to describe.
677 """
678 args = ["-ac", str(audio_format.channels)]
679 if layout := _get_channel_layout_name(audio_format.channels):
680 args += ["-channel_layout", layout]
681 return args
682
683
684def get_ffmpeg_hls_cmaf_input_args() -> list[str]:
685 """
686 Return HLS demuxer input arguments that let CMAF segments through, if any are needed.
687
688 Pass these only for a playlist from a source known to serve CMAF, never for a playlist
689 URL that a user supplied.
690 """
691 # The check this relaxes is hardening against hostile playlists, hence opt-in per caller.
692 # allowed_extensions cannot narrow it: the demuxer matches a segment URL against that
693 # option *and* against a hardcoded per-format extension list that no option reaches, so
694 # switching the check off is the only lever over the second one.
695 if get_global_cache_value(CACHE_ATTR_HLS_CMAF_BLOCKED):
696 return ["-extension_picky", "0"]
697 return []
698
699
700async def check_ffmpeg_version() -> None:
701 """Check that ffmpeg is present and usable, and cache the capabilities it reports."""
702 # check for FFmpeg presence
703 try:
704 returncode, output = await check_output("ffmpeg", "-version")
705 except FileNotFoundError:
706 raise AudioError(
707 "FFmpeg binary is missing from system. "
708 "Please install ffmpeg on your OS to enable playback."
709 )
710 if returncode != 0:
711 err_msg = "Error determining FFmpeg version on your system."
712 if returncode < 0:
713 # error below 0 is often illegal instruction
714 err_msg += " - Your CPU may be too old to run this version of FFmpeg."
715 err_msg += f" - Additional info: {returncode} {output.decode().strip()}"
716 raise AudioError(err_msg)
717 # parse version number from output
718 try:
719 version = output.decode().split("ffmpeg version ")[1].split(" ")[0].split("-")[0]
720 except IndexError:
721 raise AudioError(
722 "Error determining FFmpeg version on your system."
723 f"Additional info: {returncode} {output.decode().strip()}"
724 )
725 libsoxr_support = "enable-libsoxr" in output.decode()
726 # 7.1.1 backported a segment extension check without whitelisting CMAF, so it rejects the
727 # .cmfa segments some services serve; 7.1.2 whitelisted them, see
728 # https://trac.ffmpeg.org/ticket/11526. Probe the demuxer rather than compare versions,
729 # which builds from git report as e.g. "N-121037-g1234567". A probe that fails reads as
730 # "not blocked", so the check stays in place. Drop this once every supported build
731 # whitelists CMAF.
732 returncode, hls_options = await check_output("ffmpeg", "-hide_banner", "-h", "demuxer=hls")
733 cmaf_blocked = (
734 returncode == 0 and b"extension_picky" in hls_options and b"cmfa" not in hls_options
735 )
736 # use globals as in-memory cache
737 await set_global_cache_values(
738 {
739 CACHE_ATTR_LIBSOXR_PRESENT: libsoxr_support,
740 CACHE_ATTR_FFMPEG_VERSION: version,
741 CACHE_ATTR_HLS_CMAF_BLOCKED: cmaf_blocked,
742 }
743 )
744
745 major_version = int("".join(char for char in version.split(".")[0] if not char.isalpha()))
746 if major_version < MINIMAL_FFMPEG_VERSION:
747 raise AudioError(
748 f"FFmpeg version {version} is not supported. "
749 f"Minimal version required is {MINIMAL_FFMPEG_VERSION}."
750 )
751
752 LOGGER.info(
753 "Detected ffmpeg version %s %s",
754 version,
755 "with libsoxr support" if libsoxr_support else "",
756 )
757
758
759def _get_channel_layout_name(channels: int) -> str | None:
760 """
761 Return FFmpeg's layout name for a channel count, or None when it has no unambiguous one.
762
763 :param channels: Number of channels to name.
764 """
765 if channels == 1:
766 return "mono"
767 if channels == 2:
768 return "stereo"
769 # a wider count maps to several possible layouts (5.1 vs 5.1(side), 7.1 vs 7.1(wide), ...)
770 # and a named layout wins over -ac, so naming the wrong one would make FFmpeg misread the
771 # stream as that layout. Left unnamed, it derives the default for the count itself.
772 return None
773
774
775def _get_channel_conform_filter(input_channels: int, output_channels: int) -> str | None:
776 """
777 Return the filter that maps the source onto the output channel count, if one is needed.
778
779 :param input_channels: Channel count entering FFmpeg.
780 :param output_channels: Channel count the output is encoded at.
781 :return: The filter to run before any caller supplied ones, or None when the
782 source already carries the requested channel count.
783 """
784 if input_channels > 2 and output_channels <= 2:
785 # a single channel output needs this fold too, otherwise a mono/left/right pan
786 # would only see the front channels and silently drop the center and surround.
787 # aformat leaves the rematrix to ffmpeg, which picks the correct coefficients
788 # for whatever layout the input turns out to have (and, for an integer output,
789 # scales them to stay clip-safe). A fixed pan expression, naming channels that
790 # a given layout may not even have, can do neither.
791 return "aformat=channel_layouts=stereo"
792 if input_channels == 1 and output_channels > 1:
793 # duplicate rather than leaving the widening to ffmpeg, whose rematrix
794 # spreads the source at 1/sqrt(2) per channel and so costs 3 dB
795 return "pan=stereo|c0=c0|c1=c0"
796 return None
797
798
799def _get_overlay_volume_filter(overlay_volume: int, output_channels: int) -> str:
800 """
801 Return the filter that scales an overlay source to the requested loudness.
802
803 :param overlay_volume: Requested overlay loudness in percent.
804 :param output_channels: Channel count of the mixed output.
805 """
806 gain = overlay_volume / 100
807 if output_channels != 2:
808 # a mono source widened to more than two channels is routed to the centre at full
809 # level, so only a stereo output loses any. No overlay call site is non-stereo today.
810 return f"volume={gain}"
811 # nb_channels is evaluated where this filter sits, ahead of any layout conversion, so it
812 # still reports the source's own count: only a mono source is scaled up, leaving a stereo
813 # one and its image untouched. Comma-free, as a comma would end this filter in the graph.
814 return f"volume={gain}*{_MONO_WIDEN_COMPENSATION}^not(nb_channels-1)"
815
816
817def _build_overlay_mixer(
818 overlay_input: str, pcm_format: AudioFormat, overlay_volume: int
819) -> ComplexFilter:
820 """
821 Build the filter that mixes a looping audio overlay into the main audio.
822
823 :param overlay_input: File path or URL of the overlay audio.
824 :param pcm_format: PCM format of the main input and the mixed output.
825 :param overlay_volume: Overlay loudness relative to the main audio in percent.
826 """
827 input_args = []
828 if overlay_input.startswith("http"):
829 input_args += [
830 "-reconnect",
831 "1",
832 "-reconnect_delay_max",
833 "10",
834 "-reconnect_streamed",
835 "1",
836 ]
837 input_args += ["-stream_loop", "-1"]
838 # conform the overlay to the main stream's layout so amix sees two matching inputs;
839 # an unnameable count is left to FFmpeg's own negotiation
840 layout = _get_channel_layout_name(pcm_format.channels)
841 conform_filter = f",aformat=channel_layouts={layout}" if layout else ""
842 return ComplexFilter(
843 # the main audio is amix's first input, so duration=first follows its length;
844 # normalize=0 keeps the original levels (no averaging)
845 body="amix=inputs=2:duration=first:normalize=0",
846 inputs=[
847 ComplexFilterInput(
848 path=overlay_input,
849 # silenceremove strips a near-silent intro from the overlay source (e.g. a
850 # soft fade-in) so it becomes audible right away; it is a no-op for sources
851 # that already start at full level. It runs before volume so detection is
852 # based on the source's own levels rather than the scaled output. volume
853 # in turn has to stay ahead of the resample and conform steps, which
854 # replace the source's own channel count with the output's.
855 filters=(
856 f"silenceremove=start_periods=1:start_threshold=-40dB,"
857 f"{_get_overlay_volume_filter(overlay_volume, pcm_format.channels)},"
858 f"aresample={pcm_format.sample_rate}"
859 f"{conform_filter}"
860 ),
861 input_args=input_args,
862 )
863 ],
864 )
865
866
867def _build_filtergraph_args(
868 filter_params: list[str | ComplexFilter],
869) -> tuple[list[str], list[str]]:
870 """
871 Render a DSP filter chain to FFmpeg command-line arguments.
872
873 :param filter_params: Ordered chain of plain filter strings and/or complex
874 fragments that need extra audio inputs.
875 :return: Extra input arguments to append after the main input, and the
876 filter arguments themselves.
877 """
878 if not any(isinstance(item, ComplexFilter) for item in filter_params):
879 simple = [item for item in filter_params if isinstance(item, str) and item]
880 return [], (["-af", ",".join(simple)] if simple else [])
881
882 input_args: list[str] = []
883 parts: list[str] = []
884 pending: list[str] = []
885 current = "0:a"
886 counter = 0
887 # the main input is 0, so extra inputs are numbered from 1 in the order added
888 next_input = 1
889
890 def next_label() -> str:
891 nonlocal counter
892 counter += 1
893 return f"dsp{counter}"
894
895 def flush_pending() -> None:
896 nonlocal current
897 if not pending:
898 return
899 label = next_label()
900 parts.append(f"[{current}]{','.join(pending)}[{label}]")
901 current = label
902 pending.clear()
903
904 for item in filter_params:
905 if isinstance(item, str):
906 if item:
907 pending.append(item)
908 continue
909 # a complex fragment closes the current simple run, adds its own inputs to
910 # the command, then consumes the main pad plus those inputs
911 flush_pending()
912 source_labels: list[str] = []
913 for extra_input in item.inputs:
914 input_args += [*_INPUT_READ_ARGS, *extra_input.input_args, "-i", extra_input.path]
915 source = f"{next_input}:a"
916 next_input += 1
917 if extra_input.filters:
918 label = next_label()
919 parts.append(f"[{source}]{extra_input.filters}[{label}]")
920 source = label
921 source_labels.append(source)
922 label = next_label()
923 inputs = f"[{current}]" + "".join(f"[{sl}]" for sl in source_labels)
924 parts.append(f"{inputs}{item.body}[{label}]")
925 current = label
926 flush_pending()
927
928 return input_args, ["-filter_complex", ";".join(parts), "-map", f"[{current}]"]
929