feat: per-join crossfade durations + acrossfade curve in build_crossfade_merge_command

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,
+_ACROSSFADE_CURVES = frozenset({
-                                  out_path: str) -> list[str]:
+    "tri", "qsin", "hsin", "esin", "log", "ipar", "qua", "cub", "squ", "cbr",
-    """ffmpeg command that concatenates *clips* in order into *out_path*,
+    "par", "exp", "iqsin", "ihsin", "dese", "desi", "losi", "sinc", "isinc",
-    crossfading each join by *crossfade* seconds (0 = butt-join via concat).
+    "nofade",
-    Re-encoded per the output extension."""
+})
+
+
+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:
+    # normalize per-join params (n-1 joins)
-        # Chain acrossfade: [0][1]->[a1]; [a1][2]->[a2]; …; last label = [out].
+    xfs = [float(crossfade)] * (n - 1) if isinstance(crossfade, (int, float)) \
-        # default acrossfade curve is 'tri' (linear); per-join curves deferred
+        else [float(x) for x in crossfade]
-        parts, prev = [], "0"
+    if curves is None or isinstance(curves, str):
-        for i in range(1, len(clips)):
+        cvs = [curves or "tri"] * (n - 1)
-            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)
     else:
-        inputs = "".join(f"[{i}]" for i in range(len(clips)))
+        cvs = [str(c) for c in curves]
-        fc = f"{inputs}concat=n={len(clips)}:v=0:a=1[out]"
+    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]
 
 

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 ---