/
/
1"""Tests for the ffmpeg helper module."""
2
3from __future__ import annotations
4
5import asyncio
6import subprocess
7from array import array
8from collections.abc import AsyncGenerator, Sequence
9from math import sqrt
10from pathlib import Path
11
12import pytest
13from music_assistant_models.enums import ContentType
14from music_assistant_models.media_items import AudioFormat
15
16from music_assistant.helpers.dsp import ComplexFilter, ComplexFilterInput
17from music_assistant.helpers.ffmpeg import (
18 _INPUT_READ_ARGS,
19 FFMpeg,
20 FFMpegStreamInfo,
21 _build_filtergraph_args,
22 _build_overlay_mixer,
23 _get_overlay_volume_filter,
24 get_ffmpeg_args,
25 get_ffmpeg_overlay_stream,
26 parse_ffmpeg_duration,
27 parse_ffmpeg_stream_info,
28)
29
30
31def test_get_ffmpeg_args_does_not_mutate_filters() -> None:
32 """Automatic resampling must not alter a caller-owned filter plan."""
33 input_format = AudioFormat(
34 content_type=ContentType.PCM_F32LE,
35 sample_rate=96000,
36 bit_depth=32,
37 channels=2,
38 )
39 output_format = AudioFormat(
40 content_type=ContentType.FLAC,
41 sample_rate=48000,
42 bit_depth=16,
43 channels=2,
44 )
45 filter_params = ["volume=-1dB"]
46
47 get_ffmpeg_args(input_format, output_format, filter_params)
48
49 assert filter_params == ["volume=-1dB"]
50
51
52def test_get_ffmpeg_args_downmixes_multichannel_for_single_channel_output() -> None:
53 """A surround source is folded to stereo before the output is narrowed to one channel."""
54 input_format = AudioFormat(
55 content_type=ContentType.PCM_F32LE,
56 sample_rate=48000,
57 bit_depth=32,
58 channels=6,
59 )
60 output_format = AudioFormat(
61 content_type=ContentType.FLAC,
62 sample_rate=48000,
63 bit_depth=16,
64 channels=1,
65 )
66
67 args = get_ffmpeg_args(input_format, output_format, ["pan=mono|c0=0.5*FL+0.5*FR"])
68
69 filter_graph = args[args.index("-af") + 1]
70 assert filter_graph.index("aformat=channel_layouts=stereo") < filter_graph.index(
71 "pan=mono|c0=0.5*FL+0.5*FR"
72 )
73
74
75def _split_at_input(args: list[str]) -> tuple[list[str], list[str]]:
76 """Split generated ffmpeg args into the part describing the input and the output."""
77 idx = args.index("-i")
78 return args[:idx], args[idx + 2 :]
79
80
81@pytest.mark.parametrize(
82 ("channels", "expected_layout"),
83 [(1, "mono"), (2, "stereo")],
84)
85def test_get_ffmpeg_args_names_layout_up_to_stereo(channels: int, expected_layout: str) -> None:
86 """Mono and stereo PCM are described by both their channel count and their layout."""
87 fmt = AudioFormat(
88 content_type=ContentType.PCM_S24LE,
89 sample_rate=48000,
90 bit_depth=24,
91 channels=channels,
92 )
93
94 input_args, output_args = _split_at_input(get_ffmpeg_args(fmt, fmt, []))
95
96 for part in (input_args, output_args):
97 assert part[part.index("-ac") + 1] == str(channels)
98 assert part[part.index("-channel_layout") + 1] == expected_layout
99
100
101def test_get_ffmpeg_args_omits_layout_above_stereo() -> None:
102 """Surround PCM is described by its channel count alone, never as a stereo layout."""
103 fmt = AudioFormat(
104 content_type=ContentType.PCM_S24LE,
105 sample_rate=48000,
106 bit_depth=24,
107 channels=6,
108 )
109
110 input_args, output_args = _split_at_input(get_ffmpeg_args(fmt, fmt, []))
111
112 for part in (input_args, output_args):
113 assert part[part.index("-ac") + 1] == "6"
114 assert "-channel_layout" not in part
115
116
117def test_multichannel_pcm_folds_down_without_stretching(tmp_path: Path) -> None:
118 """Raw surround PCM keeps its real width and length, and its rear channels survive."""
119 source = tmp_path / "surround.pcm"
120 out = tmp_path / "out.flac"
121 # only the rear channels carry a tone: a downmix that drops them yields silence
122 subprocess.run( # noqa: S603
123 [ # noqa: S607
124 "ffmpeg",
125 "-y",
126 "-f",
127 "lavfi",
128 "-i",
129 "sine=frequency=1000:duration=1:sample_rate=48000",
130 "-af",
131 "pan=5.1|BL=c0|BR=c0",
132 "-f",
133 "s16le",
134 str(source),
135 ],
136 check=True,
137 capture_output=True,
138 )
139 pcm_format = AudioFormat(
140 content_type=ContentType.PCM_S16LE,
141 sample_rate=48000,
142 bit_depth=16,
143 channels=6,
144 )
145 output_format = AudioFormat(
146 content_type=ContentType.FLAC,
147 sample_rate=48000,
148 bit_depth=16,
149 channels=2,
150 )
151 args = get_ffmpeg_args(
152 pcm_format, output_format, [], input_path=str(source), output_path=str(out)
153 )
154
155 result = subprocess.run([*args, "-y"], capture_output=True, text=True, check=False) # noqa: S603
156 assert result.returncode == 0, result.stderr
157
158 duration = subprocess.run( # noqa: S603
159 [ # noqa: S607
160 "ffprobe",
161 "-v",
162 "error",
163 "-show_entries",
164 "format=duration",
165 "-of",
166 "csv=p=0",
167 str(out),
168 ],
169 capture_output=True,
170 text=True,
171 check=True,
172 ).stdout
173 assert float(duration) == pytest.approx(1.0, abs=0.05)
174 assert _rms_db(out) > -40
175
176
177def _output_args(args: list[str]) -> list[str]:
178 """Return the output section of an ffmpeg command line (everything past the input path)."""
179 return args[args.index("-i") + 2 :]
180
181
182@pytest.mark.parametrize(
183 ("content_type", "encoder_args"),
184 [
185 (ContentType.AAC, ["-f", "adts", "-c:a", "aac", "-b:a", "256k"]),
186 (ContentType.MP3, ["-f", "mp3", "-b:a", "320k"]),
187 (ContentType.WAV, ["-ar", "44100", "-acodec", "pcm_s16le", "-f", "wav"]),
188 (
189 ContentType.FLAC,
190 ["-sample_fmt", "s16", "-ar", "44100", "-f", "flac", "-compression_level", "0"],
191 ),
192 ],
193)
194def test_get_ffmpeg_args_encoded_output_declares_channels(
195 content_type: ContentType, encoder_args: list[str]
196) -> None:
197 """Every encoded output format is handed the requested channel count."""
198 input_format = AudioFormat(
199 content_type=ContentType.PCM_S16LE, sample_rate=44100, bit_depth=16, channels=2
200 )
201 output_format = AudioFormat(content_type=content_type, sample_rate=44100, bit_depth=16)
202
203 args = get_ffmpeg_args(input_format, output_format, [])
204
205 assert _output_args(args) == ["-ac", "2", "-channel_layout", "stereo", *encoder_args, "-"]
206
207
208def test_get_ffmpeg_args_single_channel_output_declares_mono() -> None:
209 """A one channel target is declared as mono rather than stereo."""
210 fmt = AudioFormat(
211 content_type=ContentType.PCM_S16LE, sample_rate=44100, bit_depth=16, channels=1
212 )
213 output_format = AudioFormat(
214 content_type=ContentType.WAV, sample_rate=44100, bit_depth=16, channels=1
215 )
216
217 args = get_ffmpeg_args(fmt, output_format, [])
218
219 assert _output_args(args) == [
220 "-ac",
221 "1",
222 "-channel_layout",
223 "mono",
224 "-ar",
225 "44100",
226 "-acodec",
227 "pcm_s16le",
228 "-f",
229 "wav",
230 "-",
231 ]
232
233
234@pytest.mark.parametrize(
235 ("output_path", "output_content_type", "expected"),
236 [
237 ("NULL", ContentType.FLAC, ["-f", "null", "-"]),
238 ("-", ContentType.NUT, ["-vn", "-dn", "-sn", "-acodec", "copy", "-f", "nut", "-"]),
239 ],
240)
241def test_get_ffmpeg_args_passthrough_sinks_omit_channels(
242 output_path: str, output_content_type: ContentType, expected: list[str]
243) -> None:
244 """The analysis sink and the cache passthrough declare no channel count of their own."""
245 input_format = AudioFormat(
246 content_type=ContentType.PCM_S16LE, sample_rate=44100, bit_depth=16, channels=1
247 )
248 output_format = AudioFormat(
249 content_type=output_content_type, sample_rate=44100, bit_depth=16, channels=1
250 )
251
252 args = get_ffmpeg_args(input_format, output_format, [], output_path=output_path)
253
254 assert _output_args(args) == expected
255
256
257def test_get_ffmpeg_args_duplicates_mono_source_for_stereo_output() -> None:
258 """A mono source is widened by duplication, ahead of the caller's own filters."""
259 input_format = AudioFormat(
260 content_type=ContentType.PCM_S16LE, sample_rate=44100, bit_depth=16, channels=1
261 )
262 output_format = AudioFormat(
263 content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16, channels=2
264 )
265
266 args = get_ffmpeg_args(input_format, output_format, ["volume=-1dB"])
267
268 assert args[args.index("-af") + 1] == "pan=stereo|c0=c0|c1=c0,volume=-1dB"
269
270
271def test_get_ffmpeg_args_mono_source_to_mono_output_is_not_widened() -> None:
272 """A mono source kept at one channel needs no channel filter at all."""
273 fmt = AudioFormat(
274 content_type=ContentType.PCM_S16LE, sample_rate=44100, bit_depth=16, channels=1
275 )
276 output_format = AudioFormat(
277 content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16, channels=1
278 )
279
280 args = get_ffmpeg_args(fmt, output_format, [])
281
282 assert "-af" not in args
283
284
285# -- parse_ffmpeg_stream_info --
286
287
288def test_parse_stream_info_mp3() -> None:
289 """Lossy MP3 line yields codec/sample rate/bit rate, but no bit depth."""
290 line = "Stream #0:0: Audio: mp3, 44100 Hz, stereo, fltp, 320 kb/s"
291 info = parse_ffmpeg_stream_info(line)
292 assert info == FFMpegStreamInfo(
293 codec=ContentType.MP3,
294 sample_rate=44100,
295 bit_depth=None,
296 bit_rate=320,
297 )
298
299
300def test_parse_stream_info_aac_with_profile_and_language() -> None:
301 """AAC line with profile annotation and language tag is parsed correctly."""
302 line = "Stream #0:0(eng): Audio: aac (LC) (mp4a / 0x6134706D), 44100 Hz, stereo, fltp, 254 kb/s"
303 info = parse_ffmpeg_stream_info(line)
304 assert info is not None
305 assert info.codec == ContentType.AAC
306 assert info.sample_rate == 44100
307 assert info.bit_rate == 254
308 assert info.bit_depth is None
309
310
311def test_parse_stream_info_flac_16bit() -> None:
312 """16-bit FLAC: bit depth is inferred from the s16 sample format token."""
313 line = "Stream #0:0: Audio: flac, 44100 Hz, stereo, s16, 1024 kb/s"
314 info = parse_ffmpeg_stream_info(line)
315 assert info == FFMpegStreamInfo(
316 codec=ContentType.FLAC,
317 sample_rate=44100,
318 bit_depth=16,
319 bit_rate=1024,
320 )
321
322
323def test_parse_stream_info_flac_24bit_in_s32() -> None:
324 """24-bit FLAC is stored in s32; the explicit "(24 bit)" annotation wins."""
325 line = "Stream #0:0: Audio: flac, 96000 Hz, stereo, s32 (24 bit)"
326 info = parse_ffmpeg_stream_info(line)
327 assert info == FFMpegStreamInfo(
328 codec=ContentType.FLAC,
329 sample_rate=96000,
330 bit_depth=24,
331 bit_rate=None,
332 )
333
334
335def test_parse_stream_info_flac_24bit_hires_with_bitrate() -> None:
336 """High-resolution 24-bit FLAC at 192k with reported bit rate."""
337 line = "Stream #0:0: Audio: flac, 192000 Hz, stereo, s32 (24 bit), 5644 kb/s"
338 info = parse_ffmpeg_stream_info(line)
339 assert info == FFMpegStreamInfo(
340 codec=ContentType.FLAC,
341 sample_rate=192000,
342 bit_depth=24,
343 bit_rate=5644,
344 )
345
346
347def test_parse_stream_info_pcm_s16le() -> None:
348 """PCM stream reports codec via try_parse, sample format gives bit depth."""
349 line = "Stream #0:0: Audio: pcm_s16le, 44100 Hz, stereo, s16, 1411 kb/s"
350 info = parse_ffmpeg_stream_info(line)
351 assert info == FFMpegStreamInfo(
352 codec=ContentType.PCM_S16LE,
353 sample_rate=44100,
354 bit_depth=16,
355 bit_rate=1411,
356 )
357
358
359def test_parse_stream_info_opus_without_bitrate() -> None:
360 """Opus often omits bit rate; we still get codec and sample rate."""
361 line = "Stream #0:0: Audio: opus, 48000 Hz, stereo, fltp"
362 info = parse_ffmpeg_stream_info(line)
363 assert info == FFMpegStreamInfo(
364 codec=ContentType.OPUS,
365 sample_rate=48000,
366 bit_depth=None,
367 bit_rate=None,
368 )
369
370
371def test_parse_stream_info_alac_planar() -> None:
372 """ALAC reported with s16p (planar) sample format still yields 16-bit depth."""
373 line = "Stream #0:0: Audio: alac (alac / 0x63616C61), 44100 Hz, stereo, s16p"
374 info = parse_ffmpeg_stream_info(line)
375 assert info is not None
376 assert info.codec == ContentType.ALAC
377 assert info.sample_rate == 44100
378 assert info.bit_depth == 16
379
380
381def test_parse_stream_info_returns_none_for_non_stream_line() -> None:
382 """Non-stream log lines must return None."""
383 assert parse_ffmpeg_stream_info("Duration: 00:03:25.78, start: 0.000000") is None
384 assert parse_ffmpeg_stream_info("[error] Invalid data found") is None
385 assert parse_ffmpeg_stream_info("") is None
386
387
388def test_parse_stream_info_ignores_video_stream() -> None:
389 """Video stream lines must not be misparsed as audio."""
390 line = "Stream #0:0: Video: h264 (High), yuv420p, 1920x1080, 5000 kb/s, 25 fps"
391 assert parse_ffmpeg_stream_info(line) is None
392
393
394def test_parse_stream_info_unknown_codec_still_yields_other_fields() -> None:
395 """Unrecognised codec token returns UNKNOWN but sample rate / bit rate are still parsed."""
396 line = "Stream #0:0: Audio: somenewcodec, 48000 Hz, stereo, 192 kb/s"
397 info = parse_ffmpeg_stream_info(line)
398 assert info is not None
399 assert info.codec == ContentType.UNKNOWN
400 assert info.sample_rate == 48000
401 assert info.bit_rate == 192
402 assert info.bit_depth is None
403
404
405# -- parse_ffmpeg_duration --
406
407
408def test_parse_duration_typical() -> None:
409 """Typical 'Duration: HH:MM:SS.ms' line yields total seconds (floor)."""
410 line = "Duration: 00:03:25.78, start: 0.000000, bitrate: 320 kb/s"
411 assert parse_ffmpeg_duration(line) == 3 * 60 + 25
412
413
414def test_parse_duration_one_hour() -> None:
415 """Hours component is honoured."""
416 assert parse_ffmpeg_duration("Duration: 01:00:00.00, bitrate: 128 kb/s") == 3600
417
418
419def test_parse_duration_under_one_second() -> None:
420 """Sub-second durations round down to 0."""
421 assert parse_ffmpeg_duration("Duration: 00:00:00.50, bitrate: 128 kb/s") == 0
422
423
424def test_parse_duration_na_returns_none() -> None:
425 """Live streams report 'Duration: N/A' — must not match."""
426 assert parse_ffmpeg_duration("Duration: N/A, start: 0.000000, bitrate: N/A") is None
427
428
429def test_parse_duration_unrelated_line_returns_none() -> None:
430 """Random log lines must return None."""
431 assert parse_ffmpeg_duration("Stream #0:0: Audio: mp3, 44100 Hz") is None
432 assert parse_ffmpeg_duration("") is None
433
434
435# -- get_ffmpeg_overlay_stream (end-to-end with a real ffmpeg process) --
436
437_PCM_FORMAT = AudioFormat(
438 content_type=ContentType.PCM_S16LE, sample_rate=44100, bit_depth=16, channels=2
439)
440_BYTES_PER_SECOND = _PCM_FORMAT.pcm_sample_size # 1 second of PCM audio
441
442
443@pytest.fixture
444def overlay_file(tmp_path: Path) -> Path:
445 """Generate a 1 second mono sine-tone wav file to use as overlay source."""
446 overlay_path = tmp_path / "overlay.wav"
447 subprocess.run( # noqa: S603
448 ["ffmpeg", "-f", "lavfi", "-i", "sine=frequency=440:duration=1", str(overlay_path)], # noqa: S607
449 check=True,
450 capture_output=True,
451 )
452 return overlay_path
453
454
455@pytest.fixture
456def overlay_file_stereo(tmp_path: Path) -> Path:
457 """Generate a stereo overlay wav carrying the same tone as ``overlay_file`` on both channels."""
458 overlay_path = tmp_path / "overlay_stereo.wav"
459 subprocess.run( # noqa: S603
460 [ # noqa: S607
461 "ffmpeg",
462 "-f",
463 "lavfi",
464 "-i",
465 "sine=frequency=440:duration=1",
466 "-af",
467 "pan=stereo|c0=c0|c1=c0",
468 str(overlay_path),
469 ],
470 check=True,
471 capture_output=True,
472 )
473 return overlay_path
474
475
476@pytest.fixture
477def overlay_file_wide_stereo(tmp_path: Path) -> Path:
478 """Generate a 1 second stereo overlay wav with the tone on the left channel only."""
479 overlay_path = tmp_path / "overlay_wide.wav"
480 subprocess.run( # noqa: S603
481 [ # noqa: S607
482 "ffmpeg",
483 "-f",
484 "lavfi",
485 "-i",
486 "sine=frequency=440:duration=1",
487 "-af",
488 "pan=stereo|c0=c0",
489 str(overlay_path),
490 ],
491 check=True,
492 capture_output=True,
493 )
494 return overlay_path
495
496
497@pytest.fixture
498def overlay_file_with_silent_intro(tmp_path: Path) -> Path:
499 """Generate a 2 second overlay wav that starts with 1s of silence then a 1s tone."""
500 overlay_path = tmp_path / "overlay_silent_intro.wav"
501 subprocess.run( # noqa: S603
502 [ # noqa: S607
503 "ffmpeg",
504 "-f",
505 "lavfi",
506 "-i",
507 "sine=frequency=440:duration=1",
508 "-af",
509 "adelay=1000:all=1",
510 str(overlay_path),
511 ],
512 check=True,
513 capture_output=True,
514 )
515 return overlay_path
516
517
518async def _silence(seconds: int) -> AsyncGenerator[bytes]:
519 """Yield the given amount of seconds of PCM silence in 1-second chunks."""
520 for _ in range(seconds):
521 yield b"\x00" * _BYTES_PER_SECOND
522
523
524async def _collect_chunks(stream: AsyncGenerator[bytes]) -> list[bytes]:
525 return [chunk async for chunk in stream]
526
527
528def _samples(pcm: bytes, channel: int | None = None) -> Sequence[int]:
529 """Return the samples of the given PCM audio, optionally for one channel only."""
530 samples = array("h")
531 samples.frombytes(pcm)
532 return samples if channel is None else samples[channel :: _PCM_FORMAT.channels]
533
534
535def _rms(samples: Sequence[int]) -> float:
536 """Return the RMS level of the given PCM samples."""
537 return sqrt(sum(sample * sample for sample in samples) / len(samples))
538
539
540async def _mix_overlay(overlay_input: Path) -> bytes:
541 """Mix the given overlay source into 1 second of silence and return the result."""
542 return b"".join(
543 await _collect_chunks(
544 get_ffmpeg_overlay_stream(
545 audio_input=_silence(1),
546 overlay_input=str(overlay_input),
547 pcm_format=_PCM_FORMAT,
548 )
549 )
550 )
551
552
553async def test_overlay_stream_mixes_loops_and_preserves_length(overlay_file: Path) -> None:
554 """The overlay is looped and mixed in while length, format and chunking stay intact."""
555 chunks = await _collect_chunks(
556 get_ffmpeg_overlay_stream(
557 audio_input=_silence(3),
558 overlay_input=str(overlay_file),
559 pcm_format=_PCM_FORMAT,
560 chunk_size=_BYTES_PER_SECOND,
561 )
562 )
563 output = b"".join(chunks)
564 # duration=first: output length exactly matches the 3s main input
565 assert len(output) == 3 * _BYTES_PER_SECOND
566 # all chunks except the last are exactly chunk_size
567 assert all(len(chunk) == _BYTES_PER_SECOND for chunk in chunks[:-1])
568 # the main input was pure silence, so any signal proves the overlay was mixed in;
569 # signal in the third second proves the 1s overlay file was looped
570 assert any(output[:_BYTES_PER_SECOND])
571 assert any(output[2 * _BYTES_PER_SECOND :])
572
573
574async def test_overlay_stream_does_not_mutate_pcm_format(overlay_file: Path) -> None:
575 """Mixing an overlay leaves the caller's PCM format unchanged."""
576 pcm_format = AudioFormat(
577 content_type=ContentType.PCM_F32LE,
578 sample_rate=48000,
579 bit_depth=32,
580 channels=2,
581 )
582 original_format = pcm_format.to_dict()
583
584 async def silence() -> AsyncGenerator[bytes]:
585 yield b"\x00" * pcm_format.pcm_sample_size
586
587 await _collect_chunks(
588 get_ffmpeg_overlay_stream(
589 audio_input=silence(),
590 overlay_input=str(overlay_file),
591 pcm_format=pcm_format,
592 )
593 )
594
595 assert pcm_format.to_dict() == original_format
596
597
598async def test_overlay_stream_applies_volume(overlay_file: Path) -> None:
599 """Overlay volume 0% silences the overlay entirely (gain is applied)."""
600 output = b"".join(
601 await _collect_chunks(
602 get_ffmpeg_overlay_stream(
603 audio_input=_silence(1),
604 overlay_input=str(overlay_file),
605 pcm_format=_PCM_FORMAT,
606 overlay_volume=0,
607 )
608 )
609 )
610 assert len(output) == _BYTES_PER_SECOND
611 assert not any(output)
612
613
614async def test_overlay_stream_trims_leading_silence(
615 overlay_file_with_silent_intro: Path,
616) -> None:
617 """A near-silent intro on the overlay source is trimmed so it plays immediately."""
618 output = b"".join(
619 await _collect_chunks(
620 get_ffmpeg_overlay_stream(
621 audio_input=_silence(1),
622 overlay_input=str(overlay_file_with_silent_intro),
623 pcm_format=_PCM_FORMAT,
624 )
625 )
626 )
627 assert len(output) == _BYTES_PER_SECOND
628 # without trimming, the first second would be the overlay's silent intro;
629 # the trim makes the tone play from the start, so the first second has signal
630 assert any(output)
631
632
633async def test_overlay_stream_level_is_independent_of_source_channel_count(
634 overlay_file: Path, overlay_file_stereo: Path
635) -> None:
636 """A mono overlay source mixes in at the same level as an identical stereo one."""
637 mono_level = _rms(_samples(await _mix_overlay(overlay_file)))
638 stereo_level = _rms(_samples(await _mix_overlay(overlay_file_stereo)))
639 assert mono_level > 0
640 # left to FFmpeg, the mono source would be spread at 1/sqrt(2) per channel
641 # and land a factor sqrt(2) (3 dB) below the stereo one
642 assert mono_level == pytest.approx(stereo_level, rel=0.02)
643
644
645async def test_overlay_stream_preserves_stereo_image(overlay_file_wide_stereo: Path) -> None:
646 """A stereo overlay keeps its channels apart instead of being folded to dual mono."""
647 output = await _mix_overlay(overlay_file_wide_stereo)
648 # the source carries the tone on the left only, so a fold would leak it into the right
649 assert _rms(_samples(output, channel=0)) > 0
650 assert not any(_samples(output, channel=1))
651
652
653# -- overlay volume filter --
654
655
656def test_overlay_volume_filter_compensates_only_for_a_stereo_output() -> None:
657 """Only the widening to stereo costs a mono source level, so only there is it scaled up."""
658 stereo_output = _get_overlay_volume_filter(100, 2)
659 assert "nb_channels" in stereo_output
660 # a comma would read as the end of this filter in the graph
661 assert "," not in stereo_output
662 # a mono source keeps its level when it is widened further, or not at all
663 assert _get_overlay_volume_filter(100, 1) == "volume=1.0"
664 assert _get_overlay_volume_filter(60, 6) == "volume=0.6"
665
666
667def test_overlay_mixer_loops_its_source() -> None:
668 """The overlay source is looped for as long as the main stream runs."""
669 (overlay,) = _build_overlay_mixer("/sound.wav", _PCM_FORMAT, 100).inputs
670 # a local file has nothing to reconnect to
671 assert overlay.input_args == ["-stream_loop", "-1"]
672
673
674def test_overlay_mixer_reconnects_for_http_sources() -> None:
675 """An http overlay source additionally gets the reconnect options."""
676 (overlay,) = _build_overlay_mixer("http://host/sound.mp3", _PCM_FORMAT, 100).inputs
677 assert "-reconnect" in overlay.input_args
678 assert overlay.input_args[-2:] == ["-stream_loop", "-1"]
679
680
681def test_overlay_args_probe_the_main_input_and_add_no_filters() -> None:
682 """Both inputs are read under our own limits and no filter is injected on top."""
683 args = get_ffmpeg_args(
684 _PCM_FORMAT, _PCM_FORMAT, [_build_overlay_mixer("/sound.wav", _PCM_FORMAT, 100)]
685 )
686 main_input = args.index("-i")
687 # the limits must reach the main input, which is the first one, and the overlay
688 assert args.index("-probesize") < main_input
689 assert args.count("-probesize") == 2
690 assert args.index("-stream_loop") > main_input
691 # the overlay format matches the output, so nothing gets resampled or reconformed
692 assert "-af" not in args
693 assert args.count("-filter_complex") == 1
694 assert not any(arg.startswith(("pan=", "aresample=resampler=")) for arg in args)
695
696
697@pytest.mark.parametrize(
698 "extra_input_args",
699 [
700 # the concat demuxer brings its own -f, and needs the whitelist for the listed files
701 ["-safe", "0", "-f", "concat", "-i", "/list.txt"],
702 # a caller raising the probe limits relies on its own values landing last
703 ["-probesize", "65536", "-analyzeduration", "5000000"],
704 ],
705 ids=["caller-supplied-input-format", "caller-raised-probe-limits"],
706)
707def test_read_args_lead_the_main_input(extra_input_args: list[str]) -> None:
708 """Every main input is opened under our read limits, which the caller may override."""
709 args = get_ffmpeg_args(_PCM_FORMAT, _PCM_FORMAT, [], extra_input_args=extra_input_args)
710 start = args.index("-protocol_whitelist")
711 end = start + len(_INPUT_READ_ARGS)
712 assert args[start:end] == _INPUT_READ_ARGS
713 # the caller's args follow ours within the same input group, so theirs win on a conflict
714 assert args[end : end + len(extra_input_args)] == extra_input_args
715 # exactly one input either way: we add ours only when the caller brings none
716 assert args.count("-i") == 1
717
718
719# -- _log_reader_task (decode-error flood guard) --
720
721
722class _FakeStream:
723 """Stand-in for a StreamReader/StreamWriter: already closed/at EOF, nothing to drain."""
724
725 def is_closing(self) -> bool:
726 return True
727
728 def at_eof(self) -> bool:
729 return True
730
731
732class _FakeProc:
733 """Minimal stand-in for asyncio.subprocess.Process — just enough for close() to run."""
734
735 def __init__(self) -> None:
736 self.pid = 12345
737 self.returncode: int | None = None
738 self.stdin: _FakeStream | None = _FakeStream()
739 self.stdout = _FakeStream()
740
741 async def communicate(self) -> tuple[bytes, bytes]:
742 self.returncode = 0
743 return b"", b""
744
745 def send_signal(self, _sig: int) -> None:
746 pass
747
748
749class _FakeProcRacingExit(_FakeProc):
750 """A no-stdin process that has already exited, so send_signal raises ProcessLookupError."""
751
752 def __init__(self) -> None:
753 super().__init__()
754 # no stdin routes close() down the send_signal(SIGINT) branch
755 self.stdin = None
756
757 def send_signal(self, _sig: int) -> None:
758 raise ProcessLookupError("no such process")
759
760
761async def test_log_reader_reports_decode_errors_once_and_aborts() -> None:
762 """
763 Crossing the decode-error threshold logs a single line and aborts the stream.
764
765 Regression test: previously every stderr line was re-promoted to ERROR for the
766 rest of the process once 50 "Invalid data" lines were seen, flooding the log
767 with thousands of lines for a single corrupted file.
768 """
769 ffmpeg = FFMpeg(audio_input="-", input_format=_PCM_FORMAT, output_format=_PCM_FORMAT)
770
771 async def fake_stderr() -> AsyncGenerator[str]:
772 for _ in range(60):
773 yield "Invalid data found when processing input"
774 # noise that a genuinely corrupted stream keeps emitting after the threshold;
775 # none of this should reach ERROR level under the fix
776 for _ in range(20):
777 yield "Reserved bit set."
778
779 ffmpeg.iter_stderr = fake_stderr # type: ignore[method-assign]
780
781 error_lines: list[str] = []
782 ffmpeg.logger.error = lambda msg, *args: error_lines.append(msg % args if args else msg) # type: ignore[method-assign]
783
784 await ffmpeg._log_reader_task()
785 assert ffmpeg._abort_task is not None
786 await ffmpeg._abort_task
787
788 assert error_lines == ["Excessive decode errors (50+) for this stream; aborting"]
789 assert ffmpeg.closed
790
791
792async def test_log_reader_below_threshold_does_not_abort() -> None:
793 """A handful of decode errors, well under the threshold, triggers no report or abort."""
794 ffmpeg = FFMpeg(audio_input="-", input_format=_PCM_FORMAT, output_format=_PCM_FORMAT)
795
796 async def fake_stderr() -> AsyncGenerator[str]:
797 for _ in range(10):
798 yield "Invalid data found when processing input"
799 yield "Reserved bit set."
800
801 ffmpeg.iter_stderr = fake_stderr # type: ignore[method-assign]
802
803 error_lines: list[str] = []
804 ffmpeg.logger.error = lambda msg, *args: error_lines.append(msg % args if args else msg) # type: ignore[method-assign]
805
806 await ffmpeg._log_reader_task()
807
808 assert error_lines == []
809 assert ffmpeg._abort_task is None
810 assert not ffmpeg.closed
811
812
813async def test_log_reader_abort_does_not_self_deadlock() -> None:
814 """
815 The detached abort task can close() the reader's own process without a self-await.
816
817 Regression test for the deadlock this PR reintroduces abort-on-close around:
818 close() does ``await asyncio.wait_for(self._stderr_reader_task, 5)``, and
819 ``_stderr_reader_task`` here is wired to the very task running
820 ``_log_reader_task`` — the same setup ``start()`` uses in production. If the
821 abort were awaited inline from within ``_log_reader_task`` instead of via a
822 detached task, that task would be awaiting itself, which asyncio turns into
823 ``RuntimeError: Task cannot await on itself`` rather than a hang.
824 """
825 ffmpeg = FFMpeg(audio_input="-", input_format=_PCM_FORMAT, output_format=_PCM_FORMAT)
826 ffmpeg.proc = _FakeProc() # type: ignore[assignment]
827
828 async def fake_stderr() -> AsyncGenerator[str]:
829 for _ in range(50):
830 yield "Invalid data found when processing input"
831
832 ffmpeg.iter_stderr = fake_stderr # type: ignore[method-assign]
833
834 reader_task = asyncio.create_task(ffmpeg._log_reader_task())
835 ffmpeg._stderr_reader_task = reader_task
836
837 await asyncio.wait_for(reader_task, timeout=2)
838 assert ffmpeg._abort_task is not None
839 await asyncio.wait_for(ffmpeg._abort_task, timeout=2)
840
841 assert ffmpeg.closed
842
843
844async def test_abort_survives_send_signal_racing_process_exit() -> None:
845 """
846 The fire-and-forget abort must not crash if the process exits before it is signalled.
847
848 close() sends SIGINT to no-stdin processes, which raises ProcessLookupError when the
849 process already exited between the returncode check and the signal. The abort task is
850 never awaited by a caller, so that race must be swallowed inside close() rather than
851 escaping as an untracked "Task exception was never retrieved".
852 """
853 ffmpeg = FFMpeg(audio_input="-", input_format=_PCM_FORMAT, output_format=_PCM_FORMAT)
854 ffmpeg.proc = _FakeProcRacingExit() # type: ignore[assignment]
855
856 async def fake_stderr() -> AsyncGenerator[str]:
857 for _ in range(50):
858 yield "Invalid data found when processing input"
859
860 ffmpeg.iter_stderr = fake_stderr # type: ignore[method-assign]
861
862 reader_task = asyncio.create_task(ffmpeg._log_reader_task())
863 ffmpeg._stderr_reader_task = reader_task
864
865 await asyncio.wait_for(reader_task, timeout=2)
866 assert ffmpeg._abort_task is not None
867 # must complete without propagating ProcessLookupError
868 await asyncio.wait_for(ffmpeg._abort_task, timeout=2)
869
870 assert ffmpeg.closed
871
872
873# -- _build_filtergraph_args (DSP chain assembly) --
874
875
876def test_build_filtergraph_all_simple_uses_af() -> None:
877 """A chain of plain filters renders to a single -af comma chain."""
878 assert _build_filtergraph_args(["equalizer=x", "volume=3dB"]) == (
879 [],
880 ["-af", "equalizer=x,volume=3dB"],
881 )
882
883
884def test_build_filtergraph_empty_returns_no_args() -> None:
885 """An empty chain produces no ffmpeg arguments."""
886 assert _build_filtergraph_args([]) == ([], [])
887
888
889def test_build_filtergraph_single_complex_fragment() -> None:
890 """A complex fragment renders a labelled -filter_complex graph with -map."""
891 result = _build_filtergraph_args(
892 [ComplexFilter("afir=irnorm=1", [ComplexFilterInput("/ir.wav", "aresample=48000")])]
893 )
894 assert result == (
895 [*_INPUT_READ_ARGS, "-i", "/ir.wav"],
896 [
897 "-filter_complex",
898 "[1:a]aresample=48000[dsp1];[0:a][dsp1]afir=irnorm=1[dsp2]",
899 "-map",
900 "[dsp2]",
901 ],
902 )
903
904
905def test_build_filtergraph_complex_between_simple_runs() -> None:
906 """Simple runs on either side of a complex fragment weave into labelled pads."""
907 result = _build_filtergraph_args(
908 [
909 "equalizer=x",
910 ComplexFilter("afir=irnorm=1", [ComplexFilterInput("/ir.wav", "aresample=48000")]),
911 "volume=2dB",
912 ]
913 )
914 assert result == (
915 [*_INPUT_READ_ARGS, "-i", "/ir.wav"],
916 [
917 "-filter_complex",
918 "[0:a]equalizer=x[dsp1];[1:a]aresample=48000[dsp2];"
919 "[dsp1][dsp2]afir=irnorm=1[dsp3];[dsp3]volume=2dB[dsp4]",
920 "-map",
921 "[dsp4]",
922 ],
923 )
924
925
926def test_build_filtergraph_multiple_inputs() -> None:
927 """A fragment with several inputs numbers them in order and feeds them to the body."""
928 result = _build_filtergraph_args(
929 [ComplexFilter("amerge", [ComplexFilterInput("/a.wav"), ComplexFilterInput("/b.wav")])]
930 )
931 assert result == (
932 [*_INPUT_READ_ARGS, "-i", "/a.wav", *_INPUT_READ_ARGS, "-i", "/b.wav"],
933 ["-filter_complex", "[0:a][1:a][2:a]amerge[dsp1]", "-map", "[dsp1]"],
934 )
935
936
937def test_build_filtergraph_input_args_precede_the_input() -> None:
938 """An input's own ffmpeg options are emitted directly before its -i."""
939 result = _build_filtergraph_args(
940 [
941 ComplexFilter(
942 "amix=inputs=2",
943 [ComplexFilterInput("/loop.wav", input_args=["-stream_loop", "-1"])],
944 )
945 ]
946 )
947 assert result == (
948 [*_INPUT_READ_ARGS, "-stream_loop", "-1", "-i", "/loop.wav"],
949 ["-filter_complex", "[0:a][1:a]amix=inputs=2[dsp1]", "-map", "[dsp1]"],
950 )
951
952
953def test_get_ffmpeg_args_uses_af_without_complex_filter() -> None:
954 """Plain filter chains keep the -af path (no -filter_complex/-map)."""
955 fmt = AudioFormat(
956 content_type=ContentType.PCM_S16LE, sample_rate=48000, bit_depth=16, channels=2
957 )
958 args = get_ffmpeg_args(fmt, fmt, ["volume=-1dB"])
959 assert "-af" in args
960 assert "-filter_complex" not in args
961
962
963def test_get_ffmpeg_args_uses_filter_complex_with_complex_filter() -> None:
964 """A complex fragment switches the whole chain to -filter_complex with -map."""
965 fmt = AudioFormat(
966 content_type=ContentType.PCM_S16LE, sample_rate=48000, bit_depth=16, channels=2
967 )
968 args = get_ffmpeg_args(
969 fmt, fmt, [ComplexFilter("afir=irnorm=1", [ComplexFilterInput("/ir.wav")])]
970 )
971 assert "-filter_complex" in args
972 assert "-map" in args
973 assert "-af" not in args
974 # the impulse response is a real input, so it never passes through graph quoting
975 assert args.count("-i") == 2
976 assert args.index("/ir.wav") > args.index("-i")
977
978
979def _rms_db(path: Path) -> float:
980 """Return the overall RMS level of an audio file in dB via ffmpeg astats."""
981 output = subprocess.run( # noqa: S603
982 [ # noqa: S607
983 "ffmpeg",
984 "-hide_banner",
985 "-nostats",
986 "-i",
987 str(path),
988 "-af",
989 "astats=measure_perchannel=none",
990 "-f",
991 "null",
992 "-",
993 ],
994 capture_output=True,
995 text=True,
996 check=True,
997 ).stderr
998 for line in output.splitlines():
999 if "RMS level dB" in line:
1000 return float(line.split("RMS level dB:")[-1])
1001 raise AssertionError("no RMS level in astats output")
1002
1003
1004def test_filtergraph_complex_runs_in_ffmpeg(tmp_path: Path) -> None:
1005 """The generated -filter_complex graph is valid and an identity IR passes audio through."""
1006 main = tmp_path / "main.wav"
1007 ir = tmp_path / "ir.wav"
1008 out = tmp_path / "out.wav"
1009 subprocess.run( # noqa: S603
1010 [ # noqa: S607
1011 "ffmpeg",
1012 "-y",
1013 "-f",
1014 "lavfi",
1015 "-i",
1016 "sine=frequency=1000:duration=1:sample_rate=48000",
1017 "-ac",
1018 "2",
1019 str(main),
1020 ],
1021 check=True,
1022 capture_output=True,
1023 )
1024 # a single-sample impulse is the identity IR: convolving with it returns the input
1025 subprocess.run( # noqa: S603
1026 [ # noqa: S607
1027 "ffmpeg",
1028 "-y",
1029 "-f",
1030 "lavfi",
1031 "-i",
1032 "aevalsrc=eq(n\\,0):d=0.01:s=48000:c=stereo",
1033 str(ir),
1034 ],
1035 check=True,
1036 capture_output=True,
1037 )
1038 input_args, filter_args = _build_filtergraph_args(
1039 [ComplexFilter("afir=irnorm=1", [ComplexFilterInput(str(ir), "aresample=48000")])]
1040 )
1041 result = subprocess.run( # noqa: S603
1042 [ # noqa: S607
1043 "ffmpeg",
1044 "-hide_banner",
1045 "-loglevel",
1046 "error",
1047 "-y",
1048 "-i",
1049 str(main),
1050 *input_args,
1051 *filter_args,
1052 str(out),
1053 ],
1054 capture_output=True,
1055 text=True,
1056 check=False,
1057 )
1058 assert result.returncode == 0, result.stderr
1059 assert out.exists()
1060 # identity IR => output level matches input level
1061 assert abs(_rms_db(out) - _rms_db(main)) < 0.5
1062
1063
1064def test_mono_source_keeps_its_level_when_widened_to_stereo(tmp_path: Path) -> None:
1065 """Widening a mono source to stereo duplicates it, where a rematrix would cost 3 dB."""
1066 main = tmp_path / "mono.wav"
1067 out = tmp_path / "stereo.wav"
1068 subprocess.run( # noqa: S603
1069 [ # noqa: S607
1070 "ffmpeg",
1071 "-y",
1072 "-f",
1073 "lavfi",
1074 "-i",
1075 "sine=frequency=1000:duration=1:sample_rate=44100",
1076 "-ac",
1077 "1",
1078 str(main),
1079 ],
1080 check=True,
1081 capture_output=True,
1082 )
1083 args = get_ffmpeg_args(
1084 AudioFormat(content_type=ContentType.WAV, sample_rate=44100, bit_depth=16, channels=1),
1085 AudioFormat(content_type=ContentType.WAV, sample_rate=44100, bit_depth=16, channels=2),
1086 [],
1087 input_path=str(main),
1088 output_path=str(out),
1089 )
1090 result = subprocess.run(args, capture_output=True, text=True, check=False) # noqa: S603
1091
1092 assert result.returncode == 0, result.stderr
1093 assert abs(_rms_db(out) - _rms_db(main)) < 0.5
1094