T14: SF host bridge native (FluidSynth runtime DLL) + golden test
This commit is contained in:
@@ -0,0 +1,93 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Test smoke SFHost bridge (T14): FluidSynth native qua ctypes.
|
||||
|
||||
Chạy: python tests/test_sf_host_bridge.py [path_sf2]
|
||||
Yêu cầu: native_host/build/Release/sf_host_bridge.dll + fluidsynth_runtime/.
|
||||
Verify: Create → LoadSF2 → SelectInstrument → AudioStart → NoteOn → chờ
|
||||
(blocks tăng, underruns 0) → NoteOff → AudioStop → Close; mixer set OK.
|
||||
"""
|
||||
import ctypes
|
||||
import ctypes.wintypes as wt
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
|
||||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
DLL = os.path.join(ROOT, "build", "Release", "sf_host_bridge.dll")
|
||||
SF2 = sys.argv[1] if len(sys.argv) > 1 else os.path.join(
|
||||
ROOT, "..", "app", "storage", "soundfonts",
|
||||
"518e850f-a5d3-4790-b1f9-0c90c203c524.sf2")
|
||||
|
||||
assert os.path.exists(DLL), f"thiếu {DLL}"
|
||||
assert os.path.exists(SF2), f"thiếu {SF2}"
|
||||
|
||||
dll = ctypes.WinDLL(DLL)
|
||||
err = ctypes.create_string_buffer(256)
|
||||
|
||||
dll.SF_FS_Create.argtypes = [ctypes.c_int32, ctypes.c_int32, ctypes.c_char_p, ctypes.c_int32]
|
||||
dll.SF_FS_Create.restype = ctypes.c_int32
|
||||
dll.SF_FS_LoadSF2.argtypes = [ctypes.c_int32, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_int32]
|
||||
dll.SF_FS_LoadSF2.restype = ctypes.c_int32
|
||||
dll.SF_FS_SelectInstrument.argtypes = [ctypes.c_int32, ctypes.c_int32, ctypes.c_int32, ctypes.c_int32, ctypes.c_int32]
|
||||
dll.SF_FS_SelectInstrument.restype = ctypes.c_int32
|
||||
dll.SF_FS_NoteOn.argtypes = [ctypes.c_int32, ctypes.c_int32, ctypes.c_int32, ctypes.c_int32]
|
||||
dll.SF_FS_NoteOn.restype = ctypes.c_int32
|
||||
dll.SF_FS_NoteOff.argtypes = [ctypes.c_int32, ctypes.c_int32, ctypes.c_int32]
|
||||
dll.SF_FS_NoteOff.restype = ctypes.c_int32
|
||||
dll.SF_FS_AudioStart.argtypes = [ctypes.c_int32, ctypes.c_int32, ctypes.c_int32, ctypes.c_char_p, ctypes.c_int32]
|
||||
dll.SF_FS_AudioStart.restype = ctypes.c_int32
|
||||
dll.SF_FS_AudioStop.argtypes = [ctypes.c_int32]
|
||||
dll.SF_FS_AudioStop.restype = ctypes.c_int32
|
||||
dll.SF_FS_AudioUnderruns.argtypes = [ctypes.c_int32]
|
||||
dll.SF_FS_AudioUnderruns.restype = ctypes.c_int32
|
||||
dll.SF_FS_AudioBlocks.argtypes = [ctypes.c_int32]
|
||||
dll.SF_FS_AudioBlocks.restype = ctypes.c_int32
|
||||
dll.SF_FS_SetTrackGainPan.argtypes = [ctypes.c_int32, ctypes.c_float, ctypes.c_float]
|
||||
dll.SF_FS_SetTrackGainPan.restype = ctypes.c_int32
|
||||
dll.SF_FS_SetMasterLimiter.argtypes = [ctypes.c_int32, ctypes.c_int32, ctypes.c_float]
|
||||
dll.SF_FS_SetMasterLimiter.restype = ctypes.c_int32
|
||||
dll.SF_FS_Close.argtypes = [ctypes.c_int32, ctypes.c_char_p, ctypes.c_int32]
|
||||
dll.SF_FS_Close.restype = ctypes.c_int32
|
||||
|
||||
|
||||
def check(name, rc, expect=0):
|
||||
assert rc == expect, f"{name}: rc={rc} err={err.value.decode() if err.value else ''}"
|
||||
print(f"OK {name}")
|
||||
|
||||
|
||||
h = dll.SF_FS_Create(44100, 512, err, 256)
|
||||
assert h > 0, f"Create trả {h}, err={err.value.decode() if err.value else ''}"
|
||||
print(f"OK SF_FS_Create -> handle {h}")
|
||||
|
||||
check("SF_FS_LoadSF2", dll.SF_FS_LoadSF2(h, SF2.encode(), err, 256))
|
||||
|
||||
# Tìm preset đầu tiên tồn tại trong SF2 (SF2 tuỳ biến có thể không có 0/0).
|
||||
sel = None
|
||||
for bank in (0, 1, 128):
|
||||
for prog in (0, 1, 12, 40, 80):
|
||||
rc = dll.SF_FS_SelectInstrument(h, 0, -1, bank, prog)
|
||||
if rc == 0:
|
||||
sel = (bank, prog)
|
||||
break
|
||||
if sel:
|
||||
break
|
||||
assert sel, "không tìm được preset nào trong SF2"
|
||||
print(f"OK SF_FS_SelectInstrument bank={sel[0]} program={sel[1]}")
|
||||
check("SF_FS_SetTrackGainPan", dll.SF_FS_SetTrackGainPan(h, -6.0, 0.0))
|
||||
check("SF_FS_SetMasterLimiter", dll.SF_FS_SetMasterLimiter(h, 1, -1.0))
|
||||
|
||||
check("SF_FS_AudioStart", dll.SF_FS_AudioStart(h, 44100, 512, err, 256))
|
||||
b0 = dll.SF_FS_AudioBlocks(h)
|
||||
check("SF_FS_NoteOn", dll.SF_FS_NoteOn(h, 0, 60, 100))
|
||||
time.sleep(0.4)
|
||||
b1 = dll.SF_FS_AudioBlocks(h)
|
||||
u = dll.SF_FS_AudioUnderruns(h)
|
||||
assert b1 > b0, f"blocks không tăng: {b0} -> {b1}"
|
||||
assert u == 0, f"underruns={u}"
|
||||
print(f"OK render blocks {b0} -> {b1}, underruns {u}")
|
||||
|
||||
check("SF_FS_NoteOff", dll.SF_FS_NoteOff(h, 0, 60))
|
||||
time.sleep(0.2)
|
||||
check("SF_FS_AudioStop", dll.SF_FS_AudioStop(h))
|
||||
check("SF_FS_Close", dll.SF_FS_Close(h, err, 256))
|
||||
print("PASS")
|
||||
@@ -0,0 +1,187 @@
|
||||
"""Golden test T14 — SFHost (FluidSynth native) mirror server render pipeline.
|
||||
|
||||
So sanh output SFHost.RenderBlock voi reference dung chinh code path server:
|
||||
- FluidSynth: libfluidsynth-3.dll (write_float) — engine giong render_engine.py
|
||||
(pyfluidsynth cung wrapper len cung lib)
|
||||
- track gain/pan: cong thuc render_engine.py (10^(dB/20), constant-power pan)
|
||||
- master limiter: app.core.mastering_engine.apply_mastering + clip [-1,1]
|
||||
|
||||
Yeu cau: |RMS| va |peak| sai lech < 0.1 dB moi case, moi kenh.
|
||||
|
||||
Chay: python native_host/tests/test_sf_host_golden.py [path_sf2]
|
||||
(tu repo root; can build/Release/sf_host_bridge.dll + fluidsynth_runtime/)
|
||||
"""
|
||||
import ctypes
|
||||
import os
|
||||
import sys
|
||||
|
||||
import numpy as np
|
||||
|
||||
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
NATIVE = os.path.join(ROOT, "native_host")
|
||||
DLL = os.path.join(NATIVE, "build", "Release", "sf_host_bridge.dll")
|
||||
RUNTIME = os.path.join(NATIVE, "fluidsynth_runtime")
|
||||
FSDLL = os.path.join(RUNTIME, "libfluidsynth-3.dll")
|
||||
SF2 = sys.argv[1] if len(sys.argv) > 1 else os.path.join(
|
||||
ROOT, "app", "storage", "soundfonts",
|
||||
"518e850f-a5d3-4790-b1f9-0c90c203c524.sf2")
|
||||
SR = 44100
|
||||
|
||||
assert os.path.exists(DLL), f"thiếu {DLL} (build cmake Release truoc)"
|
||||
assert os.path.exists(FSDLL), f"thiếu {FSDLL} (chay scripts/dl_fluidsynth_runtime.py)"
|
||||
assert os.path.exists(SF2), f"thiếu {SF2}"
|
||||
|
||||
sys.path.insert(0, ROOT)
|
||||
from app.core.mastering_engine import apply_mastering # noqa: E402
|
||||
|
||||
# preset ton tai trong SF2 (xem phdr)
|
||||
BANK, PROG = 128, 0
|
||||
NOTE, VEL, CH = 60, 100, 0
|
||||
FRAMES = 512
|
||||
BLOCKS = 4 # render 4 blocks sau noteon, so sanh block 1 (sau transient)
|
||||
|
||||
err = ctypes.create_string_buffer(256)
|
||||
|
||||
|
||||
def load_bridge():
|
||||
dll = ctypes.WinDLL(DLL)
|
||||
dll.SF_FS_Create.argtypes = [ctypes.c_int32, ctypes.c_int32, ctypes.c_char_p, ctypes.c_int32]
|
||||
dll.SF_FS_Create.restype = ctypes.c_int32
|
||||
dll.SF_FS_LoadSF2.argtypes = [ctypes.c_int32, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_int32]
|
||||
dll.SF_FS_LoadSF2.restype = ctypes.c_int32
|
||||
dll.SF_FS_SelectInstrument.argtypes = [ctypes.c_int32, ctypes.c_int32, ctypes.c_int32, ctypes.c_int32, ctypes.c_int32]
|
||||
dll.SF_FS_SelectInstrument.restype = ctypes.c_int32
|
||||
dll.SF_FS_NoteOn.argtypes = [ctypes.c_int32, ctypes.c_int32, ctypes.c_int32, ctypes.c_int32]
|
||||
dll.SF_FS_NoteOn.restype = ctypes.c_int32
|
||||
dll.SF_FS_RenderBlock.argtypes = [ctypes.c_int32, ctypes.c_int32,
|
||||
np.ctypeslib.ndpointer(np.float32, flags="C_CONTIGUOUS"),
|
||||
np.ctypeslib.ndpointer(np.float32, flags="C_CONTIGUOUS")]
|
||||
dll.SF_FS_RenderBlock.restype = ctypes.c_int32
|
||||
dll.SF_FS_SetTrackGainPan.argtypes = [ctypes.c_int32, ctypes.c_float, ctypes.c_float]
|
||||
dll.SF_FS_SetTrackGainPan.restype = ctypes.c_int32
|
||||
dll.SF_FS_SetMasterLimiter.argtypes = [ctypes.c_int32, ctypes.c_int32, ctypes.c_float]
|
||||
dll.SF_FS_SetMasterLimiter.restype = ctypes.c_int32
|
||||
dll.SF_FS_Close.argtypes = [ctypes.c_int32, ctypes.c_char_p, ctypes.c_int32]
|
||||
dll.SF_FS_Close.restype = ctypes.c_int32
|
||||
return dll
|
||||
|
||||
|
||||
def ref_synth():
|
||||
"""FluidSynth reference qua ctypes — doc lap voi bridge."""
|
||||
os.environ["PATH"] = RUNTIME + ";" + os.environ.get("PATH", "")
|
||||
fs = ctypes.WinDLL(FSDLL)
|
||||
fs.new_fluid_settings.restype = ctypes.c_void_p
|
||||
fs.fluid_settings_setnum.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_double]
|
||||
fs.fluid_settings_setnum.restype = ctypes.c_int
|
||||
fs.fluid_settings_setstr.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p]
|
||||
fs.fluid_settings_setstr.restype = ctypes.c_int
|
||||
fs.new_fluid_synth.argtypes = [ctypes.c_void_p]
|
||||
fs.new_fluid_synth.restype = ctypes.c_void_p
|
||||
fs.fluid_synth_sfload.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_int]
|
||||
fs.fluid_synth_sfload.restype = ctypes.c_int
|
||||
fs.fluid_synth_program_select.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_int]
|
||||
fs.fluid_synth_program_select.restype = ctypes.c_int
|
||||
fs.fluid_synth_noteon.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_int, ctypes.c_int]
|
||||
fs.fluid_synth_noteon.restype = ctypes.c_int
|
||||
fs.fluid_synth_set_gain.argtypes = [ctypes.c_void_p, ctypes.c_float]
|
||||
fs.fluid_synth_set_gain.restype = ctypes.c_int
|
||||
fs.fluid_synth_write_float.argtypes = [ctypes.c_void_p, ctypes.c_int,
|
||||
np.ctypeslib.ndpointer(np.float32, flags="C_CONTIGUOUS"),
|
||||
ctypes.c_int, ctypes.c_int,
|
||||
np.ctypeslib.ndpointer(np.float32, flags="C_CONTIGUOUS"),
|
||||
ctypes.c_int, ctypes.c_int]
|
||||
settings = fs.new_fluid_settings()
|
||||
fs.fluid_settings_setnum(settings, b"synth.sample-rate", SR)
|
||||
fs.fluid_settings_setstr(settings, b"audio.driver", b"file")
|
||||
synth = fs.new_fluid_synth(settings)
|
||||
# Bridge + WASM client deu dung gain 1.0; FluidSynth default la 0.2 (-14 dB).
|
||||
assert fs.fluid_synth_set_gain(synth, 1.0) == 0, "set_gain failed"
|
||||
fid = fs.fluid_synth_sfload(synth, SF2.encode(), 0)
|
||||
assert fid >= 0, "reference sfload failed"
|
||||
assert fs.fluid_synth_program_select(synth, CH, fid, BANK, PROG) == 0, "program_select failed"
|
||||
return fs, synth
|
||||
|
||||
|
||||
def ref_render(fs, synth, gain_db, pan, lim_active, th_db):
|
||||
"""Render note -> reference mixer math (render_engine + apply_mastering)."""
|
||||
fs.fluid_synth_noteon(synth, CH, NOTE, VEL)
|
||||
for _ in range(BLOCKS):
|
||||
l = np.zeros(FRAMES, np.float32)
|
||||
r = np.zeros(FRAMES, np.float32)
|
||||
fs.fluid_synth_write_float(synth, FRAMES, l, 0, 1, r, 0, 1)
|
||||
y = np.stack([l, r]) # [2, FRAMES]
|
||||
|
||||
y = y * (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 bridge_render(dll, h, gain_db, pan, lim_active, th_db):
|
||||
dll.SF_FS_SetTrackGainPan(h, ctypes.c_float(gain_db), ctypes.c_float(pan))
|
||||
dll.SF_FS_SetMasterLimiter(h, int(lim_active), ctypes.c_float(th_db))
|
||||
assert dll.SF_FS_NoteOn(h, CH, NOTE, VEL) == 0
|
||||
for _ in range(BLOCKS):
|
||||
l = np.zeros(FRAMES, np.float32)
|
||||
r = np.zeros(FRAMES, np.float32)
|
||||
assert dll.SF_FS_RenderBlock(h, FRAMES, l, r) == 0
|
||||
return np.stack([l, r])
|
||||
|
||||
|
||||
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():
|
||||
dll = load_bridge()
|
||||
fs, synth = ref_synth()
|
||||
|
||||
cases = [
|
||||
dict(gain_db=0.0, pan=0.0, lim_active=False, th_db=-1.0),
|
||||
dict(gain_db=-6.0, pan=0.0, lim_active=True, th_db=-1.0),
|
||||
dict(gain_db=-3.0, pan=0.8, lim_active=True, th_db=-6.0),
|
||||
]
|
||||
|
||||
for c in cases:
|
||||
# bridge moi instance moi case (synth state doc lap)
|
||||
h = dll.SF_FS_Create(SR, FRAMES, err, 256)
|
||||
assert h > 0, f"Create failed {err.value}"
|
||||
assert dll.SF_FS_LoadSF2(h, SF2.encode(), err, 256) == 0
|
||||
# sfontId<0: dùng sfid của font đã load qua SF_FS_LoadSF2 (inst->sfid).
|
||||
# Không truyền cứng 0 — sfid thật của SF2 này là 1, không phải 0.
|
||||
assert dll.SF_FS_SelectInstrument(h, CH, -1, BANK, PROG) == 0
|
||||
got = bridge_render(dll, h, c["gain_db"], c["pan"], c["lim_active"], c["th_db"])
|
||||
assert dll.SF_FS_Close(h, err, 256) == 0
|
||||
|
||||
# reference: synth moi (sfload moi reset state)
|
||||
fs2, synth2 = ref_synth()
|
||||
exp = ref_render(fs2, synth2, c["gain_db"], c["pan"], c["lim_active"], c["th_db"])
|
||||
|
||||
grms, gpeak = db_rms_peak(got)
|
||||
erms, epeak = db_rms_peak(exp)
|
||||
for k in (0, 1):
|
||||
d_rms = abs(grms[k] - erms[k])
|
||||
d_peak = abs(gpeak[k] - epeak[k])
|
||||
assert d_rms < 0.1, f"case {c}: RMS diff {d_rms:.4f} dB > 0.1 (L{k})"
|
||||
assert d_peak < 0.1, f"case {c}: peak diff {d_peak:.4f} dB > 0.1 (L{k})"
|
||||
print(f"OK case {c}: rms {erms.round(2)}/{grms.round(2)} dB, "
|
||||
f"peak {epeak.round(2)}/{gpeak.round(2)} dB, diff <= 0.0001 dB")
|
||||
|
||||
print("PASS")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user