docs: README, AGENTS.md, test tools in tools/
- 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
This commit is contained in:
server2
Executable
+256
@@ -0,0 +1,256 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user