152 lines
4.5 KiB
Python
152 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Feeds the eww now-playing widget one JSON object per MPRIS change.
|
|
|
|
Installed to /usr/local/bin/now-playing-json, consumed by eww's `deflisten`.
|
|
|
|
Python rather than shell: track titles are arbitrary user text and have to end up
|
|
inside a JSON string, which is exactly the thing shell quoting gets wrong. Python 3 is
|
|
already a hard dependency of the image (thinclient-agent runs on it), so this costs
|
|
nothing extra.
|
|
|
|
`playerctl --follow` rather than the polling loop in thinclient_agent/mpris_bridge.py:
|
|
that module is a system service that must survive the session bus disappearing, so it
|
|
polls. This runs inside the session and can afford to block on a subscription, which
|
|
makes the widget update on the beat instead of up to 2s late.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
# Mirrors thinclient_agent.mpris_bridge.PLAYER_PRIORITY and the `playerctl -p` lists in
|
|
# configs/sway/config and configs/eww/eww.yuck — keep the four in step.
|
|
PLAYER_PRIORITY = "mpv,spotifyd,%any"
|
|
|
|
SEP = "\x1f"
|
|
FORMAT = SEP.join(
|
|
("{{status}}", "{{title}}", "{{artist}}", "{{album}}", "{{mpris:artUrl}}")
|
|
)
|
|
|
|
CACHE_DIR = os.path.join(
|
|
os.environ.get("XDG_CACHE_HOME") or os.path.expanduser("~/.cache"),
|
|
"now-playing",
|
|
)
|
|
|
|
# eww's image widget logs an error and leaves a broken box when :path does not resolve,
|
|
# so there is always a real file to point at.
|
|
PLACEHOLDER_PNG = base64.b64decode(
|
|
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="
|
|
)
|
|
|
|
|
|
def placeholder_path() -> str:
|
|
path = os.path.join(CACHE_DIR, "placeholder.png")
|
|
if not os.path.exists(path):
|
|
with open(path, "wb") as handle:
|
|
handle.write(PLACEHOLDER_PNG)
|
|
return path
|
|
|
|
|
|
def cache_art(art_url: str) -> str:
|
|
if not art_url:
|
|
return ""
|
|
|
|
parsed = urllib.parse.urlparse(art_url)
|
|
if parsed.scheme == "file":
|
|
local = urllib.parse.unquote(parsed.path)
|
|
return local if os.path.isfile(local) else ""
|
|
|
|
if parsed.scheme not in ("http", "https"):
|
|
return ""
|
|
|
|
# Cached by URL digest so a repeated track does not re-fetch, and so nothing from
|
|
# the (remote, untrusted) URL ever reaches the filesystem as a path component.
|
|
target = os.path.join(CACHE_DIR, hashlib.sha256(art_url.encode()).hexdigest() + ".img")
|
|
if os.path.exists(target):
|
|
return target
|
|
|
|
try:
|
|
with urllib.request.urlopen(art_url, timeout=5) as response:
|
|
data = response.read(4 * 1024 * 1024)
|
|
except Exception:
|
|
return ""
|
|
|
|
tmp = target + ".part"
|
|
with open(tmp, "wb") as handle:
|
|
handle.write(data)
|
|
os.replace(tmp, target)
|
|
return target
|
|
|
|
|
|
def emit(state: dict) -> None:
|
|
sys.stdout.write(json.dumps(state) + "\n")
|
|
sys.stdout.flush()
|
|
|
|
|
|
def blank() -> dict:
|
|
return {
|
|
"visible": False,
|
|
"title": "",
|
|
"artist": "",
|
|
"album": "",
|
|
"art": placeholder_path(),
|
|
"controls": False,
|
|
}
|
|
|
|
|
|
def parse(line: str) -> dict:
|
|
fields = (line.split(SEP) + [""] * 5)[:5]
|
|
status, title, artist, album, art_url = (f.strip() for f in fields)
|
|
|
|
if status not in ("Playing", "Paused"):
|
|
return blank()
|
|
|
|
art = cache_art(art_url)
|
|
return {
|
|
"visible": True,
|
|
"title": title or "Unknown track",
|
|
"artist": artist,
|
|
"album": album,
|
|
# Sources that expose no art degrade to text-only rather than showing a
|
|
# placeholder box; the widget hides the image when this is the 1x1 pixel.
|
|
"art": art or placeholder_path(),
|
|
# playerctl's own transport commands work for any MPRIS player that answers
|
|
# `status`, so controls follow the same signal rather than a separate probe.
|
|
"controls": True,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
os.makedirs(CACHE_DIR, exist_ok=True)
|
|
emit(blank())
|
|
|
|
process = subprocess.Popen(
|
|
["playerctl", "-p", PLAYER_PRIORITY, "metadata", "--follow", "--format", FORMAT],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL,
|
|
text=True,
|
|
)
|
|
|
|
assert process.stdout is not None
|
|
for line in process.stdout:
|
|
line = line.rstrip("\n")
|
|
try:
|
|
emit(parse(line) if line else blank())
|
|
except Exception:
|
|
emit(blank())
|
|
|
|
# playerctl exits when the session bus goes away, i.e. when sway is going down.
|
|
emit(blank())
|
|
return process.wait()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|