/
/
/
1"""FFMpeg related helpers."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import time
8from collections import deque
9from collections.abc import AsyncGenerator
10from contextlib import suppress
11from typing import TYPE_CHECKING, Final
12
13from music_assistant_models.enums import ContentType
14from music_assistant_models.errors import AudioError
15from music_assistant_models.helpers import get_global_cache_value, set_global_cache_values
16
17from music_assistant.constants import VERBOSE_LOG_LEVEL
18
19from .process import AsyncProcess, check_output
20from .util import close_async_generator
21
22if TYPE_CHECKING:
23 from music_assistant_models.media_items import AudioFormat
24
25LOGGER = logging.getLogger("ffmpeg")
26MINIMAL_FFMPEG_VERSION = 6
27CACHE_ATTR_LIBSOXR_PRESENT: Final[str] = "libsoxr_present"
28
29
30class FFMpeg(AsyncProcess):
31 """FFMpeg wrapped as AsyncProcess."""
32
33 def __init__(
34 self,
35 audio_input: AsyncGenerator[bytes, None] | str | int,
36 input_format: AudioFormat,
37 output_format: AudioFormat,
38 filter_params: list[str] | None = None,
39 extra_args: list[str] | None = None,
40 extra_input_args: list[str] | None = None,
41 extra_output_args: list[str] | None = None,
42 audio_output: str | int = "-",
43 collect_log_history: bool = False,
44 loglevel: str = "info",
45 ) -> None:
46 """Initialize AsyncProcess."""
47 ffmpeg_args = get_ffmpeg_args(
48 input_format=input_format,
49 output_format=output_format,
50 filter_params=filter_params or [],
51 extra_args=extra_args or [],
52 input_path=audio_input if isinstance(audio_input, str) else "-",
53 output_path=audio_output if isinstance(audio_output, str) else "-",
54 extra_input_args=extra_input_args or [],
55 extra_output_args=extra_output_args or [],
56 loglevel=loglevel,
57 )
58 self.audio_input = audio_input
59 self.input_format = input_format
60 self.collect_log_history = collect_log_history
61 self.log_history: deque[str] = deque(maxlen=100)
62 self._stdin_feeder_task: asyncio.Task[None] | None = None
63 self._stderr_reader_task: asyncio.Task[None] | None = None
64 self._input_codec_parsed = False
65 stdin: bool | int
66 if audio_input == "-" or isinstance(audio_input, AsyncGenerator):
67 stdin = True
68 else:
69 stdin = audio_input if isinstance(audio_input, int) else False
70 stdout = audio_output if isinstance(audio_output, int) else bool(audio_output == "-")
71 super().__init__(
72 ffmpeg_args,
73 stdin=stdin,
74 stdout=stdout,
75 stderr=True,
76 )
77 self.logger = LOGGER
78
79 async def start(self) -> None:
80 """Perform Async init of process."""
81 await super().start()
82 if self.proc:
83 self.logger = LOGGER.getChild(str(self.proc.pid))
84 clean_args = []
85 for arg in self._args[1:]:
86 if arg.startswith("http"):
87 clean_args.append("<URL>")
88 elif "/" in arg and "." in arg:
89 clean_args.append("<FILE>")
90 elif arg.startswith("data:application/"):
91 clean_args.append("<DATA>")
92 else:
93 clean_args.append(arg)
94 args_str = " ".join(clean_args)
95 self.logger.log(VERBOSE_LOG_LEVEL, "started with args: %s", args_str)
96 self._stderr_reader_task = asyncio.create_task(self._log_reader_task())
97 if isinstance(self.audio_input, AsyncGenerator):
98 self._stdin_feeder_task = asyncio.create_task(self._feed_stdin())
99
100 async def communicate(
101 self,
102 input: bytes | None = None, # noqa: A002
103 timeout: float | None = None,
104 ) -> tuple[bytes, bytes]:
105 """Override communicate to avoid blocking."""
106 if self._stdin_feeder_task:
107 if not self._stdin_feeder_task.done():
108 self._stdin_feeder_task.cancel()
109 # Always await the task to consume any exception and prevent
110 # "Task exception was never retrieved" errors.
111 # Suppress CancelledError (from cancel) and any other exception
112 # since exceptions have already been propagated through the generator chain.
113 with suppress(asyncio.CancelledError, Exception):
114 await self._stdin_feeder_task
115 if self._stderr_reader_task:
116 if not self._stderr_reader_task.done():
117 self._stderr_reader_task.cancel()
118 with suppress(asyncio.CancelledError, Exception):
119 await self._stderr_reader_task
120 return await super().communicate(input, timeout)
121
122 async def _log_reader_task(self) -> None:
123 """Read ffmpeg log from stderr."""
124 decode_errors = 0
125 async for line in self.iter_stderr():
126 if self.collect_log_history:
127 self.log_history.append(line)
128 # ffmpeg logging can be quite verbose, so we only log critical errors
129 # unless verbose logging is enabled
130 if "critical" in line:
131 self.logger.error(line)
132 elif self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
133 self.logger.log(VERBOSE_LOG_LEVEL, line)
134
135 if "Invalid data found when processing input" in line:
136 decode_errors += 1
137 if decode_errors >= 50:
138 self.logger.error(line)
139
140 # if streamdetails contenttype is unknown, try parse it from the ffmpeg log
141 if line.startswith("Stream #") and ": Audio: " in line:
142 if not self._input_codec_parsed:
143 content_type_raw = line.split(": Audio: ")[1].split(" ")[0]
144 content_type_raw = content_type_raw.split(",")[0]
145 content_type = ContentType.try_parse(content_type_raw)
146 self.logger.log(
147 VERBOSE_LOG_LEVEL,
148 "Detected (input) content type: %s (%s)",
149 content_type,
150 content_type_raw,
151 )
152 if self.input_format.content_type == ContentType.UNKNOWN:
153 self.input_format.content_type = content_type
154 self.input_format.codec_type = content_type
155 self._input_codec_parsed = True
156 del line
157
158 async def _feed_stdin(self) -> None:
159 """Feed stdin with audio chunks from an AsyncGenerator."""
160 assert not isinstance(self.audio_input, str | int)
161 generator_exhausted = False
162 cancelled = False
163 status = "running"
164 chunk_count = 0
165 self.logger.log(VERBOSE_LOG_LEVEL, "Start reading audio data from source...")
166 try:
167 start = time.time()
168 async for chunk in self.audio_input:
169 chunk_count += 1
170 if self.closed:
171 return
172 await self.write(chunk)
173 generator_exhausted = True
174 except asyncio.CancelledError:
175 status = "cancelled"
176 raise
177 except Exception:
178 status = "aborted with error"
179 raise
180 finally:
181 LOGGER.log(
182 VERBOSE_LOG_LEVEL,
183 "fill_buffer_task: %s (%s chunks received) in in %.2fs",
184 status,
185 chunk_count,
186 time.time() - start,
187 )
188 if not cancelled:
189 await self.write_eof()
190 # we need to ensure that we close the async generator
191 # if we get cancelled otherwise it keeps lingering forever
192 if not generator_exhausted:
193 await close_async_generator(self.audio_input)
194
195
196async def get_ffmpeg_stream(
197 audio_input: AsyncGenerator[bytes, None] | str,
198 input_format: AudioFormat,
199 output_format: AudioFormat,
200 filter_params: list[str] | None = None,
201 extra_args: list[str] | None = None,
202 chunk_size: int | None = None,
203 extra_input_args: list[str] | None = None,
204 extra_output_args: list[str] | None = None,
205) -> AsyncGenerator[bytes, None]:
206 """
207 Get the ffmpeg audio stream as async generator.
208
209 Takes care of resampling and/or recoding if needed,
210 according to player preferences.
211 """
212 async with FFMpeg(
213 audio_input=audio_input,
214 input_format=input_format,
215 output_format=output_format,
216 filter_params=filter_params,
217 extra_args=extra_args,
218 extra_input_args=extra_input_args,
219 extra_output_args=extra_output_args,
220 collect_log_history=True,
221 ) as ffmpeg_proc:
222 # read final chunks from stdout
223 iterator = ffmpeg_proc.iter_chunked(chunk_size) if chunk_size else ffmpeg_proc.iter_any()
224 async for chunk in iterator:
225 yield chunk
226 if ffmpeg_proc.returncode not in (None, 0):
227 # unclean exit of ffmpeg - raise error with log tail
228 log_tail = "\n" + "\n".join(list(ffmpeg_proc.log_history)[-5:])
229 raise AudioError(log_tail)
230
231
232def get_ffmpeg_args( # noqa: PLR0915
233 input_format: AudioFormat,
234 output_format: AudioFormat,
235 filter_params: list[str],
236 extra_args: list[str] | None = None,
237 input_path: str = "-",
238 output_path: str = "-",
239 extra_input_args: list[str] | None = None,
240 extra_output_args: list[str] | None = None,
241 loglevel: str = "error",
242) -> list[str]:
243 """Collect all args to send to the ffmpeg process."""
244 if extra_args is None:
245 extra_args = []
246 if extra_input_args is None:
247 extra_input_args = []
248 if extra_output_args is None:
249 extra_output_args = []
250 # generic args
251 generic_args = [
252 "ffmpeg",
253 "-hide_banner",
254 "-loglevel",
255 loglevel,
256 "-nostats",
257 "-ignore_unknown",
258 "-protocol_whitelist",
259 "file,hls,http,https,tcp,tls,crypto,pipe,data,fd,rtp,udp,concat",
260 "-probesize",
261 "8096",
262 "-analyzeduration",
263 "500000", # 0.5 seconds should be enough to detect the format
264 ]
265 # collect input args
266 if "-f" in extra_input_args:
267 # input format is already specified in the extra input args
268 input_args = extra_input_args
269 else:
270 input_args = [*extra_input_args]
271 if input_path.startswith("http"):
272 # append reconnect options for direct stream from http
273 input_args += [
274 # Reconnect automatically when disconnected before EOF is hit.
275 "-reconnect",
276 "1",
277 # Set the maximum delay in seconds after which to give up reconnecting.
278 "-reconnect_delay_max",
279 "10",
280 # If set then even streamed/non seekable streams will be reconnected on errors.
281 "-reconnect_streamed",
282 "1",
283 # Reconnect automatically in case of TCP/TLS errors during connect.
284 "-reconnect_on_network_error",
285 "0",
286 # A comma separated list of HTTP status codes to reconnect on.
287 # The list can include specific status codes (e.g. 503) or the strings 4xx / 5xx.
288 "-reconnect_on_http_error",
289 "5xx,429",
290 ]
291 if input_format.content_type.is_pcm():
292 input_args += [
293 "-ac",
294 str(input_format.channels),
295 "-channel_layout",
296 "mono" if input_format.channels == 1 else "stereo",
297 "-ar",
298 str(input_format.sample_rate),
299 "-acodec",
300 input_format.content_type.name.lower(),
301 "-f",
302 input_format.content_type.value,
303 ]
304 if input_format.codec_type != ContentType.UNKNOWN:
305 input_args += ["-acodec", input_format.codec_type.name.lower()]
306
307 # add input path at the end
308 input_args += ["-i", input_path]
309
310 # collect output args
311 output_args = [
312 "-ac",
313 str(output_format.channels),
314 "-channel_layout",
315 "mono" if output_format.channels == 1 else "stereo",
316 ]
317 if output_path.upper() == "NULL":
318 # devnull stream
319 output_path = "-"
320 output_args = ["-f", "null"]
321 elif output_format.content_type.is_pcm():
322 # use explicit format identifier for pcm formats
323 output_args += [
324 "-ar",
325 str(output_format.sample_rate),
326 "-acodec",
327 output_format.content_type.name.lower(),
328 "-f",
329 output_format.content_type.value,
330 ]
331 elif output_format.content_type == ContentType.NUT:
332 # passthrough-mode (for creating the cache) using NUT container
333 output_args = [
334 "-vn",
335 "-dn",
336 "-sn",
337 "-acodec",
338 "copy",
339 "-f",
340 "nut",
341 ]
342 elif output_format.content_type == ContentType.AAC:
343 output_args = ["-f", "adts", "-c:a", "aac", "-b:a", "256k"]
344 elif output_format.content_type == ContentType.MP3:
345 output_args = ["-f", "mp3", "-b:a", "320k"]
346 elif output_format.content_type == ContentType.WAV:
347 pcm_format = ContentType.from_bit_depth(output_format.bit_depth)
348 output_args = [
349 "-ar",
350 str(output_format.sample_rate),
351 "-acodec",
352 pcm_format.name.lower(),
353 "-f",
354 "wav",
355 ]
356 elif output_format.content_type == ContentType.FLAC:
357 # use level 0 compression for fastest encoding
358 sample_fmt = "s32" if output_format.bit_depth > 16 else "s16"
359 output_args += [
360 "-sample_fmt",
361 sample_fmt,
362 "-ar",
363 str(output_format.sample_rate),
364 "-f",
365 "flac",
366 "-compression_level",
367 "0",
368 ]
369 else:
370 raise RuntimeError("Invalid/unsupported output format specified")
371
372 output_args += extra_output_args # append the extra output args
373 # append (final) output path at the end of the args
374 output_args.append(output_path)
375
376 # edge case: source file is not stereo - downmix to stereo
377 if input_format.channels > 2 and output_format.channels == 2:
378 filter_params = [
379 "pan=stereo|FL=1.0*FL+0.707*FC+0.707*SL+0.707*LFE|FR=1.0*FR+0.707*FC+0.707*SR+0.707*LFE",
380 *filter_params,
381 ]
382
383 # determine if we need to do resampling (or dithering)
384 if input_format.sample_rate != output_format.sample_rate or (
385 input_format.bit_depth > 16 and output_format.bit_depth == 16
386 ):
387 libsoxr_support = get_global_cache_value(CACHE_ATTR_LIBSOXR_PRESENT)
388 # prefer resampling with libsoxr due to its high quality
389 # but skip if loudnorm filter is present, due to this bug:
390 # https://trac.ffmpeg.org/ticket/11323
391 loudnorm_present = any("loudnorm" in f for f in filter_params)
392 if libsoxr_support and not loudnorm_present:
393 resample_filter = "aresample=resampler=soxr:precision=30"
394 else:
395 resample_filter = "aresample=resampler=swr"
396
397 # sample rate conversion
398 if input_format.sample_rate != output_format.sample_rate:
399 resample_filter += f":osr={output_format.sample_rate}"
400
401 # bit depth conversion: apply dithering when going down to 16 bits
402 # this is only needed when we need to back to 16 bits
403 # when going from 32bits FP to 24 bits no dithering is needed
404 if output_format.bit_depth == 16 and input_format.bit_depth > 16:
405 resample_filter += ":osf=s16:dither_method=triangular_hp"
406
407 filter_params.append(resample_filter)
408
409 if filter_params and "-filter_complex" not in extra_args:
410 extra_args += ["-af", ",".join(filter_params)]
411
412 return generic_args + input_args + extra_args + output_args
413
414
415async def check_ffmpeg_version() -> None:
416 """Check if ffmpeg is present (with libsoxr support)."""
417 # check for FFmpeg presence
418 try:
419 returncode, output = await check_output("ffmpeg", "-version")
420 except FileNotFoundError:
421 raise AudioError(
422 "FFmpeg binary is missing from system. "
423 "Please install ffmpeg on your OS to enable playback."
424 )
425 if returncode != 0:
426 err_msg = "Error determining FFmpeg version on your system."
427 if returncode < 0:
428 # error below 0 is often illegal instruction
429 err_msg += " - Your CPU may be too old to run this version of FFmpeg."
430 err_msg += f" - Additional info: {returncode} {output.decode().strip()}"
431 raise AudioError(err_msg)
432 # parse version number from output
433 try:
434 version = output.decode().split("ffmpeg version ")[1].split(" ")[0].split("-")[0]
435 except IndexError:
436 raise AudioError(
437 "Error determining FFmpeg version on your system."
438 f"Additional info: {returncode} {output.decode().strip()}"
439 )
440 libsoxr_support = "enable-libsoxr" in output.decode()
441 # use globals as in-memory cache
442 await set_global_cache_values({CACHE_ATTR_LIBSOXR_PRESENT: libsoxr_support})
443
444 major_version = int("".join(char for char in version.split(".")[0] if not char.isalpha()))
445 if major_version < MINIMAL_FFMPEG_VERSION:
446 raise AudioError(
447 f"FFmpeg version {version} is not supported. "
448 f"Minimal version required is {MINIMAL_FFMPEG_VERSION}."
449 )
450
451 LOGGER.info(
452 "Detected ffmpeg version %s %s",
453 version,
454 "with libsoxr support" if libsoxr_support else "",
455 )
456