From a298418156dc9aac372ea43db721a6185a46350b Mon Sep 17 00:00:00 2001 From: Ethanfel Date: Thu, 2 Jul 2026 17:31:29 +0200 Subject: [PATCH] feat: per-join crossfade durations + acrossfade curve in build_crossfade_merge_command --- core/ffmpeg.py | 62 ++++++++++++++++++++++++++++++++------------- tests/test_utils.py | 52 +++++++++++++++++++++++++++++++++++++ 2 files changed, 97 insertions(+), 17 deletions(-) diff --git a/core/ffmpeg.py b/core/ffmpeg.py index 3be88b4..c302ed4 100644 --- a/core/ffmpeg.py +++ b/core/ffmpeg.py @@ -244,32 +244,60 @@ def build_audio_clip_command(input_path: str, start: float, duration: float, ] -def build_crossfade_merge_command(clips: list[str], crossfade: float, - out_path: str) -> list[str]: - """ffmpeg command that concatenates *clips* in order into *out_path*, - crossfading each join by *crossfade* seconds (0 = butt-join via concat). - Re-encoded per the output extension.""" +_ACROSSFADE_CURVES = frozenset({ + "tri", "qsin", "hsin", "esin", "log", "ipar", "qua", "cub", "squ", "cbr", + "par", "exp", "iqsin", "ihsin", "dese", "desi", "losi", "sinc", "isinc", + "nofade", +}) + + +def build_crossfade_merge_command(clips: list[str], crossfade, + out_path: str, curves=None) -> list[str]: + """ffmpeg command concatenating *clips* into *out_path*. *crossfade* is a + single duration (applied to every join) or a per-join list of length + len(clips)-1; 0 = butt-join. *curves* is a single acrossfade curve name or a + per-join list (default 'tri', ffmpeg's default, emitted implicitly).""" if not clips: raise ValueError("no clips to merge") + n = len(clips) ext = os.path.splitext(out_path)[1].lower() codec = _AUDIO_CODEC_BY_EXT.get(ext, []) cmd = [_bin("ffmpeg"), "-y"] for c in clips: cmd += ["-i", c] - if len(clips) == 1: + if n == 1: return cmd + ["-vn", *codec, out_path] - if crossfade > 0: - # Chain acrossfade: [0][1]->[a1]; [a1][2]->[a2]; …; last label = [out]. - # default acrossfade curve is 'tri' (linear); per-join curves deferred - parts, prev = [], "0" - for i in range(1, len(clips)): - label = "out" if i == len(clips) - 1 else f"a{i}" - parts.append(f"[{prev}][{i}]acrossfade=d={round(crossfade, 3)}[{label}]") - prev = label - fc = ";".join(parts) + # normalize per-join params (n-1 joins) + xfs = [float(crossfade)] * (n - 1) if isinstance(crossfade, (int, float)) \ + else [float(x) for x in crossfade] + if curves is None or isinstance(curves, str): + cvs = [curves or "tri"] * (n - 1) else: - inputs = "".join(f"[{i}]" for i in range(len(clips))) - fc = f"{inputs}concat=n={len(clips)}:v=0:a=1[out]" + cvs = [str(c) for c in curves] + if len(xfs) != n - 1: + raise ValueError(f"crossfade list must have {n - 1} entries, got {len(xfs)}") + if len(cvs) != n - 1: + raise ValueError(f"curves list must have {n - 1} entries, got {len(cvs)}") + for cv in cvs: + if cv not in _ACROSSFADE_CURVES: + raise ValueError(f"unknown acrossfade curve: {cv!r}") + # Chain per join: [0][1]->[a1]; [a1][2]->[a2]; …; last label = [out]. + # A join uses acrossfade when its duration>0, else per-pair concat=n=2. + # The curve suffix is emitted only for non-'tri' curves ('tri' is ffmpeg's + # default, so omitting it keeps the default command strings byte-identical). + parts, prev = [], "0" + for i in range(1, n): + j = i - 1 + label = "out" if i == n - 1 else f"a{i}" + xf = xfs[j] if j < len(xfs) else 0.0 + cv = cvs[j] if j < len(cvs) else "tri" + if xf > 0: + suffix = f":c1={cv}:c2={cv}" if cv != "tri" else "" + parts.append(f"[{prev}][{i}]acrossfade=d={round(xf, 3)}{suffix}[{label}]") + else: + parts.append(f"[{prev}][{i}]concat=n=2:v=0:a=1[{label}]") + prev = label + fc = ";".join(parts) return cmd + ["-filter_complex", fc, "-map", "[out]", *codec, out_path] diff --git a/tests/test_utils.py b/tests/test_utils.py index a9c1ac6..d0b6ddb 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -139,6 +139,58 @@ def test_merge_empty_raises(): with pytest.raises(ValueError): build_crossfade_merge_command([], 0.5, "/o/o.wav") +def test_merge_per_join_crossfades(): + cmd = build_crossfade_merge_command(["/a.wav", "/b.wav", "/c.wav"], [0.5, 1.5], "/o/o.wav") + fc = cmd[cmd.index("-filter_complex") + 1] + assert "[0][1]acrossfade=d=0.5[a1]" in fc + assert "[a1][2]acrossfade=d=1.5[out]" in fc + +def test_merge_curve_emitted_when_non_default(): + cmd = build_crossfade_merge_command(["/a.wav", "/b.wav"], 0.5, "/o/o.wav", curves="exp") + fc = cmd[cmd.index("-filter_complex") + 1] + assert "acrossfade=d=0.5:c1=exp:c2=exp[out]" in fc + +def test_merge_default_curve_omits_c1c2(): + cmd = build_crossfade_merge_command(["/a.wav", "/b.wav"], 0.5, "/o/o.wav") # tri default + fc = cmd[cmd.index("-filter_complex") + 1] + assert "c1=" not in fc # tri is ffmpeg's default -> omit + +def test_merge_per_join_curves_list(): + cmd = build_crossfade_merge_command(["/a.wav", "/b.wav", "/c.wav"], 0.5, "/o/o.wav", + curves=["tri", "exp"]) + fc = cmd[cmd.index("-filter_complex") + 1] + assert "[0][1]acrossfade=d=0.5[a1]" in fc # tri -> no suffix + assert "acrossfade=d=0.5:c1=exp:c2=exp[out]" in fc # exp on 2nd join + +def test_merge_mixed_zero_crossfade(): + cmd = build_crossfade_merge_command(["/a.wav", "/b.wav", "/c.wav"], [0.0, 0.5], "/o/o.wav") + fc = cmd[cmd.index("-filter_complex") + 1] + assert "[0][1]concat=n=2:v=0:a=1[a1]" in fc # butt-join 1st + assert "acrossfade=d=0.5" in fc # crossfade 2nd + +def test_merge_crossfade_list_length_mismatch_raises(): + import pytest + with pytest.raises(ValueError): + build_crossfade_merge_command(["/a.wav", "/b.wav", "/c.wav"], [0.5], "/o/o.wav") + with pytest.raises(ValueError): + build_crossfade_merge_command(["/a.wav", "/b.wav"], [0.5, 0.5], "/o/o.wav") + +def test_merge_curves_list_length_mismatch_raises(): + import pytest + with pytest.raises(ValueError): + build_crossfade_merge_command(["/a.wav", "/b.wav", "/c.wav"], 0.5, "/o/o.wav", curves=["exp"]) + +def test_merge_unknown_curve_raises(): + import pytest + with pytest.raises(ValueError): + build_crossfade_merge_command(["/a.wav", "/b.wav"], 0.5, "/o/o.wav", curves="bogus") + +def test_merge_three_clips_zero_concat_chain(): + cmd = build_crossfade_merge_command(["/a.wav", "/b.wav", "/c.wav"], 0.0, "/o/o.wav") + fc = cmd[cmd.index("-filter_complex") + 1] + assert fc.count("concat=n=2:v=0:a=1") == 2 + assert "acrossfade" not in fc + # --- ProcessedDB ---