shawn
6008776e3c
The profile change handler was calling nam_host.set_block_size() AFTER jack_client.start(), creating a race where the JACK callback fired with the new buffer size while the old NAM subprocess was still running at the old block_size. Pipe byte counts desynced => NAM returned passthrough. + also: auto_connect disable during restart, connect_fx_ports reorder, Tone3000 gear/sort/pagination params, NAM engine switch API endpoints
3021 lines
141 KiB
Python
3021 lines
141 KiB
Python
"""FastAPI web server for Pi Multi-FX Pedal control interface.
|
||
|
||
Provides REST endpoints and WebSocket for real-time bidirectional
|
||
control of the pedal from any device on the LAN.
|
||
|
||
Run standalone for development:
|
||
python -m src.web.server
|
||
|
||
Or embed in main.py via WebServerDepss.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import logging
|
||
import math
|
||
import mimetypes
|
||
import datetime
|
||
from contextlib import asynccontextmanager
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Any, Optional
|
||
|
||
import uvicorn
|
||
from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect, Request
|
||
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
|
||
from fastapi.staticfiles import StaticFiles
|
||
|
||
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 Channel, FXBlock, FXType, Preset
|
||
from ..system.tonedownload import (
|
||
Tone3000Client,
|
||
ModelResult,
|
||
IRResult,
|
||
discover_anon_key,
|
||
format_size,
|
||
)
|
||
from ..system.audio import AudioSystem, JackAudioClient, LATENCY_PROFILES
|
||
from ..system.config import save_config
|
||
from ..ui.leds import LEDController
|
||
from ..ui.display import DisplayController
|
||
from ..midi.handler import MIDIHandler
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── Cache buster version (bumped on deploy) ─────────────────────────
|
||
# Increment this when template/static changes need to bypass browser cache
|
||
# ── Defaults ─────────────────────────────────────────────────────────────────
|
||
|
||
DEFAULT_HOST = "0.0.0.0"
|
||
DEFAULT_PORT = 80
|
||
STATIC_DIR = Path(__file__).resolve().parent / "static"
|
||
|
||
# v2 UI dist (the only UI we serve)
|
||
V2_DIST_DIR = Path(__file__).resolve().parent / "ui-v2-dist"
|
||
|
||
# ── 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 multi-channel vs stereo routing mode and per-channel state.
|
||
|
||
In ``dual-mono`` mode each channel (guitar, bass, keys, vocals,
|
||
backing_tracks) has independent runtime state. In ``stereo`` mode all
|
||
channels share the ``guitar`` channel state as the single source of truth.
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self.mode: str = _load_channel_mode()
|
||
self._channels: dict[str, ChannelState] = {
|
||
"guitar": ChannelState(),
|
||
"bass": ChannelState(),
|
||
"keys": ChannelState(),
|
||
"vocals": ChannelState(),
|
||
"backing_tracks": 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 a valid channel name,
|
||
otherwise falls back to ``"guitar"``.
|
||
"""
|
||
if self.mode == "stereo":
|
||
return "guitar"
|
||
if param in VALID_CHANNELS:
|
||
return param
|
||
return "guitar"
|
||
|
||
|
||
# ── Dependency container ─────────────────────────────────────────────────────
|
||
|
||
|
||
@dataclass
|
||
class WebServerDeps:
|
||
"""Dependencies injected into the web server from the host pedal app.
|
||
|
||
Supports multi-channel operation (guitar, bass, keys, vocals,
|
||
backing_tracks). Each channel can have its own preset manager, audio
|
||
pipeline, NAM host, and IR loader. Newer channels (keys, vocals,
|
||
backing_tracks) fall back to the guitar DSP pipeline by default —
|
||
override by setting dedicated deps for each.
|
||
|
||
The guitar fields are the defaults — callers that don't need multi-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
|
||
|
||
# Audio system (for JACK restart on profile change)
|
||
audio_system: Optional[AudioSystem] = None
|
||
# JACK audio client (needs restart when JACK restarts)
|
||
jack_audio: Optional[JackAudioClient] = None
|
||
|
||
# Full config dict (for saving profile/rate/period changes to disk)
|
||
config: Optional[dict] = None
|
||
config_path: Optional[Path] = None
|
||
|
||
# Runtime subsystems (for live apply of settings changes)
|
||
leds: Optional[LEDController] = None
|
||
midi: Optional[MIDIHandler] = None
|
||
display: Optional[DisplayController] = None
|
||
|
||
@property
|
||
def is_connected(self) -> bool:
|
||
"""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.
|
||
|
||
Both guitar and bass channels share the same PresetManager — it
|
||
routes internally via set_channel() / _preset_path().
|
||
Falls back to guitar fields for unrecognised channels.
|
||
"""
|
||
return (self.presets, self.pipeline, self.nam_host, self.ir_loader)
|
||
|
||
|
||
# ── WebSocket connection manager ─────────────────────────────────────────────
|
||
|
||
|
||
class ConnectionManager:
|
||
"""Manages active WebSocket connections for real-time push."""
|
||
|
||
def __init__(self) -> None:
|
||
self._connections: set[WebSocket] = set()
|
||
|
||
async def connect(self, ws: WebSocket) -> None:
|
||
await ws.accept()
|
||
self._connections.add(ws)
|
||
|
||
def disconnect(self, ws: WebSocket) -> None:
|
||
self._connections.discard(ws)
|
||
|
||
async def broadcast(self, message: dict[str, Any]) -> None:
|
||
"""Send a JSON message to every connected client."""
|
||
payload = json.dumps(message)
|
||
stale: list[WebSocket] = []
|
||
for ws in self._connections:
|
||
try:
|
||
await ws.send_text(payload)
|
||
except Exception:
|
||
stale.append(ws)
|
||
for ws in stale:
|
||
self._connections.discard(ws)
|
||
|
||
@property
|
||
def active_count(self) -> int:
|
||
return len(self._connections)
|
||
|
||
|
||
# ── The WebServer ────────────────────────────────────────────────────────────
|
||
|
||
|
||
class WebServer:
|
||
"""FastAPI web app wrapping REST + WebSocket endpoints for pedal control.
|
||
|
||
Typical usage from PedalApp.boot():
|
||
self.web = WebServer(WebServerDeps(presets=..., pipeline=..., ...))
|
||
self.web.start()
|
||
|
||
The server runs in a background thread to avoid blocking the JACK audio loop.
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
deps: WebServerDeps,
|
||
host: str = DEFAULT_HOST,
|
||
port: int = DEFAULT_PORT,
|
||
) -> None:
|
||
self.deps = deps
|
||
self.host = host
|
||
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:
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
logger.info("Web server lifespan started")
|
||
yield
|
||
logger.info("Web server lifespan ended")
|
||
|
||
app = FastAPI(title="Pi Multi-FX Pedal", version="0.1.0", lifespan=lifespan)
|
||
|
||
# Mount static files
|
||
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
|
||
|
||
# ── HTML pages ──────────────────────────────────────────────
|
||
|
||
@app.get("/", response_class=HTMLResponse)
|
||
async def root(request: Request):
|
||
"""Serve the v2 UI."""
|
||
v2_index = V2_DIST_DIR / "index.html"
|
||
if v2_index.exists():
|
||
content = v2_index.read_text()
|
||
return HTMLResponse(content)
|
||
return HTMLResponse("<h1>Pi Multi-FX Pedal</h1><p>UI build not found.</p>")
|
||
|
||
# Legacy routes — redirect to the v2 SPA
|
||
|
||
@app.get("/presets")
|
||
async def presets_redirect():
|
||
return RedirectResponse(url="/")
|
||
|
||
@app.get("/models")
|
||
async def models_redirect():
|
||
return RedirectResponse(url="/")
|
||
|
||
@app.get("/irs")
|
||
async def irs_redirect():
|
||
return RedirectResponse(url="/")
|
||
|
||
@app.get("/settings")
|
||
async def settings_redirect():
|
||
return RedirectResponse(url="/")
|
||
|
||
# ── REST API ────────────────────────────────────────────────
|
||
|
||
@app.get("/api/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(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(channel=Channel(channel))
|
||
result = []
|
||
for bank in banks:
|
||
presets_in_bank = []
|
||
for prog in range(4):
|
||
try:
|
||
p = pm.load(bank.number, prog, channel=Channel(channel))
|
||
presets_in_bank.append({
|
||
"name": p.name,
|
||
"bank": p.bank,
|
||
"program": p.program,
|
||
"chain": [
|
||
{
|
||
"fx_type": b.fx_type.value,
|
||
"block_id": b.block_id,
|
||
"enabled": b.enabled,
|
||
"bypass": b.bypass,
|
||
"params": dict(b.params),
|
||
"nam_model_path": b.nam_model_path,
|
||
"nam_model_name": Path(b.nam_model_path).stem if b.nam_model_path else "",
|
||
"ir_file_path": b.ir_file_path,
|
||
"ir_file_name": Path(b.ir_file_path).stem if b.ir_file_path else "",
|
||
"name": (
|
||
Path(b.nam_model_path).stem.replace("_", " ")
|
||
if b.nam_model_path
|
||
else Path(b.ir_file_path).stem.replace("_", " ")
|
||
if b.ir_file_path
|
||
else b.fx_type.value.replace("_", " ").title()
|
||
),
|
||
}
|
||
for b in p.chain
|
||
],
|
||
"master_volume": p.master_volume,
|
||
"tuner_enabled": p.tuner_enabled,
|
||
})
|
||
except FileNotFoundError:
|
||
presets_in_bank.append(None)
|
||
result.append({
|
||
"name": bank.name,
|
||
"number": bank.number,
|
||
"presets": presets_in_bank,
|
||
})
|
||
return {"banks": result}
|
||
|
||
@app.get("/api/presets/{bank}/{program}")
|
||
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:
|
||
p = pm.load(bank, program, channel=Channel(channel))
|
||
return {
|
||
"name": p.name,
|
||
"bank": p.bank,
|
||
"program": p.program,
|
||
"chain": [
|
||
{
|
||
"fx_type": b.fx_type.value,
|
||
"block_id": b.block_id,
|
||
"enabled": b.enabled,
|
||
"bypass": b.bypass,
|
||
"params": dict(b.params),
|
||
"nam_model_path": b.nam_model_path,
|
||
"nam_model_name": Path(b.nam_model_path).stem if b.nam_model_path else "",
|
||
"ir_file_path": b.ir_file_path,
|
||
"ir_file_name": Path(b.ir_file_path).stem if b.ir_file_path else "",
|
||
"name": (
|
||
Path(b.nam_model_path).stem.replace("_", " ")
|
||
if b.nam_model_path
|
||
else Path(b.ir_file_path).stem.replace("_", " ")
|
||
if b.ir_file_path
|
||
else b.fx_type.value.replace("_", " ").title()
|
||
),
|
||
}
|
||
for b in p.chain
|
||
],
|
||
"master_volume": p.master_volume,
|
||
"tuner_enabled": p.tuner_enabled,
|
||
}
|
||
except FileNotFoundError:
|
||
raise HTTPException(status_code=404, detail="Preset not found")
|
||
|
||
@app.post("/api/presets/{bank}/{program}/activate")
|
||
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(channel)
|
||
await self._manager.broadcast({
|
||
"type": "preset_changed",
|
||
"channel": channel,
|
||
"preset": {
|
||
"name": preset.name,
|
||
"bank": preset.bank,
|
||
"program": preset.program,
|
||
},
|
||
"state": state,
|
||
})
|
||
return {"ok": True, "preset": preset.name}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
@app.put("/api/presets/{bank}/{program}")
|
||
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:
|
||
chain = []
|
||
for b in data.get("chain", []):
|
||
chain.append(FXBlock(
|
||
fx_type=FXType(b["fx_type"]),
|
||
enabled=b.get("enabled", True),
|
||
bypass=b.get("bypass", False),
|
||
params=b.get("params", {}),
|
||
nam_model_path=b.get("nam_model_path", ""),
|
||
ir_file_path=b.get("ir_file_path", ""),
|
||
))
|
||
preset = Preset(
|
||
name=data.get("name", f"Preset {bank}-{program}"),
|
||
channel=Channel(channel),
|
||
bank=bank,
|
||
program=program,
|
||
chain=chain,
|
||
master_volume=data.get("master_volume", 0.8),
|
||
tuner_enabled=data.get("tuner_enabled", False),
|
||
)
|
||
pm.save(preset)
|
||
return {"ok": True, "name": preset.name}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
@app.delete("/api/presets/{bank}/{program}")
|
||
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:
|
||
pm.delete(bank, program)
|
||
return {"ok": True}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
# ── Snapshot endpoints ──────────────────────────────────────
|
||
|
||
SNAPSHOT_NAMES = {1: "Clean", 2: "Crunch", 3: "Lead", 4: "Rhythm",
|
||
5: "Ambient", 6: "Swell", 7: "Mellow", 8: "Edge"}
|
||
|
||
def _get_current_preset(pm) -> Preset | None:
|
||
"""Load the currently active preset."""
|
||
from ..presets.types import Preset
|
||
if not pm:
|
||
return None
|
||
try:
|
||
return pm.load(pm.current_bank, pm.current_program)
|
||
except Exception:
|
||
return None
|
||
|
||
def _snapshot_from_pipeline(pm, pl, slot: int, name: str = "") -> dict:
|
||
"""Capture current block states + params into a snapshot dict."""
|
||
from ..presets.types import FXBlock, FXType
|
||
preset = _get_current_preset(pm)
|
||
if not preset:
|
||
return {}
|
||
chain_data = []
|
||
for b in preset.chain:
|
||
chain_data.append({
|
||
"fx_type": b.fx_type.value,
|
||
"block_id": b.block_id,
|
||
"enabled": b.enabled,
|
||
"bypass": b.bypass,
|
||
"params": dict(b.params),
|
||
"nam_model_path": b.nam_model_path,
|
||
"ir_file_path": b.ir_file_path,
|
||
})
|
||
return {
|
||
"name": name or SNAPSHOT_NAMES.get(slot, f"Snapshot {slot}"),
|
||
"chain": chain_data,
|
||
"master_volume": preset.master_volume,
|
||
}
|
||
|
||
@app.get("/api/snapshots")
|
||
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": {}}
|
||
data = {}
|
||
for slot in range(1, 9):
|
||
snap = preset.snapshots.get(slot)
|
||
if snap:
|
||
chain = []
|
||
for b in snap.chain:
|
||
chain.append({
|
||
"fx_type": b.fx_type.value,
|
||
"block_id": b.block_id,
|
||
"enabled": b.enabled,
|
||
"bypass": b.bypass,
|
||
"params": dict(b.params),
|
||
"nam_model_path": b.nam_model_path,
|
||
"ir_file_path": b.ir_file_path,
|
||
})
|
||
data[str(slot)] = {
|
||
"name": snap.name,
|
||
"chain": chain,
|
||
"master_volume": snap.master_volume,
|
||
}
|
||
else:
|
||
data[str(slot)] = None
|
||
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, 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)
|
||
if not preset:
|
||
raise HTTPException(status_code=404, detail="No active preset")
|
||
name = (data or {}).get("name", "")
|
||
snap_data = _snapshot_from_pipeline(pm, pl, slot, name)
|
||
from ..presets.types import FXBlock, FXType, Snapshot as SnapModel
|
||
snap_chain = []
|
||
for bd in snap_data["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", ""),
|
||
)
|
||
)
|
||
preset.snapshots[slot] = SnapModel(
|
||
name=snap_data["name"],
|
||
chain=snap_chain,
|
||
master_volume=snap_data["master_volume"],
|
||
)
|
||
pm.save(preset)
|
||
# Mark current snapshot
|
||
pm._current_snapshot = slot
|
||
# Broadcast snapshot update
|
||
await self._manager.broadcast({
|
||
"type": "snapshot_saved",
|
||
"slot": slot,
|
||
"name": snap_data["name"],
|
||
})
|
||
return {"ok": True, "slot": slot, "name": snap_data["name"]}
|
||
|
||
@app.post("/api/snapshots/{slot}/recall")
|
||
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)
|
||
if not preset:
|
||
raise HTTPException(status_code=404, detail="No active preset")
|
||
snap = preset.snapshots.get(slot)
|
||
if not snap:
|
||
raise HTTPException(status_code=404, detail=f"No snapshot in slot {slot}")
|
||
|
||
# Apply snapshot state to the pipeline
|
||
if pl:
|
||
# Restore bypass states and params for each block
|
||
for i, snap_block in enumerate(snap.chain):
|
||
if i < len(preset.chain):
|
||
preset.chain[i].bypass = snap_block.bypass
|
||
preset.chain[i].enabled = snap_block.enabled
|
||
preset.chain[i].params = dict(snap_block.params)
|
||
preset.chain[i].nam_model_path = snap_block.nam_model_path
|
||
preset.chain[i].ir_file_path = snap_block.ir_file_path
|
||
# Restore master volume
|
||
preset.master_volume = snap.master_volume
|
||
# Push to pipeline
|
||
try:
|
||
pl.load_preset(preset)
|
||
except Exception:
|
||
pass
|
||
|
||
pm.save(preset)
|
||
pm._current_snapshot = slot
|
||
|
||
state = self._gather_state(channel)
|
||
await self._manager.broadcast({
|
||
"type": "snapshot_recalled",
|
||
"channel": channel,
|
||
"slot": slot,
|
||
"name": snap.name,
|
||
"state": state,
|
||
})
|
||
return {"ok": True, "slot": slot, "name": snap.name}
|
||
|
||
@app.put("/api/snapshots/{slot}")
|
||
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)
|
||
if not preset:
|
||
raise HTTPException(status_code=404, detail="No active preset")
|
||
snap = preset.snapshots.get(slot)
|
||
if not snap:
|
||
raise HTTPException(status_code=404, detail=f"No snapshot in slot {slot}")
|
||
if "name" in data:
|
||
snap.name = data["name"]
|
||
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, 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)
|
||
if not preset:
|
||
raise HTTPException(status_code=404, detail="No active preset")
|
||
preset.snapshots.pop(slot, None)
|
||
if getattr(pm, '_current_snapshot', 0) == slot:
|
||
pm._current_snapshot = 0
|
||
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, 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")
|
||
snap = preset.snapshots.get(slot)
|
||
if not snap:
|
||
raise HTTPException(status_code=404, detail=f"No snapshot in slot {slot}")
|
||
return {
|
||
"name": snap.name,
|
||
"slot": slot,
|
||
"channel": channel,
|
||
"chain": [
|
||
{
|
||
"fx_type": b.fx_type.value,
|
||
"block_id": b.block_id,
|
||
"enabled": b.enabled,
|
||
"bypass": b.bypass,
|
||
"params": dict(b.params),
|
||
}
|
||
for b in snap.chain
|
||
],
|
||
"master_volume": snap.master_volume,
|
||
}
|
||
|
||
@app.post("/api/bypass")
|
||
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:
|
||
pl.bypassed = bool(data["bypass"])
|
||
else:
|
||
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, 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:
|
||
pl._tuner_enabled = bool(data["enabled"])
|
||
else:
|
||
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(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,
|
||
"string": -1, "confidence": 0.0, "enabled": False,
|
||
}
|
||
return {
|
||
"frequency": getattr(pl, '_tuner_frequency', 0.0),
|
||
"note": getattr(pl, '_tuner_note', '--'),
|
||
"cents": getattr(pl, '_tuner_cents', 0),
|
||
"string": getattr(pl, '_tuner_string', -1),
|
||
"confidence": getattr(pl, '_tuner_confidence', 0.0),
|
||
"enabled": pl._tuner_enabled,
|
||
}
|
||
|
||
@app.put("/api/volume")
|
||
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
|
||
import asyncio
|
||
asyncio.ensure_future(self._manager.broadcast({
|
||
"type": "volume_changed",
|
||
"channel": channel,
|
||
"volume": vol,
|
||
}))
|
||
return {"ok": True, "volume": vol, "channel": channel}
|
||
|
||
# ── POST alias for volume (UI uses POST, server had PUT) ──
|
||
@app.post("/api/volume")
|
||
async def set_volume_post(data: dict, channel: str = "guitar"):
|
||
"""Set master volume — POST alias for PUT endpoint."""
|
||
_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
|
||
import asyncio
|
||
asyncio.ensure_future(self._manager.broadcast({
|
||
"type": "volume_changed",
|
||
"channel": channel,
|
||
"volume": vol,
|
||
}))
|
||
return {"ok": True, "volume": vol, "channel": channel}
|
||
|
||
# ── Toggle bypass alias (UI calls /api/bypass/toggle) ─────
|
||
@app.post("/api/bypass/toggle")
|
||
async def toggle_bypass_alias(data: Optional[dict] = None, channel: str = "guitar"):
|
||
"""Toggle or set bypass state — alias for /api/bypass."""
|
||
_pm, pl, _nam, _ir = self._channel_deps(channel)
|
||
if not pl:
|
||
raise HTTPException(status_code=503)
|
||
if data and "bypass" in data:
|
||
pl.bypassed = bool(data["bypass"])
|
||
else:
|
||
pl.bypassed = not pl.bypassed
|
||
await self._manager.broadcast({
|
||
"type": "bypass_changed",
|
||
"channel": channel,
|
||
"bypass": pl.bypassed,
|
||
})
|
||
return {"ok": True, "bypass": pl._bypassed}
|
||
|
||
# ── Block toggle (UI: PATCH /api/blocks {id, enabled}) ──
|
||
@app.patch("/api/blocks")
|
||
async def toggle_block(data: dict, channel: str = "guitar"):
|
||
"""Enable/disable a block by block_id or fx_type id — sets both enabled and bypass."""
|
||
from ..presets.types import Channel
|
||
pm, pl, nam, ir = self._channel_deps(channel)
|
||
if not pm:
|
||
raise HTTPException(status_code=503)
|
||
# Support both block_id (preferred) and id (fx_type, backward compat)
|
||
block_id = data.get("block_id") or data.get("id")
|
||
enabled = data.get("enabled")
|
||
if not block_id or enabled is None:
|
||
raise HTTPException(status_code=400, detail="Missing 'block_id'/'id' or 'enabled'")
|
||
try:
|
||
bank = pm.current_bank
|
||
program = pm.current_program
|
||
preset = pm.load(bank, program, channel=Channel(channel))
|
||
for b in preset.chain:
|
||
if b.block_id == block_id or (not data.get("block_id") and b.fx_type.value == block_id):
|
||
b.enabled = bool(enabled)
|
||
b.bypass = not bool(enabled) # sync bypass with enabled
|
||
pm.save(preset, channel=Channel(channel))
|
||
# Reload pipeline if needed
|
||
if pl:
|
||
pl.load_preset(preset)
|
||
await self._manager.broadcast({
|
||
"type": "block_toggled",
|
||
"channel": channel,
|
||
"id": block_id,
|
||
"enabled": b.enabled,
|
||
"bypass": b.bypass,
|
||
})
|
||
return {"ok": True, "id": block_id, "enabled": b.enabled, "bypass": b.bypass}
|
||
raise HTTPException(status_code=404, detail=f"Block '{block_id}' not found")
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
# ── Block param update (UI: PATCH /api/block {id, ...params}) ──
|
||
@app.patch("/api/block")
|
||
async def update_block_params(data: dict, channel: str = "guitar"):
|
||
"""Update parameters for a block by block_id or fx_type id."""
|
||
from ..presets.types import Channel, FXBlock
|
||
pm, pl, nam, ir = self._channel_deps(channel)
|
||
if not pm:
|
||
raise HTTPException(status_code=503)
|
||
# Support both block_id (preferred) and id (fx_type, backward compat)
|
||
block_id = data.get("block_id") or data.get("id")
|
||
if not block_id:
|
||
raise HTTPException(status_code=400, detail="Missing 'id'")
|
||
try:
|
||
bank = pm.current_bank
|
||
program = pm.current_program
|
||
preset = pm.load(bank, program, channel=Channel(channel))
|
||
for b in preset.chain:
|
||
if b.block_id == block_id or (not data.get("block_id") and b.fx_type.value == block_id):
|
||
# Update params from request body (skip 'id' key)
|
||
for key, val in data.items():
|
||
if key == "id":
|
||
continue
|
||
if key in ("nam_model_path", "ir_file_path", "bypass", "enabled", "subtype"):
|
||
if key == "bypass":
|
||
b.bypass = bool(val)
|
||
b.enabled = not b.bypass # sync: bypass=true → enabled=false
|
||
elif key == "enabled":
|
||
b.enabled = bool(val)
|
||
b.bypass = not b.enabled # sync: enabled=false → bypass=true
|
||
elif key == "subtype":
|
||
b.subtype = str(val)
|
||
else:
|
||
setattr(b, key, str(val))
|
||
else:
|
||
# Try setting as param even if not in defaults
|
||
b.params[key] = float(val)
|
||
pm.save(preset, channel=Channel(channel))
|
||
if pl:
|
||
pl.load_preset(preset)
|
||
await self._manager.broadcast({
|
||
"type": "block_params_changed",
|
||
"channel": channel,
|
||
"id": block_id,
|
||
"params": dict(b.params),
|
||
"bypass": b.bypass,
|
||
"enabled": b.enabled,
|
||
})
|
||
return {"ok": True, "id": block_id, "params": dict(b.params)}
|
||
raise HTTPException(status_code=404, detail=f"Block '{block_id}' not found")
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
# ── 4CM routing endpoints ────────────────────────────────
|
||
|
||
@app.get("/api/routing")
|
||
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 {
|
||
"routing_mode": pl.routing_mode,
|
||
"routing_breakpoint": pl.routing_breakpoint,
|
||
}
|
||
|
||
@app.post("/api/routing")
|
||
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.
|
||
"""
|
||
_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)
|
||
bp = int(data.get("routing_breakpoint", pl.routing_breakpoint))
|
||
pl.set_routing(mode, bp)
|
||
|
||
# Persist to current preset if preset manager is available
|
||
if pm:
|
||
try:
|
||
bank = pm.current_bank
|
||
program = pm.current_program
|
||
preset = pm.load(bank, program)
|
||
pm.set_4cm_config(preset, mode=mode, breakpoint=bp)
|
||
except Exception:
|
||
pass
|
||
|
||
await self._manager.broadcast({
|
||
"type": "routing_changed",
|
||
"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}
|
||
|
||
# ── Per-channel output routing ────────────────────────────
|
||
|
||
@app.get("/api/channel-output")
|
||
async def get_channel_output():
|
||
"""Get output routing for each channel."""
|
||
audio = self.deps.audio_system
|
||
if not audio or not audio.config:
|
||
return {"ch1": "both", "ch2": "both"}
|
||
cfg = audio.config.__dict__ if hasattr(audio.config, '__dict__') else {}
|
||
return {
|
||
"ch1": cfg.get("ch1_output", "both"),
|
||
"ch2": cfg.get("ch2_output", "both"),
|
||
}
|
||
|
||
@app.post("/api/channel-output")
|
||
async def set_channel_output(data: dict):
|
||
"""Set output routing for a channel.
|
||
|
||
Body: { "ch": 1, "port": "playback_1"|"playback_2"|"both" }
|
||
"""
|
||
ch = int(data.get("ch", 1))
|
||
port = data.get("port", "both")
|
||
if port not in ("playback_1", "playback_2", "both"):
|
||
raise HTTPException(status_code=400, detail=f"Invalid port: {port}")
|
||
audio = self.deps.audio_system
|
||
if audio and audio.config:
|
||
key = f"ch{ch}_output"
|
||
if hasattr(audio.config, 'input_device'):
|
||
setattr(audio.config, key, port)
|
||
# Reconnect JACK ports
|
||
audio.connect_fx_ports()
|
||
return {"ok": True, f"ch{ch}": port}
|
||
|
||
@app.get("/api/models")
|
||
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()
|
||
return {
|
||
"models": [
|
||
{
|
||
"name": m.name,
|
||
"path": m.path,
|
||
"size_mb": round(m.size_mb, 2),
|
||
"architecture": m.architecture,
|
||
"loaded": nam.current_model is not None and nam.current_model.name == m.name,
|
||
}
|
||
for m in models
|
||
],
|
||
"current": nam.current_model.name if nam.current_model else None,
|
||
}
|
||
|
||
@app.get("/api/nam/engine")
|
||
async def get_nam_engine(channel: str = "guitar"):
|
||
"""Get current NAM engine mode."""
|
||
_pm, _pl, nam, _ir = self._channel_deps(channel)
|
||
if not nam:
|
||
raise HTTPException(status_code=503)
|
||
mode = getattr(nam, 'engine_mode', 'cpp')
|
||
return {"engine_mode": mode}
|
||
|
||
@app.post("/api/nam/engine")
|
||
async def set_nam_engine(data: dict, channel: str = "guitar"):
|
||
"""Switch NAM engine between 'cpp' and 'pytorch'."""
|
||
_pm, _pl, nam, _ir = self._channel_deps(channel)
|
||
if not nam:
|
||
raise HTTPException(status_code=503)
|
||
mode = data.get("engine_mode", "cpp")
|
||
if mode not in ("cpp", "pytorch"):
|
||
raise HTTPException(status_code=400, detail="engine_mode must be 'cpp' or 'pytorch'")
|
||
ok = nam.set_engine(mode)
|
||
if not ok:
|
||
raise HTTPException(status_code=500, detail=f"Failed to switch to {mode} engine")
|
||
return {"ok": True, "engine_mode": mode}
|
||
|
||
@app.post("/api/models/load")
|
||
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", "")
|
||
if not path:
|
||
raise HTTPException(status_code=400, detail="path required")
|
||
ok = nam.load_model(path)
|
||
if not ok:
|
||
raise HTTPException(status_code=500, detail="Failed to load model")
|
||
await self._manager.broadcast({
|
||
"type": "model_loaded",
|
||
"model": path,
|
||
})
|
||
return {"ok": True}
|
||
|
||
@app.post("/api/models/unload")
|
||
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()
|
||
await self._manager.broadcast({
|
||
"type": "model_unloaded",
|
||
"channel": channel,
|
||
})
|
||
return {"ok": True}
|
||
|
||
@app.get("/api/irs")
|
||
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()
|
||
return {
|
||
"irs": [
|
||
{
|
||
"name": irf.name,
|
||
"path": irf.path,
|
||
"sample_rate": irf.sample_rate,
|
||
"num_taps": irf.num_taps,
|
||
"length_ms": round(irf.length_ms, 1),
|
||
"loaded": ir.current_ir is not None and ir.current_ir.name == irf.name,
|
||
}
|
||
for irf in irs
|
||
],
|
||
"current": ir.current_ir.name if ir.current_ir else None,
|
||
}
|
||
|
||
@app.post("/api/irs/load")
|
||
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", "")
|
||
if not path:
|
||
raise HTTPException(status_code=400, detail="path required")
|
||
ok = ir.load_ir(path)
|
||
if not ok:
|
||
raise HTTPException(status_code=500, detail="Failed to load IR")
|
||
await self._manager.broadcast({
|
||
"type": "ir_loaded",
|
||
"ir": path,
|
||
})
|
||
return {"ok": True}
|
||
|
||
@app.post("/api/irs/unload")
|
||
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()
|
||
await self._manager.broadcast({
|
||
"type": "ir_unloaded",
|
||
"channel": channel,
|
||
})
|
||
return {"ok": True}
|
||
|
||
# ── File upload endpoints ─────────────────────────────────
|
||
|
||
@app.post("/api/models/upload")
|
||
async def upload_model(request: Request):
|
||
"""Upload a .nam model file to the pedal's models directory."""
|
||
nam = self.deps.nam_host
|
||
if not nam:
|
||
raise HTTPException(status_code=503)
|
||
form = await request.form()
|
||
file = form.get("file")
|
||
if not file or not file.filename:
|
||
raise HTTPException(status_code=400, detail="No file provided")
|
||
if not file.filename.endswith(".nam"):
|
||
raise HTTPException(status_code=400, detail="Only .nam files accepted")
|
||
models_dir = nam._models_dir
|
||
dest = models_dir / Path(file.filename).name
|
||
content = await file.read()
|
||
dest.write_bytes(content)
|
||
logger.info("Uploaded NAM model: %s", dest.name)
|
||
return {"ok": True, "filename": dest.name}
|
||
|
||
@app.post("/api/irs/upload")
|
||
async def upload_ir(request: Request):
|
||
"""Upload a .wav IR file to the pedal's IR directory."""
|
||
ir = self.deps.ir_loader
|
||
if not ir:
|
||
raise HTTPException(status_code=503)
|
||
form = await request.form()
|
||
file = form.get("file")
|
||
if not file or not file.filename:
|
||
raise HTTPException(status_code=400, detail="No file provided")
|
||
if not file.filename.endswith(".wav"):
|
||
raise HTTPException(status_code=400, detail="Only .wav files accepted")
|
||
ir_dir = ir._ir_dir
|
||
dest = ir_dir / Path(file.filename).name
|
||
content = await file.read()
|
||
dest.write_bytes(content)
|
||
logger.info("Uploaded IR file: %s", dest.name)
|
||
return {"ok": True, "filename": dest.name}
|
||
|
||
# ── WiFi endpoints ──────────────────────────────────────
|
||
|
||
@app.get("/api/wifi/status")
|
||
async def wifi_get_status():
|
||
"""Get current WiFi client connection status."""
|
||
from ..system.network import wifi_status
|
||
return wifi_status()
|
||
|
||
@app.get("/api/wifi/scan")
|
||
async def wifi_scan():
|
||
"""Scan for available WiFi networks."""
|
||
from ..system.network import wifi_scan
|
||
return {"networks": wifi_scan()}
|
||
|
||
@app.get("/api/wifi/saved")
|
||
async def wifi_saved():
|
||
"""List saved/known WiFi networks."""
|
||
from ..system.network import wifi_saved_networks
|
||
return {"networks": wifi_saved_networks()}
|
||
|
||
@app.post("/api/wifi/connect")
|
||
async def wifi_connect(data: dict):
|
||
"""Connect to a WiFi network.
|
||
|
||
Body: {ssid: str, password: str}
|
||
"""
|
||
from ..system.network import wifi_connect
|
||
return wifi_connect(
|
||
ssid=data.get("ssid", ""),
|
||
password=data.get("password", ""),
|
||
)
|
||
|
||
@app.post("/api/wifi/disconnect")
|
||
async def wifi_disconnect():
|
||
"""Disconnect from the current WiFi network."""
|
||
from ..system.network import wifi_disconnect
|
||
return wifi_disconnect()
|
||
|
||
@app.post("/api/wifi/forget")
|
||
async def wifi_forget(data: dict):
|
||
"""Forget a saved WiFi network.
|
||
|
||
Body: {ssid: str}
|
||
"""
|
||
from ..system.network import wifi_forget
|
||
return wifi_forget(ssid=data.get("ssid", ""))
|
||
|
||
@app.get("/api/wifi/hotspot")
|
||
async def wifi_hotspot_status():
|
||
"""Get hotspot status."""
|
||
from ..system.network import hotspot_status
|
||
return hotspot_status()
|
||
|
||
@app.post("/api/wifi/hotspot/enable")
|
||
async def wifi_hotspot_enable(data: dict):
|
||
"""Enable WiFi hotspot.
|
||
|
||
Body: {ssid: str (optional), password: str (optional)}
|
||
"""
|
||
from ..system.network import hotspot_enable
|
||
return hotspot_enable(
|
||
ssid=data.get("ssid", "Pi-Pedal"),
|
||
password=data.get("password", "pedal1234"),
|
||
)
|
||
|
||
@app.post("/api/wifi/hotspot/disable")
|
||
async def wifi_hotspot_disable():
|
||
"""Disable WiFi hotspot."""
|
||
from ..system.network import hotspot_disable
|
||
return hotspot_disable()
|
||
|
||
# ── Bluetooth endpoints ──────────────────────────────────
|
||
|
||
@app.get("/api/bluetooth/status")
|
||
async def bluetooth_status():
|
||
"""Get Bluetooth adapter status."""
|
||
from ..system.bluetooth import bt_status
|
||
return bt_status()
|
||
|
||
@app.post("/api/bluetooth/power")
|
||
async def bluetooth_power(data: dict):
|
||
"""Power Bluetooth on or off.
|
||
|
||
Body: {on: bool}
|
||
"""
|
||
from ..system.bluetooth import bt_power
|
||
return bt_power(on=data.get("on", True))
|
||
|
||
@app.post("/api/bluetooth/discoverable")
|
||
async def bluetooth_discoverable(data: dict):
|
||
"""Set Bluetooth discoverable.
|
||
|
||
Body: {on: bool}
|
||
"""
|
||
from ..system.bluetooth import bt_discoverable
|
||
return bt_discoverable(on=data.get("on", True))
|
||
|
||
@app.post("/api/bluetooth/scan")
|
||
async def bluetooth_scan():
|
||
"""Scan for discoverable Bluetooth devices."""
|
||
from ..system.bluetooth import bt_scan
|
||
return bt_scan(timeout=12)
|
||
|
||
@app.get("/api/bluetooth/paired")
|
||
async def bluetooth_paired():
|
||
"""List paired Bluetooth devices."""
|
||
from ..system.bluetooth import bt_paired_devices
|
||
return {"devices": bt_paired_devices()}
|
||
|
||
@app.post("/api/bluetooth/pair")
|
||
async def bluetooth_pair(data: dict):
|
||
"""Pair with a Bluetooth device.
|
||
|
||
Body: {address: str}
|
||
"""
|
||
from ..system.bluetooth import bt_pair
|
||
return bt_pair(address=data.get("address", ""))
|
||
|
||
@app.post("/api/bluetooth/unpair")
|
||
async def bluetooth_unpair(data: dict):
|
||
"""Unpair a Bluetooth device.
|
||
|
||
Body: {address: str}
|
||
"""
|
||
from ..system.bluetooth import bt_unpair
|
||
return bt_unpair(address=data.get("address", ""))
|
||
|
||
@app.post("/api/bluetooth/connect")
|
||
async def bluetooth_connect(data: dict):
|
||
"""Connect to a paired Bluetooth device.
|
||
|
||
Body: {address: str}
|
||
"""
|
||
from ..system.bluetooth import bt_connect
|
||
return bt_connect(address=data.get("address", ""))
|
||
|
||
@app.post("/api/bluetooth/disconnect")
|
||
async def bluetooth_disconnect(data: dict):
|
||
"""Disconnect a Bluetooth device.
|
||
|
||
Body: {address: str}
|
||
"""
|
||
from ..system.bluetooth import bt_disconnect
|
||
return bt_disconnect(address=data.get("address", ""))
|
||
|
||
@app.get("/api/bluetooth/midi-status")
|
||
async def bluetooth_midi_status():
|
||
"""Get Bluetooth MIDI service status."""
|
||
from ..system.bluetooth import bt_midi_status
|
||
return bt_midi_status()
|
||
|
||
@app.post("/api/bluetooth/midi-enable")
|
||
async def bluetooth_midi_enable():
|
||
"""Enable and start the Bluetooth MIDI service."""
|
||
from ..system.bluetooth import bt_midi_enable
|
||
return bt_midi_enable()
|
||
|
||
@app.post("/api/bluetooth/midi-disable")
|
||
async def bluetooth_midi_disable():
|
||
"""Stop and disable the Bluetooth MIDI service."""
|
||
from ..system.bluetooth import bt_midi_disable
|
||
return bt_midi_disable()
|
||
|
||
@app.get("/api/bluetooth/midi-devices")
|
||
async def bluetooth_midi_devices():
|
||
"""List connected Bluetooth MIDI devices."""
|
||
from ..system.bluetooth import bt_midi_connected_devices
|
||
return {"devices": bt_midi_connected_devices()}
|
||
|
||
# ── Audio latency profile endpoints ───────────────────────
|
||
|
||
@app.get("/api/audio/profiles")
|
||
async def list_audio_profiles():
|
||
"""List all available JACK latency profiles."""
|
||
return {
|
||
"profiles": [
|
||
{
|
||
"key": k,
|
||
**v,
|
||
}
|
||
for k, v in LATENCY_PROFILES.items()
|
||
],
|
||
"current": (
|
||
self.deps.audio_system.config.profile
|
||
if self.deps.audio_system
|
||
else "standard"
|
||
),
|
||
}
|
||
|
||
@app.get("/api/audio/profile")
|
||
async def get_audio_profile():
|
||
"""Get the current JACK latency profile and audio stats."""
|
||
if not self.deps.audio_system:
|
||
return {
|
||
"profile": "standard",
|
||
"period": 256,
|
||
"nperiods": 2,
|
||
"rate": 48000,
|
||
"jack_running": False,
|
||
"xrun_count": None,
|
||
}
|
||
profile_key = self.deps.audio_system.config.profile
|
||
profile = self.deps.audio_system.config.latency_profile
|
||
xruns = self.deps.audio_system.read_xrun_count()
|
||
return {
|
||
"profile": profile_key,
|
||
"period": profile["period"],
|
||
"nperiods": profile["nperiods"],
|
||
"rate": profile["rate"],
|
||
"jack_running": True,
|
||
"xrun_count": xruns,
|
||
"input_device": self.deps.audio_system.config.input_device,
|
||
"output_device": self.deps.audio_system.config.output_device,
|
||
}
|
||
|
||
@app.post("/api/audio/profile")
|
||
async def set_audio_profile(data: dict):
|
||
"""Switch JACK latency profile and/or set custom period/rate.
|
||
|
||
Body — at least one of:
|
||
{ "profile": "stable" } → switch to a named profile
|
||
{ "period": 256, "rate": 48000 } → custom settings
|
||
{ "profile": "stable", → named profile with rate override
|
||
"rate": 44100 }
|
||
|
||
Changes are saved to ~/.pedal/config.yaml so they survive reboot.
|
||
The switch restarts JACK (brief audio dropout ~1s).
|
||
"""
|
||
audio_sys = self.deps.audio_system
|
||
if not audio_sys:
|
||
raise HTTPException(status_code=503, detail="Audio system not available")
|
||
|
||
profile_key = data.get("profile")
|
||
custom_period = data.get("period")
|
||
custom_rate = data.get("rate")
|
||
|
||
# ── Resolve target profile ───────────────────────────────
|
||
if profile_key and profile_key not in LATENCY_PROFILES and profile_key != "custom":
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"Unknown profile: {profile_key!r}. Options: {list(LATENCY_PROFILES.keys())}",
|
||
)
|
||
|
||
if profile_key and profile_key in LATENCY_PROFILES:
|
||
# Named profile — start from profile defaults
|
||
target_profile = dict(LATENCY_PROFILES[profile_key])
|
||
else:
|
||
# Custom — use current as base, or fall back to stable
|
||
current = audio_sys.config.latency_profile
|
||
target_profile = dict(current)
|
||
|
||
# Apply overrides
|
||
if custom_period is not None:
|
||
if custom_period not in (16, 32, 64, 128, 256, 512, 1024, 2048):
|
||
raise HTTPException(status_code=400, detail=f"Invalid period: {custom_period}. Must be power of 2 (16-2048)")
|
||
target_profile["period"] = custom_period
|
||
if custom_rate is not None:
|
||
if custom_rate not in (44100, 48000, 96000, 192000):
|
||
raise HTTPException(status_code=400, detail=f"Invalid rate: {custom_rate}. Must be 44100, 48000, 96000, or 192000")
|
||
target_profile["rate"] = custom_rate
|
||
|
||
# Build effective profile name
|
||
if profile_key and profile_key in LATENCY_PROFILES:
|
||
if custom_period is not None or custom_rate is not None:
|
||
effective_key = f"{profile_key}_custom"
|
||
else:
|
||
effective_key = profile_key
|
||
else:
|
||
effective_key = "custom"
|
||
|
||
old_key = audio_sys.config.profile
|
||
# Check if anything actually changed
|
||
cur_profile = audio_sys.config.latency_profile
|
||
if (audio_sys.config.profile == effective_key
|
||
and cur_profile["period"] == target_profile["period"]
|
||
and cur_profile["rate"] == target_profile["rate"]):
|
||
return {"ok": True, "profile": effective_key, "restarted": False}
|
||
|
||
# ── Apply new profile on the fly ─────────────────────────
|
||
# Update AudioConfig — store named overrides inline in profile key
|
||
audio_sys.config.profile = effective_key
|
||
|
||
logger.info(
|
||
"Switching audio: %s → %s (period=%d, rate=%d)",
|
||
old_key, effective_key,
|
||
target_profile["period"], target_profile["rate"],
|
||
)
|
||
|
||
jack_client = self.deps.jack_audio
|
||
|
||
# Update LATENCY_PROFILES with custom entry so latency_profile resolves it
|
||
LATENCY_PROFILES[effective_key] = target_profile
|
||
|
||
# ── Kill JACK server FIRST (breaks any client deadlock) ──
|
||
# jack.Client.deactivate() is a synchronous JACK protocol call
|
||
# that blocks until the RT callback thread acknowledges. If the
|
||
# RT thread is stuck (ALSA reconfiguring, xruns), deactivate()
|
||
# hangs permanently. Kill the server first to force-kill the
|
||
# RT thread, then clean up the client.
|
||
# See also: JackAudioClient.stop() which calls deactivate().
|
||
try:
|
||
audio_sys.stop_jack()
|
||
except Exception:
|
||
logger.warning("stop_jack() failed during profile switch")
|
||
if jack_client:
|
||
try:
|
||
jack_client.stop()
|
||
except Exception:
|
||
logger.warning("jack_client.stop() failed (expected if JACK was mid-callback)")
|
||
|
||
# Start JACK with new parameters
|
||
# Disable auto_connect temporarily — fx_in/fx_out ports don't
|
||
# exist until jack_client.start() recreates the JACK client below
|
||
saved_auto_connect = audio_sys.config.auto_connect
|
||
audio_sys.config.auto_connect = False
|
||
ok = audio_sys.start_jack(timeout=15)
|
||
audio_sys.config.auto_connect = saved_auto_connect
|
||
if not ok:
|
||
# Rollback on failure
|
||
audio_sys.config.profile = old_key
|
||
logger.warning("JACK restart failed — attempting rollback to %s", old_key)
|
||
audio_sys.start_jack(timeout=15)
|
||
if jack_client:
|
||
try:
|
||
jack_client.start()
|
||
except Exception:
|
||
pass
|
||
raise HTTPException(
|
||
status_code=500,
|
||
detail=f"Failed to start JACK with period={target_profile['period']}, rate={target_profile['rate']} — rolled back",
|
||
)
|
||
|
||
# Sync NAM engine block size BEFORE restarting JACK client
|
||
# so the NAM subprocess is already sized for the new period
|
||
# when the process callback starts firing.
|
||
nam_host = self.deps.nam_host
|
||
if nam_host and hasattr(nam_host, 'set_block_size'):
|
||
nam_host.set_block_size(target_profile["period"])
|
||
|
||
# Restart JACK audio client
|
||
if jack_client:
|
||
try:
|
||
jack_client.start()
|
||
except Exception as e:
|
||
logger.warning("jack_client.start() failed after profile switch: %s", e)
|
||
|
||
# Reconnect FX ports now that pipeline's JACK client is alive
|
||
if saved_auto_connect:
|
||
try:
|
||
audio_sys.connect_fx_ports()
|
||
except Exception as e:
|
||
logger.warning("connect_fx_ports() failed after profile switch: %s", e)
|
||
|
||
# ── Persist to config.yaml ──────────────────────────────
|
||
if self.deps.config is not None:
|
||
audio_cfg = self.deps.config.setdefault("audio", {})
|
||
audio_cfg["profile"] = effective_key
|
||
# Store custom period/rate in config too
|
||
audio_cfg["period"] = target_profile["period"]
|
||
audio_cfg["rate"] = target_profile["rate"]
|
||
save_config(self.deps.config, self.deps.config_path)
|
||
|
||
await self._manager.broadcast({
|
||
"type": "audio_profile_changed",
|
||
"profile": effective_key,
|
||
"period": target_profile["period"],
|
||
"rate": target_profile["rate"],
|
||
})
|
||
|
||
return {
|
||
"ok": True,
|
||
"profile": effective_key,
|
||
"period": target_profile["period"],
|
||
"rate": target_profile["rate"],
|
||
"restarted": True,
|
||
}
|
||
|
||
# ── Generic settings persistence ──────────────────────────
|
||
|
||
KEY_MAP = {
|
||
"input_device": ("audio", "input_device"),
|
||
"output_device": ("audio", "output_device"),
|
||
"channel_mode": ("audio", "mode"),
|
||
"input_pad": ("audio", "input_pad"),
|
||
"input_impedance": ("audio", "input_impedance"),
|
||
"midi_channel": ("midi", "channel"),
|
||
"midi_clock": ("midi", "clock"),
|
||
"midi_thru": ("midi", "thru"),
|
||
"auto_dim": ("display", "auto_dim"),
|
||
"screen_saver": ("display", "screen_saver"),
|
||
"tap_tempo": ("tempo", "tap_tempo"),
|
||
"tempo_division": ("tempo", "division"),
|
||
"tap_tempo_mute": ("tempo", "mute_on_tap"),
|
||
}
|
||
|
||
@app.get("/api/settings")
|
||
async def get_settings():
|
||
"""Return the full config dict (read-only snapshot)."""
|
||
if self.deps.config is None:
|
||
raise HTTPException(status_code=503, detail="Config not available")
|
||
return self.deps.config
|
||
@app.post("/api/settings")
|
||
async def save_settings(data: dict):
|
||
"""Save arbitrary settings to config.yaml.
|
||
|
||
Accepts a dict of key-value pairs. Each key is mapped to
|
||
the correct ``section.key`` in ``~/.pedal/config.yaml`` and
|
||
persisted immediately.
|
||
|
||
Special handling:
|
||
``brightness`` — UI sends 1–10, stored as 0.1–1.0 float.
|
||
"""
|
||
if self.deps.config is None:
|
||
raise HTTPException(status_code=503, detail="Config not available")
|
||
|
||
cfg = self.deps.config
|
||
n_saved = 0
|
||
|
||
for key, val in data.items():
|
||
if key == "brightness":
|
||
leds = cfg.setdefault("leds", {})
|
||
if isinstance(val, (int, float)) and val >= 1:
|
||
leds["brightness"] = val / 10.0
|
||
else:
|
||
leds["brightness"] = val
|
||
n_saved += 1
|
||
elif key in KEY_MAP:
|
||
section, skey = KEY_MAP[key]
|
||
cfg.setdefault(section, {})[skey] = val
|
||
n_saved += 1
|
||
|
||
save_config(cfg, self.deps.config_path)
|
||
logger.info("Settings saved: %d keys from %s", n_saved, data)
|
||
|
||
# ── Runtime apply for live-updatable settings ──────────
|
||
if "brightness" in data and self.deps.leds is not None:
|
||
try:
|
||
bv = data["brightness"]
|
||
if isinstance(bv, (int, float)) and bv >= 1:
|
||
bv = bv / 10.0
|
||
self.deps.leds.set_brightness(float(bv))
|
||
except Exception as e:
|
||
logger.warning("Failed to apply brightness: %s", e)
|
||
|
||
if "channel_mode" in data and self.deps.pipeline is not None:
|
||
try:
|
||
mode = str(data["channel_mode"])
|
||
self.deps.pipeline.routing_mode = "4cm" if mode == "stereo_4cm" else "mono"
|
||
logger.info("Pipeline routing_mode set to %s", self.deps.pipeline.routing_mode)
|
||
except Exception as e:
|
||
logger.warning("Failed to apply channel_mode: %s", e)
|
||
|
||
# ── JACK device hot-swap for input_device / output_device ──
|
||
if "input_device" in data or "output_device" in data:
|
||
try:
|
||
audio_sys = self.deps.audio_system
|
||
jack_client = self.deps.jack_audio
|
||
# Kill JACK server FIRST to avoid deadlock on
|
||
# jack.Client.deactivate() (same pattern as profile switch)
|
||
if "input_device" in data:
|
||
audio_sys.config.input_device = str(data["input_device"])
|
||
if "output_device" in data:
|
||
audio_sys.config.output_device = str(data["output_device"])
|
||
audio_sys.stop_jack()
|
||
if jack_client:
|
||
jack_client.stop()
|
||
ok = audio_sys.start_jack(timeout=10)
|
||
if not ok:
|
||
logger.warning("JACK restart failed after device change — rolling back")
|
||
# Restore old devices is complex, just log the failure
|
||
if jack_client:
|
||
jack_client.start()
|
||
logger.info("JACK restarted for device change: input=%s output=%s",
|
||
audio_sys.config.input_device, audio_sys.config.output_device)
|
||
except Exception as e:
|
||
logger.warning("Failed to apply device change: %s", e)
|
||
|
||
# ── MIDI channel filter ──
|
||
if "midi_channel" in data and self.deps.midi is not None:
|
||
try:
|
||
val = data["midi_channel"]
|
||
if val is None or str(val).lower() == "omni":
|
||
self.deps.midi.set_channel_filter(None)
|
||
logger.info("MIDI channel filter set to omni")
|
||
else:
|
||
ch = int(val) - 1 # UI sends "1"-"16", store as 0-15
|
||
self.deps.midi.set_channel_filter(ch)
|
||
logger.info("MIDI channel filter set to %d", ch)
|
||
except Exception as e:
|
||
logger.warning("Failed to apply midi_channel: %s", e)
|
||
|
||
# ── Display settings ──
|
||
if "auto_dim" in data and self.deps.display is not None:
|
||
try:
|
||
self.deps.display.set_auto_dim(bool(data["auto_dim"]))
|
||
except Exception as e:
|
||
logger.warning("Failed to apply auto_dim: %s", e)
|
||
|
||
if "screen_saver" in data and self.deps.display is not None:
|
||
try:
|
||
self.deps.display.set_screen_saver(str(data["screen_saver"]))
|
||
except Exception as e:
|
||
logger.warning("Failed to apply screen_saver: %s", e)
|
||
|
||
# ── Tempo settings ──
|
||
if "tap_tempo" in data and self.deps.pipeline is not None:
|
||
try:
|
||
self.deps.pipeline.set_tempo(float(data["tap_tempo"]))
|
||
except Exception as e:
|
||
logger.warning("Failed to apply tap_tempo: %s", e)
|
||
|
||
if "tempo_division" in data:
|
||
logger.info("tempo_division change noted (boot-only): %s", data["tempo_division"])
|
||
|
||
if "tap_tempo_mute" in data:
|
||
logger.info("tap_tempo_mute change noted (boot-only): %s", data["tap_tempo_mute"])
|
||
|
||
return {"ok": True, "saved": n_saved}
|
||
|
||
# ── FX block param schemas ───────────────────────────────
|
||
|
||
@app.get("/api/block-params/{fx_type}")
|
||
async def get_block_params(fx_type: str):
|
||
"""Return editable parameters for a given FX block type.
|
||
|
||
Returns a list of param definitions with name, type, min, max, default.
|
||
The web UI uses this to render appropriate sliders/knobs.
|
||
"""
|
||
param_schema = _FX_PARAM_SCHEMAS.get(fx_type, [])
|
||
return {"fx_type": fx_type, "params": param_schema}
|
||
|
||
# ── WebSocket ──────────────────────────────────────────────
|
||
|
||
@app.websocket("/ws")
|
||
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 (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()
|
||
msg = json.loads(raw)
|
||
msg_type = msg.get("type", "")
|
||
|
||
if msg_type == "ping":
|
||
await ws.send_text(json.dumps({"type": "pong"}))
|
||
elif msg_type == "get_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:
|
||
logger.warning("WebSocket error: %s", e)
|
||
finally:
|
||
self._manager.disconnect(ws)
|
||
logger.info("WebSocket client disconnected (%d active)", self._manager.active_count)
|
||
|
||
# ── Tone3000 downloader endpoints ───────────────────────
|
||
|
||
@app.get("/api/tonedownload/status")
|
||
async def tonedownload_status():
|
||
"""Check if the Tone3000 API is reachable."""
|
||
client = await self._get_tonedownload()
|
||
if not client:
|
||
return {"online": False, "error": "Client not initialized"}
|
||
ok = await client.ping()
|
||
return {"online": ok}
|
||
|
||
@app.get("/api/models/tonedownload/search")
|
||
async def tonedownload_search_models(q: str = "", page: int = 0, arch: str = "", gear: str = "", sort: str = "created_at.desc"):
|
||
"""Search Tone3000 for NAM models."""
|
||
try:
|
||
client = await self._get_tonedownload()
|
||
if not client:
|
||
return JSONResponse(
|
||
status_code=200,
|
||
content={
|
||
"results": [],
|
||
"status": "offline",
|
||
"message": "Tone3000 unavailable - using local defaults",
|
||
},
|
||
)
|
||
arch_val = arch.strip() or None
|
||
gear_val = gear.strip() or None
|
||
if not q.strip():
|
||
results = await client.get_trending_models(arch=arch_val, gear=gear_val)
|
||
else:
|
||
results = await client.search_models(q.strip(), page=page, arch=arch_val, gear=gear_val, sort=sort)
|
||
return {
|
||
"results": [
|
||
{
|
||
"id": r.id,
|
||
"name": r.display_name,
|
||
"filename": r.filename,
|
||
"size": r.size_bytes,
|
||
"size_display": format_size(r.size_bytes),
|
||
"author": r.author_username or "Unknown",
|
||
"author_avatar": r.author_avatar_url,
|
||
"thumbnail": r.thumbnail_url,
|
||
"download_url": r.download_url,
|
||
"tone_description": r.tone_description,
|
||
"tone_gear": r.tone_gear,
|
||
"pack_name": r.pack_name,
|
||
"architecture": r.architecture,
|
||
"created_at": r.created_at,
|
||
"downloads": r.downloads,
|
||
"likes": r.likes,
|
||
}
|
||
for r in results
|
||
],
|
||
"page": page,
|
||
}
|
||
except Exception as e:
|
||
logger.warning("Tone3000 model search failed: %s", e)
|
||
return JSONResponse(
|
||
status_code=200,
|
||
content={
|
||
"results": [],
|
||
"status": "offline",
|
||
"message": "Tone3000 unavailable - using local defaults",
|
||
},
|
||
)
|
||
|
||
@app.get("/api/irs/tonedownload/search")
|
||
async def tonedownload_search_irs(q: str = "", page: int = 0):
|
||
"""Search Tone3000 for IRs."""
|
||
try:
|
||
client = await self._get_tonedownload()
|
||
if not client:
|
||
return JSONResponse(
|
||
status_code=200,
|
||
content={
|
||
"results": [],
|
||
"status": "offline",
|
||
"message": "Tone3000 unavailable - using local defaults",
|
||
},
|
||
)
|
||
if not q.strip():
|
||
results = await client.get_trending_irs()
|
||
else:
|
||
results = await client.search_irs(q.strip(), page=page)
|
||
return {
|
||
"results": [
|
||
{
|
||
"id": r.id,
|
||
"name": r.display_name,
|
||
"filename": r.filename,
|
||
"size": r.size_bytes,
|
||
"size_display": format_size(r.size_bytes),
|
||
"author": r.author_username or "Unknown",
|
||
"author_avatar": r.author_avatar_url,
|
||
"thumbnail": r.thumbnail_url,
|
||
"download_url": r.download_url,
|
||
"tone_description": r.tone_description,
|
||
"created_at": r.created_at,
|
||
}
|
||
for r in results
|
||
],
|
||
}
|
||
except Exception as e:
|
||
logger.warning("Tone3000 IR search failed: %s", e)
|
||
return JSONResponse(
|
||
status_code=200,
|
||
content={
|
||
"results": [],
|
||
"status": "offline",
|
||
"message": "Tone3000 unavailable - using local defaults",
|
||
},
|
||
)
|
||
|
||
# ── Audio capture (looper/recorder) ──────────────────────
|
||
_capture_process = None
|
||
_capture_path = None
|
||
|
||
@app.post("/api/capture/start")
|
||
async def capture_start():
|
||
"""Start recording audio to a WAV file via arecord."""
|
||
nonlocal _capture_process, _capture_path
|
||
import subprocess, datetime
|
||
captures_dir = Path.home() / ".pedal" / "captures"
|
||
captures_dir.mkdir(parents=True, exist_ok=True)
|
||
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
path = captures_dir / f"capture_{ts}.wav"
|
||
try:
|
||
_capture_process = subprocess.Popen(
|
||
["arecord", "-f", "S16_LE", "-r", "48000", "-c", "2", str(path)],
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
||
)
|
||
_capture_path = path
|
||
return {"ok": True, "path": str(path), "filename": path.name}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
@app.post("/api/capture/stop")
|
||
async def capture_stop():
|
||
"""Stop the current recording."""
|
||
nonlocal _capture_process, _capture_path
|
||
if _capture_process is None:
|
||
raise HTTPException(status_code=400, detail="No active recording")
|
||
try:
|
||
_capture_process.terminate()
|
||
_capture_process.wait(timeout=5)
|
||
except Exception:
|
||
_capture_process.kill()
|
||
name = _capture_path.name if _capture_path else "unknown"
|
||
_capture_process = None
|
||
_capture_path = None
|
||
return {"ok": True, "filename": name}
|
||
|
||
@app.get("/api/captures")
|
||
async def list_captures():
|
||
"""List saved captures."""
|
||
captures_dir = Path.home() / ".pedal" / "captures"
|
||
if not captures_dir.is_dir():
|
||
return {"captures": []}
|
||
files = sorted(captures_dir.glob("*.wav"), key=lambda f: f.stat().st_mtime, reverse=True)
|
||
return {
|
||
"captures": [
|
||
{
|
||
"name": f.name,
|
||
"path": str(f),
|
||
"size": f.stat().st_size,
|
||
"size_display": f"{f.stat().st_size / 1024:.0f} KB" if f.stat().st_size < 1024*1024 else f"{f.stat().st_size / (1024*1024):.1f} MB",
|
||
"created": datetime.datetime.fromtimestamp(f.stat().st_mtime).isoformat(),
|
||
}
|
||
for f in files
|
||
]
|
||
}
|
||
|
||
@app.get("/api/captures/{filename}")
|
||
async def serve_capture(filename: str):
|
||
"""Serve a captured WAV file for download/playback."""
|
||
from fastapi.responses import FileResponse
|
||
captures_dir = Path.home() / ".pedal" / "captures"
|
||
file_path = captures_dir / filename
|
||
if not file_path.exists() or not file_path.is_file():
|
||
raise HTTPException(status_code=404, detail="Capture not found")
|
||
return FileResponse(str(file_path), media_type="audio/wav", filename=filename)
|
||
|
||
# ── Tonehub API alias routes ─────────────────────────────
|
||
# These are direct aliases for Tone3000 search, providing a
|
||
# simpler /api/tonehub/search path for the web UI.
|
||
|
||
@app.get("/api/tonehub/search")
|
||
async def tonehub_search_models(q: str = "", page: int = 0):
|
||
"""Alias for /api/models/tonedownload/search — search NAM models."""
|
||
return await tonedownload_search_models(q=q, page=page)
|
||
|
||
@app.get("/api/tonehub/search/irs")
|
||
async def tonehub_search_irs(q: str = "", page: int = 0):
|
||
"""Alias for /api/irs/tonedownload/search — search IRs."""
|
||
return await tonedownload_search_irs(q=q, page=page)
|
||
|
||
@app.post("/api/models/tonedownload/install")
|
||
async def tonedownload_install_model(data: dict):
|
||
"""Download a .nam file to the pedal's models directory."""
|
||
client = await self._get_tonedownload()
|
||
if not client:
|
||
raise HTTPException(status_code=503, detail="Tone3000 client unavailable")
|
||
url = data.get("download_url", "")
|
||
name = data.get("name", "model.nam")
|
||
if not url:
|
||
raise HTTPException(status_code=400, detail="download_url required")
|
||
model = ModelResult(
|
||
id=data.get("id", 0),
|
||
name=name,
|
||
download_url=url,
|
||
size_bytes=data.get("size"),
|
||
pack_name=data.get("pack_name"),
|
||
tone_title=data.get("display_name", name),
|
||
)
|
||
path = await client.download_model(model)
|
||
if not path:
|
||
raise HTTPException(status_code=500, detail="Download failed")
|
||
# Save metadata alongside the model file
|
||
import json
|
||
meta = {
|
||
"id": data.get("id"),
|
||
"name": data.get("name"),
|
||
"filename": data.get("filename"),
|
||
"size_bytes": data.get("size"),
|
||
"author": data.get("author"),
|
||
"author_avatar": data.get("author_avatar"),
|
||
"thumbnail": data.get("thumbnail"),
|
||
"tone_description": data.get("tone_description"),
|
||
"pack_name": data.get("pack_name"),
|
||
"architecture": data.get("architecture"),
|
||
"created_at": data.get("created_at"),
|
||
"download_date": datetime.datetime.now().isoformat(),
|
||
"source": "tone3000",
|
||
}
|
||
meta_path = path.with_suffix(".meta.json")
|
||
meta_path.write_text(json.dumps(meta, indent=2))
|
||
# Refresh the model list via NAM host
|
||
nam = self.deps.nam_host
|
||
if nam:
|
||
nam.list_available_models()
|
||
return {"ok": True, "path": str(path), "filename": path.name}
|
||
|
||
@app.post("/api/irs/tonedownload/install")
|
||
async def tonedownload_install_ir(data: dict):
|
||
"""Download a .wav IR to the pedal's IRs directory."""
|
||
client = await self._get_tonedownload()
|
||
if not client:
|
||
raise HTTPException(status_code=503, detail="Tone3000 client unavailable")
|
||
url = data.get("download_url", "")
|
||
name = data.get("name", "ir.wav")
|
||
if not url:
|
||
raise HTTPException(status_code=400, detail="download_url required")
|
||
ir = IRResult(
|
||
id=0,
|
||
name=name,
|
||
download_url=url,
|
||
size_bytes=data.get("size"),
|
||
tone_title=data.get("display_name", name),
|
||
)
|
||
path = await client.download_ir(ir)
|
||
if not path:
|
||
raise HTTPException(status_code=500, detail="Download failed")
|
||
# Save IR metadata
|
||
import json
|
||
meta = {
|
||
"id": data.get("id"),
|
||
"name": data.get("name"),
|
||
"filename": data.get("filename"),
|
||
"size_bytes": data.get("size"),
|
||
"author": data.get("author"),
|
||
"author_avatar": data.get("author_avatar"),
|
||
"thumbnail": data.get("thumbnail"),
|
||
"tone_description": data.get("tone_description"),
|
||
"created_at": data.get("created_at"),
|
||
"download_date": datetime.datetime.now().isoformat(),
|
||
"source": "tone3000",
|
||
}
|
||
meta_path = path.with_suffix(".meta.json")
|
||
meta_path.write_text(json.dumps(meta, indent=2))
|
||
ir_loader = self.deps.ir_loader
|
||
if ir_loader:
|
||
ir_loader.get_irs()
|
||
return {"ok": True, "path": str(path), "filename": path.name}
|
||
|
||
@app.get("/api/system")
|
||
async def get_system():
|
||
import os, subprocess
|
||
info = {"hostname": os.uname().nodename, "kernel": os.uname().release}
|
||
try:
|
||
with open("/proc/uptime") as f:
|
||
secs = float(f.read().split()[0])
|
||
info["uptime_seconds"] = round(secs)
|
||
d = int(secs // 86400); h = int((secs % 86400) // 3600); m = int((secs % 3600) // 60)
|
||
info["uptime"] = f"{d}d {h}h {m}m"
|
||
except: info["uptime"] = "?"
|
||
try:
|
||
r = subprocess.run(["cat","/sys/class/thermal/thermal_zone0/temp"], capture_output=True, text=True, timeout=3)
|
||
if r.returncode == 0: info["cpu_temp_c"] = round(int(r.stdout.strip()) / 1000, 1)
|
||
except: pass
|
||
try:
|
||
r = subprocess.run(["free","-m"], capture_output=True, text=True, timeout=3)
|
||
for line in r.stdout.splitlines():
|
||
p = line.split()
|
||
if p and p[0] == "Mem:":
|
||
info["memory_mb"] = {"total": int(p[1]), "used": int(p[2]), "free": int(p[3])}
|
||
except: pass
|
||
try:
|
||
r = subprocess.run(["df","-h","/"], capture_output=True, text=True, timeout=3)
|
||
parts = r.stdout.splitlines()[1].split()
|
||
info["disk"] = {"total": parts[1], "used": parts[2], "avail": parts[3], "pct": parts[4]}
|
||
except: pass
|
||
try:
|
||
with open("/proc/loadavg") as f:
|
||
info["load"] = f.read().split()[:3]
|
||
except: pass
|
||
return info
|
||
|
||
@app.get("/api/system/logs")
|
||
async def get_system_logs(lines: int = 50):
|
||
import subprocess
|
||
try:
|
||
r = subprocess.run(["journalctl","-u","pedal","--no-pager","-n",str(lines)], capture_output=True, text=True, timeout=10)
|
||
return {"logs": r.stdout.splitlines(), "exit_code": r.returncode}
|
||
except Exception as e:
|
||
return {"error": str(e), "logs": []}
|
||
|
||
@app.post("/api/system/hostname")
|
||
async def set_hostname(data: dict):
|
||
import subprocess
|
||
hostname = data.get("hostname", "").strip()
|
||
if not hostname or not hostname.replace("-","").replace("_","").isalnum():
|
||
raise HTTPException(status_code=400, detail="Invalid hostname")
|
||
try:
|
||
subprocess.run(["hostnamectl","set-hostname",hostname], check=True, timeout=10)
|
||
return {"ok": True, "hostname": hostname}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
@app.get("/api/network")
|
||
async def get_network():
|
||
import subprocess, re
|
||
info = {"ethernet": {}, "wifi": {}}
|
||
try:
|
||
r = subprocess.run(["ip","-4","addr","show","eth0"], capture_output=True, text=True, timeout=3)
|
||
m = re.search(r"inet (\S+)", r.stdout)
|
||
if m: info["ethernet"]["ip"] = m.group(1)
|
||
r2 = subprocess.run(["ip","link","show","eth0"], capture_output=True, text=True, timeout=3)
|
||
m2 = re.search(r"link/ether (\S+)", r2.stdout)
|
||
if m2: info["ethernet"]["mac"] = m2.group(1)
|
||
r3 = subprocess.run(["ip","route","show","default"], capture_output=True, text=True, timeout=3)
|
||
m3 = re.search(r"default via (\S+)", r3.stdout)
|
||
if m3: info["ethernet"]["gateway"] = m3.group(1)
|
||
except: pass
|
||
try:
|
||
with open("/etc/resolv.conf") as f:
|
||
dns = [line.split()[1] for line in f if line.startswith("nameserver")]
|
||
if dns: info["dns"] = dns
|
||
except: pass
|
||
return info
|
||
|
||
@app.post("/api/network/static")
|
||
async def set_network_static(data: dict):
|
||
import subprocess, os
|
||
ip = data.get("ip", "").strip()
|
||
gw = data.get("gateway", "").strip()
|
||
mask = data.get("netmask", "24").strip()
|
||
dns_servers = data.get("dns", "9.9.9.9 149.112.112.112")
|
||
if not ip:
|
||
raise HTTPException(status_code=400, detail="ip required")
|
||
config = f"""auto eth0
|
||
iface eth0 inet static
|
||
address {ip}/{mask}
|
||
gateway {gw}
|
||
dns-nameservers {dns_servers}
|
||
"""
|
||
try:
|
||
with open("/etc/network/interfaces.d/eth0", "w") as f:
|
||
f.write(config)
|
||
subprocess.run(["ifdown","eth0"], capture_output=True, timeout=10)
|
||
subprocess.run(["ifup","eth0"], capture_output=True, timeout=30)
|
||
return {"ok": True, "ip": ip, "mode": "static"}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
@app.post("/api/network/dhcp")
|
||
async def set_network_dhcp():
|
||
import subprocess, os
|
||
try:
|
||
os.remove("/etc/network/interfaces.d/eth0")
|
||
except FileNotFoundError: pass
|
||
try:
|
||
subprocess.run(["ifdown","eth0"], capture_output=True, timeout=10)
|
||
subprocess.run(["ifup","eth0"], capture_output=True, timeout=30)
|
||
except: pass
|
||
return {"ok": True, "mode": "dhcp"}
|
||
|
||
@app.get("/api/audio/devices")
|
||
async def list_audio_devices():
|
||
import subprocess, re
|
||
try:
|
||
r = subprocess.run(["aplay","-l"], capture_output=True, text=True, timeout=5)
|
||
cards = []
|
||
for line in r.stdout.splitlines():
|
||
m = re.match(r"card (\d+):.*?\[(.*?)\].*device (\d+):.*?\[(.*?)\]", line)
|
||
if m:
|
||
cards.append({"card": int(m.group(1)), "name": m.group(2).strip(),
|
||
"device": int(m.group(3)), "desc": m.group(4).strip()})
|
||
r2 = subprocess.run(["cat","/proc/asound/cards"], capture_output=True, text=True, timeout=3)
|
||
usb_id = None
|
||
for line in r2.stdout.splitlines():
|
||
if "USB" in line or "usb" in line.lower():
|
||
m = re.match(r"\s*(\d+)", line)
|
||
if m: usb_id = int(m.group(1))
|
||
# Add software PCM devices (bluealsa, etc.) as virtual cards
|
||
try:
|
||
r3 = subprocess.run(["aplay","-L"], capture_output=True, text=True, timeout=3)
|
||
for line in r3.stdout.splitlines():
|
||
name = line.strip()
|
||
if name.startswith("bluealsa") or name.startswith("bluealsa:"):
|
||
if not any(c.get("name","") == "Bluetooth Audio" for c in cards):
|
||
cards.append({"card": -1, "name": "Bluetooth Audio",
|
||
"device": 0, "desc": "BlueALSA (BT playback)"})
|
||
except: pass
|
||
return {"devices": cards, "usb_card": usb_id, "suggested": f"hw:{usb_id or 0},0"}
|
||
except Exception as e:
|
||
return {"error": str(e), "devices": []}
|
||
|
||
# ── Backing track source ────────────────────────────────────
|
||
|
||
@app.get("/api/backing-source")
|
||
async def get_backing_source():
|
||
"""Get current backing track source and available sources."""
|
||
import subprocess
|
||
audio = self.deps.audio_system
|
||
cfg = audio.config.__dict__ if audio and hasattr(audio.config, '__dict__') else {}
|
||
current = cfg.get("backing_source", "none")
|
||
# Check available JACK ports for possible sources
|
||
sources = [{"id": "none", "name": "None"}]
|
||
try:
|
||
r = subprocess.run(["jack_lsp"], capture_output=True, text=True, timeout=3)
|
||
for line in r.stdout.splitlines():
|
||
line = line.strip()
|
||
if line.endswith(":capture_1") or line.endswith(":capture_2"):
|
||
name = line.split(":")[0]
|
||
if name != "system" and name != "pi-multifx" and name != "bt_audio":
|
||
sources.append({"id": name, "name": f"{name} (capture)"})
|
||
# Always offer bluetooth if JACK has bt_audio ports
|
||
r2 = subprocess.run(["jack_lsp","bt_audio:capture_1"], capture_output=True, text=True, timeout=2)
|
||
if r2.returncode == 0:
|
||
sources.append({"id": "bt_audio", "name": "Bluetooth Audio"})
|
||
except: pass
|
||
return {"current": current, "sources": sources}
|
||
|
||
@app.post("/api/backing-source")
|
||
async def set_backing_source(data: dict):
|
||
"""Set backing track source and route it into the FX chain.
|
||
|
||
Body: { "source": "none" | "bt_audio" | "<jack_source>" }
|
||
"""
|
||
import subprocess
|
||
source = data.get("source", "none")
|
||
audio = self.deps.audio_system
|
||
# Save to config
|
||
if audio and hasattr(audio.config, '__dict__'):
|
||
audio.config.__dict__["backing_source"] = source
|
||
# Disconnect old backing connections
|
||
try:
|
||
r = subprocess.run(["jack_lsp","fx_in:input_2"], capture_output=True, text=True, timeout=2)
|
||
if r.returncode == 0:
|
||
subprocess.run(["jack_disconnect","bt_audio:capture_1","fx_in:input_2"],
|
||
capture_output=True, timeout=2)
|
||
subprocess.run(["jack_disconnect","bt_audio:capture_2","fx_in:input_2"],
|
||
capture_output=True, timeout=2)
|
||
except: pass
|
||
# Connect new source
|
||
if source != "none":
|
||
try:
|
||
if source == "bt_audio":
|
||
subprocess.run(["jack_connect","bt_audio:capture_1","fx_in:input_2"],
|
||
capture_output=True, timeout=3)
|
||
else:
|
||
subprocess.run(["jack_connect",f"{source}:capture_1","fx_in:input_2"],
|
||
capture_output=True, timeout=3)
|
||
except: pass
|
||
return {"ok": True, "source": source}
|
||
|
||
@app.get("/api/levels")
|
||
async def get_signal_levels():
|
||
levels = {"channels": {}}
|
||
ja = self.deps.jack_audio
|
||
if ja and hasattr(ja, 'get_levels'):
|
||
try: levels = ja.get_levels()
|
||
except: pass
|
||
if not levels.get("channels"):
|
||
for ch in ["guitar", "bass", "keys", "vocals", "backing_tracks"]:
|
||
levels["channels"][ch] = {"input_rms": 0, "input_peak": 0, "output_rms": 0, "output_peak": 0}
|
||
return levels
|
||
|
||
@app.delete("/api/models/{name:path}")
|
||
async def delete_model(name: str):
|
||
import os
|
||
from pathlib import Path
|
||
models_dir = Path(self.deps.config.get("models", {}).get("dir", "/home/oplabs/projects/pi-multifx-pedal/src/models/nam"))
|
||
target = models_dir / name
|
||
if not target.exists() or not target.is_file():
|
||
raise HTTPException(status_code=404, detail="Model not found")
|
||
if target.suffix not in (".nam", ".wav", ".onnx"):
|
||
raise HTTPException(status_code=400, detail="Not a model file")
|
||
os.remove(str(target))
|
||
nam = self.deps.nam_host
|
||
if nam: nam.list_available_models()
|
||
return {"ok": True, "deleted": name}
|
||
|
||
@app.delete("/api/irs/{name:path}")
|
||
async def delete_ir(name: str):
|
||
import os
|
||
from pathlib import Path
|
||
irs_dir = Path(self.deps.config.get("irs", {}).get("dir", "/home/oplabs/projects/pi-multifx-pedal/src/models/ir"))
|
||
target = irs_dir / name
|
||
if not target.exists() or not target.is_file():
|
||
raise HTTPException(status_code=404, detail="IR not found")
|
||
if target.suffix != ".wav":
|
||
raise HTTPException(status_code=400, detail="Not an IR file")
|
||
os.remove(str(target))
|
||
ir_loader = self.deps.ir_loader
|
||
if ir_loader: ir_loader.get_irs()
|
||
return {"ok": True, "deleted": name}
|
||
|
||
@app.get("/api/midi/mappings")
|
||
async def get_midi_mappings():
|
||
mappings = {}
|
||
if self.deps.midi and hasattr(self.deps.midi, 'get_mappings'):
|
||
try: mappings = self.deps.midi.get_mappings()
|
||
except: pass
|
||
return {"mappings": mappings}
|
||
|
||
@app.post("/api/midi/learn")
|
||
async def midi_learn(data: dict = None):
|
||
if data and "enable" in data:
|
||
enable = bool(data["enable"])
|
||
if self.deps.midi and hasattr(self.deps.midi, 'set_learn_mode'):
|
||
try: self.deps.midi.set_learn_mode(enable)
|
||
except: pass
|
||
return {"learn_active": enable}
|
||
if data and "param" in data:
|
||
if self.deps.midi and hasattr(self.deps.midi, 'set_mapping'):
|
||
try: self.deps.midi.set_mapping(data["param"], data.get("cc"))
|
||
except: pass
|
||
return {"mapped": data["param"]}
|
||
return {"learn_active": False, "mappings": {}}
|
||
|
||
@app.post("/api/backup")
|
||
async def create_backup():
|
||
import subprocess, os, datetime
|
||
backup_dir = "/tmp/pedal-backups"
|
||
os.makedirs(backup_dir, exist_ok=True)
|
||
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
filename = f"pedal-backup-{ts}.tar.gz"
|
||
outpath = os.path.join(backup_dir, filename)
|
||
cfg_path = os.path.expanduser("~/.pedal")
|
||
try:
|
||
subprocess.run(["tar","-czf",outpath,"-C",os.path.dirname(cfg_path),os.path.basename(cfg_path)], check=True, timeout=30)
|
||
size = os.path.getsize(outpath)
|
||
return {"ok": True, "filename": filename, "path": outpath, "size": size}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
@app.get("/api/backup/list")
|
||
async def list_backups():
|
||
import os, datetime
|
||
backup_dir = "/tmp/pedal-backups"
|
||
os.makedirs(backup_dir, exist_ok=True)
|
||
bks = []
|
||
for f in sorted(os.listdir(backup_dir), reverse=True):
|
||
fp = os.path.join(backup_dir, f)
|
||
if os.path.isfile(fp):
|
||
bks.append({"name": f, "size": os.path.getsize(fp),
|
||
"modified": datetime.datetime.fromtimestamp(os.path.getmtime(fp)).isoformat()})
|
||
return {"backups": bks}
|
||
|
||
@app.post("/api/restore")
|
||
async def restore_backup(data: dict):
|
||
import subprocess, os
|
||
filename = data.get("filename") if data else None
|
||
if not filename:
|
||
return {"error": "filename required"}
|
||
backup_path = os.path.join("/tmp/pedal-backups", filename)
|
||
if not os.path.exists(backup_path):
|
||
raise HTTPException(status_code=404, detail="Backup not found")
|
||
try:
|
||
subprocess.run(["tar","-xzf",backup_path,"-C",os.path.expanduser("~")], check=True, timeout=30)
|
||
return {"ok": True, "restored": filename}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
@app.get("/api/diagnostics")
|
||
async def get_diagnostics():
|
||
import subprocess
|
||
diag = {}
|
||
ja = self.deps.jack_audio
|
||
diag["jack"] = {"running": ja is not None and hasattr(ja, 'is_running') and ja.is_running()}
|
||
nam = self.deps.nam_host
|
||
diag["nam"] = {"loaded": nam is not None and nam.current_model is not None}
|
||
if diag["nam"]["loaded"]: diag["nam"]["model"] = str(nam.current_model)
|
||
midi = self.deps.midi
|
||
diag["midi"] = {"available": midi is not None}
|
||
try:
|
||
r = subprocess.run(["iwgetid"], capture_output=True, text=True, timeout=3)
|
||
diag["wifi"] = {"connected": r.returncode == 0, "ssid": r.stdout.strip() if r.returncode == 0 else None}
|
||
except: diag["wifi"] = {"connected": False}
|
||
bt = getattr(self.deps, 'bt_controller', None)
|
||
diag["bluetooth"] = {"available": bt is not None}
|
||
if self.deps.audio_system:
|
||
try:
|
||
ap = self.deps.audio_system.config
|
||
diag["audio"] = {"profile": getattr(ap, 'profile', '?'), "rate": getattr(ap, 'rate', '?')}
|
||
except: diag["audio"] = {"profile": "?"}
|
||
diag["api"] = {"status": "ok"}
|
||
return diag
|
||
|
||
|
||
|
||
# ── 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 ──────────────────────────────────
|
||
|
||
def _gather_wifi_state(self) -> dict:
|
||
"""Try to gather WiFi status without hard-failing."""
|
||
try:
|
||
from ..system.network import wifi_status, hotspot_status
|
||
ws = wifi_status()
|
||
hs = hotspot_status()
|
||
return {
|
||
"connected": ws.get("connected", False),
|
||
"ssid": ws.get("ssid"),
|
||
"ip": ws.get("ip"),
|
||
"signal": ws.get("signal", 0),
|
||
"hotspot_active": hs.get("active", False),
|
||
"hotspot_ssid": hs.get("ssid"),
|
||
"hotspot_clients": hs.get("clients", 0),
|
||
}
|
||
except Exception:
|
||
return {"connected": False, "ssid": None, "ip": None, "signal": 0,
|
||
"hotspot_active": False, "hotspot_ssid": None, "hotspot_clients": 0}
|
||
|
||
def _gather_bt_state(self) -> dict:
|
||
"""Try to gather Bluetooth status without hard-failing."""
|
||
try:
|
||
from ..system.bluetooth import bt_status, bt_midi_status
|
||
bs = bt_status()
|
||
ms = bt_midi_status()
|
||
return {
|
||
"available": bs.get("available", False),
|
||
"powered": bs.get("powered", False),
|
||
"name": bs.get("name"),
|
||
"address": bs.get("address"),
|
||
"discoverable": bs.get("discoverable", False),
|
||
"midi_enabled": ms.get("enabled", False),
|
||
"midi_running": ms.get("running", False),
|
||
}
|
||
except Exception:
|
||
return {"available": False, "powered": False, "name": None,
|
||
"address": None, "discoverable": False,
|
||
"midi_enabled": False, "midi_running": False}
|
||
|
||
# ── System state gathering ────────────────────────────────────
|
||
|
||
def _gather_system_stats(self) -> dict:
|
||
"""Gather system stats (CPU, audio config) without hard-failing."""
|
||
try:
|
||
import psutil
|
||
return {
|
||
"cpu_percent": psutil.cpu_percent(interval=None),
|
||
"sample_rate": SAMPLE_RATE if hasattr(self, 'deps') and self.deps.pipeline else 48000,
|
||
}
|
||
except Exception:
|
||
return {"cpu_percent": 0, "sample_rate": 48000}
|
||
|
||
# ── State gathering ──────────────────────────────────────────────
|
||
|
||
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 {
|
||
"connected": False,
|
||
"current_preset": None,
|
||
"bypass": False,
|
||
"tuner_enabled": False,
|
||
"tuner_frequency": 0.0,
|
||
"tuner_note": "--",
|
||
"tuner_cents": 0,
|
||
"tuner_string": -1,
|
||
"tuner_confidence": 0.0,
|
||
"master_volume": 0.8,
|
||
"routing_mode": "mono",
|
||
"routing_breakpoint": 7,
|
||
"nam_loaded": False,
|
||
"nam_model": None,
|
||
"ir_loaded": False,
|
||
"ir_name": None,
|
||
"input_level": 0,
|
||
"output_level": 0,
|
||
"cpu_percent": 0,
|
||
"sample_rate": 48000,
|
||
"wifi": {},
|
||
"bluetooth": {},
|
||
}
|
||
|
||
try:
|
||
bank = pm.current_bank
|
||
program = pm.current_program
|
||
preset = pm.load(bank, program) if pm else None
|
||
except Exception:
|
||
preset = None
|
||
|
||
def _safe_str(val) -> str | None:
|
||
"""Get a string representation, or None if the value is falsy/mock."""
|
||
if val is None:
|
||
return None
|
||
try:
|
||
s = str(val)
|
||
return s if s and not s.startswith("<MagicMock") else None
|
||
except Exception:
|
||
return None
|
||
|
||
current_model_obj = nam.current_model if nam else None
|
||
current_ir_obj = ir.current_ir if ir else None
|
||
current_model_name = _safe_str(current_model_obj.name if current_model_obj else None)
|
||
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 "—",
|
||
"bank": preset.bank if preset else 0,
|
||
"program": preset.program if preset else 0,
|
||
} if preset else None,
|
||
"bypass": pl._bypassed if pl else False,
|
||
"tuner_enabled": pl._tuner_enabled if pl else False,
|
||
"tuner_frequency": getattr(pl, '_tuner_frequency', 0.0),
|
||
"tuner_note": getattr(pl, '_tuner_note', '--'),
|
||
"tuner_cents": getattr(pl, '_tuner_cents', 0),
|
||
"tuner_string": getattr(pl, '_tuner_string', -1),
|
||
"tuner_confidence": getattr(pl, '_tuner_confidence', 0.0),
|
||
"master_volume": pl._master_volume if pl else 0.8,
|
||
"routing_mode": pl.routing_mode if pl else "mono",
|
||
"routing_breakpoint": pl.routing_breakpoint if pl else 7,
|
||
"nam_loaded": bool(nam and nam.is_loaded) if nam else False,
|
||
"nam_model": current_model_name,
|
||
"ir_loaded": bool(ir and ir.is_loaded) if ir else False,
|
||
"ir_name": current_ir_name,
|
||
"blocks": (
|
||
[
|
||
{
|
||
"fx_type": b.fx_type.value,
|
||
"block_id": b.block_id,
|
||
"enabled": b.enabled,
|
||
"bypass": b.bypass,
|
||
"params": dict(b.params),
|
||
"nam_model_path": b.nam_model_path,
|
||
"nam_model_name": Path(b.nam_model_path).stem if b.nam_model_path else "",
|
||
"ir_file_path": b.ir_file_path,
|
||
"ir_file_name": Path(b.ir_file_path).stem if b.ir_file_path else "",
|
||
"name": (
|
||
Path(b.nam_model_path).stem.replace("_", " ")
|
||
if b.nam_model_path
|
||
else Path(b.ir_file_path).stem.replace("_", " ")
|
||
if b.ir_file_path
|
||
else b.fx_type.value.replace("_", " ").title()
|
||
),
|
||
}
|
||
for b in (preset.chain if preset else [])
|
||
]
|
||
if preset
|
||
else []
|
||
),
|
||
"channel_mode": self._channel_mgr.mode,
|
||
# Audio levels from pipeline (RMS float32 normalized 0.0-1.0, scaled to 0-100)
|
||
"input_level": (
|
||
0
|
||
if pl is None or getattr(pl, '_input_level', None) is None
|
||
else max(0, min(100, round(0 if (v := float(getattr(pl, '_input_level', 0.0))) != v else v * 150)))
|
||
),
|
||
"output_level": (
|
||
0
|
||
if pl is None or getattr(pl, '_output_level', None) is None
|
||
else max(0, min(100, round(0 if (v := float(getattr(pl, '_output_level', 0.0))) != v else v * 150)))
|
||
),
|
||
# 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(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 all channel states.
|
||
|
||
Shape::
|
||
|
||
{
|
||
"channels": {
|
||
"guitar": { … },
|
||
"bass": { … },
|
||
"keys": { … },
|
||
"vocals": { … },
|
||
"backing_tracks": { … },
|
||
},
|
||
"active_channel": "guitar",
|
||
"channel_mode": "dual-mono",
|
||
"wifi": { … },
|
||
"bluetooth": { … },
|
||
"cpu_percent": …,
|
||
}
|
||
"""
|
||
guitar = self._gather_channel_state("guitar")
|
||
bass = self._gather_channel_state("bass")
|
||
keys = self._gather_channel_state("keys")
|
||
vocals = self._gather_channel_state("vocals")
|
||
backing = self._gather_channel_state("backing_tracks")
|
||
|
||
return {
|
||
"channels": {
|
||
"guitar": guitar,
|
||
"bass": bass,
|
||
"keys": keys,
|
||
"vocals": vocals,
|
||
"backing_tracks": backing,
|
||
},
|
||
"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),
|
||
keys.get("cpu_percent", 0),
|
||
vocals.get("cpu_percent", 0),
|
||
backing.get("cpu_percent", 0),
|
||
),
|
||
"sample_rate": SAMPLE_RATE,
|
||
}
|
||
|
||
# ── Tone3000 client lazy init ────────────────────────────────
|
||
|
||
async def _get_tonedownload(self) -> Optional[Tone3000Client]:
|
||
"""Lazily initialize the Tone3000 download client."""
|
||
if self._tonedownload is None:
|
||
from aiohttp import ClientSession
|
||
async with ClientSession() as session:
|
||
key = await discover_anon_key(session)
|
||
if not key:
|
||
# Fall back to hardcoded key
|
||
from ..system.tonedownload import SUPABASE_ANON_KEY
|
||
key = SUPABASE_ANON_KEY
|
||
self._tonedownload = Tone3000Client(anon_key=key)
|
||
return self._tonedownload
|
||
|
||
# ── Lifecycle ────────────────────────────────────────────────────
|
||
|
||
def start(self) -> None:
|
||
"""Start the web server in a background thread.
|
||
|
||
Uses a dedicated thread with its own asyncio event loop to
|
||
avoid conflicting with the main thread's blocking JACK loop.
|
||
"""
|
||
import threading
|
||
config = uvicorn.Config(
|
||
self._app,
|
||
host=self.host,
|
||
port=self.port,
|
||
log_level="info",
|
||
access_log=False,
|
||
)
|
||
|
||
def _run():
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
server = uvicorn.Server(config)
|
||
self._server = server
|
||
loop.run_until_complete(server.serve())
|
||
|
||
self._thread = threading.Thread(target=_run, daemon=True)
|
||
self._thread.start()
|
||
logger.info("Web server starting on http://%s:%d", self.host, self.port)
|
||
|
||
def stop(self) -> None:
|
||
"""Stop the web server gracefully."""
|
||
if self._server:
|
||
self._server.should_exit = True
|
||
logger.info("Web server stopped")
|
||
|
||
|
||
# ── FX parameter schemas ─────────────────────────────────────────────────────
|
||
# Used by the web UI to render appropriate controls per FX block type.
|
||
|
||
_FX_PARAM_SCHEMAS: dict[str, list[dict]] = {
|
||
"noise_gate": [
|
||
{"key": "threshold", "name": "Threshold", "type": "float", "min": 0.0, "max": 0.1, "step": 0.001, "default": 0.01},
|
||
{"key": "release", "name": "Release (ms)", "type": "float", "min": 10.0, "max": 500.0, "step": 5.0, "default": 100.0},
|
||
],
|
||
"compressor": [
|
||
{"key": "threshold", "name": "Threshold (dB)", "type": "float", "min": -60.0, "max": 0.0, "step": 1.0, "default": -20.0},
|
||
{"key": "ratio", "name": "Ratio", "type": "float", "min": 1.0, "max": 20.0, "step": 0.5, "default": 3.0},
|
||
{"key": "attack", "name": "Attack (ms)", "type": "float", "min": 0.5, "max": 50.0, "step": 0.5, "default": 5.0},
|
||
{"key": "release", "name": "Release (ms)", "type": "float", "min": 10.0, "max": 500.0, "step": 5.0, "default": 100.0},
|
||
{"key": "gain", "name": "Makeup Gain", "type": "float", "min": 0.0, "max": 5.0, "step": 0.1, "default": 1.0},
|
||
],
|
||
"boost": [
|
||
{"key": "gain_db", "name": "Gain (dB)", "type": "float", "min": 0.0, "max": 30.0, "step": 0.5, "default": 6.0},
|
||
],
|
||
"overdrive": [
|
||
{"key": "drive", "name": "Drive", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "tone", "name": "Tone", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "gain", "name": "Gain", "type": "float", "min": 0.0, "max": 3.0, "step": 0.1, "default": 1.0},
|
||
],
|
||
"distortion": [
|
||
{"key": "drive", "name": "Drive", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.7},
|
||
{"key": "tone", "name": "Tone", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "gain", "name": "Gain", "type": "float", "min": 0.0, "max": 3.0, "step": 0.1, "default": 1.0},
|
||
],
|
||
"fuzz": [
|
||
{"key": "drive", "name": "Drive", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.8},
|
||
{"key": "tone", "name": "Tone", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "gain", "name": "Gain", "type": "float", "min": 0.0, "max": 3.0, "step": 0.1, "default": 1.0},
|
||
],
|
||
"eq": [
|
||
{"key": "bass", "name": "Bass (dB)", "type": "float", "min": -12.0, "max": 12.0, "step": 0.5, "default": 0.0},
|
||
{"key": "mid", "name": "Mid (dB)", "type": "float", "min": -12.0, "max": 12.0, "step": 0.5, "default": 0.0},
|
||
{"key": "treble", "name": "Treble (dB)", "type": "float", "min": -12.0, "max": 12.0, "step": 0.5, "default": 0.0},
|
||
{"key": "bass_freq", "name": "Bass Freq (Hz)", "type": "float", "min": 40.0, "max": 500.0, "step": 10.0, "default": 200.0},
|
||
{"key": "mid_freq", "name": "Mid Freq (Hz)", "type": "float", "min": 200.0, "max": 5000.0, "step": 10.0, "default": 1000.0},
|
||
{"key": "treble_freq", "name": "Treble Freq (Hz)", "type": "float", "min": 1000.0, "max": 10000.0, "step": 10.0, "default": 3500.0},
|
||
{"key": "q", "name": "Q", "type": "float", "min": 0.1, "max": 10.0, "step": 0.1, "default": 0.707},
|
||
],
|
||
"chorus": [
|
||
{"key": "rate", "name": "Rate (Hz)", "type": "float", "min": 0.1, "max": 5.0, "step": 0.1, "default": 0.5},
|
||
{"key": "depth", "name": "Depth", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "delay", "name": "Delay (ms)", "type": "float", "min": 5.0, "max": 60.0, "step": 1.0, "default": 20.0},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"flanger": [
|
||
{"key": "rate", "name": "Rate (Hz)", "type": "float", "min": 0.05, "max": 3.0, "step": 0.05, "default": 0.25},
|
||
{"key": "depth", "name": "Depth", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "delay", "name": "Delay (ms)", "type": "float", "min": 0.5, "max": 15.0, "step": 0.5, "default": 5.0},
|
||
{"key": "feedback", "name": "Feedback", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.3},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"phaser": [
|
||
{"key": "rate", "name": "Rate (Hz)", "type": "float", "min": 0.05, "max": 5.0, "step": 0.05, "default": 0.4},
|
||
{"key": "depth", "name": "Depth", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "feedback", "name": "Feedback", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.3},
|
||
{"key": "stages", "name": "Stages", "type": "int", "min": 2, "max": 12, "step": 1, "default": 4},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"tremolo": [
|
||
{"key": "rate", "name": "Rate (Hz)", "type": "float", "min": 0.1, "max": 10.0, "step": 0.1, "default": 4.0},
|
||
{"key": "depth", "name": "Depth", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "shape", "name": "Waveform", "type": "select", "options": ["sine", "triangle", "square"], "default": "sine"},
|
||
],
|
||
"vibrato": [
|
||
{"key": "rate", "name": "Rate (Hz)", "type": "float", "min": 0.1, "max": 10.0, "step": 0.1, "default": 3.0},
|
||
{"key": "depth", "name": "Depth", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"delay": [
|
||
{"key": "time", "name": "Time (ms)", "type": "float", "min": 10.0, "max": 2000.0, "step": 5.0, "default": 400.0},
|
||
{"key": "feedback", "name": "Feedback", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.4},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "tap_tempo", "name": "Tap Tempo (ms)", "type": "float", "min": 20.0, "max": 2000.0, "step": 5.0, "default": 0.0},
|
||
],
|
||
"reverb": [
|
||
{"key": "size", "name": "Room Size", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "decay", "name": "Decay", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "damping", "name": "Damping", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.4},
|
||
{"key": "predelay", "name": "Pre-delay (ms)", "type": "float", "min": 0.0, "max": 100.0, "step": 5.0, "default": 30.0},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.3},
|
||
],
|
||
"volume": [
|
||
{"key": "level", "name": "Level", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.8},
|
||
],
|
||
# ── Amp / Cabinet ──────────────────────────────────────────────
|
||
"nam_amp": [
|
||
{"key": "model_path", "name": "Model Path", "type": "string", "default": ""},
|
||
],
|
||
"ir_cab": [
|
||
{"key": "wet", "name": "Wet Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 1.0},
|
||
{"key": "dry", "name": "Dry Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.0},
|
||
],
|
||
# ── Pitch & Frequency ──────────────────────────────────────────
|
||
"octaver": [
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"pitch_shifter": [
|
||
{"key": "shift", "name": "Shift (semitones)", "type": "float", "min": -12.0, "max": 12.0, "step": 0.5, "default": 0.0},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"harmonizer": [
|
||
{"key": "shift", "name": "Shift (semitones)", "type": "float", "min": -12.0, "max": 12.0, "step": 0.5, "default": 3.0},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"whammy": [
|
||
{"key": "bend", "name": "Bend", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.0},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.7},
|
||
],
|
||
"detune": [
|
||
{"key": "depth", "name": "Depth", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
# ── Modulation ─────────────────────────────────────────────────
|
||
"ring_modulator": [
|
||
{"key": "rate", "name": "Rate (Hz)", "type": "float", "min": 10.0, "max": 2000.0, "step": 10.0, "default": 100.0},
|
||
{"key": "depth", "name": "Depth", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"auto_wah": [
|
||
{"key": "sensitivity", "name": "Sensitivity", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "q", "name": "Q", "type": "float", "min": 0.5, "max": 10.0, "step": 0.1, "default": 2.0},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"envelope_filter": [
|
||
{"key": "decay", "name": "Decay", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "range", "name": "Range", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"rotary_speaker": [
|
||
{"key": "speed", "name": "Speed", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "drive", "name": "Drive", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.3},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"uni_vibe": [
|
||
{"key": "rate", "name": "Rate (Hz)", "type": "float", "min": 0.05, "max": 5.0, "step": 0.05, "default": 0.8},
|
||
{"key": "depth", "name": "Depth", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "feedback", "name": "Feedback", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.3},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"auto_pan": [
|
||
{"key": "rate", "name": "Rate (Hz)", "type": "float", "min": 0.05, "max": 5.0, "step": 0.05, "default": 0.3},
|
||
{"key": "depth", "name": "Depth", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.7},
|
||
{"key": "waveform", "name": "Waveform", "type": "select", "options": ["sine", "triangle", "square"], "default": "sine"},
|
||
],
|
||
"stereo_widener": [
|
||
{"key": "width", "name": "Width", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
# ── Drive & Saturation ─────────────────────────────────────────
|
||
"bitcrusher": [
|
||
{"key": "bits", "name": "Bits", "type": "int", "min": 1, "max": 16, "step": 1, "default": 8},
|
||
{"key": "rate", "name": "Rate Reduction", "type": "float", "min": 1.0, "max": 10.0, "step": 1.0, "default": 1.0},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"wavefolder": [
|
||
{"key": "gain", "name": "Gain", "type": "float", "min": 0.5, "max": 10.0, "step": 0.5, "default": 2.0},
|
||
{"key": "fold", "name": "Fold Stages", "type": "int", "min": 1, "max": 10, "step": 1, "default": 3},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"rectifier": [
|
||
{"key": "mode", "name": "Mode", "type": "select", "options": ["full", "half"], "default": "full"},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
# ── Dynamics ────────────────────────────────────────────────────
|
||
"expander": [
|
||
{"key": "threshold", "name": "Threshold (dB)", "type": "float", "min": -60.0, "max": 0.0, "step": 1.0, "default": -30.0},
|
||
{"key": "ratio", "name": "Ratio", "type": "float", "min": 1.0, "max": 20.0, "step": 0.5, "default": 3.0},
|
||
{"key": "attack", "name": "Attack (ms)", "type": "float", "min": 0.5, "max": 50.0, "step": 0.5, "default": 5.0},
|
||
{"key": "release", "name": "Release (ms)", "type": "float", "min": 10.0, "max": 500.0, "step": 5.0, "default": 100.0},
|
||
],
|
||
"de_esser": [
|
||
{"key": "frequency", "name": "Frequency (Hz)", "type": "float", "min": 2000.0, "max": 10000.0, "step": 100.0, "default": 6000.0},
|
||
{"key": "threshold", "name": "Threshold (dB)", "type": "float", "min": -50.0, "max": 0.0, "step": 1.0, "default": -30.0},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"transient_shaper": [
|
||
{"key": "attack", "name": "Attack Boost", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "sustain", "name": "Sustain", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
"sidechain_compressor": [
|
||
{"key": "threshold", "name": "Threshold (dB)", "type": "float", "min": -60.0, "max": 0.0, "step": 1.0, "default": -20.0},
|
||
{"key": "ratio", "name": "Ratio", "type": "float", "min": 1.0, "max": 20.0, "step": 0.5, "default": 4.0},
|
||
{"key": "attack", "name": "Attack (ms)", "type": "float", "min": 0.5, "max": 50.0, "step": 0.5, "default": 2.0},
|
||
{"key": "release", "name": "Release (ms)", "type": "float", "min": 10.0, "max": 500.0, "step": 5.0, "default": 50.0},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
# ── Filters & EQ ────────────────────────────────────────────────
|
||
"parametric_eq": [
|
||
{"key": "freq_0", "name": "Band 1 Freq (Hz)", "type": "float", "min": 20.0, "max": 20000.0, "step": 10.0, "default": 100.0},
|
||
{"key": "gain_0", "name": "Band 1 Gain (dB)", "type": "float", "min": -18.0, "max": 18.0, "step": 0.5, "default": 0.0},
|
||
{"key": "q_0", "name": "Band 1 Q", "type": "float", "min": 0.1, "max": 10.0, "step": 0.1, "default": 0.707},
|
||
{"key": "freq_1", "name": "Band 2 Freq (Hz)", "type": "float", "min": 20.0, "max": 20000.0, "step": 10.0, "default": 500.0},
|
||
{"key": "gain_1", "name": "Band 2 Gain (dB)", "type": "float", "min": -18.0, "max": 18.0, "step": 0.5, "default": 0.0},
|
||
{"key": "q_1", "name": "Band 2 Q", "type": "float", "min": 0.1, "max": 10.0, "step": 0.1, "default": 0.707},
|
||
{"key": "freq_2", "name": "Band 3 Freq (Hz)", "type": "float", "min": 20.0, "max": 20000.0, "step": 10.0, "default": 2000.0},
|
||
{"key": "gain_2", "name": "Band 3 Gain (dB)", "type": "float", "min": -18.0, "max": 18.0, "step": 0.5, "default": 0.0},
|
||
{"key": "q_2", "name": "Band 3 Q", "type": "float", "min": 0.1, "max": 10.0, "step": 0.1, "default": 0.707},
|
||
{"key": "freq_3", "name": "Band 4 Freq (Hz)", "type": "float", "min": 20.0, "max": 20000.0, "step": 10.0, "default": 5000.0},
|
||
{"key": "gain_3", "name": "Band 4 Gain (dB)", "type": "float", "min": -18.0, "max": 18.0, "step": 0.5, "default": 0.0},
|
||
{"key": "q_3", "name": "Band 4 Q", "type": "float", "min": 0.1, "max": 10.0, "step": 0.1, "default": 0.707},
|
||
],
|
||
"high_pass_filter": [
|
||
{"key": "frequency", "name": "Frequency (Hz)", "type": "float", "min": 20.0, "max": 20000.0, "step": 10.0, "default": 200.0},
|
||
{"key": "slope", "name": "Slope (dB/oct)", "type": "select", "options": [6, 12], "default": 12},
|
||
],
|
||
"low_pass_filter": [
|
||
{"key": "frequency", "name": "Frequency (Hz)", "type": "float", "min": 20.0, "max": 20000.0, "step": 10.0, "default": 5000.0},
|
||
{"key": "slope", "name": "Slope (dB/oct)", "type": "select", "options": [6, 12], "default": 12},
|
||
],
|
||
"band_pass_filter": [
|
||
{"key": "frequency", "name": "Frequency (Hz)", "type": "float", "min": 20.0, "max": 20000.0, "step": 10.0, "default": 1000.0},
|
||
{"key": "q", "name": "Q", "type": "float", "min": 0.1, "max": 20.0, "step": 0.1, "default": 0.707},
|
||
],
|
||
"notch_filter": [
|
||
{"key": "frequency", "name": "Frequency (Hz)", "type": "float", "min": 20.0, "max": 20000.0, "step": 10.0, "default": 60.0},
|
||
{"key": "q", "name": "Q", "type": "float", "min": 0.5, "max": 50.0, "step": 0.5, "default": 10.0},
|
||
],
|
||
"formant_filter": [
|
||
{"key": "vowel", "name": "Vowel", "type": "select", "options": ["a", "e", "i", "o", "u"], "default": "a"},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
],
|
||
# ── Time-Based ──────────────────────────────────────────────────
|
||
"ping_pong_delay": [
|
||
{"key": "time", "name": "Time (ms)", "type": "float", "min": 10.0, "max": 2000.0, "step": 5.0, "default": 400.0},
|
||
{"key": "feedback", "name": "Feedback", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.3},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.4},
|
||
],
|
||
"multi_tap_delay": [
|
||
{"key": "pattern", "name": "Pattern", "type": "select", "options": ["quarter", "dotted", "triplet"], "default": "quarter"},
|
||
{"key": "feedback", "name": "Feedback", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.3},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.4},
|
||
],
|
||
"reverse_delay": [
|
||
{"key": "time", "name": "Time (ms)", "type": "float", "min": 50.0, "max": 2000.0, "step": 5.0, "default": 400.0},
|
||
{"key": "feedback", "name": "Feedback", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.3},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.4},
|
||
],
|
||
"tape_echo": [
|
||
{"key": "time", "name": "Time (ms)", "type": "float", "min": 50.0, "max": 2000.0, "step": 5.0, "default": 300.0},
|
||
{"key": "wow", "name": "Wow/Flutter", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.3},
|
||
{"key": "high_cut", "name": "High Cut", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "feedback", "name": "Feedback", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.3},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.4},
|
||
],
|
||
"shimmer_reverb": [
|
||
{"key": "shift", "name": "Shift (semitones)", "type": "float", "min": -12.0, "max": 12.0, "step": 0.5, "default": 0.0},
|
||
{"key": "decay", "name": "Decay", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.4},
|
||
],
|
||
"looper": [
|
||
{"key": "record", "name": "Record", "type": "bool", "default": 0},
|
||
{"key": "overdub", "name": "Overdub", "type": "bool", "default": 0},
|
||
{"key": "play", "name": "Play", "type": "bool", "default": 0},
|
||
{"key": "stop", "name": "Stop", "type": "bool", "default": 0},
|
||
],
|
||
# ── Ambience ────────────────────────────────────────────────────
|
||
"early_reflections": [
|
||
{"key": "size", "name": "Room Size", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.5},
|
||
{"key": "decay", "name": "Decay", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.4},
|
||
{"key": "mix", "name": "Mix", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.4},
|
||
],
|
||
}
|
||
|
||
|
||
# ── Standalone runner for development ─────────────────────────────────────────
|
||
|
||
def main():
|
||
"""Run the web server standalone (for dev/testing without the full pedal)."""
|
||
logging.basicConfig(level=logging.INFO)
|
||
|
||
import uvicorn
|
||
deps = WebServerDeps()
|
||
ws = WebServer(deps)
|
||
ws.start()
|
||
|
||
logger.info("Dev server running on http://%s:%d", ws.host, ws.port)
|
||
try:
|
||
loop = asyncio.get_event_loop()
|
||
loop.run_forever()
|
||
except KeyboardInterrupt:
|
||
pass
|
||
finally:
|
||
ws.stop()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |