#!/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, artist, trackTitle, album, year, genre,
file, currentTime, duration, paused,
nextTitle, nextArtist, nextTrackTitle, nextFile, nextTitles,
coverPath, coverHash }
Next-track fields are populated when MPD reports `nextsong` in its status.
`nextTitles` is up to 3 display strings peeked from the queue starting at
the current song's successor — used by the Music Box overlay to show
"coming up" lines. The single `nextTitle`/`nextArtist`/etc. fields mirror
nextTitles[0] for overlays that only need one.
Library stats (artists/albums/songs/db_playtime) come from MPD `stats`.
That command is cheap but not free, and the results barely change between
ticks (only when files are added/removed) — so we cache and refresh every
STATS_REFRESH_SEC. Exposed under the `stats` key:
stats: { artists, albums, songs, dbPlaytime } # dbPlaytime is human-formatted
Cover art resolution (run once per track change, cached):
1. Sibling file in song's directory matching folder/cover/front ×
jpg/jpeg/png/webp (case-insensitive). Covers the Kolekcja layout
where ~/HDD/Music/Kolekcja symlinks to /media/nvme/Music/Kolekcja
and each album dir ships a folder.jpg.
2. MPD `readpicture` then `albumart` commands for embedded art (yt-dlp
--embed-thumbnail tracks under the rest of ~/HDD/Music).
Bytes are written to ./cover.jpg (next to this script). Overlays load
`
` so a hash bump cache-busts
cleanly. coverHash null = no art available; overlays should show a
placeholder. Reads MPD library root from MPD_MUSIC_DIR env var, default
~/HDD/Music (matches mpd.conf).
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"))
MPD_MUSIC_DIR = Path(os.environ.get("MPD_MUSIC_DIR", "~/HDD/Music")).expanduser()
TICK_SECONDS = 1.0 # broadcast cadence; matches the old daemon's setInterval(broadcastState, 1000)
STATS_REFRESH_SEC = 30.0 # how often to re-poll MPD `stats` for library counts
COVER_OUT = HERE / "cover.jpg" # canonical cover bytes; overlays cache-bust via ?v=
COVER_STEMS = ("folder", "cover", "front")
COVER_EXTS = (".jpg", ".jpeg", ".png", ".webp")
# ─── 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 _song_display(song: dict) -> tuple:
"""Extract (display, artist, track_title) from an MPD song dict.
Display string is built `Artist - Title` from ID3/Vorbis tags when both
are present, then degrades through tag-only → filename stem (already
"Artist - Title" for the NCS Directory naming convention) → None."""
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 display, artist, track_name
def _format_dbplaytime(seconds: float) -> str:
"""MPD's db_playtime is the total runtime of every file in the library
in seconds. Render the most-significant two units (47d 12h, 5h 22m, …)
so the stats line stays compact even at multi-day libraries."""
s = int(seconds or 0)
days, s = divmod(s, 86400)
hours, s = divmod(s, 3600)
mins, _ = divmod(s, 60)
if days: return f"{days}d {hours}h"
if hours: return f"{hours}h {mins}m"
if mins: return f"{mins}m"
return "0m"
def _status_bool(status: dict, key: str) -> bool:
"""MPD status flags arrive as "0"/"1" strings; normalize to bool."""
return str(status.get(key, "0")).strip() == "1"
def _status_volume(status: dict) -> int | None:
"""Volume is usually 0..100, but MPD can report -1 when unavailable."""
raw = status.get("volume")
if raw is None:
return None
try:
volume = int(raw)
except (TypeError, ValueError):
return None
return volume if volume >= 0 else None
# Module-level cache: MPD stats barely move between ticks (only on file
# add/remove), so we refresh every STATS_REFRESH_SEC and reuse otherwise.
_stats_cache = {"data": None, "next_refresh": 0.0}
async def refresh_stats(client) -> dict | None:
now = time.monotonic()
if _stats_cache["data"] and now < _stats_cache["next_refresh"]:
return _stats_cache["data"]
try:
s = await client.stats()
_stats_cache["data"] = {
"artists": int(s.get("artists") or 0),
"albums": int(s.get("albums") or 0),
"songs": int(s.get("songs") or 0),
"dbPlaytime": _format_dbplaytime(float(s.get("db_playtime") or 0)),
}
_stats_cache["next_refresh"] = now + STATS_REFRESH_SEC
except Exception as e:
# Keep last-known on transient failure — overlays prefer stale to
# blank for slow-moving counters like this.
print(f"[stats] refresh failed: {e}", flush=True)
return _stats_cache["data"]
def mpd_to_event(status: dict, song: dict, next_songs: list | None = None,
cover_path: str | None = None,
cover_hash: str | None = None,
stats: dict | None = None) -> dict:
"""Translate `status` + `currentsong` (+ optional `next_songs`, the next
up to 3 queued tracks) into the overlay payload. The single `nextTitle`
family mirrors `nextTitles[0]` for overlays that only need one."""
paused = status.get("state", "stop") != "play"
elapsed = float(status.get("elapsed") or 0)
duration = float(status.get("duration") or song.get("duration") or 0)
display, artist, track_name = _song_display(song)
album = _flatten_tag(song.get("album"))
# MPD's date tag is whatever the file carries — ID3v2 TYER/TDRC, Vorbis
# DATE, etc. Often a 4-digit year, sometimes a full ISO date. Take the
# leading 4 digits when present so consumers always get a clean year.
raw_date = _flatten_tag(song.get("date") or song.get("originaldate"))
year = None
if raw_date:
m = re.match(r"\s*(\d{4})", raw_date)
year = m.group(1) if m else raw_date
genre = _flatten_tag(song.get("genre"))
playback = {
"volume": _status_volume(status),
"repeat": _status_bool(status, "repeat"),
"random": _status_bool(status, "random"),
"single": _status_bool(status, "single"),
"consume": _status_bool(status, "consume"),
}
next_titles = []
next_display, next_artist, next_track, next_file = None, None, None, None
if next_songs:
for ns in next_songs:
d, a, t = _song_display(ns)
if d is not None:
next_titles.append(d)
if next_songs[0]:
next_display, next_artist, next_track = _song_display(next_songs[0])
next_file = next_songs[0].get("file")
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,
"album": album,
"year": year,
"genre": genre,
"file": song.get("file"),
"currentTime": elapsed,
"duration": duration,
"paused": paused,
"nextTitle": next_display,
"nextArtist": next_artist,
"nextTrackTitle": next_track,
"nextFile": next_file,
"nextTitles": next_titles,
"coverPath": cover_path,
"coverHash": cover_hash,
"playback": playback,
"stats": stats,
}
class SharedState:
def __init__(self):
self.data = {
"index": -1, "total": 0,
"title": None, "artist": None, "trackTitle": None,
"album": None, "year": None, "genre": None, "file": None,
"currentTime": 0, "duration": 0, "paused": True,
"nextTitle": None, "nextArtist": None, "nextTrackTitle": None, "nextFile": None,
"nextTitles": [],
"coverPath": None, "coverHash": None,
"playback": {"volume": None, "repeat": False, "random": False, "single": False, "consume": False},
"stats": None,
}
self.changed = asyncio.Event()
# ─── cover art resolution ───────────────────────────────────────────────────
class CoverState:
"""Caches the last resolved (uri → hash) so we only re-run cover lookup
on actual track change, not every status tick. The hash also lets the
overlay cache-bust without us re-reading the file each second."""
def __init__(self):
self.last_uri: str | None = None
self.last_hash: str | None = None
_cover_state = CoverState()
def _find_folder_cover(song_uri: str) -> Path | None:
"""Look in the song's own directory for folder/cover/front.{jpg,jpeg,
png,webp}. Case-insensitive. song_uri is MPD-relative (e.g.
"Kolekcja/VBS - Nikoś/01 - Track.mp3"); resolve() follows the symlink
so /media/nvme/Music/Kolekcja files are found.
Lookup is prioritized: stems in COVER_STEMS order × exts in COVER_EXTS
order. So when an album ships both `folder.jpg` and `folder.webp` (the
Kolekcja convention), the jpg wins — keeps the on-disk cache file's
extension aligned with its bytes most of the time."""
if not song_uri:
return None
try:
full = (MPD_MUSIC_DIR / song_uri).resolve()
except (OSError, RuntimeError):
return None
parent = full.parent
if not parent.is_dir():
return None
# Snapshot dir as lower-name → Path so prioritized lookup is O(1) and
# case-insensitive without re-scanning per candidate.
try:
by_lower = {p.name.lower(): p for p in parent.iterdir() if p.is_file()}
except OSError:
return None
for stem in COVER_STEMS:
for ext in COVER_EXTS:
hit = by_lower.get(f"{stem}{ext}")
if hit:
return hit
return None
async def _fetch_embedded_art(client, uri: str) -> bytes | None:
"""Try MPD `readpicture` first (preferred — works for any tagged file
with embedded art) then fall back to `albumart` (older, MPD-specific
sidecar lookup). Both stream chunks; we loop with offset until size
is reached."""
for cmd_name in ("readpicture", "albumart"):
cmd = getattr(client, cmd_name, None)
if cmd is None:
continue
try:
buf = bytearray()
offset = 0
while True:
resp = await cmd(uri, offset)
if not isinstance(resp, dict):
break
chunk = resp.get("binary")
if not chunk:
break
buf.extend(chunk)
size = int(resp.get("size") or 0)
offset = len(buf)
if size and offset >= size:
break
if buf:
return bytes(buf)
except Exception:
# MPD raises when the file lacks embedded art for the given
# command — that's the signal to try the next one.
continue
return None
async def resolve_cover(client, song_uri: str | None) -> tuple[str | None, str | None]:
"""Resolve cover art for the current song; return (coverPath, coverHash).
Writes bytes to COVER_OUT only when the hash changes. Returns (None, None)
when no art is available — overlays should render a placeholder.
Caches by `song_uri`: re-running disk reads + MPD readpicture each tick
would be wasteful. State is reset on track change (or song_uri=None).
"""
if not song_uri:
_cover_state.last_uri = None
_cover_state.last_hash = None
return None, None
if song_uri == _cover_state.last_uri:
if _cover_state.last_hash:
return f"../bridges/{COVER_OUT.name}", _cover_state.last_hash
return None, None
_cover_state.last_uri = song_uri
data: bytes | None = None
folder = _find_folder_cover(song_uri)
if folder:
try:
data = folder.read_bytes()
except OSError:
data = None
if not data:
try:
data = await _fetch_embedded_art(client, song_uri)
except Exception:
data = None
if not data:
_cover_state.last_hash = None
return None, None
digest = hashlib.sha1(data).hexdigest()[:12]
if digest != _cover_state.last_hash:
try:
COVER_OUT.write_bytes(data)
except OSError as e:
print(f"[cover] write failed: {e}", flush=True)
_cover_state.last_hash = None
return None, None
_cover_state.last_hash = digest
return f"../bridges/{COVER_OUT.name}", digest
# ─── 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()
# Peek the next up-to-3 tracks. MPD's `nextsong` is the index
# of the immediate successor; we walk forward from there until
# we hit the queue end or 3 entries, whichever comes first.
next_songs = []
if "nextsong" in status:
start = int(status["nextsong"])
total = int(status.get("playlistlength") or 0)
for i in range(start, min(start + 3, total)):
try:
rows = await client.playlistinfo(i)
if rows:
next_songs.append(rows[0])
except Exception:
break # transient — let it retry next tick
cover_path, cover_hash = await resolve_cover(client, song.get("file"))
stats = await refresh_stats(client)
state.data = mpd_to_event(status, song, next_songs,
cover_path, cover_hash, stats)
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)