261 lines
9.6 KiB
Python
261 lines
9.6 KiB
Python
"""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
|