117 lines
3.9 KiB
Python
117 lines
3.9 KiB
Python
"""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())
|