shawn
8ff584cea9
Pipeline: - process() dispatches to _process_mono() or _process_4cm() based on routing_mode - _process_4cm() splits chain at routing_breakpoint: pre blocks on ch0, post on ch1 - _process_single_block() extracted for reuse in both mono and 4cm paths - routing_mode/routing_breakpoint load from preset via load_preset() - set_routing() for runtime configuration - Properties: routing_mode (mono|4cm), routing_breakpoint with validation Web server: - GET /api/routing — current routing mode and breakpoint - POST /api/routing — set routing mode/breakpoint, persist to current preset, WS broadcast - _gather_state() includes routing_mode and routing_breakpoint Web UI: - Settings page: 4CM toggle + breakpoint slider with routing description - Dashboard: routing badge (4CM/Mono) with breakpoint info - app.js: WebSocket handler, updateRoutingUI(), toggle4cm(), set4cmBreakpoint() - style.css: .badge, .badge-4cm, .badge-mono, .routing-status, .routing-desc Tests: mock pipeline fixture updated with routing_mode/routing_breakpoint
952 lines
38 KiB
Python
952 lines
38 KiB
Python
"""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 scipy.signal import lfilter
|
|
|
|
from .nam_host import NAMHost, NAMModel, ModelSwitchMode
|
|
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)
|
|
pos = 0
|
|
while pos < n:
|
|
if self.write_idx >= self.max_len:
|
|
self.write_idx = 0
|
|
space = self.max_len - self.write_idx
|
|
chunk = min(n - pos, space)
|
|
self.buf[self.write_idx:self.write_idx + chunk] = block[pos:pos + chunk]
|
|
self.write_idx += chunk
|
|
pos += chunk
|
|
# Keep type: numpy automatically promotes on write into float32
|
|
|
|
def read_block_varying(self, delay_samples: np.ndarray) -> np.ndarray:
|
|
"""Read a block with different (fractional) delay per sample.
|
|
|
|
Fully vectorized using numpy advanced indexing.
|
|
``delay_samples`` must have shape (N,) or broadcastable to it.
|
|
"""
|
|
delays = np.asarray(delay_samples, dtype=np.float64)
|
|
int_delays = np.floor(delays).astype(np.int32)
|
|
frac = delays - int_delays
|
|
read_start = (self.write_idx - int_delays) % self.max_len
|
|
read_next = (read_start + 1) % self.max_len
|
|
return (self.buf[read_start] * (1.0 - frac)
|
|
+ self.buf[read_next] * frac)
|
|
|
|
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):
|
|
line_len = max(BLOCK_SIZE * 2, delay_samples + 1)
|
|
self.delay = _DelayLine(line_len)
|
|
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 (vectorised)
|
|
b = np.array([1.0 - self.damping], dtype=np.float64)
|
|
a = np.array([1.0, -self.damping], dtype=np.float64)
|
|
damped, _ = lfilter(b, a, delayed.astype(np.float64), zi=np.atleast_1d(self.damp_filt))
|
|
self.damp_filt = float(damped[-1])
|
|
self.buf[:] = block + damped.astype(np.float32)
|
|
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):
|
|
line_len = max(BLOCK_SIZE * 2, delay_samples + 1)
|
|
self.delay = _DelayLine(line_len)
|
|
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[NAMHost] = 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
|
|
|
|
# 4-Cable Method routing
|
|
self._routing_mode: str = "mono" # "mono" or "4cm"
|
|
self._routing_breakpoint: int = 7 # chain index where pre/post split occurs
|
|
|
|
# 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
|
|
|
|
# Set 4CM routing from preset
|
|
self._routing_mode = preset.routing_mode
|
|
self._routing_breakpoint = preset.routing_breakpoint
|
|
|
|
logger.info("Preset '%s' loaded: %d blocks, routing=%s breakpoint=%d",
|
|
preset.name, len(self._chain),
|
|
self._routing_mode, self._routing_breakpoint)
|
|
|
|
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]).
|
|
Mono mode: shape (N,) — single audio channel.
|
|
4CM mode: shape (2, N) — two channels, [guitar_in, return_in].
|
|
|
|
Returns:
|
|
Processed audio block.
|
|
Mono mode: shape (N,) — processed output.
|
|
4CM mode: shape (2, N) — [send_out, return_out].
|
|
"""
|
|
if self._bypassed:
|
|
return audio_in * self._master_volume
|
|
|
|
if self._routing_mode == "4cm":
|
|
return self._process_4cm(audio_in)
|
|
else:
|
|
return self._process_mono(audio_in)
|
|
|
|
def _process_mono(self, audio_in: np.ndarray) -> np.ndarray:
|
|
"""Process a mono block through the full chain (all blocks)."""
|
|
buf = audio_in.copy()
|
|
for idx, entry in enumerate(self._chain):
|
|
if entry["bypass"] or not entry["enabled"]:
|
|
continue
|
|
buf = self._process_single_block(buf, idx, entry)
|
|
return buf * self._master_volume
|
|
|
|
def _process_4cm(self, audio_in: np.ndarray) -> np.ndarray:
|
|
"""Process stereo block with 4CM split routing.
|
|
|
|
audio_in has shape (2, N):
|
|
ch0 = guitar input (Input 1)
|
|
ch1 = FX loop return (Input 2)
|
|
|
|
Splits at _routing_breakpoint:
|
|
pre blocks → ch0 processed through [0..breakpoint)
|
|
post blocks → ch1 processed through [breakpoint..]
|
|
|
|
Returns (2, N):
|
|
ch0 = Send output (to amp input)
|
|
ch1 = Return output (to amp FX return)
|
|
"""
|
|
ch0 = audio_in[0, :].copy()
|
|
ch1 = audio_in[1, :].copy()
|
|
bp = self._routing_breakpoint
|
|
|
|
for idx, entry in enumerate(self._chain):
|
|
if entry["bypass"] or not entry["enabled"]:
|
|
continue
|
|
|
|
if idx < bp:
|
|
# Pre-amp block — process on guitar (ch0)
|
|
ch0 = self._process_single_block(ch0, idx, entry)
|
|
else:
|
|
# Post-amp block — process on return (ch1)
|
|
ch1 = self._process_single_block(ch1, idx, entry)
|
|
|
|
out = np.zeros_like(audio_in)
|
|
out[0, :] = ch0 * self._master_volume
|
|
out[1, :] = ch1 * self._master_volume
|
|
return out
|
|
|
|
def _process_single_block(self, buf: np.ndarray, idx: int,
|
|
entry: dict) -> np.ndarray:
|
|
"""Process a single mono audio block through one FX block.
|
|
|
|
Args:
|
|
buf: Mono audio block (N,) to process.
|
|
idx: Chain index for state lookup.
|
|
entry: Chain entry dict with fx_type, params.
|
|
|
|
Returns:
|
|
Processed mono block (N,).
|
|
"""
|
|
fx_type = entry["fx_type"]
|
|
params = entry["params"]
|
|
fx_state = self._state.setdefault(f"fx_{idx}", {})
|
|
|
|
match fx_type:
|
|
case FXType.NOISE_GATE:
|
|
return self._apply_gate(buf, params, fx_state)
|
|
case FXType.COMPRESSOR:
|
|
return self._apply_compressor(buf, params, fx_state)
|
|
case FXType.BOOST:
|
|
return self._apply_boost(buf, params, fx_state)
|
|
case FXType.OVERDRIVE:
|
|
return self._apply_overdrive(buf, params, fx_state)
|
|
case FXType.DISTORTION:
|
|
return self._apply_distortion(buf, params, fx_state)
|
|
case FXType.FUZZ:
|
|
return self._apply_fuzz(buf, params, fx_state)
|
|
case FXType.EQ:
|
|
return self._apply_eq(buf, params, fx_state)
|
|
case FXType.CHORUS:
|
|
return self._apply_chorus(buf, params, fx_state)
|
|
case FXType.FLANGER:
|
|
return self._apply_flanger(buf, params, fx_state)
|
|
case FXType.PHASER:
|
|
return self._apply_phaser(buf, params, fx_state)
|
|
case FXType.TREMOLO:
|
|
return self._apply_tremolo(buf, params, fx_state)
|
|
case FXType.VIBRATO:
|
|
return self._apply_vibrato(buf, params, fx_state)
|
|
case FXType.DELAY:
|
|
return self._apply_delay(buf, params, fx_state)
|
|
case FXType.REVERB:
|
|
return self._apply_reverb(buf, params, fx_state)
|
|
case FXType.VOLUME:
|
|
return self._apply_volume(buf, params, fx_state)
|
|
case FXType.NAM_AMP:
|
|
if self.nam.is_loaded:
|
|
processed = self.nam.process(buf)
|
|
if self.nam._crossfade_buf is not None:
|
|
processed = self.nam.apply_crossfade(processed)
|
|
return processed
|
|
return buf
|
|
case FXType.IR_CAB:
|
|
if self.ir.is_loaded:
|
|
return self._apply_ir_cab(buf, params, fx_state)
|
|
return buf
|
|
case _:
|
|
return buf
|
|
|
|
# ── 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.
|
|
|
|
Uses scipy.signal.lfilter with persistent state for
|
|
zero-crosstalk between blocks.
|
|
"""
|
|
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)
|
|
|
|
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
|
|
b = np.array([b0, b1, b2], dtype=np.float64)
|
|
a = np.array([1.0, a1, a2], dtype=np.float64)
|
|
zi = state.get(f"{key}_zi", np.zeros(2, dtype=np.float64))
|
|
|
|
sig, zf = lfilter(b, a, sig, zi=zi)
|
|
state[f"{key}_zi"] = zf
|
|
|
|
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)
|
|
|
|
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")
|
|
mod_delay = base_samples + lfo * mod_range
|
|
|
|
# Vectorised read
|
|
wet = delay_line.read_block_varying(mod_delay)
|
|
|
|
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")
|
|
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
|
|
|
|
# Vectorised read
|
|
wet = delay_line.read_block_varying(mod_delay)
|
|
|
|
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))
|
|
|
|
# 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
|
|
|
|
fb_buf = state.get("fb_buf", np.zeros(len(buf), dtype=np.float64))
|
|
fb_input = buf.astype(np.float64, copy=False) + fb_buf * feedback
|
|
|
|
sig = fb_input.copy()
|
|
for stage in range(stages):
|
|
# Allpass as first-order IIR: H(z) = (coeff + z^-1) / (1 + coeff * z^-1)
|
|
# Which is lfilter(b=[coeff, 1], a=[1, coeff], ...)
|
|
# But coeff varies per sample (LFO-driven)! Can't use lfilter directly.
|
|
# Use block-constant approximation: one coeff per block at LFO centre.
|
|
freq = np.mean(freq_range)
|
|
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)
|
|
|
|
b = np.array([coeff, 1.0], dtype=np.float64)
|
|
a = np.array([1.0, coeff], dtype=np.float64)
|
|
zi = state.get(f"ap_zi_{stage}", np.zeros(1, dtype=np.float64))
|
|
sig, zf = lfilter(b, a, sig, zi=zi)
|
|
state[f"ap_zi_{stage}"] = zf
|
|
|
|
state["fb_buf"] = sig * 0.5
|
|
sig = np.clip(sig, -1.0, 1.0)
|
|
return (buf * (1.0 - mix) + sig.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 = delay_line.read_block_varying(mod_delay)
|
|
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
|
|
|
|
# ── 13. IR Cabinet Simulator ─────────────────────────────────────
|
|
|
|
def _apply_ir_cab(self, buf: np.ndarray, params: dict,
|
|
state: dict) -> np.ndarray:
|
|
"""Apply IR convolution via IRLoader.process().
|
|
|
|
Delegates to the IRLoader's FFT overlap-add engine.
|
|
Supports wet/dry mix control per-preset.
|
|
|
|
Params:
|
|
- ir_file: str (path to .wav IR) — already set via load_ir()
|
|
- enabled: bool
|
|
- wet: float 0.0-1.0
|
|
- dry: float 0.0-1.0
|
|
"""
|
|
# Update mix from preset params
|
|
wet = params.get("wet", 1.0)
|
|
dry = params.get("dry", 0.0)
|
|
self.ir.set_mix(wet=wet, dry=dry)
|
|
self.ir.enabled = params.get("enabled", True) and not params.get("bypass", False)
|
|
|
|
return self.ir.process(buf)
|
|
|
|
# ── 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
|
|
|
|
# ── 4CM routing properties ────────────────────────────────────
|
|
|
|
@property
|
|
def routing_mode(self) -> str:
|
|
return self._routing_mode
|
|
|
|
@routing_mode.setter
|
|
def routing_mode(self, value: str) -> None:
|
|
if value not in ("mono", "4cm"):
|
|
raise ValueError(f"routing_mode must be 'mono' or '4cm', got {value!r}")
|
|
self._routing_mode = value
|
|
logger.info("Routing mode: %s", value)
|
|
|
|
@property
|
|
def routing_breakpoint(self) -> int:
|
|
return self._routing_breakpoint
|
|
|
|
@routing_breakpoint.setter
|
|
def routing_breakpoint(self, value: int) -> None:
|
|
self._routing_breakpoint = max(0, value)
|
|
logger.info("Routing breakpoint: %d", self._routing_breakpoint)
|
|
|
|
def set_routing(self, mode: str, breakpoint: int = 7) -> None:
|
|
"""Set 4CM routing configuration at runtime.
|
|
|
|
Args:
|
|
mode: ``\"mono\"`` or ``\"4cm\"``.
|
|
breakpoint: Chain index where pre/post split occurs.
|
|
"""
|
|
self.routing_mode = mode
|
|
self.routing_breakpoint = breakpoint |