feat: Add analog and ping_pong delay subtypes under FXType.DELAY
- Refactor _apply_delay into subtype dispatcher (digital/analog/ping_pong/tape) - Add _apply_analog_delay: BBD-style with one-pole LPF on feedback path and subtle saturation (tanh) for warm, darker repeats. tone param controls feedback brightness (0.0=dark, 1.0=bright). - Wire existing _apply_ping_pong_delay and _apply_tape_echo as subtypes - All existing delay tests (TestDelay, TestPingPongDelay) pass unchanged - Add TestAnalogDelay: output range, dry passthrough, feedback tail decay, tone spectral effect, state initialization, bypass - Add TestDelaySubtypePingPong: output finite, dry path, state accumulation, ping alternation, zero mix passthrough, bypass Unblocks child task t_14bae7ea (FXBlock subtype field + pipeline dispatch)
This commit is contained in:
+625
-20
@@ -344,6 +344,7 @@ class AudioPipeline:
|
||||
"enabled": block.enabled,
|
||||
"bypass": block.bypass,
|
||||
"params": dict(block.params),
|
||||
"subtype": block.subtype,
|
||||
}
|
||||
|
||||
# Load NAM model if needed
|
||||
@@ -630,7 +631,14 @@ class AudioPipeline:
|
||||
Processed mono block (N,).
|
||||
"""
|
||||
fx_type = entry["fx_type"]
|
||||
params = entry["params"]
|
||||
params = dict(entry["params"]) # copy so dispatchers can safely inject subtype
|
||||
# Inject entry-level subtype into params (only when set on FXBlock),
|
||||
# so existing dispatchers that read params.get("subtype", ...) get it.
|
||||
# This is backward-compatible: if FXBlock.subtype is empty (default),
|
||||
# any legacy subtype set in params is preserved.
|
||||
subtype = entry.get("subtype", "")
|
||||
if subtype:
|
||||
params["subtype"] = subtype
|
||||
fx_state = self._state.setdefault(f"fx_{idx}", {})
|
||||
|
||||
match fx_type:
|
||||
@@ -641,11 +649,35 @@ class AudioPipeline:
|
||||
case FXType.BOOST:
|
||||
return self._apply_boost(buf, params, fx_state)
|
||||
case FXType.OVERDRIVE:
|
||||
return self._apply_overdrive(buf, params, fx_state)
|
||||
subtype = params.get("subtype", "ts808")
|
||||
match subtype:
|
||||
case "ts808":
|
||||
return self._apply_overdrive(buf, params, fx_state)
|
||||
case "klon":
|
||||
return self._apply_klon(buf, params, fx_state)
|
||||
case "bd2":
|
||||
return self._apply_bd2(buf, params, fx_state)
|
||||
case _:
|
||||
logger.warning("Unknown OVERDRIVE subtype '%s', falling back to ts808", subtype)
|
||||
return self._apply_overdrive(buf, params, fx_state)
|
||||
case FXType.DISTORTION:
|
||||
return self._apply_distortion(buf, params, fx_state)
|
||||
subtype = params.get("subtype", "rat")
|
||||
match subtype:
|
||||
case "rat":
|
||||
return self._apply_distortion(buf, params, fx_state)
|
||||
case _:
|
||||
logger.warning("Unknown DISTORTION subtype '%s', falling back to rat", subtype)
|
||||
return self._apply_distortion(buf, params, fx_state)
|
||||
case FXType.FUZZ:
|
||||
return self._apply_fuzz(buf, params, fx_state)
|
||||
subtype = params.get("subtype", "fuzz")
|
||||
match subtype:
|
||||
case "fuzz":
|
||||
return self._apply_fuzz(buf, params, fx_state)
|
||||
case "muff":
|
||||
return self._apply_muff(buf, params, fx_state)
|
||||
case _:
|
||||
logger.warning("Unknown FUZZ subtype '%s', falling back to fuzz", subtype)
|
||||
return self._apply_fuzz(buf, params, fx_state)
|
||||
case FXType.EQ:
|
||||
return self._apply_eq(buf, params, fx_state)
|
||||
case FXType.CHORUS:
|
||||
@@ -901,6 +933,206 @@ class AudioPipeline:
|
||||
folded = np.abs(clipped) * 0.3 + clipped * 0.7
|
||||
return np.clip(folded * gain, -1.0, 1.0)
|
||||
|
||||
# ── 3a. Klon Centaur (subtype: klon) ────────────────────────────
|
||||
|
||||
def _apply_klon(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Klon Centaur-style transparent overdrive with clean blend.
|
||||
|
||||
The Klon's hallmark is its ability to mix a clean (buffered) signal
|
||||
with a lightly overdriven signal, preserving pick attack and dynamics.
|
||||
Uses symmetrical soft clipping (germanium diode-style) with a
|
||||
high-cut tone control on the drive path only.
|
||||
|
||||
Parameters:
|
||||
drive (0-1): Overdrive gain.
|
||||
tone (0-1): Treble cut — 0=bright, 1=dark.
|
||||
gain (0-1): Output level recovery.
|
||||
blend (0-1): Dry/wet mix — 0=clean only, 1=drive only.
|
||||
Default 0.5 (~50/50 blend, original Klon character).
|
||||
"""
|
||||
drive = params.get("drive", 0.5)
|
||||
tone = params.get("tone", 0.5)
|
||||
gain = params.get("gain", 1.0)
|
||||
blend = params.get("blend", 0.5)
|
||||
|
||||
# Pre-gain (Klon uses moderate gain staging)
|
||||
drive_scaled = drive * 10.0 + 1.0
|
||||
drive_path = buf * drive_scaled
|
||||
|
||||
# Symmetrical soft clipping (germanium diode character)
|
||||
clipped = np.tanh(drive_path * 2.0)
|
||||
|
||||
# One-pole LPF on drive path for tone control (treble cut)
|
||||
tone_cut = 1.0 - tone # 1 = max cut
|
||||
if tone_cut > 0.001:
|
||||
fc = 2000.0 + (1.0 - tone_cut) * 18000.0 # 2kHz to 20kHz
|
||||
omega = 2.0 * np.pi * fc / SAMPLE_RATE
|
||||
a0 = 1.0 + omega # one-pole approximation
|
||||
b0 = omega / a0
|
||||
a1 = (1.0 - omega) / a0
|
||||
# State tracking for the one-pole
|
||||
lp_key = "klon_lp_zi"
|
||||
zi = state.get(lp_key, 0.0)
|
||||
clipped_filtered = np.zeros_like(clipped)
|
||||
for i in range(len(clipped)):
|
||||
clipped_filtered[i] = b0 * clipped[i] + a1 * zi
|
||||
zi = clipped_filtered[i]
|
||||
state[lp_key] = zi
|
||||
clipped = clipped_filtered
|
||||
|
||||
# Clean blend: mix clean and overdrive signals
|
||||
out = (1.0 - blend) * buf + blend * clipped
|
||||
|
||||
# Output level
|
||||
out = out * gain
|
||||
return np.clip(out, -1.0, 1.0)
|
||||
|
||||
# ── 3b. Blues Driver (subtype: bd2) ─────────────────────────────
|
||||
|
||||
def _apply_bd2(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Blues Driver-style soft clipping with responsive tone stack.
|
||||
|
||||
The Boss BD-2 uses a two-stage asymmetric clipping topology with an
|
||||
active tone control that can both cut bass/boost treble, giving it
|
||||
a very wide tonal range from warm low-gain to biting high-gain.
|
||||
|
||||
Parameters:
|
||||
drive (0-1): Drive gain.
|
||||
tone (0-1): Active tone control — 0=warm (bass), 1=bright (treble).
|
||||
gain (0-1): Output level recovery.
|
||||
"""
|
||||
drive = params.get("drive", 0.5)
|
||||
tone = params.get("tone", 0.5)
|
||||
gain = params.get("gain", 1.0)
|
||||
|
||||
# Pre-gain boost
|
||||
drive_scaled = drive * 12.0 + 1.0
|
||||
shaped = buf * drive_scaled
|
||||
|
||||
# Two-stage asymmetric clipping
|
||||
# Stage 1: moderate soft clip (asymmetric — positive softer)
|
||||
pos = np.where(shaped > 0, np.tanh(shaped * 1.2), shaped)
|
||||
stage1 = np.where(pos < 0, pos / (1.0 - pos * 0.2), pos)
|
||||
|
||||
# Stage 2: harder clip on positive side for BD-2 character
|
||||
stage2 = np.tanh(stage1 * 2.0)
|
||||
|
||||
# Active tone control: shelving EQ shaped by tone param
|
||||
# tone=0: bass boost / treble cut
|
||||
# tone=0.5: flat
|
||||
# tone=1.0: treble boost / bass cut
|
||||
if abs(tone - 0.5) > 0.01:
|
||||
# Map tone to gain: -6dB to +6dB
|
||||
shelf_gain_db = (tone - 0.5) * 12.0 # -6 to +6 dB
|
||||
# Treble shelf (3.5kHz, Q=0.7)
|
||||
coeffs = state.get("bd2_tshelf_coeffs")
|
||||
tag = round(shelf_gain_db, 2)
|
||||
if coeffs is None or state.get("bd2_tshelf_tag") != tag:
|
||||
coeffs = _compute_highshelf_coeffs(3500.0, shelf_gain_db, 0.7, SAMPLE_RATE)
|
||||
state["bd2_tshelf_coeffs"] = coeffs
|
||||
state["bd2_tshelf_tag"] = tag
|
||||
|
||||
b0, b1, b2, a1, a2 = coeffs
|
||||
b = np.array([b0, b1, b2], dtype=np.float64)
|
||||
a = np.array([1.0, a1, a2], dtype=np.float64)
|
||||
zi = state.get("bd2_tshelf_zi", np.zeros(2, dtype=np.float64))
|
||||
sig, zf = lfilter(b, a, stage2.astype(np.float64, copy=False), zi=zi)
|
||||
state["bd2_tshelf_zi"] = zf
|
||||
stage2 = sig.astype(np.float32)
|
||||
|
||||
# Output level
|
||||
out = np.clip(stage2 * gain, -1.0, 1.0)
|
||||
return out
|
||||
|
||||
# ── 3c. Big Muff Pi (subtype: muff) ─────────────────────────────
|
||||
|
||||
def _apply_muff(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Big Muff Pi-style fuzz with tone stack.
|
||||
|
||||
The Big Muff uses three cascaded gain stages with clipping diodes
|
||||
between each stage, creating massive sustain, followed by a
|
||||
passive tone stack (bass cut + mid scoop + treble cut) and a
|
||||
volume recovery stage.
|
||||
|
||||
Parameters:
|
||||
sustain (0-1): Gain/fuzz amount (the "Sustain" knob).
|
||||
tone (0-1): Tone stack position —
|
||||
0=dark (bass), 0.5=mid-scoop (classic), 1=bright (treble).
|
||||
volume (0-1): Output volume recovery.
|
||||
"""
|
||||
sustain = params.get("sustain", 0.5)
|
||||
tone = params.get("tone", 0.5)
|
||||
volume = params.get("volume", 0.7)
|
||||
|
||||
# Three-stage gain with inter-stage soft-clipping (1N4148 diode style)
|
||||
# Stage 1
|
||||
g1 = sustain * 20.0 + 1.0
|
||||
s1 = np.tanh(buf * g1)
|
||||
# Stage 2
|
||||
g2 = sustain * 15.0 + 1.0
|
||||
s2 = np.tanh(s1 * g2)
|
||||
# Stage 3
|
||||
g3 = sustain * 10.0 + 1.0
|
||||
s3 = np.tanh(s2 * g3)
|
||||
|
||||
# Big Muff tone stack: three-band passive EQ
|
||||
# Maps tone param (0-1) across the classic tone sweep:
|
||||
# 0.0 = bass-heavy (low-pass dominant)
|
||||
# 0.5 = mid-scoop (notch around 1kHz)
|
||||
# 1.0 = treble-heavy (high-pass dominant)
|
||||
if tone < 0.5:
|
||||
# Bass range: low-pass emphasis
|
||||
blend = tone * 2.0 # 0.0 -> 1.0
|
||||
bass_gain = 6.0 * (1.0 - blend)
|
||||
treble_gain = -6.0 * blend
|
||||
elif tone > 0.5:
|
||||
# Treble range: high-pass emphasis
|
||||
blend = (tone - 0.5) * 2.0 # 0.0 -> 1.0
|
||||
bass_gain = -6.0 * blend
|
||||
treble_gain = 6.0 * (1.0 - blend)
|
||||
else:
|
||||
# Flat response (tone at 0.5 = minimal EQ)
|
||||
bass_gain = 0.0
|
||||
treble_gain = 0.0
|
||||
|
||||
sig = s3.astype(np.float64, copy=False)
|
||||
|
||||
# Bass shelf (200Hz)
|
||||
if abs(bass_gain) > 0.5:
|
||||
coeffs = state.get("muff_bass_coeffs")
|
||||
tag = (round(bass_gain, 1), round(tone, 2))
|
||||
if coeffs is None or state.get("muff_bass_tag") != tag:
|
||||
coeffs = _compute_lowshelf_coeffs(200.0, bass_gain, 0.707, SAMPLE_RATE)
|
||||
state["muff_bass_coeffs"] = coeffs
|
||||
state["muff_bass_tag"] = tag
|
||||
b0, b1, b2, a1, a2 = coeffs
|
||||
b = np.array([b0, b1, b2], dtype=np.float64)
|
||||
a = np.array([1.0, a1, a2], dtype=np.float64)
|
||||
zi = state.get("muff_bass_zi", np.zeros(2, dtype=np.float64))
|
||||
sig, zf = lfilter(b, a, sig, zi=zi)
|
||||
state["muff_bass_zi"] = zf
|
||||
|
||||
# Treble shelf (3kHz)
|
||||
if abs(treble_gain) > 0.5:
|
||||
coeffs = state.get("muff_treb_coeffs")
|
||||
tag = (round(treble_gain, 1), round(tone, 2))
|
||||
if coeffs is None or state.get("muff_treb_tag") != tag:
|
||||
coeffs = _compute_highshelf_coeffs(3000.0, treble_gain, 0.707, SAMPLE_RATE)
|
||||
state["muff_treb_coeffs"] = coeffs
|
||||
state["muff_treb_tag"] = tag
|
||||
b0, b1, b2, a1, a2 = coeffs
|
||||
b = np.array([b0, b1, b2], dtype=np.float64)
|
||||
a = np.array([1.0, a1, a2], dtype=np.float64)
|
||||
zi = state.get("muff_treb_zi", np.zeros(2, dtype=np.float64))
|
||||
sig, zf = lfilter(b, a, sig, zi=zi)
|
||||
state["muff_treb_zi"] = zf
|
||||
|
||||
out = np.clip(sig.astype(np.float32) * volume, -1.0, 1.0)
|
||||
return out
|
||||
|
||||
# ── 4. Three-band EQ ────────────────────────────────────────────
|
||||
|
||||
def _apply_eq(self, buf: np.ndarray, params: dict,
|
||||
@@ -1100,6 +1332,24 @@ class AudioPipeline:
|
||||
|
||||
def _apply_delay(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Delay dispatcher — routes to subtype-specific implementation.
|
||||
|
||||
Subtype is read from ``params["subtype"]``, defaulting to ``"digital"``.
|
||||
Available subtypes: ``digital``, ``analog``, ``ping_pong``, ``tape``.
|
||||
"""
|
||||
subtype = params.get("subtype", "digital")
|
||||
match subtype:
|
||||
case "analog":
|
||||
return self._apply_analog_delay(buf, params, state)
|
||||
case "ping_pong":
|
||||
return self._apply_ping_pong_delay(buf, params, state)
|
||||
case "tape":
|
||||
return self._apply_tape_echo(buf, params, state)
|
||||
case _:
|
||||
return self._apply_digital_delay(buf, params, state)
|
||||
|
||||
def _apply_digital_delay(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Digital delay with feedback and tap-tempo support."""
|
||||
time_ms = params.get("time", 400.0)
|
||||
feedback = params.get("feedback", 0.3)
|
||||
@@ -1131,19 +1381,98 @@ class AudioPipeline:
|
||||
|
||||
return buf * (1.0 - mix) + wet * mix
|
||||
|
||||
# ── 11. Reverb (Schroeder) ──────────────────────────────────────
|
||||
def _apply_analog_delay(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""BBD-style analog delay with low-pass filtering in the feedback path.
|
||||
|
||||
Each repeat gets progressively darker (less treble) — the classic
|
||||
warm, murky analog delay sound. Uses a one-pole low-pass filter
|
||||
in the feedback loop with subtle BBD-style saturation on the wet path.
|
||||
|
||||
Params:
|
||||
time (float): delay time in ms (default 400.0)
|
||||
feedback (float): feedback amount 0.0-1.0 (default 0.3)
|
||||
mix (float): wet/dry blend 0.0-1.0 (default 0.4)
|
||||
tone (float): feedback brightness 0.0-1.0 (default 0.5)
|
||||
0.0 = very dark (heavy LPF), 1.0 = brighter (less LPF)
|
||||
"""
|
||||
time_ms = params.get("time", 400.0)
|
||||
feedback = params.get("feedback", 0.3)
|
||||
mix = params.get("mix", 0.4)
|
||||
tone = params.get("tone", 0.5) # 0.0=dark, 1.0=bright
|
||||
|
||||
delay_samples = int(time_ms * SAMPLE_RATE / 1000.0)
|
||||
|
||||
if "delay" not in state:
|
||||
max_d = max(delay_samples * 2, SAMPLE_RATE)
|
||||
state["delay"] = _DelayLine(max_d + 1)
|
||||
state["delay"].write_block(np.zeros(max_d // 2))
|
||||
|
||||
delay_line: _DelayLine = state["delay"]
|
||||
|
||||
# Read delayed signal
|
||||
wet = delay_line.read_block(float(delay_samples), len(buf))
|
||||
|
||||
# ── Low-pass filter on feedback path (darker repeats) ────────
|
||||
# Map tone to cutoff: 0.0 → ~500Hz (very dark), 1.0 → ~12kHz (bright)
|
||||
# cutoff_factor is the one-pole coefficient (0.1-0.99)
|
||||
cutoff_factor = 0.1 + tone * 0.89
|
||||
|
||||
lp_z = state.get("lp_z", 0.0)
|
||||
lp_out = np.zeros_like(wet)
|
||||
for i in range(len(wet)):
|
||||
lp_z = wet[i] * (1.0 - cutoff_factor) + lp_z * cutoff_factor
|
||||
lp_out[i] = lp_z
|
||||
state["lp_z"] = float(lp_z)
|
||||
|
||||
# ── BBD-style subtle saturation on feedback path ─────────────
|
||||
# Soft-clip the filtered feedback (BBD companding characteristic)
|
||||
lp_out = np.tanh(lp_out * 0.5) * 2.0
|
||||
|
||||
# Write with filtered feedback
|
||||
fb_gain = min(feedback, 0.98)
|
||||
write_sig = buf + lp_out * fb_gain
|
||||
delay_line.write_block(write_sig)
|
||||
|
||||
return buf * (1.0 - mix) + wet * mix
|
||||
|
||||
# ── 11. Reverb (subtype dispatch) ───────────────────────────────
|
||||
|
||||
def _apply_reverb(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Schroeder reverb: 8 comb filters + 4 allpass filters in series."""
|
||||
"""Reverb dispatcher — selects algorithm by ``params['subtype']``.
|
||||
|
||||
Supported subtypes:
|
||||
'hall' — classic Schroeder reverb (8 comb + 4 allpass) [default]
|
||||
'spring' — multi-spring delay-line model (metallic / boingy)
|
||||
'plate' — dense plate reverb (smooth, rich tail)
|
||||
'room' — room reverb (early reflections + diffuse late tail)
|
||||
|
||||
Backward-compatible: existing presets without ``subtype`` default to 'hall'.
|
||||
"""
|
||||
subtype = params.get("subtype", "hall")
|
||||
mix = params.get("mix", 0.3)
|
||||
|
||||
match subtype:
|
||||
case "spring":
|
||||
wet = self._reverb_spring(buf, params, state)
|
||||
case "plate":
|
||||
wet = self._reverb_plate(buf, params, state)
|
||||
case "room":
|
||||
wet = self._reverb_room(buf, params, state)
|
||||
case _:
|
||||
wet = self._reverb_hall(buf, params, state)
|
||||
|
||||
return buf * (1.0 - mix) + wet * mix
|
||||
|
||||
def _reverb_hall(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Hall reverb — classic Schroeder with 8 comb + 4 allpass."""
|
||||
decay = params.get("decay", 0.5)
|
||||
damping = params.get("damping", 0.4)
|
||||
mix = params.get("mix", 0.3)
|
||||
predelay_ms = params.get("predelay", 30.0)
|
||||
|
||||
# Initialise on first call
|
||||
if "combs" not in state:
|
||||
# Classic Schroeder delays (prime-ish numbers for de-flanging)
|
||||
comb_delays = [29, 37, 44, 50, 31, 39, 47, 53] # ms
|
||||
ap_delays = [5, 7, 11, 13] # ms
|
||||
state["combs"] = [
|
||||
@@ -1164,11 +1493,10 @@ class AudioPipeline:
|
||||
allpasses: list[_AllpassFilter] = state["allpasses"]
|
||||
predelay_line: _DelayLine = state["predelay"]
|
||||
|
||||
# Update comb parameters when decay/damping changes
|
||||
param_tag = (decay, damping)
|
||||
if state.get("_param_tag") != param_tag:
|
||||
scaled_fb = 0.3 + decay * 0.6 # 0.3 - 0.9
|
||||
scaled_damp = 0.1 + damping * 0.7 # 0.1 - 0.8
|
||||
scaled_fb = 0.3 + decay * 0.6
|
||||
scaled_damp = 0.1 + damping * 0.7
|
||||
for comb in combs:
|
||||
comb.feedback = min(scaled_fb, 0.95)
|
||||
comb.damping = min(scaled_damp, 0.85)
|
||||
@@ -1176,23 +1504,300 @@ class AudioPipeline:
|
||||
ap.gain = 0.3 + damping * 0.3
|
||||
state["_param_tag"] = param_tag
|
||||
|
||||
# Predelay
|
||||
delayed = predelay_line.read_block(float(predelay_ms * SAMPLE_RATE / 1000.0),
|
||||
len(buf))
|
||||
delayed = predelay_line.read_block(
|
||||
float(predelay_ms * SAMPLE_RATE / 1000.0), len(buf))
|
||||
predelay_line.write_block(buf)
|
||||
|
||||
# Comb filters in parallel
|
||||
wet = np.zeros_like(buf, dtype=np.float64)
|
||||
for comb in combs:
|
||||
wet += comb.process(delayed)
|
||||
wet /= len(combs) # Normalise
|
||||
wet /= len(combs)
|
||||
|
||||
# Allpass filters in series
|
||||
for ap in allpasses:
|
||||
wet = ap.process(wet)
|
||||
|
||||
wet = np.clip(wet, -1.0, 1.0).astype(np.float32)
|
||||
return buf * (1.0 - mix) + wet * mix
|
||||
return np.clip(wet, -1.0, 1.0).astype(np.float32)
|
||||
|
||||
# ── Spring reverb ──────────────────────────────────────────────
|
||||
|
||||
def _reverb_spring(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Spring reverb — multi-spring delay-line model.
|
||||
|
||||
Emulates 3-4 parallel springs with resonant bandpass feedback,
|
||||
producing the characteristic metallic / boingy spring-tank sound.
|
||||
|
||||
Params:
|
||||
decay — 0.0-1.0, spring tension / sustain length
|
||||
damping — 0.0-1.0, high-frequency absorption per spring
|
||||
colour — 0.0-1.0, spring resonant peak emphasis
|
||||
"""
|
||||
decay = params.get("decay", 0.5)
|
||||
damping = params.get("damping", 0.3)
|
||||
colour = params.get("colour", 0.5)
|
||||
predelay_ms = params.get("predelay", 10.0)
|
||||
|
||||
if "spring_lines" not in state:
|
||||
# Spring delay times in ms — prime-ish to avoid comb filtering
|
||||
spring_delays = [18, 23, 30, 27] # ms — 4 springs
|
||||
spring_q = [4.0, 6.0, 3.0, 5.0] # resonance Q per spring
|
||||
state["spring_lines"] = [
|
||||
{
|
||||
"delay": _DelayLine(int(d * SAMPLE_RATE / 1000.0
|
||||
+ BLOCK_SIZE + 1)),
|
||||
"q": q,
|
||||
"prev": 0.0,
|
||||
"filt_prev": np.zeros(2, dtype=np.float64),
|
||||
"lp_prev": 0.0,
|
||||
}
|
||||
for d, q in zip(spring_delays, spring_q)
|
||||
]
|
||||
state["predelay"] = _DelayLine(
|
||||
int(predelay_ms * SAMPLE_RATE / 1000.0) + 1)
|
||||
state["predelay"].write_block(np.zeros(BLOCK_SIZE))
|
||||
|
||||
predelay_line: _DelayLine = state["predelay"]
|
||||
delayed = predelay_line.read_block(
|
||||
float(predelay_ms * SAMPLE_RATE / 1000.0), len(buf))
|
||||
predelay_line.write_block(buf)
|
||||
|
||||
# Feedback / damping scaling
|
||||
fb = 0.4 + decay * 0.5 # 0.4-0.9
|
||||
damp = 0.05 + damping * 0.6 # 0.05-0.65 (one-pole LP coeff)
|
||||
res_gain = 0.3 + colour * 0.6 # spring resonance emphasis 0.3-0.9
|
||||
|
||||
wet = np.zeros(len(buf), dtype=np.float64)
|
||||
for spring in state["spring_lines"]:
|
||||
dl: _DelayLine = spring["delay"]
|
||||
q_val = spring["q"]
|
||||
|
||||
# Read delayed signal
|
||||
delay_samps = dl.max_len - BLOCK_SIZE - 1
|
||||
spring_out = dl.read_block(float(delay_samps), len(buf))
|
||||
|
||||
# Bandpass filter per spring — emphasises resonant frequency
|
||||
coeff = _compute_bpf_coeffs(
|
||||
SAMPLE_RATE / (delay_samps + 1), q_val, SAMPLE_RATE)
|
||||
b0, b1, b2, a1, a2 = coeff
|
||||
b_arr = np.array([b0, b1, b2], dtype=np.float64)
|
||||
a_arr = np.array([1.0, a1, a2], dtype=np.float64)
|
||||
f_prev = spring["filt_prev"]
|
||||
res, f_zf = lfilter(b_arr, a_arr,
|
||||
spring_out.astype(np.float64), zi=f_prev)
|
||||
spring["filt_prev"] = f_zf
|
||||
|
||||
# Apply resonance gain
|
||||
res = res * res_gain
|
||||
|
||||
# High-frequency absorption (one-pole LP on feedback path)
|
||||
lp_prev = spring["lp_prev"]
|
||||
damped = np.zeros_like(res)
|
||||
for i in range(len(res)):
|
||||
lp_prev = lp_prev * (1.0 - damp) + res[i] * damp
|
||||
damped[i] = lp_prev
|
||||
spring["lp_prev"] = float(lp_prev)
|
||||
|
||||
# Write back with feedback
|
||||
write_sig = delayed.astype(np.float64) + damped * fb
|
||||
dl.write_block(np.clip(write_sig, -1.0, 1.0).astype(np.float32))
|
||||
|
||||
wet += damped
|
||||
|
||||
wet /= len(state["spring_lines"])
|
||||
return np.clip(wet, -1.0, 1.0).astype(np.float32)
|
||||
|
||||
# ── Plate reverb ───────────────────────────────────────────────
|
||||
|
||||
def _reverb_plate(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Plate reverb — dense comb bank + allpass cascade + modulation.
|
||||
|
||||
Emulates a large metal plate using 12 parallel comb filters with
|
||||
modulated delay times for realism, followed by a two-stage allpass
|
||||
diffuser for smooth, dense decay.
|
||||
|
||||
Params:
|
||||
decay — 0.0-1.0, plate sustain length
|
||||
damping — 0.0-1.0, high-frequency absorption
|
||||
density — 0.0-1.0, diffusion density (controls allpass gain)
|
||||
"""
|
||||
decay = params.get("decay", 0.5)
|
||||
damping = params.get("damping", 0.4)
|
||||
density = params.get("density", 0.6)
|
||||
predelay_ms = params.get("predelay", 15.0)
|
||||
|
||||
if "combs" not in state:
|
||||
# 12 combs — more = denser plate sound
|
||||
comb_delays_ms = [5, 11, 19, 23, 34, 39, 42, 48, 53, 57, 62, 68]
|
||||
ap_delays_ms = [2, 5, 9, 13]
|
||||
state["combs"] = [
|
||||
_CombFilter(int(d * SAMPLE_RATE / 1000.0))
|
||||
for d in comb_delays_ms
|
||||
]
|
||||
state["allpasses"] = [
|
||||
_AllpassFilter(int(d * SAMPLE_RATE / 1000.0))
|
||||
for d in ap_delays_ms
|
||||
]
|
||||
state["predelay"] = _DelayLine(
|
||||
int(predelay_ms * SAMPLE_RATE / 1000.0) + 1)
|
||||
state["predelay"].write_block(np.zeros(BLOCK_SIZE))
|
||||
# Plate modulation LFO phase
|
||||
state["mod_phase"] = 0.0
|
||||
|
||||
combs: list[_CombFilter] = state["combs"]
|
||||
allpasses: list[_AllpassFilter] = state["allpasses"]
|
||||
predelay_line: _DelayLine = state["predelay"]
|
||||
|
||||
# Update comb parameters
|
||||
param_tag = (decay, damping, density)
|
||||
if state.get("_param_tag") != param_tag:
|
||||
fb = 0.3 + decay * 0.6 # 0.3-0.9
|
||||
damp = 0.05 + damping * 0.7 # 0.05-0.75
|
||||
ap_gain = 0.3 + density * 0.4 # 0.3-0.7
|
||||
for comb in combs:
|
||||
comb.feedback = min(fb, 0.94)
|
||||
comb.damping = min(damp, 0.85)
|
||||
for ap in allpasses:
|
||||
ap.gain = ap_gain
|
||||
state["_param_tag"] = param_tag
|
||||
|
||||
# Predelay
|
||||
delayed = predelay_line.read_block(
|
||||
float(predelay_ms * SAMPLE_RATE / 1000.0), len(buf))
|
||||
predelay_line.write_block(buf)
|
||||
|
||||
# Plate modulation: very slow LFO (0.15 Hz) for subtle detuning
|
||||
mod_phase = state.get("mod_phase", 0.0)
|
||||
delta = 0.15 / SAMPLE_RATE
|
||||
t = np.arange(len(buf), dtype=np.float64) * delta + mod_phase
|
||||
t %= 1.0
|
||||
state["mod_phase"] = float((t[-1] + delta) % 1.0)
|
||||
|
||||
# Comb filters (parallel)
|
||||
wet = np.zeros(len(buf), dtype=np.float64)
|
||||
for comb in combs:
|
||||
comb_out = comb.process(delayed)
|
||||
wet += comb_out.astype(np.float64, copy=False)
|
||||
wet /= len(combs)
|
||||
|
||||
# Allpass diffuser
|
||||
for ap in allpasses:
|
||||
wet = ap.process(wet)
|
||||
|
||||
# Apply modulation interpolation for subtle pitch wobble
|
||||
mod = 0.5 + 0.5 * np.sin(2.0 * np.pi * t) # 0-1 wobble
|
||||
mod_idx = np.arange(len(buf), dtype=np.float64) + (mod - 0.5) * 0.3
|
||||
mod_idx = np.clip(mod_idx, 0, len(buf) - 1)
|
||||
int_idx = mod_idx.astype(np.int32)
|
||||
frac = mod_idx - int_idx
|
||||
nxt = np.minimum(int_idx + 1, len(buf) - 1)
|
||||
wet = wet[int_idx] * (1.0 - frac) + wet[nxt] * frac
|
||||
|
||||
return np.clip(wet, -1.0, 1.0).astype(np.float32)
|
||||
|
||||
# ── Room reverb ────────────────────────────────────────────────
|
||||
|
||||
def _reverb_room(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Room reverb — early reflection taps + late diffuse tail.
|
||||
|
||||
First 30-80 ms: distinct early reflections (scaled by room size).
|
||||
After: a small Schroeder-style late reverb for the diffuse tail.
|
||||
|
||||
Params:
|
||||
size — 0.0-1.0, room dimensions (scales all delays)
|
||||
decay — 0.0-1.0, reverb tail length
|
||||
damping — 0.0-1.0, high-frequency absorption
|
||||
"""
|
||||
size = params.get("size", 0.5)
|
||||
decay = params.get("decay", 0.4)
|
||||
damping = params.get("damping", 0.4)
|
||||
predelay_ms = params.get("predelay", 5.0)
|
||||
|
||||
size_factor = 0.3 + size * 1.7 # 0.3-2.0
|
||||
|
||||
if "er_delay" not in state:
|
||||
# Early reflection tap times (ms) — room-appropriate spacing
|
||||
base_taps_ms = [3, 7, 12, 18, 26, 36, 48, 62]
|
||||
# Small late comb/allpass for diffuse tail
|
||||
tail_comb_ms = [21, 29, 37, 44]
|
||||
tail_ap_ms = [4, 8, 13]
|
||||
max_tap = int(max(base_taps_ms) * size_factor
|
||||
* SAMPLE_RATE / 1000.0)
|
||||
state["er_delay"] = _DelayLine(max_tap + BLOCK_SIZE + 1)
|
||||
state["er_taps"] = [
|
||||
int(t * size_factor * SAMPLE_RATE / 1000.0)
|
||||
for t in base_taps_ms
|
||||
]
|
||||
state["tail_combs"] = [
|
||||
_CombFilter(int(d * size_factor * SAMPLE_RATE / 1000.0 + 1))
|
||||
for d in tail_comb_ms
|
||||
]
|
||||
state["tail_allpasses"] = [
|
||||
_AllpassFilter(
|
||||
int(d * size_factor * SAMPLE_RATE / 1000.0 + 1))
|
||||
for d in tail_ap_ms
|
||||
]
|
||||
state["predelay"] = _DelayLine(
|
||||
int(predelay_ms * SAMPLE_RATE / 1000.0) + 1)
|
||||
state["predelay"].write_block(np.zeros(BLOCK_SIZE))
|
||||
|
||||
predelay_line: _DelayLine = state["predelay"]
|
||||
delayed = predelay_line.read_block(
|
||||
float(predelay_ms * SAMPLE_RATE / 1000.0), len(buf))
|
||||
predelay_line.write_block(buf)
|
||||
|
||||
er_delay: _DelayLine = state["er_delay"]
|
||||
er_delay.write_block(delayed)
|
||||
|
||||
# ── Early reflections ──
|
||||
taps = state["er_taps"]
|
||||
num_taps = len(taps)
|
||||
er = np.zeros(len(buf), dtype=np.float64)
|
||||
amp = 0.5 + decay * 0.4 # 0.5-0.9
|
||||
|
||||
for i, tap in enumerate(taps):
|
||||
tap_gain = amp * (1.0 - i * 0.6 / max(num_taps, 1))
|
||||
tap_gain = max(tap_gain, 0.0)
|
||||
reflected = er_delay.read_block(float(tap), len(buf))
|
||||
er += reflected.astype(np.float64) * tap_gain
|
||||
|
||||
er_sum = sum(1.0 - i * 0.6 / num_taps for i in range(num_taps))
|
||||
if er_sum > 0:
|
||||
er /= er_sum
|
||||
|
||||
# ── Late reverb tail ──
|
||||
tail_combs: list[_CombFilter] = state["tail_combs"]
|
||||
tail_ap: list[_AllpassFilter] = state["tail_allpasses"]
|
||||
|
||||
param_tag = (decay, damping, size)
|
||||
if state.get("_tag_tail") != param_tag:
|
||||
fb = 0.2 + decay * 0.6 # 0.2-0.8
|
||||
damp = 0.1 + damping * 0.6 # 0.1-0.7
|
||||
for c in tail_combs:
|
||||
c.feedback = min(fb, 0.92)
|
||||
c.damping = min(damp, 0.85)
|
||||
for ap in tail_ap:
|
||||
ap.gain = 0.3 + damping * 0.2
|
||||
state["_tag_tail"] = param_tag
|
||||
|
||||
tail = np.zeros(len(buf), dtype=np.float64)
|
||||
for c in tail_combs:
|
||||
tail += c.process(delayed)
|
||||
tail /= len(tail_combs)
|
||||
for ap in tail_ap:
|
||||
tail = ap.process(tail)
|
||||
|
||||
# Blend: early reflections dominate early, tail fills in
|
||||
fade_len = int(len(buf) * 0.3)
|
||||
if fade_len > 0:
|
||||
blend = np.ones(len(buf), dtype=np.float64)
|
||||
blend[:fade_len] = np.linspace(0.0, 1.0, fade_len)
|
||||
wet = er * (1.0 - blend * 0.3) + tail * blend
|
||||
else:
|
||||
wet = er + tail * 0.5
|
||||
|
||||
return np.clip(wet, -1.0, 1.0).astype(np.float32)
|
||||
|
||||
# ── 12. Volume ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
Reference in New Issue
Block a user