add now-playing API

app.py

@@ -1,11 +1,15 @@
+from typing import Any, Mapping
 from flask import Flask, render_template, send_file, Response
 import os
 import subprocess
 import signal
 from pathlib import Path
 import tempfile
+import socket
+import json
 
 mpv_pidfile = Path(tempfile.gettempdir()).joinpath("laas_mpv.pidfile")
+mpv_socket = Path(tempfile.gettempdir()).joinpath("mpvsocket")
 music_path = Path(os.environ.get("MUSIC_PATH", "./music"))
 mpv_process: subprocess.Popen | None = None
 app = Flask(__name__)
@@ -40,7 +44,7 @@ def playback_start():
     mpv_process = subprocess.Popen(
         [
             "mpv",
-            "--input-ipc-server=/tmp/mpvsocket",
+            f"--input-ipc-server={str(mpv_socket)}",
             "--shuffle",
             str(music_path),
             "--no-video",
@@ -141,6 +145,45 @@ def api_status():
     return Response(str(is_playing()).lower(), headers={"Content-Type": "text/json"})
 
 
+@app.route("/api/nowplaying", methods=["GET"])
+def api_nowplaying():
+    response: dict[str, Any] = {}
+
+    if is_playing():
+        sock = socket.socket(socket.AF_UNIX)
+        while True:
+            try:
+                sock.connect(str(mpv_socket).encode())
+                break
+            except (ConnectionRefusedError, FileNotFoundError):
+                pass
+
+        for prop in [
+            "filename",
+            ("metadata/by-key/artist", "artist"),
+            ("metadata/by-key/album", "album"),
+            ("media-title", "title"),
+            "playback-time",
+            "duration",
+        ]:
+            if isinstance(prop, tuple):
+                prop, key = prop
+            else:
+                key = prop
+
+            command = {"command": ["get_property", prop]}
+            sock.send((json.dumps(command) + "\n").encode())
+
+            reply = b""
+            while (data := sock.recv(1)) != b"\n":
+                reply += data
+
+            reply_json: Mapping[str, Any] = json.loads(reply.decode())
+            response[key] = reply_json["data"] if "data" in reply_json else "Unknown"
+
+    return Response(json.dumps(response), headers={"Content-Type": "text/json"})
+
+
 signal.signal(signal.SIGTERM, sigh)
 signal.signal(signal.SIGINT, sigh)
 cleanup_unclean()