301 lines
12 KiB
Python
Executable File
301 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# args_json: {"type":"object","required":["action"],"properties":{"action":{"type":"string","description":"list|status|open|playlist|play|pause|stop|next|previous"},"target":{"type":"string","description":"Target file, URI, or playlist description"},"player":{"type":"string","description":"Player name"},"library":{"type":"string","description":"Music library path"},"count":{"type":"integer","description":"Playlist track count (default: 15)"},"output":{"type":"string","description":"M3U8 output path"}}}
|
|
# ollie:prompt
|
|
# ## gui_media
|
|
#
|
|
# Control MPRIS media players.
|
|
#
|
|
# **Args**: `[action]` or `[action, target]`
|
|
# - Actions: `list`, `status`, `open`, `playlist`, `play`, `pause`, `stop`, `next`, `previous`
|
|
#
|
|
# ```
|
|
# gui_media(action="status")
|
|
# gui_media(action="open", target="/path/to/song.mp3")
|
|
# ```
|
|
# ollie:end
|
|
"""Reliable MPRIS control, including Elisa's Application.Open requirement."""
|
|
import hashlib
|
|
import json
|
|
import pathlib
|
|
import random
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
BUS = "org.freedesktop.DBus"
|
|
OBJ = "/org/mpris/MediaPlayer2"
|
|
PLAYER = "org.mpris.MediaPlayer2.Player"
|
|
PROPS = "org.freedesktop.DBus.Properties"
|
|
|
|
|
|
def call(dest, path, method, *args, check=True):
|
|
p = subprocess.run(
|
|
["gdbus", "call", "--session", "--dest", dest,
|
|
"--object-path", path, "--method", method, *args],
|
|
text=True, capture_output=True,
|
|
)
|
|
if check and p.returncode:
|
|
raise RuntimeError(p.stderr.strip() or p.stdout.strip())
|
|
return p.stdout.strip() if p.returncode == 0 else ""
|
|
|
|
|
|
def players():
|
|
out = call(BUS, "/org/freedesktop/DBus", BUS + ".ListNames")
|
|
return sorted(set(re.findall(r"org\.mpris\.MediaPlayer2[\w.-]+", out)))
|
|
|
|
|
|
def choose(hint=None, activate=False):
|
|
names = players()
|
|
if hint:
|
|
matches = [n for n in names if hint.lower() in n.lower()]
|
|
if matches:
|
|
return matches[0]
|
|
if activate and hint.lower() == "elisa":
|
|
call(BUS, "/org/freedesktop/DBus", BUS + ".StartServiceByName",
|
|
"org.kde.elisa", "0")
|
|
time.sleep(0.5)
|
|
matches = [n for n in players() if "elisa" in n.lower()]
|
|
if matches:
|
|
return matches[0]
|
|
raise RuntimeError(f"no MPRIS player matching {hint!r}")
|
|
if names:
|
|
playing = [n for n in names if prop(n, "PlaybackStatus") == "Playing"]
|
|
return (playing or names)[0]
|
|
if activate:
|
|
call(BUS, "/org/freedesktop/DBus", BUS + ".StartServiceByName",
|
|
"org.kde.elisa", "0")
|
|
time.sleep(0.5)
|
|
names = players()
|
|
if names:
|
|
return names[0]
|
|
raise RuntimeError("no MPRIS media player is available")
|
|
|
|
|
|
def prop(dest, name):
|
|
out = call(dest, OBJ, PROPS + ".Get", PLAYER, name, check=False)
|
|
m = re.search(r"<'([^']*)'>", out)
|
|
return m.group(1) if m else out
|
|
|
|
|
|
def metadata(dest):
|
|
return call(dest, OBJ, PROPS + ".Get", PLAYER, "Metadata", check=False)
|
|
|
|
|
|
def metadata_has_uri(dest, uri):
|
|
# Elisa reports file URLs with decoded spaces, while pathlib.as_uri()
|
|
# correctly sends percent-encoded spaces to Application.Open.
|
|
from urllib.parse import unquote
|
|
return uri in metadata(dest) or unquote(uri) in metadata(dest)
|
|
|
|
|
|
def uri_for(value):
|
|
if re.match(r"^[A-Za-z][A-Za-z0-9+.-]*://", value):
|
|
return value
|
|
path = pathlib.Path(value).expanduser().resolve()
|
|
if not path.is_file():
|
|
raise RuntimeError(f"media file not found: {path}")
|
|
return path.as_uri()
|
|
|
|
|
|
def clear_playlist(dest):
|
|
call(dest, OBJ, PLAYER + ".Stop", check=False)
|
|
if "elisa" not in dest.lower():
|
|
return
|
|
call(dest, "/org/kde/elisa", "org.freedesktop.Application.ActivateAction",
|
|
"clear-playlist", "[]", "{}")
|
|
# Elisa resets currentTrack to zero when the queue is empty. Metadata may
|
|
# remain stale briefly, so the index is the reliable completion signal.
|
|
for _ in range(20):
|
|
current = call(dest, OBJ, PROPS + ".Get", PLAYER, "currentTrack", check=False)
|
|
if re.search(r"<0>", current):
|
|
return
|
|
time.sleep(0.1)
|
|
raise RuntimeError("Elisa did not clear its playlist")
|
|
|
|
|
|
def open_uris(dest, uris, clear=False):
|
|
if clear:
|
|
clear_playlist(dest)
|
|
else:
|
|
call(dest, OBJ, PLAYER + ".Stop", check=False)
|
|
opened = False
|
|
if "elisa" in dest.lower():
|
|
# Elisa advertises MPRIS OpenUri but may ignore it. Its actual file-open
|
|
# path is org.freedesktop.Application.Open. JSON array syntax is also
|
|
# valid GVariant text and safely handles quotes in paths.
|
|
opened = bool(call(dest, "/org/kde/elisa",
|
|
"org.freedesktop.Application.Open",
|
|
json.dumps(uris), "{}", check=False) == "()")
|
|
if not opened:
|
|
if len(uris) != 1:
|
|
raise RuntimeError("this player cannot load a playlist through D-Bus")
|
|
call(dest, OBJ, PLAYER + ".OpenUri", uris[0])
|
|
|
|
|
|
def start_verified(dest, uri):
|
|
for _ in range(40):
|
|
if metadata_has_uri(dest, uri):
|
|
break
|
|
time.sleep(0.25)
|
|
else:
|
|
raise RuntimeError("player did not load the requested URI; playback left stopped")
|
|
|
|
call(dest, OBJ, PLAYER + ".Play")
|
|
for attempt in range(20):
|
|
if prop(dest, "PlaybackStatus") == "Playing" and metadata_has_uri(dest, uri):
|
|
return
|
|
# Elisa can expose a newly opened track as Stopped while ignoring the
|
|
# first Play. Toggling once initializes its playback state; Play then
|
|
# reliably starts the already verified track.
|
|
if attempt == 3 and "elisa" in dest.lower():
|
|
call(dest, OBJ, PLAYER + ".PlayPause", check=False)
|
|
call(dest, OBJ, PLAYER + ".Play", check=False)
|
|
time.sleep(0.25)
|
|
raise RuntimeError("requested URI loaded but did not start playing")
|
|
|
|
|
|
def open_media(dest, value):
|
|
uri = uri_for(value)
|
|
open_uris(dest, [uri])
|
|
start_verified(dest, uri)
|
|
|
|
print(f"player={dest}\nstatus=Playing\nuri={uri}")
|
|
|
|
|
|
AUDIO_EXTS = {".flac", ".mp3", ".ogg", ".opus", ".m4a", ".wav"}
|
|
MOODS = {
|
|
"chill": ({"ambient", "dream", "dreamy", "soft", "slow", "quiet", "calm", "mellow", "warm", "night", "moon", "ocean", "sea", "wave", "waves", "air", "shine", "memory", "sleep", "blue", "acoustic", "instrumental", "home"},
|
|
{"metal", "death", "war", "fire", "burn", "rage", "angry", "hate", "kill", "blood", "violent", "blast", "bore", "rancorous"}),
|
|
"focus": ({"ambient", "instrumental", "minimal", "piano", "acoustic", "memory", "space", "air"},
|
|
{"party", "dance", "rage", "chaos", "scream", "live"}),
|
|
"energetic": ({"dance", "fire", "fast", "power", "run", "alive", "electric", "party", "drive", "exalter"},
|
|
{"sleep", "slow", "quiet", "ambient", "lament"}),
|
|
"melancholy": ({"blue", "rain", "memory", "alone", "lost", "goodbye", "dead", "dream", "night", "winter", "ghost", "undertow"},
|
|
{"party", "dance", "happy", "sunshine"}),
|
|
}
|
|
|
|
|
|
def words(text):
|
|
return set(re.findall(r"[a-z0-9]+", text.lower()))
|
|
|
|
|
|
def library_tracks(root):
|
|
root = pathlib.Path(root).expanduser().resolve()
|
|
if not root.is_dir():
|
|
raise RuntimeError(f"music library not found: {root}")
|
|
tracks = []
|
|
for path in root.rglob("*"):
|
|
if path.is_file() and path.suffix.lower() in AUDIO_EXTS:
|
|
parent = path.parent.name
|
|
artist, _, album = parent.partition(" - ")
|
|
name = path.stem
|
|
title = re.sub(r"^.*? - \d+\s*", "", name)
|
|
tracks.append({"path": path, "artist": artist or parent,
|
|
"album": album, "title": title,
|
|
"text": " ".join((artist, album, title)).lower()})
|
|
if not tracks:
|
|
raise RuntimeError(f"no audio files found under {root}")
|
|
return tracks
|
|
|
|
|
|
def rank_playlist(description, tracks, count):
|
|
query = words(description)
|
|
positive, negative = set(), set()
|
|
for mood, (good, bad) in MOODS.items():
|
|
if mood in query or (mood == "chill" and query & {"relax", "relaxed", "mellow", "calm"}):
|
|
positive |= good
|
|
negative |= bad
|
|
positive |= query - {"make", "playlist", "music", "songs", "tracks", "some", "me", "a", "an", "the"}
|
|
seed = int(hashlib.sha256(description.lower().encode()).hexdigest()[:16], 16)
|
|
rng = random.Random(seed)
|
|
ranked = []
|
|
for track in tracks:
|
|
tokens = words(track["text"])
|
|
score = 4 * len(tokens & positive) - 5 * len(tokens & negative)
|
|
# Album/artist matches influence the whole selection; a tiny seeded
|
|
# jitter makes equally ranked results varied but reproducible.
|
|
score += 2 * len(words(track["album"]) & positive)
|
|
score += len(words(track["artist"]) & positive)
|
|
ranked.append((score + rng.random() * 0.2, track))
|
|
ranked.sort(key=lambda item: item[0], reverse=True)
|
|
|
|
selected, artists, albums = [], {}, {}
|
|
for _, track in ranked:
|
|
artist = track["artist"].lower()
|
|
album = (artist, track["album"].lower())
|
|
if artists.get(artist, 0) >= 3 or albums.get(album, 0) >= 2:
|
|
continue
|
|
selected.append(track)
|
|
artists[artist] = artists.get(artist, 0) + 1
|
|
albums[album] = albums.get(album, 0) + 1
|
|
if len(selected) == count:
|
|
break
|
|
return selected
|
|
|
|
|
|
def build_playlist(dest, description, root, count, output=None):
|
|
if count < 1 or count > 100:
|
|
raise RuntimeError("playlist count must be between 1 and 100")
|
|
selected = rank_playlist(description, library_tracks(root), count)
|
|
if not selected:
|
|
raise RuntimeError("no tracks matched the playlist description")
|
|
if output:
|
|
playlist = pathlib.Path(output).expanduser().resolve()
|
|
else:
|
|
slug = re.sub(r"[^a-z0-9]+", "-", description.lower()).strip("-")[:50] or "playlist"
|
|
playlist = pathlib.Path.home() / ".local/share/ollie/playlists" / f"{slug}.m3u8"
|
|
playlist.parent.mkdir(parents=True, exist_ok=True)
|
|
playlist.write_text("#EXTM3U\n" + "\n".join(str(t["path"]) for t in selected) + "\n",
|
|
encoding="utf-8")
|
|
uris = [t["path"].as_uri() for t in selected]
|
|
open_uris(dest, uris, clear=True)
|
|
start_verified(dest, uris[0])
|
|
print(f"player={dest}\nstatus=Playing\nplaylist={playlist}\ntracks={len(selected)}")
|
|
for index, track in enumerate(selected, 1):
|
|
print(f"{index:02d}. {track['artist']} — {track['title']}")
|
|
|
|
|
|
def main():
|
|
import os as _os
|
|
sys.path.insert(0, _os.path.dirname(_os.path.abspath(__file__)))
|
|
from _lib.args import parse_args as _parse_args
|
|
_a = _parse_args()
|
|
|
|
class _Args:
|
|
command = _a.require("action")
|
|
target = _a.get("target")
|
|
player = _a.get("player")
|
|
library = _a.get("library", str(pathlib.Path.home() / "music"))
|
|
count = _a.get_int("count", 15)
|
|
output = _a.get("output")
|
|
args = _Args()
|
|
|
|
if args.command == "list":
|
|
for name in players():
|
|
print(f"{name}\t{prop(name, 'PlaybackStatus')}")
|
|
return
|
|
|
|
dest = choose(args.player, activate=args.command in {"open", "playlist"})
|
|
if args.command == "open":
|
|
if not args.target:
|
|
ap.error("open requires PATH_OR_URI")
|
|
open_media(dest, args.target)
|
|
elif args.command == "playlist":
|
|
if not args.target:
|
|
ap.error("playlist requires a description")
|
|
build_playlist(dest, args.target, args.library, args.count, args.output)
|
|
elif args.command == "status":
|
|
print(f"player={dest}\nstatus={prop(dest, 'PlaybackStatus')}\nmetadata={metadata(dest)}")
|
|
else:
|
|
call(dest, OBJ, PLAYER + "." + args.command.capitalize())
|
|
print(f"player={dest}\nstatus={prop(dest, 'PlaybackStatus')}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except RuntimeError as e:
|
|
print(f"ERROR: {e}", file=sys.stderr)
|
|
sys.exit(1)
|