T13: NativeMixer mirror server pipeline (track gain/pan constant-power + brickwall limiter tanh, wired VST2/VST3 bridge, golden test 0.0001dB)

2026-08-11 16:53:15 +07:00
parent 8f1e89ce50
commit b9817e1aed
7 changed files with 404 additions and 3 deletions
@@ -0,0 +1,116 @@
+"""Golden test T13 — NativeMixer mirror server render pipeline.
+
+So sanh output native mixer (native_mixer_test.exe) voi reference dung chinh
+code path cua server `render_project`:
+- track gain/pan: cong thuc render_engine.py (10^(dB/20), constant-power pan)
+- master limiter: app.core.mastering_engine.apply_mastering (module limiter)
+  + hard clip [-1,1] nhu render_project sau apply_mastering
+
+Yeu cau: |RMS| va |peak| sai lech < 0.1 dB moi case, moi kenh.
+
+Chay:  python native_host/tests/test_native_mixer_golden.py
+(tu repo root; can native_mixer_test.exe da build + numpy/scipy)
+"""
+import os
+import subprocess
+import sys
+
+import numpy as np
+
+ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+EXE = os.path.join(ROOT, "native_host", "tests", "native_mixer_test.exe")
+SR = 44100
+
+sys.path.insert(0, ROOT)
+from app.core.mastering_engine import apply_mastering  # noqa: E402
+
+
+def make_signal(n=SR):
+    t = np.arange(n) / SR
+    sig = (
+        np.sin(2 * np.pi * 220 * t)
+        + 0.5 * np.sin(2 * np.pi * 440 * t + 0.3)
+        + 0.3 * np.sin(2 * np.pi * 880 * t + 1.1)
+    )
+    env = np.minimum(1.0, t * 20.0) * np.exp(-t * 0.8)
+    sig *= env
+    rng = np.random.default_rng(1234)
+    sig = sig + rng.standard_normal(n) * 0.05
+    # spike de limiter clip ro rang
+    for i in range(0, n, 22050):
+        sig[i] += 3.0
+    return np.stack([sig, np.roll(sig, 100)])
+
+
+def ref_process(x, gain_db, pan, lim_active, th_db):
+    y = x * (10.0 ** (gain_db / 20.0))
+    if pan != 0.0:
+        theta = ((np.clip(pan, -1.0, 1.0) + 1.0) / 2.0) * (np.pi / 2.0)
+        y = y.copy()
+        y[0, :] *= np.cos(theta)
+        y[1, :] *= np.sin(theta)
+    if lim_active:
+        settings = {
+            "masterConnected": True,
+            "isBypassed": False,
+            "chain": [{"type": "limiter", "active": True}],
+            "limActive": True,
+            "limThreshold": th_db,
+        }
+        y = apply_mastering(y, settings, SR)
+        np.clip(y, -1.0, 1.0, out=y)
+    return y
+
+
+def run_native(x, gain_db, pan, lim_active, th_db, tmp):
+    inp = os.path.join(tmp, "in.raw")
+    outp = os.path.join(tmp, "out.raw")
+    with open(inp, "wb") as f:
+        f.write(x.T.astype(np.float32).tobytes())
+    args = [EXE, inp, outp, str(gain_db), str(pan), str(int(lim_active)), str(th_db)]
+    subprocess.run(args, check=True)
+    return np.fromfile(outp, dtype=np.float32).reshape(-1, 2).T
+
+
+def db_rms_peak(y):
+    rms = 20.0 * np.log10(np.sqrt(np.mean(y * y, axis=1)) + 1e-12)
+    peak = 20.0 * np.log10(np.max(np.abs(y), axis=1) + 1e-12)
+    return rms, peak
+
+
+def main():
+    if not os.path.exists(EXE):
+        print(f"FAIL: {EXE} not found (build native_mixer_test.exe truoc)")
+        return 1
+    import tempfile
+
+    x = make_signal()
+    cases = [
+        (0.0, 0.0, False, -1.0),   # identity (limiter off)
+        (6.0, -0.5, False, -1.0),  # gain/pan, khong limiter
+        (6.0, -0.5, True, -3.0),
+        (-9.0, 0.8, True, -6.0),
+        (12.0, 0.0, True, 0.0),
+    ]
+    worst = 0.0
+    with tempfile.TemporaryDirectory() as tmp:
+        for i, (g, p, la, th) in enumerate(cases):
+            native = run_native(x, g, p, la, th, tmp)
+            ref = ref_process(x, g, p, la, th)
+            nr, npk = db_rms_peak(native)
+            rr, rpk = db_rms_peak(ref)
+            dr = np.max(np.abs(nr - rr))
+            dp = np.max(np.abs(npk - rpk))
+            worst = max(worst, dr, dp)
+            status = "OK " if (dr < 0.1 and dp < 0.1) else "FAIL"
+            print(f"case {i} gain={g} pan={p} lim={int(la)} th={th}: "
+                  f"RMS diff={dr:.4f} dB peak diff={dp:.4f} dB [{status}]")
+            if dr >= 0.1 or dp >= 0.1:
+                print(f"  native rms={nr} peak={npk}")
+                print(f"  ref    rms={rr} peak={rpk}")
+    print(f"worst diff: {worst:.4f} dB (< 0.1 -> PASS)")
+    return 0 if worst < 0.1 else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())