1083 lines
35 KiB
Python
1083 lines
35 KiB
Python
#!/usr/bin/env python3
|
||
import json
|
||
import math
|
||
import socket
|
||
import os
|
||
import queue
|
||
import signal
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import traceback
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import Optional, Dict, Any
|
||
import threading
|
||
import paho.mqtt.client as mqtt
|
||
import yaml
|
||
|
||
# ----------------------------
|
||
# 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_CONFIG = config.get("homeassistant", {})
|
||
HA_DISCOVERY_ENABLED = HA_CONFIG.get("enabled", True)
|
||
HA_DISCOVERY_PREFIX = HA_CONFIG.get("discovery_prefix", "homeassistant")
|
||
HA_DEVICE_ID = HA_CONFIG.get("device_id", "ytplayer")
|
||
HA_DEVICE_NAME = HA_CONFIG.get("device_name", "YT Player")
|
||
|
||
HA_DISCOVERY_IMPORT_ERROR: Optional[Exception] = None
|
||
if HA_DISCOVERY_ENABLED:
|
||
try:
|
||
from homeassistant_discovery import publish_discovery
|
||
except Exception as exc:
|
||
# Discovery is optional. A broken/missing helper must not prevent the
|
||
# core player and its established MQTT interface from starting.
|
||
publish_discovery = None
|
||
HA_DISCOVERY_IMPORT_ERROR = exc
|
||
else:
|
||
publish_discovery = None
|
||
|
||
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 = max(0.1, float(config["player"].get("telemetry_interval", 0.5)))
|
||
COMMANDS_PER_TICK = max(1, int(config["player"].get("commands_per_tick", 32)))
|
||
COMMAND_QUEUE_SIZE = max(1, int(config["player"].get("command_queue_size", 512)))
|
||
MPV_AUDIO_DEVICE = config["player"].get("audio_device", "pipewire/ytplayer_bus")
|
||
MPV_AUDIO_CLIENT_NAME = config["player"].get("audio_client_name", "YTPlayer")
|
||
MPV_AUDIO_CHANNELS = config["player"].get("audio_channels", "stereo")
|
||
|
||
# ----------------------------
|
||
# Runtime state
|
||
# ----------------------------
|
||
|
||
@dataclass
|
||
class PlaybackContext:
|
||
generation: int
|
||
process: subprocess.Popen
|
||
path: str
|
||
slot: Optional[int]
|
||
stop_event: threading.Event
|
||
listener_thread: Optional[threading.Thread] = None
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class QueuedCommand:
|
||
data: Dict[str, Any]
|
||
message_id: Optional[int] = None
|
||
qos: int = 0
|
||
connection_generation: Optional[int] = None
|
||
|
||
|
||
current_playback: Optional[PlaybackContext] = None
|
||
playback_generation = 0
|
||
busy_until: Optional[float] = None
|
||
state_lock = threading.RLock()
|
||
shutdown_event = threading.Event()
|
||
mqtt_generation_lock = threading.Lock()
|
||
mqtt_connection_generation = 0
|
||
|
||
command_queue: "queue.Queue[QueuedCommand]" = queue.Queue(maxsize=COMMAND_QUEUE_SIZE)
|
||
|
||
client = mqtt.Client(
|
||
mqtt.CallbackAPIVersion.VERSION2,
|
||
client_id=f"{CLIENT_ID_PREFIX}-player",
|
||
clean_session=True,
|
||
)
|
||
|
||
|
||
def log(message: str, *, error: bool = False) -> None:
|
||
"""Write each record immediately so journald timestamps are accurate."""
|
||
print(message, file=sys.stderr if error else sys.stdout, flush=True)
|
||
|
||
|
||
def log_exception(message: str) -> None:
|
||
log(message, error=True)
|
||
traceback.print_exc(file=sys.stderr)
|
||
|
||
# ----------------------------
|
||
# MQTT helpers
|
||
# ----------------------------
|
||
|
||
_USE_CURRENT_PLAYBACK = object()
|
||
|
||
|
||
def get_current_playback() -> Optional[PlaybackContext]:
|
||
with state_lock:
|
||
return current_playback
|
||
|
||
|
||
def playback_is_current(playback: PlaybackContext) -> bool:
|
||
with state_lock:
|
||
return current_playback is playback and not playback.stop_event.is_set()
|
||
|
||
|
||
def publish_status(
|
||
state: str,
|
||
*,
|
||
playback: Any = _USE_CURRENT_PLAYBACK,
|
||
include_path: bool = True,
|
||
**extra: Any,
|
||
) -> None:
|
||
"""
|
||
Publish:
|
||
- JSON status (for HA / controller) to TOPIC_STATUS
|
||
- Flat string (for Companion color mapping) to TOPIC_STATE
|
||
"""
|
||
if playback is _USE_CURRENT_PLAYBACK:
|
||
playback = get_current_playback()
|
||
|
||
payload: Dict[str, Any] = {"state": state}
|
||
if playback is not None:
|
||
if include_path and playback.path:
|
||
payload["path"] = playback.path
|
||
if playback.slot is not None:
|
||
payload["slot"] = playback.slot
|
||
payload.update(extra)
|
||
|
||
try:
|
||
# These topic names, QoS levels, retain flags, and payload shapes are
|
||
# part of the existing controller contract. Do not change them here.
|
||
client.publish(TOPIC_STATUS, json.dumps(payload), qos=1, retain=False)
|
||
client.publish(TOPIC_STATE, state, qos=0, retain=True)
|
||
except Exception:
|
||
log_exception(f"[player] failed to publish status {state!r}")
|
||
|
||
# ----------------------------
|
||
# MPV control
|
||
# ----------------------------
|
||
|
||
def _remove_mpv_socket() -> None:
|
||
try:
|
||
os.unlink(MPV_SOCKET)
|
||
except FileNotFoundError:
|
||
pass
|
||
except OSError as exc:
|
||
log(f"[player] could not remove stale MPV socket {MPV_SOCKET}: {exc}", error=True)
|
||
|
||
|
||
def _join_playback_workers(playback: PlaybackContext) -> None:
|
||
for worker in (playback.listener_thread,):
|
||
if worker is None or worker is threading.current_thread():
|
||
continue
|
||
try:
|
||
worker.join(timeout=1.5)
|
||
except RuntimeError:
|
||
# Thread construction succeeded but start() did not.
|
||
continue
|
||
if worker.is_alive():
|
||
log(
|
||
f"[player] worker {worker.name!r} did not stop promptly "
|
||
f"for generation {playback.generation}",
|
||
error=True,
|
||
)
|
||
|
||
|
||
def stop_mpv(
|
||
reason: str = "stop",
|
||
*,
|
||
expected: Optional[PlaybackContext] = None,
|
||
final_state: str = "idle",
|
||
message: Optional[str] = None,
|
||
publish_updates: bool = True,
|
||
) -> bool:
|
||
"""
|
||
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 current_playback, busy_until
|
||
|
||
with state_lock:
|
||
playback = current_playback
|
||
if expected is not None and playback is not expected:
|
||
return False
|
||
busy_until = None
|
||
if playback is not None:
|
||
playback.stop_event.set()
|
||
|
||
stop_error: Optional[str] = None
|
||
if playback is not None and playback.process.poll() is None:
|
||
try:
|
||
playback.process.terminate()
|
||
try:
|
||
playback.process.wait(timeout=5)
|
||
except subprocess.TimeoutExpired:
|
||
playback.process.kill()
|
||
playback.process.wait(timeout=2)
|
||
except Exception as exc:
|
||
stop_error = f"Failed to stop mpv: {exc}"
|
||
log(f"[player] {stop_error}", error=True)
|
||
|
||
# terminate() itself can fail. Make a separate best-effort kill so
|
||
# the daemon never abandons a live child and starts a duplicate.
|
||
if playback.process.poll() is None:
|
||
try:
|
||
playback.process.kill()
|
||
playback.process.wait(timeout=2)
|
||
except Exception as kill_exc:
|
||
log(
|
||
f"[player] Failed to kill mpv after stop error: {kill_exc}",
|
||
error=True,
|
||
)
|
||
|
||
if playback is not None and playback.process.poll() is None:
|
||
# Keep ownership of a child that could not be stopped. A later stop or
|
||
# systemd's cgroup cleanup can retry; handle_play will remain busy and
|
||
# cannot create a second MPV instance.
|
||
if publish_updates:
|
||
publish_status(
|
||
"error",
|
||
playback=playback,
|
||
message=stop_error or "Failed to stop mpv; process is still running",
|
||
)
|
||
return False
|
||
|
||
with state_lock:
|
||
if current_playback is playback:
|
||
current_playback = None
|
||
|
||
if playback is not None:
|
||
_join_playback_workers(playback)
|
||
_remove_mpv_socket()
|
||
|
||
if not publish_updates:
|
||
return True
|
||
|
||
if stop_error is not None:
|
||
# Preserve the historical error-then-idle state sequence for existing
|
||
# MQTT consumers while still completing cleanup.
|
||
publish_status("error", playback=playback, message=stop_error)
|
||
|
||
# Keep the previous slot in the clear/terminal messages, matching the
|
||
# existing controller contract, but do not carry over the old path.
|
||
publish_cleared_telemetry(
|
||
playback=None if stop_error is not None else playback,
|
||
)
|
||
|
||
extra: Dict[str, Any] = {"reason": reason}
|
||
if message is not None:
|
||
extra["message"] = message
|
||
publish_status(
|
||
final_state,
|
||
playback=playback,
|
||
include_path=final_state == "error",
|
||
**extra,
|
||
)
|
||
return True
|
||
|
||
|
||
def handle_play(cmd: Dict[str, Any]) -> None:
|
||
"""
|
||
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 current_playback, playback_generation, busy_until
|
||
|
||
log(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")
|
||
|
||
if path is not None and not isinstance(path, str):
|
||
log(f"[player] bad path value: {path!r}", error=True)
|
||
publish_status("error", message="play command path must be a string")
|
||
return
|
||
if playlist is not None and not isinstance(playlist, str):
|
||
log(f"[player] bad playlist value: {playlist!r}", error=True)
|
||
publish_status("error", message="play command playlist must be a string")
|
||
return
|
||
|
||
slot_num: Optional[int] = None
|
||
if slot is not None:
|
||
try:
|
||
if isinstance(slot, bool):
|
||
raise ValueError("boolean slots are not valid")
|
||
slot_num = int(slot)
|
||
except (TypeError, ValueError, OverflowError):
|
||
if path or playlist:
|
||
log(f"[player] ignoring bad optional slot value: {slot!r}", error=True)
|
||
slot_num = None
|
||
else:
|
||
log(f"[player] bad slot value: {slot!r}", error=True)
|
||
publish_status("error", message=f"bad slot value: {slot!r}")
|
||
return
|
||
|
||
# Decide what we're actually going to play.
|
||
final_path: Optional[str] = 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_num is not None:
|
||
final_path = f"{MEDIA_BASE}/slot{slot_num}/playlist.m3u"
|
||
|
||
if not final_path:
|
||
log("[player] no valid 'path', 'playlist' or 'slot' supplied", error=True)
|
||
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):
|
||
log(f"[player] path not found: {final_path}", error=True)
|
||
publish_status("error", message=f"path not found: {final_path}")
|
||
return
|
||
|
||
# enforce “must be stopped before new playlist”
|
||
existing = get_current_playback()
|
||
if existing is not None and existing.process.poll() is None:
|
||
log("[player] already playing, sending busy")
|
||
publish_status("busy", message="mpv already playing; stop first")
|
||
busy_until = time.monotonic() + 5
|
||
return
|
||
|
||
if existing is not None:
|
||
stop_mpv(
|
||
reason="stale_process",
|
||
expected=existing,
|
||
publish_updates=False,
|
||
)
|
||
|
||
args = [
|
||
MPV_BIN,
|
||
f"--input-ipc-server={MPV_SOCKET}",
|
||
f"--audio-device={MPV_AUDIO_DEVICE}",
|
||
f"--audio-client-name={MPV_AUDIO_CLIENT_NAME}",
|
||
f"--audio-channels={MPV_AUDIO_CHANNELS}",
|
||
# Never fall back to a physical default sink. Audio Router is the sole
|
||
# owner of the HDMI/analog choice.
|
||
"--audio-fallback-to-null=no",
|
||
# Keep warnings/errors but suppress the continuously rewritten AV
|
||
# progress line that journald stores as very large blob records.
|
||
"--quiet",
|
||
]
|
||
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
|
||
|
||
_remove_mpv_socket()
|
||
process: Optional[subprocess.Popen] = None
|
||
playback: Optional[PlaybackContext] = None
|
||
try:
|
||
log(f"[player] starting mpv: {args!r}")
|
||
process = subprocess.Popen(
|
||
args,
|
||
env=env,
|
||
stdout=None,
|
||
stderr=None,
|
||
)
|
||
with state_lock:
|
||
playback_generation += 1
|
||
playback = PlaybackContext(
|
||
generation=playback_generation,
|
||
process=process,
|
||
path=final_path,
|
||
slot=slot_num,
|
||
stop_event=threading.Event(),
|
||
)
|
||
# Establish ownership before constructing any worker. If worker
|
||
# setup raises, stop_mpv can still reap this exact child.
|
||
current_playback = playback
|
||
|
||
playback.listener_thread = threading.Thread(
|
||
target=mpv_observe_properties,
|
||
args=(playback,),
|
||
name=f"mpv-telemetry-{playback.generation}",
|
||
daemon=True,
|
||
)
|
||
except Exception as exc:
|
||
log(f"[player] failed to initialize mpv playback: {exc}", error=True)
|
||
if playback is not None:
|
||
stop_mpv(
|
||
reason="startup_error",
|
||
expected=playback,
|
||
final_state="error",
|
||
message=f"failed to start mpv: {exc}",
|
||
)
|
||
elif process is not None:
|
||
# Context construction itself failed after Popen. Reap the local
|
||
# child directly because it was never safe to publish as current.
|
||
try:
|
||
process.terminate()
|
||
try:
|
||
process.wait(timeout=2)
|
||
except subprocess.TimeoutExpired:
|
||
process.kill()
|
||
process.wait(timeout=2)
|
||
except Exception:
|
||
try:
|
||
if process.poll() is None:
|
||
process.kill()
|
||
process.wait(timeout=2)
|
||
except Exception:
|
||
log_exception("[player] failed to reap unowned mpv process")
|
||
publish_status("error", message=f"failed to start mpv: {exc}")
|
||
else:
|
||
publish_status("error", message=f"failed to start mpv: {exc}")
|
||
return
|
||
|
||
publish_status("playing", playback=playback, label=label)
|
||
try:
|
||
playback.listener_thread.start()
|
||
except Exception as exc:
|
||
log_exception("[player] failed to start playback worker")
|
||
stop_mpv(
|
||
reason="worker_start_error",
|
||
expected=playback,
|
||
final_state="error",
|
||
message=f"failed to start playback worker: {exc}",
|
||
)
|
||
|
||
def mpv_ipc(command: Any, playback: Optional[PlaybackContext] = None) -> bool:
|
||
"""Send a JSON IPC command to MPV via MPV_SOCKET."""
|
||
deadline = time.monotonic() + 3.0
|
||
while time.monotonic() < deadline:
|
||
if shutdown_event.is_set():
|
||
return False
|
||
if playback is not None and not playback_is_current(playback):
|
||
return False
|
||
try:
|
||
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
|
||
sock.settimeout(0.5)
|
||
sock.connect(MPV_SOCKET)
|
||
sock.sendall((json.dumps({"command": command}) + "\n").encode("utf-8"))
|
||
return True
|
||
except (FileNotFoundError, ConnectionRefusedError, socket.timeout):
|
||
shutdown_event.wait(0.05)
|
||
except OSError as exc:
|
||
log(f"[player] mpv_ipc error: {exc}", error=True)
|
||
return False
|
||
|
||
log(f"[player] mpv_ipc timed out waiting for {MPV_SOCKET}", error=True)
|
||
return False
|
||
|
||
def fmt_mmss(value: Optional[float]) -> Optional[str]:
|
||
if (
|
||
not isinstance(value, (int, float))
|
||
or not math.isfinite(value)
|
||
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(
|
||
playback: PlaybackContext,
|
||
*,
|
||
pos: Optional[float],
|
||
dur: Optional[float],
|
||
paused: Optional[bool],
|
||
title: Optional[str],
|
||
filename: Optional[str],
|
||
playlist_pos: Optional[int],
|
||
playlist_count: Optional[int],
|
||
) -> None:
|
||
if not playback_is_current(playback):
|
||
return
|
||
|
||
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": playback.slot,
|
||
"path": playback.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(),
|
||
}
|
||
|
||
try:
|
||
client.publish(TELEMETRY_TOPIC, json.dumps(payload), qos=0, retain=False)
|
||
except Exception:
|
||
log_exception("[player] failed to publish telemetry")
|
||
|
||
def publish_cleared_telemetry(playback: Optional[PlaybackContext] = None) -> None:
|
||
payload: Dict[str, Any] = {
|
||
"state": "telemetry",
|
||
"slot": playback.slot if playback is not None else None,
|
||
"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(),
|
||
}
|
||
|
||
try:
|
||
client.publish(TELEMETRY_TOPIC, json.dumps(payload), qos=0, retain=False)
|
||
except Exception:
|
||
log_exception("[player] failed to publish cleared telemetry")
|
||
|
||
def mpv_observe_properties(playback: PlaybackContext) -> None:
|
||
|
||
"""
|
||
Keep a persistent IPC connection to mpv and observe properties.
|
||
Publishes telemetry to MQTT at TELEMETRY_INTERVAL cadence.
|
||
"""
|
||
# 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
|
||
recv_buffer = b""
|
||
sock: Optional[socket.socket] = None
|
||
|
||
def publish_cached() -> None:
|
||
nonlocal last_pub
|
||
if not playback_is_current(playback):
|
||
return
|
||
publish_telemetry(
|
||
playback,
|
||
pos=pos,
|
||
dur=dur,
|
||
paused=paused,
|
||
title=title,
|
||
filename=filename,
|
||
playlist_pos=playlist_pos,
|
||
playlist_count=playlist_count,
|
||
)
|
||
last_pub = time.monotonic()
|
||
|
||
try:
|
||
deadline = time.monotonic() + 3.0
|
||
while playback_is_current(playback):
|
||
candidate = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
||
try:
|
||
candidate.connect(MPV_SOCKET)
|
||
sock = candidate
|
||
break
|
||
except (FileNotFoundError, ConnectionRefusedError):
|
||
candidate.close()
|
||
if time.monotonic() > deadline:
|
||
raise
|
||
playback.stop_event.wait(0.05)
|
||
else:
|
||
return
|
||
|
||
if sock is None:
|
||
return
|
||
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
|
||
log(f"[player] telemetry connected to mpv socket {MPV_SOCKET}")
|
||
publish_cached()
|
||
|
||
while playback_is_current(playback):
|
||
try:
|
||
chunk = sock.recv(65536)
|
||
if not chunk:
|
||
break
|
||
|
||
recv_buffer += chunk
|
||
lines = recv_buffer.split(b"\n")
|
||
recv_buffer = lines.pop()
|
||
for raw_line in lines:
|
||
if not raw_line.strip():
|
||
continue
|
||
try:
|
||
evt = json.loads(raw_line.decode("utf-8", errors="replace"))
|
||
except (UnicodeDecodeError, json.JSONDecodeError):
|
||
continue
|
||
|
||
if not isinstance(evt, dict):
|
||
continue
|
||
|
||
if evt.get("event") != "property-change":
|
||
continue
|
||
|
||
name = evt.get("name")
|
||
val = evt.get("data")
|
||
finite_number = (
|
||
isinstance(val, (int, float))
|
||
and not isinstance(val, bool)
|
||
and math.isfinite(val)
|
||
)
|
||
|
||
if name == "time-pos":
|
||
pos = float(val) if finite_number else None
|
||
elif name == "duration":
|
||
dur = float(val) if finite_number 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 finite_number else None
|
||
elif name == "playlist-count":
|
||
playlist_count = int(val) if finite_number else None
|
||
|
||
now = time.monotonic()
|
||
if now - last_pub >= TELEMETRY_INTERVAL:
|
||
publish_cached()
|
||
|
||
except socket.timeout:
|
||
# still periodically publish even if no new events
|
||
now = time.monotonic()
|
||
if now - last_pub >= TELEMETRY_INTERVAL:
|
||
publish_cached()
|
||
continue
|
||
|
||
except Exception as exc:
|
||
if playback_is_current(playback):
|
||
log(f"[player] mpv_observe_properties error: {exc}", error=True)
|
||
finally:
|
||
if sock is not None:
|
||
try:
|
||
sock.close()
|
||
except OSError:
|
||
pass
|
||
|
||
# ----------------------------
|
||
# Command dispatch
|
||
# ----------------------------
|
||
|
||
def decode_command(payload_str: str) -> Optional[Dict[str, Any]]:
|
||
"""Decode and validate the outer command envelope without raising."""
|
||
try:
|
||
data = json.loads(payload_str or "{}")
|
||
except (TypeError, ValueError, RecursionError):
|
||
log(f"[player] bad JSON in cmd: {payload_str!r}", error=True)
|
||
return None
|
||
|
||
if not isinstance(data, dict):
|
||
log("[player] command payload must be a JSON object", error=True)
|
||
return None
|
||
|
||
action = data.get("action")
|
||
if not isinstance(action, str) or not action.strip():
|
||
log("[player] cmd missing string 'action'", error=True)
|
||
return None
|
||
return data
|
||
|
||
|
||
def enqueue_command(
|
||
data: Dict[str, Any],
|
||
*,
|
||
message_id: Optional[int] = None,
|
||
qos: int = 0,
|
||
connection_generation: Optional[int] = None,
|
||
) -> bool:
|
||
"""Hand work to the main thread with bounded, lossless backpressure."""
|
||
queued = QueuedCommand(
|
||
data=data,
|
||
message_id=message_id,
|
||
qos=qos,
|
||
connection_generation=connection_generation,
|
||
)
|
||
while not shutdown_event.is_set():
|
||
try:
|
||
command_queue.put(queued, timeout=0.2)
|
||
return True
|
||
except queue.Full:
|
||
continue
|
||
return False
|
||
|
||
|
||
def acknowledge_command(
|
||
message_id: Optional[int],
|
||
qos: int,
|
||
connection_generation: Optional[int] = None,
|
||
) -> None:
|
||
if message_id is None or qos <= 0:
|
||
return
|
||
|
||
if connection_generation is not None:
|
||
with mqtt_generation_lock:
|
||
if connection_generation != mqtt_connection_generation:
|
||
log(
|
||
f"[player] skipping stale MQTT ack mid={message_id} from "
|
||
f"connection {connection_generation}",
|
||
error=True,
|
||
)
|
||
return
|
||
|
||
try:
|
||
ack_rc = client.ack(message_id, qos)
|
||
if ack_rc != mqtt.MQTT_ERR_SUCCESS:
|
||
log(
|
||
f"[player] MQTT ack failed mid={message_id} rc={ack_rc}",
|
||
error=True,
|
||
)
|
||
except Exception:
|
||
log_exception(f"[player] MQTT ack failed mid={message_id}")
|
||
|
||
|
||
def handle_cmd(command: Any) -> None:
|
||
"""Route one validated command on the main player thread."""
|
||
if isinstance(command, str):
|
||
data = decode_command(command)
|
||
if data is None:
|
||
return
|
||
elif isinstance(command, dict):
|
||
data = command
|
||
else:
|
||
log("[player] internal command was not an object", error=True)
|
||
return
|
||
|
||
action = data.get("action")
|
||
if not isinstance(action, str) or not action:
|
||
log("[player] cmd missing string 'action'", error=True)
|
||
return
|
||
|
||
log(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":
|
||
log("[player] STOP command")
|
||
stop_mpv("stop")
|
||
return
|
||
|
||
playback = get_current_playback()
|
||
|
||
if action == "pause":
|
||
if playback is not None and playback.process.poll() is None:
|
||
mpv_ipc(["set_property", "pause", True], playback)
|
||
publish_status("paused", playback=playback)
|
||
else:
|
||
publish_status("error", message="nothing playing to pause")
|
||
return
|
||
|
||
if action == "resume":
|
||
if playback is not None and playback.process.poll() is None:
|
||
mpv_ipc(["set_property", "pause", False], playback)
|
||
publish_status("playing", playback=playback)
|
||
else:
|
||
publish_status("error", message="nothing playing to resume")
|
||
return
|
||
|
||
if action == "toggle":
|
||
if playback is not None and playback.process.poll() is None:
|
||
mpv_ipc(["cycle", "pause"], playback)
|
||
publish_status("toggled", playback=playback)
|
||
else:
|
||
publish_status("error", message="nothing playing to toggle")
|
||
return
|
||
|
||
log(f"[player] unknown action: {action!r}", error=True)
|
||
|
||
|
||
# ----------------------------
|
||
# MQTT callbacks
|
||
# ----------------------------
|
||
|
||
def on_connect(mqtt_client, userdata, flags, rc, properties=None) -> None:
|
||
global mqtt_connection_generation
|
||
try:
|
||
if rc != 0:
|
||
log(f"[player] MQTT connect failed with code {rc}", error=True)
|
||
return
|
||
|
||
with mqtt_generation_lock:
|
||
mqtt_connection_generation += 1
|
||
|
||
log(f"[player] MQTT connected rc={rc}, subscribing {TOPIC_CMD}")
|
||
mqtt_client.subscribe(TOPIC_CMD, qos=1)
|
||
|
||
publish_status("idle", message="player online")
|
||
|
||
if HA_DISCOVERY_ENABLED:
|
||
try:
|
||
if publish_discovery is None:
|
||
raise RuntimeError("Home Assistant discovery module is unavailable")
|
||
publish_discovery(
|
||
mqtt_client,
|
||
discovery_prefix=HA_DISCOVERY_PREFIX,
|
||
device_id=HA_DEVICE_ID,
|
||
device_name=HA_DEVICE_NAME,
|
||
telemetry_topic=TELEMETRY_TOPIC,
|
||
status_topic=TOPIC_STATUS,
|
||
)
|
||
except Exception:
|
||
# Discovery is optional and must never disable core MQTT
|
||
# command handling.
|
||
log_exception("[player] Home Assistant discovery publish failed")
|
||
except Exception:
|
||
log_exception("[player] unexpected error in MQTT on_connect callback")
|
||
|
||
|
||
def on_connect_fail(mqtt_client, userdata) -> None:
|
||
log(
|
||
f"[player] MQTT connection to {MQTT_HOST}:{MQTT_PORT} failed; will retry",
|
||
error=True,
|
||
)
|
||
|
||
|
||
def on_disconnect(
|
||
mqtt_client,
|
||
userdata,
|
||
disconnect_flags,
|
||
reason_code,
|
||
properties=None,
|
||
) -> None:
|
||
if shutdown_event.is_set():
|
||
log(f"[player] MQTT disconnected rc={reason_code}")
|
||
else:
|
||
log(f"[player] MQTT disconnected rc={reason_code}; will retry", error=True)
|
||
|
||
|
||
def on_message(mqtt_client, userdata, msg) -> None:
|
||
try:
|
||
if msg.topic != TOPIC_CMD:
|
||
return
|
||
payload = msg.payload.decode("utf-8", errors="replace")
|
||
data = decode_command(payload)
|
||
message_id = getattr(msg, "mid", None)
|
||
qos = int(getattr(msg, "qos", 0))
|
||
with mqtt_generation_lock:
|
||
connection_generation = mqtt_connection_generation
|
||
if data is None:
|
||
# Validation completed synchronously, matching the old callback's
|
||
# acknowledgement point for rejected/poison messages.
|
||
acknowledge_command(message_id, qos, connection_generation)
|
||
return
|
||
|
||
enqueue_command(
|
||
data,
|
||
message_id=message_id,
|
||
qos=qos,
|
||
connection_generation=connection_generation,
|
||
)
|
||
except Exception:
|
||
# Paho otherwise terminates its network thread and leaves the process
|
||
# looking healthy to systemd while it is permanently deaf to commands.
|
||
log_exception("[player] unexpected error in MQTT on_message callback")
|
||
|
||
|
||
# ----------------------------
|
||
# Signal handling
|
||
# ----------------------------
|
||
|
||
shutdown_reason = "shutdown"
|
||
|
||
|
||
def sigterm_handler(signum, frame) -> None:
|
||
global shutdown_reason
|
||
shutdown_reason = "sigterm" if signum == signal.SIGTERM else "sigint"
|
||
log(f"[player] {signal.Signals(signum).name} received, shutting down")
|
||
shutdown_event.set()
|
||
|
||
|
||
# ----------------------------
|
||
# Main loop
|
||
# ----------------------------
|
||
|
||
def poll_current_playback() -> None:
|
||
playback = get_current_playback()
|
||
if playback is None:
|
||
return
|
||
|
||
ret = playback.process.poll()
|
||
if ret is None:
|
||
return
|
||
if ret == 0:
|
||
log("[player] mpv finished normally")
|
||
stop_mpv(reason="finished", expected=playback)
|
||
else:
|
||
log(f"[player] mpv exited with code {ret}", error=True)
|
||
publish_status("error", playback=playback, message=f"mpv exited code {ret}")
|
||
stop_mpv(reason="mpv_error", expected=playback)
|
||
|
||
|
||
def process_pending_commands() -> None:
|
||
for _ in range(COMMANDS_PER_TICK):
|
||
if shutdown_event.is_set():
|
||
return
|
||
try:
|
||
queued = command_queue.get_nowait()
|
||
except queue.Empty:
|
||
return
|
||
|
||
if isinstance(queued, QueuedCommand):
|
||
command = queued.data
|
||
message_id = queued.message_id
|
||
qos = queued.qos
|
||
connection_generation = queued.connection_generation
|
||
else:
|
||
# Backward-compatible for in-process callers/tests that populated
|
||
# the queue before QueuedCommand was introduced.
|
||
command = queued
|
||
message_id = None
|
||
qos = 0
|
||
connection_generation = None
|
||
|
||
try:
|
||
handle_cmd(command)
|
||
except Exception:
|
||
# A bad command must not terminate the daemon's main loop.
|
||
log_exception(f"[player] unexpected command failure: {command!r}")
|
||
finally:
|
||
# Match the old QoS 1 durability boundary: acknowledge only after
|
||
# main-thread execution, not merely after callback enqueueing.
|
||
acknowledge_command(message_id, qos, connection_generation)
|
||
|
||
|
||
def clear_expired_busy_status() -> None:
|
||
global busy_until
|
||
playback = get_current_playback()
|
||
if (
|
||
busy_until is not None
|
||
and time.monotonic() >= busy_until
|
||
and playback is not None
|
||
and playback.process.poll() is None
|
||
):
|
||
log("[player] clearing busy -> playing")
|
||
publish_status("playing", playback=playback)
|
||
busy_until = None
|
||
|
||
|
||
def configure_mqtt_client() -> None:
|
||
if MQTT_USER:
|
||
client.username_pw_set(MQTT_USER, MQTT_PASS or None)
|
||
|
||
client.on_connect = on_connect
|
||
client.on_connect_fail = on_connect_fail
|
||
client.on_disconnect = on_disconnect
|
||
client.on_message = on_message
|
||
client.suppress_exceptions = True
|
||
client.manual_ack_set(True)
|
||
client.reconnect_delay_set(min_delay=1, max_delay=30)
|
||
|
||
client.will_set(
|
||
TOPIC_STATUS,
|
||
json.dumps({"state": "offline"}),
|
||
qos=1,
|
||
retain=False,
|
||
)
|
||
|
||
|
||
def main() -> None:
|
||
global shutdown_reason
|
||
|
||
signal.signal(signal.SIGTERM, sigterm_handler)
|
||
signal.signal(signal.SIGINT, sigterm_handler)
|
||
shutdown_reason = "shutdown"
|
||
shutdown_event.clear()
|
||
configure_mqtt_client()
|
||
|
||
log(f"[player] connecting MQTT to {MQTT_HOST}:{MQTT_PORT}")
|
||
client.connect_async(MQTT_HOST, MQTT_PORT, keepalive=30)
|
||
loop_rc = client.loop_start()
|
||
if loop_rc != mqtt.MQTT_ERR_SUCCESS:
|
||
raise RuntimeError(f"failed to start MQTT network loop: {loop_rc}")
|
||
|
||
try:
|
||
while not shutdown_event.is_set():
|
||
poll_current_playback()
|
||
process_pending_commands()
|
||
clear_expired_busy_status()
|
||
shutdown_event.wait(0.05)
|
||
finally:
|
||
if shutdown_reason in ("sigterm", "sigint"):
|
||
# The historical signal handler published its terminal player
|
||
# state while MQTT was still running.
|
||
stop_mpv(reason=shutdown_reason)
|
||
client.loop_stop()
|
||
else:
|
||
client.loop_stop()
|
||
stop_mpv(reason=shutdown_reason)
|
||
# Do not send MQTT DISCONNECT: the broker must emit the configured
|
||
# offline LWT when the service process exits, as it did previously.
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|