feat: VSTi live playback 100% client-side (không Carla) — autosample SF2 backend (pedalboard) → FluidSynth WASM qua masterBus → mastering → main out; SonicVstiAutosample ensure+dedup+cooldown; runtime gating __enableCarlaLivePlayback; âm bắt qua ensureMidiCapture → clientSideExport
This commit is contained in:
@@ -0,0 +1,260 @@
|
||||
"""apply_mastering — offline mastering chain, mirror chính xác client WebAudio
|
||||
(app.jsx initMasterBus / applyMasteringSettings / rebuildMasteringGraph).
|
||||
|
||||
Render export (pedalboard VSTi / soundfont) chạy qua ĐÚNG chain như live
|
||||
playback: thứ tự module từ settings['chain'] (mặc định EQ -> Imager ->
|
||||
Maximizer), mỗi module bật/tắt theo settings tương ứng. Bỏ oversample
|
||||
(WaveShaper '2x'/'4x') — offline render không cần anti-alias.
|
||||
|
||||
Công thức khớp client:
|
||||
- RBJ Audio-EQ-Cookbook biquad (eqproBiquadMagDb)
|
||||
- Imager 4-band M/S crossfeed g1=(w+100)/200, g2=(100-w)/200
|
||||
- Maximizer: boost -> atan soft-clip -> upward comp (dry + gain*comp) -> hard clip ceiling
|
||||
- Compressor: soft-knee DynamicsCompressor + one-pole attack/release envelope
|
||||
- Limiter: WaveShaper tanh k=1/tLin
|
||||
- Exciter: highpass 2k -> atan(k=3), dry 1.0, wet 0.6*drive/100
|
||||
- Rebalance: M/S L/R crossfeed a=(m+s)/2, b=(m-s)/2
|
||||
"""
|
||||
import numpy as np
|
||||
from scipy.signal import sosfilt
|
||||
|
||||
|
||||
def _clamp(v, lo, hi, default=0.0):
|
||||
"""Mirror client clamp(): missing/NaN -> default (neutral), khong bao gio NaN."""
|
||||
try:
|
||||
n = float(v)
|
||||
except (TypeError, ValueError):
|
||||
return default
|
||||
if not np.isfinite(n):
|
||||
return default
|
||||
return min(hi, max(lo, n))
|
||||
|
||||
|
||||
def _biquad(kind, f0, fs, gain_db=0.0, q=0.707):
|
||||
"""RBJ biquad (b, a) normalized a0=1 — dung cong thuc client eqproBiquadMagDb."""
|
||||
f0 = min(max(float(f0), 20.0), fs * 0.45)
|
||||
w0 = 2 * np.pi * f0 / fs
|
||||
cw = np.cos(w0)
|
||||
sw = np.sin(w0)
|
||||
alpha = sw / (2 * max(0.05, q))
|
||||
A = 10 ** (_clamp(gain_db, -24, 24) / 40)
|
||||
if kind == "peaking":
|
||||
b = [1 + alpha * A, -2 * cw, 1 - alpha * A]
|
||||
a = [1 + alpha / A, -2 * cw, 1 - alpha / A]
|
||||
elif kind == "lowshelf":
|
||||
b = [A * ((A + 1) - (A - 1) * cw + 2 * np.sqrt(A) * alpha),
|
||||
2 * A * ((A - 1) - (A + 1) * cw),
|
||||
A * ((A + 1) - (A - 1) * cw - 2 * np.sqrt(A) * alpha)]
|
||||
a = [(A + 1) + (A - 1) * cw + 2 * np.sqrt(A) * alpha,
|
||||
-2 * ((A - 1) + (A + 1) * cw),
|
||||
(A + 1) + (A - 1) * cw - 2 * np.sqrt(A) * alpha]
|
||||
elif kind == "highshelf":
|
||||
b = [A * ((A + 1) + (A - 1) * cw + 2 * np.sqrt(A) * alpha),
|
||||
-2 * A * ((A - 1) + (A + 1) * cw),
|
||||
A * ((A + 1) + (A - 1) * cw - 2 * np.sqrt(A) * alpha)]
|
||||
a = [(A + 1) - (A - 1) * cw + 2 * np.sqrt(A) * alpha,
|
||||
2 * ((A - 1) - (A + 1) * cw),
|
||||
(A + 1) - (A - 1) * cw - 2 * np.sqrt(A) * alpha]
|
||||
elif kind == "highpass":
|
||||
b = [(1 + cw) / 2, -(1 + cw), (1 + cw) / 2]
|
||||
a = [1 + alpha, -2 * cw, 1 - alpha]
|
||||
elif kind == "lowpass":
|
||||
b = [(1 - cw) / 2, 1 - cw, (1 - cw) / 2]
|
||||
a = [1 + alpha, -2 * cw, 1 - alpha]
|
||||
else:
|
||||
raise ValueError(f"unsupported biquad kind: {kind}")
|
||||
b = np.asarray(b, dtype=np.float64) / a[0]
|
||||
a = np.asarray(a, dtype=np.float64) / a[0]
|
||||
return np.concatenate([b, a])
|
||||
|
||||
|
||||
def _sosfilt_chain(x, sos_list):
|
||||
"""Ap day biquad noi tiep len (2, N) — mot lan sosfilt moi kenh."""
|
||||
if not sos_list:
|
||||
return np.array(x, dtype=np.float64, copy=True)
|
||||
sos = np.asarray(sos_list, dtype=np.float64)
|
||||
out = np.empty((2, x.shape[1]), dtype=np.float64)
|
||||
for ch in range(2):
|
||||
out[ch] = sosfilt(sos, np.asarray(x[ch], dtype=np.float64))
|
||||
return out
|
||||
|
||||
|
||||
def _compressor(x, threshold_db, ratio, knee, attack_s, release_s,
|
||||
makeup_db=0.0, fs=44100.0):
|
||||
"""Soft-knee feedforward compressor (gan dung DynamicsCompressor cua WebAudio)
|
||||
— stereo-linked (detect tren max |L|,|R|), one-pole attack/release."""
|
||||
x64 = np.asarray(x, dtype=np.float64)
|
||||
n = x64.shape[1]
|
||||
if n == 0:
|
||||
return x64
|
||||
level_db = 20 * np.log10(np.max(np.abs(x64), axis=0) + 1e-12)
|
||||
t = threshold_db
|
||||
k = knee
|
||||
gr_db = np.zeros(n, dtype=np.float64)
|
||||
above = level_db >= t + k / 2
|
||||
mid = (level_db > t - k / 2) & (level_db < t + k / 2)
|
||||
gr_db[above] = (1 - 1 / ratio) * (t - level_db[above])
|
||||
gr_db[mid] = (1 - 1 / ratio) * (level_db[mid] - t + k / 2) ** 2 / (2 * k)
|
||||
|
||||
a_att = np.exp(-1 / (attack_s * fs)) if attack_s > 0 else 0.0
|
||||
a_rel = np.exp(-1 / (release_s * fs)) if release_s > 0 else 0.0
|
||||
env = np.empty(n, dtype=np.float64)
|
||||
cur = 0.0
|
||||
prev = 0.0
|
||||
for i in range(n):
|
||||
g = gr_db[i]
|
||||
if g < prev:
|
||||
cur = a_att * cur + (1 - a_att) * g
|
||||
else:
|
||||
cur = a_rel * cur + (1 - a_rel) * g
|
||||
env[i] = cur
|
||||
prev = g
|
||||
gain_lin = 10 ** ((env + makeup_db) / 20)
|
||||
return x64 * gain_lin
|
||||
|
||||
|
||||
def _apply_eq(x, s, fs):
|
||||
eq_on = bool(s.get("eqActive"))
|
||||
sos = [
|
||||
_biquad("lowshelf", 100, fs, _clamp(s.get("eqLowGain"), -24, 24) if eq_on else 0),
|
||||
_biquad("peaking", 822, fs, _clamp(s.get("eqMid1Gain"), -24, 24) if eq_on else 0, q=0.7),
|
||||
_biquad("peaking", 3200, fs, _clamp(s.get("eqMid2Gain"), -24, 24) if eq_on else 0, q=1.2),
|
||||
_biquad("highshelf", 10000, fs, _clamp(s.get("eqHighGain"), -24, 24) if eq_on else 0),
|
||||
]
|
||||
return _sosfilt_chain(x, sos)
|
||||
|
||||
|
||||
def _apply_imager(x, s, fs):
|
||||
im_on = bool(s.get("imagerActive"))
|
||||
bands = [
|
||||
[("lowpass", 100)],
|
||||
[("highpass", 100), ("lowpass", 1000)],
|
||||
[("highpass", 1000), ("lowpass", 6000)],
|
||||
[("highpass", 6000)],
|
||||
]
|
||||
widths = [s.get(f"w{i}") for i in (1, 2, 3, 4)]
|
||||
out = np.zeros_like(x)
|
||||
for bf, w in zip(bands, widths):
|
||||
width = _clamp(w, 0, 200, 100) if im_on else 100
|
||||
g1 = (width + 100) / 200
|
||||
g2 = (100 - width) / 200
|
||||
y = _sosfilt_chain(x, [_biquad(k, f, fs) for k, f in bf])
|
||||
out[0] += g1 * y[0] + g2 * y[1]
|
||||
out[1] += g1 * y[1] + g2 * y[0]
|
||||
return out
|
||||
|
||||
|
||||
def _apply_maximizer(x, s):
|
||||
on = bool(s.get("maximizerActive"))
|
||||
boost = 10 ** (_clamp(s.get("maxGain"), -60, 30) / 20) if on else 1.0
|
||||
boosted = x * boost
|
||||
soft = _clamp(s.get("maxSoftClip"), 0, 100, 0)
|
||||
if on and soft > 0:
|
||||
k = 1 + (soft / 100) * 10
|
||||
dry = np.arctan(boosted * k) / np.arctan(k)
|
||||
else:
|
||||
dry = boosted
|
||||
up = _clamp(s.get("maxUpward"), 0, 30, 0)
|
||||
if on and up > 0:
|
||||
up_gain = 10 ** (up / 20) - 1.0
|
||||
comp = _compressor(boosted, threshold_db=-30, ratio=4.0, knee=10.0,
|
||||
attack_s=0.01, release_s=0.1, makeup_db=0.0)
|
||||
out = dry + up_gain * comp
|
||||
else:
|
||||
out = dry
|
||||
ceil_db = _clamp(s.get("ceiling"), -60, 0, -0.1) if on else -0.1
|
||||
c = 10 ** (ceil_db / 20)
|
||||
return np.clip(out, -c, c)
|
||||
|
||||
|
||||
def _apply_compressor(x, s):
|
||||
if not bool(s.get("compActive")):
|
||||
return x
|
||||
makeup = 10 ** (_clamp(s.get("compMakeup"), 0, 12) / 20)
|
||||
return _compressor(x, threshold_db=_clamp(s.get("compThreshold"), -60, 0),
|
||||
ratio=_clamp(s.get("compRatio"), 1, 20), knee=8.0,
|
||||
attack_s=0.02, release_s=0.25, makeup_db=20 * np.log10(makeup))
|
||||
|
||||
|
||||
def _apply_limiter(x, s):
|
||||
if not bool(s.get("limActive")):
|
||||
return x
|
||||
t_lin = 10 ** (_clamp(s.get("limThreshold"), -24, 0, -1) / 20)
|
||||
k = 1 / max(0.02, t_lin)
|
||||
tk = np.tanh(k)
|
||||
return np.tanh(x * k) / tk
|
||||
|
||||
|
||||
def _apply_exciter(x, s, fs):
|
||||
if not bool(s.get("excActive")):
|
||||
return x
|
||||
wet = (_clamp(s.get("excDrive"), 0, 100, 0) / 100) * 0.6
|
||||
if wet <= 0:
|
||||
return x
|
||||
hp = _sosfilt_chain(x, [_biquad("highpass", 2000, fs, q=0.7)])
|
||||
k = 3
|
||||
sh = np.arctan(hp * k) / np.arctan(k)
|
||||
return x + wet * sh
|
||||
|
||||
|
||||
def _apply_rebalance(x, s):
|
||||
if not bool(s.get("rebalActive")):
|
||||
return x
|
||||
mid = 10 ** (_clamp(s.get("rebalMid"), -24, 24, 0) / 20)
|
||||
side = 10 ** (_clamp(s.get("rebalSide"), -24, 24, 0) / 20)
|
||||
a = (mid + side) / 2
|
||||
b = (mid - side) / 2
|
||||
L, R = x[0], x[1]
|
||||
return np.stack([a * L + b * R, b * L + a * R])
|
||||
|
||||
|
||||
def _apply_eqpro(x, mod, fs):
|
||||
params = mod.get("params") or {}
|
||||
amount = _clamp(params.get("amount"), 0, 200, 100) / 100.0
|
||||
sos = []
|
||||
for b in params.get("bands") or []:
|
||||
kind = b.get("type") or "peaking"
|
||||
f0 = _clamp(b.get("freq"), 20, 20000, 1000)
|
||||
q = _clamp(b.get("q"), 0.1, 18, 1.0)
|
||||
gain = (b.get("gain") or 0) * amount if b.get("active") is not False else 0
|
||||
sos.append(_biquad(kind, f0, fs, gain, q))
|
||||
return _sosfilt_chain(x, sos)
|
||||
|
||||
|
||||
def apply_mastering(buffer, settings, sample_rate):
|
||||
"""Ap mastering chain (client format mastering_settings) len (2, N) audio.
|
||||
|
||||
buffer: (2, N) — L row 0. settings: dict hoac None. Tra (2, N) float64
|
||||
(copy) da xu li; neu gate tat / khong co module -> tra copy khong doi."""
|
||||
if buffer.ndim != 2 or buffer.shape[0] != 2:
|
||||
raise ValueError("buffer phai la (2, N) stereo")
|
||||
x = np.asarray(buffer, dtype=np.float64)
|
||||
if not settings:
|
||||
return x.copy()
|
||||
if not settings.get("masterConnected") or settings.get("isBypassed"):
|
||||
return x.copy()
|
||||
chain = settings.get("chain") or []
|
||||
mods = [m for m in chain if isinstance(m, dict) and m.get("active")]
|
||||
if not mods:
|
||||
return x.copy()
|
||||
fs = float(sample_rate) if sample_rate else 44100.0
|
||||
for m in mods:
|
||||
t = m.get("type")
|
||||
if t == "eq":
|
||||
x = _apply_eq(x, settings, fs)
|
||||
elif t == "imager":
|
||||
x = _apply_imager(x, settings, fs)
|
||||
elif t == "maximizer":
|
||||
x = _apply_maximizer(x, settings)
|
||||
elif t == "compressor":
|
||||
x = _apply_compressor(x, settings)
|
||||
elif t == "limiter":
|
||||
x = _apply_limiter(x, settings)
|
||||
elif t == "exciter":
|
||||
x = _apply_exciter(x, settings, fs)
|
||||
elif t == "rebalance":
|
||||
x = _apply_rebalance(x, settings)
|
||||
elif t == "eqpro":
|
||||
x = _apply_eqpro(x, m, fs)
|
||||
# type 'carla' = pass-through (VST FX ngoai) — mirror client, bo qua
|
||||
return x
|
||||
@@ -470,10 +470,25 @@ class PythonRenderEngine:
|
||||
_cache={},
|
||||
)
|
||||
|
||||
# Normalization to prevent clipping
|
||||
max_peak = np.max(np.abs(master_buffer))
|
||||
if max_peak > 1.0:
|
||||
master_buffer /= max_peak
|
||||
# Mastering chain — mirror client WebAudio (app.jsx applyMasteringSettings /
|
||||
# rebuildMasteringGraph). Truoc day mastering_settings bi BO QUA hoan toan:
|
||||
# export WAV khong qua EQ/Imager/Maximizer -> file khac am nghe live.
|
||||
# Gio ap dung dung chain (thu tu settings['chain']) khi mastering bat.
|
||||
mastering_settings = project_json.get("mastering_settings")
|
||||
mastered = bool(mastering_settings and mastering_settings.get("masterConnected")
|
||||
and not mastering_settings.get("isBypassed"))
|
||||
if mastered:
|
||||
# Import lazy (scipy.signal) — giam thoi gian khoi dong engine
|
||||
from app.core.mastering_engine import apply_mastering
|
||||
master_buffer = apply_mastering(master_buffer, mastering_settings,
|
||||
self.sample_rate)
|
||||
# Hard clip [-1,1] — khớp WAV encoder client (browser) sau mastering
|
||||
np.clip(master_buffer, -1.0, 1.0, out=master_buffer)
|
||||
else:
|
||||
# Normalization to prevent clipping
|
||||
max_peak = np.max(np.abs(master_buffer))
|
||||
if max_peak > 1.0:
|
||||
master_buffer /= max_peak
|
||||
|
||||
# Write final output file (bit_depth: 16/24/32 → WAV PCM subtype)
|
||||
subtype_map = {16: "PCM_16", 24: "PCM_24", 32: "PCM_32"}
|
||||
|
||||
Reference in New Issue
Block a user