SonicForgeStudio/tests/test_builtin_fx_golden.py

# -*- coding: utf-8 -*-
"""Golden test PLAN_DAW_A.md 2.9: C++ BuiltinFxChain vs Python
_apply_builtin_fx_chain (render_engine.py) — SNR per-chain, kể cả order test.

Skip nếu thiếu fx_vst_bridge.exe (build/Release) hoặc scipy.
"""
import io, json, math, os, subprocess, sys, tempfile, wave

import numpy as np
import pytest

REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, REPO)

from app.core.render_engine import _apply_builtin_fx_chain  # noqa: E402

BRIDGE_CANDIDATES = [
    os.environ.get("SF_FX_BRIDGE_PATH", ""),
    os.path.join(REPO, "native_bridge", "build", "Release", "fx_vst_bridge.exe"),
    os.path.join(REPO, "install", "fx_vst_bridge.exe"),
]


def _find_bridge():
    for c in BRIDGE_CANDIDATES:
        if c and os.path.isfile(c):
            return c
    return None


@pytest.fixture(scope="module")
def bridge():
    b = _find_bridge()
    if not b:
        pytest.skip("fx_vst_bridge.exe not found (native_bridge/build/Release)")
    return b


def _write_wav(path, data, sr):
    data = np.clip(data, -1.0, 1.0)
    # data shape (2, n) → interleave (n, 2) trước khi ghi
    pcm = (data.T * 32767).astype(np.int16)
    with wave.open(path, "wb") as w:
        w.setnchannels(2)
        w.setsampwidth(2)
        w.setframerate(sr)
        w.writeframes(pcm.tobytes())


def _read_wav(path):
    import soundfile as sf
    data, sr = sf.read(path, dtype="float32", always_2d=True)
    return data.T, sr


def _render_bridge(bridge, wav_in, sr, chain):
    job = {"sample_rate": sr, "block_size": 512, "fx_chain": chain}
    with tempfile.TemporaryDirectory() as td:
        job_path = os.path.join(td, "job.json")
        out_path = os.path.join(td, "out.wav")
        with io.open(job_path, "w", encoding="utf-8") as f:
            json.dump(job, f)
        r = subprocess.run([bridge, "--render-fx", job_path, "--in", wav_in,
                            "--out", out_path],
                           capture_output=True, text=True, timeout=120)
        assert r.returncode == 0, f"bridge rc={r.returncode}: {r.stdout} {r.stderr}"
        assert os.path.isfile(out_path), f"no out wav: {r.stdout} {r.stderr}"
        return _read_wav(out_path)[0]


def _snr_db(ref, got):
    ref = ref.astype(np.float64)
    got = got.astype(np.float64)
    if ref.shape != got.shape:
        raise AssertionError(f"shape mismatch {ref.shape} vs {got.shape}")
    n = min(ref.shape[1], got.shape[1])
    if n == 0:
        return math.inf
    ref, got = ref[:, :n], got[:, :n]
    denom = float(np.sum((ref - got) ** 2))
    if denom < 1e-12:
        return math.inf
    return 10.0 * math.log10(float(np.sum(ref ** 2)) / denom)


def _run_case(bridge, sr, seed, chain):
    rng = np.random.default_rng(seed)
    n = sr // 4
    t = np.arange(n) / sr
    sig = (0.25 * np.sin(2 * np.pi * 220 * t)
           + 0.12 * np.sin(2 * np.pi * 1100 * t)
           + 0.06 * np.sin(2 * np.pi * 5000 * t))
    noise = 0.01 * rng.standard_normal(n)
    audio = np.stack([sig + noise, 0.9 * sig + 0.02 * rng.standard_normal(n)])
    audio = audio.astype(np.float32)

    with tempfile.TemporaryDirectory() as td:
        wav_in = os.path.join(td, "in.wav")
        _write_wav(wav_in, audio, sr)
        got = _render_bridge(bridge, wav_in, sr, chain)

    # Python reference dùng schema cũ {type: eq} — convert từ job schema mới
    # {type: builtin, id: eq} (bridge nhận schema mới, spec Phase 2).
    py_chain = [{"type": s.get("id", s.get("type")), "params": s.get("params") or {}}
                for s in chain]
    ref = _apply_builtin_fx_chain(audio.copy(), py_chain, sr)
    snr = _snr_db(ref, got)
    return snr


def _b(id, params):
    return {"type": "builtin", "id": id, "params": params or {}}


CHAINS = {
    "eq": [_b("eq", {"g1": 3.0, "g2": -2.0, "g3": 1.5, "g4": 2.0})],
    "eqpro": [_b("eqpro", {"amount": 80, "bands": [
        {"active": True, "type": "lowshelf", "freq": 120, "gain": 2.5, "q": 0.8},
        {"active": True, "type": "peaking", "freq": 900, "gain": -3.0, "q": 1.2},
        {"active": True, "type": "highpass", "freq": 60, "gain": 0.0, "q": 0.7},
        {"active": True, "type": "highshelf", "freq": 8000, "gain": 1.8, "q": 0.7},
    ]})],
    "compressor": [_b("compressor", {"threshold": -20.0, "ratio": 4.0, "makeup": 2.0})],
    "limiter": [_b("limiter", {"ceiling": -3.0})],
    "exciter": [_b("exciter", {"drive": 60.0})],
    "rebalance": [_b("rebalance", {"mid": 2.0, "side": -1.5})],
    "order_ec": [_b("eq", {"g1": 2.0, "g3": -1.0}),
                 _b("compressor", {"threshold": -18.0, "ratio": 3.0})],
    "order_ce": [_b("compressor", {"threshold": -18.0, "ratio": 3.0}),
                 _b("eq", {"g1": 2.0, "g3": -1.0})],
}

MIN_SNR = 30.0


@pytest.mark.parametrize("name", sorted(CHAINS))
def test_builtin_golden(bridge, name):
    snr = _run_case(bridge, 44100, seed=1234, chain=CHAINS[name])
    assert snr >= MIN_SNR, f"{name}: SNR {snr:.1f} dB < {MIN_SNR}"