From 5b2c5a3e14fabc8d466ed58f143ab9600289e280 Mon Sep 17 00:00:00 2001 From: media Date: Sun, 14 Jun 2026 14:27:26 -0500 Subject: [PATCH] Initial Commit --- config.yaml | 24 ++ player_service.py | 718 ++++++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 58 ++++ 3 files changed, 800 insertions(+) create mode 100644 config.yaml create mode 100644 player_service.py create mode 100644 requirements.txt diff --git a/config.yaml b/config.yaml new file mode 100644 index 0000000..4584aad --- /dev/null +++ b/config.yaml @@ -0,0 +1,24 @@ +mqtt: + host: 10.40.0.2 + port: 1883 + username: null + password: null + client_id_prefix: "ytplayer" + +topics: + cmd: "youtube/player/cmd" + status: "youtube/player/status" + +player: + display: ":0" + xdg_runtime_dir: "/run/user/1000" + mpv_binary: "/usr/bin/mpv" + +video_audio: + hdmi: + left: "Built-in Audio Pro 3:playback_AUX0" + right: "Built-in Audio Pro 3:playback_AUX1" + + analog: + left: "Built-in Audio Pro:playback_AUX0" + right: "Built-in Audio Pro:playback_AUX1" \ No newline at end of file diff --git a/player_service.py b/player_service.py new file mode 100644 index 0000000..b6419ff --- /dev/null +++ b/player_service.py @@ -0,0 +1,718 @@ +#!/usr/bin/env python3 +import json +import socket +import os +import signal +import subprocess +import sys +import time +from pathlib import Path +from typing import Optional, Dict, Any +import threading +import paho.mqtt.client as mqtt +import yaml +import jack + +# ---------------------------- +# Config + paths +# ---------------------------- + +BASE_DIR = Path(__file__).resolve().parent +CONFIG_PATH = BASE_DIR / "config.yaml" + +config = yaml.safe_load(CONFIG_PATH.read_text()) + +MEDIA_BASE = os.environ.get("YTPLAYER_MEDIA_BASE", "/mnt/Media/YTPlayer") + +MQTT_HOST = config["mqtt"]["host"] +MQTT_PORT = config["mqtt"].get("port", 1883) +MQTT_USER = config["mqtt"].get("username") +MQTT_PASS = config["mqtt"].get("password") +CLIENT_ID_PREFIX = config["mqtt"].get("client_id_prefix", "ytplayer") + +HA_DISCOVERY_PREFIX = config.get("homeassistant", {}).get("discovery_prefix", "homeassistant") +HA_DEVICE_ID = config.get("homeassistant", {}).get("device_id", "ytplayer") +HA_DEVICE_NAME = config.get("homeassistant", {}).get("device_name", "YT Player") + +TOPIC_CMD = config["topics"]["cmd"] # e.g. "youtube/player/cmd" +TOPIC_STATUS = config["topics"]["status"] # e.g. "youtube/player/status" +TOPIC_STATE = "youtube/player/state" # simple string state for Companion + +MPV_BIN = config["player"].get("mpv_binary", "/usr/bin/mpv") +DISPLAY = config["player"].get("display", ":0") +XDG_RUNTIME_DIR = config["player"].get("xdg_runtime_dir", "/run/user/1000") + +TELEMETRY_TOPIC = config.get("topics", {}).get("telemetry", "youtube/player/telemetry") +MPV_SOCKET = config["player"].get("ipc_socket", "/tmp/mpvsocket") +TELEMETRY_INTERVAL = float(config["player"].get("telemetry_interval", 0.5)) # seconds + +VIDEO_AUDIO_CONNECT_DELAY = float(config["player"].get("audio_connect_delay", 1.0)) + +VIDEO_HDMI_L = config["video_audio"]["hdmi"]["left"] +VIDEO_HDMI_R = config["video_audio"]["hdmi"]["right"] +VIDEO_ANALOG_L = config["video_audio"]["analog"]["left"] +VIDEO_ANALOG_R = config["video_audio"]["analog"]["right"] + +# ---------------------------- +# Runtime state +# ---------------------------- + +mpv_proc: Optional[subprocess.Popen] = None +current_path: Optional[str] = None +current_slot: Optional[int] = None +busy_until: Optional[float] = None +mpv_listener_thread: Optional["threading.Thread"] = None +mpv_listener_stop = None # will become threading.Event() + +client = mqtt.Client( + mqtt.CallbackAPIVersion.VERSION2, + client_id=f"{CLIENT_ID_PREFIX}-player", + clean_session=True, +) + +# ---------------------------- +# MQTT helpers +# ---------------------------- + +def publish_status(state: str, **extra: Any) -> None: + """ + Publish: + - JSON status (for HA / controller) to TOPIC_STATUS + - Flat string (for Companion color mapping) to TOPIC_STATE + """ + payload: Dict[str, Any] = {"state": state} + if current_path: + payload["path"] = current_path + if current_slot is not None: + payload["slot"] = current_slot + payload.update(extra) + + # JSON status + client.publish(TOPIC_STATUS, json.dumps(payload), qos=1, retain=False) + + # Flat state string + client.publish(TOPIC_STATE, state, qos=0, retain=True) + +def ha_discovery_sensor( + entity_id: str, + name: str, + value_template: str, + icon: str = "", + unit: str = "", +) -> None: + topic = f"{HA_DISCOVERY_PREFIX}/sensor/{HA_DEVICE_ID}/{entity_id}/config" + + payload: Dict[str, Any] = { + "name": name, + "unique_id": f"{HA_DEVICE_ID}_{entity_id}", + "state_topic": TELEMETRY_TOPIC, + "value_template": value_template, + + "availability_topic": TOPIC_STATUS, + "availability_template": "{% if value_json.state != 'offline' %}online{% else %}offline{% endif %}", + "payload_available": "online", + "payload_not_available": "offline", + + "device": { + "identifiers": [HA_DEVICE_ID], + "name": HA_DEVICE_NAME, + "manufacturer": "Custom", + "model": "mpv mqtt player", + }, + } + + + if icon: + payload["icon"] = icon + if unit: + payload["unit_of_measurement"] = unit + + client.publish(topic, json.dumps(payload), qos=1, retain=True) + +# ---------------------------- +# MPV control +# ---------------------------- + +def stop_mpv(reason: str = "stop") -> None: + """ + Stop mpv if running, reset state, publish idle. + + Important: we keep the slot around long enough so the controller + can see which slot just went idle/finished. + """ + global mpv_proc, current_path, current_slot + global mpv_listener_thread, mpv_listener_stop + if mpv_listener_stop is not None: + mpv_listener_stop.set() + mpv_listener_thread = None + mpv_listener_stop = None + + slot_before = current_slot + + if mpv_proc and mpv_proc.poll() is None: + try: + mpv_proc.terminate() + try: + mpv_proc.wait(timeout=5) + except subprocess.TimeoutExpired: + mpv_proc.kill() + except Exception as e: + # still publish error, with whatever slot we had + current_slot = slot_before + publish_status("error", message=f"Failed to stop mpv: {e}") + current_slot = None + + mpv_proc = None + current_path = None + # clear telemetry so HA doesn't keep stale values + publish_cleared_telemetry() + # temporarily restore slot so status includes it + current_slot = slot_before + publish_status("idle", reason=reason) + current_slot = None + + +def handle_play(cmd): + """ + cmd is a dict, expected keys: + - action: "play" + - path: absolute path to .m3u (optional) + - playlist: optional explicit playlist path + - slot: optional int (1..N) – controller usually sends this + - label/title: optional + """ + global mpv_proc, current_path, current_slot, busy_until + + print(f"[player] handle_play cmd={cmd!r}") + + path = cmd.get("path") + playlist = cmd.get("playlist") + slot = cmd.get("slot") + label = cmd.get("label") or cmd.get("title") + + # Decide what we’re actually going to play + final_path = None + + # 1) explicit path wins + if path: + final_path = path + # 2) explicit playlist from controller + elif playlist: + final_path = playlist + # 3) derive from slot: /mnt/Media/YTPlayer/slotN/playlist.m3u + elif slot is not None: + try: + slot_num = int(slot) + except (TypeError, ValueError): + print(f"[player] bad slot value: {slot!r}") + publish_status("error", message=f"bad slot value: {slot!r}") + return + final_path = f"{MEDIA_BASE}/slot{slot_num}/playlist.m3u" + + if not final_path: + print("[player] no valid 'path', 'playlist' or 'slot' supplied") + publish_status("error", message="play command missing path/playlist/slot") + return + + final_path = os.path.abspath(final_path) + if not os.path.exists(final_path): + print(f"[player] path not found: {final_path}") + publish_status("error", message=f"path not found: {final_path}") + return + + # enforce “must be stopped before new playlist” + if mpv_proc and mpv_proc.poll() is None: + print("[player] already playing, sending busy") + publish_status("busy", message="mpv already playing; stop first") + busy_until = time.time() + 5 + return + + args = [ + MPV_BIN, + f"--input-ipc-server={MPV_SOCKET}", + "--audio-client-name=MPV", + ] + if final_path.endswith(".m3u"): + # mpv wants --playlist=/path/to/file, not "--playlist" "/path" + args.append(f"--playlist={final_path}") + else: + args.append(final_path) + + env = os.environ.copy() + env["DISPLAY"] = DISPLAY + env["XDG_RUNTIME_DIR"] = XDG_RUNTIME_DIR + + try: + print(f"[player] starting mpv: {args!r}") + mpv_proc = subprocess.Popen( + args, + env=env, + stdout=None, + stderr=None, + ) + threading.Thread(target=connect_mpv_audio, daemon=True).start() + current_path = final_path + try: + current_slot_int = int(slot) if slot is not None else None + except (TypeError, ValueError): + current_slot_int = None + current_slot = current_slot_int + publish_status("playing", label=label) + # start telemetry observer thread + global mpv_listener_thread, mpv_listener_stop + mpv_listener_stop = threading.Event() + mpv_listener_thread = threading.Thread(target=mpv_observe_properties, daemon=True) + mpv_listener_thread.start() + except Exception as e: + print(f"[player] failed to start mpv: {e}") + mpv_proc = None + current_path = None + current_slot = None + publish_status("error", message=f"failed to start mpv: {e}") + +def mpv_ipc(command): + """Send a JSON IPC command to MPV via MPV_SOCKET.""" + try: + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + # Wait for socket to appear and accept connections + deadline = time.time() + 3.0 + while True: + try: + sock.connect(MPV_SOCKET) + break + except FileNotFoundError: + if time.time() > deadline: + raise + time.sleep(0.05) + except ConnectionRefusedError: + if time.time() > deadline: + raise + time.sleep(0.05) + sock.sendall((json.dumps({"command": command}) + "\n").encode("utf-8")) + sock.close() + return True + except Exception as e: + print(f"[player] mpv_ipc error: {e}") + return False + +def fmt_mmss(value: Optional[float]) -> Optional[str]: + if not isinstance(value, (int, float)) or value < 0: + return None + + total = int(value) + s = total % 60 + m = (total // 60) % 60 + h = total // 3600 + + if h > 0: + return f"{h}:{m:02d}:{s:02d}" + return f"{m}:{s:02d}" + +def publish_telemetry( + *, + pos: Optional[float], + dur: Optional[float], + paused: Optional[bool], + title: Optional[str], + filename: Optional[str], + playlist_pos: Optional[int], + playlist_count: Optional[int], +) -> None: + rem = None + if isinstance(pos, (int, float)) and isinstance(dur, (int, float)): + rem = max(0.0, float(dur) - float(pos)) + + payload: Dict[str, Any] = { + "state": "telemetry", + "slot": current_slot, + "path": current_path, + + # raw seconds + "position": pos, + "duration": dur, + "remaining": rem, + + # formatted strings + "position_fmt": fmt_mmss(pos), + "duration_fmt": fmt_mmss(dur), + "remaining_fmt": fmt_mmss(rem), + + "paused": paused, + "title": title, + "filename": filename, + "playlist_pos": playlist_pos, + "playlist_count": playlist_count, + "ts": time.time(), + } + + client.publish(TELEMETRY_TOPIC, json.dumps(payload), qos=0, retain=False) + +def publish_cleared_telemetry() -> None: + payload: Dict[str, Any] = { + "state": "telemetry", + "slot": current_slot, + "path": None, + + "position": 0, + "duration": 0, + "remaining": 0, + + "position_fmt": "0:00", + "duration_fmt": "0:00", + "remaining_fmt": "0:00", + + "paused": False, + "title": None, + "filename": None, + "playlist_pos": None, + "playlist_count": None, + "ts": time.time(), + } + + client.publish(TELEMETRY_TOPIC, json.dumps(payload), qos=0, retain=False) + +def mpv_observe_properties() -> None: + + """ + Keep a persistent IPC connection to mpv and observe properties. + Publishes telemetry to MQTT at TELEMETRY_INTERVAL cadence. + """ + global mpv_listener_stop + + # local cache + pos: Optional[float] = None + dur: Optional[float] = None + paused: Optional[bool] = None + title: Optional[str] = None + filename: Optional[str] = None + playlist_pos: Optional[int] = None + playlist_count: Optional[int] = None + + last_pub = 0.0 + + try: + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + deadline = time.time() + 3.0 + while True: + try: + sock.connect(MPV_SOCKET) + break + except (FileNotFoundError, ConnectionRefusedError): + if time.time() > deadline: + raise + time.sleep(0.05) + + sock.settimeout(1.0) + + + def send(cmd): + sock.sendall((json.dumps({"command": cmd}) + "\n").encode("utf-8")) + + # Observe useful properties + send(["observe_property", 1, "time-pos"]) # float seconds + send(["observe_property", 2, "duration"]) # float seconds + send(["observe_property", 3, "pause"]) # bool + send(["observe_property", 4, "media-title"]) # str (nice) + send(["observe_property", 5, "filename"]) # str + send(["observe_property", 6, "playlist-pos"]) # int + send(["observe_property", 7, "playlist-count"]) # int + print(f"[player] telemetry connected to mpv socket {MPV_SOCKET}") + publish_telemetry( + pos=pos, dur=dur, paused=paused, title=title, + filename=filename, playlist_pos=playlist_pos, playlist_count=playlist_count + ) + + while mpv_listener_stop is not None and not mpv_listener_stop.is_set(): + try: + chunk = sock.recv(65536) + if not chunk: + break + + for line in chunk.decode("utf-8", errors="ignore").splitlines(): + if not line.strip(): + continue + try: + evt = json.loads(line) + except json.JSONDecodeError: + continue + + if evt.get("event") != "property-change": + continue + + name = evt.get("name") + val = evt.get("data") + + if name == "time-pos": + pos = float(val) if isinstance(val, (int, float)) else None + elif name == "duration": + dur = float(val) if isinstance(val, (int, float)) else None + elif name == "pause": + paused = bool(val) if isinstance(val, bool) else None + elif name == "media-title": + title = str(val) if isinstance(val, str) else None + elif name == "filename": + filename = str(val) if isinstance(val, str) else None + elif name == "playlist-pos": + playlist_pos = int(val) if isinstance(val, (int, float)) else None + elif name == "playlist-count": + playlist_count = int(val) if isinstance(val, (int, float)) else None + + now = time.time() + if now - last_pub >= TELEMETRY_INTERVAL: + publish_telemetry( + pos=pos, + dur=dur, + paused=paused, + title=title, + filename=filename, + playlist_pos=playlist_pos, + playlist_count=playlist_count, + ) + last_pub = now + + except socket.timeout: + # still periodically publish even if no new events + now = time.time() + if now - last_pub >= TELEMETRY_INTERVAL: + publish_telemetry( + pos=pos, + dur=dur, + paused=paused, + title=title, + filename=filename, + playlist_pos=playlist_pos, + playlist_count=playlist_count, + ) + last_pub = now + continue + + except Exception as e: + print(f"[player] mpv_observe_properties error: {e}") + finally: + try: + sock.close() + except Exception: + pass + +def connect_mpv_audio(): + deadline = time.time() + 8 + + while time.time() < deadline: + try: + jack_client = jack.Client("MPVAutoRouter", no_start_server=True) + + with jack_client: + ports = [p.name for p in jack_client.get_ports()] + + left_candidates = [ + p for p in ports + if ( + p.startswith("MPV:") or p.startswith("mpv:") + ) and ( + "output_FL" in p or "out_0" in p or "output_0" in p + ) + ] + + right_candidates = [ + p for p in ports + if ( + p.startswith("MPV:") or p.startswith("mpv:") + ) and ( + "output_FR" in p or "out_1" in p or "output_1" in p + ) + ] + + if left_candidates and right_candidates: + mpv_l = left_candidates[0] + mpv_r = right_candidates[0] + + routes = [ + (mpv_l, VIDEO_HDMI_L), + (mpv_r, VIDEO_HDMI_R), + (mpv_l, VIDEO_ANALOG_L), + (mpv_r, VIDEO_ANALOG_R), + ] + + for src, dst in routes: + try: + jack_client.connect(src, dst) + print(f"[player] connected video audio: {src} -> {dst}") + except jack.JackError as e: + print(f"[player] route skipped/failed {src} -> {dst}: {e}") + + return + + print("[player] waiting for MPV audio ports...") + + except Exception as e: + print(f"[player] connect_mpv_audio retry error: {e}") + + time.sleep(0.25) + + print("[player] mpv audio ports not found after timeout") + +# ---------------------------- +# Command dispatch +# ---------------------------- + +def handle_cmd(payload_str: str) -> None: + """Decode JSON and route to play/stop/etc.""" + try: + data = json.loads(payload_str or "{}") + except json.JSONDecodeError: + print(f"[player] bad JSON in cmd: {payload_str!r}") + return + + action = data.get("action") + if not action: + print("[player] cmd missing 'action'") + return + + print(f"[player] cmd received: {data!r}") + + if action == "play": + cmd = { + "action": "play", + "path": data.get("path"), + "playlist": data.get("playlist"), + "slot": data.get("slot"), + "label": data.get("label") or data.get("title"), + } + handle_play(cmd) + return + + if action == "stop": + print("[player] STOP command") + stop_mpv("stop") + return + + elif action == "pause": + if mpv_proc and mpv_proc.poll() is None: + mpv_ipc(["set_property", "pause", True]) + publish_status("paused") + else: + publish_status("error", message="nothing playing to pause") + + elif action == "resume": + if mpv_proc and mpv_proc.poll() is None: + mpv_ipc(["set_property", "pause", False]) + publish_status("playing") + else: + publish_status("error", message="nothing playing to resume") + + elif action == "toggle": + if mpv_proc and mpv_proc.poll() is None: + mpv_ipc(["cycle", "pause"]) + publish_status("toggled") + else: + publish_status("error", message="nothing playing to toggle") + + + print(f"[player] unknown action: {action!r}") + + +# ---------------------------- +# MQTT callbacks +# ---------------------------- + +def on_connect(client, userdata, flags, rc, properties=None): + if rc == 0: + print(f"[player] MQTT connected rc={rc}, subscribing {TOPIC_CMD}") + client.subscribe(TOPIC_CMD, qos=1) + # initial idle + publish_status("idle", message="player online") + # Home Assistant MQTT Discovery sensors + ha_discovery_sensor("position", "YT Position (s)", "{{ value_json.position }}", icon="mdi:timeline-clock", unit="s") + ha_discovery_sensor("duration", "YT Duration (s)", "{{ value_json.duration }}", icon="mdi:timer", unit="s") + ha_discovery_sensor("remaining", "YT Remaining (s)", "{{ value_json.remaining }}", icon="mdi:timer-sand", unit="s") + + ha_discovery_sensor("position_fmt", "YT Position", "{{ value_json.position_fmt }}", icon="mdi:timeline-clock") + ha_discovery_sensor("duration_fmt", "YT Duration", "{{ value_json.duration_fmt }}", icon="mdi:timer") + ha_discovery_sensor("remaining_fmt", "YT Remaining", "{{ value_json.remaining_fmt }}", icon="mdi:timer-sand") + + ha_discovery_sensor("title", "YT Title", "{{ value_json.title }}", icon="mdi:music") + ha_discovery_sensor("paused", "YT Paused", "{{ value_json.paused }}", icon="mdi:pause-circle") + ha_discovery_sensor("slot", "YT Slot", "{{ value_json.slot }}", icon="mdi:playlist-play") + + else: + sys.stderr.write(f"[player] MQTT connect failed with code {rc}\n") + + +def on_message(client, userdata, msg): + if msg.topic == TOPIC_CMD: + payload = msg.payload.decode("utf-8", errors="ignore") + handle_cmd(payload) + + +# ---------------------------- +# Signal handling +# ---------------------------- + +def sigterm_handler(signum, frame): + print("[player] SIGTERM received, stopping mpv") + stop_mpv(reason="sigterm") + sys.exit(0) + + +# ---------------------------- +# Main loop +# ---------------------------- + +def main() -> None: + global mpv_proc, busy_until + + signal.signal(signal.SIGTERM, sigterm_handler) + signal.signal(signal.SIGINT, sigterm_handler) + + if MQTT_USER: + client.username_pw_set(MQTT_USER, MQTT_PASS or None) + + client.on_connect = on_connect + client.on_message = on_message + + client.will_set( + TOPIC_STATUS, + json.dumps({"state": "offline"}), + qos=1, + retain=False, + ) + + print(f"[player] connecting MQTT to {MQTT_HOST}:{MQTT_PORT}") + client.connect(MQTT_HOST, MQTT_PORT, keepalive=30) + client.loop_start() + + try: + while True: + now = time.time() + + # Watch mpv exit + if mpv_proc is not None: + ret = mpv_proc.poll() + if ret is not None: + if ret == 0: + print("[player] mpv finished normally") + stop_mpv(reason="finished") + else: + print(f"[player] mpv exited with code {ret}") + publish_status("error", message=f"mpv exited code {ret}") + stop_mpv(reason="mpv_error") + + # If we previously reported "busy", re-assert "playing" after delay + if ( + busy_until is not None + and now >= busy_until + and mpv_proc is not None + and mpv_proc.poll() is None + ): + print("[player] clearing busy → playing") + publish_status("playing") + busy_until = None + + time.sleep(0.2) + finally: + client.loop_stop() + stop_mpv(reason="shutdown") + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..fe752dd --- /dev/null +++ b/requirements.txt @@ -0,0 +1,58 @@ +attrs==22.2.0 +Brotli==1.0.9 +certifi==2022.9.24 +cffi==1.15.1 +chardet==5.1.0 +charset-normalizer==3.0.1 +crit==3.17.1 +cupshelpers==1.0 +dbus-python==1.3.2 +distro==1.8.0 +distro-info==1.5+deb12u1 +docker==5.0.3 +docker-compose==1.29.2 +dockerpty==0.4.1 +docopt==0.6.2 +gyp==0.1 +httplib2==0.20.4 +idna==3.3 +JACK-Client==0.5.3 +jsonpointer==2.3 +jsonschema==4.10.3 +Mako==1.2.4.dev0 +Markdown==3.4.1 +MarkupSafe==2.1.2 +mutagen==1.46.0 +numpy==1.24.2 +paho-mqtt==2.1.0 +ply==3.11 +protobuf==4.21.12 +pycairo==1.20.1 +pycparser==2.21 +pycryptodomex==3.11.0 +pycups==2.0.1 +pycurl==7.45.2 +Pygments==2.14.0 +PyGObject==3.42.2 +pyparsing==3.0.9 +pyrsistent==0.18.1 +PySimpleSOAP==1.16.2 +pysmbc==1.0.23 +python-apt==2.6.0 +python-debian==0.1.49 +python-debianbts==4.0.1 +python-dotenv==0.21.0 +pyxattr==0.8.1 +PyYAML==6.0 +reportbug==12.0.0 +requests==2.28.1 +rfc3987==1.3.8 +six==1.16.0 +texttable==1.6.7 +unattended-upgrades==0.1 +uritemplate==4.1.1 +urllib3==1.26.12 +webcolors==1.11.1 +websocket-client==1.2.3 +websockets==10.4 +yt-dlp==2025.9.26