diff --git a/src/dsp/pipeline.py b/src/dsp/pipeline.py index f44c5b1..ba9bdff 100644 --- a/src/dsp/pipeline.py +++ b/src/dsp/pipeline.py @@ -15,6 +15,7 @@ same effect type at different positions in the chain. from __future__ import annotations import logging +import threading from dataclasses import dataclass, field from typing import Optional @@ -230,14 +231,14 @@ class _CombFilter: __slots__ = ("delay", "feedback", "damping", "damp_filt", "buf", "_buf_size") - def __init__(self, delay_samples: int): - line_len = max(BLOCK_SIZE * 2, delay_samples + 1) + def __init__(self, delay_samples: int, block_size: int = 256): + line_len = max(block_size * 2, delay_samples + 1) self.delay = _DelayLine(line_len) self.feedback: float = 0.5 self.damping: float = 0.5 # low-pass damping coefficient self.damp_filt: float = 0.0 # state variable for damping - self.buf = np.zeros(BLOCK_SIZE, dtype=np.float32) - self._buf_size = BLOCK_SIZE + self.buf = np.zeros(block_size, dtype=np.float32) + self._buf_size = block_size def process(self, block: np.ndarray) -> np.ndarray: # Resize internal buffer if block size changed (e.g. JACK period switch) @@ -263,12 +264,12 @@ class _AllpassFilter: __slots__ = ("delay", "gain", "buf", "_buf_size") - def __init__(self, delay_samples: int): - line_len = max(BLOCK_SIZE * 2, delay_samples + 1) + def __init__(self, delay_samples: int, block_size: int = 256): + line_len = max(block_size * 2, delay_samples + 1) self.delay = _DelayLine(line_len) self.gain: float = 0.5 - self.buf = np.zeros(BLOCK_SIZE, dtype=np.float32) - self._buf_size = BLOCK_SIZE + self.buf = np.zeros(block_size, dtype=np.float32) + self._buf_size = block_size def process(self, block: np.ndarray) -> np.ndarray: # Resize internal buffer if block size changed (e.g. JACK period switch) @@ -324,12 +325,16 @@ class AudioPipeline: # Cached filter coefficients per block self._coeffs: dict[str, tuple] = {} + # Runtime audio params (may be updated via set_audio_profile) + self._block_size: int = 256 + self._sample_rate: int = 48000 + # VU meter level tracking — updated on every process() call # Smoothed RMS levels (0.0–1.0) read by web server for live VU meters self._input_level: float = 0.0 self._output_level: float = 0.0 - # Smoothing factor: ~50ms time constant at 48kHz/256 block - self._vu_alpha: float = np.exp(-BLOCK_SIZE / (0.05 * SAMPLE_RATE)) + # Smoothing factor: ~50ms time constant — recomputed on profile change + self._vu_alpha: float = np.exp(-256 / (0.05 * 48000)) # ── Tuner / pitch detection state ──────────────────────────────── self._tuner_frequency: float = 0.0 # detected fundamental freq (Hz) @@ -341,14 +346,36 @@ class AudioPipeline: self._pitch_buffer: np.ndarray = np.array([], dtype=np.float32) self._pitch_buffer_max: int = 2048 # ~43ms at 48kHz + # Thread-safety lock — protects config state swapped by load_preset() + # and read by process(). process() snapshots under this lock briefly; + # load_preset() holds it only for the atomic swap (not during model I/O). + self._lock = threading.Lock() + logger.info("Audio pipeline initialized (block=%d, sr=%d)", - BLOCK_SIZE, SAMPLE_RATE) + self._block_size, self._sample_rate) + + @property + def sample_rate(self) -> int: + """Current sample rate in Hz.""" + return self._sample_rate + + @property + def block_size(self) -> int: + """Current audio block size (frames per callback).""" + return self._block_size def load_preset(self, preset: Preset) -> None: - """Load a complete preset (NAM, IR, and FX chain).""" - self._chain = [] - self._state = {} - self._coeffs = {} + """Load a complete preset (NAM, IR, and FX chain). + + Builds the new chain list off-thread, then swaps atomically under + ``_lock`` so that the audio thread's ``process()`` never sees a + half-constructed chain. + """ + new_chain: list[dict] = [] + new_routing_mode = preset.routing_mode + new_routing_breakpoint = preset.routing_breakpoint + new_master_volume = preset.master_volume + new_tuner_enabled = preset.tuner_enabled for block in preset.chain: entry = { @@ -359,7 +386,7 @@ class AudioPipeline: "subtype": block.subtype, } - # Load NAM model if needed + # Load NAM model if needed (may do I/O — don't hold pipeline lock) if block.fx_type == FXType.NAM_AMP and block.nam_model_path: self.nam.load_model(block.nam_model_path) @@ -367,14 +394,17 @@ class AudioPipeline: if block.fx_type == FXType.IR_CAB and block.ir_file_path: self.ir.load_ir(block.ir_file_path) - self._chain.append(entry) + new_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 + # Atomic swap under lock — keep the window as short as possible + with self._lock: + self._chain = new_chain + self._state = {} + self._coeffs = {} + self._master_volume = new_master_volume + self._tuner_enabled = new_tuner_enabled + self._routing_mode = new_routing_mode + self._routing_breakpoint = new_routing_breakpoint logger.info("Preset '%s' loaded: %d blocks, routing=%s breakpoint=%d", preset.name, len(self._chain), @@ -393,8 +423,21 @@ class AudioPipeline: Mono mode: shape (N,) — processed output. 4CM mode: shape (2, N) — [send_out, return_out]. """ + # ── Snapshot config under lock (brief — no I/O or heavy work) ── + with self._lock: + tuner_enabled = self._tuner_enabled + bypassed = self._bypassed + routing_mode = self._routing_mode + routing_breakpoint = self._routing_breakpoint + master_volume = self._master_volume + # Shallow-copy the chain list so iteration is isolated + chain = list(self._chain) + # Snapshot the state dict — process() mutates entries inside it + # but load_preset() replaces the whole dict under lock + state = self._state + # ── Tuner mode: mute output, keep input tracking for pitch detection ── - if self._tuner_enabled: + if tuner_enabled: # Still track input level for tuner display if audio_in.ndim == 1: in_rms = np.sqrt(np.mean(audio_in ** 2) + _EPS) @@ -413,13 +456,14 @@ class AudioPipeline: self._detect_pitch(audio_in) return np.zeros_like(audio_in) - if self._bypassed: - return audio_in * self._master_volume + if bypassed: + return audio_in * master_volume - if self._routing_mode == "4cm": - return self._process_4cm(audio_in) + if routing_mode == "4cm": + return self._process_4cm(audio_in, master_volume, chain, state, + routing_breakpoint) else: - return self._process_mono(audio_in) + return self._process_mono(audio_in, master_volume, chain, state) # ── Pitch detection for tuner ────────────────────────────────────────────── @@ -472,8 +516,8 @@ class AudioPipeline: # 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 + min_lag = int(self._sample_rate / 1600) # ~30 + max_lag = min(int(self._sample_rate / 50), len(buf) // 2) # ~960 if max_lag <= min_lag: self._tuner_confidence = 0.0 @@ -483,10 +527,7 @@ class AudioPipeline: # 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 + # Simple autocorrelation lag_slice = corr[min_lag:max_lag + 1] # Find the first peak in the autocorrelation @@ -517,7 +558,7 @@ class AudioPipeline: best_lag = best_lag + correction # Fundamental frequency - freq = SAMPLE_RATE / best_lag if best_lag > 0 else 0 + freq = self._sample_rate / best_lag if best_lag > 0 else 0 # Clip confidence to 0-1 range self._tuner_confidence = min(1.0, max(0.0, confidence / 10.0)) @@ -554,7 +595,10 @@ class AudioPipeline: self._tuner_string = si + 1 break - def _process_mono(self, audio_in: np.ndarray) -> np.ndarray: + def _process_mono(self, audio_in: np.ndarray, + master_volume: float, + chain: list[dict], + state: dict[str, dict]) -> np.ndarray: """Process a mono block through the full chain (all blocks).""" # Update input VU level (RMS with envelope smoothing) in_rms = np.sqrt(np.mean(audio_in ** 2) + _EPS) @@ -564,11 +608,11 @@ class AudioPipeline: ) buf = audio_in.copy() - for idx, entry in enumerate(self._chain): + for idx, entry in enumerate(chain): if entry["bypass"] or not entry["enabled"]: continue - buf = self._process_single_block(buf, idx, entry) - out = np.clip(buf * self._master_volume, -1.0, 1.0) + buf = self._process_single_block(buf, idx, entry, state) + out = np.clip(buf * master_volume, -1.0, 1.0) # Update output VU level out_rms = np.sqrt(np.mean(out ** 2) + _EPS) @@ -579,14 +623,18 @@ class AudioPipeline: return out - def _process_4cm(self, audio_in: np.ndarray) -> np.ndarray: + def _process_4cm(self, audio_in: np.ndarray, + master_volume: float, + chain: list[dict], + state: dict[str, dict], + routing_breakpoint: int) -> np.ndarray: """Process stereo block with 4CM split routing. audio_in has shape (2, N): ch0 = guitar input (Input 1) ch1 = FX loop return (Input 2) - Splits at _routing_breakpoint: + Splits at routing_breakpoint: pre blocks → ch0 processed through [0..breakpoint) post blocks → ch1 processed through [breakpoint..] @@ -604,22 +652,22 @@ class AudioPipeline: + in_rms * (1.0 - self._vu_alpha) ) - bp = self._routing_breakpoint + bp = routing_breakpoint - for idx, entry in enumerate(self._chain): + for idx, entry in enumerate(chain): if entry["bypass"] or not entry["enabled"]: continue if idx < bp: # Pre-amp block — process on guitar (ch0) - ch0 = self._process_single_block(ch0, idx, entry) + ch0 = self._process_single_block(ch0, idx, entry, state) else: # Post-amp block — process on return (ch1) - ch1 = self._process_single_block(ch1, idx, entry) + ch1 = self._process_single_block(ch1, idx, entry, state) out = np.zeros_like(audio_in) - out[0, :] = np.clip(ch0 * self._master_volume, -1.0, 1.0) - out[1, :] = np.clip(ch1 * self._master_volume, -1.0, 1.0) + out[0, :] = np.clip(ch0 * master_volume, -1.0, 1.0) + out[1, :] = np.clip(ch1 * master_volume, -1.0, 1.0) # Update output VU level from the processed effect return (ch1) out_rms = np.sqrt(np.mean(out ** 2) + _EPS) @@ -631,13 +679,15 @@ class AudioPipeline: return out def _process_single_block(self, buf: np.ndarray, idx: int, - entry: dict) -> np.ndarray: + entry: dict, + state: dict[str, dict]) -> np.ndarray: """Process a single mono audio block through one FX block. Args: buf: Mono audio block (N,) to process. idx: Chain index for state lookup. entry: Chain entry dict with fx_type, params. + state: Per-block DSP state dict (snapshotted by ``process()``). Returns: Processed mono block (N,). @@ -651,7 +701,7 @@ class AudioPipeline: subtype = entry.get("subtype", "") if subtype: params["subtype"] = subtype - fx_state = self._state.setdefault(f"fx_{idx}", {}) + fx_state = state.setdefault(f"fx_{idx}", {}) match fx_type: case FXType.NOISE_GATE: @@ -809,11 +859,10 @@ class AudioPipeline: # ── LFO helpers ───────────────────────────────────────────────── - @staticmethod - def _lfo_phase(rate_hz: float, state: dict, block_size: int) -> np.ndarray: + def _lfo_phase(self, rate_hz: float, state: dict, block_size: int) -> np.ndarray: """Generate LFO phase ramp (0->1), update state.""" phase = state.get("phase", 0.0) - delta = rate_hz / SAMPLE_RATE + delta = rate_hz / self._sample_rate t = np.arange(block_size, dtype=np.float64) * delta + phase t %= 1.0 state["phase"] = float(t[-1] + delta) % 1.0 @@ -851,7 +900,7 @@ class AudioPipeline: envelope = rms else: # Exponential release — time constant per block - release_coeff = np.exp(-BLOCK_SIZE / (release_ms * SAMPLE_RATE / 1000.0)) + release_coeff = np.exp(-self._block_size / (release_ms * self._sample_rate / 1000.0)) envelope = envelope * release_coeff + rms * (1.0 - release_coeff) state["envelope"] = envelope @@ -876,9 +925,9 @@ class AudioPipeline: envelope = state.get("envelope", 0.0) if rms > envelope: - alpha = np.exp(-BLOCK_SIZE / (attack_ms * SAMPLE_RATE / 1000.0)) + alpha = np.exp(-self._block_size / (attack_ms * self._sample_rate / 1000.0)) else: - alpha = np.exp(-BLOCK_SIZE / (release_ms * SAMPLE_RATE / 1000.0)) + alpha = np.exp(-self._block_size / (release_ms * self._sample_rate / 1000.0)) envelope = envelope * alpha + rms * (1.0 - alpha) state["envelope"] = envelope @@ -993,7 +1042,7 @@ class AudioPipeline: 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 + omega = 2.0 * np.pi * fc / self._sample_rate a0 = 1.0 + omega # one-pole approximation b0 = omega / a0 a1 = (1.0 - omega) / a0 @@ -1056,7 +1105,7 @@ class AudioPipeline: 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) + coeffs = _compute_highshelf_coeffs(3500.0, shelf_gain_db, 0.7, self._sample_rate) state["bd2_tshelf_coeffs"] = coeffs state["bd2_tshelf_tag"] = tag @@ -1131,7 +1180,7 @@ class AudioPipeline: 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) + coeffs = _compute_lowshelf_coeffs(200.0, bass_gain, 0.707, self._sample_rate) state["muff_bass_coeffs"] = coeffs state["muff_bass_tag"] = tag b0, b1, b2, a1, a2 = coeffs @@ -1146,7 +1195,7 @@ class AudioPipeline: 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) + coeffs = _compute_highshelf_coeffs(3000.0, treble_gain, 0.707, self._sample_rate) state["muff_treb_coeffs"] = coeffs state["muff_treb_tag"] = tag b0, b1, b2, a1, a2 = coeffs @@ -1189,7 +1238,7 @@ class AudioPipeline: 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) + coeffs = compute_fn(freq, gain_db, q, self._sample_rate) state[f"{key}_coeffs"] = coeffs state[f"{key}_tag"] = param_tag @@ -1213,11 +1262,11 @@ class AudioPipeline: 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 + base_samples = delay_base * self._sample_rate / 1000.0 + mod_range = depth * 5.0 * self._sample_rate / 1000.0 if "delay" not in state: - max_d = int(base_samples + mod_range + 10.0 * SAMPLE_RATE / 1000.0) + 1 + max_d = int(base_samples + mod_range + 10.0 * self._sample_rate / 1000.0) + 1 state["delay"] = _DelayLine(max_d) state["delay"].write_block(np.zeros(max_d)) @@ -1244,11 +1293,11 @@ class AudioPipeline: 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 + base_samples = delay_base * self._sample_rate / 1000.0 + mod_range = depth * 5.0 * self._sample_rate / 1000.0 if "delay" not in state: - max_d = int(base_samples + mod_range + 10.0 * SAMPLE_RATE / 1000.0) + 1 + max_d = int(base_samples + mod_range + 10.0 * self._sample_rate / 1000.0) + 1 state["delay"] = _DelayLine(max_d) state["delay"].write_block(np.zeros(max_d)) @@ -1299,7 +1348,7 @@ class AudioPipeline: # 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 + w = 2.0 * np.pi * freq / self._sample_rate tan_half_w = np.tan(w / 2.0) coeff = (1.0 - tan_half_w) / (1.0 + tan_half_w) @@ -1337,11 +1386,11 @@ class AudioPipeline: 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 + base_samples = 2.0 * self._sample_rate / 1000.0 # fixed ~2ms base + mod_range = depth * 3.0 * self._sample_rate / 1000.0 if "delay" not in state: - max_d = int(base_samples + mod_range + 5.0 * SAMPLE_RATE / 1000.0) + 1 + max_d = int(base_samples + mod_range + 5.0 * self._sample_rate / 1000.0) + 1 state["delay"] = _DelayLine(max_d) state["delay"].write_block(np.zeros(max_d)) @@ -1386,11 +1435,11 @@ class AudioPipeline: if tap_tempo > 0: time_ms = tap_tempo - delay_samples = int(time_ms * SAMPLE_RATE / 1000.0) + delay_samples = int(time_ms * self._sample_rate / 1000.0) if "delay" not in state: # Allocate 2x requested delay for headroom - max_d = max(delay_samples * 2, SAMPLE_RATE) # at least 1s + max_d = max(delay_samples * 2, self._sample_rate) # at least 1s state["delay"] = _DelayLine(max_d + 1) state["delay"].write_block(np.zeros(max_d // 2)) @@ -1426,10 +1475,10 @@ class AudioPipeline: 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) + delay_samples = int(time_ms * self._sample_rate / 1000.0) if "delay" not in state: - max_d = max(delay_samples * 2, SAMPLE_RATE) + max_d = max(delay_samples * 2, self._sample_rate) state["delay"] = _DelayLine(max_d + 1) state["delay"].write_block(np.zeros(max_d // 2)) @@ -1501,17 +1550,17 @@ class AudioPipeline: 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)) + _CombFilter(int(d * self._sample_rate / 1000.0), block_size=self._block_size) for d in comb_delays ] state["allpasses"] = [ - _AllpassFilter(int(d * SAMPLE_RATE / 1000.0)) + _AllpassFilter(int(d * self._sample_rate / 1000.0), block_size=self._block_size) for d in ap_delays ] state["predelay"] = _DelayLine( - int(predelay_ms * SAMPLE_RATE / 1000.0) + 1 + int(predelay_ms * self._sample_rate / 1000.0) + 1 ) - state["predelay"].write_block(np.zeros(BLOCK_SIZE)) + state["predelay"].write_block(np.zeros(self._block_size)) state["_computed"] = False combs: list[_CombFilter] = state["combs"] @@ -1530,7 +1579,7 @@ class AudioPipeline: state["_param_tag"] = param_tag delayed = predelay_line.read_block( - float(predelay_ms * SAMPLE_RATE / 1000.0), len(buf)) + float(predelay_ms * self._sample_rate / 1000.0), len(buf)) predelay_line.write_block(buf) wet = np.zeros_like(buf, dtype=np.float64) @@ -1568,8 +1617,8 @@ class AudioPipeline: 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)), + "delay": _DelayLine(int(d * self._sample_rate / 1000.0 + + self._block_size + 1)), "q": q, "prev": 0.0, "filt_prev": np.zeros(2, dtype=np.float64), @@ -1578,12 +1627,12 @@ class AudioPipeline: 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)) + int(predelay_ms * self._sample_rate / 1000.0) + 1) + state["predelay"].write_block(np.zeros(self._block_size)) predelay_line: _DelayLine = state["predelay"] delayed = predelay_line.read_block( - float(predelay_ms * SAMPLE_RATE / 1000.0), len(buf)) + float(predelay_ms * self._sample_rate / 1000.0), len(buf)) predelay_line.write_block(buf) # Feedback / damping scaling @@ -1597,12 +1646,12 @@ class AudioPipeline: q_val = spring["q"] # Read delayed signal - delay_samps = dl.max_len - BLOCK_SIZE - 1 + delay_samps = dl.max_len - self._block_size - 1 spring_out = dl.read_block(float(delay_samps), len(buf)) # Bandpass filter per spring — emphasises resonant frequency coeff = _compute_bpf_coeffs( - SAMPLE_RATE / (delay_samps + 1), q_val, SAMPLE_RATE) + self._sample_rate / (delay_samps + 1), q_val, self._sample_rate) b0, b1, b2, a1, a2 = coeff b_arr = np.array([b0, b1, b2], dtype=np.float64) a_arr = np.array([1.0, a1, a2], dtype=np.float64) @@ -1656,16 +1705,16 @@ class AudioPipeline: 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)) + _CombFilter(int(d * self._sample_rate / 1000.0), block_size=self._block_size) for d in comb_delays_ms ] state["allpasses"] = [ - _AllpassFilter(int(d * SAMPLE_RATE / 1000.0)) + _AllpassFilter(int(d * self._sample_rate / 1000.0), block_size=self._block_size) for d in ap_delays_ms ] state["predelay"] = _DelayLine( - int(predelay_ms * SAMPLE_RATE / 1000.0) + 1) - state["predelay"].write_block(np.zeros(BLOCK_SIZE)) + int(predelay_ms * self._sample_rate / 1000.0) + 1) + state["predelay"].write_block(np.zeros(self._block_size)) # Plate modulation LFO phase state["mod_phase"] = 0.0 @@ -1688,12 +1737,12 @@ class AudioPipeline: # Predelay delayed = predelay_line.read_block( - float(predelay_ms * SAMPLE_RATE / 1000.0), len(buf)) + float(predelay_ms * self._sample_rate / 1000.0), len(buf)) predelay_line.write_block(buf) # Plate modulation: very slow LFO (0.15 Hz) for subtle detuning mod_phase = state.get("mod_phase", 0.0) - delta = 0.15 / SAMPLE_RATE + delta = 0.15 / self._sample_rate t = np.arange(len(buf), dtype=np.float64) * delta + mod_phase t %= 1.0 state["mod_phase"] = float((t[-1] + delta) % 1.0) @@ -1748,28 +1797,29 @@ class AudioPipeline: 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) + * self._sample_rate / 1000.0) + state["er_delay"] = _DelayLine(max_tap + self._block_size + 1) state["er_taps"] = [ - int(t * size_factor * SAMPLE_RATE / 1000.0) + int(t * size_factor * self._sample_rate / 1000.0) for t in base_taps_ms ] state["tail_combs"] = [ - _CombFilter(int(d * size_factor * SAMPLE_RATE / 1000.0 + 1)) + _CombFilter(int(d * size_factor * self._sample_rate / 1000.0 + 1), block_size=self._block_size) for d in tail_comb_ms ] state["tail_allpasses"] = [ _AllpassFilter( - int(d * size_factor * SAMPLE_RATE / 1000.0 + 1)) + int(d * size_factor * self._sample_rate / 1000.0 + 1), + block_size=self._block_size) for d in tail_ap_ms ] state["predelay"] = _DelayLine( - int(predelay_ms * SAMPLE_RATE / 1000.0) + 1) - state["predelay"].write_block(np.zeros(BLOCK_SIZE)) + int(predelay_ms * self._sample_rate / 1000.0) + 1) + state["predelay"].write_block(np.zeros(self._block_size)) predelay_line: _DelayLine = state["predelay"] delayed = predelay_line.read_block( - float(predelay_ms * SAMPLE_RATE / 1000.0), len(buf)) + float(predelay_ms * self._sample_rate / 1000.0), len(buf)) predelay_line.write_block(buf) er_delay: _DelayLine = state["er_delay"] @@ -1900,7 +1950,7 @@ class AudioPipeline: 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 + grain_size = int(0.040 * self._sample_rate) # 1920 samples at 48kHz hop_in = grain_size hop_out = int(grain_size / factor) if factor > 0 else grain_size @@ -1953,7 +2003,7 @@ class AudioPipeline: return buf factor = 2.0 ** (shift_semitones / 12.0) - grain_size = int(0.040 * SAMPLE_RATE) + grain_size = int(0.040 * self._sample_rate) hop_out = int(grain_size / factor) if "ring" not in state: @@ -2000,7 +2050,7 @@ class AudioPipeline: shift_semitones = bend * 12.0 if "ring" not in state: - grain_size = int(0.040 * SAMPLE_RATE) + grain_size = int(0.040 * self._sample_rate) state["ring"] = np.zeros(grain_size * 2, dtype=np.float32) state["wpos"] = 0 state["rpos"] = 0 @@ -2042,11 +2092,11 @@ class AudioPipeline: 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 + base_samples = 1.0 * self._sample_rate / 1000.0 # 1ms base + mod_range = depth * 3.0 * self._sample_rate / 1000.0 if "delay" not in state: - max_d = int(base_samples + mod_range + 5.0 * SAMPLE_RATE / 1000.0) + 1 + max_d = int(base_samples + mod_range + 5.0 * self._sample_rate / 1000.0) + 1 state["delay"] = _DelayLine(max_d) state["delay"].write_block(np.zeros(max_d)) @@ -2097,7 +2147,7 @@ class AudioPipeline: cutoff = np.clip(cutoff, 200.0, 5000.0) # BPF biquad - coeffs = _compute_bpf_coeffs(cutoff, q, SAMPLE_RATE) + coeffs = _compute_bpf_coeffs(cutoff, q, self._sample_rate) b0, b1, b2, a1, a2 = coeffs b = np.array([b0, b1, b2], dtype=np.float64) a = np.array([1.0, a1, a2], dtype=np.float64) @@ -2125,13 +2175,13 @@ class AudioPipeline: if rms > env: env = rms # instant attack else: - env = env * np.exp(-BLOCK_SIZE / (decay * 0.1 * SAMPLE_RATE)) + env = env * np.exp(-self._block_size / (decay * 0.1 * self._sample_rate)) state["envelope"] = env cutoff = 200.0 + env * freq_range * 3800.0 cutoff = np.clip(cutoff, 100.0, 4000.0) - coeffs = _compute_lpf_coeffs(cutoff, 0.707, SAMPLE_RATE) + coeffs = _compute_lpf_coeffs(cutoff, 0.707, self._sample_rate) b0, b1, b2, a1, a2 = coeffs b = np.array([b0, b1, b2], dtype=np.float64) a = np.array([1.0, a1, a2], dtype=np.float64) @@ -2160,10 +2210,10 @@ class AudioPipeline: # 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))) + state["rotor_delay"] = _DelayLine(int(0.020 * self._sample_rate + 1)) + state["horn_delay"] = _DelayLine(int(0.015 * self._sample_rate + 1)) + state["rotor_delay"].write_block(np.zeros(int(0.020 * self._sample_rate))) + state["horn_delay"].write_block(np.zeros(int(0.015 * self._sample_rate))) rotor_phase = self._lfo_phase(rotor_rate, state, len(buf)) horn_phase = self._lfo_phase(horn_rate, state, len(buf)) @@ -2177,11 +2227,11 @@ class AudioPipeline: horn_delay_s = 0.001 + horn_lfo * 0.002 # 1-3ms rotor_wet = state["rotor_delay"].read_block_varying( - rotor_delay_s * SAMPLE_RATE) + rotor_delay_s * self._sample_rate) state["rotor_delay"].write_block(buf) horn_wet = state["horn_delay"].read_block_varying( - horn_delay_s * SAMPLE_RATE) + horn_delay_s * self._sample_rate) state["horn_delay"].write_block(buf * 0.7) # Mix with subtle drive (soft clip) @@ -2219,7 +2269,7 @@ class AudioPipeline: 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 + w = 2.0 * np.pi * freq_mean / self._sample_rate tan_half_w = np.tan(w / 2.0) coeff = (1.0 - tan_half_w) / (1.0 + tan_half_w) @@ -2270,12 +2320,12 @@ class AudioPipeline: # 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))) + state["delay"] = _DelayLine(int(0.030 * self._sample_rate + 1)) + state["delay"].write_block(np.zeros(int(0.030 * self._sample_rate))) delay_line: _DelayLine = state["delay"] # Read delayed signal for "side" channel - side = delay_line.read_block(0.010 * SAMPLE_RATE, len(buf)) + side = delay_line.read_block(0.010 * self._sample_rate, len(buf)) delay_line.write_block(buf) # Mid = original, Side = delayed difference @@ -2371,9 +2421,9 @@ class AudioPipeline: envelope = state.get("envelope", 0.0) if rms > envelope: - alpha = np.exp(-BLOCK_SIZE / (attack_ms * SAMPLE_RATE / 1000.0)) + alpha = np.exp(-self._block_size / (attack_ms * self._sample_rate / 1000.0)) else: - alpha = np.exp(-BLOCK_SIZE / (release_ms * SAMPLE_RATE / 1000.0)) + alpha = np.exp(-self._block_size / (release_ms * self._sample_rate / 1000.0)) envelope = envelope * alpha + rms * (1.0 - alpha) state["envelope"] = envelope @@ -2405,7 +2455,7 @@ class AudioPipeline: # Bandpass the sibilance band q = 3.0 # narrow Q for sibilance band - coeffs = _compute_bpf_coeffs(freq, q, SAMPLE_RATE) + coeffs = _compute_bpf_coeffs(freq, q, self._sample_rate) b0, b1, b2, a1, a2 = coeffs b = np.array([b0, b1, b2], dtype=np.float64) a = np.array([1.0, a1, a2], dtype=np.float64) @@ -2489,9 +2539,9 @@ class AudioPipeline: envelope = state.get("envelope", 0.0) if rms > envelope: - alpha = np.exp(-BLOCK_SIZE / (attack_ms * SAMPLE_RATE / 1000.0)) + alpha = np.exp(-self._block_size / (attack_ms * self._sample_rate / 1000.0)) else: - alpha = np.exp(-BLOCK_SIZE / (release_ms * SAMPLE_RATE / 1000.0)) + alpha = np.exp(-self._block_size / (release_ms * self._sample_rate / 1000.0)) envelope = envelope * alpha + rms * (1.0 - alpha) state["envelope"] = envelope @@ -2527,7 +2577,7 @@ class AudioPipeline: if freq == 0.0 or gain_db == 0.0: continue - coeffs = _compute_peaking_coeffs(freq, gain_db, q, SAMPLE_RATE) + coeffs = _compute_peaking_coeffs(freq, gain_db, q, self._sample_rate) b0, b1, b2, a1, a2 = coeffs b = np.array([b0, b1, b2], dtype=np.float64) a = np.array([1.0, a1, a2], dtype=np.float64) @@ -2546,7 +2596,7 @@ class AudioPipeline: 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) + coeffs = _compute_hpf_coeffs(freq, q, self._sample_rate) b0, b1, b2, a1, a2 = coeffs b = np.array([b0, b1, b2], dtype=np.float64) a = np.array([1.0, a1, a2], dtype=np.float64) @@ -2557,7 +2607,7 @@ class AudioPipeline: # 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)) + alpha = 1.0 / (1.0 + self._sample_rate / (2.0 * np.pi * freq)) zi1 = state.get("hpf_zi_1", np.float64(0.0)) out = np.zeros(len(buf), dtype=np.float64) for i in range(len(buf)): @@ -2576,7 +2626,7 @@ class AudioPipeline: slope = params.get("slope", 12.0) q = 0.707 if slope >= 12 else 0.5 - coeffs = _compute_lpf_coeffs(freq, q, SAMPLE_RATE) + coeffs = _compute_lpf_coeffs(freq, q, self._sample_rate) b0, b1, b2, a1, a2 = coeffs b = np.array([b0, b1, b2], dtype=np.float64) a = np.array([1.0, a1, a2], dtype=np.float64) @@ -2585,7 +2635,7 @@ class AudioPipeline: state["lpf_zi"] = zf if slope < 12: - alpha = 1.0 / (1.0 + SAMPLE_RATE / (2.0 * np.pi * freq)) + alpha = 1.0 / (1.0 + self._sample_rate / (2.0 * np.pi * freq)) zi1 = state.get("lpf_zi_1", np.float64(0.0)) out = np.zeros(len(buf), dtype=np.float64) for i in range(len(buf)): @@ -2601,7 +2651,7 @@ class AudioPipeline: """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) + coeffs = _compute_bpf_coeffs(freq, q, self._sample_rate) b0, b1, b2, a1, a2 = coeffs b = np.array([b0, b1, b2], dtype=np.float64) a = np.array([1.0, a1, a2], dtype=np.float64) @@ -2615,7 +2665,7 @@ class AudioPipeline: """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) + coeffs = _compute_notch_coeffs(freq, q, self._sample_rate) b0, b1, b2, a1, a2 = coeffs b = np.array([b0, b1, b2], dtype=np.float64) a = np.array([1.0, a1, a2], dtype=np.float64) @@ -2645,7 +2695,7 @@ class AudioPipeline: 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) + coeffs = _compute_peaking_coeffs(freq, 12.0, q_val, self._sample_rate) b0, b1, b2, a1, a2 = coeffs b = np.array([b0, b1, b2], dtype=np.float64) a = np.array([1.0, a1, a2], dtype=np.float64) @@ -2667,9 +2717,9 @@ class AudioPipeline: feedback = params.get("feedback", 0.3) mix = params.get("mix", 0.4) - delay_samples = int(time_ms * SAMPLE_RATE / 1000.0) + delay_samples = int(time_ms * self._sample_rate / 1000.0) if "delay" not in state: - max_d = max(delay_samples * 2, SAMPLE_RATE) + max_d = max(delay_samples * 2, self._sample_rate) state["delay"] = _DelayLine(max_d + 1) state["delay"].write_block(np.zeros(max_d // 2)) state["ping"] = 1 # 1 = left, -1 = right @@ -2706,9 +2756,9 @@ class AudioPipeline: } taps = tap_patterns.get(pattern, tap_patterns["quarter"]) - max_delay = int(max(taps) * SAMPLE_RATE / 1000.0) + max_delay = int(max(taps) * self._sample_rate / 1000.0) if "delay" not in state: - max_d = max(max_delay * 2, SAMPLE_RATE) + max_d = max(max_delay * 2, self._sample_rate) state["delay"] = _DelayLine(max_d + 1) state["delay"].write_block(np.zeros(max_d // 2)) @@ -2717,7 +2767,7 @@ class AudioPipeline: # 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_samples = int(tap_ms * self._sample_rate / 1000.0) tap_sig = delay_line.read_block(float(tap_samples), len(buf)) wet += tap_sig * (1.0 / len(taps)) # Normalise @@ -2734,9 +2784,9 @@ class AudioPipeline: feedback = params.get("feedback", 0.3) mix = params.get("mix", 0.4) - delay_samples = int(time_ms * SAMPLE_RATE / 1000.0) + delay_samples = int(time_ms * self._sample_rate / 1000.0) if "delay" not in state: - max_d = max(delay_samples * 2, SAMPLE_RATE) + max_d = max(delay_samples * 2, self._sample_rate) state["delay"] = _DelayLine(max_d + 1) state["delay"].write_block(np.zeros(max_d // 2)) state["read_pos"] = 0 @@ -2773,9 +2823,9 @@ class AudioPipeline: feedback = params.get("feedback", 0.3) mix = params.get("mix", 0.4) - delay_samples = int(time_ms * SAMPLE_RATE / 1000.0) + delay_samples = int(time_ms * self._sample_rate / 1000.0) if "delay" not in state: - max_d = max(delay_samples * 2, SAMPLE_RATE) + max_d = max(delay_samples * 2, self._sample_rate) state["delay"] = _DelayLine(max_d + 1) state["delay"].write_block(np.zeros(max_d // 2)) @@ -2821,10 +2871,10 @@ class AudioPipeline: 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))) + state["combs"] = [_CombFilter(int(d * self._sample_rate / 1000.0), block_size=self._block_size) for d in comb_delays] + state["allpasses"] = [_AllpassFilter(int(d * self._sample_rate / 1000.0), block_size=self._block_size) for d in ap_delays] + state["predelay"] = _DelayLine(int(30.0 * self._sample_rate / 1000.0 + 1)) + state["predelay"].write_block(np.zeros(int(30.0 * self._sample_rate / 1000.0))) combs: list[_CombFilter] = state["combs"] allpasses: list[_AllpassFilter] = state["allpasses"] @@ -2839,7 +2889,7 @@ class AudioPipeline: ap.gain = 0.3 + decay * 0.3 # Predelay - delayed = predelay_line.read_block(30.0 * SAMPLE_RATE / 1000.0, len(buf)) + delayed = predelay_line.read_block(30.0 * self._sample_rate / 1000.0, len(buf)) predelay_line.write_block(buf) # Comb + allpass reverb @@ -2854,7 +2904,7 @@ class AudioPipeline: if shift != 0.0: factor = 2.0 ** (shift / 12.0) if "shift_ring" not in state: - grain_size = int(0.040 * SAMPLE_RATE) + grain_size = int(0.040 * self._sample_rate) state["shift_ring"] = np.zeros(grain_size * 2, dtype=np.float32) state["swpos"] = 0 state["srpos"] = 0 @@ -2892,7 +2942,7 @@ class AudioPipeline: Controls: record, overdub, play, stop. State machine: idle -> recording -> playing -> overdub -> idle """ - max_buffer = int(30.0 * SAMPLE_RATE) # 30 seconds + max_buffer = int(30.0 * self._sample_rate) # 30 seconds if "looper_buf" not in state: state["looper_buf"] = np.zeros(max_buffer, dtype=np.float32) @@ -2970,10 +3020,10 @@ class AudioPipeline: # 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] + taps = [int(t * size_factor * self._sample_rate / 1000.0) for t in base_taps] if "delay" not in state: - max_d = max(taps) + BLOCK_SIZE + 1 + max_d = max(taps) + self._block_size + 1 state["delay"] = _DelayLine(max_d) delay_line: _DelayLine = state["delay"] @@ -2992,6 +3042,44 @@ class AudioPipeline: return (buf * (1.0 - mix) + wet * mix).astype(np.float32) + def set_audio_profile(self, block_size: int, sample_rate: int) -> None: + """Update block size and sample rate at runtime. + + Recomputes profile-dependent constants and clears DSP state so + effects reinitialise with the new buffer sizes on the next + process() call. + + Called by the web server when the JACK latency profile changes + (``POST /api/audio/profile``). Follows the same pattern as + :meth:`NAMEngineRouter.set_block_size`. + + Args: + block_size: JACK period size in frames (e.g. 64, 128, 256, 512). + sample_rate: Sample rate in Hz (e.g. 44100, 48000, 96000). + """ + changed = (self._block_size != block_size + or self._sample_rate != sample_rate) + if not changed: + return + self._block_size = block_size + self._sample_rate = sample_rate + self._vu_alpha = np.exp(-block_size / (0.05 * sample_rate)) + # Clear DSP state — effects will reinit with new block/sample rate + self._state.clear() + self._coeffs.clear() + logger.info("Audio profile updated: block=%d, sr=%d", + block_size, sample_rate) + + @property + def block_size(self) -> int: + """Current JACK period / block size in frames.""" + return self._block_size + + @property + def sample_rate(self) -> int: + """Current audio sample rate in Hz.""" + return self._sample_rate + # ── Properties ───────────────────────────────────────────────── @property @@ -3041,6 +3129,13 @@ class AudioPipeline: self._routing_breakpoint = max(0, value) logger.info("Routing breakpoint: %d", self._routing_breakpoint) + # ── Chain access (read-only for external inspection) ──────────── + + @property + def chain(self) -> list[dict]: + """Read-only access to the DSP chain.""" + return list(self._chain) + def set_routing(self, mode: str, breakpoint: int = 7) -> None: """Set 4CM routing configuration at runtime.