""" OPLabs Mixer App - FastAPI Backend Serves the React SPA and provides REST/WebSocket access to the C++ Mixer Daemon via Unix domain socket IPC. Architecture: Browser <-> FastAPI (:8081) <-> Unix Socket <-> MixerDaemon Also retains backward compatibility with PiPedal via WebSocket proxy. """ import asyncio import json import logging import os import time import uuid from contextlib import asynccontextmanager from pathlib import Path from typing import Optional import websockets from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from mixer_ipc import MixerIpcClient, MixerIpcError from ws_proxy import PiPedalWSProxy logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s") logger = logging.getLogger("mixer-app") # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- PIPEDAL_HOST = "192.168.0.245" PIPEDAL_PORT = 8080 PIPEDAL_WS_PATH = "/pipedal" MIXER_SOCKET_PATH = "/tmp/oplabs-mixer.sock" # OPLabs plugin URIs we care about OPLABS_PLUGIN_URIS = { "http://ourpad.casa/plugins/oplabs-band-channel", "http://ourpad.casa/plugins/oplabs-band-bus", "http://ourpad.casa/plugins/oplabs-band-bus#", # alternate form } FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend" / "dist" SCENES_FILE = Path(__file__).resolve().parent / "scenes.json" # --------------------------------------------------------------------------- # Global state # --------------------------------------------------------------------------- pipedal_proxy: Optional[PiPedalWSProxy] = None pipedal_state: dict = { "connected": False, "plugins": [], # plugin descriptors from PiPedal "pedalboard": None, # current pedalboard state "oplabs_instances": [], # discovered OPLabs plugin instances } # Mixer daemon IPC client (set in lifespan) mixer_client: Optional[MixerIpcClient] = None mixer_daemon_connected: bool = False # --------------------------------------------------------------------------- # Scene storage helpers # --------------------------------------------------------------------------- def _load_scenes() -> list[dict]: """Load scenes from the scenes.json file.""" if not SCENES_FILE.exists(): return [] try: data = json.loads(SCENES_FILE.read_text()) return data.get("scenes", []) except (json.JSONDecodeError, Exception) as e: logger.warning("Failed to load scenes: %s", e) return [] def _save_scenes(scenes: list[dict]): """Save scenes to the scenes.json file.""" SCENES_FILE.parent.mkdir(parents=True, exist_ok=True) SCENES_FILE.write_text(json.dumps({"scenes": scenes}, indent=2)) # --------------------------------------------------------------------------- # PiPedal WS client for backend-internal use # --------------------------------------------------------------------------- async def pipedal_request(message_name: str, body=None, timeout: float = 10.0): """Connect to PiPedal WS, send a request, wait for reply, then disconnect.""" url = f"ws://{PIPEDAL_HOST}:{PIPEDAL_PORT}{PIPEDAL_WS_PATH}" reply_to = hash(message_name + str(time.time())) & 0x7FFFFFFF msg = [{"message": message_name, "replyTo": reply_to}] if body is not None: msg.append(body) try: async with websockets.connect(url, ping_interval=30, ping_timeout=10, close_timeout=5) as ws: ws.settimeout(timeout) # Send hello first to register as subscriber hello_reply_to = 0x10000001 hello_msg = json.dumps([{"message": "hello", "replyTo": hello_reply_to}]) await ws.send(hello_msg) # Wait for hello reply (discard other push messages) deadline = time.time() + timeout got_hello = False while time.time() < deadline: remaining = deadline - time.time() raw = await asyncio.wait_for(ws.recv(), timeout=remaining) data = json.loads(raw) if isinstance(data, list) and len(data) >= 1: hdr = data[0] if hdr.get("reply") == hello_reply_to: got_hello = True break if not got_hello: logger.warning("Hello reply not received") return None # Send actual request await ws.send(json.dumps(msg)) # Wait for matching reply (discard interleaved push messages) while time.time() < deadline: remaining = deadline - time.time() if remaining <= 0: break raw = await asyncio.wait_for(ws.recv(), timeout=remaining) data = json.loads(raw) if isinstance(data, list) and len(data) >= 1: hdr = data[0] if hdr.get("reply") == reply_to: body_result = data[1] if len(data) > 1 else None return body_result logger.warning("Request '%s' timed out (no matching reply)", message_name) return None except asyncio.TimeoutError: logger.warning("Request '%s' timed out", message_name) return None except websockets.exceptions.ConnectionClosed: logger.warning("Request '%s': connection closed", message_name) return None except Exception as e: logger.warning("Request '%s' failed: %s", message_name, e) return None # --------------------------------------------------------------------------- # OPLabs plugin discovery # --------------------------------------------------------------------------- def is_oplabs_plugin(uri: str) -> bool: """Check if a plugin URI is one of our OPLabs plugins.""" return uri in OPLABS_PLUGIN_URIS def discover_oplabs_from_pedalboard(pedalboard: dict) -> list[dict]: """Extract OPLabs plugin instances from a pedalboard state dict.""" instances = [] if not pedalboard: return instances def extract_from_item(item: dict): uri = item.get("uri", "") if is_oplabs_plugin(uri): control_values = item.get("controlValues", []) cv_map = {cv["key"]: cv["value"] for cv in control_values if "key" in cv} instances.append({ "instanceId": item.get("instanceId"), "uri": uri, "pluginName": item.get("pluginName", ""), "title": item.get("title", ""), "isEnabled": item.get("isEnabled", True), "controlValues": cv_map, "portValues": cv_map, # alias for clarity }) # Handle split items (nested chains) if "chaining" in item: for chain in item["chaining"]: if isinstance(chain, dict) and "items" in chain: for sub_item in chain["items"]: extract_from_item(sub_item) items = pedalboard.get("items", []) for item in items: extract_from_item(item) return instances def fetch_pedalboard_state(): """Fetch pedalboard state from PiPedal (blocking, for REST API).""" global pipedal_state try: result = asyncio.run(pipedal_request("currentPedalboard")) if result is not None: pipedal_state["pedalboard"] = result pipedal_state["oplabs_instances"] = discover_oplabs_from_pedalboard(result) pipedal_state["connected"] = True return True except Exception as e: logger.warning("Failed to fetch pedalboard state: %s", e) pipedal_state["connected"] = False return False # --------------------------------------------------------------------------- # Mixer Daemon IPC helpers # --------------------------------------------------------------------------- async def with_mixer_client(method_name: str, *args, **kwargs): """Call a method on the MixerIpcClient, handling connection errors.""" global mixer_client, mixer_daemon_connected if not mixer_client or not mixer_client.is_connected: mixer_daemon_connected = False raise MixerIpcError("Mixer daemon not connected") try: result = await getattr(mixer_client, method_name)(*args, **kwargs) mixer_daemon_connected = True return result except MixerIpcError: raise except Exception as e: mixer_daemon_connected = False raise MixerIpcError(f"Mixer daemon communication error: {e}") # --------------------------------------------------------------------------- # FastAPI app # --------------------------------------------------------------------------- @asynccontextmanager async def lifespan(app: FastAPI): """Startup / shutdown.""" global pipedal_proxy, mixer_client # Start PiPedal proxy pipedal_proxy = PiPedalWSProxy(host=PIPEDAL_HOST, port=PIPEDAL_PORT, path=PIPEDAL_WS_PATH) logger.info("Mixer app starting - PiPedal at %s:%d", PIPEDAL_HOST, PIPEDAL_PORT) # Try connecting to Mixer Daemon mixer_client = MixerIpcClient(socket_path=MIXER_SOCKET_PATH) connected = await mixer_client.connect() if connected: logger.info("Connected to Mixer Daemon at %s", MIXER_SOCKET_PATH) else: logger.warning("Mixer Daemon not available at %s - mixer endpoints disabled", MIXER_SOCKET_PATH) yield # Shutdown if mixer_client: await mixer_client.close() if pipedal_proxy: await pipedal_proxy.close_upstream() logger.info("Mixer app shutting down") app = FastAPI( title="OPLabs Mixer App", version="0.2.0", lifespan=lifespan, ) # --------------------------------------------------------------------------- # Mixer Daemon REST endpoints (new) # --------------------------------------------------------------------------- @app.get("/api/mixer/state") async def mixer_get_state(): """Get full mixer state from the mixer daemon.""" try: state = await with_mixer_client("get_state") return state except MixerIpcError as e: return JSONResponse({"error": str(e)}, status_code=503) @app.post("/api/mixer/channel/{channel_index}/volume") async def mixer_set_channel_volume(channel_index: int, body: dict): """Set channel volume in dB.""" volume_db = body.get("volumeDb", body.get("volume", -96.0)) try: await with_mixer_client("set_channel_volume", channel_index, volume_db) return {"status": "ok"} except MixerIpcError as e: return JSONResponse({"error": str(e)}, status_code=503) @app.post("/api/mixer/channel/{channel_index}/pan") async def mixer_set_channel_pan(channel_index: int, body: dict): """Set channel pan (-1.0 to +1.0).""" pan = body.get("pan", 0.0) try: await with_mixer_client("set_channel_pan", channel_index, pan) return {"status": "ok"} except MixerIpcError as e: return JSONResponse({"error": str(e)}, status_code=503) @app.post("/api/mixer/channel/{channel_index}/mute") async def mixer_set_channel_mute(channel_index: int, body: dict): """Set channel mute state.""" mute = body.get("mute", True) try: await with_mixer_client("set_channel_mute", channel_index, mute) return {"status": "ok"} except MixerIpcError as e: return JSONResponse({"error": str(e)}, status_code=503) @app.post("/api/mixer/channel/{channel_index}/solo") async def mixer_set_channel_solo(channel_index: int, body: dict): """Set channel solo state.""" solo = body.get("solo", True) try: await with_mixer_client("set_channel_solo", channel_index, solo) return {"status": "ok"} except MixerIpcError as e: return JSONResponse({"error": str(e)}, status_code=503) @app.post("/api/mixer/channel/add") async def mixer_add_channel(body: dict): """Add a new channel. Returns {channelIndex: N}.""" physical_input = body.get("physicalInputIndex", 0) try: idx = await with_mixer_client("add_channel", physical_input) return {"channelIndex": idx} except MixerIpcError as e: return JSONResponse({"error": str(e)}, status_code=503) @app.delete("/api/mixer/channel/{channel_index}") async def mixer_remove_channel(channel_index: int): """Remove a channel by index.""" try: await with_mixer_client("remove_channel", channel_index) return {"status": "ok"} except MixerIpcError as e: return JSONResponse({"error": str(e)}, status_code=503) @app.get("/api/mixer/output-routes") async def mixer_get_output_routes(): """Get output routing table.""" try: routes = await with_mixer_client("get_output_routes") return {"routes": routes} except MixerIpcError as e: return JSONResponse({"error": str(e)}, status_code=503) @app.post("/api/mixer/output-routes") async def mixer_set_output_routes(body: dict): """Set output routing table.""" routes = body.get("routes", []) try: await with_mixer_client("set_output_routes", routes) return {"status": "ok"} except MixerIpcError as e: return JSONResponse({"error": str(e)}, status_code=503) @app.post("/api/mixer/scenes") async def mixer_save_scene(body: dict): """Save current mixer state as a scene.""" name = body.get("name", "Unnamed Scene") try: result = await with_mixer_client("save_scene", name) return {"scene": result} except MixerIpcError as e: return JSONResponse({"error": str(e)}, status_code=503) @app.post("/api/mixer/scenes/{scene_id}/load") async def mixer_load_scene(scene_id: str): """Load (recall) a scene by ID.""" try: ok = await with_mixer_client("load_scene", scene_id) if ok: return {"status": "ok"} return JSONResponse({"error": "Scene not found"}, status_code=404) except MixerIpcError as e: return JSONResponse({"error": str(e)}, status_code=503) @app.get("/api/mixer/scenes") async def mixer_list_scenes(): """List available scenes.""" try: scenes = await with_mixer_client("list_scenes") return {"scenes": scenes} except MixerIpcError as e: return JSONResponse({"error": str(e)}, status_code=503) @app.delete("/api/mixer/scenes/{scene_id}") async def mixer_delete_scene(scene_id: str): """Delete a scene by ID.""" try: ok = await with_mixer_client("delete_scene", scene_id) if ok: return {"status": "deleted"} return JSONResponse({"error": "Scene not found"}, status_code=404) except MixerIpcError as e: return JSONResponse({"error": str(e)}, status_code=503) # --------------------------------------------------------------------------- # Scene REST endpoints (legacy - file-based, via PiPedal) # --------------------------------------------------------------------------- class SceneStateChannel(BaseModel): instanceId: int params: dict class SceneState(BaseModel): channels: list[SceneStateChannel] = [] buses: list[SceneStateChannel] = [] class SceneCreate(BaseModel): name: str midiPc: Optional[int] = None state: SceneState class SceneUpdate(BaseModel): name: Optional[str] = None midiPc: Optional[int] = None state: Optional[SceneState] = None @app.get("/api/scenes") async def list_scenes(): """List all saved scenes (legacy file-based).""" scenes = _load_scenes() return {"scenes": scenes} @app.post("/api/scenes") async def create_scene(scene: SceneCreate): """Save a new scene (legacy file-based).""" scenes = _load_scenes() new_scene = { "id": str(uuid.uuid4()), "name": scene.name, "midiPc": scene.midiPc, "state": scene.state.model_dump(), } scenes.append(new_scene) _save_scenes(scenes) return {"scene": new_scene} @app.put("/api/scenes/{scene_id}") async def update_scene(scene_id: str, update: SceneUpdate): """Update an existing scene.""" scenes = _load_scenes() for sc in scenes: if sc["id"] == scene_id: if update.name is not None: sc["name"] = update.name if update.midiPc is not None: sc["midiPc"] = update.midiPc if update.state is not None: sc["state"] = update.state.model_dump() _save_scenes(scenes) return {"scene": sc} return JSONResponse({"error": "Scene not found"}, status_code=404) @app.delete("/api/scenes/{scene_id}") async def delete_scene(scene_id: str): """Delete a scene.""" scenes = _load_scenes() filtered = [sc for sc in scenes if sc["id"] != scene_id] if len(filtered) == len(scenes): return JSONResponse({"error": "Scene not found"}, status_code=404) _save_scenes(filtered) return {"status": "deleted"} @app.post("/api/scenes/{scene_id}/recall") async def recall_scene(scene_id: str): """Recall a scene via PiPedal.""" scenes = _load_scenes() scene = None for sc in scenes: if sc["id"] == scene_id: scene = sc break if not scene: return JSONResponse({"error": "Scene not found"}, status_code=404) state = scene.get("state", {}) channels = state.get("channels", []) buses = state.get("buses", []) messages: list[dict] = [] for ch in channels: instance_id = ch["instanceId"] for key, value in ch["params"].items(): messages.append({ "message": "setControlValue", "body": {"instanceId": instance_id, "key": key, "value": value}, }) for bus in buses: instance_id = bus["instanceId"] for key, value in bus["params"].items(): messages.append({ "message": "setControlValue", "body": {"instanceId": instance_id, "key": key, "value": value}, }) if not messages: return {"status": "recalled", "message_count": 0} url = f"ws://{PIPEDAL_HOST}:{PIPEDAL_PORT}{PIPEDAL_WS_PATH}" try: async with websockets.connect(url, ping_interval=30, ping_timeout=10, close_timeout=5) as ws: hello_reply_to = 0x10000001 await ws.send(json.dumps([{"message": "hello", "replyTo": hello_reply_to}])) deadline = time.time() + 10.0 got_hello = False while time.time() < deadline: remaining = deadline - time.time() raw = await asyncio.wait_for(ws.recv(), timeout=remaining) data = json.loads(raw) if isinstance(data, list) and len(data) >= 1: hdr = data[0] if hdr.get("reply") == hello_reply_to: got_hello = True break if not got_hello: logger.warning("Recall: hello reply not received") return JSONResponse({"error": "PiPedal not responding"}, status_code=502) sent = 0 for msg in messages: reply_to = (hash(str(msg)) & 0x7FFFFFFF) wire_msg = json.dumps([ {"message": msg["message"], "replyTo": reply_to}, msg["body"], ]) await ws.send(wire_msg) sent += 1 if sent % 5 == 0: await asyncio.sleep(0.01) logger.info("Recalled scene '%s' - sent %d control messages", scene["name"], sent) return {"status": "recalled", "message_count": sent} except Exception as e: logger.warning("Recall failed: %s", e) return JSONResponse({"error": f"Recall failed: {str(e)}"}, status_code=502) # --------------------------------------------------------------------------- # PiPedal REST endpoints (legacy) # --------------------------------------------------------------------------- @app.get("/api/discover") async def discover_plugins(): """Discover OPLabs plugin instances from PiPedal.""" ok = fetch_pedalboard_state() return { "connected": pipedal_state["connected"], "oplabs_instances": pipedal_state["oplabs_instances"], "pedalboard_name": pipedal_state["pedalboard"].get("name") if pipedal_state["pedalboard"] else None, } @app.get("/api/plugins") async def list_pipedal_plugins(): """List all available plugins from PiPedal.""" result = await pipedal_request("plugins") return {"plugins": result if result else []} @app.get("/api/ping") async def ping(): """Health check.""" return { "status": "ok", "pipedal_host": PIPEDAL_HOST, "pipedal_port": PIPEDAL_PORT, "pipedal_connected": pipedal_state["connected"], "mixer_daemon_connected": mixer_daemon_connected if mixer_client else False, } # --------------------------------------------------------------------------- # WebSocket endpoints # --------------------------------------------------------------------------- @app.websocket("/ws/pipedal") async def websocket_proxy(ws: WebSocket): """Proxy WebSocket: browser <-> local FastAPI <-> PiPedal.""" await ws.accept() logger.info("Browser WS connected (PiPedal proxy)") proxy = PiPedalWSProxy(host=PIPEDAL_HOST, port=PIPEDAL_PORT, path=PIPEDAL_WS_PATH) connected = await proxy.connect_upstream() if not connected: await ws.send_json({ "type": "error", "message": f"Cannot connect to PiPedal at {proxy.upstream_url}" }) await ws.close() return pipedal_state["connected"] = True await ws.send_json({"type": "connected", "pipedal": proxy.upstream_url}) try: await proxy.relay_loop(ws) except WebSocketDisconnect: logger.info("Browser WS disconnected") except Exception as e: logger.warning("WebSocket proxy error: %s", e) finally: pipedal_state["connected"] = False await proxy.close_upstream() @app.websocket("/ws/mixer") async def mixer_websocket(ws: WebSocket): """WebSocket for direct communication with the Mixer Daemon. Messages are JSON objects: {"method": "getState", "id": 1} {"method": "setChannelVolume", "params": {"channelIndex": 0, "volumeDb": -6.0}, "id": 2} Responses mirror the IPC protocol: {"id": 1, "result": {...}, "error": null} """ await ws.accept() logger.info("Browser WS connected (Mixer Daemon)") if not mixer_client or not mixer_client.is_connected: await ws.send_json({ "type": "error", "message": "Mixer daemon not available" }) await ws.close() return await ws.send_json({"type": "connected", "daemon": MIXER_SOCKET_PATH}) try: while True: raw = await ws.receive_text() try: msg = json.loads(raw) except json.JSONDecodeError: await ws.send_json({"id": -1, "error": "Invalid JSON"}) continue method = msg.get("method", "") params = msg.get("params", {}) req_id = msg.get("id", -1) # Map WebSocket methods to MixerIpcClient methods method_map = { "getState": ("get_state", None), "setChannelVolume": ("set_channel_volume", ["channelIndex", "volumeDb"]), "setChannelPan": ("set_channel_pan", ["channelIndex", "pan"]), "setChannelMute": ("set_channel_mute", ["channelIndex", "mute"]), "setChannelSolo": ("set_channel_solo", ["channelIndex", "solo"]), "addChannel": ("add_channel", ["physicalInputIndex"]), "removeChannel": ("remove_channel", ["channelIndex"]), "getOutputRoutes": ("get_output_routes", None), "setOutputRoutes": ("set_output_routes", ["routes"]), "saveScene": ("save_scene", ["name"]), "loadScene": ("load_scene", ["sceneId"]), "listScenes": ("list_scenes", None), "deleteScene": ("delete_scene", ["sceneId"]), } if method not in method_map: await ws.send_json({"id": req_id, "error": f"Unknown method: {method}"}) continue client_method, param_keys = method_map[method] try: if param_keys is None: result = await getattr(mixer_client, client_method)() else: args = [params.get(k) for k in param_keys] result = await getattr(mixer_client, client_method)(*args) await ws.send_json({"id": req_id, "result": result, "error": None}) except MixerIpcError as e: await ws.send_json({"id": req_id, "error": str(e)}) except WebSocketDisconnect: logger.info("Mixer WS disconnected") except Exception as e: logger.warning("Mixer WS error: %s", e) # --------------------------------------------------------------------------- # Serve SPA (must be last, after all API routes) # --------------------------------------------------------------------------- SPA_INDEX = FRONTEND_DIR / "index.html" if FRONTEND_DIR.exists(): app.mount("/assets", StaticFiles(directory=str(FRONTEND_DIR / "assets")), name="assets") @app.get("/{full_path:path}") async def serve_spa(full_path: str): """Serve the SPA - return index.html for all non-API routes.""" if full_path.startswith("api/") or full_path.startswith("ws/"): return JSONResponse({"error": "not found"}, status_code=404) if not SPA_INDEX.exists(): return JSONResponse( {"error": "Frontend not built. Run 'npm run build' in frontend/"}, status_code=503, ) return FileResponse(str(SPA_INDEX)) else: logger.warning("Frontend dist not found at %s. SPA serving disabled.", FRONTEND_DIR) @app.get("/") async def root(): return {"message": "OPLabs Mixer App API", "docs": "/docs"} # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- if __name__ == "__main__": import uvicorn uvicorn.run("app:app", host="0.0.0.0", port=8081, reload=True)