nickmin/gti_radiostudio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Strip compressor/limiter DSP, add per-channel pre-fader gain, PFL/HP blend, LUFS metering from external port, MIDI gain binding

Browse Source
This commit is contained in:
nickmin
2026-05-15 16:42:10 +10:00
parent 4d6a639c08
commit 2b81a80a39
5 changed files with 193 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files
audio_engine.py
+50 -135
View File
@@ -147,19 +147,12 @@ class DeckState:
return f"-{int(remaining//60):02d}:{int(remaining%60):02d}"
class ExternalInputMonitor:
def __init__(self, buffer_size=600):
self.buffer = np.zeros(buffer_size, dtype=np.float32)
self.write_pos = 0
self.buffer_size = buffer_size
self.monitoring = False
class MixerChannel:
def __init__(self, name: str, index: int):
self.name = name
self.index = index
self.volume = 1.0
self.gain = 1.0 # pre-fader gain trim
self.muted = False
self.solo = False
self.pfl = False
@@ -169,71 +162,12 @@ class MixerChannel:
self.in_port_R = None
class Compressor:
def __init__(self, sr: int):
self.sr = sr
self.threshold_db = -20.0
self.ratio = 4.0
self.attack_ms = 5.0
self.release_ms = 100.0
self.makeup_gain_db = 0.0
self.bypassed = False
self._envelope = 0.0
self._attack_coeff = self._time_constant(self.attack_ms)
self._release_coeff = self._time_constant(self.release_ms)
def _time_constant(self, ms: float) -> float:
return np.exp(-1.0 / (ms * self.sr / 1000.0))
def set_attack(self, ms: float):
self.attack_ms = ms
self._attack_coeff = self._time_constant(ms)
def set_release(self, ms: float):
self.release_ms = ms
self._release_coeff = self._time_constant(ms)
def process(self, audio: np.ndarray):
if self.bypassed:
return
threshold_linear = 10 ** (self.threshold_db / 20.0)
makeup_linear = 10 ** (self.makeup_gain_db / 20.0)
for ch in range(audio.shape[0]):
for i in range(audio.shape[1]):
sample = audio[ch, i]
level = abs(sample)
if level > self._envelope:
self._envelope += (1 - self._attack_coeff) * (level - self._envelope)
else:
self._envelope += (1 - self._release_coeff) * (level - self._envelope)
if self._envelope > threshold_linear:
env_db = 20 * np.log10(max(self._envelope, 1e-10))
gain_db = (self.threshold_db - env_db) * (1 - 1.0 / self.ratio)
gain = 10 ** (gain_db / 20.0)
else:
gain = 1.0
audio[ch, i] = sample * gain * makeup_linear
class Limiter:
def __init__(self):
self.ceiling_db = -0.1
self.bypassed = False
def process(self, audio: np.ndarray):
if self.bypassed:
return
ceiling = 10 ** (self.ceiling_db / 20.0)
np.clip(audio, -ceiling, ceiling, out=audio)
class MasterBus:
def __init__(self, sr: int):
self.compressor = Compressor(sr)
self.limiter = Limiter()
self.vu_peak = 0.0
self.lufs_current = -70.0
self.volume = 1.0
self.pfl_hp_blend = 0.5 # 0.0 = PFL only, 0.5 = equal, 1.0 = HP only
self._lufs_accum = 0
@@ -258,16 +192,9 @@ class AudioEngine:
),
}
# External input monitors (deck3, deck4 — pass-through with waveform)
self.ext_monitors = {}
self.ext_ports = {}
for ext_key, label in [('deck3', 'deck3'), ('deck4', 'deck4')]:
in_L = self.client.inports.register(f"{label}_in_L")
in_R = self.client.inports.register(f"{label}_in_R")
out_L = self.client.outports.register(f"{label}_out_L")
out_R = self.client.outports.register(f"{label}_out_R")
self.ext_ports[ext_key] = (in_L, in_R, out_L, out_R)
self.ext_monitors[ext_key] = ExternalInputMonitor()
# LUFS metering input (external processed signal)
self.lufs_in_L = self.client.inports.register("lufs_in_L")
self.lufs_in_R = self.client.inports.register("lufs_in_R")
# Mixer channels: 8 stereo input ports
self.mixer = {
@@ -326,6 +253,14 @@ class AudioEngine:
if 0 <= index < NUM_MIXER_CHANNELS:
self.mixer['channels'][index].volume = max(0.0, min(1.0, volume))
def set_channel_gain(self, index: int, gain: float):
if 0 <= index < NUM_MIXER_CHANNELS:
self.mixer['channels'][index].gain = max(0.0, min(2.0, gain))
def set_channel_gain(self, index: int, gain: float):
if 0 <= index < NUM_MIXER_CHANNELS:
self.mixer['channels'][index].gain = max(0.0, min(2.0, gain))
def set_channel_mute(self, index: int, muted: bool):
if 0 <= index < NUM_MIXER_CHANNELS:
self.mixer['channels'][index].muted = muted
@@ -411,33 +346,8 @@ class AudioEngine:
deck.playing = False
deck.position = track.num_samples - 1
# ── External input monitors (deck3, deck4) ─────────────────────
gate_linear = 10 ** (-46.0 / 20.0)
for ext_key in ('deck3', 'deck4'):
in_L, in_R, out_L, out_R = self.ext_ports[ext_key]
in_buf_L = in_L.get_array()
in_buf_R = in_R.get_array()
out_buf_L = out_L.get_array()
out_buf_R = out_R.get_array()
out_buf_L[:] = in_buf_L[:]
out_buf_R[:] = in_buf_R[:]
monitor = self.ext_monitors[ext_key]
if not monitor.monitoring:
monitor.buffer[:] = 0.0
continue
rms = np.sqrt(np.mean((in_buf_L ** 2 + in_buf_R ** 2) / 2.0))
if rms > gate_linear:
db = 20 * np.log10(max(rms, 1e-10))
display = max(0.0, (db + 46.0) / 46.0)
display = min(1.0, display)
else:
display = 0.0
monitor.buffer[monitor.write_pos] = display
monitor.write_pos = (monitor.write_pos + 1) % monitor.buffer_size
# ── LUFS metering from external processing input ────────────────
lufs_audio = np.array([self.lufs_in_L.get_array(), self.lufs_in_R.get_array()])
# ── Mixer ───────────────────────────────────────────────────────
master_buf_L = self.master_out[0].get_array()
@@ -470,11 +380,12 @@ class AudioEngine:
mute = ch.muted or (any_solo and not ch.solo)
is_mic = ch.index < 2
g = ch.gain
# PFL: pre-fader, pre-mute signal to PFL bus
# PFL: pre-fader, pre-mute signal to PFL bus (with gain applied)
if ch.pfl:
pfl_buf_L[:] += in_L[:]
pfl_buf_R[:] += in_R[:]
pfl_buf_L[:] += in_L[:] * g
pfl_buf_R[:] += in_R[:] * g
if mute:
ch.vu_peak = 0.0
@@ -486,8 +397,8 @@ class AudioEngine:
peak = 0.0
for i in range(frames):
s_L = in_L[i] * vol
s_R = in_R[i] * vol
s_L = in_L[i] * g * vol
s_R = in_R[i] * g * vol
if goes_to_main:
master_buf_L[i] += s_L
master_buf_R[i] += s_R
@@ -506,34 +417,38 @@ class AudioEngine:
studio_buf_L[:] = master_buf_L[:]
studio_buf_R[:] = master_buf_R[:]
# ── Master DSP ──────────────────────────────────────────────────
master_audio = np.array([master_buf_L, master_buf_R])
# ── PFL / Headphone blend ───────────────────────────────────────
blend = self.mixer['master'].pfl_hp_blend
hp_gain = blend # 0..1 how much normal HP mix
pfl_gain = 1.0 - blend # 0..1 how much PFL
any_pfl = any(ch.pfl for ch in self.mixer['channels'])
if any_pfl and pfl_gain > 0.0:
headphone_buf_L[:] = headphone_buf_L * hp_gain + pfl_buf_L * pfl_gain
headphone_buf_R[:] = headphone_buf_R * hp_gain + pfl_buf_R * pfl_gain
elif any_pfl and hp_gain == 0.0:
headphone_buf_L[:] = pfl_buf_L[:]
headphone_buf_R[:] = pfl_buf_R[:]
# else: no PFL active, headphone stays as-is
self.mixer['master'].compressor.process(master_audio)
self.mixer['master'].limiter.process(master_audio)
# ── PFL / Headphone blend ───────────────────────────────────────
blend = self.mixer['master'].pfl_hp_blend
hp_gain = blend # 0..1 how much normal HP mix
pfl_gain = 1.0 - blend # 0..1 how much PFL
any_pfl = any(ch.pfl for ch in self.mixer['channels'])
if any_pfl and pfl_gain > 0.0:
headphone_buf_L[:] = headphone_buf_L * hp_gain + pfl_buf_L * pfl_gain
headphone_buf_R[:] = headphone_buf_R * hp_gain + pfl_buf_R * pfl_gain
elif any_pfl and hp_gain == 0.0:
headphone_buf_L[:] = pfl_buf_L[:]
headphone_buf_R[:] = pfl_buf_R[:]
# else: no PFL active, headphone stays as-is
master_buf_L[:] = master_audio[0]
master_buf_R[:] = master_audio[1]
# Apply same DSP to headphone (without mics)
hp_audio = np.array([headphone_buf_L, headphone_buf_R])
self.mixer['master'].compressor.process(hp_audio)
self.mixer['master'].limiter.process(hp_audio)
headphone_buf_L[:] = hp_audio[0]
headphone_buf_R[:] = hp_audio[1]
# Apply same DSP to studio
studio_audio = np.array([studio_buf_L, studio_buf_R])
self.mixer['master'].compressor.process(studio_audio)
self.mixer['master'].limiter.process(studio_audio)
studio_buf_L[:] = studio_audio[0]
studio_buf_R[:] = studio_audio[1]
master_peak = max(np.max(np.abs(master_audio[0])), np.max(np.abs(master_audio[1])))
# ── Master VU and LUFS ─────────────────────────────────────────
master_peak = max(
np.max(np.abs(master_buf_L)), np.max(np.abs(master_buf_R))
)
self.mixer['master'].vu_peak = master_peak
# Apply master volume fader (master bus only – headphone and studio
# maintain independent levels)
vol = self.mixer['master'].volume
if vol < 1.0:
master_buf_L[:] *= vol
@@ -543,7 +458,7 @@ class AudioEngine:
master._lufs_accum += frames
if master._lufs_accum >= 512:
master._lufs_accum -= 512
self._calculate_lufs(master_audio)
self._calculate_lufs(lufs_audio)
def _calculate_lufs(self, audio: np.ndarray):
mean_square = np.mean(audio ** 2)