shawn/pi-multifx-pedal
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

v2 UI at root, instrument type selector, Gitea push

Browse Source
This commit is contained in:
Shawn
2026-06-15 05:53:11 +00:00
parent e2f81fd309
commit a69aee357c
2 changed files with 1385 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/web/server.py
+655 -46
View File
@@ -42,6 +42,10 @@ from ..system.tonedownload import (
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__)
@@ -207,6 +211,15 @@ class WebServerDeps:
# 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)."""
@@ -316,12 +329,25 @@ class WebServer:
logger.info("Mounting React UI from %s at /ui", UI_DIST_DIR)
app.mount("/ui", StaticFiles(directory=str(UI_DIST_DIR), html=True), name="ui")
# Mount V2 React UI (if dist exists)
V2_DIST_DIR = Path(__file__).resolve().parent / "ui-v2-dist"
if V2_DIST_DIR.is_dir():
logger.info("Mounting V2 React UI from %s at /v2", V2_DIST_DIR)
app.mount("/v2", StaticFiles(directory=str(V2_DIST_DIR), html=True), name="ui-v2")
# ── HTML pages ──────────────────────────────────────────────
@app.get("/", response_class=HTMLResponse)
async def root(request: Request):
"""Serve the React SPA (or legacy dashboard as fallback)."""
# If we have a React build, serve it at /
"""Serve the v2 UI (or React SPA / legacy dashboard as fallback)."""
# Prefer the v2 UI if available
v2_dir = Path(__file__).resolve().parent / "ui-v2-dist"
if v2_dir.is_dir():
v2_index = v2_dir / "index.html"
if v2_index.exists():
content = v2_index.read_text()
return HTMLResponse(content)
# Fallback: React build
if UI_DIST_DIR and UI_DIST_DIR.is_dir():
index_html = UI_DIST_DIR / "index.html"
if index_html.exists():
@@ -394,11 +420,21 @@ class WebServer:
"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
],
@@ -429,11 +465,21 @@ class WebServer:
"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
],
@@ -535,6 +581,7 @@ class WebServer:
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),
@@ -562,6 +609,7 @@ class WebServer:
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),
@@ -721,6 +769,7 @@ class WebServer:
"chain": [
{
"fx_type": b.fx_type.value,
"block_id": b.block_id,
"enabled": b.enabled,
"bypass": b.bypass,
"params": dict(b.params),
@@ -737,13 +786,13 @@ class WebServer:
if not pl:
raise HTTPException(status_code=503)
if data and "bypass" in data:
pl._bypassed = bool(data["bypass"])
pl.bypassed = bool(data["bypass"])
else:
pl._bypassed = not pl._bypassed
pl.bypassed = not pl.bypassed
await self._manager.broadcast({
"type": "bypass_changed",
"channel": channel,
"bypass": pl._bypassed,
"bypass": pl.bypassed,
})
return {"ok": True, "bypass": pl._bypassed}
@@ -790,6 +839,12 @@ class WebServer:
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) ──
@@ -801,6 +856,12 @@ class WebServer:
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) ─────
@@ -811,40 +872,48 @@ class WebServer:
if not pl:
raise HTTPException(status_code=503)
if data and "bypass" in data:
pl._bypassed = bool(data["bypass"])
pl.bypassed = bool(data["bypass"])
else:
pl._bypassed = not pl._bypassed
pl.bypassed = not pl.bypassed
await self._manager.broadcast({
"type": "bypass_changed",
"channel": channel,
"bypass": pl._bypassed,
"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 its fx_type id — sets both enabled and bypass."""
"""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)
block_id = data.get("id")
# 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 'id' or 'enabled'")
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.fx_type.value == block_id:
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) # <-- added: sync bypass with 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:
@@ -853,12 +922,13 @@ class WebServer:
# ── 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 its fx_type id."""
"""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)
block_id = data.get("id")
# 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:
@@ -866,7 +936,7 @@ class WebServer:
program = pm.current_program
preset = pm.load(bank, program, channel=Channel(channel))
for b in preset.chain:
if b.fx_type.value == block_id:
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":
@@ -874,8 +944,10 @@ class WebServer:
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:
@@ -886,6 +958,14 @@ class WebServer:
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:
@@ -1015,6 +1095,10 @@ class WebServer:
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")
@@ -1064,6 +1148,10 @@ class WebServer:
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 ─────────────────────────────────
@@ -1321,64 +1409,291 @@ class WebServer:
@app.post("/api/audio/profile")
async def set_audio_profile(data: dict):
"""Switch JACK latency profile (restarts JACK server).
"""Switch JACK latency profile and/or set custom period/rate.
Body: { "profile": "standard" | "low" }
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 }
The switch takes effect immediately by restarting JACK with the
new period / nperiods / priority settings.
Changes are saved to ~/.pedal/config.yaml so they survive reboot.
The switch restarts JACK (brief audio dropout ~1s).
"""
profile_key = data.get("profile", "")
if profile_key not in LATENCY_PROFILES:
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())}",
)
audio_sys = self.deps.audio_system
if not audio_sys:
raise HTTPException(status_code=503, detail="Audio system not available")
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
if old_key == profile_key:
return {"ok": True, "profile": profile_key, "restarted": False}
# Apply new profile
audio_sys.config.profile = profile_key
profile = audio_sys.config.latency_profile
logger.info("Switching audio profile: %s → %s (period=%d, nperiods=%d)",
old_key, profile_key, profile["period"], profile["nperiods"])
# Stop JACK audio client first (holds a live JACK C client)
# 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"],
)
# Stop JACK audio client first
jack_client = self.deps.jack_audio
if jack_client:
jack_client.stop()
# Restart JACK server with new parameters
ok = audio_sys.restart_jack(timeout=10)
# 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) ──
# This interrupts the audio callback thread, releasing any lock
# that jack_client.stop() would need.
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
ok = audio_sys.start_jack(timeout=10)
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=10)
if jack_client:
jack_client.start()
raise HTTPException(status_code=500, detail=f"Failed to start JACK with profile {profile_key!r} — rolled back to {old_key!r}")
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",
)
# Restart JACK audio client
if jack_client:
jack_client.start()
# Sync NAM engine block size with new period
try:
jack_client.start()
except Exception as e:
logger.warning("jack_client.start() failed after profile switch: %s", e)
# Sync NAM engine block size
nam_host = self.deps.nam_host
if nam_host and hasattr(nam_host, 'set_block_size'):
nam_host.set_block_size(profile["period"])
nam_host.set_block_size(target_profile["period"])
# ── 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": profile_key,
"period": profile["period"],
"nperiods": profile["nperiods"],
"profile": effective_key,
"period": target_profile["period"],
"rate": target_profile["rate"],
})
return {
"ok": True,
"profile": profile_key,
"period": profile["period"],
"nperiods": profile["nperiods"],
"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
if jack_client:
jack_client.stop()
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()
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}")
@@ -1680,6 +1995,275 @@ class WebServer:
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))
return {"devices": cards, "usb_card": usb_id, "suggested": f"hw:{usb_id or 0},0"}
except Exception as e:
return {"error": str(e), "devices": []}
@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.
@@ -1853,6 +2437,31 @@ class WebServer:
"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": (