"""FX/Audio pipeline — the main real-time signal chain. Runs on RPi 4B under JACK, connecting: Guitar -> Gate -> Comp -> Boost -> NAM Amp -> IR Cab -> EQ -> Mod -> Delay -> Reverb -> Volume -> Out Each block can be bypassed per-preset. The pipeline manages block-level audio routing using numpy arrays for zero-copy inter-block communication. All DSP state is stored per-block-instance in self._state, keyed by chain index. This allows multiple instances of the same effect type at different positions in the chain. """ from __future__ import annotations import logging from dataclasses import dataclass, field from typing import Optional import numpy as np from scipy.signal import lfilter from .nam_host import NAMHost, NAMModel, ModelSwitchMode from .ir_loader import IRLoader, IRFile from ..presets.types import FXBlock, FXType, Preset logger = logging.getLogger(__name__) BLOCK_SIZE = 256 # Samples per JACK callback SAMPLE_RATE = 48000 # Standard guitar audio rate # ── Biquad coefficient helpers ───────────────────────────────────── _EPS = 1e-10 def _compute_lowshelf_coeffs(freq: float, gain_db: float, q: float, sr: float) -> tuple: """RBJ low-shelf biquad coefficients.""" a = 10 ** (gain_db / 40.0) omega = 2 * np.pi * freq / sr sn = np.sin(omega) cs = np.cos(omega) beta = np.sqrt(a) / q # sqrt(A) / Q if gain_db >= 0: b0 = a * (a + 1 - (a - 1) * cs + beta * sn) b1 = 2 * a * (a - 1 - (a + 1) * cs) b2 = a * (a + 1 - (a - 1) * cs - beta * sn) a0 = a + 1 + (a - 1) * cs + beta * sn a1 = -2 * a * (a - 1 + (a + 1) * cs) a2 = a + 1 + (a - 1) * cs - beta * sn else: b0 = a * (a + 1 + (a - 1) * cs + beta * sn) b1 = -2 * a * (a - 1 + (a + 1) * cs) b2 = a * (a + 1 + (a - 1) * cs - beta * sn) a0 = a + 1 - (a - 1) * cs + beta * sn a1 = 2 * a * (a - 1 - (a + 1) * cs) a2 = a + 1 - (a - 1) * cs - beta * sn return (b0 / a0, b1 / a0, b2 / a0, a1 / a0, a2 / a0) def _compute_highshelf_coeffs(freq: float, gain_db: float, q: float, sr: float) -> tuple: """RBJ high-shelf biquad coefficients.""" a = 10 ** (gain_db / 40.0) omega = 2 * np.pi * freq / sr sn = np.sin(omega) cs = np.cos(omega) beta = np.sqrt(a) / q if gain_db >= 0: b0 = a * (a + 1 + (a - 1) * cs + beta * sn) b1 = -2 * a * (a - 1 + (a + 1) * cs) b2 = a * (a + 1 + (a - 1) * cs - beta * sn) a0 = a + 1 - (a - 1) * cs + beta * sn a1 = 2 * a * (a - 1 - (a + 1) * cs) a2 = a + 1 - (a - 1) * cs - beta * sn else: b0 = a * (a + 1 - (a - 1) * cs + beta * sn) b1 = 2 * a * (a - 1 - (a + 1) * cs) b2 = a * (a + 1 - (a - 1) * cs - beta * sn) a0 = a + 1 + (a - 1) * cs + beta * sn a1 = -2 * a * (a - 1 + (a + 1) * cs) a2 = a + 1 + (a - 1) * cs - beta * sn return (b0 / a0, b1 / a0, b2 / a0, a1 / a0, a2 / a0) def _compute_peaking_coeffs(freq: float, gain_db: float, q: float, sr: float) -> tuple: """RBJ peaking biquad coefficients.""" a = 10 ** (gain_db / 40.0) omega = 2 * np.pi * freq / sr sn = np.sin(omega) cs = np.cos(omega) alpha = sn / (2 * q) b0 = 1 + alpha * a b1 = -2 * cs b2 = 1 - alpha * a a0 = 1 + alpha / a a1 = -2 * cs a2 = 1 - alpha / a return (b0 / a0, b1 / a0, b2 / a0, a1 / a0, a2 / a0) 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") def __init__(self, delay_samples: int): line_len = max(BLOCK_SIZE * 2, delay_samples + 1) self.delay = _DelayLine(line_len) self.feedback: float = 0.5 self.damping: float = 0.5 # low-pass damping coefficient self.damp_filt: float = 0.0 # state variable for damping self.buf = np.zeros(BLOCK_SIZE, dtype=np.float32) def process(self, block: np.ndarray) -> np.ndarray: self.buf[:] = block # Write with feedback: out[n] = in[n] + feedback * damped_delayed delayed = self.delay.add_to_block(self.buf, self.delay.max_len - 1, self.feedback) # One-pole low-pass on feedback path (vectorised) b = np.array([1.0 - self.damping], dtype=np.float64) a = np.array([1.0, -self.damping], dtype=np.float64) damped, _ = lfilter(b, a, delayed.astype(np.float64), zi=np.atleast_1d(self.damp_filt)) self.damp_filt = float(damped[-1]) self.buf[:] = block + damped.astype(np.float32) self.delay.write_block(self.buf) return self.buf class _AllpassFilter: """Allpass filter for Schroeder reverb.""" __slots__ = ("delay", "gain", "buf") def __init__(self, delay_samples: int): line_len = max(BLOCK_SIZE * 2, delay_samples + 1) self.delay = _DelayLine(line_len) self.gain: float = 0.5 self.buf = np.zeros(BLOCK_SIZE, dtype=np.float32) def process(self, block: np.ndarray) -> np.ndarray: # out[n] = -gain * in[n] + delay[n - D] + gain * delay_output[n - D] # Standard allpass: out = -g * in + delayed + g * delayed_out # But block-wise: read delayed, write in + g * delayed, output = -g * in + delayed self.buf[:] = block delayed = self.delay.add_to_block(self.buf, self.delay.max_len - 1, self.gain) # Write: buf + gain * delayed self.buf[:] = block + delayed * self.gain self.delay.write_block(self.buf) # Output: -gain * block + delayed return -self.gain * block + delayed # ── Audio Pipeline ───────────────────────────────────────────────── class AudioPipeline: """Orchestrates the real-time audio FX chain. The pipeline processes audio block-by-block, chaining effect modules in order. Each module receives a numpy array of audio samples and returns processed samples. Effect state (delay buffers, LFO phases, envelope followers, filter memory) is stored per-instance in self._state. """ def __init__( self, nam_host: Optional[NAMHost] = None, ir_loader: Optional[IRLoader] = None, ): self.nam = nam_host or NAMHost() self.ir = ir_loader or IRLoader() # Signal chain — list of (FXType, enabled, bypass, params) self._chain: list[dict] = [] self._master_volume: float = 0.8 self._tuner_enabled: bool = False self._bypassed: bool = False # Global bypass # 4-Cable Method routing self._routing_mode: str = "mono" # "mono" or "4cm" self._routing_breakpoint: int = 7 # chain index where pre/post split occurs # Per-block DSP state: {f"fx_{idx}": {state_dict}} self._state: dict[str, dict] = {} # Cached filter coefficients per block self._coeffs: dict[str, tuple] = {} # 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 at 48kHz/256 block self._vu_alpha: float = np.exp(-BLOCK_SIZE / (0.05 * SAMPLE_RATE)) # ── 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 logger.info("Audio pipeline initialized (block=%d, sr=%d)", BLOCK_SIZE, SAMPLE_RATE) def load_preset(self, preset: Preset) -> None: """Load a complete preset (NAM, IR, and FX chain).""" self._chain = [] self._state = {} self._coeffs = {} for block in preset.chain: entry = { "fx_type": block.fx_type, "enabled": block.enabled, "bypass": block.bypass, "params": dict(block.params), } # Load NAM model if needed if block.fx_type == FXType.NAM_AMP and block.nam_model_path: self.nam.load_model(block.nam_model_path) # Load IR if needed if block.fx_type == FXType.IR_CAB and block.ir_file_path: self.ir.load_ir(block.ir_file_path) self._chain.append(entry) self._master_volume = preset.master_volume self._tuner_enabled = preset.tuner_enabled # Set 4CM routing from preset self._routing_mode = preset.routing_mode self._routing_breakpoint = preset.routing_breakpoint logger.info("Preset '%s' loaded: %d blocks, routing=%s breakpoint=%d", preset.name, len(self._chain), self._routing_mode, self._routing_breakpoint) def process(self, audio_in: np.ndarray) -> np.ndarray: """Process a block of audio through the entire FX chain. Args: audio_in: numpy array of PCM samples (float32 [-1, 1]). Mono mode: shape (N,) — single audio channel. 4CM mode: shape (2, N) — two channels, [guitar_in, return_in]. Returns: Processed audio block. Mono mode: shape (N,) — processed output. 4CM mode: shape (2, N) — [send_out, return_out]. """ # ── Tuner mode: mute output, keep input tracking for pitch detection ── if self._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 self._bypassed: return audio_in * self._master_volume if self._routing_mode == "4cm": return self._process_4cm(audio_in) else: return self._process_mono(audio_in) # ── 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(SAMPLE_RATE / 1600) # ~30 max_lag = min(int(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:] # Normalize by energy at each lag (YIN-style) energy = np.cumsum(buf ** 2) energy = energy[max(1, min_lag):max_lag + len(buf) - len(corr) + min_lag + 1] # Fallback: use 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 = 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) -> 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(self._chain): if entry["bypass"] or not entry["enabled"]: continue buf = self._process_single_block(buf, idx, entry) out = buf * self._master_volume # 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) -> 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 = self._routing_breakpoint for idx, entry in enumerate(self._chain): if entry["bypass"] or not entry["enabled"]: continue if idx < bp: # Pre-amp block — process on guitar (ch0) ch0 = self._process_single_block(ch0, idx, entry) else: # Post-amp block — process on return (ch1) ch1 = self._process_single_block(ch1, idx, entry) out = np.zeros_like(audio_in) out[0, :] = ch0 * self._master_volume out[1, :] = ch1 * self._master_volume # 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) -> np.ndarray: """Process a single mono audio block through one FX block. Args: buf: Mono audio block (N,) to process. idx: Chain index for state lookup. entry: Chain entry dict with fx_type, params. Returns: Processed mono block (N,). """ fx_type = entry["fx_type"] params = entry["params"] fx_state = self._state.setdefault(f"fx_{idx}", {}) match fx_type: case FXType.NOISE_GATE: return self._apply_gate(buf, params, fx_state) case FXType.COMPRESSOR: return self._apply_compressor(buf, params, fx_state) case FXType.BOOST: return self._apply_boost(buf, params, fx_state) case FXType.OVERDRIVE: return self._apply_overdrive(buf, params, fx_state) case FXType.DISTORTION: return self._apply_distortion(buf, params, fx_state) case FXType.FUZZ: return self._apply_fuzz(buf, params, fx_state) case FXType.EQ: return self._apply_eq(buf, params, fx_state) case FXType.CHORUS: return self._apply_chorus(buf, params, fx_state) case FXType.FLANGER: return self._apply_flanger(buf, params, fx_state) case FXType.PHASER: return self._apply_phaser(buf, params, fx_state) case FXType.TREMOLO: return self._apply_tremolo(buf, params, fx_state) case FXType.VIBRATO: return self._apply_vibrato(buf, params, fx_state) case FXType.DELAY: return self._apply_delay(buf, params, fx_state) case FXType.REVERB: return self._apply_reverb(buf, params, fx_state) case FXType.VOLUME: return self._apply_volume(buf, params, fx_state) case FXType.NAM_AMP: if self.nam.is_loaded: processed = self.nam.process(buf) if self.nam._crossfade_buf is not None: processed = self.nam.apply_crossfade(processed) return processed return buf case FXType.IR_CAB: if self.ir.is_loaded: return self._apply_ir_cab(buf, params, fx_state) return buf # ── 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 ───────────────────────────────────────────────── @staticmethod def _lfo_phase(rate_hz: float, state: dict, block_size: int) -> np.ndarray: """Generate LFO phase ramp (0->1), update state.""" phase = state.get("phase", 0.0) delta = rate_hz / SAMPLE_RATE t = np.arange(block_size, dtype=np.float64) * delta + phase t %= 1.0 state["phase"] = float(t[-1] + delta) % 1.0 return t @staticmethod def _lfo_wave(phase: np.ndarray, shape: str = "sine") -> np.ndarray: """Generate LFO waveform from phase array.""" match shape: case "sine": return 0.5 + 0.5 * np.sin(2 * np.pi * phase) case "triangle": return 2.0 * np.abs(2.0 * phase - 1.0) - 1.0 # Returns in [-1, 1]; normalise below case "square": return np.where(phase < 0.5, 1.0, 0.0) case _: return 0.5 + 0.5 * np.sin(2 * np.pi * phase) # ── Effect implementations ────────────────────────────────────── # ── 1. Noise Gate ─────────────────────────────────────────────── def _apply_gate(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Noise gate with adjustable threshold and release envelope.""" threshold = params.get("threshold", 0.01) release_ms = params.get("release", 100.0) envelope = state.get("envelope", 0.0) rms = np.sqrt(np.mean(buf ** 2) + _EPS) if rms >= threshold: # Instant attack envelope = rms else: # Exponential release — time constant per block release_coeff = np.exp(-BLOCK_SIZE / (release_ms * SAMPLE_RATE / 1000.0)) envelope = envelope * release_coeff + rms * (1.0 - release_coeff) state["envelope"] = envelope if envelope < threshold: return np.zeros_like(buf) return buf # ── 2. Compressor ─────────────────────────────────────────────── def _apply_compressor(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Compressor with threshold (dB), ratio, attack, release, make-up gain.""" threshold_db = params.get("threshold", -20.0) # dB ratio = params.get("ratio", 3.0) attack_ms = params.get("attack", 5.0) release_ms = params.get("release", 100.0) makeup = params.get("gain", 1.0) # RMS envelope with attack/release shaping rms = np.sqrt(np.mean(buf ** 2) + _EPS) envelope = state.get("envelope", 0.0) if rms > envelope: alpha = np.exp(-BLOCK_SIZE / (attack_ms * SAMPLE_RATE / 1000.0)) else: alpha = np.exp(-BLOCK_SIZE / (release_ms * SAMPLE_RATE / 1000.0)) envelope = envelope * alpha + rms * (1.0 - alpha) state["envelope"] = envelope # Compute gain reduction in dB domain if envelope > 1e-10: env_db = 20.0 * np.log10(envelope) else: env_db = -120.0 if env_db > threshold_db: # gain_db = threshold + (env - threshold) / ratio - env gain_db = threshold_db + (env_db - threshold_db) / ratio - env_db else: gain_db = 0.0 gain_lin = 10 ** (gain_db / 20.0) return np.clip(buf * gain_lin * makeup, -1.0, 1.0) # ── 3. Boost / Overdrive / Distortion / Fuzz ──────────────────── def _apply_boost(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Clean boost with linear gain.""" gain_db = params.get("gain_db", 6.0) gain_linear = 10 ** (gain_db / 20.0) return np.clip(buf * gain_linear, -1.0, 1.0) def _apply_overdrive(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Tube-style overdrive with asymmetric soft clipping.""" drive = params.get("drive", 0.5) tone = params.get("tone", 0.5) gain = params.get("gain", 1.0) drive_scaled = drive * 15.0 + 1.0 shaped = buf * drive_scaled # Asymmetric soft clipping (tube-like) # Positive half clips softer than negative (tube asymmetry) pos = np.where(shaped > 0, shaped / (1.0 + shaped * 0.3), shaped) neg = np.where(pos < 0, pos / (1.0 - pos * 0.5), pos) out = np.tanh(neg) # Final polish with tanh return np.clip(out * gain, -1.0, 1.0) def _apply_distortion(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Harder asymmetric clipping with diode-style transfer.""" drive = params.get("drive", 0.7) tone = params.get("tone", 0.5) gain = params.get("gain", 1.0) drive_scaled = drive * 30.0 + 1.0 shaped = buf * drive_scaled # Diode-style asymmetric clipping clipped = np.where( shaped > 0, np.clip(shaped, 0, 0.8) / (1.0 + np.abs(np.clip(shaped, 0, 0.8)) * 0.5), np.clip(shaped, -0.6, 0) / (1.0 + np.abs(np.clip(shaped, -0.6, 0)) * 0.3), ) return np.clip(clipped * gain, -1.0, 1.0) def _apply_fuzz(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Octave-fuzzy hard clipping with gated sustain.""" drive = params.get("drive", 0.8) tone = params.get("tone", 0.5) gain = params.get("gain", 1.0) drive_scaled = drive * 50.0 + 1.0 shaped = buf * drive_scaled # Hard square-wave clip with asymmetric gate clipped = np.sign(shaped) * (1.0 - np.exp(-np.abs(shaped) * 2.0)) # Foldover for octave effect folded = np.abs(clipped) * 0.3 + clipped * 0.7 return np.clip(folded * gain, -1.0, 1.0) # ── 4. Three-band EQ ──────────────────────────────────────────── def _apply_eq(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """3-band EQ: bass shelf, mid peaking, treble shelf. Uses scipy.signal.lfilter with persistent state for zero-crosstalk between blocks. """ bass_gain = params.get("bass", 0.0) # dB mid_gain = params.get("mid", 0.0) # dB treble_gain = params.get("treble", 0.0) # dB bass_freq = params.get("bass_freq", 200.0) mid_freq = params.get("mid_freq", 1000.0) treble_freq = params.get("treble_freq", 3500.0) q = params.get("q", 0.707) sig = buf.astype(np.float64, copy=False) for band_name, freq, gain_db, compute_fn in [ ("bass", bass_freq, bass_gain, _compute_lowshelf_coeffs), ("mid", mid_freq, mid_gain, _compute_peaking_coeffs), ("treble", treble_freq, treble_gain, _compute_highshelf_coeffs), ]: if gain_db == 0.0: continue key = f"eq_{band_name}" coeffs = state.get(f"{key}_coeffs") param_tag = (bass_freq, mid_freq, treble_freq, bass_gain, mid_gain, treble_gain, q) if coeffs is None or state.get(f"{key}_tag") != param_tag: coeffs = compute_fn(freq, gain_db, q, SAMPLE_RATE) state[f"{key}_coeffs"] = coeffs state[f"{key}_tag"] = param_tag b0, b1, b2, a1, a2 = coeffs b = np.array([b0, b1, b2], dtype=np.float64) a = np.array([1.0, a1, a2], dtype=np.float64) zi = state.get(f"{key}_zi", np.zeros(2, dtype=np.float64)) sig, zf = lfilter(b, a, sig, zi=zi) state[f"{key}_zi"] = zf return np.clip(sig, -1.0, 1.0).astype(np.float32) # ── 5. Chorus ─────────────────────────────────────────────────── def _apply_chorus(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Chorus with LFO-driven modulated delay line (stereo-ish).""" rate = params.get("rate", 0.5) # Hz depth = params.get("depth", 0.5) # 0.0-1.0 mix = params.get("mix", 0.5) # wet/dry delay_base = params.get("delay", 20.0) # ms (typical chorus: 15-30ms) base_samples = delay_base * SAMPLE_RATE / 1000.0 mod_range = depth * 5.0 * SAMPLE_RATE / 1000.0 if "delay" not in state: max_d = int(base_samples + mod_range + 10.0 * SAMPLE_RATE / 1000.0) + 1 state["delay"] = _DelayLine(max_d) state["delay"].write_block(np.zeros(max_d)) delay_line: _DelayLine = state["delay"] phase = self._lfo_phase(rate, state, len(buf)) lfo = self._lfo_wave(phase, "sine") mod_delay = base_samples + lfo * mod_range # Vectorised read wet = delay_line.read_block_varying(mod_delay) delay_line.write_block(buf) return buf * (1.0 - mix) + wet * mix # ── 6. Flanger ────────────────────────────────────────────────── def _apply_flanger(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Flanger with swept comb filter and feedback.""" rate = params.get("rate", 0.25) # Hz (slower than chorus) depth = params.get("depth", 0.7) # 0.0-1.0 feedback = params.get("feedback", 0.3) mix = params.get("mix", 0.5) # wet/dry delay_base = params.get("delay", 5.0) # ms (typical flanger: 1-10ms) base_samples = delay_base * SAMPLE_RATE / 1000.0 mod_range = depth * 5.0 * SAMPLE_RATE / 1000.0 if "delay" not in state: max_d = int(base_samples + mod_range + 10.0 * SAMPLE_RATE / 1000.0) + 1 state["delay"] = _DelayLine(max_d) state["delay"].write_block(np.zeros(max_d)) delay_line: _DelayLine = state["delay"] phase = self._lfo_phase(rate, state, len(buf)) lfo = self._lfo_wave(phase, "sine") mod_delay = base_samples + lfo * mod_range # Feedback buffer feedback_buf = state.get("fb_buf", np.zeros(len(buf), dtype=np.float32)) # Blend feedback into input fb_input = buf + feedback_buf * feedback # Vectorised read wet = delay_line.read_block_varying(mod_delay) delay_line.write_block(fb_input) # Store feedback for next block state["fb_buf"] = wet * 0.5 return buf * (1.0 - mix) + wet * mix # ── 7. Phaser ─────────────────────────────────────────────────── def _apply_phaser(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Phaser with allpass filter cascade and feedback.""" rate = params.get("rate", 0.4) # Hz depth = params.get("depth", 0.5) # 0.0-1.0 feedback = params.get("feedback", 0.3) mix = params.get("mix", 0.5) stages = int(params.get("stages", 4)) # Map LFO to centre frequency sweep: 200-2000 Hz phase = self._lfo_phase(rate, state, len(buf)) lfo = self._lfo_wave(phase, "sine") freq_range = 200.0 + lfo * depth * 1800.0 fb_buf = state.get("fb_buf", np.zeros(len(buf), dtype=np.float64)) fb_input = buf.astype(np.float64, copy=False) + fb_buf * feedback sig = fb_input.copy() for stage in range(stages): # Allpass as first-order IIR: H(z) = (coeff + z^-1) / (1 + coeff * z^-1) # Which is lfilter(b=[coeff, 1], a=[1, coeff], ...) # But coeff varies per sample (LFO-driven)! Can't use lfilter directly. # Use block-constant approximation: one coeff per block at LFO centre. freq = np.mean(freq_range) w = 2.0 * np.pi * freq / SAMPLE_RATE tan_half_w = np.tan(w / 2.0) coeff = (1.0 - tan_half_w) / (1.0 + tan_half_w) b = np.array([coeff, 1.0], dtype=np.float64) a = np.array([1.0, coeff], dtype=np.float64) zi = state.get(f"ap_zi_{stage}", np.zeros(1, dtype=np.float64)) sig, zf = lfilter(b, a, sig, zi=zi) state[f"ap_zi_{stage}"] = zf state["fb_buf"] = sig * 0.5 sig = np.clip(sig, -1.0, 1.0) return (buf * (1.0 - mix) + sig.astype(np.float32) * mix).astype(np.float32) # ── 8. Tremolo ────────────────────────────────────────────────── def _apply_tremolo(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Tremolo with configurable LFO shape.""" rate = params.get("rate", 4.0) # Hz depth = params.get("depth", 0.7) shape = params.get("shape", "sine") # sine / triangle / square phase = self._lfo_phase(rate, state, len(buf)) lfo = self._lfo_wave(phase, shape) # LFO is 0-1; tremolo scales between full volume and attenuated mod = 1.0 - depth * (1.0 - lfo) return buf * mod # ── 9. Vibrato ────────────────────────────────────────────────── def _apply_vibrato(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Vibrato — modulated delay with 100% wet (pitch modulation).""" rate = params.get("rate", 3.0) # Hz depth = params.get("depth", 0.5) # cents equivalent base_samples = 2.0 * SAMPLE_RATE / 1000.0 # fixed ~2ms base mod_range = depth * 3.0 * SAMPLE_RATE / 1000.0 if "delay" not in state: max_d = int(base_samples + mod_range + 5.0 * SAMPLE_RATE / 1000.0) + 1 state["delay"] = _DelayLine(max_d) state["delay"].write_block(np.zeros(max_d)) delay_line: _DelayLine = state["delay"] phase = self._lfo_phase(rate, state, len(buf)) lfo = self._lfo_wave(phase, "sine") mod_delay = base_samples + lfo * mod_range wet = delay_line.read_block_varying(mod_delay) delay_line.write_block(buf) return wet # ── 10. Delay ─────────────────────────────────────────────────── def _apply_delay(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Digital delay with feedback and tap-tempo support.""" time_ms = params.get("time", 400.0) feedback = params.get("feedback", 0.3) mix = params.get("mix", 0.4) tap_tempo = params.get("tap_tempo", 0.0) # Tap tempo overrides time_ms when > 0 if tap_tempo > 0: time_ms = tap_tempo delay_samples = int(time_ms * SAMPLE_RATE / 1000.0) if "delay" not in state: # Allocate 2x requested delay for headroom max_d = max(delay_samples * 2, SAMPLE_RATE) # at least 1s state["delay"] = _DelayLine(max_d + 1) state["delay"].write_block(np.zeros(max_d // 2)) delay_line: _DelayLine = state["delay"] # Read delayed signal wet = delay_line.read_block(float(delay_samples), len(buf)) # Write dry + feedback (no self-oscillation guard) # clips feedback automatically fb_gain = min(feedback, 0.98) write_sig = buf + wet * fb_gain delay_line.write_block(write_sig) return buf * (1.0 - mix) + wet * mix # ── 11. Reverb (Schroeder) ────────────────────────────────────── def _apply_reverb(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Schroeder reverb: 8 comb filters + 4 allpass filters in series.""" decay = params.get("decay", 0.5) damping = params.get("damping", 0.4) mix = params.get("mix", 0.3) predelay_ms = params.get("predelay", 30.0) # Initialise on first call if "combs" not in state: # Classic Schroeder delays (prime-ish numbers for de-flanging) comb_delays = [29, 37, 44, 50, 31, 39, 47, 53] # ms ap_delays = [5, 7, 11, 13] # ms state["combs"] = [ _CombFilter(int(d * SAMPLE_RATE / 1000.0)) for d in comb_delays ] state["allpasses"] = [ _AllpassFilter(int(d * SAMPLE_RATE / 1000.0)) for d in ap_delays ] state["predelay"] = _DelayLine( int(predelay_ms * SAMPLE_RATE / 1000.0) + 1 ) state["predelay"].write_block(np.zeros(BLOCK_SIZE)) state["_computed"] = False combs: list[_CombFilter] = state["combs"] allpasses: list[_AllpassFilter] = state["allpasses"] predelay_line: _DelayLine = state["predelay"] # Update comb parameters when decay/damping changes param_tag = (decay, damping) if state.get("_param_tag") != param_tag: scaled_fb = 0.3 + decay * 0.6 # 0.3 - 0.9 scaled_damp = 0.1 + damping * 0.7 # 0.1 - 0.8 for comb in combs: comb.feedback = min(scaled_fb, 0.95) comb.damping = min(scaled_damp, 0.85) for ap in allpasses: ap.gain = 0.3 + damping * 0.3 state["_param_tag"] = param_tag # Predelay delayed = predelay_line.read_block(float(predelay_ms * SAMPLE_RATE / 1000.0), len(buf)) predelay_line.write_block(buf) # Comb filters in parallel wet = np.zeros_like(buf, dtype=np.float64) for comb in combs: wet += comb.process(delayed) wet /= len(combs) # Normalise # Allpass filters in series for ap in allpasses: wet = ap.process(wet) wet = np.clip(wet, -1.0, 1.0).astype(np.float32) return buf * (1.0 - mix) + wet * mix # ── 12. Volume ────────────────────────────────────────────────── def _apply_volume(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Simple volume/level control.""" level = params.get("level", 1.0) return buf * level # ── 13. IR Cabinet Simulator ───────────────────────────────────── def _apply_ir_cab(self, buf: np.ndarray, params: dict, state: dict) -> np.ndarray: """Apply IR convolution via IRLoader.process(). Delegates to the IRLoader's FFT overlap-add engine. Supports wet/dry mix control per-preset. Params: - ir_file: str (path to .wav IR) — already set via load_ir() - enabled: bool - wet: float 0.0-1.0 - dry: float 0.0-1.0 """ # Update mix from preset params wet = params.get("wet", 1.0) dry = params.get("dry", 0.0) self.ir.set_mix(wet=wet, dry=dry) self.ir.enabled = params.get("enabled", True) and not params.get("bypass", False) return self.ir.process(buf) # ── 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 * 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 * 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 * 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 * SAMPLE_RATE / 1000.0 # 1ms base mod_range = depth * 3.0 * SAMPLE_RATE / 1000.0 if "delay" not in state: max_d = int(base_samples + mod_range + 5.0 * SAMPLE_RATE / 1000.0) + 1 state["delay"] = _DelayLine(max_d) state["delay"].write_block(np.zeros(max_d)) delay_line: _DelayLine = state["delay"] phase = self._lfo_phase(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 buf * (1.0 - mix) + wet * mix # ── 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, 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 buf * (1.0 - mix) + wet * mix # ── 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(-BLOCK_SIZE / (decay * 0.1 * 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, 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 buf * (1.0 - mix) + wet * mix # ── 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 * SAMPLE_RATE + 1)) state["horn_delay"] = _DelayLine(int(0.015 * SAMPLE_RATE + 1)) state["rotor_delay"].write_block(np.zeros(int(0.020 * SAMPLE_RATE))) state["horn_delay"].write_block(np.zeros(int(0.015 * 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 * SAMPLE_RATE) state["rotor_delay"].write_block(buf) horn_wet = state["horn_delay"].read_block_varying( horn_delay_s * 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 / 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 * SAMPLE_RATE + 1)) state["delay"].write_block(np.zeros(int(0.030 * SAMPLE_RATE))) delay_line: _DelayLine = state["delay"] # Read delayed signal for "side" channel side = delay_line.read_block(0.010 * 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(-BLOCK_SIZE / (attack_ms * SAMPLE_RATE / 1000.0)) else: alpha = np.exp(-BLOCK_SIZE / (release_ms * SAMPLE_RATE / 1000.0)) envelope = envelope * alpha + rms * (1.0 - alpha) state["envelope"] = envelope 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, 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(-BLOCK_SIZE / (attack_ms * SAMPLE_RATE / 1000.0)) else: alpha = np.exp(-BLOCK_SIZE / (release_ms * SAMPLE_RATE / 1000.0)) envelope = envelope * alpha + rms * (1.0 - alpha) state["envelope"] = envelope 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, 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, 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 + 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, 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 + 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, 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, 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, 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 buf * (1.0 - mix) + wet * mix # ── 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 * SAMPLE_RATE / 1000.0) if "delay" not in state: max_d = max(delay_samples * 2, SAMPLE_RATE) state["delay"] = _DelayLine(max_d + 1) state["delay"].write_block(np.zeros(max_d // 2)) 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) * SAMPLE_RATE / 1000.0) if "delay" not in state: max_d = max(max_delay * 2, 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 * 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(buf + wet * fb_gain) return buf * (1.0 - mix) + wet * mix # ── 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 * SAMPLE_RATE / 1000.0) if "delay" not in state: max_d = max(delay_samples * 2, SAMPLE_RATE) state["delay"] = _DelayLine(max_d + 1) state["delay"].write_block(np.zeros(max_d // 2)) 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(buf + out * fb_gain) return buf * (1.0 - mix) + out * mix # ── 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 * SAMPLE_RATE / 1000.0) if "delay" not in state: max_d = max(delay_samples * 2, SAMPLE_RATE) state["delay"] = _DelayLine(max_d + 1) state["delay"].write_block(np.zeros(max_d // 2)) delay_line: _DelayLine = state["delay"] # 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(buf + wet * fb_gain) return buf * (1.0 - mix) + wet * mix # ── 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 * SAMPLE_RATE / 1000.0)) for d in comb_delays] state["allpasses"] = [_AllpassFilter(int(d * SAMPLE_RATE / 1000.0)) for d in ap_delays] state["predelay"] = _DelayLine(int(30.0 * SAMPLE_RATE / 1000.0 + 1)) state["predelay"].write_block(np.zeros(int(30.0 * 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 * 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 * 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 buf * (1.0 - mix) + wet * mix # ── 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 * 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 * SAMPLE_RATE / 1000.0) for t in base_taps] if "delay" not in state: max_d = max(taps) + 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) # ── Properties ───────────────────────────────────────────────── @property def master_volume(self) -> float: return self._master_volume @master_volume.setter def master_volume(self, value: float) -> None: self._master_volume = max(0.0, min(1.0, value)) @property def bypassed(self) -> bool: return self._bypassed @bypassed.setter def bypassed(self, value: bool) -> None: self._bypassed = value logger.info("Global bypass: %s", "ON" if value else "OFF") @property def tuner_enabled(self) -> bool: return self._tuner_enabled @tuner_enabled.setter def tuner_enabled(self, value: bool) -> None: self._tuner_enabled = value # ── 4CM routing properties ──────────────────────────────────── @property def routing_mode(self) -> str: return self._routing_mode @routing_mode.setter def routing_mode(self, value: str) -> None: if value not in ("mono", "4cm"): raise ValueError(f"routing_mode must be 'mono' or '4cm', got {value!r}") self._routing_mode = value logger.info("Routing mode: %s", value) @property def routing_breakpoint(self) -> int: return self._routing_breakpoint @routing_breakpoint.setter def routing_breakpoint(self, value: int) -> None: self._routing_breakpoint = max(0, value) logger.info("Routing breakpoint: %d", self._routing_breakpoint) def set_routing(self, mode: str, breakpoint: int = 7) -> None: """Set 4CM routing configuration at runtime. Args: mode: ``\"mono\"`` or ``\"4cm\"``. breakpoint: Chain index where pre/post split occurs. """ self.routing_mode = mode self.routing_breakpoint = breakpoint