feat: separate preset banks per channel (guitar/bass)

- Channel enum (GUITAR, BASS) with channel field on Preset dataclass
- Channel-prefixed storage: {root}/{channel}/bank_{bank}/preset_{program}.json
- PresetManager channel awareness: set_channel(), current_channel property
- Per-channel state persistence (channel_state.json per channel dir)
- Legacy migration: flat bank_* dirs auto-migrate to guitar/ on first boot
- All CRUD methods accept optional channel parameter
- API endpoints accept channel param on GET/PUT/DELETE /api/presets
- /api/channel GET/POST for channel switching
- 10 new channel tests (independence, switching, migration, scoping)
- Factory presets install to specified channel
This commit is contained in:
2026-06-12 23:22:47 -04:00
parent 0a21297f77
commit bfe1434f75
10 changed files with 1100 additions and 352 deletions
+87 -8
View File
@@ -20,15 +20,31 @@ const T = {
const API = window.location.origin;
const WS_URL = API.replace(/^http/, 'ws') + '/ws';
// ── Channel state (module-level, synced from localStorage) ─────
let _channel = 'guitar';
try {
const saved = localStorage.getItem('pedal_channel');
if (saved === 'guitar' || saved === 'bass') _channel = saved;
} catch {}
export function getChannel() { return _channel; }
export function setChannel(ch) { _channel = ch; }
function _ch(url) {
const c = _channel;
if (!c) return url;
return url + (url.includes('?') ? '&' : '?') + 'channel=' + encodeURIComponent(c);
}
// ── API helpers ────────────────────────────────────────────────
async function apiGet(path) {
const r = await fetch(`${API}${path}`);
const r = await fetch(`${API}${_ch(path)}`);
if (!r.ok) throw new Error(`${path} => ${r.status}`);
return r.json();
}
async function apiPost(path, data) {
const r = await fetch(`${API}${path}`, {
const r = await fetch(`${API}${_ch(path)}`, {
method: 'POST', headers: { 'Content-Type': 'application/json' },
body: data != null ? JSON.stringify(data) : undefined,
});
@@ -37,7 +53,7 @@ async function apiPost(path, data) {
}
async function apiPut(path, data) {
const r = await fetch(`${API}${path}`, {
const r = await fetch(`${API}${_ch(path)}`, {
method: 'PUT', headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(data),
});
@@ -46,7 +62,7 @@ async function apiPut(path, data) {
}
async function apiDelete(path) {
const r = await fetch(`${API}${path}`, { method: 'DELETE' });
const r = await fetch(`${API}${_ch(path)}`, { method: 'DELETE' });
if (!r.ok) throw new Error(`${path} => ${r.status}`);
return r.json();
}
@@ -54,7 +70,7 @@ async function apiDelete(path) {
async function apiUpload(path, file) {
const fd = new FormData();
fd.append('file', file);
const r = await fetch(`${API}${path}`, { method: 'POST', body: fd });
const r = await fetch(`${API}${_ch(path)}`, { method: 'POST', body: fd });
if (!r.ok) throw new Error(`${path} => ${r.status}`);
return r.json();
}
@@ -86,6 +102,7 @@ function usePedalState() {
if (msg.type === 'bypass_changed') setState(prev => prev ? { ...prev, bypass: msg.bypass } : prev);
if (msg.type === 'tuner_changed') setState(prev => prev ? { ...prev, tuner_enabled: msg.tuner_enabled } : prev);
if (msg.type === 'routing_changed') setState(prev => prev ? { ...prev, routing_mode: msg.routing_mode, routing_breakpoint: msg.routing_breakpoint } : prev);
if (msg.type === 'channel_mode_changed') setState(prev => prev ? { ...prev, channel_mode: msg.channel_mode } : prev);
if (msg.type === 'model_loaded' || msg.type === 'ir_loaded') {
apiGet('/api/state').then(s => setState(s));
}
@@ -1285,11 +1302,13 @@ function SettingsScreen({ state, connected, refresh }) {
const [showConnect, setShowConnect] = useState(null);
const [routingMode, setRoutingMode] = useState(state?.routing_mode || 'mono');
const [routingBp, setRoutingBp] = useState(state?.routing_breakpoint || 7);
const [channelMode, setChannelMode] = useState(state?.channel_mode || 'dual-mono');
useEffect(() => {
if (state?.routing_mode) setRoutingMode(state.routing_mode);
if (state?.routing_breakpoint != null) setRoutingBp(state.routing_breakpoint);
}, [state?.routing_mode, state?.routing_breakpoint]);
if (state?.channel_mode) setChannelMode(state.channel_mode);
}, [state?.routing_mode, state?.routing_breakpoint, state?.channel_mode]);
const scanWifi = async () => {
setWifiScanning(true);
@@ -1351,6 +1370,13 @@ function SettingsScreen({ state, connected, refresh }) {
try { await apiPost('/api/routing', { routing_mode: mode, routing_breakpoint: bp }); } catch (e) { /* ignore */ }
};
const handleChannelMode = async (mode) => {
try {
await apiPost('/api/channel-mode', { channel_mode: mode });
setChannelMode(mode);
} catch (e) { /* ignore */ }
};
useEffect(() => { if (connected) { scanWifi(); loadBtPaired(); } }, [connected]);
const wifiState = state?.wifi;
@@ -1521,6 +1547,7 @@ function SettingsScreen({ state, connected, refresh }) {
{/* ── Routing Tab ───────────────────────────────────────────*/}
{tab === 'routing' && (
<>
<div className="card-sm">
<div className="section-label">4CM Routing</div>
<div className="setting-row">
@@ -1546,6 +1573,27 @@ function SettingsScreen({ state, connected, refresh }) {
</div>
)}
</div>
{/* Channel Mode */}
<div className="card-sm">
<div className="section-label">Channel Mode</div>
<div className="setting-row">
<div>
<div className="setting-label">Routing</div>
<div className="setting-desc">Dual-mono: two independent channels · Stereo: linked as L/R pair</div>
</div>
<select value={channelMode} onChange={e => handleChannelMode(e.target.value)}
style={{ background: T.surface, border: `1px solid ${T.border}`, borderRadius: 6, padding: "6px 10px", color: T.textPrimary, fontSize: 13, outline: "none" }}>
<option value="dual-mono">Dual-Mono</option>
<option value="stereo">Stereo</option>
</select>
</div>
<div style={{ fontSize: 10, color: T.textDim, marginTop: 4, padding: '4px 0' }}>
{channelMode === 'dual-mono'
? 'Each channel controlled independently from its own phone.'
: 'Channels A (Left) and B (Right) are controlled together.'}
</div>
</div>
</>
)}
</div>
</div>
@@ -1857,10 +1905,23 @@ const TABS = [
export default function App() {
const [tab, setTab] = useState("rig");
const [showSnapshots, setShowSnapshots] = useState(false);
const [channel, setChannelState] = useState(() => {
try { return localStorage.getItem('pedal_channel') || 'guitar'; } catch { return 'guitar'; }
});
const { state, connected, refresh } = usePedalState();
useEffect(() => { document.title = "Pi Multi-FX Pedal"; }, []);
// Sync channel to localStorage + module-level API var
useEffect(() => {
try { localStorage.setItem('pedal_channel', channel); } catch (e) {}
setChannel(channel);
}, [channel]);
const handleChannel = (ch) => {
setChannelState(ch);
};
return (
<>
<style>{css}</style>
@@ -1870,12 +1931,30 @@ export default function App() {
{/* Status bar */}
<div style={{ display: "flex", justifyContent: "space-between", alignItems: "center",
padding: "6px 12px 4px", background: T.panel, borderBottom: `1px solid ${T.border}`, flexShrink: 0 }}>
padding: "4px 10px", background: T.panel, borderBottom: `1px solid ${T.border}`, flexShrink: 0 }}>
<div style={{ display: "flex", gap: 6, alignItems: "center" }}>
<div style={{ width: 6, height: 6, borderRadius: "50%",
background: connected ? T.green : T.red,
boxShadow: connected ? `0 0 6px ${T.green}` : "none" }} />
<span className="mono" style={{ fontSize: 9, color: T.textSec }}>PI MULTI-FX</span>
{/* Channel toggle — hidden in stereo mode */}
{state?.channel_mode === 'stereo' ? (
<span className="badge badge-amber" style={{ fontSize: 9, padding: "2px 7px", letterSpacing: ".08em" }}>
STEREO
</span>
) : (
<div className="toggle" style={{ margin: 0, height: 22 }}>
<button className={`toggle-btn ${channel === 'guitar' ? 'active' : ''}`}
onClick={() => handleChannel('guitar')}
style={{ padding: '2px 8px', fontSize: 10, fontWeight: 700, letterSpacing: '.1em' }}>
GUITAR
</button>
<button className={`toggle-btn ${channel === 'bass' ? 'active' : ''}`}
onClick={() => handleChannel('bass')}
style={{ padding: '2px 8px', fontSize: 10, fontWeight: 700, letterSpacing: '.1em' }}>
BASS
</button>
</div>
)}
{state?.tuner_enabled && <span className="badge badge-amber" style={{ fontSize: 8, padding: "1px 5px" }}>TUNER</span>}
</div>
<div style={{ display: "flex", gap: 8, alignItems: "center" }}>
+7
View File
@@ -0,0 +1,7 @@
from .types import Bank, Channel, FXBlock, FXType, MIDIMapping, Preset, Snapshot
from .manager import PresetManager
__all__ = [
"Bank", "Channel", "FXBlock", "FXType", "MIDIMapping",
"Preset", "PresetManager", "Snapshot",
]
+256 -56
View File
@@ -19,7 +19,7 @@ import threading
from pathlib import Path
from typing import Optional
from .types import Bank, FXBlock, FXType, MIDIMapping, Preset, Snapshot
from .types import Bank, Channel, FXBlock, FXType, MIDIMapping, Preset, Snapshot
logger = logging.getLogger(__name__)
@@ -44,6 +44,7 @@ def _preset_to_dict(preset: Preset) -> dict:
"""Serialize a Preset to a JSON-compatible dict."""
return {
"name": preset.name,
"channel": preset.channel.value,
"bank": preset.bank,
"program": preset.program,
"master_volume": preset.master_volume,
@@ -137,6 +138,7 @@ def _preset_from_dict(data: dict) -> Preset:
return Preset(
name=data["name"],
channel=Channel(data.get("channel", "guitar")),
bank=data.get("bank", 0),
program=data.get("program", 0),
chain=chain,
@@ -171,6 +173,7 @@ class PresetManager:
self._pipeline = audio_pipeline
# Runtime state
self._current_channel: Channel = Channel.GUITAR
self._current_bank: int = 0
self._current_program: int = 0
self._dirty = False
@@ -185,10 +188,17 @@ class PresetManager:
# Current snapshot slot (0 = none selected)
self._current_snapshot: int = 0
logger.info("PresetManager root: %s", self._dir)
# Migrate legacy flat presets → gtr channel on first boot
self._migrate_legacy_presets()
logger.info("PresetManager root: %s (channel=%s)", self._dir, self._current_channel.value)
# ── Public navigation API ───────────────────────────────────────────────
@property
def current_channel(self) -> Channel:
return self._current_channel
@property
def current_bank(self) -> int:
return self._current_bank
@@ -197,10 +207,43 @@ class PresetManager:
def current_program(self) -> int:
return self._current_program
def set_channel(self, channel: Channel) -> Preset:
"""Switch the active channel and restore its last preset.
Args:
channel: The channel to switch to.
Returns:
The currently active Preset in the new channel.
"""
if channel == self._current_channel:
try:
return self.load(self._current_bank, self._current_program)
except FileNotFoundError:
pass
return self._select_and_activate(self._current_bank, self._current_program)
# Save current channel state first
self.save_state()
self._current_channel = channel
# Try restoring channel-specific state
restored = self._restore_channel_state()
if restored:
return restored
# Default to bank 0, program 0 in the new channel
try:
return self._select_and_activate(0, 0)
except Exception:
# Channel is empty — create first blank preset
return self._select_and_activate(0, 0)
@property
def current_preset_path(self) -> Path:
"""Path to the slot for the current (bank, program)."""
return self._preset_path(self._current_bank, self._current_program)
"""Path to the slot for the current (channel, bank, program)."""
return self._preset_path(self._current_channel, self._current_bank, self._current_program)
def preset_up(self) -> Preset:
"""Select next preset in current bank (wraps 3→0).
@@ -240,17 +283,18 @@ class PresetManager:
"""
return self._switch_bank(-1)
def select(self, bank: int, program: int) -> Preset:
def select(self, bank: int, program: int, channel: Channel | None = None) -> Preset:
"""Directly select a specific (bank, program) slot.
Args:
bank: Bank number.
program: Preset index (0-3) within the bank.
channel: Optional channel override (defaults to current).
Returns:
The activated Preset.
"""
return self._select_and_activate(bank, program)
return self._select_and_activate(bank, program, channel or self._current_channel)
def midi_pc(self, channel: int, program: int) -> Preset:
"""Handle MIDI Program Change: bank=channel, program=program.
@@ -267,15 +311,17 @@ class PresetManager:
# ── CRUD ────────────────────────────────────────────────────────────────
def save(self, preset: Preset, *, auto: bool = False) -> None:
def save(self, preset: Preset, *, auto: bool = False, channel: Channel | None = None) -> None:
"""Save a preset to its (bank, program) slot on disk.
Args:
preset: The preset to persist.
auto: If True, this was triggered by auto-save and the
dirty/current-state write is implicitly handled.
channel: Channel to save under (defaults to preset.channel).
"""
path = self._preset_path(preset.bank, preset.program)
ch = channel or preset.channel
path = self._preset_path(ch, preset.bank, preset.program)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(_preset_to_dict(preset), indent=2, sort_keys=True),
@@ -287,16 +333,17 @@ class PresetManager:
self._known_banks.add(preset.bank)
# Save/update bank metadata
self._save_bank_meta(preset.bank)
self._save_bank_meta(ch, preset.bank)
logger.info("Saved preset '%s'%s", preset.name, path)
def load(self, bank: int, program: int) -> Preset:
def load(self, bank: int, program: int, channel: Channel | None = None) -> Preset:
"""Load a preset from disk.
Args:
bank: Bank number.
program: Preset index (0-3).
channel: Channel to load from (defaults to current).
Returns:
The deserialized Preset.
@@ -305,41 +352,52 @@ class PresetManager:
FileNotFoundError: If the preset slot doesn't exist.
json.JSONDecodeError: If the file is corrupt.
"""
path = self._preset_path(bank, program)
ch = channel or self._current_channel
path = self._preset_path(ch, bank, program)
if not path.exists():
raise FileNotFoundError(
f"No preset at (bank={bank}, program={program}): {path}"
f"No preset at (channel={ch.value}, bank={bank}, program={program}): {path}"
)
data = json.loads(path.read_text(encoding="utf-8"))
preset = _preset_from_dict(data)
# Ensure metadata is correct even if file was manually altered
preset.channel = ch
preset.bank = bank
preset.program = program
return preset
def delete(self, bank: int, program: int) -> None:
"""Delete a preset slot from disk."""
path = self._preset_path(bank, program)
def delete(self, bank: int, program: int, channel: Channel | None = None) -> None:
"""Delete a preset slot from disk.
Args:
bank: Bank number.
program: Preset index.
channel: Channel to delete from (defaults to current).
"""
ch = channel or self._current_channel
path = self._preset_path(ch, bank, program)
if path.exists():
path.unlink()
logger.info("Deleted preset at (bank=%d, program=%d)", bank, program)
logger.info("Deleted preset at (channel=%s, bank=%d, program=%d)", ch.value, bank, program)
self._dirty = True
def rename(self, bank: int, program: int, new_name: str) -> Preset:
def rename(self, bank: int, program: int, new_name: str, channel: Channel | None = None) -> Preset:
"""Rename a preset and re-save it.
Args:
bank: Bank number.
program: Preset index.
new_name: Replacement name.
channel: Channel (defaults to current).
Returns:
The renamed Preset.
"""
preset = self.load(bank, program)
preset = self.load(bank, program, channel)
preset.name = new_name
self.save(preset)
logger.info("Renamed preset at (%d,%d) → '%s'", bank, program, new_name)
self.save(preset, channel=channel)
logger.info("Renamed preset at (%s,%d,%d) → '%s'",
(channel or self._current_channel).value, bank, program, new_name)
return preset
def reorder(self, bank: int, program: int, new_program: int) -> None:
@@ -415,10 +473,22 @@ class PresetManager:
# ── Bank management ─────────────────────────────────────────────────────
def list_banks(self) -> list[Bank]:
"""Scan the preset directory and return known banks sorted by number."""
def list_banks(self, channel: Channel | None = None) -> list[Bank]:
"""Scan the preset directory and return known banks sorted by number.
Args:
channel: Channel to list banks for (defaults to current).
Returns:
List of Bank objects sorted by number.
"""
ch = channel or self._current_channel
ch_dir = self._dir / ch.value
if not ch_dir.is_dir():
return []
banks: dict[int, Bank] = {}
for meta_path in sorted(self._dir.glob("bank_*/bank.json")):
for meta_path in sorted(ch_dir.glob("bank_*/bank.json")):
try:
data = json.loads(meta_path.read_text(encoding="utf-8"))
num = data.get("number", 0)
@@ -434,7 +504,7 @@ class PresetManager:
continue
# Also discover banks by scanning dirs without bank.json
for p_dir in sorted(self._dir.glob("bank_*")):
for p_dir in sorted(ch_dir.glob("bank_*")):
try:
num = int(p_dir.name.split("_")[1])
if num not in banks:
@@ -460,20 +530,25 @@ class PresetManager:
# ── Auto-restore ────────────────────────────────────────────────────────
def save_state(self) -> None:
"""Persist the current (bank, program) for power-cycle restore.
"""Persist the current (channel, bank, program) for power-cycle restore.
Call this whenever the active preset changes to guarantee
Call this whenever the active preset or channel changes to guarantee
the pedal comes back to the same sound after a power cycle.
"""
state = {"current_bank": self._current_bank, "current_program": self._current_program}
state = {
"current_channel": self._current_channel.value,
"current_bank": self._current_bank,
"current_program": self._current_program,
}
state_path = self._dir / "state.json"
state_path.write_text(
json.dumps(state, indent=2), encoding="utf-8"
)
logger.debug("State saved: bank=%d program=%d", self._current_bank, self._current_program)
logger.debug("State saved: channel=%s bank=%d program=%d",
self._current_channel.value, self._current_bank, self._current_program)
def restore_state(self) -> Optional[Preset]:
"""Restore the last active preset from state.json.
"""Restore the last active (channel, bank, program) from state.json.
Returns:
The restored Preset, or None if no state file exists or the
@@ -486,17 +561,20 @@ class PresetManager:
try:
state = json.loads(state_path.read_text(encoding="utf-8"))
ch = Channel(state.get("current_channel", "guitar"))
bank = state["current_bank"]
program = state["current_program"]
preset = self.load(bank, program)
except (FileNotFoundError, json.JSONDecodeError, KeyError):
preset = self.load(bank, program, ch)
except (FileNotFoundError, json.JSONDecodeError, KeyError, ValueError):
logger.warning("Could not restore state, falling back to first preset")
return None
self._current_channel = ch
self._current_bank = bank
self._current_program = program
self._dirty = False
logger.info("Restored state: bank=%d program=%d '%s'", bank, program, preset.name)
logger.info("Restored state: channel=%s bank=%d program=%d '%s'",
ch.value, bank, program, preset.name)
return preset
# ── Activation ──────────────────────────────────────────────────────────
@@ -514,7 +592,8 @@ class PresetManager:
# ── Factory presets ─────────────────────────────────────────────────────
def install_factory_presets(self, overwrite: bool = False) -> int:
def install_factory_presets(self, overwrite: bool = False,
channel: Channel = Channel.GUITAR) -> int:
"""Install bundled factory presets into the preset store.
Scans ``FACTORY_PRESET_DIR`` for ``bank_*/preset_*.json`` files
@@ -523,6 +602,7 @@ class PresetManager:
Args:
overwrite: If True, overwrite existing presets at the same
(bank, program) slots. If False, skip occupied slots.
channel: Which channel to install factory presets into (default GUITAR).
Returns:
Number of factory presets installed.
@@ -542,14 +622,14 @@ class PresetManager:
logger.warning("Skipping malformed factory preset path: %s", src)
continue
dest = self._preset_path(bank_num, prog_num)
dest = self._preset_path(channel, bank_num, prog_num)
if dest.exists() and not overwrite:
continue
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, dest)
self._known_banks.add(bank_num)
self._save_bank_meta(bank_num)
self._save_bank_meta(channel, bank_num)
count += 1
logger.debug("Installed factory preset: %s", dest)
@@ -595,19 +675,31 @@ class PresetManager:
# ── Internal helpers ────────────────────────────────────────────────────
def _preset_path(self, bank: int, program: int) -> Path:
"""Compute the on-disk path for a (bank, program) slot."""
return self._dir / f"bank_{bank}" / f"preset_{program}.json"
def _preset_path(self, channel: Channel, bank: int, program: int) -> Path:
"""Compute the on-disk path for a (channel, bank, program) slot."""
return self._dir / channel.value / f"bank_{bank}" / f"preset_{program}.json"
def _bank_dir(self, bank: int) -> Path:
return self._dir / f"bank_{bank}"
def _channel_dir(self, channel: Channel) -> Path:
"""Return the root directory for a given channel."""
return self._dir / channel.value
def _bank_meta_path(self, bank: int) -> Path:
return self._bank_dir(bank) / "bank.json"
def _bank_dir(self, channel: Channel, bank: int) -> Path:
return self._channel_dir(channel) / f"bank_{bank}"
def _get_or_create_bank(self, number: int, name: str = "") -> Bank:
"""Return an existing bank or create a new one on disk."""
meta_path = self._bank_meta_path(number)
def _bank_meta_path(self, channel: Channel, bank: int) -> Path:
return self._bank_dir(channel, bank) / "bank.json"
def _get_or_create_bank(self, number: int, name: str = "",
channel: Channel | None = None) -> Bank:
"""Return an existing bank or create a new one on disk.
Args:
number: Bank number.
name: Optional display name.
channel: Channel (defaults to current).
"""
ch = channel or self._current_channel
meta_path = self._bank_meta_path(ch, number)
if meta_path.exists():
try:
data = json.loads(meta_path.read_text(encoding="utf-8"))
@@ -620,14 +712,15 @@ class PresetManager:
# Create new bank
bank = Bank(name=name or f"Bank {number}", number=number)
self._save_bank_meta(number, bank.name)
self._save_bank_meta(ch, number, bank.name)
self._known_banks.add(number)
logger.info("Created bank %d: '%s'", number, bank.name)
logger.info("Created bank %d: '%s' (channel=%s)", number, bank.name, ch.value)
return bank
def _save_bank_meta(self, bank_num: int, name: str | None = None) -> None:
def _save_bank_meta(self, channel: Channel, bank_num: int,
name: str | None = None) -> None:
"""Write or update a bank's metadata file."""
meta_path = self._bank_meta_path(bank_num)
meta_path = self._bank_meta_path(channel, bank_num)
existing = {}
if meta_path.exists():
try:
@@ -673,6 +766,7 @@ class PresetManager:
except FileNotFoundError:
preset = Preset(
name=f"Empty {new_bank.name}",
channel=self._current_channel,
bank=new_bank.number,
program=self._current_program,
)
@@ -680,11 +774,14 @@ class PresetManager:
self._current_bank = new_bank.number
self._current_program = preset.program
self.activate(preset)
self._save_channel_state()
self.save_state()
logger.info("Switched to bank %d, preset '%s'", new_bank.number, preset.name)
logger.info("Switched to bank %d, preset '%s' (channel=%s)",
new_bank.number, preset.name, self._current_channel.value)
return new_bank, preset
def _select_and_activate(self, bank: int, program: int) -> Preset:
def _select_and_activate(self, bank: int, program: int,
channel: Channel | None = None) -> Preset:
"""Select a (bank, program) slot and activate it.
If the slot is empty, creates a blank preset in that slot first.
@@ -692,22 +789,125 @@ class PresetManager:
Args:
bank: Bank number.
program: Preset index (0-3).
channel: Channel to operate on (defaults to current).
Returns:
The activated Preset.
"""
ch = channel or self._current_channel
program = max(0, min(PRESETS_PER_BANK - 1, program))
try:
preset = self.load(bank, program)
preset = self.load(bank, program, ch)
except FileNotFoundError:
# Create a blank placeholder preset
bank_obj = self._get_or_create_bank(bank)
preset = Preset(name=f"New {bank_obj.name} #{program}", bank=bank, program=program)
self.save(preset, auto=True)
bank_obj = self._get_or_create_bank(bank, channel=ch)
preset = Preset(
name=f"New {bank_obj.name} #{program}",
channel=ch, bank=bank, program=program,
)
self.save(preset, auto=True, channel=ch)
self._current_channel = ch
self._current_bank = bank
self._current_program = program
self.activate(preset)
self._save_channel_state(ch)
self.save_state()
return preset
return preset
# ── Channel state helpers ───────────────────────────────────────────
def _channel_state_path(self, channel: Channel) -> Path:
"""Path to the per-channel state file."""
return self._channel_dir(channel) / "channel_state.json"
def _save_channel_state(self, channel: Channel | None = None) -> None:
"""Persist the channel-specific (bank, program) state."""
ch = channel or self._current_channel
state = {
"current_bank": self._current_bank,
"current_program": self._current_program,
}
path = self._channel_state_path(ch)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(state, indent=2), encoding="utf-8")
logger.debug("Channel state saved for %s: bank=%d program=%d",
ch.value, self._current_bank, self._current_program)
def _restore_channel_state(self, channel: Channel | None = None) -> Optional[Preset]:
"""Restore the per-channel (bank, program) state.
Args:
channel: Channel to restore (defaults to current).
Returns:
The restored Preset, or None if no saved state exists.
"""
ch = channel or self._current_channel
path = self._channel_state_path(ch)
if not path.exists():
return None
try:
state = json.loads(path.read_text(encoding="utf-8"))
bank = state["current_bank"]
program = state["current_program"]
preset = self.load(bank, program, ch)
except (FileNotFoundError, json.JSONDecodeError, KeyError):
return None
self._current_bank = bank
self._current_program = program
logger.info("Restored channel state for %s: bank=%d program=%d '%s'",
ch.value, bank, program, preset.name)
return preset
# ── Legacy migration ────────────────────────────────────────────────
def _migrate_legacy_presets(self) -> None:
"""Migrate flat bank_* directories under root into gtr/ subdirectory.
Before dual-channel support, presets were stored as:
{root}/bank_0/preset_0.json
Now they should live under:
{root}/gtr/bank_0/preset_0.json
This runs once on first boot with the new layout.
"""
ch_dir = self._dir / "guitar"
if ch_dir.exists():
# Already migrated or new install — nothing to do
return
# Check if there are legacy flat bank_* dirs
legacy_banks = sorted(self._dir.glob("bank_*"))
if not legacy_banks:
return # Fresh install, nothing to migrate
logger.info("Migrating %d legacy bank directories → gtr channel ...", len(legacy_banks))
ch_dir.mkdir(parents=True, exist_ok=True)
for src in legacy_banks:
dest = ch_dir / src.name
try:
shutil.move(str(src), str(dest))
logger.debug(" migrated %s%s", src.name, dest)
except OSError as e:
logger.warning(" failed to migrate %s: %s", src.name, e)
# Also migrate state.json if it exists at root
legacy_state = self._dir / "state.json"
if legacy_state.exists():
try:
state = json.loads(legacy_state.read_text(encoding="utf-8"))
if "current_channel" not in state:
state["current_channel"] = "guitar"
legacy_state.write_text(
json.dumps(state, indent=2), encoding="utf-8"
)
except (json.JSONDecodeError, OSError):
pass
logger.info("Legacy preset migration complete")
+7
View File
@@ -7,6 +7,12 @@ from dataclasses import dataclass, field
from typing import Optional
class Channel(enum.StrEnum):
"""Guitar or Bass signal channel — each has its own independent preset bank."""
GUITAR = "guitar"
BASS = "bass"
class FXType(enum.StrEnum):
"""Types of effects in the pedal signal chain."""
# Existing
@@ -101,6 +107,7 @@ class Snapshot:
class Preset:
"""A complete pedal preset — full signal chain state."""
name: str
channel: Channel = Channel.GUITAR
bank: int = 0
program: int = 0
chain: list[FXBlock] = field(default_factory=list)
+404 -99
View File
@@ -32,7 +32,7 @@ from ..dsp.nam_host import NAMHost
from ..dsp.ir_loader import IRLoader
from ..dsp.pipeline import AudioPipeline, SAMPLE_RATE
from ..presets.manager import PresetManager
from ..presets.types import FXBlock, FXType, Preset
from ..presets.types import Channel, FXBlock, FXType, Preset
from ..system.tonedownload import (
Tone3000Client,
ModelResult,
@@ -64,22 +64,156 @@ else:
]
UI_DIST_DIR = next((d for d in _candidates if d.is_dir()), None)
# ── Channel support ──────────────────────────────────────────────────────────────
VALID_CHANNELS: frozenset = frozenset({"guitar", "bass"})
def _resolve_channel(channel: str = "guitar") -> str:
"""Normalize a channel string; fall back to 'guitar' for invalid values."""
if isinstance(channel, str) and channel.lower() in VALID_CHANNELS:
return channel.lower()
return "guitar"
# ── Channel mode persistence ────────────────────────────────────────────────
CHANNEL_MODE_FILE = Path.home() / ".pedal" / "channel_mode.json"
def _load_channel_mode() -> str:
"""Load channel mode from disk; default to 'dual-mono'."""
try:
if CHANNEL_MODE_FILE.exists():
data = json.loads(CHANNEL_MODE_FILE.read_text())
mode = data.get("mode", "dual-mono")
if mode in ("dual-mono", "stereo"):
return mode
except Exception:
pass
return "dual-mono"
def _save_channel_mode(mode: str) -> None:
"""Persist channel mode to disk."""
CHANNEL_MODE_FILE.parent.mkdir(parents=True, exist_ok=True)
CHANNEL_MODE_FILE.write_text(json.dumps({"mode": mode}))
@dataclass
class ChannelState:
"""Per-channel runtime state (volume, bypass, tuner)."""
volume: float = 0.8
bypass: bool = False
tuner_enabled: bool = False
current_snapshot: int = 0
class ChannelManager:
"""Manages dual-channel vs stereo routing mode and per-channel state.
In ``dual-mono`` mode each channel (guitar / bass) has independent
runtime state. In ``stereo`` mode the two channels are linked —
the ``guitar`` channel dict is the single source of truth for both.
"""
def __init__(self) -> None:
self.mode: str = _load_channel_mode()
self._channels: dict[str, ChannelState] = {
"guitar": ChannelState(),
"bass": ChannelState(),
}
# ── Mode management ────────────────────────────────────────────────
def set_mode(self, mode: str) -> None:
if mode not in ("dual-mono", "stereo"):
raise ValueError(f"Unknown channel mode: {mode!r}")
self.mode = mode
_save_channel_mode(mode)
# ── Per-channel helpers ────────────────────────────────────────────
def state_for(self, channel: str) -> ChannelState:
"""Return the ``ChannelState`` for a given channel.
In stereo mode both channels read from the ``guitar`` slot.
Falls back to ``guitar`` for unknown channel names.
"""
if self.mode == "stereo":
return self._channels["guitar"]
return self._channels.get(channel, self._channels["guitar"])
def sync_to_stereo(self) -> None:
"""When switching *to* stereo, copy bass -> guitar so guitar has
the most recently changed values from either channel."""
if self.mode == "stereo":
g = self._channels["guitar"]
b = self._channels["bass"]
# Pick the "latest" non-default values
for attr in ("volume", "bypass", "tuner_enabled", "current_snapshot"):
bv = getattr(b, attr)
if bv != getattr(ChannelState(), attr):
setattr(g, attr, bv)
def resolve_channel(self, param: str | None) -> str:
"""Return the active channel name.
In stereo mode always returns ``"guitar"`` (the canonical slot).
In dual-mono returns *param* if it's ``"guitar"`` or ``"bass"``,
otherwise falls back to ``"guitar"``.
"""
if self.mode == "stereo":
return "guitar"
if param in ("guitar", "bass"):
return param
return "guitar"
# ── Dependency container ─────────────────────────────────────────────────────
@dataclass
class WebServerDeps:
"""Dependencies injected into the web server from the host pedal app."""
"""Dependencies injected into the web server from the host pedal app.
Supports dual-channel operation (guitar + bass). Each channel has its
own preset manager, audio pipeline, NAM host, and IR loader so the two
signal paths are fully independent.
The guitar fields are the defaults — callers that don't need dual-channel
can just set the guitar fields as before and everything works.
"""
# Guitar channel (default)
presets: Optional[PresetManager] = None
pipeline: Optional[AudioPipeline] = None
nam_host: Optional[NAMHost] = None
ir_loader: Optional[IRLoader] = None
# Bass channel (separate from guitar)
bass_presets: Optional[PresetManager] = None
bass_pipeline: Optional[AudioPipeline] = None
bass_nam_host: Optional[NAMHost] = None
bass_ir_loader: Optional[IRLoader] = None
@property
def is_connected(self) -> bool:
"""True if we're wired to a live pedal instance."""
return self.presets is not None
"""True if we're wired to a live pedal instance (any channel)."""
return self.presets is not None or self.bass_presets is not None
def get_deps(self, channel: str) -> tuple:
"""Return (presets, pipeline, nam_host, ir_loader) for the given channel.
Falls back to guitar fields for unrecognised channels.
"""
if _resolve_channel(channel) == "bass":
return (
self.bass_presets,
self.bass_pipeline,
self.bass_nam_host,
self.bass_ir_loader,
)
return (self.presets, self.pipeline, self.nam_host, self.ir_loader)
# ── WebSocket connection manager ─────────────────────────────────────────────
@@ -139,11 +273,21 @@ class WebServer:
self.port = port
self._manager = ConnectionManager()
self._channel_mgr = ChannelManager()
self._server: Optional[uvicorn.Server] = None
self._task: Optional[asyncio.Task] = None
self._tonedownload: Optional[Tone3000Client] = None
self._app = self._build_app()
# ── Channel dep resolution ──────────────────────────────────────
def _channel_deps(self, channel: str = "guitar") -> tuple:
"""Shorthand for self.deps.get_deps(channel).
Returns (presets, pipeline, nam_host, ir_loader).
"""
return self.deps.get_deps(channel)
# ── App factory ─────────────────────────────────────────────────────
def _build_app(self) -> FastAPI:
@@ -177,8 +321,8 @@ class WebServer:
if index_html.exists():
content = index_html.read_text()
return HTMLResponse(content)
# Fallback: legacy dashboard
state = self._gather_state()
# Fallback: legacy dashboard (guitar channel state)
state = self._gather_state(channel="guitar")
ctx = {"request": request}
ctx.update(state)
return templates.TemplateResponse(request, "dashboard.html", ctx)
@@ -201,7 +345,7 @@ class WebServer:
@app.get("/settings", response_class=HTMLResponse)
async def settings_page(request: Request):
"""Settings page."""
state = self._gather_state()
state = self._gather_state(channel="guitar")
ctx = {"request": request}
ctx.update(state)
return templates.TemplateResponse(request, "settings.html", ctx)
@@ -209,17 +353,25 @@ class WebServer:
# ── REST API ────────────────────────────────────────────────
@app.get("/api/state")
async def get_state():
"""Full current pedal state snapshot."""
state = self._gather_state()
async def get_state(channel: Optional[str] = None, combined: bool = False):
"""Full pedal state snapshot.
Without parameters returns the active (guitar) channel's flat state.
``?channel=guitar|bass`` returns per-channel state.
``?combined=true`` returns both channel states in a combined view.
"""
if combined:
state = self._gather_state_combined()
else:
state = self._gather_state(channel=channel)
if not state:
raise HTTPException(status_code=503, detail="Pedal not connected")
return state
@app.get("/api/presets")
async def list_presets():
"""List all banks and their presets."""
pm = self.deps.presets
async def list_presets(channel: str = "guitar"):
"""List all banks and their presets for the given channel."""
pm, _pl, _nam, _ir = self._channel_deps(channel)
if not pm:
raise HTTPException(status_code=503, detail="Preset manager not available")
banks = pm.list_banks()
@@ -257,9 +409,9 @@ class WebServer:
return {"banks": result}
@app.get("/api/presets/{bank}/{program}")
async def get_preset(bank: int, program: int):
"""Get a specific preset."""
pm = self.deps.presets
async def get_preset(bank: int, program: int, channel: str = "guitar"):
"""Get a specific preset for the given channel."""
pm, _pl, _nam, _ir = self._channel_deps(channel)
if not pm:
raise HTTPException(status_code=503)
try:
@@ -286,17 +438,18 @@ class WebServer:
raise HTTPException(status_code=404, detail="Preset not found")
@app.post("/api/presets/{bank}/{program}/activate")
async def activate_preset(bank: int, program: int):
"""Activate a preset (load into pipeline)."""
pm = self.deps.presets
async def activate_preset(bank: int, program: int, channel: str = "guitar"):
"""Activate a preset (load into pipeline) for the given channel."""
pm, _pl, _nam, _ir = self._channel_deps(channel)
if not pm:
raise HTTPException(status_code=503)
try:
preset = pm.select(bank, program)
pm.save_state()
state = self._gather_state()
state = self._gather_state(channel)
await self._manager.broadcast({
"type": "preset_changed",
"channel": channel,
"preset": {
"name": preset.name,
"bank": preset.bank,
@@ -309,9 +462,9 @@ class WebServer:
raise HTTPException(status_code=500, detail=str(e))
@app.put("/api/presets/{bank}/{program}")
async def save_preset(bank: int, program: int, data: dict):
"""Save/update a preset."""
pm = self.deps.presets
async def save_preset(bank: int, program: int, data: dict, channel: str = "guitar"):
"""Save/update a preset for the given channel."""
pm, _pl, _nam, _ir = self._channel_deps(channel)
if not pm:
raise HTTPException(status_code=503)
try:
@@ -327,6 +480,7 @@ class WebServer:
))
preset = Preset(
name=data.get("name", f"Preset {bank}-{program}"),
channel=Channel(channel),
bank=bank,
program=program,
chain=chain,
@@ -339,9 +493,9 @@ class WebServer:
raise HTTPException(status_code=500, detail=str(e))
@app.delete("/api/presets/{bank}/{program}")
async def delete_preset(bank: int, program: int):
"""Delete a preset."""
pm = self.deps.presets
async def delete_preset(bank: int, program: int, channel: str = "guitar"):
"""Delete a preset for the given channel."""
pm, _pl, _nam, _ir = self._channel_deps(channel)
if not pm:
raise HTTPException(status_code=503)
try:
@@ -388,9 +542,9 @@ class WebServer:
}
@app.get("/api/snapshots")
async def list_snapshots():
"""List snapshot slots for the current preset."""
pm = self.deps.presets
async def list_snapshots(channel: str = "guitar"):
"""List snapshot slots for the current preset on the given channel."""
pm, _pl, _nam, _ir = self._channel_deps(channel)
preset = _get_current_preset(pm)
if not preset:
return {"snapshots": {}}
@@ -418,10 +572,9 @@ class WebServer:
return {"snapshots": data, "current_snapshot": getattr(pm, '_current_snapshot', 0)}
@app.post("/api/snapshots/{slot}/save")
async def save_snapshot(slot: int, data: Optional[dict] = None):
"""Save current state to a snapshot slot (1-8)."""
pm = self.deps.presets
pl = self.deps.pipeline
async def save_snapshot(slot: int, data: Optional[dict] = None, channel: str = "guitar"):
"""Save current state to a snapshot slot (1-8) for the given channel."""
pm, pl, _nam, _ir = self._channel_deps(channel)
if not pm or not (1 <= slot <= 8):
raise HTTPException(status_code=400, detail="Invalid slot (1-8)")
preset = _get_current_preset(pm)
@@ -459,10 +612,9 @@ class WebServer:
return {"ok": True, "slot": slot, "name": snap_data["name"]}
@app.post("/api/snapshots/{slot}/recall")
async def recall_snapshot(slot: int):
"""Recall a snapshot: restore block states + params from slot."""
pm = self.deps.presets
pl = self.deps.pipeline
async def recall_snapshot(slot: int, channel: str = "guitar"):
"""Recall a snapshot: restore block states + params from slot for the given channel."""
pm, pl, _nam, _ir = self._channel_deps(channel)
if not pm or not (1 <= slot <= 8):
raise HTTPException(status_code=400, detail="Invalid slot (1-8)")
preset = _get_current_preset(pm)
@@ -493,9 +645,10 @@ class WebServer:
pm.save(preset)
pm._current_snapshot = slot
state = self._gather_state()
state = self._gather_state(channel)
await self._manager.broadcast({
"type": "snapshot_recalled",
"channel": channel,
"slot": slot,
"name": snap.name,
"state": state,
@@ -503,9 +656,9 @@ class WebServer:
return {"ok": True, "slot": slot, "name": snap.name}
@app.put("/api/snapshots/{slot}")
async def update_snapshot(slot: int, data: dict):
"""Rename or update a snapshot slot."""
pm = self.deps.presets
async def update_snapshot(slot: int, data: dict, channel: str = "guitar"):
"""Rename or update a snapshot slot for the given channel."""
pm, _pl, _nam, _ir = self._channel_deps(channel)
if not pm or not (1 <= slot <= 8):
raise HTTPException(status_code=400, detail="Invalid slot (1-8)")
preset = _get_current_preset(pm)
@@ -519,15 +672,16 @@ class WebServer:
pm.save(preset)
await self._manager.broadcast({
"type": "snapshot_updated",
"channel": channel,
"slot": slot,
"name": snap.name,
})
return {"ok": True, "slot": slot, "name": snap.name}
@app.delete("/api/snapshots/{slot}")
async def delete_snapshot(slot: int):
"""Delete a snapshot slot."""
pm = self.deps.presets
async def delete_snapshot(slot: int, channel: str = "guitar"):
"""Delete a snapshot slot for the given channel."""
pm, _pl, _nam, _ir = self._channel_deps(channel)
if not pm or not (1 <= slot <= 8):
raise HTTPException(status_code=400, detail="Invalid slot (1-8)")
preset = _get_current_preset(pm)
@@ -539,14 +693,15 @@ class WebServer:
pm.save(preset)
await self._manager.broadcast({
"type": "snapshot_deleted",
"channel": channel,
"slot": slot,
})
return {"ok": True, "slot": slot}
@app.get("/api/snapshots/{slot}")
async def get_snapshot(slot: int):
"""Get a single snapshot slot."""
pm = self.deps.presets
async def get_snapshot(slot: int, channel: str = "guitar"):
"""Get a single snapshot slot for the given channel."""
pm, _pl, _nam, _ir = self._channel_deps(channel)
preset = _get_current_preset(pm)
if not preset:
raise HTTPException(status_code=404, detail="No active preset")
@@ -556,6 +711,7 @@ class WebServer:
return {
"name": snap.name,
"slot": slot,
"channel": channel,
"chain": [
{
"fx_type": b.fx_type.value,
@@ -569,9 +725,9 @@ class WebServer:
}
@app.post("/api/bypass")
async def toggle_bypass(data: Optional[dict] = None):
"""Toggle or set bypass state."""
pl = self.deps.pipeline
async def toggle_bypass(data: Optional[dict] = None, channel: str = "guitar"):
"""Toggle or set bypass state for the given channel."""
_pm, pl, _nam, _ir = self._channel_deps(channel)
if not pl:
raise HTTPException(status_code=503)
if data and "bypass" in data:
@@ -580,14 +736,15 @@ class WebServer:
pl._bypassed = not pl._bypassed
await self._manager.broadcast({
"type": "bypass_changed",
"channel": channel,
"bypass": pl._bypassed,
})
return {"ok": True, "bypass": pl._bypassed}
@app.post("/api/tuner")
async def toggle_tuner(data: Optional[dict] = None):
"""Toggle tuner mode."""
pl = self.deps.pipeline
async def toggle_tuner(data: Optional[dict] = None, channel: str = "guitar"):
"""Toggle tuner mode for the given channel."""
_pm, pl, _nam, _ir = self._channel_deps(channel)
if not pl:
raise HTTPException(status_code=503)
if data and "enabled" in data:
@@ -596,14 +753,15 @@ class WebServer:
pl._tuner_enabled = not pl._tuner_enabled
await self._manager.broadcast({
"type": "tuner_changed",
"channel": channel,
"tuner_enabled": pl._tuner_enabled,
})
return {"ok": True, "tuner_enabled": pl._tuner_enabled}
@app.get("/api/tuner/pitch")
async def get_tuner_pitch():
"""Get current tuner pitch detection data (fast poll endpoint)."""
pl = self.deps.pipeline
async def get_tuner_pitch(channel: str = "guitar"):
"""Get current tuner pitch detection data for the given channel (fast poll endpoint)."""
_pm, pl, _nam, _ir = self._channel_deps(channel)
if not pl:
return {
"frequency": 0.0, "note": "--", "cents": 0,
@@ -619,21 +777,21 @@ class WebServer:
}
@app.put("/api/volume")
async def set_volume(data: dict):
"""Set master volume (0.0-1.0)."""
pl = self.deps.pipeline
async def set_volume(data: dict, channel: str = "guitar"):
"""Set master volume (0.0-1.0) for the given channel."""
_pm, pl, _nam, _ir = self._channel_deps(channel)
if not pl:
raise HTTPException(status_code=503)
vol = max(0.0, min(1.0, float(data.get("volume", 0.8))))
pl._master_volume = vol
return {"ok": True, "volume": vol}
return {"ok": True, "volume": vol, "channel": channel}
# ── 4CM routing endpoints ────────────────────────────────
@app.get("/api/routing")
async def get_routing():
"""Get current routing mode and breakpoint."""
pl = self.deps.pipeline
async def get_routing(channel: str = "guitar"):
"""Get current routing mode and breakpoint for the given channel."""
_pm, pl, _nam, _ir = self._channel_deps(channel)
if not pl:
raise HTTPException(status_code=503)
return {
@@ -642,15 +800,15 @@ class WebServer:
}
@app.post("/api/routing")
async def set_routing(data: dict):
"""Set routing mode and/or breakpoint.
async def set_routing(data: dict, channel: str = "guitar"):
"""Set routing mode and/or breakpoint for the given channel.
Body:
routing_mode (optional): ``\"mono\"`` or ``\"4cm\"``.
routing_breakpoint (optional): chain index for split.
"""
pl = self.deps.pipeline
pm = self.deps.presets
_pm, pl, _nam, _ir = self._channel_deps(channel)
pm, _pl2, _nam2, _ir2 = self._channel_deps(channel)
if not pl:
raise HTTPException(status_code=503)
mode = data.get("routing_mode", pl.routing_mode)
@@ -669,16 +827,47 @@ class WebServer:
await self._manager.broadcast({
"type": "routing_changed",
"channel": channel,
"routing_mode": pl.routing_mode,
"routing_breakpoint": pl.routing_breakpoint,
})
return {"ok": True, "routing_mode": pl.routing_mode,
"routing_breakpoint": pl.routing_breakpoint}
# ── Channel mode endpoints ───────────────────────────────
@app.get("/api/channel-mode")
async def get_channel_mode():
"""Get the current channel routing mode."""
return {"channel_mode": self._channel_mgr.mode}
@app.post("/api/channel-mode")
async def set_channel_mode(data: dict):
"""Set channel routing mode.
Body: { "channel_mode": "dual-mono" | "stereo" }
"""
mode = data.get("channel_mode", "")
if mode not in ("dual-mono", "stereo"):
raise HTTPException(
status_code=400,
detail=f"Invalid channel_mode: {mode!r}. Must be 'dual-mono' or 'stereo'.",
)
was = self._channel_mgr.mode
self._channel_mgr.set_mode(mode)
if mode == "stereo" and was == "dual-mono":
self._channel_mgr.sync_to_stereo()
logger.info("Channel mode switched: %s%s", was, mode)
await self._manager.broadcast({
"type": "channel_mode_changed",
"channel_mode": mode,
})
return {"ok": True, "channel_mode": mode}
@app.get("/api/models")
async def list_models():
"""List available NAM models."""
nam = self.deps.nam_host
async def list_models(channel: str = "guitar"):
"""List available NAM models for the given channel."""
_pm, _pl, nam, _ir = self._channel_deps(channel)
if not nam:
raise HTTPException(status_code=503)
models = nam.list_available_models()
@@ -697,9 +886,9 @@ class WebServer:
}
@app.post("/api/models/load")
async def load_model(data: dict):
"""Load a NAM model by path."""
nam = self.deps.nam_host
async def load_model(data: dict, channel: str = "guitar"):
"""Load a NAM model by path for the given channel."""
_pm, _pl, nam, _ir = self._channel_deps(channel)
if not nam:
raise HTTPException(status_code=503)
path = data.get("path", "")
@@ -715,18 +904,18 @@ class WebServer:
return {"ok": True}
@app.post("/api/models/unload")
async def unload_model():
"""Unload the current NAM model."""
nam = self.deps.nam_host
async def unload_model(channel: str = "guitar"):
"""Unload the current NAM model for the given channel."""
_pm, _pl, nam, _ir = self._channel_deps(channel)
if not nam:
raise HTTPException(status_code=503)
nam.unload()
return {"ok": True}
@app.get("/api/irs")
async def list_irs():
"""List available IR files."""
ir = self.deps.ir_loader
async def list_irs(channel: str = "guitar"):
"""List available IR files for the given channel."""
_pm, _pl, _nam, ir = self._channel_deps(channel)
if not ir:
raise HTTPException(status_code=503)
irs = ir.get_irs()
@@ -746,9 +935,9 @@ class WebServer:
}
@app.post("/api/irs/load")
async def load_ir(data: dict):
"""Load an IR file by path."""
ir = self.deps.ir_loader
async def load_ir(data: dict, channel: str = "guitar"):
"""Load an IR file by path for the given channel."""
_pm, _pl, _nam, ir = self._channel_deps(channel)
if not ir:
raise HTTPException(status_code=503)
path = data.get("path", "")
@@ -764,9 +953,9 @@ class WebServer:
return {"ok": True}
@app.post("/api/irs/unload")
async def unload_ir():
"""Unload the current IR."""
ir = self.deps.ir_loader
async def unload_ir(channel: str = "guitar"):
"""Unload the current IR for the given channel."""
_pm, _pl, _nam, ir = self._channel_deps(channel)
if not ir:
raise HTTPException(status_code=503)
ir.unload()
@@ -999,10 +1188,16 @@ class WebServer:
async def websocket_endpoint(ws: WebSocket):
await self._manager.connect(ws)
logger.info("WebSocket client connected (%d active)", self._manager.active_count)
# Track per-connection channel preference
_ws_channel: str = "guitar"
try:
# Send initial state snapshot on connect
state = self._gather_state()
await ws.send_text(json.dumps({"type": "connected", "state": state}))
# Send initial state snapshot on connect (combined view)
state = self._gather_state_combined()
await ws.send_text(json.dumps({
"type": "connected",
"state": state,
"channels": list(VALID_CHANNELS),
}))
while True:
raw = await ws.receive_text()
@@ -1012,8 +1207,20 @@ class WebServer:
if msg_type == "ping":
await ws.send_text(json.dumps({"type": "pong"}))
elif msg_type == "get_state":
state = self._gather_state()
await ws.send_text(json.dumps({"type": "state", "state": state}))
ch = msg.get("channel", _ws_channel)
_ws_channel = _resolve_channel(ch)
state = self._gather_state(channel=_ws_channel)
await ws.send_text(json.dumps({
"type": "state",
"channel": _ws_channel,
"state": state,
}))
elif msg_type == "set_channel":
_ws_channel = _resolve_channel(msg.get("channel", "guitar"))
await ws.send_text(json.dumps({
"type": "channel_set",
"channel": _ws_channel,
}))
except WebSocketDisconnect:
pass
except Exception as e:
@@ -1264,6 +1471,42 @@ class WebServer:
ir_loader.get_irs()
return {"ok": True, "path": str(path), "filename": path.name}
# ── Channel-prefixed route aliases ──────────────────────
# Every state/control endpoint that accepts ?channel= is also
# available at /api/channel/{channel}/... for explicit routing.
_ch_base = "/api/channel/{channel}"
# State
app.add_api_route(f"{_ch_base}/state", get_state, methods=["GET"])
# Presets
app.add_api_route(f"{_ch_base}/presets", list_presets, methods=["GET"])
app.add_api_route(f"{_ch_base}/presets/{{bank}}/{{program}}", get_preset, methods=["GET"]) # noqa: F821
app.add_api_route(f"{_ch_base}/presets/{{bank}}/{{program}}/activate", activate_preset, methods=["POST"]) # noqa: F821
app.add_api_route(f"{_ch_base}/presets/{{bank}}/{{program}}", save_preset, methods=["PUT"]) # noqa: F821
app.add_api_route(f"{_ch_base}/presets/{{bank}}/{{program}}", delete_preset, methods=["DELETE"]) # noqa: F821
# Snapshots
app.add_api_route(f"{_ch_base}/snapshots", list_snapshots, methods=["GET"])
app.add_api_route(f"{_ch_base}/snapshots/{{slot}}", get_snapshot, methods=["GET"])
app.add_api_route(f"{_ch_base}/snapshots/{{slot}}/save", save_snapshot, methods=["POST"])
app.add_api_route(f"{_ch_base}/snapshots/{{slot}}/recall", recall_snapshot, methods=["POST"])
app.add_api_route(f"{_ch_base}/snapshots/{{slot}}", update_snapshot, methods=["PUT"])
app.add_api_route(f"{_ch_base}/snapshots/{{slot}}", delete_snapshot, methods=["DELETE"])
# Controller
app.add_api_route(f"{_ch_base}/bypass", toggle_bypass, methods=["POST"])
app.add_api_route(f"{_ch_base}/tuner", toggle_tuner, methods=["POST"])
app.add_api_route(f"{_ch_base}/tuner/pitch", get_tuner_pitch, methods=["GET"])
app.add_api_route(f"{_ch_base}/volume", set_volume, methods=["PUT"])
app.add_api_route(f"{_ch_base}/routing", get_routing, methods=["GET"])
app.add_api_route(f"{_ch_base}/routing", set_routing, methods=["POST"])
# Models & IRs
app.add_api_route(f"{_ch_base}/models", list_models, methods=["GET"])
app.add_api_route(f"{_ch_base}/models/load", load_model, methods=["POST"])
app.add_api_route(f"{_ch_base}/models/unload", unload_model, methods=["POST"])
app.add_api_route(f"{_ch_base}/irs", list_irs, methods=["GET"])
app.add_api_route(f"{_ch_base}/irs/load", load_ir, methods=["POST"])
app.add_api_route(f"{_ch_base}/irs/unload", unload_ir, methods=["POST"])
return app
# ── Network/BT state gathering ──────────────────────────────────
@@ -1322,12 +1565,14 @@ class WebServer:
# ── State gathering ──────────────────────────────────────────────
def _gather_state(self) -> dict[str, Any]:
"""Collect a snapshot of the current pedal state."""
pm = self.deps.presets
pl = self.deps.pipeline
nam = self.deps.nam_host
ir = self.deps.ir_loader
def _gather_channel_state(self, channel: str) -> dict[str, Any]:
"""Collect a snapshot for a single channel.
Returns the full state dict for the requested channel. When the
channel's preset manager is not connected the dict has a flat
``connected: False`` marker.
"""
pm, pl, nam, ir = self._channel_deps(channel)
if not pm:
return {
@@ -1378,6 +1623,7 @@ class WebServer:
current_ir_name = _safe_str(current_ir_obj.name if current_ir_obj else None)
return {
"channel": channel,
"connected": True,
"current_preset": {
"name": preset.name if preset else "",
@@ -1398,15 +1644,74 @@ class WebServer:
"nam_model": current_model_name,
"ir_loaded": bool(ir and ir.is_loaded) if ir else False,
"ir_name": current_ir_name,
"channel_mode": self._channel_mgr.mode,
# Audio levels from pipeline (RMS float32 normalized 0.0-1.0, scaled to 0-100)
"input_level": min(100, round((pl._input_level or 0.0) * 150)) if pl else 0,
"output_level": min(100, round((pl._output_level or 0.0) * 150)) if pl else 0,
"input_level": (
min(100, round((getattr(pl, '_input_level', 0.0) or 0.0) * 150))
if pl and isinstance(getattr(pl, '_input_level', None), (int, float))
else 0
),
"output_level": (
min(100, round((getattr(pl, '_output_level', 0.0) or 0.0) * 150))
if pl and isinstance(getattr(pl, '_output_level', None), (int, float))
else 0
),
# System stats
"cpu_percent": self._gather_system_stats().get("cpu_percent", 0),
"sample_rate": SAMPLE_RATE,
"wifi": self._gather_wifi_state(),
"bluetooth": self._gather_bt_state(),
"current_snapshot": getattr(self.deps.presets, '_current_snapshot', 0) if self.deps.presets else 0,
"current_snapshot": getattr(pm, '_current_snapshot', 0) if pm else 0,
}
def _gather_state(self, channel: Optional[str] = None) -> dict[str, Any]:
"""Collect pedal state snapshot.
By default (``channel=None``) returns the guitar channel's flat
state — this preserves backward compatibility for existing clients
that expect ``connected``, ``current_preset``, etc. at the top
level.
Pass ``channel=\"guitar\"`` or ``channel=\"bass\"`` for explicit
per-channel state.
"""
ch = channel if channel is not None else "guitar"
return self._gather_channel_state(ch)
def _gather_state_combined(self) -> dict[str, Any]:
"""Return a combined view with both channel states.
Shape::
{
\"channels\": {
\"guitar\": { … },
\"bass\": { … },
},
\"active_channel\": \"guitar\",
\"channel_mode\": \"dual-mono\",
\"wifi\": { … },
\"bluetooth\": { … },
\"cpu_percent\": …,
}
"""
guitar = self._gather_channel_state("guitar")
bass = self._gather_channel_state("bass")
return {
"channels": {
"guitar": guitar,
"bass": bass,
},
"active_channel": "guitar",
"channel_mode": self._channel_mgr.mode,
"wifi": guitar.get("wifi", {}),
"bluetooth": guitar.get("bluetooth", {}),
"cpu_percent": max(
guitar.get("cpu_percent", 0),
bass.get("cpu_percent", 0),
),
"sample_rate": SAMPLE_RATE,
}
# ── Tone3000 client lazy init ────────────────────────────────
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
+1 -1
View File
@@ -5,7 +5,7 @@
<link rel="icon" type="image/svg+xml" href="/ui/favicon.svg" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>Pi Multi-FX Pedal</title>
<script type="module" crossorigin src="/ui/assets/index-DKZHEo9m.js"></script>
<script type="module" crossorigin src="/ui/assets/index-BiNA-JeT.js"></script>
<link rel="stylesheet" crossorigin href="/ui/assets/index-CuJgR-s6.css">
</head>
<body>
+148 -4
View File
@@ -27,7 +27,7 @@ from src.presets.manager import (
_preset_from_dict,
_preset_to_dict,
)
from src.presets.types import FXBlock, FXType, MIDIMapping, Preset
from src.presets.types import Channel, FXBlock, FXType, MIDIMapping, Preset
# ── Fixtures ────────────────────────────────────────────────────────────────
@@ -152,7 +152,7 @@ class TestSaveLoad:
def test_save_creates_json_file(self, manager: PresetManager, sample_preset: Preset, preset_dir: Path):
manager.save(sample_preset)
path = preset_dir / "bank_0" / "preset_0.json"
path = preset_dir / "guitar" / "bank_0" / "preset_0.json"
assert path.exists()
content = path.read_text(encoding="utf-8")
assert "Crunch Rhythm" in content
@@ -513,8 +513,152 @@ class TestEdgeCases:
assert manager.current_program == 1
def test_corrupted_json_raises(self, manager: PresetManager, preset_dir: Path):
path = preset_dir / "bank_0" / "preset_0.json"
path = preset_dir / "guitar" / "bank_0" / "preset_0.json"
path.parent.mkdir(parents=True)
path.write_text("not valid json", encoding="utf-8")
with pytest.raises(json.JSONDecodeError):
manager.load(0, 0)
manager.load(0, 0)
# ── Channel independence ────────────────────────────────────────────────
class TestChannelIndependence:
"""Guitar and bass channels each have their own preset bank namespace."""
def test_channel_serialization_round_trip(self):
"""Channel field survives to_dict / from_dict."""
p = Preset(name="GTR Lead", channel=Channel.GUITAR, bank=0, program=0)
d = _preset_to_dict(p)
assert d["channel"] == "guitar"
restored = _preset_from_dict(d)
assert restored.channel == Channel.GUITAR
p2 = Preset(name="Bass Slap", channel=Channel.BASS, bank=0, program=0)
d2 = _preset_to_dict(p2)
assert d2["channel"] == "bass"
restored2 = _preset_from_dict(d2)
assert restored2.channel == Channel.BASS
def test_channel_defaults_to_guitar(self):
"""Presets without explicit channel default to guitar."""
p = Preset(name="Default", bank=0, program=0)
assert p.channel == Channel.GUITAR
def test_guitar_and_bass_independent_banks(self, manager: PresetManager):
"""Guitar and bass presets in the same (bank, program) don't collide."""
gtr = Preset(name="GTR Crunch", channel=Channel.GUITAR, bank=0, program=0)
bass = Preset(name="Bass Slap", channel=Channel.BASS, bank=0, program=0)
# Save guitar via current channel (defaults to guitar)
manager.save(gtr)
# Save bass explicitly
manager.save(bass, channel=Channel.BASS)
# Load guitar
loaded_gtr = manager.load(0, 0, Channel.GUITAR)
assert loaded_gtr.name == "GTR Crunch"
assert loaded_gtr.channel == Channel.GUITAR
# Load bass
loaded_bass = manager.load(0, 0, Channel.BASS)
assert loaded_bass.name == "Bass Slap"
assert loaded_bass.channel == Channel.BASS
def test_channel_switch_restores_preset(self, manager: PresetManager):
"""Switching channels restores the last active state for that channel."""
manager.save(Preset(name="GTR Lead", channel=Channel.GUITAR, bank=0, program=0))
manager.save(Preset(name="GTR Clean", channel=Channel.GUITAR, bank=0, program=1))
manager.save(Preset(name="Bass Slap", channel=Channel.BASS, bank=1, program=2))
# Select guitar preset (bank=0, program=1)
manager.select(0, 1, Channel.GUITAR)
# Select bass preset so its channel state is saved
manager.select(1, 2, Channel.BASS)
# Switch to guitar — should go to (0, 1)
manager.set_channel(Channel.GUITAR)
assert manager.current_channel == Channel.GUITAR
assert manager.current_bank == 0
assert manager.current_program == 1
# Switch back to bass — should restore (1, 2)
manager.set_channel(Channel.BASS)
assert manager.current_channel == Channel.BASS
assert manager.current_bank == 1
assert manager.current_program == 2
def test_set_same_channel_noop(self, manager: PresetManager):
"""set_channel to the same channel preserves current position."""
manager.save(Preset(name="Only", channel=Channel.GUITAR, bank=0, program=2))
manager.select(0, 2, Channel.GUITAR)
result = manager.set_channel(Channel.GUITAR)
assert result.name == "Only"
assert manager.current_bank == 0
assert manager.current_program == 2
def test_current_channel_property(self, manager: PresetManager):
"""current_channel returns the active channel."""
assert manager.current_channel == Channel.GUITAR
manager.set_channel(Channel.BASS)
assert manager.current_channel == Channel.BASS
manager.set_channel(Channel.GUITAR)
assert manager.current_channel == Channel.GUITAR
def test_list_banks_channel_scoped(self, manager: PresetManager):
"""list_banks only returns banks for the specified channel."""
manager.save(Preset(name="GTR_A", channel=Channel.GUITAR, bank=0, program=0))
manager.save(Preset(name="BASS_A", channel=Channel.BASS, bank=5, program=0))
gtr_banks = manager.list_banks(Channel.GUITAR)
assert len(gtr_banks) == 1
assert gtr_banks[0].number == 0
bass_banks = manager.list_banks(Channel.BASS)
assert len(bass_banks) == 1
assert bass_banks[0].number == 5
def test_migrate_legacy_presets(self, tmp_path: Path):
"""Legacy flat bank_* dirs are migrated to guitar/ on first boot."""
# Create legacy layout: bank_0/preset_0.json at root
legacy_bank = tmp_path / "bank_0"
legacy_bank.mkdir()
(legacy_bank / "preset_0.json").write_text(
json.dumps({"name": "Legacy Tone", "bank": 0, "program": 0}),
encoding="utf-8",
)
# Create PresetManager — should migrate
pm = PresetManager(preset_dir=str(tmp_path), audio_pipeline=None)
# Old location should be gone
assert not (tmp_path / "bank_0").exists()
# New location should exist
guitar_path = tmp_path / "guitar" / "bank_0" / "preset_0.json"
assert guitar_path.exists()
# Preset should be loadable from guitar channel
preset = pm.load(0, 0, Channel.GUITAR)
assert preset.name == "Legacy Tone"
def test_migration_skips_when_already_migrated(self, tmp_path: Path):
"""Migration doesn't run if guitar/ dir already exists."""
guitar_dir = tmp_path / "guitar"
guitar_dir.mkdir(parents=True)
(guitar_dir / "bank_0").mkdir()
(guitar_dir / "bank_0" / "preset_0.json").write_text(
json.dumps({"name": "Already Migrated", "channel": "guitar", "bank": 0, "program": 0}),
encoding="utf-8",
)
pm = PresetManager(preset_dir=str(tmp_path), audio_pipeline=None)
preset = pm.load(0, 0, Channel.GUITAR)
assert preset.name == "Already Migrated"
def test_fresh_install_no_migration(self, tmp_path: Path):
"""Fresh install with no files doesn't raise or create guitar dir."""
pm = PresetManager(preset_dir=str(tmp_path), audio_pipeline=None)
assert not (tmp_path / "guitar").exists()
assert pm.list_banks() == []
+6
View File
@@ -32,6 +32,7 @@ def mock_presets():
pm = MagicMock()
pm.current_bank = 0
pm.current_program = 0
pm._current_snapshot = 0
pm.load.return_value = Preset(name="Default", bank=0, program=0)
return pm
@@ -41,6 +42,11 @@ def mock_pipeline():
pl = MagicMock()
pl._bypassed = False
pl._tuner_enabled = False
pl._tuner_frequency = 0.0
pl._tuner_note = '--'
pl._tuner_cents = 0
pl._tuner_string = -1
pl._tuner_confidence = 0.0
pl._master_volume = 0.8
pl.routing_mode = "mono"
pl.routing_breakpoint = 7