shawn
659e53766a
Root cause: factory presets for bass (banks 17-20) sat in the root factory dir alongside guitar presets, and install_factory_presets() used rglob() which picked up ALL presets into the GUITAR channel. Changes: - Move bank_17-20 from presets/factory/ → presets/factory/bass/ (matching the existing keys/, vocals/, backing_tracks/ layout) - Fix install_factory_presets() to scan channel-specific subdirectory: GUITAR scans root bank_* dirs, all others scan <channel>/bank_* - Fix main.py to iterate over all 5 channels when installing factory presets (was only installing into GUITAR) - Import Channel enum in main.py Tests: 56/56 preset manager tests pass. Integration verified: - GUITAR: 68 presets (banks 0-16) - BASS: 16 presets (banks 17-20, SansAmp/Darkglass/Ampeg SVT) - KEYS: 12 presets (banks 21-23) - VOCALS: 12 presets (banks 24-26) - BACKING_TRACKS: 12 presets (banks 27-29)
925 lines
35 KiB
Python
925 lines
35 KiB
Python
"""Preset and bank manager — save, load, navigate, and auto-restore presets.
|
||
|
||
The PresetManager owns the on-disk preset store and provides:
|
||
- Human-readable JSON preset storage
|
||
- Bank structure with 4 presets per bank
|
||
- Footswitch navigation (preset-up/down wraps within bank, bank-up/down switches banks)
|
||
- MIDI Program Change routing (bank=channel, program=preset)
|
||
- Auto-restore of last active preset on power cycle
|
||
- Factory preset installation
|
||
- Preset rename and reorder
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import shutil
|
||
import threading
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
from .types import Bank, Channel, FXBlock, FXType, MIDIMapping, Preset, Snapshot
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── Constants ───────────────────────────────────────────────────────────────
|
||
|
||
PRESETS_PER_BANK = 4
|
||
"""Number of presets per bank (convention: footswitch 1-4 within a bank)."""
|
||
|
||
AUTO_SAVE_DELAY_S = 1.0
|
||
"""Debounce delay in seconds before writing auto-save state to disk."""
|
||
|
||
FACTORY_PRESET_DIR = Path(__file__).resolve().parent.parent.parent / "presets" / "factory"
|
||
"""Location of bundled factory preset JSON files."""
|
||
|
||
FACTORY_IR_DIR = FACTORY_PRESET_DIR / "irs"
|
||
"""Location of bundled factory cabinet IR .wav files."""
|
||
|
||
|
||
# ── Serialisation helpers ───────────────────────────────────────────────────
|
||
|
||
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,
|
||
"routing_mode": preset.routing_mode,
|
||
"routing_breakpoint": preset.routing_breakpoint,
|
||
"tuner_enabled": preset.tuner_enabled,
|
||
"chain": [
|
||
{
|
||
"fx_type": block.fx_type.value,
|
||
"enabled": block.enabled,
|
||
"bypass": block.bypass,
|
||
"params": dict(block.params),
|
||
"nam_model_path": block.nam_model_path,
|
||
"ir_file_path": block.ir_file_path,
|
||
}
|
||
for block in preset.chain
|
||
],
|
||
"midi_mappings": {
|
||
key: {
|
||
"cc_number": mapping.cc_number,
|
||
"channel": mapping.channel,
|
||
"min_val": mapping.min_val,
|
||
"max_val": mapping.max_val,
|
||
}
|
||
for key, mapping in preset.midi_mappings.items()
|
||
},
|
||
"snapshots": {
|
||
str(slot): {
|
||
"name": snap.name,
|
||
"master_volume": snap.master_volume,
|
||
"chain": [
|
||
{
|
||
"fx_type": block.fx_type.value,
|
||
"enabled": block.enabled,
|
||
"bypass": block.bypass,
|
||
"params": dict(block.params),
|
||
"nam_model_path": block.nam_model_path,
|
||
"ir_file_path": block.ir_file_path,
|
||
}
|
||
for block in snap.chain
|
||
],
|
||
}
|
||
for slot, snap in preset.snapshots.items()
|
||
},
|
||
}
|
||
|
||
|
||
def _preset_from_dict(data: dict) -> Preset:
|
||
"""Deserialize a Preset from a JSON-compatible dict."""
|
||
chain = []
|
||
for block_data in data.get("chain", []):
|
||
chain.append(
|
||
FXBlock(
|
||
fx_type=FXType(block_data["fx_type"]),
|
||
enabled=block_data.get("enabled", True),
|
||
bypass=block_data.get("bypass", False),
|
||
params=dict(block_data.get("params", {})),
|
||
nam_model_path=block_data.get("nam_model_path", ""),
|
||
ir_file_path=block_data.get("ir_file_path", ""),
|
||
)
|
||
)
|
||
|
||
midi_mappings = {}
|
||
for key, md in data.get("midi_mappings", {}).items():
|
||
midi_mappings[key] = MIDIMapping(
|
||
cc_number=md.get("cc_number", 0),
|
||
channel=md.get("channel", 0),
|
||
min_val=md.get("min_val", 0.0),
|
||
max_val=md.get("max_val", 1.0),
|
||
)
|
||
|
||
snapshots = {}
|
||
for slot_str, snap_data in data.get("snapshots", {}).items():
|
||
snap_chain = []
|
||
for bd in snap_data.get("chain", []):
|
||
snap_chain.append(
|
||
FXBlock(
|
||
fx_type=FXType(bd["fx_type"]),
|
||
enabled=bd.get("enabled", True),
|
||
bypass=bd.get("bypass", False),
|
||
params=dict(bd.get("params", {})),
|
||
nam_model_path=bd.get("nam_model_path", ""),
|
||
ir_file_path=bd.get("ir_file_path", ""),
|
||
)
|
||
)
|
||
snapshots[int(slot_str)] = Snapshot(
|
||
name=snap_data.get("name", f"Snapshot {slot_str}"),
|
||
chain=snap_chain,
|
||
master_volume=snap_data.get("master_volume", 0.8),
|
||
)
|
||
|
||
return Preset(
|
||
name=data["name"],
|
||
channel=Channel(data.get("channel", "guitar")),
|
||
bank=data.get("bank", 0),
|
||
program=data.get("program", 0),
|
||
chain=chain,
|
||
midi_mappings=midi_mappings,
|
||
master_volume=data.get("master_volume", 0.8),
|
||
routing_mode=data.get("routing_mode", "mono"),
|
||
routing_breakpoint=data.get("routing_breakpoint", 7),
|
||
tuner_enabled=data.get("tuner_enabled", False),
|
||
snapshots=snapshots,
|
||
)
|
||
|
||
|
||
# ── The PresetManager ───────────────────────────────────────────────────────
|
||
|
||
|
||
class PresetManager:
|
||
"""Manages preset/bank storage, navigation, MIDI binding, and auto-save.
|
||
|
||
Args:
|
||
preset_dir: Directory for the preset store (created if missing).
|
||
audio_pipeline: Optional AudioPipeline to activate presets on.
|
||
When set, ``activate()`` calls ``pipeline.load_preset()``.
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
preset_dir: str | Path = "~/.pedal/presets",
|
||
audio_pipeline: object = None,
|
||
) -> None:
|
||
self._dir = Path(preset_dir).expanduser().resolve()
|
||
self._dir.mkdir(parents=True, exist_ok=True)
|
||
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
|
||
self._lock = threading.Lock()
|
||
|
||
# Register of known bank numbers (populated lazily)
|
||
self._known_banks: set[int] = set()
|
||
|
||
# Auto-save debounce timer
|
||
self._auto_save_timer: Optional[threading.Timer] = None
|
||
|
||
# Current snapshot slot (0 = none selected)
|
||
self._current_snapshot: int = 0
|
||
|
||
# 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
|
||
|
||
@property
|
||
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 (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).
|
||
|
||
Returns:
|
||
The activated Preset.
|
||
"""
|
||
bank = self._get_or_create_bank(self._current_bank)
|
||
prog = (self._current_program + 1) % PRESETS_PER_BANK
|
||
return self._select_and_activate(self._current_bank, prog)
|
||
|
||
def preset_down(self) -> Preset:
|
||
"""Select previous preset in current bank (wraps 0→3).
|
||
|
||
Returns:
|
||
The activated Preset.
|
||
"""
|
||
prog = (self._current_program - 1) % PRESETS_PER_BANK
|
||
return self._select_and_activate(self._current_bank, prog)
|
||
|
||
def bank_up(self) -> tuple[Bank, Preset]:
|
||
"""Move to the next higher-numbered bank.
|
||
|
||
If there are no banks above the current one, wraps to the
|
||
lowest known bank number.
|
||
|
||
Returns:
|
||
(Bank, Preset) — the new bank and its currently active preset.
|
||
"""
|
||
return self._switch_bank(+1)
|
||
|
||
def bank_down(self) -> tuple[Bank, Preset]:
|
||
"""Move to the next lower-numbered bank (wraps around).
|
||
|
||
Returns:
|
||
(Bank, Preset) — the new bank and its currently active preset.
|
||
"""
|
||
return self._switch_bank(-1)
|
||
|
||
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, channel or self._current_channel)
|
||
|
||
def midi_pc(self, channel: int, program: int) -> Preset:
|
||
"""Handle MIDI Program Change: bank=channel, program=program.
|
||
|
||
Args:
|
||
channel: MIDI channel (used as bank number).
|
||
program: MIDI program number (used as preset index).
|
||
|
||
Returns:
|
||
The activated Preset.
|
||
"""
|
||
logger.info("MIDI PC: bank=%d, program=%d", channel, program)
|
||
return self._select_and_activate(channel, program)
|
||
|
||
# ── CRUD ────────────────────────────────────────────────────────────────
|
||
|
||
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).
|
||
"""
|
||
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),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
if not auto:
|
||
self._dirty = True
|
||
self._known_banks.add(preset.bank)
|
||
|
||
# Save/update bank metadata
|
||
self._save_bank_meta(ch, preset.bank)
|
||
|
||
logger.info("Saved preset '%s' → %s", preset.name, path)
|
||
|
||
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.
|
||
|
||
Raises:
|
||
FileNotFoundError: If the preset slot doesn't exist.
|
||
json.JSONDecodeError: If the file is corrupt.
|
||
"""
|
||
ch = channel or self._current_channel
|
||
path = self._preset_path(ch, bank, program)
|
||
if not path.exists():
|
||
raise FileNotFoundError(
|
||
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, 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 (channel=%s, bank=%d, program=%d)", ch.value, bank, program)
|
||
self._dirty = True
|
||
|
||
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, channel)
|
||
preset.name = 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:
|
||
"""Move a preset to a different slot within the same bank.
|
||
|
||
If the target slot is occupied the two presets are swapped.
|
||
If the current active preset is being moved, tracking is updated.
|
||
|
||
Args:
|
||
bank: Bank number.
|
||
program: Current program index.
|
||
new_program: Target program index (0-3).
|
||
"""
|
||
if program == new_program:
|
||
return
|
||
if not (0 <= new_program < PRESETS_PER_BANK):
|
||
raise ValueError(f"Program index {new_program} out of range [0, {PRESETS_PER_BANK})")
|
||
|
||
# Load both existing presets (or create empty placeholder for empty slot)
|
||
try:
|
||
src = self.load(bank, program)
|
||
except FileNotFoundError:
|
||
raise FileNotFoundError(f"Cannot reorder: no preset at (bank={bank}, program={program})")
|
||
|
||
try:
|
||
dst = self.load(bank, new_program)
|
||
src.program = new_program
|
||
dst.program = program
|
||
self.save(dst)
|
||
except FileNotFoundError:
|
||
src.program = new_program
|
||
|
||
self.save(src)
|
||
|
||
# Update current tracking if we moved the active preset
|
||
with self._lock:
|
||
if self._current_bank == bank and self._current_program == program:
|
||
self._current_program = new_program
|
||
|
||
logger.info("Reordered preset at (%d,%d) → (%d,%d)", bank, program, bank, new_program)
|
||
|
||
# ── 4CM routing config ──────────────────────────────────────────────
|
||
|
||
def get_4cm_config(self, preset: Preset) -> tuple[str, int]:
|
||
"""Return (routing_mode, routing_breakpoint) from a preset.
|
||
|
||
Returns:
|
||
Tuple of ``(routing_mode, routing_breakpoint)``.
|
||
"""
|
||
return (preset.routing_mode, preset.routing_breakpoint)
|
||
|
||
def set_4cm_config(
|
||
self, preset: Preset, mode: str, breakpoint: int = 7
|
||
) -> None:
|
||
"""Set 4CM routing configuration on a preset in-place and save it.
|
||
|
||
Args:
|
||
preset: The preset to modify.
|
||
mode: ``\"mono\"`` or ``\"4cm\"``.
|
||
breakpoint: Chain index where the pre/post split occurs
|
||
(default 7 = after IR_CAB, before mod FX).
|
||
|
||
Raises:
|
||
ValueError: If ``mode`` is not ``\"mono\"`` or ``\"4cm\"``.
|
||
"""
|
||
if mode not in ("mono", "4cm"):
|
||
raise ValueError(f"routing_mode must be 'mono' or '4cm', got {mode!r}")
|
||
preset.routing_mode = mode
|
||
preset.routing_breakpoint = breakpoint
|
||
self.save(preset)
|
||
logger.info("Set 4CM config on '%s': mode=%s breakpoint=%d",
|
||
preset.name, mode, breakpoint)
|
||
|
||
# ── Bank management ─────────────────────────────────────────────────────
|
||
|
||
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(ch_dir.glob("bank_*/bank.json")):
|
||
try:
|
||
data = json.loads(meta_path.read_text(encoding="utf-8"))
|
||
num = data.get("number", 0)
|
||
preset_count = sum(
|
||
1 for _ in meta_path.parent.glob("preset_*.json")
|
||
)
|
||
banks[num] = Bank(
|
||
name=data.get("name", f"Bank {num}"),
|
||
number=num,
|
||
presets=[None] * preset_count, # Placeholder
|
||
)
|
||
except (json.JSONDecodeError, KeyError):
|
||
continue
|
||
|
||
# Also discover banks by scanning dirs without bank.json
|
||
for p_dir in sorted(ch_dir.glob("bank_*")):
|
||
try:
|
||
num = int(p_dir.name.split("_")[1])
|
||
if num not in banks:
|
||
banks[num] = Bank(name=f"Bank {num}", number=num)
|
||
except (ValueError, IndexError):
|
||
continue
|
||
|
||
self._known_banks = set(banks.keys())
|
||
return [banks[k] for k in sorted(banks)]
|
||
|
||
def get_or_create_bank(self, number: int, name: str = "") -> Bank:
|
||
"""Return an existing bank or create a new one.
|
||
|
||
Args:
|
||
number: Bank number.
|
||
name: Optional display name (defaults to f"Bank {number}").
|
||
|
||
Returns:
|
||
The Bank object.
|
||
"""
|
||
return self._get_or_create_bank(number, name)
|
||
|
||
# ── Auto-restore ────────────────────────────────────────────────────────
|
||
|
||
def save_state(self) -> None:
|
||
"""Persist the current (channel, bank, program) for power-cycle restore.
|
||
|
||
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_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: 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 (channel, bank, program) from state.json.
|
||
|
||
Returns:
|
||
The restored Preset, or None if no state file exists or the
|
||
saved slot is empty.
|
||
"""
|
||
state_path = self._dir / "state.json"
|
||
if not state_path.exists():
|
||
logger.info("No saved state to restore")
|
||
return None
|
||
|
||
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, 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: channel=%s bank=%d program=%d '%s'",
|
||
ch.value, bank, program, preset.name)
|
||
return preset
|
||
|
||
# ── Activation ──────────────────────────────────────────────────────────
|
||
|
||
def activate(self, preset: Preset) -> None:
|
||
"""Apply a preset to the audio pipeline (if one is connected).
|
||
|
||
Args:
|
||
preset: The preset to activate.
|
||
"""
|
||
if self._pipeline is not None:
|
||
self._pipeline.load_preset(preset)
|
||
logger.info("Activated preset '%s' (bank=%d, program=%d)",
|
||
preset.name, preset.bank, preset.program)
|
||
|
||
# ── Factory presets ─────────────────────────────────────────────────────
|
||
|
||
def install_factory_presets(self, overwrite: bool = False,
|
||
channel: Channel = Channel.GUITAR) -> int:
|
||
"""Install bundled factory presets into the preset store.
|
||
|
||
Scans the channel-specific subdirectory of ``FACTORY_PRESET_DIR``
|
||
for ``bank_*/preset_*.json`` files and copies them into the user
|
||
preset store. For GUITAR the root of ``FACTORY_PRESET_DIR`` is
|
||
scanned (backward-compatible with legacy ``bank_0``–``bank_16``).
|
||
All other channels use ``FACTORY_PRESET_DIR / channel.value``.
|
||
|
||
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.
|
||
"""
|
||
if not FACTORY_PRESET_DIR.is_dir():
|
||
logger.warning("Factory preset directory not found: %s", FACTORY_PRESET_DIR)
|
||
return 0
|
||
|
||
# Determine source directory — root for GUITAR, subdir for all others
|
||
source_dir = FACTORY_PRESET_DIR if channel == Channel.GUITAR \
|
||
else FACTORY_PRESET_DIR / channel.value
|
||
|
||
if not source_dir.is_dir():
|
||
logger.warning("No factory presets for channel %s: %s",
|
||
channel.value, source_dir)
|
||
return 0
|
||
|
||
count = 0
|
||
for src in sorted(source_dir.glob("bank_*/preset_*.json")):
|
||
# Derive (bank, program) from file path
|
||
bank_dir = src.parent
|
||
try:
|
||
bank_num = int(bank_dir.name.split("_")[1])
|
||
prog_num = int(src.stem.split("_")[1])
|
||
except (ValueError, IndexError):
|
||
logger.warning("Skipping malformed factory preset path: %s", src)
|
||
continue
|
||
|
||
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(channel, bank_num)
|
||
count += 1
|
||
logger.debug("Installed factory preset: %s", dest)
|
||
|
||
if count:
|
||
self._dirty = True
|
||
logger.info("Installed %d factory presets for %s", count, channel.value)
|
||
return count
|
||
|
||
def install_factory_irs(self, overwrite: bool = False) -> int:
|
||
"""Install bundled factory cabinet IR .wav files into the IR directory.
|
||
|
||
Copies IR files from ``FACTORY_IR_DIR`` into the IR loader's default
|
||
directory (``~/.pedal/irs/``). Existing files are skipped unless
|
||
``overwrite=True``.
|
||
|
||
Args:
|
||
overwrite: If True, overwrite existing IR files with the same name.
|
||
|
||
Returns:
|
||
Number of factory IR files installed.
|
||
"""
|
||
if not FACTORY_IR_DIR.is_dir():
|
||
logger.warning("Factory IR directory not found: %s", FACTORY_IR_DIR)
|
||
return 0
|
||
|
||
from ..dsp.ir_loader import DEFAULT_IR_DIR
|
||
dest = DEFAULT_IR_DIR
|
||
dest.mkdir(parents=True, exist_ok=True)
|
||
|
||
count = 0
|
||
for src in sorted(FACTORY_IR_DIR.glob("*.wav")):
|
||
dst = dest / src.name
|
||
if dst.exists() and not overwrite:
|
||
logger.debug("Skipping existing IR: %s", dst.name)
|
||
continue
|
||
shutil.copy2(src, dst)
|
||
count += 1
|
||
logger.debug("Installed factory IR: %s", dst.name)
|
||
|
||
if count:
|
||
logger.info("Installed %d factory cab IRs into %s", count, dest)
|
||
return count
|
||
|
||
# ── Internal helpers ────────────────────────────────────────────────────
|
||
|
||
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 _channel_dir(self, channel: Channel) -> Path:
|
||
"""Return the root directory for a given channel."""
|
||
return self._dir / channel.value
|
||
|
||
def _bank_dir(self, channel: Channel, bank: int) -> Path:
|
||
return self._channel_dir(channel) / f"bank_{bank}"
|
||
|
||
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"))
|
||
return Bank(
|
||
name=data.get("name", name or f"Bank {number}"),
|
||
number=data.get("number", number),
|
||
)
|
||
except (json.JSONDecodeError, KeyError):
|
||
pass
|
||
|
||
# Create new bank
|
||
bank = Bank(name=name or f"Bank {number}", number=number)
|
||
self._save_bank_meta(ch, number, bank.name)
|
||
self._known_banks.add(number)
|
||
logger.info("Created bank %d: '%s' (channel=%s)", number, bank.name, ch.value)
|
||
return bank
|
||
|
||
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(channel, bank_num)
|
||
existing = {}
|
||
if meta_path.exists():
|
||
try:
|
||
existing = json.loads(meta_path.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
meta = {
|
||
"number": bank_num,
|
||
"name": name or existing.get("name", f"Bank {bank_num}"),
|
||
"preset_count": PRESETS_PER_BANK,
|
||
}
|
||
meta_path.parent.mkdir(parents=True, exist_ok=True)
|
||
meta_path.write_text(json.dumps(meta, indent=2), encoding="utf-8")
|
||
|
||
def _switch_bank(self, direction: int) -> tuple[Bank, Preset]:
|
||
"""Move +1 or -1 through known banks, wrapping at edges.
|
||
|
||
Args:
|
||
direction: +1 for bank_up, -1 for bank_down.
|
||
|
||
Returns:
|
||
(Bank, Preset) of the new active slot.
|
||
"""
|
||
banks = self.list_banks()
|
||
if not banks:
|
||
raise RuntimeError("No banks available")
|
||
|
||
current_idx = next(
|
||
(i for i, b in enumerate(banks) if b.number == self._current_bank),
|
||
None,
|
||
)
|
||
if current_idx is None:
|
||
current_idx = 0
|
||
|
||
new_idx = (current_idx + direction) % len(banks)
|
||
new_bank = banks[new_idx]
|
||
|
||
# Activate preset at same program index within the new bank,
|
||
# defaulting to a blank preset if the slot is empty
|
||
try:
|
||
preset = self.load(new_bank.number, self._current_program)
|
||
except FileNotFoundError:
|
||
preset = Preset(
|
||
name=f"Empty {new_bank.name}",
|
||
channel=self._current_channel,
|
||
bank=new_bank.number,
|
||
program=self._current_program,
|
||
)
|
||
|
||
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' (channel=%s)",
|
||
new_bank.number, preset.name, self._current_channel.value)
|
||
return new_bank, 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.
|
||
|
||
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, ch)
|
||
except FileNotFoundError:
|
||
# Create a blank placeholder preset
|
||
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
|
||
|
||
# ── 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") |