Build: Web UI + mDNS (pedal.local)
FastAPI web server with REST + WebSocket, 5 Jinja2 HTML pages, mobile-friendly dark theme CSS, page-specific JS controllers, and Avahi mDNS service advertisement for pedal.local. Files: src/web/server.py — FastAPI app with 17 REST endpoints + /ws WebSocket src/web/__init__.py — Package init with WebServer, WebServerDeps exports src/web/templates/*.html — Dashboard, Presets, Models, IRs, Settings (base) src/web/static/style.css — Dark mobile-first CSS (80-82% max-width) src/web/static/websocket.js — Auto-reconnecting WebSocket manager src/web/static/app.js — Dashboard live updates + REST helpers src/web/static/presets.js — Bank/preset CRUD modal UI src/web/static/models.js — NAM model list/load/upload src/web/static/irs.js — IR file list/load/upload etc/avahi/pi-multifx-pedal.service — Avahi service advertisement scripts/setup-mdns.sh — One-shot mDNS setup (avahi, hostname, restart) main.py — Wire WebServer into boot/shutdown requirements.txt — Add fastapi, uvicorn, jinja2, python-multipart tests/test_web.py — 36 tests (pages, REST, WebSocket, errors) All 36 tests pass.
This commit is contained in:
@@ -0,0 +1,723 @@
|
||||
"""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 logging
|
||||
import time
|
||||
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
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.templating import Jinja2Templates
|
||||
|
||||
from ..dsp.nam_host import NAMHost
|
||||
from ..dsp.ir_loader import IRLoader
|
||||
from ..dsp.pipeline import AudioPipeline
|
||||
from ..presets.manager import PresetManager
|
||||
from ..presets.types import FXBlock, FXType, Preset
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ── Defaults ─────────────────────────────────────────────────────────────────
|
||||
|
||||
DEFAULT_HOST = "0.0.0.0"
|
||||
DEFAULT_PORT = 8080
|
||||
TEMPLATES_DIR = Path(__file__).resolve().parent / "templates"
|
||||
STATIC_DIR = Path(__file__).resolve().parent / "static"
|
||||
|
||||
# ── Dependency container ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@dataclass
|
||||
class WebServerDeps:
|
||||
"""Dependencies injected into the web server from the host pedal app."""
|
||||
|
||||
presets: Optional[PresetManager] = None
|
||||
pipeline: Optional[AudioPipeline] = None
|
||||
nam_host: Optional[NAMHost] = None
|
||||
ir_loader: Optional[IRLoader] = None
|
||||
|
||||
@property
|
||||
def is_connected(self) -> bool:
|
||||
"""True if we're wired to a live pedal instance."""
|
||||
return self.presets is not None
|
||||
|
||||
|
||||
# ── 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._server: Optional[uvicorn.Server] = None
|
||||
self._task: Optional[asyncio.Task] = None
|
||||
self._app = self._build_app()
|
||||
|
||||
# ── App factory ─────────────────────────────────────────────────────
|
||||
|
||||
def _build_app(self) -> FastAPI:
|
||||
|
||||
templates = Jinja2Templates(directory=str(TEMPLATES_DIR))
|
||||
|
||||
@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 dashboard(request: Request):
|
||||
"""Main dashboard — current preset, bypass, tuner, volume."""
|
||||
state = self._gather_state()
|
||||
ctx = {"request": request}
|
||||
ctx.update(state)
|
||||
return templates.TemplateResponse(request, "dashboard.html", ctx)
|
||||
|
||||
@app.get("/presets", response_class=HTMLResponse)
|
||||
async def preset_manager(request: Request):
|
||||
"""Preset manager page — browse banks, load/save/edit."""
|
||||
return templates.TemplateResponse(request, "presets.html", {"request": request})
|
||||
|
||||
@app.get("/models", response_class=HTMLResponse)
|
||||
async def model_manager(request: Request):
|
||||
"""NAM model manager page."""
|
||||
return templates.TemplateResponse(request, "models.html", {"request": request})
|
||||
|
||||
@app.get("/irs", response_class=HTMLResponse)
|
||||
async def ir_manager(request: Request):
|
||||
"""IR manager page."""
|
||||
return templates.TemplateResponse(request, "irs.html", {"request": request})
|
||||
|
||||
@app.get("/settings", response_class=HTMLResponse)
|
||||
async def settings_page(request: Request):
|
||||
"""Settings page."""
|
||||
state = self._gather_state()
|
||||
ctx = {"request": request}
|
||||
ctx.update(state)
|
||||
return templates.TemplateResponse(request, "settings.html", ctx)
|
||||
|
||||
# ── REST API ────────────────────────────────────────────────
|
||||
|
||||
@app.get("/api/state")
|
||||
async def get_state():
|
||||
"""Full current pedal state snapshot."""
|
||||
state = self._gather_state()
|
||||
if not state:
|
||||
raise HTTPException(status_code=503, detail="Pedal not connected")
|
||||
return state
|
||||
|
||||
@app.get("/api/presets")
|
||||
async def list_presets():
|
||||
"""List all banks and their presets."""
|
||||
pm = self.deps.presets
|
||||
if not pm:
|
||||
raise HTTPException(status_code=503, detail="Preset manager not available")
|
||||
banks = pm.list_banks()
|
||||
result = []
|
||||
for bank in banks:
|
||||
presets_in_bank = []
|
||||
for prog in range(4):
|
||||
try:
|
||||
p = pm.load(bank.number, prog)
|
||||
presets_in_bank.append({
|
||||
"name": p.name,
|
||||
"bank": p.bank,
|
||||
"program": p.program,
|
||||
"chain": [
|
||||
{
|
||||
"fx_type": b.fx_type.value,
|
||||
"enabled": b.enabled,
|
||||
"bypass": b.bypass,
|
||||
"params": dict(b.params),
|
||||
"nam_model_path": b.nam_model_path,
|
||||
"ir_file_path": b.ir_file_path,
|
||||
}
|
||||
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):
|
||||
"""Get a specific preset."""
|
||||
pm = self.deps.presets
|
||||
if not pm:
|
||||
raise HTTPException(status_code=503)
|
||||
try:
|
||||
p = pm.load(bank, program)
|
||||
return {
|
||||
"name": p.name,
|
||||
"bank": p.bank,
|
||||
"program": p.program,
|
||||
"chain": [
|
||||
{
|
||||
"fx_type": b.fx_type.value,
|
||||
"enabled": b.enabled,
|
||||
"bypass": b.bypass,
|
||||
"params": dict(b.params),
|
||||
"nam_model_path": b.nam_model_path,
|
||||
"ir_file_path": b.ir_file_path,
|
||||
}
|
||||
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):
|
||||
"""Activate a preset (load into pipeline)."""
|
||||
pm = self.deps.presets
|
||||
if not pm:
|
||||
raise HTTPException(status_code=503)
|
||||
try:
|
||||
preset = pm.select(bank, program)
|
||||
pm.save_state()
|
||||
state = self._gather_state()
|
||||
await self._manager.broadcast({
|
||||
"type": "preset_changed",
|
||||
"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):
|
||||
"""Save/update a preset."""
|
||||
pm = self.deps.presets
|
||||
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}"),
|
||||
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):
|
||||
"""Delete a preset."""
|
||||
pm = self.deps.presets
|
||||
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))
|
||||
|
||||
@app.post("/api/bypass")
|
||||
async def toggle_bypass(data: Optional[dict] = None):
|
||||
"""Toggle or set bypass state."""
|
||||
pl = self.deps.pipeline
|
||||
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",
|
||||
"bypass": pl._bypassed,
|
||||
})
|
||||
return {"ok": True, "bypass": pl._bypassed}
|
||||
|
||||
@app.post("/api/tuner")
|
||||
async def toggle_tuner(data: Optional[dict] = None):
|
||||
"""Toggle tuner mode."""
|
||||
pl = self.deps.pipeline
|
||||
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",
|
||||
"tuner_enabled": pl._tuner_enabled,
|
||||
})
|
||||
return {"ok": True, "tuner_enabled": pl._tuner_enabled}
|
||||
|
||||
@app.put("/api/volume")
|
||||
async def set_volume(data: dict):
|
||||
"""Set master volume (0.0-1.0)."""
|
||||
pl = self.deps.pipeline
|
||||
if not pl:
|
||||
raise HTTPException(status_code=503)
|
||||
vol = max(0.0, min(1.0, float(data.get("volume", 0.8))))
|
||||
pl._master_volume = vol
|
||||
return {"ok": True, "volume": vol}
|
||||
|
||||
@app.get("/api/models")
|
||||
async def list_models():
|
||||
"""List available NAM models."""
|
||||
nam = self.deps.nam_host
|
||||
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.post("/api/models/load")
|
||||
async def load_model(data: dict):
|
||||
"""Load a NAM model by path."""
|
||||
nam = self.deps.nam_host
|
||||
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():
|
||||
"""Unload the current NAM model."""
|
||||
nam = self.deps.nam_host
|
||||
if not nam:
|
||||
raise HTTPException(status_code=503)
|
||||
nam.unload()
|
||||
return {"ok": True}
|
||||
|
||||
@app.get("/api/irs")
|
||||
async def list_irs():
|
||||
"""List available IR files."""
|
||||
ir = self.deps.ir_loader
|
||||
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):
|
||||
"""Load an IR file by path."""
|
||||
ir = self.deps.ir_loader
|
||||
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():
|
||||
"""Unload the current IR."""
|
||||
ir = self.deps.ir_loader
|
||||
if not ir:
|
||||
raise HTTPException(status_code=503)
|
||||
ir.unload()
|
||||
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}
|
||||
|
||||
@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)
|
||||
try:
|
||||
# Send initial state snapshot on connect
|
||||
state = self._gather_state()
|
||||
await ws.send_text(json.dumps({"type": "connected", "state": state}))
|
||||
|
||||
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":
|
||||
state = self._gather_state()
|
||||
await ws.send_text(json.dumps({"type": "state", "state": state}))
|
||||
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)
|
||||
|
||||
return app
|
||||
|
||||
# ── State gathering ──────────────────────────────────────────────
|
||||
|
||||
def _gather_state(self) -> dict[str, Any]:
|
||||
"""Collect a snapshot of the current pedal state."""
|
||||
pm = self.deps.presets
|
||||
pl = self.deps.pipeline
|
||||
nam = self.deps.nam_host
|
||||
ir = self.deps.ir_loader
|
||||
|
||||
if not pm:
|
||||
return {"connected": False}
|
||||
|
||||
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 {
|
||||
"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,
|
||||
"master_volume": pl._master_volume if pl else 0.8,
|
||||
"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,
|
||||
}
|
||||
|
||||
# ── 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": "low_gain", "name": "Low (dB)", "type": "float", "min": -12.0, "max": 12.0, "step": 0.5, "default": 0.0},
|
||||
{"key": "mid_gain", "name": "Mid (dB)", "type": "float", "min": -12.0, "max": 12.0, "step": 0.5, "default": 0.0},
|
||||
{"key": "high_gain", "name": "High (dB)", "type": "float", "min": -12.0, "max": 12.0, "step": 0.5, "default": 0.0},
|
||||
],
|
||||
"chorus": [
|
||||
{"key": "rate", "name": "Rate (Hz)", "type": "float", "min": 0.1, "max": 5.0, "step": 0.1, "default": 1.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},
|
||||
],
|
||||
"flanger": [
|
||||
{"key": "rate", "name": "Rate (Hz)", "type": "float", "min": 0.05, "max": 3.0, "step": 0.05, "default": 0.5},
|
||||
{"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},
|
||||
],
|
||||
"phaser": [
|
||||
{"key": "rate", "name": "Rate (Hz)", "type": "float", "min": 0.05, "max": 5.0, "step": 0.05, "default": 0.5},
|
||||
{"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},
|
||||
],
|
||||
"tremolo": [
|
||||
{"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},
|
||||
],
|
||||
"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},
|
||||
],
|
||||
"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.4},
|
||||
{"key": "damping", "name": "Damping", "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.3},
|
||||
],
|
||||
"volume": [
|
||||
{"key": "level", "name": "Level", "type": "float", "min": 0.0, "max": 1.0, "step": 0.01, "default": 0.8},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
# ── 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://0.0.0.0:8080")
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
loop.run_forever()
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
finally:
|
||||
ws.stop()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user