""" 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. """ from __future__ import annotations 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 base64 import websockets from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from mixer_ipc import MixerIpcClient, MixerIpcError from ws_proxy import PiPedalWSProxy from jack_endpoints import router as jack_router from fx_inserts import router as fx_router 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" MIXER_SCENES_FILE = Path(__file__).resolve().parent / "mixer-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 # Connected mixer WebSocket clients for push event broadcasting mixer_ws_clients: set[WebSocket] = set() # HTTP Basic Auth config (from env — for NPMplus reverse proxy compat) AUTH_ENABLED = os.environ.get("MIXER_AUTH_ENABLED", "").lower() in ("1", "true", "yes") AUTH_USERNAME = os.environ.get("MIXER_AUTH_USERNAME", "oplabs") AUTH_PASSWORD = os.environ.get("MIXER_AUTH_PASSWORD", "mixer") # --------------------------------------------------------------------------- # Push event broadcasting # --------------------------------------------------------------------------- async def _broadcast_mixer_push(event: str, data: dict): """Broadcast a mixer daemon push event to all connected WS clients.""" if not mixer_ws_clients: return message = json.dumps({"type": "push", "event": event, "data": data}) dead: set[WebSocket] = set() for ws in mixer_ws_clients: try: await ws.send_text(message) except Exception: dead.add(ws) mixer_ws_clients -= dead # --------------------------------------------------------------------------- # 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)) # --------------------------------------------------------------------------- # Mixer scene storage helpers (separate from legacy PiPedal scenes) # --------------------------------------------------------------------------- def _load_mixer_scenes() -> list[dict]: """Load mixer scenes from the mixer-scenes.json file.""" if not MIXER_SCENES_FILE.exists(): return [] try: data = json.loads(MIXER_SCENES_FILE.read_text()) return data.get("scenes", []) except (json.JSONDecodeError, Exception) as e: logger.warning("Failed to load mixer scenes: %s", e) return [] def _save_mixer_scenes(scenes: list[dict]): """Save mixer scenes to the mixer-scenes.json file.""" MIXER_SCENES_FILE.parent.mkdir(parents=True, exist_ok=True) MIXER_SCENES_FILE.write_text(json.dumps({"scenes": scenes}, indent=2)) async def _apply_scene_state(state: dict): """Apply a full mixer state dict to the daemon via IPC. Expects the state format returned by mixer_client.get_state(): {channels: [{channelIndex, volume, pan, mute, solo, label, ...}], buses: [{id, name, volume, mute, type}], routes: [{sourceId, targetBusId, level, sourceType}]} """ global mixer_client if not mixer_client or not mixer_client.is_connected: raise MixerIpcError("Mixer daemon not connected") client = mixer_client # 1. Apply channel states channels = state.get("channels", []) for ch in channels: idx = ch.get("channelIndex") if idx is None: continue if "volume" in ch: await client.set_channel_volume(idx, ch["volume"]) if "pan" in ch: await client.set_channel_pan(idx, ch["pan"]) if "mute" in ch: await client.set_channel_mute(idx, ch["mute"]) if "solo" in ch: await client.set_channel_solo(idx, ch["solo"]) if ch.get("label"): await client.set_channel_label(idx, ch["label"]) # 2. Apply bus states buses = state.get("buses", []) for bus in buses: bus_id = bus.get("id") if bus_id is None: continue if "volume" in bus: await client.set_bus_volume(bus_id, bus["volume"]) if "mute" in bus: await client.set_bus_mute(bus_id, bus["mute"]) # 3. Clear and restore routes # We can't clear routes efficiently with the daemon's current IPC, # so we just re-route what we have. New routes are added on top. routes = state.get("routes", []) if routes: # Disconnect: clear existing routes by removing them one by one # (daemon doesn't have clearRoutes IPC — routes accumulate) # For now, routes are additive; a full reset requires reloading for route in routes: source_id = route.get("sourceId") target_bus_id = route.get("targetBusId") level = route.get("level", 0.0) source_type = route.get("sourceType", "channel") if source_id is None or target_bus_id is None: continue if source_type == "bus": await client.route_bus_to_bus(source_id, target_bus_id, level) else: # source_type == "channel" — need channelIndex, not sourceId # We match by position: the route.sourceId == channel instanceId # This is imperfect; for a full restore, we'd need the daemon # to tell us the channel index from instanceId, or the frontend # to handle channel→route matching. pass # 4. Apply output routes output_routes = state.get("outputRoutes") if output_routes is not None: await client.set_output_routes(output_routes) logger.info("Scene state applied: %d channels, %d buses, %d routes", len(channels), len(buses), len(routes)) # --------------------------------------------------------------------------- # 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 with retry mixer_client = MixerIpcClient(socket_path=MIXER_SOCKET_PATH) connected = await _connect_with_retry(mixer_client, retries=5, delay=3.0) if connected: mixer_daemon_connected = True logger.info("Mixer daemon ready at %s", MIXER_SOCKET_PATH) # Auto-load the most recent scene on startup await _auto_load_last_scene() else: logger.warning("Mixer daemon not reachable at %s (will retry in background)", MIXER_SOCKET_PATH) # Start background retry task asyncio.create_task(_background_daemon_reconnect(mixer_client)) yield # Shutdown if mixer_client: await mixer_client.close() if pipedal_proxy: await pipedal_proxy.close_upstream() logger.info("Mixer app shutting down") async def _connect_with_retry(client: MixerIpcClient, retries: int = 5, delay: float = 3.0) -> bool: """Try connecting to the daemon with exponential backoff.""" for attempt in range(1, retries + 1): ok = await client.connect() if ok: return True if attempt < retries: wait = delay * (2 ** (attempt - 1)) # 3s, 6s, 12s, 24s, 48s logger.info("Daemon connect attempt %d/%d failed, retrying in %.1fs...", attempt, retries, wait) await asyncio.sleep(wait) else: logger.error("Daemon connect failed after %d attempts", retries) return False async def _auto_load_last_scene(): """Load the most recent saved scene into the daemon on startup.""" global mixer_client if not mixer_client or not mixer_client.is_connected: logger.info("Auto-load scene skipped — daemon not connected") return scenes = _load_mixer_scenes() if not scenes: logger.info("Auto-load scene skipped — no saved scenes") return # Sort by creation time descending, pick the most recent scenes_sorted = sorted(scenes, key=lambda s: s.get("createdAt", 0), reverse=True) latest = scenes_sorted[0] state = latest.get("state") if not state: logger.warning("Auto-load scene '%s' has no state data, skipping", latest.get("name", "unnamed")) return try: await _apply_scene_state(state) logger.info("Auto-loaded scene '%s' on startup (%d channels, %d buses)", latest.get("name", "unnamed"), len(state.get("channels", [])), len(state.get("buses", []))) except MixerIpcError as e: logger.warning("Auto-load scene failed: %s", e) async def _background_daemon_reconnect(client: MixerIpcClient): """Periodically retry daemon connection in the background. Once connected, loads the last saved scene. Runs until daemon connects or the app shuts down. """ global mixer_daemon_connected retry_delay = 5.0 max_delay = 60.0 while not client.is_connected: await asyncio.sleep(retry_delay) ok = await client.connect() if ok: mixer_daemon_connected = True logger.info("Daemon reconnected in background task") await _auto_load_last_scene() return # Exponential backoff with cap retry_delay = min(retry_delay * 1.5, max_delay) logger.debug("Daemon reconnect retry in %.1fs...", retry_delay) app = FastAPI( title="OPLabs Mixer App", version="0.2.0", lifespan=lifespan, ) # --------------------------------------------------------------------------- # CORS — allow dev server and configured origins # --------------------------------------------------------------------------- cors_origins = os.environ.get("MIXER_CORS_ORIGINS", "http://localhost:5173,http://localhost:5174") app.add_middleware( CORSMiddleware, allow_origins=cors_origins.split(",") if cors_origins else ["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # --------------------------------------------------------------------------- # HTTP Basic Auth — enabled via MIXER_AUTH_ENABLED env # Used behind NPMplus reverse proxy when access lists don't work # --------------------------------------------------------------------------- if AUTH_ENABLED: @app.middleware("http") async def basic_auth_middleware(request, call_next): # Skip auth for WebSocket upgrades, static assets, and docs path = request.url.path if (path.startswith("/ws/") or path.startswith("/assets/") or path.startswith("/docs") or path.startswith("/openapi")): return await call_next(request) auth_header = request.headers.get("Authorization", "") if not auth_header.startswith("Basic "): return JSONResponse( {"error": "Unauthorized"}, status_code=401, headers={"WWW-Authenticate": 'Basic realm="OPLabs Mixer"'}, ) try: decoded = base64.b64decode(auth_header[6:]).decode("utf-8") username, password = decoded.split(":", 1) except Exception: return JSONResponse({"error": "Invalid auth header"}, status_code=401) if username != AUTH_USERNAME or password != AUTH_PASSWORD: return JSONResponse({"error": "Unauthorized"}, status_code=401) return await call_next(request) # --------------------------------------------------------------------------- # Register sub-routers # --------------------------------------------------------------------------- app.include_router(jack_router) app.include_router(fx_router) # --------------------------------------------------------------------------- # 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 (file-based persistence). Captures the full mixer daemon state and stores it to mixer-scenes.json. """ name = body.get("name", "Unnamed Scene") try: state = await with_mixer_client("get_state") scenes = _load_mixer_scenes() scene = { "id": str(uuid.uuid4()), "name": name, "createdAt": time.time(), "state": state, } scenes.append(scene) _save_mixer_scenes(scenes) return {"scene": {"id": scene["id"], "name": scene["name"], "createdAt": scene["createdAt"]}} 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 — restore daemon state from file.""" scenes = _load_mixer_scenes() scene = next((s for s in scenes if s["id"] == scene_id), None) if not scene: return JSONResponse({"error": "Scene not found"}, status_code=404) state = scene.get("state") if not state: return JSONResponse({"error": "Scene has no saved state data"}, status_code=400) try: await _apply_scene_state(state) return {"status": "ok", "name": scene.get("name", "")} 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 (metadata only, no full state).""" scenes = _load_mixer_scenes() return {"scenes": [ {"id": s["id"], "name": s.get("name", ""), "createdAt": s.get("createdAt", 0)} for s in scenes ]} @app.delete("/api/mixer/scenes/{scene_id}") async def mixer_delete_scene(scene_id: str): """Delete a scene by ID (file-based).""" scenes = _load_mixer_scenes() filtered = [s for s in scenes if s["id"] != scene_id] if len(filtered) == len(scenes): return JSONResponse({"error": "Scene not found"}, status_code=404) _save_mixer_scenes(filtered) return {"status": "deleted"} # --------------------------------------------------------------------------- # 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} Push events from the daemon are forwarded to all connected clients. """ 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 # Register for push event broadcasts mixer_ws_clients.add(ws) 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"]), # Scene methods — handled via backend REST /api/mixer/scenes/* # (removed from WS proxy — use REST endpoints instead) } 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) finally: mixer_ws_clients.discard(ws) # --------------------------------------------------------------------------- # 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)