"""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 import threading from dataclasses import dataclass, field from typing import Optional import numpy as np from scipy.signal import lfilter from .nam_router import NAMEngineRouter 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) def _compute_hpf_coeffs(freq: float, q: float, sr: float) -> tuple: """RBJ high-pass biquad coefficients.""" omega = 2 * np.pi * freq / sr sn = np.sin(omega) cs = np.cos(omega) alpha = sn / (2 * q) b0 = (1 + cs) / 2 b1 = -(1 + cs) b2 = (1 + cs) / 2 a0 = 1 + alpha a1 = -2 * cs a2 = 1 - alpha return (b0 / a0, b1 / a0, b2 / a0, a1 / a0, a2 / a0) def _compute_lpf_coeffs(freq: float, q: float, sr: float) -> tuple: """RBJ low-pass biquad coefficients.""" omega = 2 * np.pi * freq / sr sn = np.sin(omega) cs = np.cos(omega) alpha = sn / (2 * q) b0 = (1 - cs) / 2 b1 = 1 - cs b2 = (1 - cs) / 2 a0 = 1 + alpha a1 = -2 * cs a2 = 1 - alpha return (b0 / a0, b1 / a0, b2 / a0, a1 / a0, a2 / a0) def _compute_bpf_coeffs(freq: float, q: float, sr: float) -> tuple: """RBJ band-pass biquad coefficients.""" omega = 2 * np.pi * freq / sr sn = np.sin(omega) cs = np.cos(omega) alpha = sn / (2 * q) b0 = sn / 2 b1 = 0.0 b2 = -sn / 2 a0 = 1 + alpha a1 = -2 * cs a2 = 1 - alpha return (b0 / a0, b1 / a0, b2 / a0, a1 / a0, a2 / a0) def _compute_notch_coeffs(freq: float, q: float, sr: float) -> tuple: """RBJ notch biquad coefficients.""" omega = 2 * np.pi * freq / sr sn = np.sin(omega) cs = np.cos(omega) alpha = sn / (2 * q) b0 = 1.0 b1 = -2 * cs b2 = 1.0 a0 = 1 + alpha a1 = -2 * cs a2 = 1 - alpha 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", "_buf_size") def __init__(self, delay_samples: int, block_size: int = 256): 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) self._buf_size = block_size def process(self, block: np.ndarray) -> np.ndarray: # Resize internal buffer if block size changed (e.g. JACK period switch) n = len(block) if n != self._buf_size: self.buf = np.zeros(n, dtype=np.float32) self._buf_size = n 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", "_buf_size") def __init__(self, delay_samples: int, block_size: int = 256): 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) self._buf_size = block_size def process(self, block: np.ndarray) -> np.ndarray: # Resize internal buffer if block size changed (e.g. JACK period switch) n = len(block) if n != self._buf_size: self.buf = np.zeros(n, dtype=np.float32) self._buf_size = n # 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[NAMEngineRouter] = None, ir_loader: Optional[IRLoader] = None, ): self.nam = nam_host or NAMEngineRouter() 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] = {} # Runtime audio params (may be updated via set_audio_profile) self._block_size: int = 256 self._sample_rate: int = 48000 # VU meter level tracking — updated on every process() call # Smoothed RMS levels (0.0–1.0) read by web server for live VU meters self._input_level: float = 0.0 self._output_level: float = 0.0 # Smoothing factor: ~50ms time constant — recomputed on profile change self._vu_alpha: float = np.exp(-256 / (0.05 * 48000)) # ── Tuner / pitch detection state ──────────────────────────────── self._tuner_frequency: float = 0.0 # detected fundamental freq (Hz) self._tuner_note: str = "--" # closest note name self._tuner_cents: float = 0.0 # cent deviation from closest note self._tuner_string: int = -1 # string number (1-6) or -1 if not matched self._tuner_confidence: float = 0.0 # 0.0 to 1.0 # Pitch detection buffer (keep last N samples for analysis) self._pitch_buffer: np.ndarray = np.array([], dtype=np.float32) self._pitch_buffer_max: int = 2048 # ~43ms at 48kHz # Thread-safety lock — protects config state swapped by load_preset() # and read by process(). process() snapshots under this lock briefly; # load_preset() holds it only for the atomic swap (not during model I/O). self._lock = threading.Lock() logger.info("Audio pipeline initialized (block=%d, sr=%d)", self._block_size, self._sample_rate) @property def sample_rate(self) -> int: """Current sample rate in Hz.""" return self._sample_rate @property def block_size(self) -> int: """Current audio block size (frames per callback).""" return self._block_size def load_preset(self, preset: Preset) -> None: """Load a complete preset (NAM, IR, and FX chain). Builds the new chain list off-thread, then swaps atomically under ``_lock`` so that the audio thread's ``process()`` never sees a half-constructed chain. """ new_chain: list[dict] = [] new_routing_mode = preset.routing_mode new_routing_breakpoint = preset.routing_breakpoint new_master_volume = preset.master_volume new_tuner_enabled = preset.tuner_enabled for block in preset.chain: entry = { "fx_type": block.fx_type, "enabled": block.enabled, "bypass": block.bypass, "params": dict(block.params), "subtype": block.subtype, } # Load NAM model if needed (may do I/O — don't hold pipeline lock) 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) new_chain.append(entry) # Atomic swap under lock — keep the window as short as possible with self._lock: self._chain = new_chain self._state = {} self._coeffs = {} self._master_volume = new_master_volume self._tuner_enabled = new_tuner_enabled self._routing_mode = new_routing_mode self._routing_breakpoint = new_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]. """ # ── Snapshot config under lock (brief — no I/O or heavy work) ── with self._lock: tuner_enabled = self._tuner_enabled bypassed = self._bypassed routing_mode = self._routing_mode routing_breakpoint = self._routing_breakpoint master_volume = self._master_volume # Shallow-copy the chain list so iteration is isolated chain = list(self._chain) # Snapshot the state dict — process() mutates entries inside it # but load_preset() replaces the whole dict under lock state = self._state # ── Tuner mode: mute output, keep input tracking for pitch detection ── if tuner_enabled: # Still track input level for tuner display if audio_in.ndim == 1: in_rms = np.sqrt(np.mean(audio_in ** 2) + _EPS) self._input_level = ( self._input_level * self._vu_alpha + in_rms * (1.0 - self._vu_alpha) ) else: ch0 = audio_in[0, :] in_rms = np.sqrt(np.mean(ch0 ** 2) + _EPS) self._input_level = ( self._input_level * self._vu_alpha + in_rms * (1.0 - self._vu_alpha) ) # Run pitch detection on input self._detect_pitch(audio_in) return np.zeros_like(audio_in) if bypassed: return audio_in * master_volume if routing_mode == "4cm": return self._process_4cm(audio_in, master_volume, chain, state, routing_breakpoint) else: return self._process_mono(audio_in, master_volume, chain, state) # ── Pitch detection for tuner ────────────────────────────────────────────── # Standard tuning frequencies (E2, A2, D3, G3, B3, E4) — guitar strings _STRING_FREQS = [82.41, 110.0, 146.83, 196.0, 246.94, 329.63] _STRING_NAMES = ["E", "A", "D", "G", "B", "e"] # Note names in chromatic order (C = 0) _NOTE_NAMES = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"] def _detect_pitch(self, audio_in: np.ndarray) -> None: """Run pitch detection on the input buffer using autocorrelation. Updates ``self._tuner_frequency``, ``self._tuner_note``, ``self._tuner_cents``, and ``self._tuner_string``. Args: audio_in: Input audio block — mono (N,) or stereo (2, N). """ # Extract mono channel if audio_in.ndim == 2: signal = audio_in[0, :].copy() else: signal = audio_in.copy() # Append to rolling pitch buffer self._pitch_buffer = np.concatenate([self._pitch_buffer, signal]) if len(self._pitch_buffer) > self._pitch_buffer_max: self._pitch_buffer = self._pitch_buffer[-self._pitch_buffer_max:] # Need enough signal for meaningful analysis if len(self._pitch_buffer) < 512: self._tuner_confidence = 0.0 return # Simple autocorrelation pitch detection buf = self._pitch_buffer # Remove DC offset buf = buf - np.mean(buf) # Check if there's enough amplitude rms = np.sqrt(np.mean(buf ** 2)) if rms < 0.002: # Silence threshold self._tuner_confidence = 0.0 self._tuner_frequency = 0.0 self._tuner_note = "--" self._tuner_cents = 0.0 self._tuner_string = -1 return # Autocorrelation: find the fundamental period # Search lag range: 30 to 1024 samples (46.9Hz to 1600Hz at 48kHz) min_lag = int(self._sample_rate / 1600) # ~30 max_lag = min(int(self._sample_rate / 50), len(buf) // 2) # ~960 if max_lag <= min_lag: self._tuner_confidence = 0.0 return corr = np.correlate(buf, buf, mode='full') # Take only the second half (positive lags) corr = corr[len(corr) // 2:] # Simple autocorrelation lag_slice = corr[min_lag:max_lag + 1] # Find the first peak in the autocorrelation diffs = np.diff(lag_slice) # Look for zero crossings in diff (peaks: positive→negative) peaks = [] for i in range(1, len(diffs)): if diffs[i-1] > 0 and diffs[i] <= 0: peaks.append((min_lag + i, lag_slice[i])) if not peaks: self._tuner_confidence = 0.0 return # Pick the strongest peak best_lag, best_val = max(peaks, key=lambda x: x[1]) # Confidence based on relative peak strength noise_floor = np.mean(np.abs(corr[min_lag:max_lag + 1])) confidence = best_val / (noise_floor + 1e-10) # Parabolic interpolation for sub-sample accuracy if best_lag > min_lag and best_lag < max_lag: idx = best_lag - min_lag if 0 < idx < len(lag_slice) - 1: y0, y1, y2 = lag_slice[idx-1], lag_slice[idx], lag_slice[idx+1] if y0 + y2 - 2 * y1 != 0: correction = (y0 - y2) / (2 * (y0 + y2 - 2 * y1)) best_lag = best_lag + correction # Fundamental frequency freq = self._sample_rate / best_lag if best_lag > 0 else 0 # Clip confidence to 0-1 range self._tuner_confidence = min(1.0, max(0.0, confidence / 10.0)) if freq < 30 or freq > 1600: self._tuner_confidence = 0.0 return self._tuner_frequency = freq # ── Convert frequency to note name ── # A4 = 440Hz, MIDI note 69 midi_note = 12 * np.log2(freq / 440.0) + 69 midi_rounded = round(midi_note) cents = int(100 * (midi_note - midi_rounded)) # Clamp to valid MIDI range if midi_rounded < 0 or midi_rounded > 127: self._tuner_note = "--" self._tuner_confidence = 0.0 return octave = (midi_rounded // 12) - 1 note_idx = midi_rounded % 12 note_name = self._NOTE_NAMES[note_idx] self._tuner_note = f"{note_name}{octave}" self._tuner_cents = cents # ── Guess guitar string ── self._tuner_string = -1 for si, sf in enumerate(self._STRING_FREQS): # +/- 3 semitones from the string's fundamental if abs(freq - sf) / sf < 0.2: self._tuner_string = si + 1 break def _process_mono(self, audio_in: np.ndarray, master_volume: float, chain: list[dict], state: dict[str, dict]) -> np.ndarray: """Process a mono block through the full chain (all blocks).""" # Update input VU level (RMS with envelope smoothing) in_rms = np.sqrt(np.mean(audio_in ** 2) + _EPS) self._input_level = ( self._input_level * self._vu_alpha + in_rms * (1.0 - self._vu_alpha) ) buf = audio_in.copy() for idx, entry in enumerate(chain): if entry["bypass"] or not entry["enabled"]: continue buf = self._process_single_block(buf, idx, entry, state) out = np.clip(buf * master_volume, -1.0, 1.0) # Update output VU level out_rms = np.sqrt(np.mean(out ** 2) + _EPS) self._output_level = ( self._output_level * self._vu_alpha + out_rms * (1.0 - self._vu_alpha) ) return out def _process_4cm(self, audio_in: np.ndarray, master_volume: float, chain: list[dict], state: dict[str, dict], routing_breakpoint: int) -> 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() # Update input VU level from the guitar input channel (ch0) in_rms = np.sqrt(np.mean(ch0 ** 2) + _EPS) self._input_level = ( self._input_level * self._vu_alpha + in_rms * (1.0 - self._vu_alpha) ) bp = routing_breakpoint for idx, entry in enumerate(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, state) else: # Post-amp block — process on return (ch1) ch1 = self._process_single_block(ch1, idx, entry, state) out = np.zeros_like(audio_in) out[0, :] = np.clip(ch0 * master_volume, -1.0, 1.0) out[1, :] = np.clip(ch1 * master_volume, -1.0, 1.0) # Update output VU level from the processed effect return (ch1) out_rms = np.sqrt(np.mean(out ** 2) + _EPS) self._output_level = ( self._output_level * self._vu_alpha + out_rms * (1.0 - self._vu_alpha) ) return out def _process_single_block(self, buf: np.ndarray, idx: int, entry: dict, state: dict[str, 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. state: Per-block DSP state dict (snapshotted by ``process()``). Returns: Processed mono block (N,). """ fx_type = entry["fx_type"] 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 = 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: 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: 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: 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: 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 np.clip(self._apply_volume(buf, params, fx_state), -1.0, 1.0) case FXType.NAM_AMP: # Use C++ NeuralAudio engine when a .nam file is loaded. if self.nam.is_loaded: # Apply input gain as pre-amp drive (level 0.0-1.0 maps to 0-1.0x gain) level = params.get("level", 0.75) drive = np.clip(level, 0.0, 1.0) if drive != 1.0: buf = buf * drive # Single process call through the C++ engine processed = self.nam.process(buf) # Crossfade on preset switch if self.nam._crossfade_buf is not None: processed = self.nam.apply_crossfade(processed) # Clip output to prevent digital distortion return np.clip(processed, -1.0, 1.0) logger.debug("NAM_AMP: engine not loaded, passing through") return buf case FXType.IR_CAB: if self.ir.is_loaded: return self._apply_ir_cab(buf, params, fx_state) return buf # ── Pitch & Frequency ────────────────────────────────────── case FXType.OCTAVER: return self._apply_octaver(buf, params, fx_state) case FXType.PITCH_SHIFTER: return self._apply_pitch_shifter(buf, params, fx_state) case FXType.HARMONIZER: return self._apply_harmonizer(buf, params, fx_state) case FXType.WHAMMY: return self._apply_whammy(buf, params, fx_state) case FXType.DETUNE: return self._apply_detune(buf, params, fx_state) # ── Modulation ───────────────────────────────────────────── case FXType.RING_MODULATOR: return self._apply_ring_modulator(buf, params, fx_state) case FXType.AUTO_WAH: return self._apply_auto_wah(buf, params, fx_state) case FXType.ENVELOPE_FILTER: return self._apply_envelope_filter(buf, params, fx_state) case FXType.ROTARY_SPEAKER: return self._apply_rotary_speaker(buf, params, fx_state) case FXType.UNI_VIBE: return self._apply_uni_vibe(buf, params, fx_state) case FXType.AUTO_PAN: return self._apply_auto_pan(buf, params, fx_state) case FXType.STEREO_WIDENER: return self._apply_stereo_widener(buf, params, fx_state) # ── Drive & Saturation ───────────────────────────────────── case FXType.BITCRUSHER: return self._apply_bitcrusher(buf, params, fx_state) case FXType.WAVEFOLDER: return self._apply_wavefolder(buf, params, fx_state) case FXType.RECTIFIER: return self._apply_rectifier(buf, params, fx_state) # ── Dynamics ─────────────────────────────────────────────── case FXType.EXPANDER: return self._apply_expander(buf, params, fx_state) case FXType.DE_ESSER: return self._apply_de_esser(buf, params, fx_state) case FXType.TRANSIENT_SHAPER: return self._apply_transient_shaper(buf, params, fx_state) case FXType.SIDECHAIN_COMPRESSOR: return self._apply_sidechain_compressor(buf, params, fx_state) # ── Filters & EQ ────────────────────────────────────────── case FXType.PARAMETRIC_EQ: return self._apply_parametric_eq(buf, params, fx_state) case FXType.HIGH_PASS_FILTER: return self._apply_hpf(buf, params, fx_state) case FXType.LOW_PASS_FILTER: return self._apply_lpf(buf, params, fx_state) case FXType.BAND_PASS_FILTER: return self._apply_bpf(buf, params, fx_state) case FXType.NOTCH_FILTER: return self._apply_notch(buf, params, fx_state) case FXType.FORMANT_FILTER: return self._apply_formant_filter(buf, params, fx_state) # ── Time-Based ───────────────────────────────────────────── case FXType.PING_PONG_DELAY: return self._apply_ping_pong_delay(buf, params, fx_state) case FXType.MULTI_TAP_DELAY: return self._apply_multi_tap_delay(buf, params, fx_state) case FXType.REVERSE_DELAY: return self._apply_reverse_delay(buf, params, fx_state) case FXType.TAPE_ECHO: return self._apply_tape_echo(buf, params, fx_state) case FXType.SHIMMER_REVERB: return self._apply_shimmer_reverb(buf, params, fx_state) case FXType.LOOPER: return self._apply_looper(buf, params, fx_state) # ── Ambience ────────────────────────────────────────────── case FXType.EARLY_REFLECTIONS: return self._apply_early_reflections(buf, params, fx_state) case _: return buf # ── LFO helpers ───────────────────────────────────────────────── def _lfo_phase(self, 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 / self._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(-self._block_size / (release_ms * self._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(-self._block_size / (attack_ms * self._sample_rate / 1000.0)) else: alpha = np.exp(-self._block_size / (release_ms * self._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) # ── 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 / self._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, self._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 = np.clip(sig, -1.0, 1.0).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, self._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, self._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(np.clip(sig, -1.0, 1.0).astype(np.float32) * volume, -1.0, 1.0) return out # ── 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, self._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 * self._sample_rate / 1000.0 mod_range = depth * 5.0 * self._sample_rate / 1000.0 if "delay" not in state: max_d = int(base_samples + mod_range + 10.0 * self._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 np.clip(buf * (1.0 - mix) + wet * mix, -1.0, 1.0) # ── 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 * self._sample_rate / 1000.0 mod_range = depth * 5.0 * self._sample_rate / 1000.0 if "delay" not in state: max_d = int(base_samples + mod_range + 10.0 * self._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 np.clip(buf * (1.0 - mix) + wet * mix, -1.0, 1.0) # ── 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 / self._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 * self._sample_rate / 1000.0 # fixed ~2ms base mod_range = depth * 3.0 * self._sample_rate / 1000.0 if "delay" not in state: max_d = int(base_samples + mod_range + 5.0 * self._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: """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) 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 * self._sample_rate / 1000.0) if "delay" not in state: # Allocate 2x requested delay for headroom max_d = max(delay_samples * 2, self._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 + clipped feedback to prevent delay-line runaway fb_gain = min(feedback, 0.98) write_sig = np.clip(buf + wet * fb_gain, -1.0, 1.0) delay_line.write_block(write_sig) return np.clip(buf * (1.0 - mix) + wet * mix, -1.0, 1.0) 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 * self._sample_rate / 1000.0) if "delay" not in state: max_d = max(delay_samples * 2, self._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) # Write with clipped feedback fb_gain = min(feedback, 0.98) write_sig = np.clip(buf + lp_out * fb_gain, -1.0, 1.0) delay_line.write_block(write_sig) return np.clip(buf * (1.0 - mix) + wet * mix, -1.0, 1.0) # ── 11. Reverb (subtype dispatch) ─────────────────────────────── def _apply_reverb(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """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 np.clip(buf * (1.0 - mix) + wet * mix, -1.0, 1.0) 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) predelay_ms = params.get("predelay", 30.0) if "combs" not in state: comb_delays = [29, 37, 44, 50, 31, 39, 47, 53] # ms ap_delays = [5, 7, 11, 13] # ms state["combs"] = [ _CombFilter(int(d * self._sample_rate / 1000.0), block_size=self._block_size) for d in comb_delays ] state["allpasses"] = [ _AllpassFilter(int(d * self._sample_rate / 1000.0), block_size=self._block_size) for d in ap_delays ] state["predelay"] = _DelayLine( int(predelay_ms * self._sample_rate / 1000.0) + 1 ) state["predelay"].write_block(np.zeros(self._block_size)) state["_computed"] = False combs: list[_CombFilter] = state["combs"] allpasses: list[_AllpassFilter] = state["allpasses"] predelay_line: _DelayLine = state["predelay"] param_tag = (decay, damping) if state.get("_param_tag") != param_tag: 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) for ap in allpasses: ap.gain = 0.3 + damping * 0.3 state["_param_tag"] = param_tag delayed = predelay_line.read_block( float(predelay_ms * self._sample_rate / 1000.0), len(buf)) predelay_line.write_block(buf) wet = np.zeros_like(buf, dtype=np.float64) for comb in combs: wet += comb.process(delayed) wet /= len(combs) for ap in allpasses: wet = ap.process(wet) 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 * self._sample_rate / 1000.0 + self._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 * self._sample_rate / 1000.0) + 1) state["predelay"].write_block(np.zeros(self._block_size)) predelay_line: _DelayLine = state["predelay"] delayed = predelay_line.read_block( float(predelay_ms * self._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 - self._block_size - 1 spring_out = dl.read_block(float(delay_samps), len(buf)) # Bandpass filter per spring — emphasises resonant frequency coeff = _compute_bpf_coeffs( self._sample_rate / (delay_samps + 1), q_val, self._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 * self._sample_rate / 1000.0), block_size=self._block_size) for d in comb_delays_ms ] state["allpasses"] = [ _AllpassFilter(int(d * self._sample_rate / 1000.0), block_size=self._block_size) for d in ap_delays_ms ] state["predelay"] = _DelayLine( int(predelay_ms * self._sample_rate / 1000.0) + 1) state["predelay"].write_block(np.zeros(self._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 * self._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 / self._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 * self._sample_rate / 1000.0) state["er_delay"] = _DelayLine(max_tap + self._block_size + 1) state["er_taps"] = [ int(t * size_factor * self._sample_rate / 1000.0) for t in base_taps_ms ] state["tail_combs"] = [ _CombFilter(int(d * size_factor * self._sample_rate / 1000.0 + 1), block_size=self._block_size) for d in tail_comb_ms ] state["tail_allpasses"] = [ _AllpassFilter( int(d * size_factor * self._sample_rate / 1000.0 + 1), block_size=self._block_size) for d in tail_ap_ms ] state["predelay"] = _DelayLine( int(predelay_ms * self._sample_rate / 1000.0) + 1) state["predelay"].write_block(np.zeros(self._block_size)) predelay_line: _DelayLine = state["predelay"] delayed = predelay_line.read_block( float(predelay_ms * self._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 ────────────────────────────────────────────────── 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) # ── 14. Octaver ────────────────────────────────────────────────── def _apply_octaver(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Sub-octave via full-wave rectification + divide-by-2. Mix control blends between dry and octave-down. """ mix = params.get("mix", 0.5) # Full-wave rectification produces even harmonics including sub-octave rectified = np.abs(buf) # Low-pass to isolate fundamental (below ~300Hz corner) # Simple one-pole: alpha ~ 0.95 at 48kHz alpha = state.get("lp_alpha", 0.95) lp = np.zeros_like(buf) prev = state.get("lp_prev", 0.0) for i in range(len(buf)): prev = alpha * prev + (1.0 - alpha) * rectified[i] lp[i] = prev state["lp_prev"] = prev # Scale to match input level sub = lp * 0.5 out = buf * (1.0 - mix) + sub * mix return np.clip(out, -1.0, 1.0) # ── 15. Pitch Shifter ──────────────────────────────────────────── # BETA — test on RPi 4B for xruns def _apply_pitch_shifter(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Granular/overlap-add pitch shifter for subtle shifts (±2 semitones). CPU-heavy, tagged as beta. """ # BETA — test on RPi 4B for xruns shift_semitones = params.get("shift", 0.0) mix = params.get("mix", 0.5) if shift_semitones == 0.0: return buf # Pitch factor: 2^(shift/12) factor = 2.0 ** (shift_semitones / 12.0) # Windowed overlap-add with grain size ~40ms grain_size = int(0.040 * self._sample_rate) # 1920 samples at 48kHz hop_in = grain_size hop_out = int(grain_size / factor) if factor > 0 else grain_size if "ring" not in state: state["ring"] = np.zeros(grain_size * 2, dtype=np.float32) state["write_pos"] = 0 state["read_pos"] = 0 ring: np.ndarray = state["ring"] wpos = state["write_pos"] rpos = state["read_pos"] # Write input into ring buffer for i, s in enumerate(buf): ring[wpos % len(ring)] = s wpos += 1 # Read with resampling via linear interpolation out = np.zeros_like(buf) for i in range(len(buf)): idx_floor = int(np.floor(rpos)) frac = rpos - idx_floor a = ring[idx_floor % len(ring)] b = ring[(idx_floor + 1) % len(ring)] out[i] = a + frac * (b - a) rpos += factor if rpos >= wpos: rpos = wpos - 1 state["write_pos"] = wpos state["read_pos"] = rpos # Apply raised-cosine window window = 0.5 * (1.0 - np.cos(2.0 * np.pi * np.arange(len(buf)) / len(buf))) out = out * window return buf * (1.0 - mix) + out.astype(np.float32) * mix # ── 16. Harmonizer ────────────────────────────────────────────── def _apply_harmonizer(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Pitch shift + dry mix, with MIDI-note-aware shift amount. shift parameter is in semitones (e.g. 3 = minor third above). """ shift_semitones = params.get("shift", 3.0) mix = params.get("mix", 0.5) if shift_semitones == 0.0: return buf factor = 2.0 ** (shift_semitones / 12.0) grain_size = int(0.040 * self._sample_rate) hop_out = int(grain_size / factor) if "ring" not in state: state["ring"] = np.zeros(grain_size * 2, dtype=np.float32) state["wpos"] = 0 state["rpos"] = 0 ring = state["ring"] wpos = state["wpos"] rpos = state["rpos"] for s in buf: ring[wpos % len(ring)] = s wpos += 1 out = np.zeros_like(buf) for i in range(len(buf)): idx = int(np.floor(rpos)) frac = rpos - idx a = ring[idx % len(ring)] b = ring[(idx + 1) % len(ring)] out[i] = a + frac * (b - a) rpos += factor if rpos >= wpos: rpos = wpos - 1 state["wpos"] = wpos state["rpos"] = rpos window = 0.5 * (1.0 - np.cos(2.0 * np.pi * np.arange(len(buf)) / len(buf))) out = out * window return buf * (1.0 - mix) + out.astype(np.float32) * mix # ── 17. Whammy ────────────────────────────────────────────────── def _apply_whammy(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Continuous pitch bend via parameter (expression pedal target). bend parameter: 0.0 = no shift, 1.0 = +12 semitones up. """ bend = params.get("bend", 0.0) # 0.0-1.0 maps to 0-12 semitones mix = params.get("mix", 0.7) shift_semitones = bend * 12.0 if "ring" not in state: grain_size = int(0.040 * self._sample_rate) state["ring"] = np.zeros(grain_size * 2, dtype=np.float32) state["wpos"] = 0 state["rpos"] = 0 factor = 2.0 ** (shift_semitones / 12.0) ring = state["ring"] wpos = state["wpos"] rpos = state["rpos"] for s in buf: ring[wpos % len(ring)] = s wpos += 1 out = np.zeros_like(buf) for i in range(len(buf)): idx = int(np.floor(rpos)) frac = rpos - idx a = ring[idx % len(ring)] b = ring[(idx + 1) % len(ring)] out[i] = a + frac * (b - a) rpos += factor if rpos >= wpos: rpos = wpos - 1 state["wpos"] = wpos state["rpos"] = rpos window = 0.5 * (1.0 - np.cos(2.0 * np.pi * np.arange(len(buf)) / len(buf))) out = out * window return buf * (1.0 - mix) + out.astype(np.float32) * mix # ── 18. Detune ────────────────────────────────────────────────── def _apply_detune(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Very short modulated delay (0.5-5ms), chorus-like thickening. Detunes by shifting pitch via modulated read of a tiny delay line. """ depth = params.get("depth", 0.5) mix = params.get("mix", 0.5) base_samples = 1.0 * self._sample_rate / 1000.0 # 1ms base mod_range = depth * 3.0 * self._sample_rate / 1000.0 if "delay" not in state: max_d = int(base_samples + mod_range + 5.0 * self._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(4.0, state, len(buf)) # 4Hz LFO 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 np.clip(buf * (1.0 - mix) + wet * mix, -1.0, 1.0) # ── 19. Ring Modulator ─────────────────────────────────────────── def _apply_ring_modulator(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Multiply signal with sine LFO. Rate, Depth, Mix.""" rate = params.get("rate", 100.0) # Hz depth = params.get("depth", 0.5) mix = params.get("mix", 0.5) phase = self._lfo_phase(rate, state, len(buf)) lfo = self._lfo_wave(phase, "sine") carrier = lfo * 2.0 - 1.0 # map to [-1, 1] ring = buf * carrier rm_signal = buf * (1.0 - depth) + ring * depth return buf * (1.0 - mix) + rm_signal * mix # ── 20. Auto-Wah ──────────────────────────────────────────────── def _apply_auto_wah(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Envelope follower drives bandpass filter cutoff. Sensitivity, Q, Mix. """ sensitivity = params.get("sensitivity", 0.5) q = params.get("q", 2.0) mix = params.get("mix", 0.5) # Envelope follower rms = np.sqrt(np.mean(buf ** 2) + _EPS) env = state.get("envelope", 0.0) env = env * 0.9 + rms * 0.1 # smoothed state["envelope"] = env # Map envelope to cutoff frequency: 400-4000 Hz cutoff = 400.0 + env * sensitivity * 3600.0 cutoff = np.clip(cutoff, 200.0, 5000.0) # BPF biquad coeffs = _compute_bpf_coeffs(cutoff, q, self._sample_rate) 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("bpf_zi", np.zeros(2, dtype=np.float64)) sig, zf = lfilter(b, a, buf.astype(np.float64), zi=zi) state["bpf_zi"] = zf wet = sig.astype(np.float32) return np.clip(buf * (1.0 - mix) + wet * mix, -1.0, 1.0) # ── 21. Envelope Filter ────────────────────────────────────────── def _apply_envelope_filter(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Filter cutoff follows pick attack envelope. Decay, Range, Mix. """ decay = params.get("decay", 0.5) freq_range = params.get("range", 0.5) mix = params.get("mix", 0.5) # Envelope follower with decay rms = np.sqrt(np.mean(buf ** 2) + _EPS) env = state.get("envelope", 0.0) if rms > env: env = rms # instant attack else: env = env * np.exp(-self._block_size / (decay * 0.1 * self._sample_rate)) state["envelope"] = env cutoff = 200.0 + env * freq_range * 3800.0 cutoff = np.clip(cutoff, 100.0, 4000.0) coeffs = _compute_lpf_coeffs(cutoff, 0.707, self._sample_rate) 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("lpf_zi", np.zeros(2, dtype=np.float64)) sig, zf = lfilter(b, a, buf.astype(np.float64), zi=zi) state["lpf_zi"] = zf wet = sig.astype(np.float32) return np.clip(buf * (1.0 - mix) + wet * mix, -1.0, 1.0) # ── 22. Rotary Speaker ────────────────────────────────────────── def _apply_rotary_speaker(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Dual modulated allpass filters (Leslie sim). Speed, Drive, Mix. """ speed = params.get("speed", 0.5) # 0.0-1.0, slow-fast drive = params.get("drive", 0.3) mix = params.get("mix", 0.5) # Map speed: slow rotor ~0.5Hz, fast ~6.5Hz rotor_rate = 0.5 + speed * 6.0 # Horn: ~2x rotor speed horn_rate = rotor_rate * 2.0 # Dual modulated delays (rotor + horn) if "rotor_delay" not in state: state["rotor_delay"] = _DelayLine(int(0.020 * self._sample_rate + 1)) state["horn_delay"] = _DelayLine(int(0.015 * self._sample_rate + 1)) state["rotor_delay"].write_block(np.zeros(int(0.020 * self._sample_rate))) state["horn_delay"].write_block(np.zeros(int(0.015 * self._sample_rate))) rotor_phase = self._lfo_phase(rotor_rate, state, len(buf)) horn_phase = self._lfo_phase(horn_rate, state, len(buf)) # Rotor: deeper modulation, slower rotor_lfo = self._lfo_wave(rotor_phase, "sine") rotor_delay_s = 0.003 + rotor_lfo * 0.005 # 3-8ms # Horn: shallower, faster horn_lfo = self._lfo_wave(horn_phase, "sine") horn_delay_s = 0.001 + horn_lfo * 0.002 # 1-3ms rotor_wet = state["rotor_delay"].read_block_varying( rotor_delay_s * self._sample_rate) state["rotor_delay"].write_block(buf) horn_wet = state["horn_delay"].read_block_varying( horn_delay_s * self._sample_rate) state["horn_delay"].write_block(buf * 0.7) # Mix with subtle drive (soft clip) combined = rotor_wet + horn_wet if drive > 0: combined = np.tanh(combined * (1.0 + drive * 2.0)) return buf * (1.0 - mix) + combined.astype(np.float32) * mix # ── 23. Uni-Vibe ──────────────────────────────────────────────── def _apply_uni_vibe(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Photoresistor-style phaser. Rate, Depth, Feedback, Mix. """ rate = params.get("rate", 0.8) depth = params.get("depth", 0.5) feedback = params.get("feedback", 0.3) mix = params.get("mix", 0.5) phase = self._lfo_phase(rate, state, len(buf)) lfo = self._lfo_wave(phase, "sine") # Uni-Vibe uses 4 allpass stages with LFO-driven modulation # Centre frequency sweeps between 200-2000 Hz if "fb_buf" not in state: state["fb_buf"] = np.zeros(len(buf), dtype=np.float64) fb_buf = state["fb_buf"] fb_input = buf.astype(np.float64, copy=False) + fb_buf * feedback # Frequency per sample varies with LFO sig = fb_input.copy() for stage in range(4): freq = 200.0 + lfo * depth * 1800.0 freq_mean = float(np.mean(freq)) w = 2.0 * np.pi * freq_mean / self._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"uv_zi_{stage}", np.zeros(1, dtype=np.float64)) sig, zf = lfilter(b, a, sig, zi=zi) state[f"uv_zi_{stage}"] = zf state["fb_buf"] = sig * 0.3 sig = np.clip(sig, -1.0, 1.0) return (buf * (1.0 - mix) + sig.astype(np.float32) * mix).astype(np.float32) # ── 24. Auto-Pan ──────────────────────────────────────────────── def _apply_auto_pan(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """LFO-driven stereo pan. Rate, Depth, Waveform. Currently returns mono with pan applied; full stereo needs 2-channel pipeline. Simulated here as amplitude modulation. """ rate = params.get("rate", 0.3) depth = params.get("depth", 0.7) waveform = params.get("waveform", "sine") phase = self._lfo_phase(rate, state, len(buf)) lfo = self._lfo_wave(phase, waveform) # Map LFO (0-1) to pan [-1, 1] pan = (lfo * 2.0 - 1.0) * depth # Apply pan as amplitude modulation left_gain = np.clip(1.0 - pan, 0.0, 1.0) right_gain = np.clip(1.0 + pan, 0.0, 1.0) # Mono output: average both channels mod = (left_gain + right_gain) * 0.5 return (buf * mod).astype(np.float32) # ── 25. Stereo Widener ────────────────────────────────────────── def _apply_stereo_widener(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Mid/side processing, widen by boosting side channel. Width, Mix. Mono input: creates synthetic side from filtered difference. """ width = params.get("width", 0.5) mix = params.get("mix", 0.5) # From mono: create synthetic stereo by delaying a copy if "delay" not in state: state["delay"] = _DelayLine(int(0.030 * self._sample_rate + 1)) state["delay"].write_block(np.zeros(int(0.030 * self._sample_rate))) delay_line: _DelayLine = state["delay"] # Read delayed signal for "side" channel side = delay_line.read_block(0.010 * self._sample_rate, len(buf)) delay_line.write_block(buf) # Mid = original, Side = delayed difference mid = buf * 0.5 + side * 0.5 side_boosted = (buf - side) * (0.5 + width * 0.5) # Mix: blend widened signal with original widened = mid + side_boosted return (buf * (1.0 - mix) + widened * mix).astype(np.float32) # ── 26. Bitcrusher ────────────────────────────────────────────── def _apply_bitcrusher(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Reduce bit depth + sample rate. Bits, Rate, Mix.""" bits = params.get("bits", 8) # 1-16 rate_reduction = params.get("rate", 1.0) # 1-10, higher = more crush mix = params.get("mix", 0.5) # Bit depth reduction n_levels = max(2, 2 ** min(bits, 16)) crushed = np.round(buf * n_levels) / n_levels # Sample rate reduction (hold each sample for N samples) hold = max(1, int(rate_reduction)) if hold > 1: out = np.zeros_like(crushed) held_val = state.get("held_val", 0.0) hold_count = state.get("hold_count", 0) for i in range(len(crushed)): if hold_count >= hold: held_val = crushed[i] hold_count = 0 out[i] = held_val hold_count += 1 state["held_val"] = float(held_val) state["hold_count"] = hold_count crushed = out return buf * (1.0 - mix) + crushed * mix # ── 27. Wavefolder ────────────────────────────────────────────── def _apply_wavefolder(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Fold signal back at thresholds (Klanggebiet-style). Gain, Fold, Mix. """ gain = params.get("gain", 2.0) fold = params.get("fold", 3.0) mix = params.get("mix", 0.5) shaped = buf * gain # Multi-stage wavefolding for _ in range(int(fold)): # Fold: if abs > 1, fold back above = np.abs(shaped) > 1.0 shaped = np.where(above, np.sign(shaped) * (2.0 - np.abs(shaped)), shaped) return (buf * (1.0 - mix) + shaped * mix).astype(np.float32) # ── 28. Rectifier ─────────────────────────────────────────────── def _apply_rectifier(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Full/half-wave rectification for octave-up fuzz. Mode (full/half), Mix. """ mode = params.get("mode", "full") # "full" or "half" mix = params.get("mix", 0.5) if mode == "full": rectified = np.abs(buf) # full-wave: all positive, octave up else: rectified = np.where(buf > 0, buf, 0.0) # half-wave return buf * (1.0 - mix) + rectified * mix # ── 29. Expander ──────────────────────────────────────────────── def _apply_expander(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Inverse compressor, widens dynamic range. Threshold, Ratio, Attack, Release. """ threshold_db = params.get("threshold", -30.0) ratio = params.get("ratio", 3.0) attack_ms = params.get("attack", 5.0) release_ms = params.get("release", 100.0) rms = np.sqrt(np.mean(buf ** 2) + _EPS) envelope = state.get("envelope", 0.0) if rms > envelope: alpha = np.exp(-self._block_size / (attack_ms * self._sample_rate / 1000.0)) else: alpha = np.exp(-self._block_size / (release_ms * self._sample_rate / 1000.0)) envelope = envelope * alpha + rms * (1.0 - alpha) state["envelope"] = envelope if envelope > 1e-10: env_db = 20.0 * np.log10(envelope) else: env_db = -120.0 if env_db < threshold_db: # Expand: gain_db = (env - threshold) * (ratio - 1) gain_db = (env_db - threshold_db) * (ratio - 1.0) else: gain_db = 0.0 gain_lin = 10 ** (gain_db / 20.0) return np.clip(buf * gain_lin, -1.0, 1.0) # ── 30. De-esser ──────────────────────────────────────────────── def _apply_de_esser(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Compress sibilance band (EQ split + compressor). Frequency, Threshold, Mix. """ freq = params.get("frequency", 6000.0) # sibilance centre threshold = params.get("threshold", -30.0) mix = params.get("mix", 0.5) # Bandpass the sibilance band q = 3.0 # narrow Q for sibilance band coeffs = _compute_bpf_coeffs(freq, q, self._sample_rate) 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("bpf_zi", np.zeros(2, dtype=np.float64)) sibilance, zf = lfilter(b, a, buf.astype(np.float64), zi=zi) state["bpf_zi"] = zf # RMS of sibilance band sib_rms = np.sqrt(np.mean(sibilance ** 2) + _EPS) # Compress sibilance if sib_rms > (10.0 ** (threshold / 20.0)): gain = 0.3 # heavy reduction else: gain = 1.0 # Split: clean signal minus reduced sibilance clean = buf.astype(np.float64) - sibilance * (1.0 - gain) return (buf * (1.0 - mix) + clean.astype(np.float32) * mix).astype(np.float32) # ── 31. Transient Shaper ──────────────────────────────────────── def _apply_transient_shaper(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Attack boost/sustain per note. Attack, Sustain, Mix.""" attack = params.get("attack", 0.5) # boost amount sustain = params.get("sustain", 0.5) # sustain amount mix = params.get("mix", 0.5) # Envelope follower with fast attack/slow release rms = np.sqrt(np.mean(buf ** 2) + _EPS) env = state.get("envelope", 0.0) if rms > env: env = env * 0.5 + rms * 0.5 # fast attack else: env = env * 0.995 + rms * 0.005 # slow release state["envelope"] = env # Differentiate to find transients if "prev" not in state: state["prev"] = np.float32(0.0) diff = np.abs(buf.astype(np.float64) - state["prev"]) state["prev"] = np.float32(buf[-1]) # Transient detection: diff > envelope * threshold trans_thresh = max(env * 0.1, 0.001) is_transient = diff > trans_thresh # Attack boost on transients boost = np.ones_like(buf) boost[is_transient] = 1.0 + attack * 2.0 # Sustain: amplify quiet tail tail = env > 0.0 sustain_gain = 1.0 + (sustain - 0.5) * 2.0 shaped = buf * boost shaped[~is_transient] = buf[~is_transient] * sustain_gain return (buf * (1.0 - mix) + shaped.astype(np.float32) * mix).astype(np.float32) # ── 32. Sidechain Compressor ──────────────────────────────────── def _apply_sidechain_compressor(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Compression triggered by external input (4CM return channel). Threshold, Ratio, Attack, Release, Mix. NOTE: In mono mode, simulates sidechain from the input signal itself. """ threshold_db = params.get("threshold", -20.0) ratio = params.get("ratio", 4.0) attack_ms = params.get("attack", 2.0) release_ms = params.get("release", 50.0) mix = params.get("mix", 0.5) # Sidechain signal = the input itself (in mono mode) # In 4CM mode, the return channel would be the sidechain sidechain = buf rms = np.sqrt(np.mean(sidechain ** 2) + _EPS) envelope = state.get("envelope", 0.0) if rms > envelope: alpha = np.exp(-self._block_size / (attack_ms * self._sample_rate / 1000.0)) else: alpha = np.exp(-self._block_size / (release_ms * self._sample_rate / 1000.0)) envelope = envelope * alpha + rms * (1.0 - alpha) state["envelope"] = envelope if envelope > 1e-10: env_db = 20.0 * np.log10(envelope) else: env_db = -120.0 if env_db > threshold_db: gain_db = threshold_db + (env_db - threshold_db) / ratio - env_db else: gain_db = 0.0 gain_lin = 10 ** (gain_db / 20.0) compressed = np.clip(buf * gain_lin, -1.0, 1.0) return buf * (1.0 - mix) + compressed * mix # ── 33. Parametric EQ ────────────────────────────────────────── def _apply_parametric_eq(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """N-band biquad peaking filters. Takes array-style params: freq_0, gain_0, q_0, freq_1, ... Up to 4 bands. """ # Parse bands from flattened params sig = buf.astype(np.float64, copy=False) for band in range(4): freq = params.get(f"freq_{band}", 0.0) gain_db = params.get(f"gain_{band}", 0.0) q = params.get(f"q_{band}", 0.707) if freq == 0.0 or gain_db == 0.0: continue coeffs = _compute_peaking_coeffs(freq, gain_db, q, self._sample_rate) 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"peq_zi_{band}", np.zeros(2, dtype=np.float64)) sig, zf = lfilter(b, a, sig, zi=zi) state[f"peq_zi_{band}"] = zf return np.clip(sig, -1.0, 1.0).astype(np.float32) # ── 34-37. HPF / LPF / BPF / Notch ────────────────────────────── def _apply_hpf(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """High-pass filter. Frequency, Slope (6/12dB oct).""" freq = params.get("frequency", 200.0) slope = params.get("slope", 12.0) q = 0.707 if slope >= 12 else 0.5 # 12dB = Q=0.707 (Butter), 6dB = Q=0.5 coeffs = _compute_hpf_coeffs(freq, q, self._sample_rate) 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("hpf_zi", np.zeros(2, dtype=np.float64)) sig, zf = lfilter(b, a, buf.astype(np.float64), zi=zi) state["hpf_zi"] = zf # If 6dB slope, mix in a first-order version (less aggressive) if slope < 12: # First-order HPF: simpler - just one pole alpha = 1.0 / (1.0 + self._sample_rate / (2.0 * np.pi * freq)) zi1 = state.get("hpf_zi_1", np.float64(0.0)) out = np.zeros(len(buf), dtype=np.float64) for i in range(len(buf)): zi1 = alpha * (zi1 + buf[i] - state.get("hpf_last", np.float64(0.0))) out[i] = zi1 state["hpf_last"] = np.float64(buf[i]) state["hpf_zi_1"] = zi1 return np.clip(out, -1.0, 1.0).astype(np.float32) return np.clip(sig, -1.0, 1.0).astype(np.float32) def _apply_lpf(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Low-pass filter. Frequency, Slope (6/12dB).""" freq = params.get("frequency", 5000.0) slope = params.get("slope", 12.0) q = 0.707 if slope >= 12 else 0.5 coeffs = _compute_lpf_coeffs(freq, q, self._sample_rate) 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("lpf_zi", np.zeros(2, dtype=np.float64)) sig, zf = lfilter(b, a, buf.astype(np.float64), zi=zi) state["lpf_zi"] = zf if slope < 12: alpha = 1.0 / (1.0 + self._sample_rate / (2.0 * np.pi * freq)) zi1 = state.get("lpf_zi_1", np.float64(0.0)) out = np.zeros(len(buf), dtype=np.float64) for i in range(len(buf)): zi1 = zi1 + alpha * (buf[i] - zi1) out[i] = zi1 state["lpf_zi_1"] = zi1 return np.clip(out, -1.0, 1.0).astype(np.float32) return np.clip(sig, -1.0, 1.0).astype(np.float32) def _apply_bpf(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Band-pass filter. Frequency, Q.""" freq = params.get("frequency", 1000.0) q = params.get("q", 0.707) coeffs = _compute_bpf_coeffs(freq, q, self._sample_rate) 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("bpf_zi", np.zeros(2, dtype=np.float64)) sig, zf = lfilter(b, a, buf.astype(np.float64), zi=zi) state["bpf_zi"] = zf return np.clip(sig, -1.0, 1.0).astype(np.float32) def _apply_notch(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Notch filter. Frequency, Q.""" freq = params.get("frequency", 60.0) q = params.get("q", 10.0) # High Q for narrow notch coeffs = _compute_notch_coeffs(freq, q, self._sample_rate) 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("notch_zi", np.zeros(2, dtype=np.float64)) sig, zf = lfilter(b, a, buf.astype(np.float64), zi=zi) state["notch_zi"] = zf return np.clip(sig, -1.0, 1.0).astype(np.float32) # ── 38. Formant Filter ────────────────────────────────────────── def _apply_formant_filter(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Fixed resonant peaks (vowel shapes). Vowel (a/e/i/o/u), Mix.""" vowel = params.get("vowel", "a") mix = params.get("mix", 0.5) # Formant frequencies and bandwidths for vowels (F1, F2, F3) formants = { "a": [(700, 80), (1100, 90), (2500, 120)], "e": [(400, 60), (1800, 80), (2600, 100)], "i": [(300, 50), (2300, 80), (2900, 100)], "o": [(450, 70), (800, 80), (2700, 110)], "u": [(300, 50), (700, 70), (2700, 110)], } bands = formants.get(vowel, formants["a"]) sig = buf.astype(np.float64, copy=False) for stage, (freq, bw) in enumerate(bands): q_val = freq / bw # Q from bandwidth coeffs = _compute_peaking_coeffs(freq, 12.0, q_val, self._sample_rate) 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"form_zi_{stage}", np.zeros(2, dtype=np.float64)) sig, zf = lfilter(b, a, sig, zi=zi) state[f"form_zi_{stage}"] = zf wet = np.clip(sig, -1.0, 1.0).astype(np.float32) return np.clip(buf * (1.0 - mix) + wet * mix, -1.0, 1.0) # ── 39. Ping-pong Delay ───────────────────────────────────────── def _apply_ping_pong_delay(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Alternate left/right taps. Time, Feedback, Mix. Mono input: creates stereo effect by alternating. """ time_ms = params.get("time", 400.0) feedback = params.get("feedback", 0.3) mix = params.get("mix", 0.4) delay_samples = int(time_ms * self._sample_rate / 1000.0) if "delay" not in state: max_d = max(delay_samples * 2, self._sample_rate) state["delay"] = _DelayLine(max_d + 1) state["delay"].write_block(np.zeros(max_d // 2)) state["ping"] = 1 # 1 = left, -1 = right delay_line: _DelayLine = state["delay"] wet = delay_line.read_block(float(delay_samples), len(buf)) fb_gain = min(feedback, 0.98) # Alternate pan per block pan = state.get("ping", 1) panned_wet = wet * (0.5 + pan * 0.5) write_sig = buf + panned_wet * fb_gain delay_line.write_block(write_sig) state["ping"] = -pan return buf * (1.0 - mix) + panned_wet * mix # ── 40. Multi-tap Delay ───────────────────────────────────────── def _apply_multi_tap_delay(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Rhythmic repeats (dotted eighth, triplet). Pattern, Feedback, Mix.""" pattern = params.get("pattern", "quarter") # quarter/dotted/triplet feedback = params.get("feedback", 0.3) mix = params.get("mix", 0.4) # Tap times in ms, relative to quarter note = 500ms (120 BPM default) quarter = 500.0 tap_patterns = { "quarter": [quarter], "dotted": [quarter * 1.5], "triplet": [quarter / 3.0, quarter * 2.0 / 3.0, quarter], } taps = tap_patterns.get(pattern, tap_patterns["quarter"]) max_delay = int(max(taps) * self._sample_rate / 1000.0) if "delay" not in state: max_d = max(max_delay * 2, self._sample_rate) state["delay"] = _DelayLine(max_d + 1) state["delay"].write_block(np.zeros(max_d // 2)) delay_line: _DelayLine = state["delay"] # Sum all tap reads wet = np.zeros(len(buf), dtype=np.float32) for tap_ms in taps: tap_samples = int(tap_ms * self._sample_rate / 1000.0) tap_sig = delay_line.read_block(float(tap_samples), len(buf)) wet += tap_sig * (1.0 / len(taps)) # Normalise fb_gain = min(feedback, 0.98) delay_line.write_block(np.clip(buf + wet * fb_gain, -1.0, 1.0)) return np.clip(buf * (1.0 - mix) + wet * mix, -1.0, 1.0) # ── 41. Reverse Delay ─────────────────────────────────────────── def _apply_reverse_delay(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Buffer + reverse before playback. Time, Feedback, Mix.""" time_ms = params.get("time", 400.0) feedback = params.get("feedback", 0.3) mix = params.get("mix", 0.4) delay_samples = int(time_ms * self._sample_rate / 1000.0) if "delay" not in state: max_d = max(delay_samples * 2, self._sample_rate) state["delay"] = _DelayLine(max_d + 1) state["delay"].write_block(np.zeros(max_d // 2)) state["read_pos"] = 0 delay_line: _DelayLine = state["delay"] rpos = state.get("read_pos", 0) # Read reversed: read from buffer start, write at current pos # Reverse: read from (write_pos - read_offset) going backwards wpos = delay_line.write_idx buf_len = len(buf) out = np.zeros(buf_len, dtype=np.float32) for i in range(buf_len): # Read sample at (wpos - rpos) mod max_len rev_idx = (wpos - rpos) % delay_line.max_len out[i] = delay_line.buf[rev_idx] rpos = (rpos + 1) % delay_samples state["read_pos"] = rpos % delay_samples fb_gain = min(feedback, 0.98) delay_line.write_block(np.clip(buf + out * fb_gain, -1.0, 1.0)) return np.clip(buf * (1.0 - mix) + out * mix, -1.0, 1.0) # ── 42. Tape Echo ─────────────────────────────────────────────── def _apply_tape_echo(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Delay with wow/flutter + high-cut feedback. Time, Wow, High-Cut, Mix.""" time_ms = params.get("time", 300.0) wow = params.get("wow", 0.3) # wow/flutter intensity high_cut = params.get("high_cut", 0.5) # 0.0-1.0 high cut feedback = params.get("feedback", 0.3) mix = params.get("mix", 0.4) delay_samples = int(time_ms * self._sample_rate / 1000.0) if "delay" not in state: max_d = max(delay_samples * 2, self._sample_rate) state["delay"] = _DelayLine(max_d + 1) state["delay"].write_block(np.zeros(max_d // 2)) delay_line: _DelayLine = state["delay"] # Wow/flutter: LFO on read position (wobble) phase = self._lfo_phase(4.0, state, len(buf)) # 4Hz wow lfo = self._lfo_wave(phase, "sine") wow_offset = (lfo * 2.0 - 1.0) * wow * 3.0 # up to ±3 samples wobble read_delay = float(delay_samples) + wow_offset wet = delay_line.read_block_varying(read_delay) # High-cut: one-pole low-pass on feedback if high_cut > 0: cutoff_factor = 1.0 - high_cut * 0.9 # 0.1-1.0 lp_prev = state.get("lp_prev", 0.0) lp_out = np.zeros_like(wet) for i in range(len(wet)): lp_prev = lp_prev * cutoff_factor + wet[i] * (1.0 - cutoff_factor) lp_out[i] = lp_prev state["lp_prev"] = float(lp_prev) wet = lp_out.astype(np.float32) fb_gain = min(feedback, 0.98) delay_line.write_block(np.clip(buf + wet * fb_gain, -1.0, 1.0)) return np.clip(buf * (1.0 - mix) + wet * mix, -1.0, 1.0) # ── 43. Shimmer Reverb ────────────────────────────────────────── # BETA — test on RPi 4B for xruns def _apply_shimmer_reverb(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Reverb + pitch shift + re-verb. Shift, Decay, Mix. CPU-heavy, tagged as beta. """ # BETA — test on RPi 4B for xruns shift = params.get("shift", 0.0) # semitones decay = params.get("decay", 0.5) mix = params.get("mix", 0.4) # Use regular reverb as base if "combs" not in state: comb_delays = [29, 37, 44, 50, 31, 39, 47, 53] ap_delays = [5, 7, 11, 13] state["combs"] = [_CombFilter(int(d * self._sample_rate / 1000.0), block_size=self._block_size) for d in comb_delays] state["allpasses"] = [_AllpassFilter(int(d * self._sample_rate / 1000.0), block_size=self._block_size) for d in ap_delays] state["predelay"] = _DelayLine(int(30.0 * self._sample_rate / 1000.0 + 1)) state["predelay"].write_block(np.zeros(int(30.0 * self._sample_rate / 1000.0))) combs: list[_CombFilter] = state["combs"] allpasses: list[_AllpassFilter] = state["allpasses"] predelay_line: _DelayLine = state["predelay"] scaled_fb = 0.3 + decay * 0.6 scaled_damp = 0.1 + decay * 0.5 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 + decay * 0.3 # Predelay delayed = predelay_line.read_block(30.0 * self._sample_rate / 1000.0, len(buf)) predelay_line.write_block(buf) # Comb + allpass reverb wet = np.zeros_like(buf, dtype=np.float64) for comb in combs: wet += comb.process(delayed) wet /= len(combs) for ap in allpasses: wet = ap.process(wet) # Pitch shift the reverb tail (shimmer!) if shift != 0.0: factor = 2.0 ** (shift / 12.0) if "shift_ring" not in state: grain_size = int(0.040 * self._sample_rate) state["shift_ring"] = np.zeros(grain_size * 2, dtype=np.float32) state["swpos"] = 0 state["srpos"] = 0 ring = state["shift_ring"] wpos = state["swpos"] rpos = state["srpos"] wet_float = wet.astype(np.float32) for i, s in enumerate(wet_float): ring[wpos % len(ring)] = s wpos += 1 shifted_out = np.zeros_like(buf, dtype=np.float32) for i in range(len(buf)): idx = int(np.floor(rpos)) frac = rpos - idx a = ring[idx % len(ring)] b = ring[(idx + 1) % len(ring)] shifted_out[i] = a + frac * (b - a) rpos += factor if rpos >= wpos: rpos = wpos - 1 state["swpos"] = wpos state["srpos"] = rpos window = 0.5 * (1.0 - np.cos(2.0 * np.pi * np.arange(len(buf)) / len(buf))) shifted_out = shifted_out * window wet = wet + shifted_out.astype(np.float64) * 0.5 # feedback shimmer wet = np.clip(wet, -1.0, 1.0).astype(np.float32) return np.clip(buf * (1.0 - mix) + wet * mix, -1.0, 1.0) # ── 44. Looper ───────────────────────────────────────────────── def _apply_looper(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Record/overdub/playback (~30s buffer = 6MB). Controls: record, overdub, play, stop. State machine: idle -> recording -> playing -> overdub -> idle """ max_buffer = int(30.0 * self._sample_rate) # 30 seconds if "looper_buf" not in state: state["looper_buf"] = np.zeros(max_buffer, dtype=np.float32) state["looper_pos"] = 0 state["looper_len"] = 0 state["looper_mode"] = "idle" looper_buf: np.ndarray = state["looper_buf"] pos = state.get("looper_pos", 0) buf_len = state.get("looper_len", 0) mode = state.get("looper_mode", "idle") # Control actions if params.get("record", 0): mode = "recording" state["looper_len"] = 0 state["looper_pos"] = 0 looper_buf[:] = 0.0 elif params.get("overdub", 0): mode = "overdub" state["looper_pos"] = 0 elif params.get("play", 0): mode = "playing" state["looper_pos"] = 0 elif params.get("stop", 0): mode = "idle" state["looper_mode"] = mode if mode == "idle": return buf # passthrough out = np.zeros_like(buf) n = len(buf) if mode == "recording": # Write input to buffer end = min(pos + n, max_buffer) looper_buf[pos:end] = buf[:end - pos] state["looper_pos"] = pos + n state["looper_len"] = max(state["looper_len"], pos + n) out = buf * 0.5 # Monitor at half volume during recording elif mode == "overdub": # Play existing + record new for i in range(n): looper_buf[pos % max_buffer] = ( looper_buf[pos % max_buffer] + buf[i] ) * 0.7 # Mix down to avoid clipping out[i] = looper_buf[pos % max_buffer] pos += 1 state["looper_pos"] = pos state["looper_len"] = max(state["looper_len"], pos) elif mode == "playing": # Play from buffer for i in range(n): out[i] = looper_buf[pos % buf_len] if buf_len > 0 else 0.0 pos += 1 state["looper_pos"] = pos return out.astype(np.float32) # ── 45. Early Reflections ─────────────────────────────────────── def _apply_early_reflections(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Sparse delay taps + allpass for room simulation. Size, Decay, Mix. """ room_size = params.get("size", 0.5) decay = params.get("decay", 0.4) mix = params.get("mix", 0.4) # Early reflection tap times (ms) scaled by room size base_taps = [5, 12, 20, 30, 45, 65] size_factor = 0.5 + room_size * 2.0 # 0.5-2.5x taps = [int(t * size_factor * self._sample_rate / 1000.0) for t in base_taps] if "delay" not in state: max_d = max(taps) + self._block_size + 1 state["delay"] = _DelayLine(max_d) delay_line: _DelayLine = state["delay"] delay_line.write_block(buf) # Read reflections with decreasing amplitude wet = np.zeros(len(buf), dtype=np.float32) amplitude = 0.6 + decay * 0.5 # 0.6-1.1 for i, tap in enumerate(taps): gain = amplitude * (1.0 - i / len(taps)) reflected = delay_line.read_block(float(tap), len(buf)) wet += reflected * gain # Normalize wet = wet / max(len(taps), 1) * 0.5 return (buf * (1.0 - mix) + wet * mix).astype(np.float32) def set_audio_profile(self, block_size: int, sample_rate: int) -> None: """Update block size and sample rate at runtime. Recomputes profile-dependent constants and clears DSP state so effects reinitialise with the new buffer sizes on the next process() call. Called by the web server when the JACK latency profile changes (``POST /api/audio/profile``). Follows the same pattern as :meth:`NAMEngineRouter.set_block_size`. Args: block_size: JACK period size in frames (e.g. 64, 128, 256, 512). sample_rate: Sample rate in Hz (e.g. 44100, 48000, 96000). """ changed = (self._block_size != block_size or self._sample_rate != sample_rate) if not changed: return self._block_size = block_size self._sample_rate = sample_rate self._vu_alpha = np.exp(-block_size / (0.05 * sample_rate)) # Clear DSP state — effects will reinit with new block/sample rate self._state.clear() self._coeffs.clear() logger.info("Audio profile updated: block=%d, sr=%d", block_size, sample_rate) @property def block_size(self) -> int: """Current JACK period / block size in frames.""" return self._block_size @property def sample_rate(self) -> int: """Current audio sample rate in Hz.""" return self._sample_rate # ── 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) # ── Chain access (read-only for external inspection) ──────────── @property def chain(self) -> list[dict]: """Read-only access to the DSP chain.""" return list(self._chain) 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