Martin Bauer
a7fb56c9fe
Requiring 3.13 meant the device needed an interpreter the distribution does not
have, which is what dragged uv in, and uv then had to be matched to the Pi's
32-bit userland by hand and to build Pillow from source because no armv7 wheel
exists for a 3.13 ABI. Dropping to 3.11 removes all of that: apt provides the
interpreter and piwheels has prebuilt armhf wheels for the native dependencies.
The 3.13-only syntax was shallow - PEP 695 throughout, which converts back
mechanically:
type X = Y -> X: TypeAlias = Y
type Handler[E: Event] -> E = TypeVar("E", bound=Event) plus a plain alias,
which is generic anyway because it carries a TypeVar
def f[T: Bound](...) -> a module-level TypeVar
Also drop the one @override (3.12, and static-only), and stop the lirc test
harness calling Server.close_clients(), which is 3.13: the scripted handler now
releases its connection when asked, which is what that call was there to force.
Verified on 3.11.14 - 485 passed, mypy strict clean - and still 496 passed on
the 3.14 dev venv, which additionally has the analysis extra.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
347 lines
12 KiB
Python
347 lines
12 KiB
Python
"""One vocabulary for driving a simulated mouse, shared by three front-ends.
|
|
|
|
The same verbs are typed at the interactive prompt, listed in a scenario file, and
|
|
called from pytest - so a bug reproduced by hand becomes a regression test by pasting
|
|
the session into a ``.txt`` file.
|
|
|
|
place fuchs
|
|
wait 1s
|
|
press right
|
|
expect track 1
|
|
|
|
Under :class:`~musicmouse.clock.FakeClock` ``wait 1s`` costs microseconds, so scenarios
|
|
are cheap enough to run on every commit.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from typing import TypeVar
|
|
|
|
from musicmouse.app import App
|
|
from musicmouse.clock import Clock, FakeClock
|
|
from musicmouse.events import (
|
|
ButtonEvent,
|
|
NextTrackRequested,
|
|
PauseRequested,
|
|
PlayAlbumRequested,
|
|
PlayRequested,
|
|
PreviousTrackRequested,
|
|
RfidTokenRead,
|
|
RotaryTurned,
|
|
SeekRequested,
|
|
SetVolumeRequested,
|
|
TouchButtonPressed,
|
|
TouchButtonReleased,
|
|
)
|
|
from musicmouse.hardware import (
|
|
NO_FIGURE_TAG,
|
|
Button,
|
|
ButtonAction,
|
|
LedZone,
|
|
RotaryDirection,
|
|
TouchButton,
|
|
)
|
|
from musicmouse.simulator.fake_transport import FakeTransport
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
__all__ = ["ExpectationError", "ScriptError", "SimulatorDriver"]
|
|
|
|
|
|
class ScriptError(Exception):
|
|
"""A scenario line could not be understood."""
|
|
|
|
|
|
class ExpectationError(AssertionError):
|
|
"""An ``expect`` line did not hold."""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Duration:
|
|
seconds: float
|
|
|
|
@classmethod
|
|
def parse(cls, text: str) -> Duration:
|
|
raw = text.strip().lower()
|
|
scale = 1.0
|
|
for suffix, factor in (("ms", 0.001), ("s", 1.0), ("m", 60.0)):
|
|
if raw.endswith(suffix):
|
|
raw = raw.removesuffix(suffix)
|
|
scale = factor
|
|
break
|
|
try:
|
|
return cls(float(raw) * scale)
|
|
except ValueError:
|
|
raise ScriptError(f"{text!r} is not a duration (try '1s', '500ms', '2')") from None
|
|
|
|
|
|
class SimulatorDriver:
|
|
def __init__(self, app: App, transport: FakeTransport, clock: Clock) -> None:
|
|
self.app = app
|
|
self.transport = transport
|
|
self.clock = clock
|
|
|
|
# ------------------------------------------------------------------- inputs
|
|
|
|
async def place(self, figure: str) -> None:
|
|
"""Put a figure on the reader."""
|
|
try:
|
|
tag = self.app.config.figures[figure].id
|
|
except KeyError:
|
|
known = ", ".join(sorted(self.app.config.figures))
|
|
raise ScriptError(f"unknown figure {figure!r} (configured: {known})") from None
|
|
await self.tag(tag)
|
|
|
|
async def tag(self, tag_id: bytes) -> None:
|
|
self.transport.inject(RfidTokenRead(tag_id=tag_id, source="simulator"))
|
|
await self.settle()
|
|
|
|
async def remove(self) -> None:
|
|
"""Take whatever is on the reader off it."""
|
|
await self.tag(NO_FIGURE_TAG)
|
|
|
|
async def press(self, button: str, action: str = "pressed") -> None:
|
|
self.transport.inject(
|
|
ButtonEvent(
|
|
button=_enum_by_name(Button, button, "button"),
|
|
action=_enum_by_name(ButtonAction, action, "button action"),
|
|
source="simulator",
|
|
)
|
|
)
|
|
await self.settle()
|
|
|
|
async def touch(self, button: str) -> None:
|
|
self.transport.inject(
|
|
TouchButtonPressed(
|
|
button=_enum_by_name(TouchButton, button, "touch button"), source="simulator"
|
|
)
|
|
)
|
|
await self.settle()
|
|
|
|
async def release(self, button: str) -> None:
|
|
self.transport.inject(
|
|
TouchButtonReleased(
|
|
button=_enum_by_name(TouchButton, button, "touch button"), source="simulator"
|
|
)
|
|
)
|
|
await self.settle()
|
|
|
|
async def turn(self, steps: int) -> None:
|
|
"""Turn the rotary encoder; negative steps turn it down."""
|
|
self.transport.inject(
|
|
RotaryTurned(
|
|
position=0,
|
|
increment=abs(steps),
|
|
direction=RotaryDirection.UP if steps >= 0 else RotaryDirection.DOWN,
|
|
source="simulator",
|
|
)
|
|
)
|
|
await self.settle()
|
|
|
|
async def disconnect(self) -> None:
|
|
self.transport.disconnect()
|
|
self.app.mouse.on_disconnected("simulated disconnect")
|
|
await self.settle()
|
|
|
|
async def reconnect(self) -> None:
|
|
self.transport.reconnect()
|
|
self.app.mouse.on_connected()
|
|
await self.settle()
|
|
|
|
# ------------------------------------------------------------------ intents
|
|
|
|
async def emit_play(self) -> None:
|
|
self.app.bus.emit(PlayRequested(source="simulator"))
|
|
await self.settle()
|
|
|
|
async def emit_pause(self) -> None:
|
|
self.app.bus.emit(PauseRequested(source="simulator"))
|
|
await self.settle()
|
|
|
|
async def emit_next(self) -> None:
|
|
self.app.bus.emit(NextTrackRequested(source="simulator"))
|
|
await self.settle()
|
|
|
|
async def emit_prev(self) -> None:
|
|
self.app.bus.emit(PreviousTrackRequested(source="simulator"))
|
|
await self.settle()
|
|
|
|
async def set_volume(self, volume: int) -> None:
|
|
self.app.bus.emit(SetVolumeRequested(volume=volume, source="simulator"))
|
|
await self.settle()
|
|
|
|
async def play_album(self, title: str, track_index: int = 0) -> None:
|
|
"""Start a library album by title - what the web front-end does."""
|
|
album = next(
|
|
(a for a in self.app.library.albums if a.title.lower() == title.lower()), None
|
|
)
|
|
if album is None:
|
|
known = ", ".join(sorted(a.title for a in self.app.library.albums))
|
|
raise ScriptError(f"unknown album {title!r} (known: {known})")
|
|
self.app.bus.emit(
|
|
PlayAlbumRequested(
|
|
album_id=album.id, track_index=track_index, source="simulator"
|
|
)
|
|
)
|
|
await self.settle()
|
|
|
|
async def seek(self, position: float) -> None:
|
|
self.app.bus.emit(SeekRequested(position=position, source="simulator"))
|
|
await self.settle()
|
|
|
|
# -------------------------------------------------------------------- time
|
|
|
|
async def wait(self, seconds: float) -> None:
|
|
await self.clock.advance(seconds)
|
|
await self.settle()
|
|
|
|
async def settle(self) -> None:
|
|
"""Let every queued event, and everything it triggers, be handled."""
|
|
await self.app.bus.drain()
|
|
|
|
# ------------------------------------------------------------------ queries
|
|
|
|
def status(self) -> str:
|
|
player = self.app.player
|
|
track = player.current_track
|
|
return (
|
|
f"figure={self.app.mouse.active_figure or '-'} "
|
|
f"{'playing' if player.is_playing else 'paused'} "
|
|
f"track={player.track_index}{f' ({track.title})' if track else ''} "
|
|
f"volume={player.volume} "
|
|
f"buttons={self.app.mouse.button_led_brightness:.2f}"
|
|
)
|
|
|
|
def leds(self) -> str:
|
|
return "\n".join(
|
|
f" {zone:>5}: {self.app.mouse.effect(zone) or '-'}" for zone in LedZone
|
|
)
|
|
|
|
def check(self, key: str, value: str) -> None:
|
|
"""Assert one property. Raises :class:`ExpectationError` if it does not hold."""
|
|
actual = self._lookup(key)
|
|
expected = value.strip()
|
|
if actual != expected:
|
|
raise ExpectationError(f"expected {key} to be {expected!r}, but it is {actual!r}")
|
|
|
|
def _lookup(self, key: str) -> str:
|
|
player = self.app.player
|
|
match key:
|
|
case "playing":
|
|
return "true" if player.is_playing else "false"
|
|
case "figure":
|
|
return self.app.mouse.active_figure or "none"
|
|
case "playlist":
|
|
return player.playlist.name if player.playlist else "none"
|
|
case "album":
|
|
album = self.app.album_for(player.playlist)
|
|
return album.title if album else "none"
|
|
case "position":
|
|
return f"{player.position:.1f}"
|
|
case "track":
|
|
return str(player.track_index)
|
|
case "title":
|
|
track = player.current_track
|
|
return track.title if track else "none"
|
|
case "volume":
|
|
return str(player.volume)
|
|
case "brightness":
|
|
return f"{self.app.mouse.button_led_brightness:.2f}"
|
|
case "ring" | "mouse" | "shelf":
|
|
effect = self.app.mouse.effect(LedZone(key))
|
|
return type(effect).__name__ if effect is not None else "none"
|
|
case _:
|
|
raise ScriptError(
|
|
f"unknown property {key!r} (try: playing, figure, playlist, album, "
|
|
f"track, title, position, volume, brightness, ring, mouse, shelf)"
|
|
)
|
|
|
|
# ------------------------------------------------------------------ scripts
|
|
|
|
async def execute(self, line: str) -> str | None:
|
|
"""Run one scenario line. Returns text to show, if any."""
|
|
stripped = line.split("#", 1)[0].strip()
|
|
if not stripped:
|
|
return None
|
|
verb, *args = stripped.split()
|
|
return await self._dispatch(verb.lower(), args)
|
|
|
|
async def run_script(self, text: str) -> None:
|
|
for number, line in enumerate(text.splitlines(), start=1):
|
|
try:
|
|
if (output := await self.execute(line)) is not None:
|
|
print(output)
|
|
except (ScriptError, ExpectationError) as exc:
|
|
raise type(exc)(f"line {number}: {exc}\n {line.strip()}") from None
|
|
|
|
async def _dispatch(self, verb: str, args: list[str]) -> str | None:
|
|
match verb, args:
|
|
case ("place" | "rfid", [figure]):
|
|
await self.place(figure)
|
|
case ("remove", []):
|
|
await self.remove()
|
|
case ("press", [button]):
|
|
await self.press(button)
|
|
case ("press", [button, action]):
|
|
await self.press(button, action)
|
|
case ("touch", [button]):
|
|
await self.touch(button)
|
|
case ("release", [button]):
|
|
await self.release(button)
|
|
case ("turn", [steps]):
|
|
await self.turn(_int(steps))
|
|
case ("next", []):
|
|
await self.emit_next()
|
|
case ("prev" | "previous", []):
|
|
await self.emit_prev()
|
|
case ("play", []):
|
|
await self.emit_play()
|
|
case ("pause", []):
|
|
await self.emit_pause()
|
|
case ("volume", [level]):
|
|
await self.set_volume(_int(level))
|
|
case ("album", [*words]) if words:
|
|
await self.play_album(" ".join(words))
|
|
case ("seek", [position]):
|
|
await self.seek(Duration.parse(position).seconds)
|
|
case ("disconnect", []):
|
|
await self.disconnect()
|
|
case ("reconnect", []):
|
|
await self.reconnect()
|
|
case ("wait", [duration]):
|
|
await self.wait(Duration.parse(duration).seconds)
|
|
case ("expect", [key, *rest]) if rest:
|
|
self.check(key, " ".join(rest))
|
|
case ("status", []):
|
|
return self.status()
|
|
case ("leds", []):
|
|
return self.leds()
|
|
case _:
|
|
raise ScriptError(f"don't know how to {' '.join([verb, *args])!r}")
|
|
return None
|
|
|
|
|
|
def _int(text: str) -> int:
|
|
try:
|
|
return int(text)
|
|
except ValueError:
|
|
raise ScriptError(f"{text!r} is not a whole number") from None
|
|
|
|
|
|
_EnumT = TypeVar("_EnumT", Button, ButtonAction, TouchButton)
|
|
|
|
|
|
def _enum_by_name(enum: type[_EnumT], name: str, what: str) -> _EnumT:
|
|
try:
|
|
return enum[name.upper()]
|
|
except KeyError:
|
|
options = ", ".join(member.name.lower() for member in enum)
|
|
raise ScriptError(f"unknown {what} {name!r} (try: {options})") from None
|
|
|
|
|
|
def fake_clock_for(app: App) -> FakeClock:
|
|
"""A clock whose ``advance`` also drains the bus, for deterministic scenarios."""
|
|
return FakeClock(idle=app.bus.drain)
|