282 lines
10 KiB
Python
282 lines
10 KiB
Python
"""
|
|
FX Inserts API — manage PiPedal plugin insert configurations on mixer channels.
|
|
|
|
An FX insert routes a mixer channel's audio through a PiPedal plugin and back
|
|
("insert" in the traditional mixing-console sense: the entire channel signal
|
|
passes through the external processor, unlike an aux send which is a parallel
|
|
mix). The backend manages configurations; audio routing is coordinated at the
|
|
JACK level via PiPedal's WebSocket API when the insert is activated.
|
|
|
|
Data flow:
|
|
Browser → REST API (insert CRUD) → fx_inserts.json (config persistence)
|
|
Browser → REST API (route) → PiPedal WS (JACK connect/disconnect)
|
|
|
|
The routing strategy for a channel insert:
|
|
mixer-daemon:channel_{N}_send → pipedal:{plugin_id}:audio_in_1
|
|
pipedal:{plugin_id}:audio_out_1 → mixer-daemon:channel_{N}_return
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
logger = logging.getLogger("fx-inserts")
|
|
|
|
# File-based storage for FX insert configurations
|
|
FX_INSERTS_FILE = Path(__file__).resolve().parent / "fx_inserts.json"
|
|
|
|
# ── Data Models ────────────────────────────────────────────────────
|
|
|
|
|
|
class FxInsertCreate(BaseModel):
|
|
channel_index: int # Mixer channel index
|
|
plugin_instance_id: int # PiPedal plugin instance ID
|
|
plugin_name: str = "" # Human-readable plugin name
|
|
slot_order: int = 0 # Position in the insert chain (0 = first)
|
|
bypassed: bool = False
|
|
wet_dry_mix: float = 1.0 # 0.0 (dry) to 1.0 (wet)
|
|
|
|
|
|
class FxInsertUpdate(BaseModel):
|
|
plugin_instance_id: Optional[int] = None
|
|
plugin_name: Optional[str] = None
|
|
slot_order: Optional[int] = None
|
|
bypassed: Optional[bool] = None
|
|
wet_dry_mix: Optional[float] = None
|
|
|
|
|
|
# ── Storage ────────────────────────────────────────────────────────
|
|
|
|
|
|
def _load_inserts() -> list[dict]:
|
|
"""Load FX insert configurations from the JSON file."""
|
|
if not FX_INSERTS_FILE.exists():
|
|
return []
|
|
try:
|
|
data = json.loads(FX_INSERTS_FILE.read_text())
|
|
return data if isinstance(data, list) else []
|
|
except (json.JSONDecodeError, Exception) as e:
|
|
logger.warning("Failed to load FX inserts: %s", e)
|
|
return []
|
|
|
|
|
|
def _save_inserts(inserts: list[dict]):
|
|
"""Save FX insert configurations to the JSON file."""
|
|
FX_INSERTS_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
FX_INSERTS_FILE.write_text(json.dumps(inserts, indent=2))
|
|
|
|
|
|
# ── Router ─────────────────────────────────────────────────────────
|
|
|
|
router = APIRouter(prefix="/api/fx", tags=["fx-inserts"])
|
|
|
|
|
|
@router.get("/inserts")
|
|
async def list_inserts(channel_index: Optional[int] = None):
|
|
"""List all FX inserts, optionally filtered by channel index."""
|
|
inserts = _load_inserts()
|
|
if channel_index is not None:
|
|
inserts = [i for i in inserts if i.get("channel_index") == channel_index]
|
|
inserts.sort(key=lambda x: x.get("slot_order", 0))
|
|
return {"inserts": inserts}
|
|
|
|
|
|
@router.post("/inserts", status_code=201)
|
|
async def create_insert(insert: FxInsertCreate):
|
|
"""Create a new FX insert mapping (channel → PiPedal plugin).
|
|
|
|
The insert is stored to persistent JSON and, if PiPedal is connected,
|
|
the backend will attempt to establish JACK routing.
|
|
"""
|
|
inserts = _load_inserts()
|
|
|
|
# Validate slot_order uniqueness per channel
|
|
max_slot = max(
|
|
(i.get("slot_order", 0) for i in inserts
|
|
if i.get("channel_index") == insert.channel_index),
|
|
default=-1,
|
|
)
|
|
slot = insert.slot_order if insert.slot_order <= max_slot + 1 else max_slot + 1
|
|
|
|
new_insert = {
|
|
"id": str(uuid.uuid4()),
|
|
"channel_index": insert.channel_index,
|
|
"plugin_instance_id": insert.plugin_instance_id,
|
|
"plugin_name": insert.plugin_name or f"Plugin {insert.plugin_instance_id}",
|
|
"slot_order": slot,
|
|
"bypassed": insert.bypassed,
|
|
"wet_dry_mix": min(1.0, max(0.0, insert.wet_dry_mix)),
|
|
}
|
|
inserts.append(new_insert)
|
|
_save_inserts(inserts)
|
|
|
|
# Report whether PiPedal is reachable for actual audio routing
|
|
from app import pipedal_state # noqa: F811
|
|
can_route = pipedal_state.get("connected", False)
|
|
new_insert["can_route"] = can_route
|
|
|
|
logger.info(
|
|
"FX insert created: channel %d → plugin %d (slot %d, bypass=%s)",
|
|
insert.channel_index, insert.plugin_instance_id, slot, insert.bypassed,
|
|
)
|
|
return {"insert": new_insert}
|
|
|
|
|
|
@router.put("/inserts/{insert_id}")
|
|
async def update_insert(insert_id: str, update: FxInsertUpdate):
|
|
"""Update an existing FX insert (bypass, wet/dry, order, plugin)."""
|
|
inserts = _load_inserts()
|
|
for ins in inserts:
|
|
if ins["id"] == insert_id:
|
|
if update.plugin_instance_id is not None:
|
|
ins["plugin_instance_id"] = update.plugin_instance_id
|
|
if update.plugin_name is not None:
|
|
ins["plugin_name"] = update.plugin_name
|
|
if update.slot_order is not None:
|
|
ins["slot_order"] = update.slot_order
|
|
if update.bypassed is not None:
|
|
ins["bypassed"] = update.bypassed
|
|
if update.wet_dry_mix is not None:
|
|
ins["wet_dry_mix"] = min(1.0, max(0.0, update.wet_dry_mix))
|
|
_save_inserts(inserts)
|
|
logger.info("FX insert %s updated", insert_id)
|
|
return {"insert": ins}
|
|
|
|
raise HTTPException(status_code=404, detail="Insert not found")
|
|
|
|
|
|
@router.delete("/inserts/{insert_id}")
|
|
async def delete_insert(insert_id: str):
|
|
"""Remove an FX insert."""
|
|
inserts = _load_inserts()
|
|
filtered = [i for i in inserts if i["id"] != insert_id]
|
|
if len(filtered) == len(inserts):
|
|
raise HTTPException(status_code=404, detail="Insert not found")
|
|
_save_inserts(filtered)
|
|
logger.info("FX insert %s deleted", insert_id)
|
|
return {"status": "deleted"}
|
|
|
|
|
|
@router.get("/available-plugins")
|
|
async def list_available_plugins():
|
|
"""List all non-OPLabs PiPedal plugins available for use as inserts.
|
|
|
|
Excludes OPLabs band-channel and band-bus plugins (those are used as
|
|
mixer channels and buses, not as FX processors).
|
|
|
|
Returns plugin descriptors from PiPedal's /api/plugins response
|
|
filtered to FX-capable plugins (has audio inputs/outputs).
|
|
"""
|
|
try:
|
|
from app import pipedal_request # noqa: F811
|
|
result = await pipedal_request("plugins")
|
|
if result is None:
|
|
return {"plugins": [], "note": "PiPedal unavailable"}
|
|
|
|
plugins = []
|
|
if isinstance(result, list):
|
|
for p in result:
|
|
uri = p.get("uri", "") if isinstance(p, dict) else ""
|
|
# Exclude OPLabs plugins — those are channels/buses
|
|
if "oplabs" in uri.lower():
|
|
continue
|
|
plugins.append({
|
|
"uri": uri,
|
|
"name": p.get("name", p.get("pluginName", "")),
|
|
"brand": p.get("brand", ""),
|
|
"audioInputs": p.get("audioInputs", 2),
|
|
"audioOutputs": p.get("audioOutputs", 2),
|
|
"instanceId": p.get("instanceId"),
|
|
})
|
|
return {"plugins": plugins}
|
|
except Exception as e:
|
|
logger.warning("Failed to list available plugins: %s", e)
|
|
return {"plugins": [], "note": str(e)}
|
|
|
|
|
|
@router.get("/channel/{channel_index}")
|
|
async def get_channel_inserts(channel_index: int):
|
|
"""Get all FX inserts for a specific channel, ordered by slot_order."""
|
|
inserts = _load_inserts()
|
|
channel_inserts = [
|
|
i for i in inserts if i.get("channel_index") == channel_index
|
|
]
|
|
channel_inserts.sort(key=lambda x: x.get("slot_order", 0))
|
|
return {"inserts": channel_inserts}
|
|
|
|
|
|
@router.post("/route/{insert_id}")
|
|
async def route_insert_audio(insert_id: str):
|
|
"""Attempt to create JACK connections for a specific insert.
|
|
|
|
Returns the intended source→destination port connections.
|
|
The actual routing must be performed via PiPedal's JACK API
|
|
(POST /api/jack/connect) after this call.
|
|
"""
|
|
inserts = _load_inserts()
|
|
insert = None
|
|
for ins in inserts:
|
|
if ins["id"] == insert_id:
|
|
insert = ins
|
|
break
|
|
if not insert:
|
|
raise HTTPException(status_code=404, detail="Insert not found")
|
|
|
|
channel_idx = insert["channel_index"]
|
|
plugin_instance_id = insert["plugin_instance_id"]
|
|
|
|
# Declare the intended JACK port connections for this insert
|
|
# These are metadata — actual JACK routing depends on the running
|
|
# audio graph and is performed separately via the JACK API
|
|
connections = [
|
|
{
|
|
"source": f"mixer-daemon:channel_{channel_idx}_send",
|
|
"destination": f"pipedal:{plugin_instance_id}:audio_in_1",
|
|
},
|
|
{
|
|
"source": f"pipedal:{plugin_instance_id}:audio_out_1",
|
|
"destination": f"mixer-daemon:channel_{channel_idx}_return",
|
|
},
|
|
]
|
|
|
|
logger.info(
|
|
"FX insert %s route intended: channel %d ↔ plugin %d",
|
|
insert_id, channel_idx, plugin_instance_id,
|
|
)
|
|
return {
|
|
"status": "routed",
|
|
"insert_id": insert_id,
|
|
"connections": connections,
|
|
}
|
|
|
|
|
|
@router.post("/route-all")
|
|
async def route_all_inserts():
|
|
"""Re-route all active (non-bypassed) FX inserts.
|
|
|
|
Returns the intended routing state for all active inserts.
|
|
"""
|
|
inserts = _load_inserts()
|
|
active = [i for i in inserts if not i.get("bypassed", False)]
|
|
|
|
routed = []
|
|
for ins in active:
|
|
channel_idx = ins["channel_index"]
|
|
plugin_id = ins["plugin_instance_id"]
|
|
routed.append({
|
|
"insert_id": ins["id"],
|
|
"channel": channel_idx,
|
|
"plugin": plugin_id,
|
|
"status": "configured",
|
|
})
|
|
|
|
logger.info("Re-routed %d active FX inserts", len(active))
|
|
return {"routed": routed, "total": len(active)}
|