Add additive audio mixing for video SFX

app/operations/mix_audio.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import json
+from collections.abc import Sequence
+from pathlib import Path
+from typing import Any
+
+from app.core.exceptions import InputError
+from app.models.media import InputMedia, OperationResult
+from app.operations.common import execute, output_path, require_inputs
+from app.services.ffmpeg_service import FFmpegService
+from app.services.validator import bounded_number
+
+
+def _track_params(params: dict[str, Any], count: int) -> list[dict[str, float]]:
+    raw = params.get("audio_tracks")
+    if raw in (None, ""):
+        return [
+            {
+                "start": bounded_number(params, "start", 0, 0, 86_400),
+                "volume": bounded_number(params, "volume", 1, 0, 10),
+            }
+            for _ in range(count)
+        ]
+
+    if isinstance(raw, str):
+        try:
+            raw = json.loads(raw)
+        except json.JSONDecodeError as exc:
+            raise InputError("audio_tracks must be valid JSON") from exc
+
+    if not isinstance(raw, list) or len(raw) != count:
+        raise InputError("audio_tracks must contain one object for each additional audio input")
+
+    tracks: list[dict[str, float]] = []
+    for index, item in enumerate(raw, start=1):
+        if not isinstance(item, dict):
+            raise InputError(f"audio_tracks[{index - 1}] must be an object")
+        try:
+            start = bounded_number(item, "start", 0, 0, 86_400)
+            volume = bounded_number(item, "volume", 1, 0, 10)
+        except InputError as exc:
+            raise InputError(f"Invalid audio_tracks[{index - 1}] configuration") from exc
+        tracks.append({"start": start, "volume": volume})
+    return tracks
+
+
+async def add_audio(
+    ffmpeg: FFmpegService,
+    inputs: Sequence[InputMedia],
+    params: dict[str, Any],
+    output_dir: Path,
+) -> OperationResult:
+    require_inputs(inputs, 2)
+    tracks = _track_params(params, len(inputs) - 1)
+    output = output_path(output_dir, "audio-mix", "mp4")
+
+    filters: list[str] = []
+    mix_inputs = ["[0:a]"]
+    for offset, track in enumerate(tracks, start=1):
+        delay_ms = round(track["start"] * 1000)
+        filters.append(
+            f"[{offset}:a]volume={track['volume']:.6g},adelay={delay_ms}:all=1[a{offset}]"
+        )
+        mix_inputs.append(f"[a{offset}]")
+
+    filters.append(
+        "".join(mix_inputs)
+        + f"amix=inputs={len(mix_inputs)}:duration=first:dropout_transition=0:normalize=0[mixed]"
+    )
+
+    args: list[str | Path] = []
+    for media in inputs:
+        args.extend(["-i", media.temp_path])
+    args.extend(
+        [
+            "-filter_complex",
+            ";".join(filters),
+            "-map",
+            "0:v:0",
+            "-map",
+            "[mixed]",
+            "-c:v",
+            "copy",
+            "-c:a",
+            "aac",
+            "-b:a",
+            str(params.get("audio_bitrate", "192k")),
+            "-movflags",
+            "+faststart",
+            output,
+        ]
+    )
+    return await execute(
+        ffmpeg,
+        args,
+        output,
+        "video.add_audio",
+        {"audio_tracks": tracks, "preserve_video": True, "mix_mode": "additive"},
+    )