add gui_media tool and docs
This commit is contained in:
Levi Neely
parent
4a29efc121
commit
a12e00e231
|
|
@ -230,3 +230,30 @@ call_tool: calls=[{tool: "gui_brightness", args: ["down", "25"]}]
|
|||
```
|
||||
|
||||
**Note**: The underlying PowerDevil API uses absolute values (0 to ~10000). This tool handles the conversion — always pass percentages.
|
||||
|
||||
---
|
||||
|
||||
## gui_media
|
||||
|
||||
Control MPRIS media players through the session D-Bus. Use this instead of launching a player process directly. It reliably opens local files in Elisa, waits for the requested URI to load, starts playback, and verifies that the same URI is playing.
|
||||
|
||||
**Args**: `[command]`, `[command, target]`, optionally followed by `["--player", name]`
|
||||
|
||||
Commands: `list`, `status`, `open <path-or-uri>`, `playlist <description>`, `play`, `pause`, `stop`, `next`, `previous`
|
||||
|
||||
Playlist options: `--count N` (default 15), `--library PATH` (default `~/music`), `--output PATH` (default `~/.local/share/ollie/playlists/<description>.m3u8`)
|
||||
|
||||
```
|
||||
call_tool: calls=[{tool: "gui_media", args: ["list"]}]
|
||||
call_tool: calls=[{tool: "gui_media", args: ["status", "--player", "elisa"]}]
|
||||
call_tool: calls=[{tool: "gui_media", args: ["open", "/home/user/Music/track.flac", "--player", "elisa"]}]
|
||||
call_tool: calls=[{tool: "gui_media", args: ["playlist", "chill dreamy late night", "--count", "12", "--player", "elisa"]}]
|
||||
call_tool: calls=[{tool: "gui_media", args: ["pause", "--player", "elisa"]}]
|
||||
call_tool: calls=[{tool: "gui_media", args: ["play", "--player", "elisa"]}]
|
||||
```
|
||||
|
||||
**Player selection**: `--player` accepts a case-insensitive substring of an MPRIS bus name. Without it, the tool prefers a currently playing player, then the first available player. `open` activates Elisa over D-Bus if no player is running.
|
||||
|
||||
**Playlist matching**: Uses local artist, album, title, and path text with mood profiles for chill/relaxed, focus, energetic, and melancholy descriptions. Results are deterministic and limited to three tracks per artist and two per album for variety. Local files may lack genre, mood, or BPM tags, so describe artists, albums, or title themes when you need tighter matching.
|
||||
|
||||
**Safety**: `playlist` clears Elisa's existing queue before loading the generated tracks. `open` and `playlist` stop existing playback first and fail without resuming the old source if loading cannot be verified.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,284 @@
|
|||
#!/usr/bin/env python3
|
||||
# description: Control MPRIS players and build local playlists. Args: <list|status|open|playlist|play|pause|stop|next|previous> [target] [options]
|
||||
"""Reliable MPRIS control, including Elisa's Application.Open requirement."""
|
||||
import argparse
|
||||
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():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("command", choices=["list", "status", "open", "playlist",
|
||||
"play", "pause", "stop", "next", "previous"])
|
||||
ap.add_argument("target", nargs="?", help="path, URI, or playlist description")
|
||||
ap.add_argument("--player")
|
||||
ap.add_argument("--library", default=str(pathlib.Path.home() / "music"))
|
||||
ap.add_argument("--count", type=int, default=15)
|
||||
ap.add_argument("--output", help="M3U8 output path")
|
||||
args = ap.parse_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)
|
||||
Reference in New Issue