660 lines
28 KiB
Python
660 lines
28 KiB
Python
"""Native audio service (T15) — một engine duy nhất cho track instrument.
|
||
|
||
Load các bridge native_host DLL qua ctypes (SF host / VST2 / VST3) và giữ
|
||
instance theo track_id. Mọi track instrument (SF + VSTi) đi qua CÙNG mixer
|
||
native (track gain/pan + master limiter + master fader) — hết drift Bug 2:
|
||
JS không còn route instrument qua masterBus WebAudio.
|
||
|
||
- Live: AudioStart (WASAPI) → note_on/note_off qua SPSC trên audio thread.
|
||
- Offline SF: SF_FS_RenderBlock (write_float + mixer native trong C++).
|
||
- Offline VST2: SF_VST2_Process (raw plugin) + mixer_math (mirror NativeMixer/
|
||
mastering_engine.py) trong Python — bridge Process không áp mixer.
|
||
- VST3 offline: bridge chưa có export Process — matrix ghi limitation (T15);
|
||
VST3 live dùng AudioStart + SendNoteOn như VST2 (AudioEngine chung).
|
||
|
||
master_gain (dB) áp cho MỌI instance — mirror masterBus.output.gain WebAudio
|
||
(sau limiter + clip, linear, không re-clip).
|
||
"""
|
||
import ctypes
|
||
import os
|
||
import threading
|
||
|
||
import numpy as np
|
||
|
||
from app.config import settings
|
||
|
||
_RES_DIR = os.getenv("SF_RESOURCE_DIR")
|
||
_NATIVE_DIR = (
|
||
os.getenv("SONICFORGE_NATIVE_DIR")
|
||
or (os.path.join(_RES_DIR, "native_host") if _RES_DIR else None)
|
||
or os.path.join(settings.BASE_DIR, "native_host", "build", "Release")
|
||
)
|
||
|
||
_ERR = None # khoi tao buffer sau khi dinh nghia _errbuf()
|
||
_LOCK = threading.RLock() # reentrant: _hidden_hwnd() goi trong ensure_*
|
||
|
||
|
||
def native_dir() -> str:
|
||
return _NATIVE_DIR
|
||
|
||
|
||
def _dll(name: str) -> str:
|
||
return os.path.join(_NATIVE_DIR, name)
|
||
|
||
|
||
def _errbuf():
|
||
return ctypes.create_string_buffer(256)
|
||
|
||
|
||
def _errval(buf) -> str:
|
||
return buf.value.decode(errors="replace") if buf and buf.value else ""
|
||
|
||
_ERR = _errbuf() # module-level err buffer (dung chung, 256 bytes)
|
||
|
||
|
||
def _hidden_hwnd():
|
||
"""Tao 1 hidden window dung lam parent cho VST3 editor headless (live).
|
||
SF_VST3_Attach yeu cau parent_hwnd != NULL; window 0-size khong hien thi."""
|
||
import ctypes as _ct
|
||
from ctypes import wintypes as _wt
|
||
if getattr(_hidden_hwnd, "_hwnd", 0):
|
||
return _hidden_hwnd._hwnd
|
||
with _LOCK:
|
||
if getattr(_hidden_hwnd, "_hwnd", 0):
|
||
return _hidden_hwnd._hwnd
|
||
_u = _ct.windll.user32
|
||
_k = _ct.windll.kernel32
|
||
_u.DefWindowProcW.argtypes = [_wt.HWND, _wt.UINT, _ct.c_void_p, _ct.c_void_p]
|
||
_u.DefWindowProcW.restype = _ct.c_long
|
||
_WNDPROC = _ct.WINFUNCTYPE(_ct.c_long, _wt.HWND, _wt.UINT, _ct.c_void_p, _ct.c_void_p) # WPARAM/LPARAM 64-bit
|
||
class _WC(_ct.Structure):
|
||
_fields_ = [("style", _ct.c_uint), ("lpfnWndProc", _WNDPROC), ("cbClsExtra", _ct.c_int),
|
||
("cbWndExtra", _ct.c_int), ("hInstance", _wt.HINSTANCE), ("hIcon", _wt.HICON),
|
||
("hCursor", _wt.HANDLE), ("hbrBackground", _wt.HBRUSH), ("lpszMenuName", _wt.LPCWSTR),
|
||
("lpszClassName", _wt.LPCWSTR)]
|
||
_wc = _WC()
|
||
_wc.lpfnWndProc = _WNDPROC(_u.DefWindowProcW)
|
||
_wc.lpszClassName = "SonicForgeNativeHidden"
|
||
_wc.hInstance = _k.GetModuleHandleW(None)
|
||
if _u.RegisterClassW(_ct.byref(_wc)) or _k.GetLastError() == 1410: # 1410 = class ton tai
|
||
_h = _u.CreateWindowExW(0, "SonicForgeNativeHidden", "sf", 0, 0, 0, 0, 0,
|
||
None, None, _wc.hInstance, None)
|
||
_hidden_hwnd._hwnd = _h or 0
|
||
return _hidden_hwnd._hwnd
|
||
|
||
|
||
def is_available() -> bool:
|
||
return os.path.isfile(_dll("sf_host_bridge.dll"))
|
||
|
||
|
||
# ── mixer math (offline VST) — MIRROR NativeMixer.cpp + mastering_engine.py ──
|
||
def _clamp_db(v, lo, hi, default):
|
||
if v != v: # NaN
|
||
return default
|
||
return max(lo, min(hi, v))
|
||
|
||
|
||
def apply_track_mixer(y, gain_db=0.0, pan=0.0, lim_active=False,
|
||
threshold_db=-1.0, master_lin=1.0):
|
||
"""y: (2, N) float32. Áp track gain/pan + limiter + master fader."""
|
||
y = y * (10.0 ** (gain_db / 20.0))
|
||
if pan != 0.0:
|
||
theta = ((max(-1.0, min(1.0, pan)) + 1.0) / 2.0) * (np.pi / 2.0)
|
||
y = y.copy()
|
||
y[0, :] *= np.cos(theta)
|
||
y[1, :] *= np.sin(theta)
|
||
if lim_active:
|
||
t = _clamp_db(threshold_db, -24.0, 0.0, -1.0)
|
||
t_lin = 10.0 ** (t / 20.0)
|
||
k = 1.0 / max(0.02, t_lin)
|
||
tk = np.tanh(k)
|
||
y = np.tanh(y * k) / tk
|
||
np.clip(y, -1.0, 1.0, out=y)
|
||
y = y * master_lin
|
||
return y
|
||
|
||
|
||
class _SfCtx:
|
||
"""ctypes wrapper một instance sf_host_bridge (handle theo track)."""
|
||
|
||
def __init__(self, dll: ctypes.WinDLL):
|
||
self.dll = dll
|
||
self.handle = 0
|
||
|
||
def create(self, sr, block):
|
||
self.handle = self.dll.SF_FS_Create(sr, block, _ERR, 256)
|
||
return self.handle
|
||
|
||
def close(self):
|
||
if self.handle:
|
||
try:
|
||
self.dll.SF_FS_Close(self.handle, _ERR, 256)
|
||
finally:
|
||
self.handle = 0
|
||
|
||
def __getattr__(self, name):
|
||
return getattr(self.dll, "SF_FS_" + name)
|
||
|
||
|
||
class _Vst2Ctx:
|
||
def __init__(self, dll: ctypes.WinDLL):
|
||
self.dll = dll
|
||
self.handle = 0
|
||
|
||
def create(self, plugin_path):
|
||
w = ctypes.c_int32(0)
|
||
h = ctypes.c_int32(0)
|
||
self.handle = self.dll.SF_VST2_Load(plugin_path.encode("utf-8"), 0,
|
||
ctypes.byref(w), ctypes.byref(h),
|
||
_ERR, 256)
|
||
return self.handle
|
||
|
||
def close(self):
|
||
if self.handle:
|
||
try:
|
||
self.dll.SF_VST2_Close(self.handle, _ERR, 256)
|
||
finally:
|
||
self.handle = 0
|
||
|
||
def __getattr__(self, name):
|
||
return getattr(self.dll, "SF_VST2_" + name)
|
||
|
||
|
||
class _Vst3Ctx:
|
||
def __init__(self, dll: ctypes.WinDLL):
|
||
self.dll = dll
|
||
self.handle = 0
|
||
|
||
def create(self, plugin_path):
|
||
# Attach can plugin_name = ten class VST3 (ClassInfo::name) + parent
|
||
# hwnd. Ten class thuong = ten file/folder .vst3; hidden window cho
|
||
# live headless. Neu ten class khong khop -> rc 0 -> caller fallback.
|
||
base = os.path.basename(plugin_path.rstrip("\\/"))
|
||
if base.lower().endswith(".vst3"):
|
||
plugin_name = base[:-5]
|
||
else:
|
||
plugin_name = os.path.splitext(base)[0]
|
||
w = ctypes.c_int32(0)
|
||
h = ctypes.c_int32(0)
|
||
self.handle = self.dll.SF_VST3_Attach(plugin_path.encode("utf-8"),
|
||
plugin_name.encode("utf-8"),
|
||
_hidden_hwnd(),
|
||
ctypes.byref(w), ctypes.byref(h),
|
||
_ERR, 256)
|
||
return self.handle
|
||
|
||
def close(self):
|
||
if self.handle:
|
||
try:
|
||
self.dll.SF_VST3_Close(self.handle, _ERR, 256)
|
||
finally:
|
||
self.handle = 0
|
||
|
||
def __getattr__(self, name):
|
||
return getattr(self.dll, "SF_VST3_" + name)
|
||
|
||
|
||
class NativeAudioService:
|
||
"""Singleton: registry track_id → instance; master_gain toàn cục."""
|
||
|
||
def __init__(self):
|
||
self._sf = {}
|
||
self._vst2 = {}
|
||
self._vst3 = {}
|
||
self._sf_dll = None
|
||
self._vst2_dll = None
|
||
self._vst3_dll = None
|
||
self.master_gain_db = 0.0
|
||
self._master_lin = 1.0
|
||
self.sample_rate = 44100
|
||
self.block_size = 256
|
||
|
||
# ── master fader (IPC set_master_gain) ──
|
||
def set_master_gain(self, gain_db):
|
||
with _LOCK:
|
||
self.master_gain_db = float(gain_db)
|
||
self._master_lin = 0.0 if self.master_gain_db <= -50 else \
|
||
10.0 ** (self.master_gain_db / 20.0)
|
||
for ctx in list(self._sf.values()) + list(self._vst2.values()) + \
|
||
list(self._vst3.values()):
|
||
if ctx and ctx.handle:
|
||
try:
|
||
ctx.SetMasterGain(ctx.handle, self._master_lin)
|
||
except Exception:
|
||
pass
|
||
return {"ok": True, "master_gain_db": self.master_gain_db,
|
||
"linear": self._master_lin}
|
||
|
||
def _load_sf_dll(self):
|
||
if self._sf_dll is None:
|
||
if not os.path.isfile(_dll("sf_host_bridge.dll")):
|
||
raise RuntimeError(f"thiếu {_dll('sf_host_bridge.dll')} — "
|
||
f"build native_host Release (T9–T14)")
|
||
dll = ctypes.WinDLL(_dll("sf_host_bridge.dll"))
|
||
i32 = ctypes.c_int32
|
||
dll.SF_FS_Create.argtypes = [i32, i32, ctypes.c_char_p, i32]
|
||
dll.SF_FS_Create.restype = i32
|
||
dll.SF_FS_LoadSF2.argtypes = [i32, ctypes.c_char_p, ctypes.c_char_p, i32]
|
||
dll.SF_FS_LoadSF2.restype = i32
|
||
dll.SF_FS_SelectInstrument.argtypes = [i32, i32, i32, i32, i32]
|
||
dll.SF_FS_SelectInstrument.restype = i32
|
||
dll.SF_FS_NoteOn.argtypes = [i32, i32, i32, i32]
|
||
dll.SF_FS_NoteOn.restype = i32
|
||
dll.SF_FS_NoteOff.argtypes = [i32, i32, i32]
|
||
dll.SF_FS_NoteOff.restype = i32
|
||
dll.SF_FS_RenderBlock.argtypes = [i32, i32,
|
||
np.ctypeslib.ndpointer(np.float32, flags="C_CONTIGUOUS"),
|
||
np.ctypeslib.ndpointer(np.float32, flags="C_CONTIGUOUS")]
|
||
dll.SF_FS_RenderBlock.restype = i32
|
||
dll.SF_FS_AudioStart.argtypes = [i32, i32, i32, ctypes.c_char_p, i32]
|
||
dll.SF_FS_AudioStart.restype = i32
|
||
dll.SF_FS_AudioStop.argtypes = [i32]
|
||
dll.SF_FS_AudioStop.restype = i32
|
||
dll.SF_FS_AudioUnderruns.argtypes = [i32]
|
||
dll.SF_FS_AudioUnderruns.restype = i32
|
||
dll.SF_FS_AudioBlocks.argtypes = [i32]
|
||
dll.SF_FS_AudioBlocks.restype = i32
|
||
for fn in ("SetTrackGainPan", "SetMasterLimiter"):
|
||
getattr(dll, "SF_FS_" + fn).argtypes = [i32, ctypes.c_float, ctypes.c_float]
|
||
getattr(dll, "SF_FS_" + fn).restype = i32
|
||
dll.SF_FS_SetMasterGain.argtypes = [i32, ctypes.c_float]
|
||
dll.SF_FS_SetMasterGain.restype = i32
|
||
dll.SF_FS_Close.argtypes = [i32, ctypes.c_char_p, i32]
|
||
dll.SF_FS_Close.restype = i32
|
||
self._sf_dll = dll
|
||
return self._sf_dll
|
||
|
||
def ensure_sf(self, track_id, sf_path, bank=0, program=0, channel=0,
|
||
live=True, gain_db=None, pan=None):
|
||
"""Tạo (hoặc tái dùng) instance SF cho track. sf_path: đường dẫn SF2/SF3."""
|
||
track_id = str(track_id)
|
||
with _LOCK:
|
||
dll = self._load_sf_dll()
|
||
ctx = self._sf.get(track_id)
|
||
if ctx is None:
|
||
ctx = _SfCtx(dll)
|
||
h = ctx.create(self.sample_rate, self.block_size)
|
||
if h <= 0:
|
||
raise RuntimeError(f"SF_FS_Create fail: {_errval(_ERR)}")
|
||
rc = ctx.LoadSF2(h, os.fspath(sf_path).encode("utf-8"), _ERR, 256)
|
||
if rc != 0:
|
||
ctx.close()
|
||
raise RuntimeError(f"SF_FS_LoadSF2 fail rc={rc}: {_errval(_ERR)}")
|
||
# sfontId<0 → dùng sfid của font vừa load (T14)
|
||
rc = ctx.SelectInstrument(h, channel, -1, int(bank), int(program))
|
||
if rc != 0:
|
||
ctx.close()
|
||
raise RuntimeError(f"SF_FS_SelectInstrument fail rc={rc}: "
|
||
f"bank={bank} prog={program}")
|
||
ctx.SetMasterGain(h, self._master_lin)
|
||
if live:
|
||
rc = ctx.AudioStart(h, self.sample_rate, self.block_size, _ERR, 256)
|
||
if rc != 0:
|
||
ctx.close()
|
||
raise RuntimeError(f"SF_FS_AudioStart fail rc={rc}: {_errval(_ERR)}")
|
||
self._sf[track_id] = ctx
|
||
else:
|
||
h = ctx.handle
|
||
ctx.SelectInstrument(h, channel, -1, int(bank), int(program))
|
||
if gain_db is not None and pan is not None:
|
||
ctx.SetTrackGainPan(ctx.handle, float(gain_db), float(pan))
|
||
return {"ok": True, "track_id": track_id, "handle": ctx.handle}
|
||
|
||
def sf_note_on(self, track_id, channel, pitch, velocity=100):
|
||
track_id = str(track_id)
|
||
with _LOCK:
|
||
ctx = self._sf.get(track_id)
|
||
if not ctx or not ctx.handle:
|
||
raise RuntimeError("track SF chưa ensure")
|
||
return {"ok": ctx.NoteOn(ctx.handle, int(channel), int(pitch),
|
||
int(max(1, min(127, velocity)))) == 0}
|
||
|
||
def sf_note_off(self, track_id, channel, pitch):
|
||
track_id = str(track_id)
|
||
with _LOCK:
|
||
ctx = self._sf.get(track_id)
|
||
if not ctx or not ctx.handle:
|
||
raise RuntimeError("track SF chưa ensure")
|
||
return {"ok": ctx.NoteOff(ctx.handle, int(channel), int(pitch)) == 0}
|
||
|
||
def set_track_gain_pan(self, track_id, gain_db, pan):
|
||
"""Áp gain/pan cho instance track (SF/VST2/VST3)."""
|
||
track_id = str(track_id)
|
||
with _LOCK:
|
||
ctx = (self._sf.get(track_id) or self._vst2.get(track_id) or
|
||
self._vst3.get(track_id))
|
||
if not ctx or not ctx.handle:
|
||
raise RuntimeError("track chưa ensure")
|
||
rc = ctx.SetTrackGainPan(ctx.handle, float(gain_db), float(pan))
|
||
return {"ok": rc == 0, "track_id": track_id,
|
||
"gain_db": gain_db, "pan": pan}
|
||
|
||
def sf_stats(self, track_id):
|
||
track_id = str(track_id)
|
||
with _LOCK:
|
||
ctx = self._sf.get(track_id)
|
||
if not ctx or not ctx.handle:
|
||
return None
|
||
return {"blocks": ctx.AudioBlocks(ctx.handle),
|
||
"underruns": ctx.AudioUnderruns(ctx.handle)}
|
||
|
||
def sf_audio_stop(self, track_id):
|
||
track_id = str(track_id)
|
||
with _LOCK:
|
||
ctx = self._sf.get(track_id)
|
||
if not ctx or not ctx.handle:
|
||
return {"ok": False}
|
||
return {"ok": ctx.AudioStop(ctx.handle) == 0}
|
||
|
||
def render_sf_offline(self, sf_path, bank, program, notes, bpm=120.0,
|
||
sr=44100, gain_db=0.0, pan=0.0, lim_active=False,
|
||
threshold_db=-1.0, master_gain_db=0.0):
|
||
"""Render offline 1 đoạn MIDI bằng instance TẠM (không AudioStart) —
|
||
RenderBlock áp mixer native (gain/pan/limiter/master) trong C++."""
|
||
with _LOCK:
|
||
dll = self._load_sf_dll()
|
||
ctx = _SfCtx(dll)
|
||
h = ctx.create(sr, 256)
|
||
if h <= 0:
|
||
raise RuntimeError(f"SF_FS_Create fail: {_errval(_ERR)}")
|
||
try:
|
||
rc = ctx.LoadSF2(h, os.fspath(sf_path).encode("utf-8"), _ERR, 256)
|
||
if rc != 0:
|
||
raise RuntimeError(f"SF_FS_LoadSF2 fail rc={rc}")
|
||
rc = ctx.SelectInstrument(h, 0, -1, int(bank), int(program))
|
||
if rc != 0:
|
||
raise RuntimeError(f"SF_FS_SelectInstrument fail rc={rc}")
|
||
ctx.SetTrackGainPan(h, float(gain_db), float(pan))
|
||
ctx.SetMasterLimiter(h, int(lim_active), float(threshold_db))
|
||
master_lin = 0.0 if master_gain_db <= -50 else \
|
||
10.0 ** (float(master_gain_db) / 20.0)
|
||
ctx.SetMasterGain(h, master_lin)
|
||
|
||
beat_sec = 60.0 / max(30.0, bpm)
|
||
events = []
|
||
total = 0.0
|
||
for ev in notes:
|
||
start_s = float(ev.get("start_beat", 0)) * beat_sec
|
||
dur_s = float(ev.get("duration_beats", 1)) * beat_sec
|
||
pitch = int(ev.get("note", 60))
|
||
vel = int(max(1, min(127, float(ev.get("velocity", 100)))))
|
||
events.append((int(start_s * sr), "on", pitch, vel))
|
||
events.append((int((start_s + dur_s) * sr), "off", pitch, 0))
|
||
total = max(total, start_s + dur_s)
|
||
events.sort(key=lambda e: e[0])
|
||
n = int((max(total, 0.25) + 0.5) * sr)
|
||
out = np.zeros((2, n), dtype=np.float32)
|
||
block = 256
|
||
pos = 0
|
||
ev_idx = 0
|
||
while pos < n:
|
||
while ev_idx < len(events) and events[ev_idx][0] <= pos:
|
||
_, kind, pitch, vel = events[ev_idx]
|
||
if kind == "on":
|
||
ctx.NoteOn(h, 0, pitch, vel)
|
||
else:
|
||
ctx.NoteOff(h, 0, pitch)
|
||
ev_idx += 1
|
||
take = min(block, n - pos)
|
||
l = np.zeros(take, np.float32)
|
||
r = np.zeros(take, np.float32)
|
||
ctx.RenderBlock(h, take, l, r)
|
||
out[0, pos:pos + take] = l
|
||
out[1, pos:pos + take] = r
|
||
pos += take
|
||
return out
|
||
finally:
|
||
ctx.close()
|
||
|
||
# ── VST2 (live + offline qua SF_VST2_Process) ──
|
||
def _load_vst2_dll(self):
|
||
if self._vst2_dll is None:
|
||
if not os.path.isfile(_dll("vst2_host_bridge.dll")):
|
||
raise RuntimeError(f"thiếu {_dll('vst2_host_bridge.dll')}")
|
||
dll = ctypes.WinDLL(_dll("vst2_host_bridge.dll"))
|
||
i32 = ctypes.c_int32
|
||
dll.SF_VST2_Load.argtypes = [ctypes.c_char_p, ctypes.c_void_p,
|
||
ctypes.POINTER(i32), ctypes.POINTER(i32),
|
||
ctypes.c_char_p, i32]
|
||
dll.SF_VST2_Load.restype = i32
|
||
dll.SF_VST2_SendNoteOn.argtypes = [i32, i32, i32, i32]
|
||
dll.SF_VST2_SendNoteOn.restype = i32
|
||
dll.SF_VST2_SendNoteOff.argtypes = [i32, i32, i32]
|
||
dll.SF_VST2_SendNoteOff.restype = i32
|
||
dll.SF_VST2_Process.argtypes = [i32,
|
||
np.ctypeslib.ndpointer(np.float32, flags="C_CONTIGUOUS"),
|
||
i32, i32]
|
||
dll.SF_VST2_Process.restype = i32
|
||
dll.SF_VST2_AudioStart.argtypes = [i32, i32, i32, ctypes.c_char_p, i32]
|
||
dll.SF_VST2_AudioStart.restype = i32
|
||
dll.SF_VST2_AudioStop.argtypes = [i32]
|
||
dll.SF_VST2_AudioStop.restype = i32
|
||
dll.SF_VST2_AudioUnderruns.argtypes = [i32]
|
||
dll.SF_VST2_AudioUnderruns.restype = i32
|
||
dll.SF_VST2_AudioBlocks.argtypes = [i32]
|
||
dll.SF_VST2_AudioBlocks.restype = i32
|
||
dll.SF_VST2_Close.argtypes = [i32, ctypes.c_char_p, i32]
|
||
dll.SF_VST2_Close.restype = i32
|
||
for fn in ("SetTrackGainPan", "SetMasterLimiter"):
|
||
getattr(dll, "SF_VST2_" + fn).argtypes = [i32, ctypes.c_float, ctypes.c_float]
|
||
getattr(dll, "SF_VST2_" + fn).restype = i32
|
||
dll.SF_VST2_SetMasterGain.argtypes = [i32, ctypes.c_float]
|
||
dll.SF_VST2_SetMasterGain.restype = i32
|
||
self._vst2_dll = dll
|
||
return self._vst2_dll
|
||
|
||
def ensure_vst2(self, track_id, plugin_path, live=True, gain_db=None, pan=None):
|
||
track_id = str(track_id)
|
||
with _LOCK:
|
||
dll = self._load_vst2_dll()
|
||
ctx = self._vst2.get(track_id)
|
||
if ctx is None:
|
||
ctx = _Vst2Ctx(dll)
|
||
h = ctx.create(plugin_path)
|
||
if h <= 0:
|
||
raise RuntimeError(f"SF_VST2_Load fail: {_errval(_ERR)}")
|
||
ctx.SetMasterGain(h, self._master_lin)
|
||
if live:
|
||
rc = ctx.AudioStart(h, self.sample_rate, self.block_size, _ERR, 256)
|
||
if rc != 0:
|
||
ctx.close()
|
||
raise RuntimeError(f"SF_VST2_AudioStart fail rc={rc}: {_errval(_ERR)}")
|
||
self._vst2[track_id] = ctx
|
||
if gain_db is not None and pan is not None:
|
||
ctx.SetTrackGainPan(ctx.handle, float(gain_db), float(pan))
|
||
return {"ok": True, "track_id": track_id, "handle": ctx.handle}
|
||
|
||
def vst2_note_on(self, track_id, channel, pitch, velocity=100):
|
||
track_id = str(track_id)
|
||
with _LOCK:
|
||
ctx = self._vst2.get(track_id)
|
||
if not ctx or not ctx.handle:
|
||
raise RuntimeError("track VST2 chưa ensure")
|
||
return {"ok": ctx.SendNoteOn(ctx.handle, int(channel), int(pitch),
|
||
int(max(1, min(127, velocity)))) == 0}
|
||
|
||
def vst2_note_off(self, track_id, channel, pitch):
|
||
track_id = str(track_id)
|
||
with _LOCK:
|
||
ctx = self._vst2.get(track_id)
|
||
if not ctx or not ctx.handle:
|
||
raise RuntimeError("track VST2 chưa ensure")
|
||
return {"ok": ctx.SendNoteOff(ctx.handle, int(channel), int(pitch)) == 0}
|
||
|
||
def vst2_stats(self, track_id):
|
||
track_id = str(track_id)
|
||
with _LOCK:
|
||
ctx = self._vst2.get(track_id)
|
||
if not ctx or not ctx.handle:
|
||
return None
|
||
return {"blocks": ctx.AudioBlocks(ctx.handle),
|
||
"underruns": ctx.AudioUnderruns(ctx.handle)}
|
||
|
||
def render_vst2_offline(self, plugin_path, notes, bpm=120.0, sr=44100,
|
||
gain_db=0.0, pan=0.0, lim_active=False,
|
||
threshold_db=-1.0, master_gain_db=0.0, tail_sec=0.5):
|
||
"""Offline VST2: SF_VST2_Process (raw) + mixer_math (Python mirror).
|
||
|
||
tail_sec: duoi im lang sau note cuoi (autosample muon release tail)."""
|
||
with _LOCK:
|
||
dll = self._load_vst2_dll()
|
||
ctx = _Vst2Ctx(dll)
|
||
h = ctx.create(plugin_path)
|
||
if h <= 0:
|
||
raise RuntimeError(f"SF_VST2_Load fail: {_errval(_ERR)}")
|
||
try:
|
||
beat_sec = 60.0 / max(30.0, bpm)
|
||
events = []
|
||
total = 0.0
|
||
for ev in notes:
|
||
start_s = float(ev.get("start_beat", 0)) * beat_sec
|
||
dur_s = float(ev.get("duration_beats", 1)) * beat_sec
|
||
pitch = int(ev.get("note", 60))
|
||
vel = int(max(1, min(127, float(ev.get("velocity", 100)))))
|
||
events.append((int(start_s * sr), "on", pitch, vel))
|
||
events.append((int((start_s + dur_s) * sr), "off", pitch, 0))
|
||
total = max(total, start_s + dur_s)
|
||
events.sort(key=lambda e: e[0])
|
||
n = int((max(total, 0.25) + float(tail_sec)) * sr)
|
||
out = np.zeros((2, n), dtype=np.float32)
|
||
block = 256
|
||
pos = 0
|
||
ev_idx = 0
|
||
buf = np.zeros((4, block), np.float32) # inputs+outputs (fake: 2+2)
|
||
while pos < n:
|
||
while ev_idx < len(events) and events[ev_idx][0] <= pos:
|
||
_, kind, pitch, vel = events[ev_idx]
|
||
if kind == "on":
|
||
ctx.SendNoteOn(h, 0, pitch, vel)
|
||
else:
|
||
ctx.SendNoteOff(h, 0, pitch)
|
||
ev_idx += 1
|
||
take = min(block, n - pos)
|
||
buf.fill(0)
|
||
ctx.Process(h, buf, 4, take)
|
||
# ponytail: giả định plugin synth 0-input → output ở buf[0:2]
|
||
# (bridge xếp outputs tại offset numInputs; khi cần plugin
|
||
# có input → thêm export getNumIO rồi đọc offset đó)
|
||
out[0, pos:pos + take] = buf[0, :take]
|
||
out[1, pos:pos + take] = buf[1, :take]
|
||
pos += take
|
||
master_lin = 0.0 if master_gain_db <= -50 else \
|
||
10.0 ** (float(master_gain_db) / 20.0)
|
||
return apply_track_mixer(out, gain_db, pan, lim_active,
|
||
threshold_db, master_lin)
|
||
finally:
|
||
ctx.close()
|
||
|
||
# ── VST3: live qua bridge (offline: limitation — chưa có Process export) ──
|
||
def _load_vst3_dll(self):
|
||
if self._vst3_dll is None:
|
||
if not os.path.isfile(_dll("vst3_host_bridge.dll")):
|
||
raise RuntimeError(f"thiếu {_dll('vst3_host_bridge.dll')}")
|
||
dll = ctypes.WinDLL(_dll("vst3_host_bridge.dll"))
|
||
i32 = ctypes.c_int32
|
||
dll.SF_VST3_Attach.argtypes = [ctypes.c_char_p, ctypes.c_char_p,
|
||
ctypes.c_void_p, ctypes.POINTER(i32),
|
||
ctypes.POINTER(i32), ctypes.c_char_p, i32]
|
||
dll.SF_VST3_Attach.restype = i32
|
||
dll.SF_VST3_SendNoteOn.argtypes = [i32, i32, i32, i32]
|
||
dll.SF_VST3_SendNoteOn.restype = i32
|
||
dll.SF_VST3_SendNoteOff.argtypes = [i32, i32, i32]
|
||
dll.SF_VST3_SendNoteOff.restype = i32
|
||
dll.SF_VST3_AudioStart.argtypes = [i32, i32, i32, ctypes.c_char_p, i32]
|
||
dll.SF_VST3_AudioStart.restype = i32
|
||
dll.SF_VST3_AudioStop.argtypes = [i32]
|
||
dll.SF_VST3_AudioStop.restype = i32
|
||
dll.SF_VST3_AudioUnderruns.argtypes = [i32]
|
||
dll.SF_VST3_AudioUnderruns.restype = i32
|
||
dll.SF_VST3_AudioBlocks.argtypes = [i32]
|
||
dll.SF_VST3_Close.argtypes = [i32, ctypes.c_char_p, i32]
|
||
dll.SF_VST3_Close.restype = i32
|
||
for fn in ("SetTrackGainPan", "SetMasterLimiter"):
|
||
getattr(dll, "SF_VST3_" + fn).argtypes = [i32, ctypes.c_float, ctypes.c_float]
|
||
getattr(dll, "SF_VST3_" + fn).restype = i32
|
||
dll.SF_VST3_SetMasterGain.argtypes = [i32, ctypes.c_float]
|
||
dll.SF_VST3_SetMasterGain.restype = i32
|
||
self._vst3_dll = dll
|
||
return self._vst3_dll
|
||
|
||
def ensure_vst3(self, track_id, plugin_path, live=True, gain_db=None, pan=None):
|
||
track_id = str(track_id)
|
||
with _LOCK:
|
||
dll = self._load_vst3_dll()
|
||
ctx = self._vst3.get(track_id)
|
||
if ctx is None:
|
||
ctx = _Vst3Ctx(dll)
|
||
h = ctx.create(plugin_path)
|
||
if h <= 0:
|
||
raise RuntimeError(f"SF_VST3_Attach fail: {_errval(_ERR)}")
|
||
ctx.SetMasterGain(h, self._master_lin)
|
||
if live:
|
||
rc = ctx.AudioStart(h, self.sample_rate, self.block_size, _ERR, 256)
|
||
if rc != 0:
|
||
ctx.close()
|
||
raise RuntimeError(f"SF_VST3_AudioStart fail rc={rc}: {_errval(_ERR)}")
|
||
self._vst3[track_id] = ctx
|
||
if gain_db is not None and pan is not None:
|
||
ctx.SetTrackGainPan(ctx.handle, float(gain_db), float(pan))
|
||
return {"ok": True, "track_id": track_id, "handle": ctx.handle}
|
||
|
||
def vst3_note_on(self, track_id, channel, pitch, velocity=100):
|
||
track_id = str(track_id)
|
||
with _LOCK:
|
||
ctx = self._vst3.get(track_id)
|
||
if not ctx or not ctx.handle:
|
||
raise RuntimeError("track VST3 chưa ensure")
|
||
return {"ok": ctx.SendNoteOn(ctx.handle, int(channel), int(pitch),
|
||
int(max(1, min(127, velocity)))) == 0}
|
||
|
||
def vst3_note_off(self, track_id, channel, pitch):
|
||
track_id = str(track_id)
|
||
with _LOCK:
|
||
ctx = self._vst3.get(track_id)
|
||
if not ctx or not ctx.handle:
|
||
raise RuntimeError("track VST3 chưa ensure")
|
||
return {"ok": ctx.SendNoteOff(ctx.handle, int(channel), int(pitch)) == 0}
|
||
|
||
def close_track(self, track_id):
|
||
track_id = str(track_id)
|
||
with _LOCK:
|
||
for reg in (self._sf, self._vst2, self._vst3):
|
||
ctx = reg.pop(track_id, None)
|
||
if ctx:
|
||
try:
|
||
ctx.close()
|
||
except Exception:
|
||
pass
|
||
return {"ok": True}
|
||
|
||
def close_all(self):
|
||
with _LOCK:
|
||
for reg in (self._sf, self._vst2, self._vst3):
|
||
for ctx in list(reg.values()):
|
||
try:
|
||
ctx.close()
|
||
except Exception:
|
||
pass
|
||
reg.clear()
|
||
return {"ok": True}
|
||
|
||
def status(self):
|
||
return {
|
||
"available": is_available(),
|
||
"dll_dir": _NATIVE_DIR,
|
||
"master_gain_db": self.master_gain_db,
|
||
"sf_tracks": sorted(self._sf.keys()),
|
||
"vst2_tracks": sorted(self._vst2.keys()),
|
||
"vst3_tracks": sorted(self._vst3.keys()),
|
||
}
|
||
|
||
|
||
_service = None
|
||
|
||
|
||
def get_service() -> NativeAudioService:
|
||
global _service
|
||
if _service is None:
|
||
_service = NativeAudioService()
|
||
return _service
|