server2
a8cd23546d
- README: features, layout, requirements, install, IPC, runtime files, limits - AGENTS.md: the user's ground rules (branch + approval, never touch real players' sound), architecture, measured performance rules, type-check and test loop, desktop test driver and its pitfalls, open threads - tools/: amane-remote.sh and fake_mpris.py, so the test loop doesn't depend on the Hermes skill - .gitignore: the Cargo files amane generates
257 lines
10 KiB
Python
Executable File
257 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Fake MPRIS players for visual tests of media widgets.
|
|
|
|
usage: fake_mpris.py PROFILE (0, 1, 2, ... see PROFILES)
|
|
|
|
Each process owns org.mpris.MediaPlayer2.fake<PROFILE>, answers the calls a bar
|
|
makes (GetAll, PlayPause, Next, Previous, Seek) and, when the profile says so,
|
|
plays through `paplay` as its own child, so mixers that match sound streams to
|
|
players by process ancestry find it: "silence" (zeros) or "synth" (a beat, a
|
|
chord and a sweep, for visualisers and equalizers). Set FAKE_MPRIS_SINK to a
|
|
null sink (`pactl load-module module-null-sink sink_name=...`) so the synth is
|
|
never audible. Covers are generated PNGs in a folder with a space in its name
|
|
(exercises percent-encoded file:// links). Profiles cover: long title +
|
|
playing, >1 h paused podcast, live stream without cover or sound, synth.
|
|
Needs PyGObject (gi) and paplay. SIGTERM stops paplay too.
|
|
"""
|
|
|
|
import array
|
|
import math
|
|
import os
|
|
import random
|
|
import signal
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import urllib.parse
|
|
import zlib
|
|
|
|
from gi.repository import Gio, GLib
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
RATE = 44100
|
|
|
|
# identity, playing, sound ("silence", "synth" or None), tracks of (title, artist, seconds, cover colours or None)
|
|
PROFILES = [
|
|
("Lo-Fi Lab", True, "silence", [
|
|
("Late Night Tape Hiss (Extended Rain Mix) feat. Somebody With A Long Name",
|
|
"Nujabes Tribute Collective", 214, ((255, 140, 60), (200, 40, 140))),
|
|
("Second Track", "Lo-Fi Lab Band", 180, ((250, 210, 70), (60, 160, 120))),
|
|
]),
|
|
("Podcast Deck", False, "silence", [
|
|
("Episode 142: Shipping Small", "The Lazy Dev Show", 3725, ((40, 190, 200), (20, 30, 90))),
|
|
]),
|
|
("Radio Stream", True, None, [
|
|
("Live from the Basement", "Night FM", 0, None),
|
|
]),
|
|
("Synth Test", True, "synth", [
|
|
("Chest Pain (I Love)", "Malcolm Todd", 200, ((120, 110, 190), (30, 20, 60))),
|
|
]),
|
|
]
|
|
|
|
|
|
def synth(out, stop):
|
|
"""Endless s16le stereo: a kick every 0.5 s, hats, a breathing chord and a slow sweep."""
|
|
frame = 0
|
|
chunk = 2048
|
|
while not stop.is_set():
|
|
samples = array.array("h")
|
|
for _ in range(chunk):
|
|
t = frame / RATE
|
|
beat = t % 0.5
|
|
kick = math.sin(2 * math.pi * (50 + 90 * math.exp(-beat * 30)) * beat) * math.exp(-beat * 8)
|
|
hat = (random.random() * 2 - 1) * math.exp(-(t % 0.25) * 60) * 0.25
|
|
chord = sum(math.sin(2 * math.pi * f * t) for f in (220.0, 277.2, 329.6)) / 3 * (0.3 + 0.2 * math.sin(t))
|
|
sweep = math.sin(2 * math.pi * (400 + 3000 * (0.5 + 0.5 * math.sin(t * 0.7))) * t) * 0.15
|
|
value = int(max(-1.0, min(1.0, 0.5 * kick + hat + chord + sweep)) * 20000)
|
|
samples.append(value)
|
|
samples.append(value)
|
|
frame += 1
|
|
try:
|
|
out.write(samples.tobytes())
|
|
out.flush()
|
|
except (BrokenPipeError, ValueError):
|
|
return
|
|
|
|
INTERFACES = """
|
|
<node>
|
|
<interface name="org.mpris.MediaPlayer2">
|
|
<method name="Raise"/>
|
|
<method name="Quit"/>
|
|
<property name="Identity" type="s" access="read"/>
|
|
<property name="CanQuit" type="b" access="read"/>
|
|
<property name="CanRaise" type="b" access="read"/>
|
|
<property name="HasTrackList" type="b" access="read"/>
|
|
<property name="SupportedUriSchemes" type="as" access="read"/>
|
|
<property name="SupportedMimeTypes" type="as" access="read"/>
|
|
</interface>
|
|
<interface name="org.mpris.MediaPlayer2.Player">
|
|
<method name="PlayPause"/>
|
|
<method name="Play"/>
|
|
<method name="Pause"/>
|
|
<method name="Stop"/>
|
|
<method name="Next"/>
|
|
<method name="Previous"/>
|
|
<method name="Seek"><arg name="Offset" type="x" direction="in"/></method>
|
|
<method name="SetPosition">
|
|
<arg name="TrackId" type="o" direction="in"/>
|
|
<arg name="Position" type="x" direction="in"/>
|
|
</method>
|
|
<property name="PlaybackStatus" type="s" access="read"/>
|
|
<property name="Metadata" type="a{sv}" access="read"/>
|
|
<property name="Position" type="x" access="read"/>
|
|
<property name="Rate" type="d" access="read"/>
|
|
<property name="Volume" type="d" access="read"/>
|
|
<property name="CanGoNext" type="b" access="read"/>
|
|
<property name="CanGoPrevious" type="b" access="read"/>
|
|
<property name="CanPlay" type="b" access="read"/>
|
|
<property name="CanPause" type="b" access="read"/>
|
|
<property name="CanSeek" type="b" access="read"/>
|
|
<property name="CanControl" type="b" access="read"/>
|
|
</interface>
|
|
</node>
|
|
"""
|
|
|
|
|
|
def png(path, colours, size=256):
|
|
"""A diagonal two-colour gradient with a lighter disc, written with zlib only."""
|
|
(r0, g0, b0), (r1, g1, b1) = colours
|
|
rows = []
|
|
for y in range(size):
|
|
row = bytearray(b"\x00")
|
|
for x in range(size):
|
|
t = (x + y) / (2 * size - 2)
|
|
disc = (x - size * 0.62) ** 2 + (y - size * 0.38) ** 2 < (size * 0.22) ** 2
|
|
lift = 60 if disc else 0
|
|
row += bytes(min(255, int(a + (b - a) * t) + lift) for a, b in ((r0, r1), (g0, g1), (b0, b1)))
|
|
rows.append(bytes(row))
|
|
|
|
def chunk(kind, data):
|
|
return struct.pack(">I", len(data)) + kind + data + struct.pack(">I", zlib.crc32(kind + data))
|
|
|
|
header = struct.pack(">IIBBBBB", size, size, 8, 2, 0, 0, 0)
|
|
with open(path, "wb") as out:
|
|
out.write(b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", header) + chunk(b"IDAT", zlib.compress(b"".join(rows))) + chunk(b"IEND", b""))
|
|
|
|
|
|
class Player:
|
|
def __init__(self, index):
|
|
self.index = index
|
|
self.identity, self.playing, sound, self.tracks = PROFILES[index]
|
|
self.track = 0
|
|
self.offset = 0.0 # seconds into the track at `since`
|
|
self.since = time.monotonic()
|
|
self.sound = None
|
|
self.stop_synth = threading.Event()
|
|
if sound:
|
|
command = ["paplay", "--raw", "--format=s16le", f"--rate={RATE}", "--channels=2",
|
|
f"--client-name={self.identity}"]
|
|
if os.environ.get("FAKE_MPRIS_SINK"):
|
|
command.append(f"--device={os.environ['FAKE_MPRIS_SINK']}")
|
|
if sound == "silence":
|
|
self.sound = subprocess.Popen(command + ["/dev/zero"])
|
|
else:
|
|
# paplay reads stdin when no file is given
|
|
self.sound = subprocess.Popen(command, stdin=subprocess.PIPE)
|
|
threading.Thread(target=synth, args=(self.sound.stdin, self.stop_synth), daemon=True).start()
|
|
|
|
covers = os.path.join(HERE, "fake covers")
|
|
os.makedirs(covers, exist_ok=True)
|
|
self.covers = []
|
|
for number, (_, _, _, colours) in enumerate(self.tracks):
|
|
if colours is None:
|
|
self.covers.append("")
|
|
continue
|
|
path = os.path.join(covers, f"player{index}-track{number}.png")
|
|
if not os.path.exists(path):
|
|
png(path, colours)
|
|
self.covers.append("file://" + urllib.parse.quote(path))
|
|
|
|
def position(self):
|
|
length = self.tracks[self.track][2]
|
|
moved = time.monotonic() - self.since if self.playing else 0.0
|
|
seconds = self.offset + moved
|
|
return min(seconds, length) if length else seconds
|
|
|
|
def rebase(self, seconds):
|
|
self.offset = max(0.0, seconds)
|
|
self.since = time.monotonic()
|
|
|
|
def metadata(self):
|
|
title, artist, length, _ = self.tracks[self.track]
|
|
data = {
|
|
"mpris:trackid": GLib.Variant("o", f"/org/mpris/MediaPlayer2/fake/track{self.track}"),
|
|
"xesam:title": GLib.Variant("s", title),
|
|
"xesam:artist": GLib.Variant("as", [artist]),
|
|
"mpris:artUrl": GLib.Variant("s", self.covers[self.track]),
|
|
}
|
|
if length:
|
|
data["mpris:length"] = GLib.Variant("x", length * 1_000_000)
|
|
return GLib.Variant("a{sv}", data)
|
|
|
|
def get(self, _connection, _sender, _path, _interface, name):
|
|
constant = {
|
|
"Identity": GLib.Variant("s", self.identity),
|
|
"CanQuit": GLib.Variant("b", False), "CanRaise": GLib.Variant("b", False),
|
|
"HasTrackList": GLib.Variant("b", False),
|
|
"SupportedUriSchemes": GLib.Variant("as", []), "SupportedMimeTypes": GLib.Variant("as", []),
|
|
"Rate": GLib.Variant("d", 1.0), "Volume": GLib.Variant("d", 1.0),
|
|
}
|
|
if name in constant:
|
|
return constant[name]
|
|
if name == "PlaybackStatus":
|
|
return GLib.Variant("s", "Playing" if self.playing else "Paused")
|
|
if name == "Metadata":
|
|
return self.metadata()
|
|
if name == "Position":
|
|
return GLib.Variant("x", int(self.position() * 1_000_000))
|
|
return GLib.Variant("b", True) # every Can* capability
|
|
|
|
def call(self, _connection, _sender, _path, _interface, method, parameters, invocation):
|
|
if method in ("PlayPause", "Play", "Pause", "Stop"):
|
|
here = self.position()
|
|
self.playing = {"PlayPause": not self.playing, "Play": True}.get(method, False)
|
|
self.rebase(here)
|
|
elif method in ("Next", "Previous"):
|
|
step = 1 if method == "Next" else -1
|
|
self.track = (self.track + step) % len(self.tracks)
|
|
self.rebase(0)
|
|
elif method == "Seek":
|
|
self.rebase(self.position() + parameters.unpack()[0] / 1_000_000)
|
|
elif method == "SetPosition":
|
|
self.rebase(parameters.unpack()[1] / 1_000_000)
|
|
print(f"{self.identity}: {method} {parameters.unpack() if parameters else ''} "
|
|
f"-> {'playing' if self.playing else 'paused'} at {self.position():.1f}s", flush=True)
|
|
invocation.return_value(None)
|
|
|
|
def stop(self, *_):
|
|
self.stop_synth.set()
|
|
if self.sound:
|
|
self.sound.terminate()
|
|
sys.exit(0)
|
|
|
|
|
|
def main():
|
|
player = Player(int(sys.argv[1]))
|
|
signal.signal(signal.SIGTERM, player.stop)
|
|
signal.signal(signal.SIGINT, player.stop)
|
|
|
|
node = Gio.DBusNodeInfo.new_for_xml(INTERFACES)
|
|
|
|
def acquired(connection, _name):
|
|
for interface in node.interfaces:
|
|
connection.register_object("/org/mpris/MediaPlayer2", interface, player.call, player.get, None)
|
|
|
|
Gio.bus_own_name(Gio.BusType.SESSION, f"org.mpris.MediaPlayer2.fake{player.index}",
|
|
Gio.BusNameOwnerFlags.NONE, acquired, None, None)
|
|
|
|
# GLib's loop ignores Python signal handlers unless it wakes up now and then
|
|
GLib.timeout_add(250, lambda: True)
|
|
GLib.MainLoop().run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|