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