initial commmit
A video player controller for the video player node
This commit is contained in:
Media
@@ -0,0 +1,838 @@
|
||||
import os
|
||||
import time
|
||||
import json
|
||||
import threading
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from dataclasses import dataclass, asdict
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
import yaml
|
||||
import paho.mqtt.client as mqtt
|
||||
|
||||
# --- Config / paths ---
|
||||
|
||||
MEDIA_BASE = Path(os.environ.get("YTPLAYER_MEDIA_BASE", "/mnt/Media/YTPlayer"))
|
||||
DATA_BASE = Path(os.environ.get("YTPLAYER_DATA_BASE", "/controller-data"))
|
||||
STATE_PATH = DATA_BASE / "playlists.yaml"
|
||||
|
||||
MAX_SLOTS = int(os.environ.get("YTPLAYER_MAX_SLOTS", "10"))
|
||||
RETENTION_SECONDS = int(os.environ.get("YTPLAYER_RETENTION_SECONDS", str(48 * 3600)))
|
||||
|
||||
MQTT_HOST = os.environ.get("MQTT_HOST", "10.40.0.2")
|
||||
MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883"))
|
||||
MQTT_CLIENT_ID = os.environ.get("MQTT_CLIENT_ID", "yt-controller")
|
||||
|
||||
TOPIC_INGEST_PROGRESS = "youtube/ingest/progress"
|
||||
TOPIC_PLAYBACK_BASE = "youtube/media/playback" # e.g. youtube/media/playback/1/title
|
||||
TOPIC_PLAYER_STATUS = "youtube/player/status" # JSON from player_service
|
||||
TOPIC_PLAYER_CMD = "youtube/player/cmd" # commands to player_service
|
||||
TOPIC_CONTROLLER_CMD = "youtube/controller/cmd" # commands from HA / StreamDeck
|
||||
|
||||
MEDIA_BASE.mkdir(parents=True, exist_ok=True)
|
||||
DATA_BASE.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# --- MQTT client ---
|
||||
|
||||
import logging, sys
|
||||
logging.basicConfig(stream=sys.stdout, level=logging.INFO)
|
||||
|
||||
def _on_connect(client, userdata, flags, rc, properties=None):
|
||||
print(f"[mqtt] connected rc={rc}")
|
||||
if rc == 0:
|
||||
client.subscribe(TOPIC_PLAYER_STATUS, qos=1)
|
||||
client.subscribe(TOPIC_CONTROLLER_CMD, qos=1)
|
||||
print(f"[controller] subscribed to {TOPIC_PLAYER_STATUS} and {TOPIC_CONTROLLER_CMD}")
|
||||
else:
|
||||
print(f"[mqtt] connect failed rc={rc}")
|
||||
|
||||
def _on_message(client, userdata, msg):
|
||||
if msg.topic == TOPIC_PLAYER_STATUS:
|
||||
try:
|
||||
data = json.loads(msg.payload.decode("utf-8") or "{}")
|
||||
except Exception:
|
||||
return
|
||||
|
||||
state_str = data.get("state")
|
||||
slot_val = data.get("slot")
|
||||
|
||||
if not state_str or slot_val is None:
|
||||
return
|
||||
|
||||
try:
|
||||
slot_num = int(slot_val)
|
||||
except (TypeError, ValueError):
|
||||
return
|
||||
|
||||
print(f"[controller] player status: slot={slot_num} state={state_str}")
|
||||
_update_slot_state_from_player(slot_num, state_str)
|
||||
|
||||
elif msg.topic == TOPIC_CONTROLLER_CMD:
|
||||
_handle_control_cmd(msg.payload.decode("utf-8", errors="ignore"))
|
||||
|
||||
def _handle_control_cmd(payload: str):
|
||||
"""
|
||||
Handle messages on TOPIC_CONTROLLER_CMD, e.g.:
|
||||
{"action": "delete_slot", "slot": 2, "url": "..."}
|
||||
{"action": "play", "slot": 1}
|
||||
{"action": "stop"}
|
||||
"""
|
||||
try:
|
||||
data = json.loads(payload or "{}")
|
||||
except json.JSONDecodeError:
|
||||
print(f"[controller] bad JSON on control cmd: {payload!r}")
|
||||
return
|
||||
|
||||
action = data.get("action")
|
||||
if not action:
|
||||
print("[controller] control cmd missing 'action'")
|
||||
return
|
||||
|
||||
# delete slot
|
||||
if action == "delete_slot":
|
||||
slot = data.get("slot")
|
||||
try:
|
||||
slot_num = int(slot)
|
||||
except (TypeError, ValueError):
|
||||
print(f"[controller] delete_slot: bad slot {slot!r}")
|
||||
return
|
||||
|
||||
expected_url = data.get("url") # optional safety
|
||||
delete_slot(slot_num, expected_url=expected_url)
|
||||
return
|
||||
|
||||
# play / stop → hand off to the orchestrator
|
||||
if action in ("play", "stop"):
|
||||
handle_play_request(data)
|
||||
return
|
||||
|
||||
print(f"[controller] unknown control action: {action!r}")
|
||||
|
||||
def _on_disconnect(client, userdata, rc, properties=None):
|
||||
print(f"[mqtt] disconnected rc={rc}")
|
||||
|
||||
mqtt_client = mqtt.Client(client_id=MQTT_CLIENT_ID, clean_session=True)
|
||||
|
||||
# If your broker ever needs auth, you'd uncomment:
|
||||
# mqtt_client.username_pw_set(os.environ.get("MQTT_USERNAME"), os.environ.get("MQTT_PASSWORD"))
|
||||
|
||||
mqtt_client.enable_logger() # send Paho logs to Python logging
|
||||
mqtt_client.on_connect = _on_connect
|
||||
mqtt_client.on_disconnect = _on_disconnect
|
||||
mqtt_client.on_message = _on_message
|
||||
|
||||
print(f"[controller] attempting MQTT connect to {MQTT_HOST}:{MQTT_PORT} …")
|
||||
mqtt_client.connect(MQTT_HOST, MQTT_PORT, keepalive=30)
|
||||
|
||||
mqtt_client.loop_start()
|
||||
print("[controller] MQTT loop started")
|
||||
|
||||
mqtt_client.on_message = _on_message
|
||||
mqtt_client.subscribe(TOPIC_PLAYER_STATUS, qos=1)
|
||||
mqtt_client.subscribe(TOPIC_CONTROLLER_CMD, qos=1)
|
||||
print(f"[controller] subscribed to {TOPIC_PLAYER_STATUS} and {TOPIC_CONTROLLER_CMD}")
|
||||
|
||||
|
||||
def mqtt_publish(topic: str, payload, **kwargs):
|
||||
"""
|
||||
Small wrapper that JSON-encodes non-strings and prints what we publish.
|
||||
Keeps logs useful and avoids repeating code.
|
||||
"""
|
||||
if not isinstance(payload, str):
|
||||
payload = json.dumps(payload)
|
||||
print(f"[mqtt] publish {topic} -> {payload}")
|
||||
mqtt_client.publish(topic, payload, **kwargs)
|
||||
|
||||
|
||||
# --- State model ---
|
||||
|
||||
@dataclass
|
||||
class PlaylistSlot:
|
||||
slot: int
|
||||
url: str
|
||||
yt_id: Optional[str]
|
||||
title: str
|
||||
path: str
|
||||
m3u: str
|
||||
created_at: float
|
||||
expires_at: float
|
||||
item_count: int
|
||||
state: str # "downloading" | "ready" | "error" | "playing" | "busy"
|
||||
|
||||
def to_dict(self):
|
||||
d = asdict(self)
|
||||
# timestamps as ISO-ish strings
|
||||
d["created_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(self.created_at))
|
||||
d["expires_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(self.expires_at))
|
||||
return d
|
||||
|
||||
@staticmethod
|
||||
def from_dict(d: Dict):
|
||||
def parse_ts(s: str) -> float:
|
||||
try:
|
||||
# naive ISO Z
|
||||
return time.mktime(time.strptime(s, "%Y-%m-%dT%H:%M:%SZ"))
|
||||
except Exception:
|
||||
return float(d.get("created_at_ts", time.time()))
|
||||
|
||||
created_at = parse_ts(d.get("created_at", "1970-01-01T00:00:00Z"))
|
||||
expires_at = parse_ts(d.get("expires_at", "1970-01-01T00:00:00Z"))
|
||||
|
||||
return PlaylistSlot(
|
||||
slot=int(d["slot"]),
|
||||
url=d["url"],
|
||||
yt_id=d.get("yt_id"),
|
||||
title=d.get("title", ""),
|
||||
path=d["path"],
|
||||
m3u=d["m3u"],
|
||||
created_at=created_at,
|
||||
expires_at=expires_at,
|
||||
item_count=int(d.get("item_count", 0)),
|
||||
state=d.get("state", "ready"),
|
||||
)
|
||||
|
||||
|
||||
class ControllerState:
|
||||
def __init__(self):
|
||||
self.slots: Dict[int, PlaylistSlot] = {}
|
||||
|
||||
def load(self):
|
||||
if not STATE_PATH.exists():
|
||||
return
|
||||
data = yaml.safe_load(STATE_PATH.read_text()) or {}
|
||||
slots = data.get("slots", {})
|
||||
self.slots.clear()
|
||||
for k, v in slots.items():
|
||||
try:
|
||||
slot = PlaylistSlot.from_dict(v)
|
||||
self.slots[slot.slot] = slot
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
def save(self):
|
||||
data = {"slots": {str(s): slot.to_dict() for s, slot in self.slots.items()}}
|
||||
STATE_PATH.write_text(yaml.safe_dump(data, sort_keys=True))
|
||||
|
||||
def prune_expired(self):
|
||||
"""
|
||||
Remove any slots whose expires_at is in the past, delete their folders,
|
||||
compact remaining slots, save, and reannounce MQTT so ghosts disappear.
|
||||
"""
|
||||
now = time.time()
|
||||
changed = False
|
||||
to_delete: List[int] = []
|
||||
|
||||
for s, slot in list(self.slots.items()):
|
||||
if now >= slot.expires_at:
|
||||
to_delete.append(s)
|
||||
|
||||
for s in to_delete:
|
||||
slot = self.slots.pop(s, None)
|
||||
if not slot:
|
||||
continue
|
||||
|
||||
try:
|
||||
p = Path(slot.path)
|
||||
if p.exists():
|
||||
for child in p.iterdir():
|
||||
try:
|
||||
child.unlink()
|
||||
except IsADirectoryError:
|
||||
# ignore unexpected subdirs
|
||||
pass
|
||||
p.rmdir()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
changed = True
|
||||
|
||||
if changed:
|
||||
# compact slot numbers (1..N), save, then resync MQTT topics
|
||||
self.compact_slots()
|
||||
self.save()
|
||||
# this is the helper you already have below
|
||||
reannounce_slots()
|
||||
|
||||
def compact_slots(self):
|
||||
"""Shift slots down so there are no gaps: 1..N"""
|
||||
if not self.slots:
|
||||
return
|
||||
new_slots: Dict[int, PlaylistSlot] = {}
|
||||
for new_index, slot in enumerate(sorted(self.slots.values(), key=lambda sl: sl.slot), start=1):
|
||||
if slot.slot != new_index:
|
||||
old_dir = Path(slot.path)
|
||||
new_dir = MEDIA_BASE / f"slot{new_index}"
|
||||
# rename folder if it exists
|
||||
if old_dir.exists() and old_dir != new_dir:
|
||||
new_dir.parent.mkdir(parents=True, exist_ok=True)
|
||||
old_dir.rename(new_dir)
|
||||
slot.slot = new_index
|
||||
slot.path = str(new_dir)
|
||||
slot.m3u = str(new_dir / "playlist.m3u")
|
||||
new_slots[new_index] = slot
|
||||
self.slots = new_slots
|
||||
|
||||
def get_next_free_slot(self) -> Optional[int]:
|
||||
self.prune_expired()
|
||||
used = set(self.slots.keys())
|
||||
for i in range(1, MAX_SLOTS + 1):
|
||||
if i not in used:
|
||||
return i
|
||||
return None
|
||||
|
||||
|
||||
state = ControllerState()
|
||||
state.load()
|
||||
state.prune_expired()
|
||||
|
||||
|
||||
# --- Helpers ---
|
||||
|
||||
def delete_slot(slot_num: int, expected_url: Optional[str] = None):
|
||||
"""
|
||||
Delete a slot from state + filesystem, then compact and republish.
|
||||
|
||||
expected_url is optional: if provided and doesn't match, we ignore
|
||||
the delete (safety against stale retained messages if someone ever
|
||||
accidentally turns retain on).
|
||||
"""
|
||||
state.load()
|
||||
|
||||
slot = state.slots.get(slot_num)
|
||||
if not slot:
|
||||
print(f"[controller] delete_slot: slot {slot_num} not found, ignoring")
|
||||
return
|
||||
|
||||
if expected_url and slot.url != expected_url:
|
||||
print(f"[controller] delete_slot: url mismatch for slot {slot_num}, ignoring")
|
||||
return
|
||||
|
||||
print(f"[controller] deleting slot {slot_num}: {slot.title!r}")
|
||||
|
||||
# Remove files on disk
|
||||
slot_dir = Path(slot.path)
|
||||
if slot_dir.exists():
|
||||
for child in slot_dir.iterdir():
|
||||
try:
|
||||
child.unlink()
|
||||
except IsADirectoryError:
|
||||
# no subdirs expected, but don't crash
|
||||
pass
|
||||
try:
|
||||
slot_dir.rmdir()
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
# Remove from state
|
||||
state.slots.pop(slot_num, None)
|
||||
|
||||
# Compact and save
|
||||
state.compact_slots()
|
||||
state.save()
|
||||
|
||||
# Re-publish MQTT for all remaining slots, so HA/Deck are in sync
|
||||
reannounce_slots()
|
||||
|
||||
|
||||
def reannounce_slots():
|
||||
"""
|
||||
Re-publish MQTT state/title/count for all existing slots,
|
||||
and mark any unused slots as empty.
|
||||
"""
|
||||
state.load()
|
||||
|
||||
# Re-announce real slots
|
||||
for s, slot in sorted(state.slots.items()):
|
||||
publish_slot_topics(slot)
|
||||
|
||||
# Clear any slots above the current highest used index
|
||||
highest = max(state.slots.keys(), default=0)
|
||||
for s in range(highest + 1, MAX_SLOTS + 1):
|
||||
base = f"{TOPIC_PLAYBACK_BASE}/{s}"
|
||||
mqtt_publish(f"{base}/title", "", retain=True)
|
||||
mqtt_publish(f"{base}/state", "empty", retain=True)
|
||||
mqtt_publish(f"{base}/count", "0", retain=True)
|
||||
# also clear item names
|
||||
for i in range(1, 100):
|
||||
mqtt_publish(f"{base}/items/{i}", "", retain=True)
|
||||
|
||||
|
||||
def publish_slot_topics(slot: PlaylistSlot):
|
||||
"""Publish the title/state/count for one slot to its MQTT topics."""
|
||||
base = f"{TOPIC_PLAYBACK_BASE}/{slot.slot}"
|
||||
mqtt_publish(f"{base}/title", slot.title, retain=True)
|
||||
mqtt_publish(f"{base}/state", slot.state, retain=True)
|
||||
mqtt_publish(f"{base}/count", slot.item_count, retain=True)
|
||||
|
||||
def set_slot_busy_temporarily(slot: PlaylistSlot, seconds: int = 5):
|
||||
"""
|
||||
Briefly mark a slot as 'busy', then restore after a delay.
|
||||
"""
|
||||
base = f"{TOPIC_PLAYBACK_BASE}/{slot.slot}"
|
||||
mqtt_publish(f"{base}/state", "busy", retain=True)
|
||||
|
||||
def _clear_busy_later(slot_number: int, delay: int):
|
||||
time.sleep(delay)
|
||||
|
||||
state.load()
|
||||
sl = state.slots.get(slot_number)
|
||||
|
||||
if sl is None:
|
||||
base2 = f"{TOPIC_PLAYBACK_BASE}/{slot_number}"
|
||||
mqtt_publish(f"{base2}/title", "", retain=True)
|
||||
mqtt_publish(f"{base2}/state", "empty", retain=True)
|
||||
mqtt_publish(f"{base2}/count", 0, retain=True)
|
||||
return
|
||||
|
||||
publish_slot_topics(sl)
|
||||
|
||||
t = threading.Thread(
|
||||
target=_clear_busy_later,
|
||||
args=(slot.slot, seconds),
|
||||
daemon=True,
|
||||
)
|
||||
t.start()
|
||||
|
||||
|
||||
|
||||
def _parse_track_title_from_filename(path: Path) -> str:
|
||||
stem = path.stem # "000 - Song Name [id]"
|
||||
# strip leading "000 - "
|
||||
if " - " in stem:
|
||||
stem = stem.split(" - ", 1)[1]
|
||||
# strip trailing " [id]"
|
||||
if " [" in stem:
|
||||
stem = stem.rsplit(" [", 1)[0]
|
||||
return stem
|
||||
|
||||
|
||||
def _update_slot_state_from_player(slot_num: int, new_state: str):
|
||||
"""
|
||||
Mirror player state ("playing", "busy", "idle", "error", etc.)
|
||||
into our per-slot state + MQTT topics.
|
||||
"""
|
||||
slot_obj = state.slots.get(slot_num)
|
||||
if not slot_obj:
|
||||
return
|
||||
|
||||
# Map "idle" / "finished" from player back to "ready" for the slot UI
|
||||
if new_state in ("idle", "finished"):
|
||||
slot_obj.state = "ready"
|
||||
mqtt_state = "ready"
|
||||
else:
|
||||
slot_obj.state = new_state
|
||||
mqtt_state = new_state
|
||||
|
||||
state.save()
|
||||
publish_slot_topics(slot_obj)
|
||||
|
||||
|
||||
def _get_title_for_url(url: str) -> str:
|
||||
"""
|
||||
Try to detect a human-friendly title for either:
|
||||
- a playlist (use playlist_title)
|
||||
- a single video (fall back to video title)
|
||||
|
||||
If we can't figure it out, just return the URL.
|
||||
"""
|
||||
try:
|
||||
# One line: playlist_title|||video_title
|
||||
out = subprocess.check_output(
|
||||
[
|
||||
"yt-dlp",
|
||||
"--flat-playlist",
|
||||
"--print", "%(playlist_title)s|||%(title)s",
|
||||
"--playlist-end", "1",
|
||||
url,
|
||||
],
|
||||
stderr=subprocess.DEVNULL,
|
||||
text=True,
|
||||
).strip()
|
||||
|
||||
if not out:
|
||||
return url
|
||||
|
||||
line = out.splitlines()[0]
|
||||
playlist_title, _, video_title = line.partition("|||")
|
||||
|
||||
playlist_title = (playlist_title or "").strip()
|
||||
video_title = (video_title or "").strip()
|
||||
|
||||
# If playlist_title looks real (not empty/NA), use it
|
||||
if playlist_title and playlist_title.upper() != "NA":
|
||||
return playlist_title
|
||||
|
||||
# Otherwise fall back to video title
|
||||
if video_title:
|
||||
return video_title
|
||||
|
||||
return url
|
||||
except Exception:
|
||||
return url
|
||||
|
||||
|
||||
def _download_playlist_to_slot(url: str, slot_num: int):
|
||||
"""
|
||||
Runs yt-dlp into /mnt/Media/YTPlayer/slotN, builds playlist.m3u,
|
||||
updates YAML, and publishes MQTT topics.
|
||||
"""
|
||||
slot_dir = MEDIA_BASE / f"slot{slot_num}"
|
||||
slot_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
title = _get_title_for_url(url)
|
||||
now = time.time()
|
||||
expires_at = now + RETENTION_SECONDS
|
||||
|
||||
# mark slot as downloading
|
||||
slot = PlaylistSlot(
|
||||
slot=slot_num,
|
||||
url=url,
|
||||
yt_id=None,
|
||||
title=title,
|
||||
path=str(slot_dir),
|
||||
m3u=str(slot_dir / "playlist.m3u"),
|
||||
created_at=now,
|
||||
expires_at=expires_at,
|
||||
item_count=0,
|
||||
state="downloading",
|
||||
)
|
||||
state.slots[slot_num] = slot
|
||||
state.save()
|
||||
|
||||
# Tell HA/Deck: this slot is downloading now
|
||||
publish_slot_topics(slot)
|
||||
|
||||
mqtt_publish(
|
||||
TOPIC_INGEST_PROGRESS,
|
||||
{"slot": slot_num, "state": "downloading", "title": title},
|
||||
)
|
||||
|
||||
# run yt-dlp
|
||||
try:
|
||||
subprocess.run(
|
||||
[
|
||||
"yt-dlp",
|
||||
"--cookies",
|
||||
"/cookies.txt", # inside the container (mount your exported cookies here)
|
||||
"-o",
|
||||
str(slot_dir / "%(playlist_index)03d - %(title)s [%(id)s].%(ext)s"),
|
||||
"--yes-playlist",
|
||||
"--no-overwrites", # keep existing files, only fetch new ones
|
||||
"-f",
|
||||
"bv*[height<=1080]+ba/b[height<=1080]/b[height<=1080]/b",
|
||||
url,
|
||||
],
|
||||
check=True,
|
||||
)
|
||||
except subprocess.CalledProcessError as e:
|
||||
slot.state = "error"
|
||||
state.save()
|
||||
publish_slot_topics(slot)
|
||||
mqtt_publish(
|
||||
TOPIC_INGEST_PROGRESS,
|
||||
{"slot": slot_num, "state": "error", "message": f"yt-dlp failed: {e}"},
|
||||
)
|
||||
return
|
||||
|
||||
# --- build .m3u in sorted order ---
|
||||
media_files = sorted(
|
||||
[
|
||||
p
|
||||
for p in slot_dir.iterdir()
|
||||
if p.is_file() and p.suffix.lower() in {".mp4", ".mkv", ".webm"}
|
||||
]
|
||||
)
|
||||
|
||||
if not media_files:
|
||||
slot.state = "error"
|
||||
state.save()
|
||||
publish_slot_topics(slot)
|
||||
mqtt_publish(
|
||||
TOPIC_INGEST_PROGRESS,
|
||||
{"slot": slot_num, "state": "error", "message": "no media files downloaded"},
|
||||
)
|
||||
return
|
||||
|
||||
item_titles: List[str] = []
|
||||
m3u_path = slot_dir / "playlist.m3u"
|
||||
with m3u_path.open("w", encoding="utf-8") as f:
|
||||
for p in media_files:
|
||||
# Use a path relative to the slot directory so that if the
|
||||
# whole slotN folder gets renamed (compaction), the playlist
|
||||
# still works.
|
||||
rel = os.path.relpath(p, slot_dir)
|
||||
f.write(rel + "\n")
|
||||
item_titles.append(_parse_track_title_from_filename(p))
|
||||
|
||||
|
||||
slot.item_count = len(media_files)
|
||||
slot.state = "ready"
|
||||
state.slots[slot_num] = slot
|
||||
state.save()
|
||||
|
||||
# publish MQTT info for HA + StreamDeck
|
||||
publish_slot_topics(slot)
|
||||
|
||||
# items: 1-based index
|
||||
base = f"{TOPIC_PLAYBACK_BASE}/{slot_num}"
|
||||
for i, name in enumerate(item_titles, start=1):
|
||||
mqtt_publish(f"{base}/items/{i}", name, retain=True)
|
||||
|
||||
mqtt_publish(
|
||||
TOPIC_INGEST_PROGRESS,
|
||||
{"slot": slot_num, "state": "ready", "title": title, "count": slot.item_count},
|
||||
)
|
||||
|
||||
def publish_slot_topics(slot: PlaylistSlot):
|
||||
"""Publish the title/state/count for one slot to its MQTT topics."""
|
||||
base = f"{TOPIC_PLAYBACK_BASE}/{slot.slot}"
|
||||
mqtt_publish(f"{base}/title", slot.title, retain=True)
|
||||
mqtt_publish(f"{base}/state", slot.state, retain=True)
|
||||
mqtt_publish(f"{base}/count", slot.item_count, retain=True)
|
||||
|
||||
# --- Controller → Player orchestration ---
|
||||
|
||||
def handle_play_request(cmd: Dict) -> None:
|
||||
"""
|
||||
Handle commands from HA/Companion on TOPIC_CONTROLLER_CMD.
|
||||
|
||||
Supported payloads:
|
||||
- {"action": "play", "slot": N}
|
||||
- {"action": "play", "playlist": "/some/path.m3u"} # future perm playlist
|
||||
- {"action": "stop"}
|
||||
"""
|
||||
action = cmd.get("action")
|
||||
|
||||
# Which slot is currently playing?
|
||||
playing_slot: Optional[int] = None
|
||||
for s, slot in state.slots.items():
|
||||
if slot.state == "playing":
|
||||
playing_slot = s
|
||||
break
|
||||
|
||||
# --- STOP command ---
|
||||
if action == "stop":
|
||||
# Forward stop to player
|
||||
mqtt_publish(TOPIC_PLAYER_CMD, {"action": "stop"})
|
||||
|
||||
# Any slot marked playing goes back to ready
|
||||
changed = False
|
||||
for s, slot in state.slots.items():
|
||||
if slot.state == "playing":
|
||||
slot.state = "ready"
|
||||
base = f"{TOPIC_PLAYBACK_BASE}/{s}"
|
||||
mqtt_publish(f"{base}/state", "ready", retain=True)
|
||||
changed = True
|
||||
if changed:
|
||||
state.save()
|
||||
return
|
||||
|
||||
# --- PLAY command ---
|
||||
if action != "play":
|
||||
# ignore unknown for now
|
||||
return
|
||||
|
||||
slot_num_raw = cmd.get("slot")
|
||||
playlist = cmd.get("playlist") or cmd.get("path")
|
||||
label = cmd.get("label")
|
||||
|
||||
slot_obj: Optional[PlaylistSlot] = None
|
||||
|
||||
if slot_num_raw is not None:
|
||||
try:
|
||||
slot_num = int(slot_num_raw)
|
||||
except (TypeError, ValueError):
|
||||
print(f"[controller] bad slot in play cmd: {slot_num_raw!r}")
|
||||
return
|
||||
slot_obj = state.slots.get(slot_num)
|
||||
if not slot_obj:
|
||||
print(f"[controller] no such slot {slot_num}")
|
||||
return
|
||||
|
||||
# If we have a slot but no playlist, derive path from the slot.
|
||||
if slot_obj is not None and not playlist:
|
||||
playlist = slot_obj.m3u
|
||||
label = label or slot_obj.title
|
||||
|
||||
# If we still have no playlist/path, bail.
|
||||
if not playlist:
|
||||
print("[controller] play cmd missing playlist/path and no valid slot provided")
|
||||
return
|
||||
|
||||
# Enforce only one slot at a time for now
|
||||
if playing_slot is not None and slot_obj is not None and playing_slot != slot_obj.slot:
|
||||
# briefly mark this requested slot as busy, then restore it
|
||||
set_slot_busy_temporarily(slot_obj, seconds=5)
|
||||
|
||||
mqtt_publish(
|
||||
TOPIC_INGEST_PROGRESS,
|
||||
{
|
||||
"slot": slot_obj.slot,
|
||||
"state": "busy",
|
||||
"message": f"slot {playing_slot} already playing; stop first",
|
||||
},
|
||||
)
|
||||
return
|
||||
|
||||
|
||||
# If we're starting a slot-backed playlist, mark that slot as playing
|
||||
if slot_obj is not None:
|
||||
slot_obj.state = "playing"
|
||||
state.save()
|
||||
base = f"{TOPIC_PLAYBACK_BASE}/{slot_obj.slot}"
|
||||
mqtt_publish(f"{base}/state", "playing", retain=True)
|
||||
|
||||
# Build command for the player
|
||||
play_cmd = {
|
||||
"action": "play",
|
||||
"playlist": playlist,
|
||||
}
|
||||
if slot_obj is not None:
|
||||
play_cmd["slot"] = slot_obj.slot
|
||||
if label:
|
||||
play_cmd["label"] = label
|
||||
|
||||
mqtt_publish(TOPIC_PLAYER_CMD, play_cmd)
|
||||
print(f"[controller] sent play cmd to player: {play_cmd}")
|
||||
|
||||
|
||||
# --- HA Discovery ---
|
||||
|
||||
HA_DISCOVERY_PREFIX = os.environ.get("HA_DISCOVERY_PREFIX", "homeassistant")
|
||||
HA_DEVICE = {
|
||||
"identifiers": ["ytplayer-controller"],
|
||||
"name": "YouTube Controller",
|
||||
"manufacturer": "River of Glory",
|
||||
"model": "YT Controller",
|
||||
}
|
||||
|
||||
def ha_discovery_publish():
|
||||
"""
|
||||
Publish MQTT Discovery configs for:
|
||||
- sensor.yt_slot_<n>_title
|
||||
- sensor.yt_slot_<n>_state
|
||||
- sensor.yt_slot_<n>_count
|
||||
"""
|
||||
for n in range(1, MAX_SLOTS + 1):
|
||||
base = f"{TOPIC_PLAYBACK_BASE}/{n}"
|
||||
obj_id = f"yt_slot_{n}"
|
||||
|
||||
# Title
|
||||
cfg_title = {
|
||||
"name": f"YT Slot {n} Title",
|
||||
"state_topic": f"{base}/title",
|
||||
"unique_id": f"{obj_id}_title",
|
||||
"icon": "mdi:playlist-music",
|
||||
"device": HA_DEVICE,
|
||||
}
|
||||
mqtt_publish(
|
||||
f"{HA_DISCOVERY_PREFIX}/sensor/{obj_id}_title/config",
|
||||
cfg_title,
|
||||
retain=True,
|
||||
)
|
||||
|
||||
# State (ready/downloading/playing/busy/error/idle)
|
||||
cfg_state = {
|
||||
"name": f"YT Slot {n} State",
|
||||
"state_topic": f"{base}/state",
|
||||
"unique_id": f"{obj_id}_state",
|
||||
"icon": "mdi:circle-small",
|
||||
"device": HA_DEVICE,
|
||||
}
|
||||
mqtt_publish(
|
||||
f"{HA_DISCOVERY_PREFIX}/sensor/{obj_id}_state/config",
|
||||
cfg_state,
|
||||
retain=True,
|
||||
)
|
||||
|
||||
# Count
|
||||
cfg_count = {
|
||||
"name": f"YT Slot {n} Count",
|
||||
"state_topic": f"{base}/count",
|
||||
"unique_id": f"{obj_id}_count",
|
||||
"icon": "mdi:numeric",
|
||||
"device": HA_DEVICE,
|
||||
}
|
||||
mqtt_publish(
|
||||
f"{HA_DISCOVERY_PREFIX}/sensor/{obj_id}_count/config",
|
||||
cfg_count,
|
||||
retain=True,
|
||||
)
|
||||
|
||||
|
||||
# --- Public API used by Flask app ---
|
||||
|
||||
def enqueue_download(url: str, submitter: Optional[str], note: Optional[str]) -> Dict:
|
||||
"""
|
||||
Called by Flask when a new playlist is submitted.
|
||||
- If this URL is already in a slot, reuse that slot and only grab new videos.
|
||||
- Otherwise, allocate a new slot.
|
||||
Returns a small dict summarizing what happened, for UI or HA.
|
||||
"""
|
||||
state.load()
|
||||
state.prune_expired()
|
||||
|
||||
# 1) Check if this URL already exists in a slot
|
||||
existing_slot_obj = None
|
||||
for sl in state.slots.values():
|
||||
if sl.url == url:
|
||||
existing_slot_obj = sl
|
||||
break
|
||||
|
||||
if existing_slot_obj:
|
||||
slot_num = existing_slot_obj.slot
|
||||
print(f"[controller] enqueue_download: reusing existing slot {slot_num} for URL {url!r}")
|
||||
reused = True
|
||||
else:
|
||||
# 2) Allocate a new slot
|
||||
slot_num = state.get_next_free_slot()
|
||||
reused = False
|
||||
|
||||
if slot_num is None:
|
||||
# queue full
|
||||
expires = [sl.expires_at for sl in state.slots.values()]
|
||||
oldest = min(expires) if expires else time.time()
|
||||
return {
|
||||
"status": "queue_full",
|
||||
"oldest_expires_at": oldest,
|
||||
"oldest_expires_in_seconds": int(oldest - time.time()),
|
||||
}
|
||||
|
||||
# 3) Spawn background thread so Flask doesn't block
|
||||
t = threading.Thread(
|
||||
target=_download_playlist_to_slot,
|
||||
args=(url, slot_num),
|
||||
daemon=True,
|
||||
)
|
||||
t.start()
|
||||
|
||||
# 4) Simple MQTT log of submission
|
||||
mqtt_publish(
|
||||
TOPIC_INGEST_PROGRESS,
|
||||
{
|
||||
"slot": slot_num,
|
||||
"state": "queued",
|
||||
"url": url,
|
||||
"submitter": submitter,
|
||||
"note": note,
|
||||
"reused_slot": reused,
|
||||
},
|
||||
)
|
||||
|
||||
return {
|
||||
"status": "queued",
|
||||
"slot": slot_num,
|
||||
"reused_slot": reused,
|
||||
}
|
||||
|
||||
|
||||
# --- Post-define initialization that needs functions above ---
|
||||
|
||||
print("[controller] running HA discovery publish…")
|
||||
ha_discovery_publish()
|
||||
print("[controller] ha_discovery_publish() done")
|
||||
|
||||
# Re-announce any existing slots so HA / Deck see them after a restart
|
||||
for s, slot in state.slots.items():
|
||||
publish_slot_topics(slot)
|
||||
print("[controller] re-published slot states")
|
||||
Reference in New Issue
Block a user