Initial scaffold: Pi Multi-FX Pedal with NAM A2, IR cab, multi-FX, MIDI, stomp UI
This commit is contained in:
@@ -0,0 +1,824 @@
|
||||
"""FX/Audio pipeline — the main real-time signal chain.
|
||||
|
||||
Runs on RPi 4B under JACK, connecting:
|
||||
Guitar -> Gate -> Comp -> Boost -> NAM Amp -> IR Cab -> EQ -> Mod -> Delay -> Reverb -> Volume -> Out
|
||||
|
||||
Each block can be bypassed per-preset. The pipeline manages
|
||||
block-level audio routing using numpy arrays for zero-copy
|
||||
inter-block communication.
|
||||
|
||||
All DSP state is stored per-block-instance in self._state,
|
||||
keyed by chain index. This allows multiple instances of the
|
||||
same effect type at different positions in the chain.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Optional
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .nam_host import NAMHost, NAMModel
|
||||
from .ir_loader import IRLoader, IRFile
|
||||
from ..presets.types import FXBlock, FXType, Preset
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
BLOCK_SIZE = 256 # Samples per JACK callback
|
||||
SAMPLE_RATE = 48000 # Standard guitar audio rate
|
||||
|
||||
# ── Biquad coefficient helpers ─────────────────────────────────────
|
||||
|
||||
_EPS = 1e-10
|
||||
|
||||
|
||||
def _compute_lowshelf_coeffs(freq: float, gain_db: float, q: float, sr: float) -> tuple:
|
||||
"""RBJ low-shelf biquad coefficients."""
|
||||
a = 10 ** (gain_db / 40.0)
|
||||
omega = 2 * np.pi * freq / sr
|
||||
sn = np.sin(omega)
|
||||
cs = np.cos(omega)
|
||||
beta = np.sqrt(a) / q # sqrt(A) / Q
|
||||
if gain_db >= 0:
|
||||
b0 = a * (a + 1 - (a - 1) * cs + beta * sn)
|
||||
b1 = 2 * a * (a - 1 - (a + 1) * cs)
|
||||
b2 = a * (a + 1 - (a - 1) * cs - beta * sn)
|
||||
a0 = a + 1 + (a - 1) * cs + beta * sn
|
||||
a1 = -2 * a * (a - 1 + (a + 1) * cs)
|
||||
a2 = a + 1 + (a - 1) * cs - beta * sn
|
||||
else:
|
||||
b0 = a * (a + 1 + (a - 1) * cs + beta * sn)
|
||||
b1 = -2 * a * (a - 1 + (a + 1) * cs)
|
||||
b2 = a * (a + 1 + (a - 1) * cs - beta * sn)
|
||||
a0 = a + 1 - (a - 1) * cs + beta * sn
|
||||
a1 = 2 * a * (a - 1 - (a + 1) * cs)
|
||||
a2 = a + 1 - (a - 1) * cs - beta * sn
|
||||
return (b0 / a0, b1 / a0, b2 / a0, a1 / a0, a2 / a0)
|
||||
|
||||
|
||||
def _compute_highshelf_coeffs(freq: float, gain_db: float, q: float, sr: float) -> tuple:
|
||||
"""RBJ high-shelf biquad coefficients."""
|
||||
a = 10 ** (gain_db / 40.0)
|
||||
omega = 2 * np.pi * freq / sr
|
||||
sn = np.sin(omega)
|
||||
cs = np.cos(omega)
|
||||
beta = np.sqrt(a) / q
|
||||
if gain_db >= 0:
|
||||
b0 = a * (a + 1 + (a - 1) * cs + beta * sn)
|
||||
b1 = -2 * a * (a - 1 + (a + 1) * cs)
|
||||
b2 = a * (a + 1 + (a - 1) * cs - beta * sn)
|
||||
a0 = a + 1 - (a - 1) * cs + beta * sn
|
||||
a1 = 2 * a * (a - 1 - (a + 1) * cs)
|
||||
a2 = a + 1 - (a - 1) * cs - beta * sn
|
||||
else:
|
||||
b0 = a * (a + 1 - (a - 1) * cs + beta * sn)
|
||||
b1 = 2 * a * (a - 1 - (a + 1) * cs)
|
||||
b2 = a * (a + 1 - (a - 1) * cs - beta * sn)
|
||||
a0 = a + 1 + (a - 1) * cs + beta * sn
|
||||
a1 = -2 * a * (a - 1 + (a + 1) * cs)
|
||||
a2 = a + 1 + (a - 1) * cs - beta * sn
|
||||
return (b0 / a0, b1 / a0, b2 / a0, a1 / a0, a2 / a0)
|
||||
|
||||
|
||||
def _compute_peaking_coeffs(freq: float, gain_db: float, q: float, sr: float) -> tuple:
|
||||
"""RBJ peaking biquad coefficients."""
|
||||
a = 10 ** (gain_db / 40.0)
|
||||
omega = 2 * np.pi * freq / sr
|
||||
sn = np.sin(omega)
|
||||
cs = np.cos(omega)
|
||||
alpha = sn / (2 * q)
|
||||
b0 = 1 + alpha * a
|
||||
b1 = -2 * cs
|
||||
b2 = 1 - alpha * a
|
||||
a0 = 1 + alpha / a
|
||||
a1 = -2 * cs
|
||||
a2 = 1 - alpha / a
|
||||
return (b0 / a0, b1 / a0, b2 / a0, a1 / a0, a2 / a0)
|
||||
|
||||
|
||||
# ── Circular delay line (block-vectorised) ─────────────────────────
|
||||
|
||||
class _DelayLine:
|
||||
"""Vectorised circular buffer with linear interpolation."""
|
||||
|
||||
__slots__ = ("buf", "max_len", "write_idx")
|
||||
|
||||
def __init__(self, max_delay_samples: int):
|
||||
self.buf = np.zeros(max_delay_samples, dtype=np.float32)
|
||||
self.max_len = max_delay_samples
|
||||
self.write_idx = 0
|
||||
|
||||
def write_block(self, block: np.ndarray) -> None:
|
||||
n = len(block)
|
||||
space = self.max_len - self.write_idx
|
||||
if n <= space:
|
||||
self.buf[self.write_idx:self.write_idx + n] = block
|
||||
else:
|
||||
first_part = n - space
|
||||
self.buf[self.write_idx:] = block[:space]
|
||||
self.buf[:first_part] = block[space:]
|
||||
self.write_idx = (self.write_idx + n) % self.max_len
|
||||
# Keep type: numpy automatically promotes on write into float32
|
||||
|
||||
def read_block(self, delay_samples: float, n_samples: int) -> np.ndarray:
|
||||
"""Read n_samples with linear interpolation at a fractional delay."""
|
||||
n_delay = int(delay_samples)
|
||||
frac = delay_samples - n_delay
|
||||
read_start = (self.write_idx - n_delay) % self.max_len
|
||||
indices = (read_start + np.arange(n_samples)) % self.max_len
|
||||
next_indices = (indices + 1) % self.max_len
|
||||
return self.buf[indices] * (1.0 - frac) + self.buf[next_indices] * frac
|
||||
|
||||
def add_to_block(self, block: np.ndarray, delay_samples: float,
|
||||
gain: float) -> np.ndarray:
|
||||
"""Add delayed + gained signal to block (for feedback loops)."""
|
||||
n_delay = int(delay_samples)
|
||||
frac = delay_samples - n_delay
|
||||
read_start = (self.write_idx - n_delay) % self.max_len
|
||||
indices = (read_start + np.arange(len(block))) % self.max_len
|
||||
next_indices = (indices + 1) % self.max_len
|
||||
delayed = self.buf[indices] * (1.0 - frac) + self.buf[next_indices] * frac
|
||||
return delayed * gain
|
||||
|
||||
def read_all(self) -> np.ndarray:
|
||||
"""Return the full buffer (for debugging / IR export)."""
|
||||
return self.buf.copy()
|
||||
|
||||
|
||||
# ── Schroeder reverb helpers ───────────────────────────────────────
|
||||
|
||||
class _CombFilter:
|
||||
"""Comb filter for Schroeder reverb."""
|
||||
|
||||
__slots__ = ("delay", "feedback", "damping", "damp_filt", "buf")
|
||||
|
||||
def __init__(self, delay_samples: int):
|
||||
self.delay = _DelayLine(delay_samples + 1)
|
||||
self.feedback: float = 0.5
|
||||
self.damping: float = 0.5 # low-pass damping coefficient
|
||||
self.damp_filt: float = 0.0 # state variable for damping
|
||||
self.buf = np.zeros(BLOCK_SIZE, dtype=np.float32)
|
||||
|
||||
def process(self, block: np.ndarray) -> np.ndarray:
|
||||
self.buf[:] = block
|
||||
# Write with feedback: out[n] = in[n] + feedback * damped_delayed
|
||||
delayed = self.delay.add_to_block(self.buf, self.delay.max_len - 1, self.feedback)
|
||||
# One-pole low-pass on feedback path
|
||||
damped = np.zeros_like(delayed)
|
||||
for i in range(len(delayed)):
|
||||
self.damp_filt = (1.0 - self.damping) * delayed[i] + self.damping * self.damp_filt
|
||||
damped[i] = self.damp_filt
|
||||
self.buf[:] = block + damped
|
||||
self.delay.write_block(self.buf)
|
||||
return self.buf
|
||||
|
||||
|
||||
class _AllpassFilter:
|
||||
"""Allpass filter for Schroeder reverb."""
|
||||
|
||||
__slots__ = ("delay", "gain", "buf")
|
||||
|
||||
def __init__(self, delay_samples: int):
|
||||
self.delay = _DelayLine(delay_samples + 1)
|
||||
self.gain: float = 0.5
|
||||
self.buf = np.zeros(BLOCK_SIZE, dtype=np.float32)
|
||||
|
||||
def process(self, block: np.ndarray) -> np.ndarray:
|
||||
# out[n] = -gain * in[n] + delay[n - D] + gain * delay_output[n - D]
|
||||
# Standard allpass: out = -g * in + delayed + g * delayed_out
|
||||
# But block-wise: read delayed, write in + g * delayed, output = -g * in + delayed
|
||||
self.buf[:] = block
|
||||
delayed = self.delay.add_to_block(self.buf, self.delay.max_len - 1, self.gain)
|
||||
# Write: buf + gain * delayed
|
||||
self.buf[:] = block + delayed * self.gain
|
||||
self.delay.write_block(self.buf)
|
||||
# Output: -gain * block + delayed
|
||||
return -self.gain * block + delayed
|
||||
|
||||
|
||||
# ── Audio Pipeline ─────────────────────────────────────────────────
|
||||
|
||||
class AudioPipeline:
|
||||
"""Orchestrates the real-time audio FX chain.
|
||||
|
||||
The pipeline processes audio block-by-block, chaining
|
||||
effect modules in order. Each module receives a numpy
|
||||
array of audio samples and returns processed samples.
|
||||
Effect state (delay buffers, LFO phases, envelope followers,
|
||||
filter memory) is stored per-instance in self._state.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
nam_host: Optional[NAMIHost] = None,
|
||||
ir_loader: Optional[IRLoader] = None,
|
||||
):
|
||||
self.nam = nam_host or NAMHost()
|
||||
self.ir = ir_loader or IRLoader()
|
||||
|
||||
# Signal chain — list of (FXType, enabled, bypass, params)
|
||||
self._chain: list[dict] = []
|
||||
self._master_volume: float = 0.8
|
||||
self._tuner_enabled: bool = False
|
||||
self._bypassed: bool = False # Global bypass
|
||||
|
||||
# Per-block DSP state: {f"fx_{idx}": {state_dict}}
|
||||
self._state: dict[str, dict] = {}
|
||||
|
||||
# Cached filter coefficients per block
|
||||
self._coeffs: dict[str, tuple] = {}
|
||||
|
||||
logger.info("Audio pipeline initialized (block=%d, sr=%d)",
|
||||
BLOCK_SIZE, SAMPLE_RATE)
|
||||
|
||||
def load_preset(self, preset: Preset) -> None:
|
||||
"""Load a complete preset (NAM, IR, and FX chain)."""
|
||||
self._chain = []
|
||||
self._state = {}
|
||||
self._coeffs = {}
|
||||
|
||||
for block in preset.chain:
|
||||
entry = {
|
||||
"fx_type": block.fx_type,
|
||||
"enabled": block.enabled,
|
||||
"bypass": block.bypass,
|
||||
"params": dict(block.params),
|
||||
}
|
||||
|
||||
# Load NAM model if needed
|
||||
if block.fx_type == FXType.NAM_AMP and block.nam_model_path:
|
||||
self.nam.load_model(block.nam_model_path)
|
||||
|
||||
# Load IR if needed
|
||||
if block.fx_type == FXType.IR_CAB and block.ir_file_path:
|
||||
self.ir.load_ir(block.ir_file_path)
|
||||
|
||||
self._chain.append(entry)
|
||||
|
||||
self._master_volume = preset.master_volume
|
||||
self._tuner_enabled = preset.tuner_enabled
|
||||
|
||||
logger.info("Preset '%s' loaded: %d blocks", preset.name, len(self._chain))
|
||||
|
||||
def process(self, audio_in: np.ndarray) -> np.ndarray:
|
||||
"""Process a block of audio through the entire FX chain.
|
||||
|
||||
Args:
|
||||
audio_in: numpy array of PCM samples (float32 [-1, 1]).
|
||||
|
||||
Returns:
|
||||
Processed audio block.
|
||||
"""
|
||||
if self._bypassed:
|
||||
return audio_in * self._master_volume
|
||||
|
||||
buf = audio_in.copy()
|
||||
|
||||
for idx, entry in enumerate(self._chain):
|
||||
if entry["bypass"] or not entry["enabled"]:
|
||||
continue
|
||||
|
||||
fx_type = entry["fx_type"]
|
||||
params = entry["params"]
|
||||
fx_state = self._state.setdefault(f"fx_{idx}", {})
|
||||
|
||||
match fx_type:
|
||||
case FXType.NOISE_GATE:
|
||||
buf = self._apply_gate(buf, params, fx_state)
|
||||
case FXType.COMPRESSOR:
|
||||
buf = self._apply_compressor(buf, params, fx_state)
|
||||
case FXType.BOOST:
|
||||
buf = self._apply_boost(buf, params, fx_state)
|
||||
case FXType.OVERDRIVE:
|
||||
buf = self._apply_overdrive(buf, params, fx_state)
|
||||
case FXType.DISTORTION:
|
||||
buf = self._apply_distortion(buf, params, fx_state)
|
||||
case FXType.FUZZ:
|
||||
buf = self._apply_fuzz(buf, params, fx_state)
|
||||
case FXType.EQ:
|
||||
buf = self._apply_eq(buf, params, fx_state)
|
||||
case FXType.CHORUS:
|
||||
buf = self._apply_chorus(buf, params, fx_state)
|
||||
case FXType.FLANGER:
|
||||
buf = self._apply_flanger(buf, params, fx_state)
|
||||
case FXType.PHASER:
|
||||
buf = self._apply_phaser(buf, params, fx_state)
|
||||
case FXType.TREMOLO:
|
||||
buf = self._apply_tremolo(buf, params, fx_state)
|
||||
case FXType.VIBRATO:
|
||||
buf = self._apply_vibrato(buf, params, fx_state)
|
||||
case FXType.DELAY:
|
||||
buf = self._apply_delay(buf, params, fx_state)
|
||||
case FXType.REVERB:
|
||||
buf = self._apply_reverb(buf, params, fx_state)
|
||||
case FXType.VOLUME:
|
||||
buf = self._apply_volume(buf, params, fx_state)
|
||||
case _:
|
||||
pass # NAM/IR handled externally
|
||||
|
||||
return buf * self._master_volume
|
||||
|
||||
# ── LFO helpers ─────────────────────────────────────────────────
|
||||
|
||||
@staticmethod
|
||||
def _lfo_phase(rate_hz: float, state: dict, block_size: int) -> np.ndarray:
|
||||
"""Generate LFO phase ramp (0->1), update state."""
|
||||
phase = state.get("phase", 0.0)
|
||||
delta = rate_hz / SAMPLE_RATE
|
||||
t = np.arange(block_size, dtype=np.float64) * delta + phase
|
||||
t %= 1.0
|
||||
state["phase"] = float(t[-1] + delta) % 1.0
|
||||
return t
|
||||
|
||||
@staticmethod
|
||||
def _lfo_wave(phase: np.ndarray, shape: str = "sine") -> np.ndarray:
|
||||
"""Generate LFO waveform from phase array."""
|
||||
match shape:
|
||||
case "sine":
|
||||
return 0.5 + 0.5 * np.sin(2 * np.pi * phase)
|
||||
case "triangle":
|
||||
return 2.0 * np.abs(2.0 * phase - 1.0) - 1.0
|
||||
# Returns in [-1, 1]; normalise below
|
||||
case "square":
|
||||
return np.where(phase < 0.5, 1.0, 0.0)
|
||||
case _:
|
||||
return 0.5 + 0.5 * np.sin(2 * np.pi * phase)
|
||||
|
||||
# ── Effect implementations ──────────────────────────────────────
|
||||
|
||||
# ── 1. Noise Gate ───────────────────────────────────────────────
|
||||
|
||||
def _apply_gate(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Noise gate with adjustable threshold and release envelope."""
|
||||
threshold = params.get("threshold", 0.01)
|
||||
release_ms = params.get("release", 100.0)
|
||||
|
||||
envelope = state.get("envelope", 0.0)
|
||||
rms = np.sqrt(np.mean(buf ** 2) + _EPS)
|
||||
|
||||
if rms >= threshold:
|
||||
# Instant attack
|
||||
envelope = rms
|
||||
else:
|
||||
# Exponential release — time constant per block
|
||||
release_coeff = np.exp(-BLOCK_SIZE / (release_ms * SAMPLE_RATE / 1000.0))
|
||||
envelope = envelope * release_coeff + rms * (1.0 - release_coeff)
|
||||
|
||||
state["envelope"] = envelope
|
||||
|
||||
if envelope < threshold:
|
||||
return np.zeros_like(buf)
|
||||
return buf
|
||||
|
||||
# ── 2. Compressor ───────────────────────────────────────────────
|
||||
|
||||
def _apply_compressor(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Compressor with threshold (dB), ratio, attack, release, make-up gain."""
|
||||
threshold_db = params.get("threshold", -20.0) # dB
|
||||
ratio = params.get("ratio", 3.0)
|
||||
attack_ms = params.get("attack", 5.0)
|
||||
release_ms = params.get("release", 100.0)
|
||||
makeup = params.get("gain", 1.0)
|
||||
|
||||
# RMS envelope with attack/release shaping
|
||||
rms = np.sqrt(np.mean(buf ** 2) + _EPS)
|
||||
envelope = state.get("envelope", 0.0)
|
||||
|
||||
if rms > envelope:
|
||||
alpha = np.exp(-BLOCK_SIZE / (attack_ms * SAMPLE_RATE / 1000.0))
|
||||
else:
|
||||
alpha = np.exp(-BLOCK_SIZE / (release_ms * SAMPLE_RATE / 1000.0))
|
||||
|
||||
envelope = envelope * alpha + rms * (1.0 - alpha)
|
||||
state["envelope"] = envelope
|
||||
|
||||
# Compute gain reduction in dB domain
|
||||
if envelope > 1e-10:
|
||||
env_db = 20.0 * np.log10(envelope)
|
||||
else:
|
||||
env_db = -120.0
|
||||
|
||||
if env_db > threshold_db:
|
||||
# gain_db = threshold + (env - threshold) / ratio - env
|
||||
gain_db = threshold_db + (env_db - threshold_db) / ratio - env_db
|
||||
else:
|
||||
gain_db = 0.0
|
||||
|
||||
gain_lin = 10 ** (gain_db / 20.0)
|
||||
return np.clip(buf * gain_lin * makeup, -1.0, 1.0)
|
||||
|
||||
# ── 3. Boost / Overdrive / Distortion / Fuzz ────────────────────
|
||||
|
||||
def _apply_boost(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Clean boost with linear gain."""
|
||||
gain_db = params.get("gain_db", 6.0)
|
||||
gain_linear = 10 ** (gain_db / 20.0)
|
||||
return np.clip(buf * gain_linear, -1.0, 1.0)
|
||||
|
||||
def _apply_overdrive(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Tube-style overdrive with asymmetric soft clipping."""
|
||||
drive = params.get("drive", 0.5)
|
||||
tone = params.get("tone", 0.5)
|
||||
gain = params.get("gain", 1.0)
|
||||
|
||||
drive_scaled = drive * 15.0 + 1.0
|
||||
shaped = buf * drive_scaled
|
||||
|
||||
# Asymmetric soft clipping (tube-like)
|
||||
# Positive half clips softer than negative (tube asymmetry)
|
||||
pos = np.where(shaped > 0, shaped / (1.0 + shaped * 0.3), shaped)
|
||||
neg = np.where(pos < 0, pos / (1.0 - pos * 0.5), pos)
|
||||
out = np.tanh(neg) # Final polish with tanh
|
||||
|
||||
return np.clip(out * gain, -1.0, 1.0)
|
||||
|
||||
def _apply_distortion(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Harder asymmetric clipping with diode-style transfer."""
|
||||
drive = params.get("drive", 0.7)
|
||||
tone = params.get("tone", 0.5)
|
||||
gain = params.get("gain", 1.0)
|
||||
|
||||
drive_scaled = drive * 30.0 + 1.0
|
||||
shaped = buf * drive_scaled
|
||||
|
||||
# Diode-style asymmetric clipping
|
||||
clipped = np.where(
|
||||
shaped > 0,
|
||||
np.clip(shaped, 0, 0.8) / (1.0 + np.abs(np.clip(shaped, 0, 0.8)) * 0.5),
|
||||
np.clip(shaped, -0.6, 0) / (1.0 + np.abs(np.clip(shaped, -0.6, 0)) * 0.3),
|
||||
)
|
||||
return np.clip(clipped * gain, -1.0, 1.0)
|
||||
|
||||
def _apply_fuzz(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Octave-fuzzy hard clipping with gated sustain."""
|
||||
drive = params.get("drive", 0.8)
|
||||
tone = params.get("tone", 0.5)
|
||||
gain = params.get("gain", 1.0)
|
||||
|
||||
drive_scaled = drive * 50.0 + 1.0
|
||||
shaped = buf * drive_scaled
|
||||
|
||||
# Hard square-wave clip with asymmetric gate
|
||||
clipped = np.sign(shaped) * (1.0 - np.exp(-np.abs(shaped) * 2.0))
|
||||
# Foldover for octave effect
|
||||
folded = np.abs(clipped) * 0.3 + clipped * 0.7
|
||||
return np.clip(folded * gain, -1.0, 1.0)
|
||||
|
||||
# ── 4. Three-band EQ ────────────────────────────────────────────
|
||||
|
||||
def _apply_eq(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""3-band EQ: bass shelf, mid peaking, treble shelf."""
|
||||
bass_gain = params.get("bass", 0.0) # dB
|
||||
mid_gain = params.get("mid", 0.0) # dB
|
||||
treble_gain = params.get("treble", 0.0) # dB
|
||||
bass_freq = params.get("bass_freq", 200.0)
|
||||
mid_freq = params.get("mid_freq", 1000.0)
|
||||
treble_freq = params.get("treble_freq", 3500.0)
|
||||
q = params.get("q", 0.707)
|
||||
|
||||
sig = buf.astype(np.float64, copy=False)
|
||||
|
||||
# Cache biquad coefficients per block position — recompute only
|
||||
# when params change (checked via hash). Each band gets its own
|
||||
# state sub-key.
|
||||
for band_name, freq, gain_db, compute_fn in [
|
||||
("bass", bass_freq, bass_gain, _compute_lowshelf_coeffs),
|
||||
("mid", mid_freq, mid_gain, _compute_peaking_coeffs),
|
||||
("treble", treble_freq, treble_gain, _compute_highshelf_coeffs),
|
||||
]:
|
||||
if gain_db == 0.0:
|
||||
continue
|
||||
key = f"eq_{band_name}"
|
||||
coeffs = state.get(f"{key}_coeffs")
|
||||
param_tag = (bass_freq, mid_freq, treble_freq, bass_gain, mid_gain, treble_gain, q)
|
||||
if coeffs is None or state.get(f"{key}_tag") != param_tag:
|
||||
coeffs = compute_fn(freq, gain_db, q, SAMPLE_RATE)
|
||||
state[f"{key}_coeffs"] = coeffs
|
||||
state[f"{key}_tag"] = param_tag
|
||||
|
||||
b0, b1, b2, a1, a2 = coeffs
|
||||
x1 = state.get(f"{key}_x1", 0.0)
|
||||
x2 = state.get(f"{key}_x2", 0.0)
|
||||
y1 = state.get(f"{key}_y1", 0.0)
|
||||
y2 = state.get(f"{key}_y2", 0.0)
|
||||
|
||||
for i in range(len(sig)):
|
||||
x0 = sig[i]
|
||||
y0 = b0 * x0 + b1 * x1 + b2 * x2 - a1 * y1 - a2 * y2
|
||||
x2, x1 = x1, x0
|
||||
y2, y1 = y1, y0
|
||||
sig[i] = y0
|
||||
|
||||
state[f"{key}_x1"] = x1
|
||||
state[f"{key}_x2"] = x2
|
||||
state[f"{key}_y1"] = y1
|
||||
state[f"{key}_y2"] = y2
|
||||
|
||||
return np.clip(sig, -1.0, 1.0).astype(np.float32)
|
||||
|
||||
# ── 5. Chorus ───────────────────────────────────────────────────
|
||||
|
||||
def _apply_chorus(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Chorus with LFO-driven modulated delay line (stereo-ish)."""
|
||||
rate = params.get("rate", 0.5) # Hz
|
||||
depth = params.get("depth", 0.5) # 0.0-1.0
|
||||
mix = params.get("mix", 0.5) # wet/dry
|
||||
delay_base = params.get("delay", 20.0) # ms (typical chorus: 15-30ms)
|
||||
|
||||
# Convert to samples
|
||||
base_samples = delay_base * SAMPLE_RATE / 1000.0
|
||||
mod_range = depth * 5.0 * SAMPLE_RATE / 1000.0 # up to 5ms of modulation
|
||||
|
||||
if "delay" not in state:
|
||||
max_d = int(base_samples + mod_range + 10.0 * SAMPLE_RATE / 1000.0) + 1
|
||||
state["delay"] = _DelayLine(max_d)
|
||||
# Warm up delay buffer
|
||||
state["delay"].write_block(np.zeros(max_d))
|
||||
|
||||
delay_line: _DelayLine = state["delay"]
|
||||
phase = self._lfo_phase(rate, state, len(buf))
|
||||
lfo = self._lfo_wave(phase, "sine") # 0-1 range
|
||||
mod_delay = base_samples + lfo * mod_range
|
||||
|
||||
# Read modulated delayed signal
|
||||
wet = np.zeros_like(buf)
|
||||
for i in range(len(buf)):
|
||||
wet[i] = delay_line.read_block(mod_delay[i], 1)[0]
|
||||
|
||||
# Write dry to delay line
|
||||
delay_line.write_block(buf)
|
||||
|
||||
return buf * (1.0 - mix) + wet * mix
|
||||
|
||||
# ── 6. Flanger ──────────────────────────────────────────────────
|
||||
|
||||
def _apply_flanger(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Flanger with swept comb filter and feedback."""
|
||||
rate = params.get("rate", 0.25) # Hz (slower than chorus)
|
||||
depth = params.get("depth", 0.7) # 0.0-1.0
|
||||
feedback = params.get("feedback", 0.3)
|
||||
mix = params.get("mix", 0.5) # wet/dry
|
||||
delay_base = params.get("delay", 5.0) # ms (typical flanger: 1-10ms)
|
||||
|
||||
base_samples = delay_base * SAMPLE_RATE / 1000.0
|
||||
mod_range = depth * 5.0 * SAMPLE_RATE / 1000.0
|
||||
|
||||
if "delay" not in state:
|
||||
max_d = int(base_samples + mod_range + 10.0 * SAMPLE_RATE / 1000.0) + 1
|
||||
state["delay"] = _DelayLine(max_d)
|
||||
state["delay"].write_block(np.zeros(max_d))
|
||||
|
||||
delay_line: _DelayLine = state["delay"]
|
||||
phase = self._lfo_phase(rate, state, len(buf))
|
||||
lfo = self._lfo_wave(phase, "sine") # 0-1
|
||||
mod_delay = base_samples + lfo * mod_range
|
||||
|
||||
# Feedback buffer
|
||||
feedback_buf = state.get("fb_buf", np.zeros(len(buf), dtype=np.float32))
|
||||
|
||||
# Blend feedback into input
|
||||
fb_input = buf + feedback_buf * feedback
|
||||
|
||||
wet = np.zeros_like(buf)
|
||||
for i in range(len(fb_input)):
|
||||
wet[i] = delay_line.read_block(mod_delay[i], 1)[0]
|
||||
|
||||
delay_line.write_block(fb_input)
|
||||
|
||||
# Store feedback for next block
|
||||
state["fb_buf"] = wet * 0.5
|
||||
|
||||
return buf * (1.0 - mix) + wet * mix
|
||||
|
||||
# ── 7. Phaser ───────────────────────────────────────────────────
|
||||
|
||||
def _apply_phaser(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Phaser with allpass filter cascade and feedback."""
|
||||
rate = params.get("rate", 0.4) # Hz
|
||||
depth = params.get("depth", 0.5) # 0.0-1.0
|
||||
feedback = params.get("feedback", 0.3)
|
||||
mix = params.get("mix", 0.5)
|
||||
stages = int(params.get("stages", 4)) # number of allpass stages
|
||||
|
||||
# Map LFO to centre frequency sweep: 200-2000 Hz
|
||||
phase = self._lfo_phase(rate, state, len(buf))
|
||||
lfo = self._lfo_wave(phase, "sine")
|
||||
freq_range = 200.0 + lfo * depth * 1800.0 # 200-2000 Hz sweep
|
||||
|
||||
# Pre-compute allpass coefficients per sample
|
||||
fb_buf = state.get("fb_buf", np.zeros(len(buf), dtype=np.float64))
|
||||
fb_input = buf.astype(np.float64, copy=False) + fb_buf * feedback
|
||||
|
||||
out = np.zeros(len(buf), dtype=np.float64)
|
||||
|
||||
for i in range(len(buf)):
|
||||
freq = freq_range[i]
|
||||
# Allpass coefficient: a = (1 - tan(w/2)) / (1 + tan(w/2))
|
||||
w = 2.0 * np.pi * freq / SAMPLE_RATE
|
||||
tan_half_w = np.tan(w / 2.0)
|
||||
coeff = (1.0 - tan_half_w) / (1.0 + tan_half_w)
|
||||
|
||||
x = fb_input[i]
|
||||
for stage in range(stages):
|
||||
# Load state for this stage
|
||||
s_delay = state.get(f"ap_delay_{stage}", 0.0)
|
||||
s_out = state.get(f"ap_out_{stage}", 0.0)
|
||||
# Allpass: out[n] = coeff * in[n] + delay[n-1] - coeff * out[n-1]
|
||||
y = coeff * x + s_delay - coeff * s_out
|
||||
state[f"ap_delay_{stage}"] = x
|
||||
state[f"ap_out_{stage}"] = y
|
||||
x = y
|
||||
out[i] = x
|
||||
|
||||
state["fb_buf"] = out * 0.5
|
||||
out = np.clip(out, -1.0, 1.0)
|
||||
return (buf * (1.0 - mix) + out.astype(np.float32) * mix).astype(np.float32)
|
||||
|
||||
# ── 8. Tremolo ──────────────────────────────────────────────────
|
||||
|
||||
def _apply_tremolo(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Tremolo with configurable LFO shape."""
|
||||
rate = params.get("rate", 4.0) # Hz
|
||||
depth = params.get("depth", 0.7)
|
||||
shape = params.get("shape", "sine") # sine / triangle / square
|
||||
|
||||
phase = self._lfo_phase(rate, state, len(buf))
|
||||
lfo = self._lfo_wave(phase, shape)
|
||||
|
||||
# LFO is 0-1; tremolo scales between full volume and attenuated
|
||||
mod = 1.0 - depth * (1.0 - lfo)
|
||||
return buf * mod
|
||||
|
||||
# ── 9. Vibrato ──────────────────────────────────────────────────
|
||||
|
||||
def _apply_vibrato(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Vibrato — modulated delay with 100% wet (pitch modulation)."""
|
||||
rate = params.get("rate", 3.0) # Hz
|
||||
depth = params.get("depth", 0.5) # cents equivalent
|
||||
|
||||
base_samples = 2.0 * SAMPLE_RATE / 1000.0 # fixed ~2ms base
|
||||
mod_range = depth * 3.0 * SAMPLE_RATE / 1000.0
|
||||
|
||||
if "delay" not in state:
|
||||
max_d = int(base_samples + mod_range + 5.0 * SAMPLE_RATE / 1000.0) + 1
|
||||
state["delay"] = _DelayLine(max_d)
|
||||
state["delay"].write_block(np.zeros(max_d))
|
||||
|
||||
delay_line: _DelayLine = state["delay"]
|
||||
phase = self._lfo_phase(rate, state, len(buf))
|
||||
lfo = self._lfo_wave(phase, "sine")
|
||||
mod_delay = base_samples + lfo * mod_range
|
||||
|
||||
wet = np.zeros_like(buf)
|
||||
for i in range(len(buf)):
|
||||
wet[i] = delay_line.read_block(mod_delay[i], 1)[0]
|
||||
|
||||
delay_line.write_block(buf)
|
||||
return wet
|
||||
|
||||
# ── 10. Delay ───────────────────────────────────────────────────
|
||||
|
||||
def _apply_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)
|
||||
mix = params.get("mix", 0.4)
|
||||
tap_tempo = params.get("tap_tempo", 0.0)
|
||||
|
||||
# Tap tempo overrides time_ms when > 0
|
||||
if tap_tempo > 0:
|
||||
time_ms = tap_tempo
|
||||
|
||||
delay_samples = int(time_ms * SAMPLE_RATE / 1000.0)
|
||||
|
||||
if "delay" not in state:
|
||||
# Allocate 2x requested delay for headroom
|
||||
max_d = max(delay_samples * 2, SAMPLE_RATE) # at least 1s
|
||||
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))
|
||||
|
||||
# Write dry + feedback (no self-oscillation guard)
|
||||
# clips feedback automatically
|
||||
fb_gain = min(feedback, 0.98)
|
||||
write_sig = buf + wet * fb_gain
|
||||
delay_line.write_block(write_sig)
|
||||
|
||||
return buf * (1.0 - mix) + wet * mix
|
||||
|
||||
# ── 11. Reverb (Schroeder) ──────────────────────────────────────
|
||||
|
||||
def _apply_reverb(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Schroeder reverb: 8 comb filters + 4 allpass filters in series."""
|
||||
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"] = [
|
||||
_CombFilter(int(d * SAMPLE_RATE / 1000.0))
|
||||
for d in comb_delays
|
||||
]
|
||||
state["allpasses"] = [
|
||||
_AllpassFilter(int(d * SAMPLE_RATE / 1000.0))
|
||||
for d in ap_delays
|
||||
]
|
||||
state["predelay"] = _DelayLine(
|
||||
int(predelay_ms * SAMPLE_RATE / 1000.0) + 1
|
||||
)
|
||||
state["predelay"].write_block(np.zeros(BLOCK_SIZE))
|
||||
state["_computed"] = False
|
||||
|
||||
combs: list[_CombFilter] = state["combs"]
|
||||
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
|
||||
for comb in combs:
|
||||
comb.feedback = min(scaled_fb, 0.95)
|
||||
comb.damping = min(scaled_damp, 0.85)
|
||||
for ap in allpasses:
|
||||
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))
|
||||
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
|
||||
|
||||
# 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
|
||||
|
||||
# ── 12. Volume ──────────────────────────────────────────────────
|
||||
|
||||
def _apply_volume(self, buf: np.ndarray, params: dict,
|
||||
state: dict) -> np.ndarray:
|
||||
"""Simple volume/level control."""
|
||||
level = params.get("level", 1.0)
|
||||
return buf * level
|
||||
|
||||
# ── Properties ─────────────────────────────────────────────────
|
||||
|
||||
@property
|
||||
def master_volume(self) -> float:
|
||||
return self._master_volume
|
||||
|
||||
@master_volume.setter
|
||||
def master_volume(self, value: float) -> None:
|
||||
self._master_volume = max(0.0, min(1.0, value))
|
||||
|
||||
@property
|
||||
def bypassed(self) -> bool:
|
||||
return self._bypassed
|
||||
|
||||
@bypassed.setter
|
||||
def bypassed(self, value: bool) -> None:
|
||||
self._bypassed = value
|
||||
logger.info("Global bypass: %s", "ON" if value else "OFF")
|
||||
|
||||
@property
|
||||
def tuner_enabled(self) -> bool:
|
||||
return self._tuner_enabled
|
||||
|
||||
@tuner_enabled.setter
|
||||
def tuner_enabled(self, value: bool) -> None:
|
||||
self._tuner_enabled = value
|
||||
Reference in New Issue
Block a user