#!/usr/bin/env python3 """ MPD → OBS WebSocket bridge. Re-broadcasts MPD playback state as `mpd:state` CustomEvents on the OBS WebSocket bus. The loading overlay (loading/index.html) already listens for those events to drive its now-playing terminal line — this script replaces the custom music/index.html browser daemon as the source of truth. Payload shape matches the original daemon's broadcastState() so the overlay needs zero changes: { _type, index, total, title, file, currentTime, duration, paused } Reads OBS WS credentials from ../vendor/obs-config.js (auto-generated by scripts/setup.sh from plugin_config/obs-websocket/config.json — single source of truth). Run via systemd: systemctl --user enable --now obs-mpd-bridge.service Tail logs with: journalctl --user -u obs-mpd-bridge -f """ import asyncio import base64 import hashlib import json import os import re import sys import time from pathlib import Path import websockets from mpd.asyncio import MPDClient HERE = Path(__file__).resolve().parent OBS_CONFIG_JS = HERE.parent / "vendor" / "obs-config.js" MPD_HOST = os.environ.get("MPD_HOST", "localhost") MPD_PORT = int(os.environ.get("MPD_PORT", "6600")) TICK_SECONDS = 1.0 # broadcast cadence; matches the old daemon's setInterval(broadcastState, 1000) # ─── config + auth ────────────────────────────────────────────────────────── def load_obs_config(path: Path): """Extract `url` and `password` from the JS config file by regex — avoids requiring a JS parser for two trivial string assignments.""" text = path.read_text() url_m = re.search(r"url:\s*'([^']+)'", text) pw_m = re.search(r"password:\s*'([^']*)'", text) if not url_m: raise RuntimeError(f"no `url` field in {path}") return url_m.group(1), (pw_m.group(1) if pw_m else "") def obs_auth_response(password: str, salt: str, challenge: str) -> str: """OBS WebSocket v5 auth: base64(sha256(b64(sha256(pw+salt)) + challenge)).""" secret_b64 = base64.b64encode( hashlib.sha256((password + salt).encode()).digest() ).decode() return base64.b64encode( hashlib.sha256((secret_b64 + challenge).encode()).digest() ).decode() # ─── state translation ────────────────────────────────────────────────────── def _flatten_tag(v): """MPD returns multi-value tags as lists (e.g. two ARTIST lines for a feature collab). Join with `, ` and trim; treat empty as None.""" if v is None: return None if isinstance(v, list): v = ", ".join(x for x in v if x) v = v.strip() return v or None def mpd_to_event(status: dict, song: dict) -> dict: """Translate `status` + `currentsong` into the overlay's expected payload. Display string in `title` is built `Artist - Title` from ID3/Vorbis tags when both are present, then degrades through tag-only → filename stem (which is already "Artist - Title" for the NCS Directory naming convention) → None. The separate `artist` field is emitted for future overlay uses that want to style artist + title differently.""" paused = status.get("state", "stop") != "play" elapsed = float(status.get("elapsed") or 0) duration = float(status.get("duration") or song.get("duration") or 0) artist = _flatten_tag(song.get("artist") or song.get("albumartist")) track_name = _flatten_tag(song.get("title")) if artist and track_name: display = f"{artist} - {track_name}" elif track_name: display = track_name elif song.get("file"): display = Path(song["file"]).stem.strip() or None else: display = None return { "index": int(status["song"]) if "song" in status else -1, "total": int(status.get("playlistlength") or 0), "title": display, "artist": artist, "trackTitle": track_name, "file": song.get("file"), "currentTime": elapsed, "duration": duration, "paused": paused, } class SharedState: def __init__(self): self.data = { "index": -1, "total": 0, "title": None, "artist": None, "trackTitle": None, "file": None, "currentTime": 0, "duration": 0, "paused": True, } self.changed = asyncio.Event() # ─── MPD side ─────────────────────────────────────────────────────────────── async def mpd_loop(state: SharedState): """Poll MPD status every TICK_SECONDS, push into shared state.""" while True: client = MPDClient() try: await client.connect(MPD_HOST, MPD_PORT) print(f"[mpd] connected → {MPD_HOST}:{MPD_PORT} " f"(proto {client.mpd_version})", flush=True) while True: status = await client.status() song = await client.currentsong() state.data = mpd_to_event(status, song) state.changed.set() await asyncio.sleep(TICK_SECONDS) except Exception as e: print(f"[mpd] {type(e).__name__}: {e} — reconnect in 2s", flush=True) try: client.disconnect() except Exception: pass await asyncio.sleep(2) # ─── OBS WS side ──────────────────────────────────────────────────────────── async def obs_loop(state: SharedState, url: str, password: str): """Connect to OBS WS, identify, broadcast state changes as mpd:state. Quiet when OBS isn't running: logs only state transitions (down→up, up→down) and uses exponential backoff so the journal doesn't fill with reconnect spam during long OBS-off windows. """ backoff = 2 # current retry delay in seconds BACKOFF_MAX = 30 # cap; OBS doesn't take long to wake up once started prev_status = None # None | "up" | "down" — only log on transitions while True: try: async with websockets.connect(url, ping_interval=20) as ws: # ── Hello (op 0) → Identify (op 1) → Identified (op 2) ── hello = json.loads(await ws.recv()) if hello.get("op") != 0: raise RuntimeError(f"expected Hello, got op={hello.get('op')}") identify = {"rpcVersion": 1, "eventSubscriptions": 0} if hello["d"].get("authentication"): a = hello["d"]["authentication"] identify["authentication"] = obs_auth_response( password, a["salt"], a["challenge"], ) await ws.send(json.dumps({"op": 1, "d": identify})) ident = json.loads(await ws.recv()) if ident.get("op") != 2: raise RuntimeError(f"expected Identified, got op={ident.get('op')}") if prev_status != "up": print(f"[obs] connected → {url} (identified ✓)", flush=True) prev_status = "up" backoff = 2 # reset for next disconnect # ── Broadcast loop + recv drain ── # We send ~1 BroadcastCustomEvent per second; OBS replies to # each with an op=7 RequestResponse we don't care about. If we # never recv, the socket's read buffer fills and eventually # TCP backpressure stalls our sends — so a parallel task # consumes and discards everything inbound. async def drain(): async for _ in ws: pass async def broadcast(): # Wake on every state change. A safety timeout slightly # larger than TICK_SECONDS guards against a stalled MPD # loop — we still emit a heartbeat so the overlay knows # we're up. while True: try: await asyncio.wait_for( state.changed.wait(), timeout=TICK_SECONDS + 1.0, ) except asyncio.TimeoutError: pass state.changed.clear() payload = dict(state.data) payload["_type"] = "mpd:state" request = { "op": 6, "d": { "requestType": "BroadcastCustomEvent", "requestId": f"r_{time.monotonic_ns()}", "requestData": {"eventData": payload}, }, } await ws.send(json.dumps(request)) await asyncio.gather(drain(), broadcast()) except Exception as e: # Only log the first time OBS becomes unreachable, or when an # established connection drops. Routine "still down" retries # are silent. if prev_status == "up": print(f"[obs] disconnected ({type(e).__name__}: {e}) — " f"retrying with backoff", flush=True) elif prev_status is None: print(f"[obs] unreachable ({type(e).__name__}) — " f"will keep retrying silently", flush=True) prev_status = "down" await asyncio.sleep(backoff) backoff = min(backoff * 2, BACKOFF_MAX) # ─── entrypoint ───────────────────────────────────────────────────────────── async def main(): if not OBS_CONFIG_JS.exists(): print(f"[err] missing {OBS_CONFIG_JS} — run scripts/setup.sh first", file=sys.stderr, flush=True) sys.exit(1) url, password = load_obs_config(OBS_CONFIG_JS) state = SharedState() await asyncio.gather(mpd_loop(state), obs_loop(state, url, password)) if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: sys.exit(0)