From 9f7214dcd63e3fda143a238a093d87847d6ac31d Mon Sep 17 00:00:00 2001
From: Jeroen Schweitzer <jpmschweitzer@gmail.com>
Date: Wed, 18 Feb 2026 13:11:33 +0100
Subject: [PATCH] feat(db): add audio-batch processor and --post flag to
 audio-generate

Reduces prompt approvals from ~30 per batch to 2 (one manifest write,
one batch run). audio-generate --post chains trim+normalize+convert
after generation. audio-batch processes a JSON manifest of multiple
assets sequentially, supporting both SAO generation and harmonic
synthesis methods. Includes --dry-run, --only, and --skip-existing.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 db/connectors/audio-batch        |   4 +
 db/connectors/audio_batch.py     | 309 +++++++++++++++++++++++++++++++
 db/connectors/audio_connector.py |  60 +++++-
 3 files changed, 368 insertions(+), 5 deletions(-)
 create mode 100755 db/connectors/audio-batch
 create mode 100644 db/connectors/audio_batch.py

diff --git a/db/connectors/audio-batch b/db/connectors/audio-batch
new file mode 100755
index 000000000..ba8468a0c
--- /dev/null
+++ b/db/connectors/audio-batch
@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+# Batch audio generation from a manifest file. Whitelistable command.
+# Usage: audio-batch manifest.json [--dry-run] [--only ID,ID,...] [--skip-existing]
+exec python3 "$(dirname "$0")/audio_batch.py" "$@"
diff --git a/db/connectors/audio_batch.py b/db/connectors/audio_batch.py
new file mode 100644
index 000000000..419bce4f6
--- /dev/null
+++ b/db/connectors/audio_batch.py
@@ -0,0 +1,309 @@
+#!/usr/bin/env python3
+"""Batch audio generation from a manifest file.
+
+Processes multiple assets sequentially: SAO generation or harmonic synthesis,
+followed by post-processing (trim, normalize, convert to OGG).
+
+Usage:
+    python3 audio_batch.py manifest.json [--dry-run] [--only ID,ID,...] [--skip-existing]
+"""
+
+import json
+import math
+import os
+import struct
+import subprocess
+import sys
+import wave
+
+import numpy as np
+
+
+def load_manifest(path):
+    with open(path) as f:
+        return json.load(f)
+
+
+def resolve_paths(manifest, manifest_dir):
+    """Resolve output_dir and gen_dir relative to the git root."""
+    # Find git root by walking up from manifest_dir
+    # Check for .git as file (worktree) or directory (regular repo)
+    git_root = manifest_dir
+    while git_root != "/":
+        if os.path.exists(os.path.join(git_root, ".git")):
+            break
+        git_root = os.path.dirname(git_root)
+    else:
+        git_root = manifest_dir
+
+    output_dir = os.path.join(git_root, manifest.get("output_dir", "client/assets/audio"))
+    gen_dir = os.path.join(git_root, manifest.get("gen_dir", "client/assets/audio/gen"))
+    os.makedirs(output_dir, exist_ok=True)
+    os.makedirs(gen_dir, exist_ok=True)
+    return output_dir, gen_dir, git_root
+
+
+def get_default(manifest, asset, key):
+    """Get a value from the asset, falling back to manifest defaults."""
+    defaults = manifest.get("defaults", {})
+    return asset.get(key, defaults.get(key))
+
+
+def run_sao_generate(asset, manifest, gen_dir, output_dir, script_dir):
+    """Generate audio via Stable Audio Open + post-processing."""
+    filename = asset["filename"]
+    base_name = os.path.splitext(filename)[0]
+    wav_path = os.path.join(gen_dir, base_name + ".wav")
+    ogg_path = os.path.join(output_dir, filename)
+
+    prompt = asset["prompt"]
+    duration = asset.get("duration", 10)
+    steps = get_default(manifest, asset, "steps") or 100
+    cfg = get_default(manifest, asset, "cfg") or 7
+    timeout = get_default(manifest, asset, "timeout") or 600
+
+    # Run audio-generate with --post
+    cmd = [
+        sys.executable, os.path.join(script_dir, "audio_connector.py"),
+        "generate", prompt,
+        "--duration", str(duration),
+        "--steps", str(steps),
+        "--cfg", str(cfg),
+        "--output", wav_path,
+        "--output-ogg", ogg_path,
+        "--timeout", str(timeout),
+    ]
+
+    result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 60)
+    if result.returncode != 0:
+        stderr = result.stderr.strip()
+        try:
+            err = json.loads(result.stdout)
+            return {"ok": False, "error": err.get("error", stderr)}
+        except (json.JSONDecodeError, ValueError):
+            return {"ok": False, "error": stderr or "generation failed"}
+
+    try:
+        return json.loads(result.stdout)
+    except json.JSONDecodeError:
+        return {"ok": False, "error": f"Unexpected output: {result.stdout[:200]}"}
+
+
+def synthesize_harmonic(params, wav_path):
+    """Synthesize audio from harmonic parameters."""
+    sr = 44100
+    duration = params["duration"]
+    fundamental = params["fundamental"]
+    harmonics = params.get("harmonics", [])
+    attack_ms = params.get("attack_ms", 10)
+    sustain_ratio = params.get("sustain_ratio", 0.2)
+    decay = params.get("decay", "exponential")
+
+    n = int(sr * duration)
+    t = np.linspace(0, duration, n, endpoint=False)
+
+    # Fundamental
+    signal = np.sin(2 * np.pi * fundamental * t)
+
+    # Add harmonics
+    for h in harmonics:
+        freq = h["freq"]
+        db = h["db"]
+        amplitude = 10 ** (db / 20)
+        signal = signal + amplitude * np.sin(2 * np.pi * freq * t)
+
+    # Envelope: attack + sustain + decay
+    attack_s = attack_ms / 1000
+    attack_env = np.minimum(t / attack_s, 1.0) if attack_s > 0 else np.ones(n)
+
+    sustain_end = duration * sustain_ratio
+    if decay == "exponential":
+        # Decay rate: reach -60dB by end of duration
+        decay_rate = 6.9 / (duration - sustain_end) if duration > sustain_end else 10
+        decay_env = np.where(t < sustain_end, 1.0, np.exp(-decay_rate * (t - sustain_end)))
+    else:
+        # Linear decay
+        decay_env = np.where(t < sustain_end, 1.0,
+                             1.0 - (t - sustain_end) / (duration - sustain_end))
+
+    envelope = attack_env * decay_env
+    signal = signal * envelope
+
+    # Normalize to peak
+    peak = np.max(np.abs(signal))
+    if peak > 0:
+        signal = signal / peak * 0.9
+
+    # Write WAV
+    int_samples = np.clip(signal * 32767, -32767, 32767).astype(np.int16)
+    with wave.open(wav_path, "w") as f:
+        f.setnchannels(1)
+        f.setsampwidth(2)
+        f.setframerate(sr)
+        f.writeframes(int_samples.tobytes())
+
+    return wav_path
+
+
+def run_synth(asset, manifest, gen_dir, output_dir, script_dir):
+    """Synthesize audio from harmonic parameters + post-process."""
+    filename = asset["filename"]
+    base_name = os.path.splitext(filename)[0]
+    wav_path = os.path.join(gen_dir, base_name + "_synth.wav")
+    ogg_path = os.path.join(output_dir, filename)
+
+    synth_params = asset.get("synth")
+    if not synth_params:
+        return {"ok": False, "error": "No synth parameters provided"}
+
+    synth_type = synth_params.get("type", "harmonic")
+    if synth_type != "harmonic":
+        return {"ok": False, "error": f"Unknown synth type: {synth_type}"}
+
+    try:
+        synthesize_harmonic(synth_params, wav_path)
+    except Exception as e:
+        return {"ok": False, "error": f"Synthesis failed: {e}"}
+
+    # Post-process: normalize + convert (skip trim for synth — no silence to trim)
+    post_script = os.path.join(script_dir, "audio_post.py")
+    lufs = get_default(manifest, asset, "lufs") or -16
+    quality = get_default(manifest, asset, "quality") or 6
+
+    # Normalize
+    norm_path = os.path.join(gen_dir, base_name + "_norm.wav")
+    cmd = [sys.executable, post_script, "normalize", wav_path, "--output", norm_path,
+           "--lufs", str(lufs)]
+    result = subprocess.run(cmd, capture_output=True, text=True)
+    if result.returncode != 0:
+        return {"ok": False, "error": f"Normalize failed: {result.stderr.strip()}"}
+
+    # Convert to OGG
+    cmd = [sys.executable, post_script, "convert", norm_path, "--output", ogg_path,
+           "--quality", str(quality)]
+    result = subprocess.run(cmd, capture_output=True, text=True)
+    if result.returncode != 0:
+        return {"ok": False, "error": f"Convert failed: {result.stderr.strip()}"}
+
+    # Clean up intermediate
+    try:
+        os.remove(norm_path)
+    except OSError:
+        pass
+
+    ogg_size = os.path.getsize(ogg_path)
+    return {
+        "ok": True,
+        "file": wav_path,
+        "ogg_file": ogg_path,
+        "ogg_size_bytes": ogg_size,
+        "synth_params": synth_params,
+        "post_processed": True,
+    }
+
+
+def main():
+    if len(sys.argv) < 2:
+        print("Usage: audio_batch.py manifest.json [--dry-run] [--only ID,ID,...] [--skip-existing]",
+              file=sys.stderr)
+        sys.exit(1)
+
+    manifest_path = sys.argv[1]
+    dry_run = "--dry-run" in sys.argv
+    skip_existing = "--skip-existing" in sys.argv
+
+    only_ids = None
+    for i, arg in enumerate(sys.argv):
+        if arg == "--only" and i + 1 < len(sys.argv):
+            only_ids = set(sys.argv[i + 1].split(","))
+
+    manifest = load_manifest(manifest_path)
+    manifest_dir = os.path.dirname(os.path.abspath(manifest_path))
+    output_dir, gen_dir, git_root = resolve_paths(manifest, manifest_dir)
+    script_dir = os.path.dirname(os.path.abspath(__file__))
+
+    assets = manifest.get("assets", [])
+    if only_ids:
+        assets = [a for a in assets if a["id"] in only_ids]
+
+    # Health check if any SAO assets
+    sao_assets = [a for a in assets if a.get("method") == "sao"]
+    if sao_assets and not dry_run:
+        print(f"Checking SAO API health...", file=sys.stderr)
+        health_cmd = [sys.executable, os.path.join(script_dir, "audio_connector.py"), "health"]
+        result = subprocess.run(health_cmd, capture_output=True, text=True)
+        if result.returncode != 0:
+            print(json.dumps({"ok": False, "error": "SAO API health check failed",
+                              "details": result.stdout.strip()}))
+            sys.exit(1)
+        print(f"  SAO API is up.", file=sys.stderr)
+
+    total = len(assets)
+    results = []
+    success = 0
+    failed = 0
+    skipped = 0
+
+    print(f"Processing {total} assets from {os.path.basename(manifest_path)}...", file=sys.stderr)
+    if dry_run:
+        print("  (dry run — no generation will occur)", file=sys.stderr)
+
+    for i, asset in enumerate(assets, 1):
+        asset_id = asset["id"]
+        filename = asset["filename"]
+        method = asset.get("method", "sao")
+
+        print(f"\n[{i}/{total}] {asset_id}: {filename} ({method})", file=sys.stderr)
+
+        if skip_existing:
+            ogg_path = os.path.join(output_dir, filename)
+            if os.path.exists(ogg_path):
+                print(f"  Skipping — already exists", file=sys.stderr)
+                results.append({"id": asset_id, "status": "skipped", "reason": "exists"})
+                skipped += 1
+                continue
+
+        if dry_run:
+            print(f"  Would generate: {filename}", file=sys.stderr)
+            if method == "sao":
+                print(f"  Prompt: {asset.get('prompt', '(none)')[:80]}...", file=sys.stderr)
+            elif method == "synth":
+                synth = asset.get("synth", {})
+                print(f"  Synth: {synth.get('fundamental')}Hz, {synth.get('duration')}s",
+                      file=sys.stderr)
+            results.append({"id": asset_id, "status": "dry_run"})
+            continue
+
+        if method == "sao":
+            result = run_sao_generate(asset, manifest, gen_dir, output_dir, script_dir)
+        elif method == "synth":
+            result = run_synth(asset, manifest, gen_dir, output_dir, script_dir)
+        else:
+            result = {"ok": False, "error": f"Unknown method: {method}"}
+
+        result["id"] = asset_id
+        if result.get("ok"):
+            success += 1
+            result["status"] = "success"
+            print(f"  OK → {result.get('ogg_file', filename)}", file=sys.stderr)
+        else:
+            failed += 1
+            result["status"] = "failed"
+            print(f"  FAILED: {result.get('error', 'unknown')}", file=sys.stderr)
+
+        results.append(result)
+
+    # Summary
+    summary = {
+        "ok": failed == 0,
+        "total": total,
+        "success": success,
+        "failed": failed,
+        "skipped": skipped,
+        "results": results,
+    }
+    print(json.dumps(summary, indent=2))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/db/connectors/audio_connector.py b/db/connectors/audio_connector.py
index 1323a8e70..5e01260ca 100644
--- a/db/connectors/audio_connector.py
+++ b/db/connectors/audio_connector.py
@@ -6,12 +6,13 @@ Talks to the Stable Audio Open Gradio app at tower-of-joy:11500.
 Uses the async Gradio API pattern: POST to submit, SSE stream for results.
 
 Usage:
-    python3 audio_connector.py generate "prompt text" [--duration 10] [--steps 100] [--cfg 7] [--output file.wav]
+    python3 audio_connector.py generate "prompt text" [--duration 10] [--steps 100] [--cfg 7] [--output file.wav] [--post]
     python3 audio_connector.py health
 """
 
 import json
 import os
+import subprocess
 import sys
 import time
 import urllib.error
@@ -56,7 +57,31 @@ def health():
         }, indent=2))
         sys.exit(1)
 
-def generate(prompt, duration=10.0, steps=100, cfg=7.0, output=None, timeout=600):
+def post_process(wav_path, ogg_path=None, lufs=-16, quality=6, threshold=-50):
+    """Run trim + normalize + convert on a WAV file via audio-post pipeline."""
+    script_dir = os.path.dirname(os.path.abspath(__file__))
+    post_script = os.path.join(script_dir, "audio_post.py")
+    if ogg_path is None:
+        ogg_path = os.path.splitext(wav_path)[0] + ".ogg"
+    cmd = [
+        sys.executable, post_script, "pipeline", wav_path,
+        "--output", ogg_path,
+        "--lufs", str(lufs),
+        "--quality", str(quality),
+        "--threshold", str(threshold),
+    ]
+    print(f"  Post-processing → {os.path.basename(ogg_path)}...", file=sys.stderr)
+    result = subprocess.run(cmd, capture_output=True, text=True)
+    if result.returncode != 0:
+        return {"ok": False, "error": f"Post-processing failed: {result.stderr.strip()}"}
+    try:
+        return json.loads(result.stdout)
+    except json.JSONDecodeError:
+        return {"ok": True, "output": ogg_path}
+
+
+def generate(prompt, duration=10.0, steps=100, cfg=7.0, output=None, timeout=600,
+             post=False, output_ogg=None):
     """
     Generate audio from a text prompt.
 
@@ -67,6 +92,8 @@ def generate(prompt, duration=10.0, steps=100, cfg=7.0, output=None, timeout=600
         cfg: Classifier-free guidance scale (default 7)
         output: Output file path (default: auto-named in current directory)
         timeout: Maximum wait time in seconds (default 600 = 10 minutes)
+        post: If True, run trim+normalize+convert after generation
+        output_ogg: OGG output path when post=True (default: same basename .ogg)
     """
     base = get_base_url()
     api_url = f"{base}/gradio_api/call/generate_audio"
@@ -216,7 +243,7 @@ def generate(prompt, duration=10.0, steps=100, cfg=7.0, output=None, timeout=600
                 shutil.copyfileobj(resp, f)
 
         file_size = os.path.getsize(output)
-        print(json.dumps({
+        result_json = {
             "ok": True,
             "file": output,
             "size_bytes": file_size,
@@ -225,7 +252,20 @@ def generate(prompt, duration=10.0, steps=100, cfg=7.0, output=None, timeout=600
             "cfg": cfg,
             "prompt": prompt,
             "generation_time_s": elapsed
-        }, indent=2))
+        }
+
+        if post:
+            post_result = post_process(output, ogg_path=output_ogg)
+            if not post_result.get("ok"):
+                result_json["post_processed"] = False
+                result_json["post_error"] = post_result.get("error", "unknown")
+            else:
+                result_json["post_processed"] = True
+                result_json["ogg_file"] = post_result.get("output", output_ogg)
+                ogg_size = os.path.getsize(result_json["ogg_file"])
+                result_json["ogg_size_bytes"] = ogg_size
+
+        print(json.dumps(result_json, indent=2))
 
     except Exception as e:
         print(json.dumps({
@@ -258,6 +298,8 @@ def main():
         cfg = 7.0
         output = None
         timeout = 600
+        post = False
+        output_ogg = None
 
         # Parse optional args
         i = 3
@@ -277,11 +319,19 @@ def main():
             elif sys.argv[i] == "--timeout" and i + 1 < len(sys.argv):
                 timeout = int(sys.argv[i + 1])
                 i += 2
+            elif sys.argv[i] == "--post":
+                post = True
+                i += 1
+            elif sys.argv[i] == "--output-ogg" and i + 1 < len(sys.argv):
+                output_ogg = sys.argv[i + 1]
+                post = True  # --output-ogg implies --post
+                i += 2
             else:
                 print(f"Unknown argument: {sys.argv[i]}", file=sys.stderr)
                 sys.exit(1)
 
-        generate(prompt, duration=duration, steps=steps, cfg=cfg, output=output, timeout=timeout)
+        generate(prompt, duration=duration, steps=steps, cfg=cfg, output=output,
+                 timeout=timeout, post=post, output_ogg=output_ogg)
     else:
         print(f"Unknown command: {cmd}", file=sys.stderr)
         sys.exit(1)