diff --git a/src/dsp/pipeline.py b/src/dsp/pipeline.py index 88a342f..b5499e5 100644 --- a/src/dsp/pipeline.py +++ b/src/dsp/pipeline.py @@ -242,6 +242,10 @@ class AudioPipeline: 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] = {} @@ -278,72 +282,139 @@ class AudioPipeline: self._master_volume = preset.master_volume self._tuner_enabled = preset.tuner_enabled - logger.info("Preset '%s' loaded: %d blocks", preset.name, len(self._chain)) + # Set 4CM routing from preset + self._routing_mode = preset.routing_mode + self._routing_breakpoint = preset.routing_breakpoint + + logger.info("Preset '%s' loaded: %d blocks, routing=%s breakpoint=%d", + preset.name, len(self._chain), + self._routing_mode, self._routing_breakpoint) def process(self, audio_in: np.ndarray) -> np.ndarray: """Process a block of audio through the entire FX chain. Args: audio_in: numpy array of PCM samples (float32 [-1, 1]). + Mono mode: shape (N,) — single audio channel. + 4CM mode: shape (2, N) — two channels, [guitar_in, return_in]. Returns: Processed audio block. + Mono mode: shape (N,) — processed output. + 4CM mode: shape (2, N) — [send_out, return_out]. """ if self._bypassed: return audio_in * self._master_volume + if self._routing_mode == "4cm": + return self._process_4cm(audio_in) + else: + return self._process_mono(audio_in) + + def _process_mono(self, audio_in: np.ndarray) -> np.ndarray: + """Process a mono block through the full chain (all blocks).""" buf = audio_in.copy() + for idx, entry in enumerate(self._chain): + if entry["bypass"] or not entry["enabled"]: + continue + buf = self._process_single_block(buf, idx, entry) + return buf * self._master_volume + + def _process_4cm(self, audio_in: np.ndarray) -> np.ndarray: + """Process stereo block with 4CM split routing. + + audio_in has shape (2, N): + ch0 = guitar input (Input 1) + ch1 = FX loop return (Input 2) + + Splits at _routing_breakpoint: + pre blocks → ch0 processed through [0..breakpoint) + post blocks → ch1 processed through [breakpoint..] + + Returns (2, N): + ch0 = Send output (to amp input) + ch1 = Return output (to amp FX return) + """ + ch0 = audio_in[0, :].copy() + ch1 = audio_in[1, :].copy() + bp = self._routing_breakpoint for idx, entry in enumerate(self._chain): if entry["bypass"] or not entry["enabled"]: continue - fx_type = entry["fx_type"] - params = entry["params"] - fx_state = self._state.setdefault(f"fx_{idx}", {}) + 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) - match fx_type: - case FXType.NOISE_GATE: - buf = self._apply_gate(buf, params, fx_state) - case FXType.COMPRESSOR: - buf = self._apply_compressor(buf, params, fx_state) - case FXType.BOOST: - buf = self._apply_boost(buf, params, fx_state) - case FXType.OVERDRIVE: - buf = self._apply_overdrive(buf, params, fx_state) - case FXType.DISTORTION: - buf = self._apply_distortion(buf, params, fx_state) - case FXType.FUZZ: - buf = self._apply_fuzz(buf, params, fx_state) - case FXType.EQ: - buf = self._apply_eq(buf, params, fx_state) - case FXType.CHORUS: - buf = self._apply_chorus(buf, params, fx_state) - case FXType.FLANGER: - buf = self._apply_flanger(buf, params, fx_state) - case FXType.PHASER: - buf = self._apply_phaser(buf, params, fx_state) - case FXType.TREMOLO: - buf = self._apply_tremolo(buf, params, fx_state) - case FXType.VIBRATO: - buf = self._apply_vibrato(buf, params, fx_state) - case FXType.DELAY: - buf = self._apply_delay(buf, params, fx_state) - case FXType.REVERB: - buf = self._apply_reverb(buf, params, fx_state) - case FXType.VOLUME: - buf = self._apply_volume(buf, params, fx_state) - case FXType.NAM_AMP: - if self.nam.is_loaded: - buf = self.nam.process(buf) - # Apply crossfade if a model switch is in progress - if self.nam._crossfade_buf is not None: - buf = self.nam.apply_crossfade(buf) - case FXType.IR_CAB: - if self.ir.is_loaded: - buf = self._apply_ir_cab(buf, params, fx_state) + out = np.zeros_like(audio_in) + out[0, :] = ch0 * self._master_volume + out[1, :] = ch1 * self._master_volume + return out - return buf * self._master_volume + def _process_single_block(self, buf: np.ndarray, idx: int, + entry: dict) -> np.ndarray: + """Process a single mono audio block through one FX block. + + Args: + buf: Mono audio block (N,) to process. + idx: Chain index for state lookup. + entry: Chain entry dict with fx_type, params. + + Returns: + Processed mono block (N,). + """ + fx_type = entry["fx_type"] + params = entry["params"] + fx_state = self._state.setdefault(f"fx_{idx}", {}) + + match fx_type: + case FXType.NOISE_GATE: + return self._apply_gate(buf, params, fx_state) + case FXType.COMPRESSOR: + return self._apply_compressor(buf, params, fx_state) + case FXType.BOOST: + return self._apply_boost(buf, params, fx_state) + case FXType.OVERDRIVE: + return self._apply_overdrive(buf, params, fx_state) + case FXType.DISTORTION: + return self._apply_distortion(buf, params, fx_state) + case FXType.FUZZ: + return self._apply_fuzz(buf, params, fx_state) + case FXType.EQ: + return self._apply_eq(buf, params, fx_state) + case FXType.CHORUS: + return self._apply_chorus(buf, params, fx_state) + case FXType.FLANGER: + return self._apply_flanger(buf, params, fx_state) + case FXType.PHASER: + return self._apply_phaser(buf, params, fx_state) + case FXType.TREMOLO: + return self._apply_tremolo(buf, params, fx_state) + case FXType.VIBRATO: + return self._apply_vibrato(buf, params, fx_state) + case FXType.DELAY: + return self._apply_delay(buf, params, fx_state) + case FXType.REVERB: + return self._apply_reverb(buf, params, fx_state) + case FXType.VOLUME: + return self._apply_volume(buf, params, fx_state) + case FXType.NAM_AMP: + if self.nam.is_loaded: + processed = self.nam.process(buf) + if self.nam._crossfade_buf is not None: + processed = self.nam.apply_crossfade(processed) + return processed + return buf + case FXType.IR_CAB: + if self.ir.is_loaded: + return self._apply_ir_cab(buf, params, fx_state) + return buf + case _: + return buf # ── LFO helpers ───────────────────────────────────────────────── @@ -846,4 +917,36 @@ class AudioPipeline: @tuner_enabled.setter def tuner_enabled(self, value: bool) -> None: - self._tuner_enabled = value \ No newline at end of file + 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 \ No newline at end of file diff --git a/src/web/server.py b/src/web/server.py index 617737e..48361e5 100644 --- a/src/web/server.py +++ b/src/web/server.py @@ -356,6 +356,53 @@ class WebServer: pl._master_volume = vol return {"ok": True, "volume": vol} + # ── 4CM routing endpoints ──────────────────────────────── + + @app.get("/api/routing") + async def get_routing(): + """Get current routing mode and breakpoint.""" + pl = self.deps.pipeline + if not pl: + raise HTTPException(status_code=503) + return { + "routing_mode": pl.routing_mode, + "routing_breakpoint": pl.routing_breakpoint, + } + + @app.post("/api/routing") + async def set_routing(data: dict): + """Set routing mode and/or breakpoint. + + Body: + routing_mode (optional): ``\"mono\"`` or ``\"4cm\"``. + routing_breakpoint (optional): chain index for split. + """ + pl = self.deps.pipeline + pm = self.deps.presets + if not pl: + raise HTTPException(status_code=503) + mode = data.get("routing_mode", pl.routing_mode) + bp = int(data.get("routing_breakpoint", pl.routing_breakpoint)) + pl.set_routing(mode, bp) + + # Persist to current preset if preset manager is available + if pm: + try: + bank = pm.current_bank + program = pm.current_program + preset = pm.load(bank, program) + pm.set_4cm_config(preset, mode=mode, breakpoint=bp) + except Exception: + pass + + await self._manager.broadcast({ + "type": "routing_changed", + "routing_mode": pl.routing_mode, + "routing_breakpoint": pl.routing_breakpoint, + }) + return {"ok": True, "routing_mode": pl.routing_mode, + "routing_breakpoint": pl.routing_breakpoint} + @app.get("/api/models") async def list_models(): """List available NAM models.""" @@ -578,6 +625,8 @@ class WebServer: "bypass": pl._bypassed if pl else False, "tuner_enabled": pl._tuner_enabled if pl else False, "master_volume": pl._master_volume if pl else 0.8, + "routing_mode": pl.routing_mode if pl else "mono", + "routing_breakpoint": pl.routing_breakpoint if pl else 7, "nam_loaded": bool(nam and nam.is_loaded) if nam else False, "nam_model": current_model_name, "ir_loaded": bool(ir and ir.is_loaded) if ir else False, diff --git a/src/web/static/app.js b/src/web/static/app.js index eb6da0a..e206a73 100644 --- a/src/web/static/app.js +++ b/src/web/static/app.js @@ -80,6 +80,74 @@ pedalWS.on('tuner_changed', (msg) => { } }); +pedalWS.on('routing_changed', (msg) => { + updateRoutingUI(msg.routing_mode, msg.routing_breakpoint); +}); + +/* ── 4CM Routing UI ────────────────────────────────────────────── */ + +function updateRoutingUI(mode, breakpoint) { + // Dashboard badge + const indicator = document.getElementById('routing-mode-indicator'); + const info = document.getElementById('routing-info'); + if (indicator) { + indicator.setAttribute('data-mode', mode); + if (mode === '4cm') { + indicator.innerHTML = '4CM'; + } else { + indicator.innerHTML = 'Mono'; + } + } + if (info) { + info.textContent = mode === '4cm' ? `Breakpoint: ${breakpoint}` : 'Full chain'; + } + + // Settings page controls + const toggleBtn = document.getElementById('4cm-toggle-btn'); + if (toggleBtn) { + toggleBtn.textContent = mode === '4cm' ? 'ON' : 'OFF'; + toggleBtn.classList.toggle('active', mode === '4cm'); + } + const bpRow = document.getElementById('breakpoint-row'); + if (bpRow) { + bpRow.style.opacity = mode === '4cm' ? '1' : '0.4'; + bpRow.style.pointerEvents = mode === '4cm' ? 'auto' : 'none'; + } + const bpSlider = document.getElementById('4cm-breakpoint'); + const bpText = document.getElementById('4cm-breakpoint-text'); + if (bpSlider) bpSlider.value = breakpoint; + if (bpText) bpText.textContent = breakpoint; + + const desc = document.getElementById('4cm-routing-desc'); + if (desc) { + if (mode === '4cm') { + desc.innerHTML = `Input 1 (Guitar) → Pre blocks [0..${breakpoint}) → Send  |  Input 2 (Return) → Post blocks [${breakpoint}..] → Output`; + } else { + desc.textContent = 'Mono: Guitar → Full chain → Output'; + } + } +} + +async function toggle4cm() { + const btn = document.getElementById('4cm-toggle-btn'); + const currentMode = btn && btn.classList.contains('active') ? '4cm' : 'mono'; + const newMode = currentMode === '4cm' ? 'mono' : '4cm'; + try { + await apiPost('/routing', { routing_mode: newMode }); + } catch (e) { + console.warn('Failed to toggle 4CM:', e); + } +} + +async function set4cmBreakpoint(val) { + const bp = parseInt(val); + try { + await apiPost('/routing', { routing_breakpoint: bp }); + } catch (e) { + console.warn('Failed to set breakpoint:', e); + } +} + function updateDashboardState(state) { if (!state || !state.connected) return; @@ -128,6 +196,11 @@ function updateDashboardState(state) { // IR const irNameEl = document.getElementById('ir-name'); if (irNameEl) irNameEl.textContent = state.ir_name || 'None loaded'; + + // 4CM Routing + if (state.routing_mode) { + updateRoutingUI(state.routing_mode, state.routing_breakpoint); + } } /* ── Dashboard action handlers ─────────────────────────────────── */ diff --git a/src/web/static/style.css b/src/web/static/style.css index 6985a1a..2e84156 100644 --- a/src/web/static/style.css +++ b/src/web/static/style.css @@ -651,4 +651,47 @@ body { .mb-8 { margin-bottom: 8px; } .flex { display: flex; } .flex-1 { flex: 1; } -.gap-8 { gap: 8px; } \ No newline at end of file +.gap-8 { gap: 8px; } + +/* ── Routing / 4CM ────────────────────────────────────────────────── */ + +.routing-status { + display: flex; + align-items: center; + gap: 10px; +} + +.routing-indicator { + display: inline-flex; +} + +.badge { + display: inline-block; + padding: 3px 10px; + border-radius: 10px; + font-size: 0.75rem; + font-weight: 700; + letter-spacing: 0.5px; + text-transform: uppercase; +} + +.badge-4cm { + background: var(--warning); + color: #1a1a1a; +} + +.badge-mono { + background: var(--text-muted); + color: var(--text-primary); +} + +.routing-info { + font-size: 0.82rem; + color: var(--text-secondary); +} + +.routing-desc { + font-size: 0.78rem; + color: var(--text-secondary); + line-height: 1.4; +} \ No newline at end of file diff --git a/src/web/templates/dashboard.html b/src/web/templates/dashboard.html index 53984c5..d44b47b 100644 --- a/src/web/templates/dashboard.html +++ b/src/web/templates/dashboard.html @@ -85,5 +85,30 @@ + + +
+
+

Routing

+
+
+
+ + {% if routing_mode == '4cm' %} + 4CM + {% else %} + Mono + {% endif %} + + + {% if routing_mode == '4cm' %} + Breakpoint: {{ routing_breakpoint }} + {% else %} + Full chain + {% endif %} + +
+
+
{% endblock %} \ No newline at end of file diff --git a/src/web/templates/settings.html b/src/web/templates/settings.html index 5859406..e20cc5e 100644 --- a/src/web/templates/settings.html +++ b/src/web/templates/settings.html @@ -52,6 +52,46 @@ +
+
+

4-Cable Method (4CM)

+
+
+
+
4CM Mode
+
+ +
+
+
+
Breakpoint
+
+ + {{ routing_breakpoint }} +
+
+
+
Routing
+
+ + {% if routing_mode == '4cm' %} + Input 1 (Guitar) → Pre blocks [0..{{ routing_breakpoint }}) → Send +  |  Input 2 (Return) → Post blocks [{{ routing_breakpoint }}..] → Output + {% else %} + Mono: Guitar → Full chain → Output + {% endif %} + +
+
+
+
+

About

diff --git a/tests/test_web.py b/tests/test_web.py index 7505984..79b39c8 100644 --- a/tests/test_web.py +++ b/tests/test_web.py @@ -42,6 +42,8 @@ def mock_pipeline(): pl._bypassed = False pl._tuner_enabled = False pl._master_volume = 0.8 + pl.routing_mode = "mono" + pl.routing_breakpoint = 7 return pl