From 5e21f1b78ad1562901fa0e54b78ffeb3a7ff9516 Mon Sep 17 00:00:00 2001 From: tianrking Date: Thu, 16 Jul 2026 13:33:17 +0800 Subject: [PATCH] fix(audio): schedule delayed track fades correctly --- tests/tools/test_audio_mixer_track_fades.py | 95 +++++++++++++++++++++ tools/audio/audio_mixer.py | 63 ++++++++------ 2 files changed, 130 insertions(+), 28 deletions(-) create mode 100644 tests/tools/test_audio_mixer_track_fades.py diff --git a/tests/tools/test_audio_mixer_track_fades.py b/tests/tools/test_audio_mixer_track_fades.py new file mode 100644 index 00000000..caac6f14 --- /dev/null +++ b/tests/tools/test_audio_mixer_track_fades.py @@ -0,0 +1,95 @@ +"""Regression tests for delayed per-track fades in ``audio_mixer``.""" + +import shutil +import subprocess +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + +from tools.audio.audio_mixer import AudioMixer # noqa: E402 + + +@pytest.mark.parametrize("operation", ["mix", "full_mix"]) +def test_delayed_track_fades_use_the_source_timeline(tmp_path, monkeypatch, operation): + """Fade source audio before delay, and start fade-out at its real end.""" + tone = tmp_path / "tone.wav" + tone.write_bytes(b"stub") + commands = [] + + def fake_run(self, cmd, **kwargs): + commands.append(list(cmd)) + if cmd[0] == "ffprobe": + return SimpleNamespace(stdout="2.0\n", stderr="") + return SimpleNamespace(stdout="", stderr="") + + monkeypatch.setattr(AudioMixer, "run_command", fake_run) + result = AudioMixer().execute( + { + "operation": operation, + "tracks": [{ + "path": str(tone), + "role": "speech", + "start_seconds": 1, + "fade_in_seconds": 0.25, + "fade_out_seconds": 0.5, + }], + "ducking": {"enabled": False}, + "normalize": False, + "output_path": str(tmp_path / "out.wav"), + } + ) + + assert result.success, result.error + ffmpeg_cmd = next(cmd for cmd in commands if cmd[0] == "ffmpeg") + filter_graph = ffmpeg_cmd[ffmpeg_cmd.index("-filter_complex") + 1] + assert ( + "afade=t=in:d=0.25,afade=t=out:st=1.5:d=0.5,adelay=1000|1000" + in filter_graph + ) + + +@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg required") +def test_delayed_track_with_fade_out_remains_audible_after_its_start(tmp_path): + """End-to-end regression for #347: delayed audio must not be silenced.""" + tone = tmp_path / "tone.wav" + output = tmp_path / "mixed.wav" + subprocess.run( + ["ffmpeg", "-y", "-f", "lavfi", "-i", "sine=frequency=440:duration=2", str(tone)], + capture_output=True, + check=True, + timeout=30, + ) + + result = AudioMixer().execute( + { + "operation": "mix", + "tracks": [{ + "path": str(tone), + "role": "music", + "start_seconds": 1, + "fade_out_seconds": 0.5, + }], + "normalize": False, + "output_path": str(output), + } + ) + assert result.success, result.error + + measured = subprocess.run( + [ + "ffmpeg", "-ss", "1.25", "-t", "0.25", "-i", str(output), + "-vn", "-af", "volumedetect", "-f", "null", "-", + ], + capture_output=True, + text=True, + check=True, + timeout=30, + ) + mean_line = next(line for line in measured.stderr.splitlines() if "mean_volume" in line) + mean_db = float(mean_line.split("mean_volume:")[1].strip().split(" ")[0]) + assert mean_db > -60, f"delayed tone was unexpectedly silent: {mean_db} dB" diff --git a/tools/audio/audio_mixer.py b/tools/audio/audio_mixer.py index f3520236..189a153a 100644 --- a/tools/audio/audio_mixer.py +++ b/tools/audio/audio_mixer.py @@ -213,6 +213,39 @@ class AudioMixer(BaseTool): target = max(-40.0, min(0.0, target)) return f"[{in_label}]loudnorm=I={target}:LRA=11:TP=-1.5[{out_label}]" + def _track_filters(self, track: dict[str, Any]) -> list[str]: + """Build per-track filters on the source timeline before scheduling it. + + ``afade=t=out`` defaults to ``st=0``. Applying it after ``adelay`` + therefore fades the delay silence instead of the source audio, leaving + a delayed track silent by the time it starts. Fade source samples first + and add the timeline delay last so both fades follow the track itself. + """ + filters = [] + volume = track.get("volume", 1.0) + delay_ms = int(track.get("start_seconds", 0) * 1000) + fade_in = track.get("fade_in_seconds", 0) + fade_out = track.get("fade_out_seconds", 0) + + if volume != 1.0: + filters.append(f"volume={volume}") + if fade_in > 0: + filters.append(f"afade=t=in:d={fade_in}") + if fade_out > 0: + duration_cmd = [ + "ffprobe", "-v", "error", + "-show_entries", "format=duration", + "-of", "csv=p=0", + track["path"], + ] + duration = float(self.run_command(duration_cmd).stdout.strip().split("\n")[0]) + fade_start = max(0.0, duration - float(fade_out)) + filters.append(f"afade=t=out:st={fade_start}:d={fade_out}") + if delay_ms > 0: + filters.append(f"adelay={delay_ms}|{delay_ms}") + + return filters + def execute(self, inputs: dict[str, Any]) -> ToolResult: operation = inputs["operation"] start = time.time() @@ -256,20 +289,7 @@ class AudioMixer(BaseTool): for i, track in enumerate(tracks): input_args.extend(["-i", track["path"]]) - volume = track.get("volume", 1.0) - delay_ms = int(track.get("start_seconds", 0) * 1000) - fade_in = track.get("fade_in_seconds", 0) - fade_out = track.get("fade_out_seconds", 0) - - filters = [] - if volume != 1.0: - filters.append(f"volume={volume}") - if delay_ms > 0: - filters.append(f"adelay={delay_ms}|{delay_ms}") - if fade_in > 0: - filters.append(f"afade=t=in:d={fade_in}") - if fade_out > 0: - filters.append(f"afade=t=out:d={fade_out}") + filters = self._track_filters(track) if filters: filter_chain = ",".join(filters) @@ -500,20 +520,7 @@ class AudioMixer(BaseTool): for i, track in enumerate(all_tracks): input_args.extend(["-i", track["path"]]) - volume = track.get("volume", 1.0) - delay_ms = int(track.get("start_seconds", 0) * 1000) - fade_in = track.get("fade_in_seconds", 0) - fade_out = track.get("fade_out_seconds", 0) - - filters = [] - if volume != 1.0: - filters.append(f"volume={volume}") - if delay_ms > 0: - filters.append(f"adelay={delay_ms}|{delay_ms}") - if fade_in > 0: - filters.append(f"afade=t=in:d={fade_in}") - if fade_out > 0: - filters.append(f"afade=t=out:d={fade_out}") + filters = self._track_filters(track) if filters: filter_chain = ",".join(filters)