Web frontend

This commit is contained in:
2026-08-27 12:32:20 +02:00
parent d44c24ec97
commit edb6e5e027
97 changed files with 9535 additions and 195 deletions

Binary file not shown.

After

Width:  |  Height:  |  Size: 207 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 277 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 207 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 650 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 104 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 91 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 567 KiB

File diff suppressed because one or more lines are too long

View File

@@ -1,21 +1,22 @@
# MusicMouse backend
The host application: it reads RFID tags, buttons and touch areas from the ESP32
firmware over serial, plays music through VLC, drives three LED strips, and exposes
everything to Home Assistant over MQTT.
firmware over serial, plays music through VLC, drives three LED strips, indexes the
music collection, and exposes everything to Home Assistant over MQTT and to a browser
over HTTP.
```
ESP32 ⇄ MusicMouseDevice ─┐ ┌─► MqttService (state out, intents in)
ESP32 ⇄ MusicMouseDevice ─┐ ┌─► MqttService (state out, intents in)
VLC ⇄ VlcPlayer ────────┼──► EventBus ──────────►┤
broker ⇄ MqttService ────── ▲ └─► (a web service would go here)
broker ⇄ MqttService ────── ▲ └─► WebService (state out, intents in)
browser ⇄ WebService ───────┘
reactions/*.py ── call actions on ──► device / player
```
Three objects own the outside world, one bus carries everything, and the *reactions*
are the only place that decides what should happen. Adding a new way to control the
mouse means adding a service that emits the same intents - no device or reaction
changes.
changes. The web front-end was added exactly that way.
## Running it
@@ -28,6 +29,32 @@ See `config.yml.example` for the schema and `musicmouse.service` for the systemd
Config problems are reported all at once with the path to each one; unknown keys are
errors, not silent no-ops.
### On a host with no mouse attached
The web front-end is a complete way to drive the player, so the backend is useful on a
machine with no serial port and no sound card worth grabbing. Two config keys say so,
each by taking the literal value `simulate`:
| Setting | `simulate` gives you | Warned about as |
|---|---|---|
| `serial_port` | no serial link; RFID, buttons and LEDs are inert | "running without the mouse" |
| `alsa_device` | the simulator's player: everything works, nothing is audible | "nothing will be audible" |
Both keys are **required**. Leaving one out is a config error, not a shortcut to
simulation - running blind or silent has to be asked for, so a config that lost a line
fails loudly instead of booting into something that looks like it is working. The
warnings are repeated on every boot for the same reason.
`--no-hardware` forces the serial half regardless of the config, for a one-off run:
```sh
python -m musicmouse --config ./config.yml --no-hardware
```
Note that `alsa_device` has no "just use whatever" value. VLC's own default would seize
whatever the desktop is playing through, which is the wrong thing to do silently - name
a device (`"default"` is the system one) when the machine is meant to make noise.
### Without hardware
The simulator runs the entire app - real bus, real device, real reactions, real MQTT
@@ -65,7 +92,9 @@ the prompt becomes a regression test by pasting it into a `.txt` file.
| `musicmouse/events.py` | The vocabulary: **input** (something happened), **intent** (something was requested), **state** (something changed). |
| `musicmouse/devices/` | `mouse.py` (firmware), `player.py` (VLC), `wire.py` (pure codec), `serial_link.py` (transport + reconnect). |
| `musicmouse/reactions/` | The policy. `@on(SomeEvent)` functions that get the app and act. |
| `musicmouse/library/` | The music collection: scanning, tags, cover art and its colours, the cache, and the seams for future track analysis. |
| `musicmouse/services/mqtt/` | Home Assistant entities: three lights, a player sensor, a volume number, transport buttons, device triggers, a tag scanner. |
| `musicmouse/services/web/` | The browser front-end's API: the library, a state websocket, command endpoints, and parent-mode settings. |
| `musicmouse/simulator/` | Fake transport and player, the driver vocabulary, the REPL and the script runner. |
| `musicmouse/config.py` | Pydantic schema, validation, and human-readable error formatting. |
@@ -118,3 +147,89 @@ trigger, action `light.toggle` on `light.kinderzimmer_fluter`.
Trigger on the corresponding `*_touched` device trigger and call `light.turn_on` with
that data.
## The library
`general.library.root` is the one path to the music. The shelves under it are fixed
names rather than settings (`musicmouse/library/sections.py`), because each has quirks
the scanner has to know about:
| Folder | Shown as | Why it is special |
|---|---|---|
| `Figuren/<figure>/` | an album per figurine | the folder name *is* the figure name from the config |
| `Musik/<Artist> - <Album>/` | music, grouped by artist | plain ID3 |
| `Hörbücher/<Artist> - <Album>/` | audiobooks, grouped by character | `album_artist` is a credit list; only the name before the first comma groups usefully |
| `Kinderpodcasts/<Show>/` | audiobooks, newest episode first | the tags are useless here - `artist` is the presenter list and `album` is the feed name, so the *folder* is the show |
Only files whose suffix is in `audio_extensions` are read, and dotfiles are skipped, so
a podcast downloader's `archive.json` and its half-finished `.download.tmp` never reach
a playlist.
Each album carries three colours, pulled out of its cover art with Pillow (or
synthesised from a hash of its id when it has none). The frontend paints cards with
them and the LED strips run the first of them, so shelf and screen agree.
### The cache
`general.library.cache` is a directory, not a file, because its contents cost wildly
different amounts to produce:
```
index.json cheap: tags and structure. Rebuilt freely.
covers/<album_id>.jpg medium: art extracted from an ID3 APIC frame
analysis/<track_key>.json expensive: reserved for offline audio analysis
```
Deleting the whole directory is safe; deleting it throws away analysis that is minutes
of DSP per track, which is why anything expensive is keyed by *file content* rather than
by album id - renaming a folder or re-sorting a section then costs nothing.
An entry is reused whenever its files' sizes and mtimes are unchanged. That means a
change to how the scanner derives a title, artist or series is invisible until the cache
is invalidated: bump `_INDEX_VERSION` in `musicmouse/library/cache.py` when you touch
that logic.
**Track analysis is not implemented.** `musicmouse/library/analysis.py` fixes the shape
of the results - scalars (`tempo`, `energy`, `valence`, `brightness`) travel inline with
the index, and a beat grid lives in its own file and is fetched per track - so an
analyzer can be added later without touching the scanner, the API or the frontend. It
will go behind an optional dependency group, and because results are content-keyed
files, they can equally well be computed on a workstation and the `analysis/` folder
copied to the device.
## The web front-end
`web/` in the repo root, served by this backend when `general.web.static_dir` is set.
| Route | Purpose |
|---|---|
| `GET /api/library` | Every album with its tracks and colours. ~90 kB, sent once. |
| `GET /api/albums/{id}/cover` | The cover, or 404 - the client paints the album's colours instead. |
| `POST /api/library/refresh` | Rescan in the background; connected clients are told when it lands. |
| `GET /api/state` | Snapshot: what is playing, where, how loud. |
| `WS /api/ws` | Push only. A snapshot on connect, then a frame per change, plus the position at 2 Hz while playing. |
| `POST /api/play` | `{album_id, track_index?}` |
| `POST /api/resume` `/pause` `/next` `/previous` `/seek` | transport |
| `POST /api/volume` | `{percent}` or `{delta_percent}` |
| `GET` `PUT /api/settings` | parent mode |
Two decisions worth knowing:
**Search is not an endpoint.** The whole index goes to the browser and filtering happens
there, which is what makes the design's type-to-search feel instant.
**Volume is a percentage at this boundary.** `max_volume` is a parent's business, not a
child's, so it never crosses into the browser: `100 %` means whatever ceiling is
configured, and the mapping lives in `services/web/settings.py` so MQTT, the rotary
encoder and the firmware carry on in device units.
### Parent mode
`?parentMode=1` reveals a settings panel for the volume limits, the rotary step and the
button brightness. Saving writes `config.yml` back through ruamel's round-trip loader,
so the file keeps its comments, and a new ceiling applies to the running player rather
than waiting for a restart.
This **hides** the settings; it does not protect them. There is no authentication on any
endpoint, which matches a device on a home network - put it behind a reverse proxy if
that is not good enough.

94
python-backend/config.yml Normal file
View File

@@ -0,0 +1,94 @@
# Example config for the MusicMouse backend.
#
# python -m musicmouse --config /media/musicmouse/config.yml
#
# Unknown keys are rejected rather than ignored, and every problem in the file is
# reported at once, so a typo fails at startup with the path to the offending line.
# Keep the real config (with credentials) off the repo - on the device only.
general:
# The music collection. One path; the shelves underneath it are fixed names, not
# settings, because each one has its own quirks the code already knows about:
#
# <root>/Figuren/<figure name>/ one folder per figurine
# <root>/Musik/<Artist> - <Album>/ albums, grouped by artist
# <root>/Hörbücher/<Artist> - <Album>/ audiobooks, grouped by character
# <root>/Kinderpodcasts/<Show>/ shows, newest episode first
#
# A cover.jpg next to the audio is used if present, otherwise the art is pulled out
# of the files' tags. Relative paths resolve against this file's directory.
library:
root: /home/martin/Music
# Scan results, extracted cover art and track analysis. Safe to delete: the index
# is rebuilt on the next start. Deleting it does throw away track analysis, which
# is expensive to recompute.
cache: .musicmouse-cache
# Serial port the ESP32 firmware is on. A dropped link is retried, not fatal.
# Required - use "simulate" to run without the mouse attached, which is a complete
# setup on its own because the web front-end can drive the player by itself. RFID,
# buttons and LEDs then do nothing, and startup says so every boot.
serial_port: "simulate"
baudrate: 115200
reconnect_interval: 5.0
# ALSA output device passed to VLC, e.g. "hw:0,0", or "default" for the system
# default output. Required - use "simulate" for a player that makes no sound, which
# is handy when working on the web UI on a machine whose audio you would rather not
# commandeer. Startup says so every boot.
#
# Both of these are required rather than optional on purpose: running blind or silent
# has to be asked for, so a config that lost a line fails loudly instead of booting
# into something that looks like it is working.
alsa_device: "default"
# Volume, 0..100. min/max clamp everything, including the rotary encoder.
min_volume: 0
max_volume: 80
initial_volume: 40
volume_increment: 5 # per rotary-encoder click
# Backlight of the prev/next buttons while a figure is playing, 0..1.
button_leds_brightness: 0.5
# Which files count as music. Anything else - a podcast downloader's archive.json,
# a half-finished .tmp - is ignored.
audio_extensions: [".mp3", ".ogg", ".oga", ".opus", ".flac", ".wav", ".m4a", ".aac"]
# The web front-end. Omit the whole section to run without it.
#
# There is no authentication: this is a device on a home network. The settings panel
# at ?parentMode=1 is hidden from the child, not protected from them - it writes back
# to this file. Put it behind a reverse proxy if that is not good enough.
web:
host: "0.0.0.0"
port: 8080
# Built frontend to serve at /. Omit to expose only the JSON API.
static_dir: ../web/dist
# Home Assistant integration. Omit the whole section to run without MQTT.
# The backend exposes three lights, a player sensor, a volume slider, transport
# buttons, device triggers for every button/touch area, and a tag scanner.
#mqtt:
# server: "homeassistant.local"
# port: 1883
# user: "musicmouse"
# password: "REPLACE_WITH_MQTT_PASSWORD"
# base_topic: "musicmouse"
# discovery_prefix: "homeassistant"
# device_id: "musicmouse"
# device_name: "Music Mouse"
# reconnect_interval: 10.0
# One entry per figurine. The key is the figure name and the subfolder name.
figures:
fuchs:
# RFID tag id, 5 bytes as hex. Must be unique across figures.
id: "04a1b2c3d4"
# Exactly four colours: primary, secondary, background, accent.
# Either "#rrggbb" (RGB) or "wNN" (white channel only, hex).
colors: ["#ff6600", "#ffcc00", "#331100", "wff"]
eule:
id: "04b2c3d4e5"
colors: ["#3355ff", "#66aaff", "#001133", "#ffffff"]

View File

@@ -7,17 +7,39 @@
# Keep the real config (with credentials) off the repo - on the device only.
general:
# Root folder holding one subfolder per figure. Relative paths resolve against
# this file's directory. Each figure plays <figure_folder>/<figure name>,
# in alphabetical order by filename.
figure_folder: music
# The music collection. One path; the shelves underneath it are fixed names, not
# settings, because each one has its own quirks the code already knows about:
#
# <root>/Figuren/<figure name>/ one folder per figurine
# <root>/Musik/<Artist> - <Album>/ albums, grouped by artist
# <root>/Hörbücher/<Artist> - <Album>/ audiobooks, grouped by character
# <root>/Kinderpodcasts/<Show>/ shows, newest episode first
#
# A cover.jpg next to the audio is used if present, otherwise the art is pulled out
# of the files' tags. Relative paths resolve against this file's directory.
library:
root: /home/martin/Music
# Scan results, extracted cover art and track analysis. Safe to delete: the index
# is rebuilt on the next start. Deleting it does throw away track analysis, which
# is expensive to recompute.
cache: .musicmouse-cache
# Serial port the ESP32 firmware is on. A dropped link is retried, not fatal.
# Required - use "simulate" to run without the mouse attached, which is a complete
# setup on its own because the web front-end can drive the player by itself. RFID,
# buttons and LEDs then do nothing, and startup says so every boot.
serial_port: "/dev/ttyUSB0"
baudrate: 115200
reconnect_interval: 5.0
# ALSA output device passed to VLC, e.g. "hw:0,0". Omit for VLC's default.
# ALSA output device passed to VLC, e.g. "hw:0,0", or "default" for the system
# default output. Required - use "simulate" for a player that makes no sound, which
# is handy when working on the web UI on a machine whose audio you would rather not
# commandeer. Startup says so every boot.
#
# Both of these are required rather than optional on purpose: running blind or silent
# has to be asked for, so a config that lost a line fails loudly instead of booting
# into something that looks like it is working.
alsa_device: "softvol_effects"
# Volume, 0..100. min/max clamp everything, including the rotary encoder.
@@ -29,9 +51,21 @@ general:
# Backlight of the prev/next buttons while a figure is playing, 0..1.
button_leds_brightness: 0.5
# Which files count as music. Anything else in a figure folder is ignored.
# Which files count as music. Anything else - a podcast downloader's archive.json,
# a half-finished .tmp - is ignored.
audio_extensions: [".mp3", ".ogg", ".oga", ".opus", ".flac", ".wav", ".m4a", ".aac"]
# The web front-end. Omit the whole section to run without it.
#
# There is no authentication: this is a device on a home network. The settings panel
# at ?parentMode=1 is hidden from the child, not protected from them - it writes back
# to this file. Put it behind a reverse proxy if that is not good enough.
web:
host: "0.0.0.0"
port: 8080
# Built frontend to serve at /. Omit to expose only the JSON API.
static_dir: ../web/dist
# Home Assistant integration. Omit the whole section to run without MQTT.
# The backend exposes three lights, a player sensor, a volume slider, transport
# buttons, device triggers for every button/touch area, and a tag scanner.
@@ -54,7 +88,13 @@ figures:
# Exactly four colours: primary, secondary, background, accent.
# Either "#rrggbb" (RGB) or "wNN" (white channel only, hex).
colors: ["#ff6600", "#ffcc00", "#331100", "wff"]
# "music" (default) or "book". Every other shelf is named after what is on it, so
# its type is obvious; a figure folder is named after the figurine, so this is the
# one thing that has to be said out loud. The web UI draws albums square and
# audiobooks taller than wide, so getting it wrong is visible at a glance.
kind: music
eule:
id: "04b2c3d4e5"
colors: ["#3355ff", "#66aaff", "#001133", "#ffffff"]
kind: book

View File

@@ -1,12 +1,13 @@
"""Entry point and composition root.
python -m musicmouse --config /media/musicmouse/config.yml
python -m musicmouse --config ./config.yml --no-hardware
python -m musicmouse --config ./config.yml --simulate
python -m musicmouse --config ./config.yml --simulate --script scenarios/smoke.txt
This is the only module that knows which concrete implementations are in play; the
difference between "real mouse" and "simulated mouse" is which transport and which
player get built here.
difference between "real mouse", "no mouse attached" and "simulated mouse" is which
transport and which player get built here.
"""
from __future__ import annotations
@@ -24,13 +25,16 @@ from musicmouse import __version__
from musicmouse.app import App
from musicmouse.bus import EventBus
from musicmouse.clock import RealClock
from musicmouse.config import Config, ConfigError, build_playlists, load_config
from musicmouse.config import SIMULATE, Config, ConfigError, GeneralConfig, load_config
from musicmouse.devices.mouse import MusicMouseDevice
from musicmouse.devices.null_transport import NullTransport
from musicmouse.devices.player import Player, VlcPlayer
from musicmouse.devices.serial_link import SerialLink
from musicmouse.library import MusicLibrary
from musicmouse.reactions import register_all
from musicmouse.services.base import Service
from musicmouse.services.mqtt import MqttService, build_entities
from musicmouse.services.web import WebService
_log = logging.getLogger("musicmouse")
@@ -48,6 +52,12 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
action="store_true",
help="run against fake hardware and a fake player (no serial port, no audio)",
)
parser.add_argument(
"--no-hardware",
action="store_true",
help="real audio and a real web front-end, but no serial port: for a host with "
"no mouse attached",
)
parser.add_argument(
"--script",
type=Path,
@@ -64,6 +74,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
args = parser.parse_args(argv)
if args.script and not args.simulate:
parser.error("--script only makes sense together with --simulate")
if args.simulate and args.no_hardware:
parser.error("--simulate already runs without hardware; drop --no-hardware")
return args
@@ -85,7 +97,11 @@ def main(argv: list[str] | None = None) -> int:
print(f"error: {exc}", file=sys.stderr)
return 2
runner = run_simulated(config, args.script) if args.simulate else run_real(config)
runner = (
run_simulated(config, args.config, args.script)
if args.simulate
else run_real(config, args.config, hardware=wants_hardware(config, args.no_hardware))
)
try:
asyncio.run(runner)
except KeyboardInterrupt:
@@ -96,43 +112,53 @@ def main(argv: list[str] | None = None) -> int:
# ------------------------------------------------------------------------- real
async def run_real(config: Config) -> None:
async def run_real(config: Config, config_path: Path, *, hardware: bool = True) -> None:
bus = EventBus()
await bus.start()
clock = RealClock()
general = config.general
link = SerialLink(
general.serial_port,
general.baudrate,
reconnect_interval=general.reconnect_interval,
clock=clock,
)
mouse = MusicMouseDevice(bus, link, config.tag_map, port=general.serial_port)
link.attach(
mouse.feed, on_connect=mouse.on_connected, on_disconnect=mouse.on_disconnected
)
link: SerialLink | None = None
if hardware:
link = SerialLink(
general.serial_port,
general.baudrate,
reconnect_interval=general.reconnect_interval,
clock=clock,
)
mouse = MusicMouseDevice(bus, link, config.tag_map, port=general.serial_port)
link.attach(
mouse.feed, on_connect=mouse.on_connected, on_disconnect=mouse.on_disconnected
)
else:
mouse = MusicMouseDevice(bus, NullTransport(), config.tag_map, port="none")
player: Player = VlcPlayer(
bus,
alsa_device=general.alsa_device,
clock=clock,
**VlcPlayer.volume_kwargs(general),
)
player = _build_player(bus, general, clock=clock)
app = _build_app(config, bus, mouse, player, clock=clock)
services = _build_services(app, mouse, player, clock=clock)
library = await build_library(config)
app = _build_app(config, bus, mouse, player, library, clock=clock)
services = _build_services(
app, mouse, player, clock=clock, config_path=config_path
)
_log.info(
"MusicMouse %s starting: %d figures, serial %s, mqtt %s",
"MusicMouse %s starting: %d figures, %d albums, serial %s, audio %s, mqtt %s",
__version__,
len(config.figures),
general.serial_port,
len(library.albums),
# Report what the config says, with a marker when --no-hardware overrode it,
# so the line never disagrees with the file it was started from.
general.serial_port + ("" if link or general.serial_simulated else " (no link)"),
general.alsa_device,
general.mqtt.server if general.mqtt else "disabled",
)
try:
await _run_forever(
[link.run(), player.run(), *(service.run() for service in services)]
[
*([link.run()] if link else []),
player.run(),
*(service.run() for service in services),
]
)
finally:
player.close()
@@ -142,7 +168,7 @@ async def run_real(config: Config) -> None:
# -------------------------------------------------------------------- simulated
async def run_simulated(config: Config, script: Path | None) -> None:
async def run_simulated(config: Config, config_path: Path, script: Path | None) -> None:
# Imported here so the production path never touches the simulator.
from musicmouse.simulator.harness import build_simulation
from musicmouse.simulator.repl import run_repl
@@ -154,7 +180,9 @@ async def run_simulated(config: Config, script: Path | None) -> None:
config, clock=RealClock() if script is None else None, track_duration=5.0
)
services = _build_services(sim.app, sim.app.mouse, sim.player, clock=RealClock())
services = _build_services(
sim.app, sim.app.mouse, sim.player, clock=RealClock(), config_path=config_path
)
tasks = [asyncio.create_task(service.run(), name=service.name) for service in services]
try:
if script is not None:
@@ -170,11 +198,68 @@ async def run_simulated(config: Config, script: Path | None) -> None:
# ---------------------------------------------------------------------- wiring
def wants_hardware(config: Config, no_hardware_flag: bool) -> bool:
"""Whether to open a serial port at all.
``serial_port: simulate`` says the same thing ``--no-hardware`` does. It is asked
for rather than inferred from a missing key, but it is still worth saying out loud
every boot - a mouse whose RFID reader does nothing should say why.
"""
if no_hardware_flag:
return False
if config.general.serial_simulated:
_log.warning(
'general.serial_port is "%s": running without the mouse. '
"The web front-end still works; RFID, buttons and LEDs do not.",
SIMULATE,
)
return False
return True
def _build_player(bus: EventBus, general: GeneralConfig, *, clock: RealClock) -> Player:
"""A real player, or a silent one when the config asked for that.
``alsa_device: simulate`` gets the simulator's player: everything above it behaves
identically, it just makes no sound. Useful for working on the web UI without
commandeering the machine's audio.
"""
if not general.audio_simulated:
return VlcPlayer(
bus,
alsa_device=general.alsa_device,
clock=clock,
**VlcPlayer.volume_kwargs(general),
)
_log.warning(
'general.alsa_device is "%s": using a simulated player, so nothing will be '
'audible. Set it to "default" for the system default output.',
SIMULATE,
)
# Imported here rather than at module scope so the real audio path never loads the
# simulator - and so this still runs on a machine without libVLC at all.
from musicmouse.simulator.fake_player import FakePlayer
return FakePlayer(bus, clock=clock, **FakePlayer.volume_kwargs(general))
async def build_library(config: Config) -> MusicLibrary:
library_config = config.general.library
return await MusicLibrary.build(
library_config.root,
library_config.cache,
frozenset(config.general.audio_extensions),
figure_kinds=config.figure_kinds,
)
def _build_app(
config: Config,
bus: EventBus,
mouse: MusicMouseDevice,
player: Player,
library: MusicLibrary,
*,
clock: RealClock,
) -> App:
@@ -183,7 +268,11 @@ def _build_app(
bus=bus,
mouse=mouse,
player=player,
playlists=build_playlists(config),
library=library,
# One source of truth: the figure path and the web path must hand the player the
# same Playlist object for the same folder, because `play_figure` resumes on an
# identity check.
playlists=library.figure_playlists(),
clock=clock,
)
register_all(bus, app)
@@ -191,15 +280,30 @@ def _build_app(
def _build_services(
app: App, mouse: MusicMouseDevice, player: Player, *, clock: RealClock
app: App,
mouse: MusicMouseDevice,
player: Player,
*,
clock: RealClock,
config_path: Path,
) -> list[Service]:
"""Every front-end. A web service would be one more line here."""
if app.config.general.mqtt is None:
_log.info("No mqtt section in the config: Home Assistant integration is off")
return []
"""Every front-end. Each one only speaks intents, so they cannot conflict."""
services: list[Service] = []
mqtt_config = app.config.general.mqtt
entities = build_entities(app.bus, mqtt_config, mouse, player)
return [MqttService(app.bus, mqtt_config, entities, clock=clock)]
if mqtt_config is None:
_log.info("No mqtt section in the config: Home Assistant integration is off")
else:
entities = build_entities(app.bus, mqtt_config, mouse, player)
services.append(MqttService(app.bus, mqtt_config, entities, clock=clock))
web_config = app.config.general.web
if web_config is None:
_log.info("No web section in the config: the web front-end is off")
else:
services.append(WebService(app, web_config, config_path))
return services
async def _run_forever(coroutines: list[Coroutine[Any, Any, None]]) -> None:

View File

@@ -10,6 +10,7 @@ from musicmouse.clock import Clock, RealClock
from musicmouse.config import Config, FigureColors
from musicmouse.devices.mouse import MusicMouseDevice
from musicmouse.devices.player import Player
from musicmouse.library import MusicLibrary
from musicmouse.media import Playlist
_log = logging.getLogger(__name__)
@@ -25,6 +26,10 @@ class AppState:
#: instead of starting over. Cleared once its playlist runs out.
last_partially_played_figure: str | None = None
#: Whether the MQTT broker is currently reachable. The firmware's equivalent is
#: readable off the transport; a broker's is not, so it is remembered here.
mqtt_connected: bool = False
@dataclass
class App:
@@ -32,6 +37,7 @@ class App:
bus: EventBus
mouse: MusicMouseDevice
player: Player
library: MusicLibrary
playlists: dict[str, Playlist]
clock: Clock = field(default_factory=RealClock)
state: AppState = field(default_factory=AppState)

View File

@@ -9,7 +9,7 @@ from __future__ import annotations
import logging
from pathlib import Path
from typing import Annotated, Any, Self
from typing import Annotated, Any, Final, Literal, Self
from pydantic import (
BaseModel,
@@ -26,24 +26,31 @@ from ruamel.yaml.error import YAMLError
from musicmouse.color import ColorRGBW, parse_color
from musicmouse.hardware import NO_FIGURE_TAG, RFID_TAG_LENGTH
from musicmouse.media import Playlist, build_playlist
_log = logging.getLogger(__name__)
__all__ = [
"SIMULATE",
"Config",
"ConfigError",
"FigureColors",
"FigureConfig",
"GeneralConfig",
"LibraryConfig",
"MqttConfig",
"build_playlists",
"WebConfig",
"format_validation_error",
"load_config",
]
DEFAULT_AUDIO_EXTENSIONS = (".mp3", ".ogg", ".oga", ".opus", ".flac", ".wav", ".m4a", ".aac")
#: Stand-in value for ``serial_port`` and ``alsa_device``. Running without the mouse or
#: without sound is a supported setup, but it has to be *asked for*: a missing key is an
#: error, so a config that lost a line fails loudly instead of booting into a silent
#: mouse that looks like it is working.
SIMULATE: Final = "simulate"
class ConfigError(Exception):
"""Raised with an already human-readable, multi-line message."""
@@ -101,6 +108,17 @@ class FigureColors(_Strict):
return data
def _resolve_folder(folder: Path, info: ValidationInfo, *, must_exist: bool) -> Path:
"""Make a configured path absolute against the config file, and optionally check it."""
context = info.context or {}
base = context.get("config_dir")
if base is not None and not folder.is_absolute():
folder = (Path(base) / folder).resolve()
if must_exist and context.get("check_paths", True) and not folder.is_dir():
raise ValueError(f"no such directory: {folder}")
return folder
class MqttConfig(_Strict):
server: str
port: int = Field(default=1883, ge=1, le=65535)
@@ -113,18 +131,63 @@ class MqttConfig(_Strict):
reconnect_interval: float = Field(default=10.0, gt=0)
class GeneralConfig(_Strict):
#: Root folder holding one subfolder per figure.
figure_folder: Path
class LibraryConfig(_Strict):
"""Where the music lives.
serial_port: str = "/dev/ttyUSB0"
One path. The shelves underneath it - ``Figuren``, ``Musik``, ``Hoerbuecher``,
``Kinderpodcasts`` - are fixed names, not settings; see
:mod:`musicmouse.library.sections`.
"""
root: Path
#: Scan results, extracted cover art and track analysis. Relative to this file.
cache: Path = Path(".musicmouse-cache")
@field_validator("root")
@classmethod
def _resolve_root(cls, folder: Path, info: ValidationInfo) -> Path:
return _resolve_folder(folder, info, must_exist=True)
@field_validator("cache")
@classmethod
def _resolve_cache(cls, folder: Path, info: ValidationInfo) -> Path:
return _resolve_folder(folder, info, must_exist=False)
@property
def figure_folder(self) -> Path:
return self.root / "Figuren"
class WebConfig(_Strict):
"""The web front-end. Omit the whole section to run without it."""
#: A LAN appliance with no auth; binding to all interfaces is the point.
host: str = "0.0.0.0"
port: int = Field(default=8080, ge=1, le=65535)
#: Built frontend to serve at ``/``. Omit to expose only the JSON API.
static_dir: Path | None = None
@field_validator("static_dir")
@classmethod
def _resolve_static(cls, folder: Path | None, info: ValidationInfo) -> Path | None:
return None if folder is None else _resolve_folder(folder, info, must_exist=False)
class GeneralConfig(_Strict):
library: LibraryConfig
#: Serial port the ESP32 is on, or ``"simulate"`` to run without the mouse: the
#: web front-end is a complete way to drive the player on its own. Required.
serial_port: str
baudrate: int = Field(default=115200, gt=0)
reconnect_interval: float = Field(default=5.0, gt=0)
#: ALSA output device passed to VLC, e.g. "hw:0,0"; null for VLC's default.
alsa_device: str | None = None
#: ALSA output device passed to VLC, e.g. ``"hw:0,0"`` or ``"default"``, or
#: ``"simulate"`` for a player that makes no sound. Required.
alsa_device: str
mqtt: MqttConfig | None = None
web: WebConfig | None = None
min_volume: int = Field(default=0, ge=0, le=200)
max_volume: int = Field(default=100, ge=0, le=200)
@@ -134,6 +197,14 @@ class GeneralConfig(_Strict):
audio_extensions: tuple[str, ...] = DEFAULT_AUDIO_EXTENSIONS
@property
def serial_simulated(self) -> bool:
return self.serial_port == SIMULATE
@property
def audio_simulated(self) -> bool:
return self.alsa_device == SIMULATE
@model_validator(mode="after")
def _check_volumes(self) -> Self:
if self.min_volume > self.max_volume:
@@ -147,22 +218,15 @@ class GeneralConfig(_Strict):
)
return self
@field_validator("figure_folder")
@classmethod
def _resolve_figure_folder(cls, folder: Path, info: ValidationInfo) -> Path:
context = info.context or {}
base = context.get("config_dir")
if base is not None and not folder.is_absolute():
folder = (Path(base) / folder).resolve()
if context.get("check_paths", True) and not folder.is_dir():
raise ValueError(f"no such directory: {folder}")
return folder
class FigureConfig(_Strict):
#: RFID tag id as hex, e.g. "04a1b2c3d4".
id: TagId
colors: FigureColors
#: What this figure holds. Unlike the other shelves a figure folder is named after
#: the figurine rather than its contents, so nothing on disk says whether it is an
#: album or an audiobook - and the browse view draws the two differently.
kind: Literal["music", "book"] = "music"
class Config(_Strict):
@@ -185,16 +249,13 @@ class Config(_Strict):
"""Tag id -> figure name, as handed to the device."""
return {figure.id: name for name, figure in self.figures.items()}
@property
def figure_kinds(self) -> dict[str, Literal["music", "book"]]:
"""Figure name -> what it holds, as handed to the library scanner."""
return {name: figure.kind for name, figure in self.figures.items()}
def folder_for(self, figure: str) -> Path:
return self.general.figure_folder / figure
def build_playlists(config: Config) -> dict[str, Playlist]:
"""One playlist per figure, from ``<figure_folder>/<figure_name>``, alphabetically."""
return {
name: build_playlist(name, config.folder_for(name), config.general.audio_extensions)
for name in config.figures
}
return self.general.library.figure_folder / figure
def format_validation_error(error: ValidationError) -> str:
@@ -209,6 +270,8 @@ def format_validation_error(error: ValidationError) -> str:
message = message.removeprefix(prefix)
if entry["type"] == "extra_forbidden":
message = "unknown option (check the spelling against config.yml.example)"
elif entry["type"] == "missing":
message = "required (see config.yml.example)"
lines.append(f" {location or '<root>'}: {message}")
plural = "s" if len(lines) != 1 else ""
return f"{len(lines)} problem{plural} in the config file:\n" + "\n".join(lines)

View File

@@ -0,0 +1,22 @@
"""A transport with nothing on the other end.
The web front-end is a complete way to drive the mouse, so the backend has to be
useful on a machine that has no mouse attached - a spare Pi, a laptop, a container.
Rather than making :class:`~musicmouse.devices.mouse.MusicMouseDevice` optional and
teaching every reaction to cope with its absence, the device is built as usual against
a transport that drops what it is handed. Lighting reactions still run; their bytes go
nowhere.
"""
from __future__ import annotations
__all__ = ["NullTransport"]
class NullTransport:
def write(self, data: bytes) -> None:
pass
@property
def connected(self) -> bool:
return False

View File

@@ -43,16 +43,33 @@ class Player(Protocol):
def track_index(self) -> int: ...
@property
def current_track(self) -> Track | None: ...
@property
def position(self) -> float:
"""Seconds into the current track. ``0.0`` when nothing is loaded.
Read on demand rather than announced: a progress bar wants this twice a second,
and an event at that rate would flood the bus, the MQTT service and the log for
the benefit of one front-end.
"""
...
@property
def duration(self) -> float:
"""Length of the current track in seconds, or ``0.0`` when unknown."""
...
def set_playlist(self, playlist: Playlist) -> None: ...
def play(self) -> None: ...
def play_from_start(self) -> None: ...
def play_track(self, index: int) -> None: ...
def pause(self) -> None: ...
def stop(self) -> None: ...
def next_track(self) -> None: ...
def previous_track(self) -> None: ...
def seek(self, position: float) -> None: ...
def set_volume(self, volume: int, *, source: EventSource = "system") -> None: ...
def change_volume(self, delta: int, *, source: EventSource = "system") -> None: ...
def set_volume_limits(self, minimum: int, maximum: int) -> None: ...
async def run(self) -> None:
"""Long-running task, if the implementation needs one."""
@@ -113,6 +130,32 @@ class PlayerBase:
return None
return self._playlist[self._index]
# ------------------------------------------------------------------ actions
def set_volume(self, volume: int, *, source: EventSource = "system") -> None:
clamped = self._clamp(volume)
if clamped == self._volume:
return
self._volume = clamped
self._apply_volume(clamped)
self._announce_volume(source)
def change_volume(self, delta: int, *, source: EventSource = "system") -> None:
self.set_volume(self._volume + delta, source=source)
def set_volume_limits(self, minimum: int, maximum: int) -> None:
"""Re-clamp to a new allowed range, and pull the current volume into it.
Parent mode edits these while the mouse is playing, so they cannot only be
constructor arguments.
"""
self._min_volume = minimum
self._max_volume = maximum
self.set_volume(self._volume)
def _apply_volume(self, volume: int) -> None:
"""Push the new volume at whatever actually makes sound. No-op by default."""
# ---------------------------------------------------------------- internals
def _clamp(self, volume: int) -> int:
@@ -181,6 +224,17 @@ class VlcPlayer(PlayerBase):
self._attach_events()
self._media_player.audio_set_volume(self._volume)
# -------------------------------------------------------------------- state
@property
def position(self) -> float:
# libVLC reports -1 for both of these until a media is actually opened.
return max(0.0, float(self._media_player.get_time()) / 1000)
@property
def duration(self) -> float:
return max(0.0, float(self._media_player.get_length()) / 1000)
# ------------------------------------------------------------------ actions
def set_playlist(self, playlist: Playlist) -> None:
@@ -201,10 +255,13 @@ class VlcPlayer(PlayerBase):
self._list_player.play()
def play_from_start(self) -> None:
self.play_track(0)
def play_track(self, index: int) -> None:
if self._playlist is None or not self._playlist:
_log.warning("Nothing to play: the playlist is empty")
return
self._list_player.play_item_at_index(0)
self._list_player.play_item_at_index(max(0, min(index, len(self._playlist) - 1)))
def pause(self) -> None:
self._media_player.set_pause(1)
@@ -218,16 +275,11 @@ class VlcPlayer(PlayerBase):
def previous_track(self) -> None:
self._list_player.previous()
def set_volume(self, volume: int, *, source: EventSource = "system") -> None:
clamped = self._clamp(volume)
if clamped == self._volume:
return
self._volume = clamped
self._media_player.audio_set_volume(clamped)
self._announce_volume(source)
def seek(self, position: float) -> None:
self._media_player.set_time(int(max(0.0, position) * 1000))
def change_volume(self, delta: int, *, source: EventSource = "system") -> None:
self.set_volume(self._volume + delta, source=source)
def _apply_volume(self, volume: int) -> None:
self._media_player.audio_set_volume(volume)
async def run(self) -> None:
"""Poll for state libVLC does not reliably report by event."""
@@ -249,6 +301,21 @@ class VlcPlayer(PlayerBase):
self._volume = volume
self._announce_volume()
self._set_playing(bool(self._list_player.is_playing()))
self._sync_index()
def _sync_index(self) -> None:
"""Read which track is playing back off libVLC.
Asking the media player which media it holds works on every build; the
``MediaListPlayerNextItemSet`` payload does not - on some it arrives as a bare
int rather than a Media, and the index would then never move off zero.
"""
media = self._media_player.get_media()
if media is None:
return
index = self._mrl_to_index.get(media.get_mrl())
if index is not None:
self._set_index(index)
def _attach_events(self) -> None:
vlc = self._vlc
@@ -274,11 +341,6 @@ class VlcPlayer(PlayerBase):
self._set_playing(False)
self._announce_playlist_finished()
def _on_next_item(self, event: Any) -> None:
try:
mrl = event.u.media.get_mrl()
except AttributeError: # pragma: no cover - depends on the libVLC build
return
index = self._mrl_to_index.get(mrl)
if index is not None:
self._set_index(index)
def _on_next_item(self, _event: Any) -> None:
# The event says *when* to look; what it carries is not portable, so ignore it.
self._sync_index()

View File

@@ -15,6 +15,7 @@ from __future__ import annotations
import struct
from dataclasses import dataclass, replace
from enum import IntEnum
from typing import override
from musicmouse.effects import (
EffectAlexaSwipeConfig,
@@ -366,6 +367,7 @@ class SetButtonBrightness:
button: Button
brightness: float
@override
def __repr__(self) -> str:
return f"{self.button.slug} backlight <- {self.brightness:.2f}"

View File

@@ -178,6 +178,20 @@ class PlayFigureRequested(IntentEvent):
restart: bool = True
@dataclass(frozen=True, slots=True, kw_only=True)
class PlayAlbumRequested(IntentEvent):
"""Start any album from the library, figure or not."""
album_id: str
track_index: int = 0
@dataclass(frozen=True, slots=True, kw_only=True)
class SeekRequested(IntentEvent):
#: Seconds from the start of the current track.
position: float
@dataclass(frozen=True, slots=True, kw_only=True)
class VolumeChangeRequested(IntentEvent):
delta: int

View File

@@ -0,0 +1,164 @@
"""The music collection: what exists, what it is called, and what colour it is.
The scan is the only part of this backend that touches hundreds of files, so it runs
in a worker thread and its results are cached. Everything the rest of the app sees is
plain immutable data.
"""
from __future__ import annotations
import asyncio
import logging
from collections.abc import Mapping
from dataclasses import replace
from pathlib import Path
from musicmouse.library.analysis import ANALYZER_VERSION, Analyzer, BeatGrid, NullAnalyzer
from musicmouse.library.cache import Fingerprint, LibraryCache
from musicmouse.library.models import Album, LibraryTrack, album_id, track_key
from musicmouse.library.scanner import scan_library
from musicmouse.library.sections import SECTIONS, AlbumKind
from musicmouse.media import Playlist
_log = logging.getLogger(__name__)
__all__ = [
"ANALYZER_VERSION",
"SECTIONS",
"Album",
"Analyzer",
"BeatGrid",
"LibraryCache",
"LibraryTrack",
"MusicLibrary",
"NullAnalyzer",
"album_id",
"track_key",
]
class MusicLibrary:
"""An immutable index of albums, rebuilt wholesale rather than mutated in place."""
def __init__(
self,
root: Path,
cache: LibraryCache,
extensions: frozenset[str],
*,
analyzer: Analyzer | None = None,
figure_kinds: Mapping[str, AlbumKind] | None = None,
) -> None:
self.root = root
self.cache = cache
self.extensions = extensions
self.analyzer: Analyzer = analyzer or NullAnalyzer()
#: What each figure holds. The only thing a folder name cannot say.
self.figure_kinds: Mapping[str, AlbumKind] = figure_kinds or {}
self._entries: dict[str, tuple[Album, Fingerprint]] = {}
# -------------------------------------------------------------------- reading
@property
def albums(self) -> list[Album]:
return [album for album, _ in self._entries.values()]
def get(self, identifier: str | None) -> Album | None:
entry = self._entries.get(identifier) if identifier else None
return entry[0] if entry else None
def figure_playlists(self) -> dict[str, Playlist]:
"""One playlist per figure folder, keyed by figure name.
The figure path and the web path must hand the player the *same* object for the
same album: ``reactions.playback.play_figure`` resumes on an identity check.
"""
return {
album.figure: album.to_playlist()
for album, _ in self._entries.values()
if album.figure is not None
}
def beats(self, identifier: str, index: int) -> BeatGrid | None:
album = self.get(identifier)
if album is None or not 0 <= index < len(album.tracks):
return None
return self.cache.load_beats(track_key(album.tracks[index].path))
# -------------------------------------------------------------------- writing
async def refresh(self) -> None:
"""Rescan from disk. Blocking work happens off the loop; the swap is atomic."""
known = dict(self._entries)
entries = await asyncio.to_thread(
scan_library,
self.root,
self.extensions,
self.cache,
known=known,
figure_kinds=self.figure_kinds,
)
self._entries = await asyncio.to_thread(self._with_analysis, entries)
await asyncio.to_thread(self.cache.store_index, self._entries)
def _with_analysis(
self, entries: dict[str, tuple[Album, Fingerprint]]
) -> dict[str, tuple[Album, Fingerprint]]:
"""Fold cached analysis results into the freshly scanned index.
Skipped entirely while ``analysis/`` is empty, which is the normal case until an
analyzer has actually been run - no point stat-ing 900 files that cannot exist.
"""
if not any(self.cache.analysis.glob("*.json")):
return entries
out: dict[str, tuple[Album, Fingerprint]] = {}
for identifier, (album, fingerprint) in entries.items():
tracks = tuple(
replace(track, analysis=self.cache.load_analysis(track_key(track.path)))
for track in album.tracks
)
out[identifier] = (replace(album, tracks=tracks), fingerprint)
return out
@classmethod
async def build(
cls,
root: Path,
cache_dir: Path,
extensions: frozenset[str],
*,
analyzer: Analyzer | None = None,
figure_kinds: Mapping[str, AlbumKind] | None = None,
) -> MusicLibrary:
cache = LibraryCache(cache_dir)
library = cls(root, cache, extensions, analyzer=analyzer, figure_kinds=figure_kinds)
library._entries = await asyncio.to_thread(cache.load_index)
await library.refresh()
return library
async def analyze_pending(self) -> int:
"""Run the analyzer over tracks that have no current result.
Never called during startup: the index is what playback needs, and analysis is
minutes of DSP per track. Results are keyed by file content, so they can equally
well be produced on a faster machine and the ``analysis/`` folder copied over.
"""
analyzer = self.analyzer
if analyzer.version < ANALYZER_VERSION:
return 0
done = 0
for album in self.albums:
for track in album.tracks:
key = track_key(track.path)
cached = self.cache.load_analysis(key)
if cached is not None and cached.version >= analyzer.version:
continue
analysis, grid = await asyncio.to_thread(analyzer.analyze, track.path)
if grid is not None:
self.cache.store_beats(key, grid)
self.cache.store_analysis(key, analysis)
done += 1
if done:
_log.info("Analyzed %d tracks", done)
return done

View File

@@ -0,0 +1,106 @@
"""Offline audio analysis: the seam, not the implementation.
Nothing here computes anything yet. What it does is fix the shape of the results so
that the analyzer can arrive later without touching the scanner, the cache format, the
API contract or the frontend's data flow.
Two rules hold the design together:
* **Scalars travel with the index, time series do not.** A 25-minute podcast at 120 BPM
has ~3000 beats; 900 tracks of that inside ``GET /api/library`` would be tens of
megabytes. :class:`TrackAnalysis` is a handful of floats and rides along; the beat
grid lives in its own file and is fetched for the one track that is playing.
* **Every field is optional with a default.** Adding ``danceability`` later needs no
migration and no cache wipe: unknown keys on disk are dropped on load, missing ones
fall back to the default. Only :data:`ANALYZER_VERSION` moving past what a file
records marks that file stale.
"""
from __future__ import annotations
from dataclasses import asdict, dataclass, fields
from pathlib import Path
from typing import Any, Final, Protocol
__all__ = [
"ANALYZER_VERSION",
"Analyzer",
"BeatGrid",
"NullAnalyzer",
"TrackAnalysis",
]
#: Bumped when an analyzer's output changes meaning. Cached results recorded under a
#: lower version are recomputed; results at or above it are left alone.
ANALYZER_VERSION: Final = 1
@dataclass(frozen=True, slots=True)
class TrackAnalysis:
"""What an animation may want to know about a track, in a few dozen bytes."""
#: The :data:`ANALYZER_VERSION` that produced this. ``0`` means "never analyzed".
version: int = 0
tempo: float | None = None
#: 0..1 overall loudness and drive.
energy: float | None = None
#: 0..1 mood, dark and calm through bright and happy.
valence: float | None = None
#: 0..1 spectral centroid. Drives the background hue.
brightness: float | None = None
#: Whether a beat grid for this track exists on disk.
beats: bool = False
@property
def is_analyzed(self) -> bool:
return self.version >= ANALYZER_VERSION
def to_json(self) -> dict[str, Any]:
return asdict(self)
@classmethod
def from_json(cls, data: dict[str, Any]) -> TrackAnalysis:
"""Load leniently: unknown keys are dropped, missing keys take their default.
This is what makes the cache survive an analyzer that grew a field.
"""
known = {f.name for f in fields(cls)}
return cls(**{k: v for k, v in data.items() if k in known})
@dataclass(frozen=True, slots=True)
class BeatGrid:
"""Beat onsets in seconds with a 0..1 strength each, for beat-synced animation."""
times: tuple[float, ...]
strengths: tuple[float, ...]
def to_json(self) -> dict[str, Any]:
# Flat pairs: half the JSON punctuation of a list of objects.
return {"beats": [x for pair in zip(self.times, self.strengths, strict=True) for x in pair]}
@classmethod
def from_json(cls, data: dict[str, Any]) -> BeatGrid:
flat: list[float] = data["beats"]
return cls(tuple(flat[0::2]), tuple(flat[1::2]))
class Analyzer(Protocol):
"""Turns one audio file into cacheable analysis results.
Implementations are CPU-bound and run off the event loop. The real one will live
behind an optional dependency group so the device never installs numpy to play music.
"""
version: int
def analyze(self, path: Path) -> tuple[TrackAnalysis, BeatGrid | None]: ...
class NullAnalyzer:
"""Analyzes nothing. Keeps the wiring exercised until a real analyzer lands."""
version = 0
def analyze(self, path: Path) -> tuple[TrackAnalysis, BeatGrid | None]: # noqa: ARG002
return TrackAnalysis(), None

View File

@@ -0,0 +1,204 @@
"""Where scan results live between runs.
A directory rather than a single file, because the three kinds of content cost wildly
different amounts to produce::
<cache_dir>/
├── index.json cheap: tags and structure. Thrown away freely.
├── covers/<album_id>.jpg medium: art pulled out of an ID3 APIC frame
└── analysis/<track_key>.json expensive: minutes of DSP per track
analysis/<track_key>.beats.json
That split is the whole point. A rescan must be free to rebuild ``index.json`` without
destroying analysis, so everything expensive is keyed by a *content* key (see
:func:`~musicmouse.library.models.track_key`) rather than by album id - renaming a
folder or re-sorting a section then costs nothing.
"""
from __future__ import annotations
import json
import logging
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any, cast
from musicmouse.library.analysis import BeatGrid, TrackAnalysis
from musicmouse.library.models import Album, LibraryTrack
from musicmouse.library.sections import SECTIONS, AlbumKind
_log = logging.getLogger(__name__)
__all__ = ["Fingerprint", "LibraryCache"]
#: Bump whenever the scanner's *derivation* changes - how a title, artist or series is
#: worked out - not just when the JSON shape does. A cached entry is reused whenever its
#: files are untouched, so otherwise a change to that logic is invisible until somebody
#: edits their music folder.
_INDEX_VERSION = 3
@dataclass(frozen=True, slots=True)
class Fingerprint:
"""What makes a folder's cached entry still valid: its files, sizes and mtimes."""
files: tuple[tuple[str, int, int], ...]
@classmethod
def of(cls, paths: list[Path]) -> Fingerprint:
entries: list[tuple[str, int, int]] = []
for path in paths:
stat = path.stat()
entries.append((path.name, stat.st_size, int(stat.st_mtime)))
return cls(tuple(entries))
def to_json(self) -> list[list[Any]]:
return [list(entry) for entry in self.files]
@classmethod
def from_json(cls, data: list[list[Any]]) -> Fingerprint:
return cls(tuple((str(n), int(s), int(m)) for n, s, m in data))
def _write_atomic(path: Path, payload: str) -> None:
"""Write via a sibling temp file so a crash never leaves a half-written cache."""
temp = path.with_name(f"{path.name}.tmp{os.getpid()}")
temp.write_text(payload, encoding="utf-8")
temp.replace(path)
class LibraryCache:
def __init__(self, directory: Path) -> None:
self.directory = directory
self.covers = directory / "covers"
self.analysis = directory / "analysis"
def prepare(self) -> None:
for folder in (self.directory, self.covers, self.analysis):
folder.mkdir(parents=True, exist_ok=True)
# -------------------------------------------------------------------- covers
def cover_path(self, album_id: str) -> Path:
return self.covers / f"{album_id}.jpg"
def store_cover(self, album_id: str, data: bytes) -> Path:
path = self.cover_path(album_id)
path.write_bytes(data)
return path
# ------------------------------------------------------------------ analysis
def load_analysis(self, key: str) -> TrackAnalysis | None:
path = self.analysis / f"{key}.json"
try:
return TrackAnalysis.from_json(json.loads(path.read_text(encoding="utf-8")))
except (OSError, ValueError):
return None
def store_analysis(self, key: str, analysis: TrackAnalysis) -> None:
_write_atomic(self.analysis / f"{key}.json", json.dumps(analysis.to_json()))
def load_beats(self, key: str) -> BeatGrid | None:
path = self.analysis / f"{key}.beats.json"
try:
return BeatGrid.from_json(json.loads(path.read_text(encoding="utf-8")))
except (OSError, ValueError, KeyError):
return None
def store_beats(self, key: str, grid: BeatGrid) -> None:
_write_atomic(self.analysis / f"{key}.beats.json", json.dumps(grid.to_json()))
# --------------------------------------------------------------------- index
def load_index(self) -> dict[str, tuple[Album, Fingerprint]]:
path = self.directory / "index.json"
try:
raw = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError):
return {}
if raw.get("version") != _INDEX_VERSION:
_log.info("Library index is from an older version; rescanning from scratch")
return {}
out: dict[str, tuple[Album, Fingerprint]] = {}
for entry in raw.get("albums", []):
try:
out[entry["id"]] = (_album_from_json(entry), Fingerprint.from_json(entry["files"]))
except (KeyError, TypeError, ValueError):
_log.debug("Dropping unreadable index entry %r", entry.get("id"))
return out
def store_index(self, albums: dict[str, tuple[Album, Fingerprint]]) -> None:
payload = {
"version": _INDEX_VERSION,
"albums": [
_album_to_json(album) | {"files": fingerprint.to_json()}
for album, fingerprint in albums.values()
],
}
self.prepare()
_write_atomic(self.directory / "index.json", json.dumps(payload, ensure_ascii=False))
# ------------------------------------------------------------------ serialisation
def _album_to_json(album: Album) -> dict[str, Any]:
return {
"id": album.id,
"section": album.section,
"kind": album.kind,
"title": album.title,
"artist": album.artist,
"series": album.series,
"figure": album.figure,
"colors": list(album.colors),
"folder": str(album.folder),
"cover": str(album.cover) if album.cover else None,
"tracks": [
{
"path": str(track.path),
"title": track.title,
"duration": track.duration,
"analysis": track.analysis.to_json() if track.analysis else None,
}
for track in album.tracks
],
}
def _album_from_json(data: dict[str, Any]) -> Album:
section = str(data["section"])
if section not in SECTIONS:
raise ValueError(f"unknown section {section!r}")
# Read back what was stored rather than re-deriving it from the section: a figure's
# kind comes from the config, so the section cannot answer for it.
kind = str(data["kind"])
if kind not in ("music", "book"):
raise ValueError(f"unknown kind {kind!r}")
red, green, blue = data["colors"]
return Album(
id=str(data["id"]),
section=section,
kind=cast("AlbumKind", kind),
title=str(data["title"]),
artist=str(data["artist"]),
series=data["series"],
figure=data["figure"],
colors=(str(red), str(green), str(blue)),
folder=Path(data["folder"]),
cover=Path(data["cover"]) if data["cover"] else None,
tracks=tuple(
LibraryTrack(
path=Path(track["path"]),
title=str(track["title"]),
duration=float(track["duration"]),
analysis=(
TrackAnalysis.from_json(track["analysis"]) if track.get("analysis") else None
),
)
for track in data["tracks"]
),
)

View File

@@ -0,0 +1,82 @@
"""Three colours per album, from its cover art.
The frontend paints an album card with these and the LED strips run an effect in them,
so a web-started album looks the same on the shelf as it does on the screen. Extraction
happens once per album ever - the result is cached next to the cover.
"""
from __future__ import annotations
import colorsys
import hashlib
import logging
from io import BytesIO
from typing import cast
from musicmouse.library.models import AlbumColors
_log = logging.getLogger(__name__)
__all__ = ["colors_from_cover", "colors_from_id"]
#: Quantize to this many candidates before ranking. More just returns near-duplicates.
_PALETTE_SIZE = 8
#: Downscale first: colour proportions survive, the decode gets ~100x cheaper.
_SAMPLE_SIZE = (64, 64)
#: Greys and near-blacks read as mud on an RGBW strip and as dirt on a card, so a
#: candidate has to clear both bars to count as one of an album's colours.
_MIN_SATURATION = 0.25
_MIN_VALUE = 0.25
def _hex(r: int, g: int, b: int) -> str:
return f"#{r:02x}{g:02x}{b:02x}"
def colors_from_id(album_id: str) -> AlbumColors:
"""Synthesise a palette when there is no cover to take one from.
Analogous plus complementary off one hashed hue: distinct per album, never muddy.
"""
hue = int(hashlib.sha1(album_id.encode()).hexdigest()[:4], 16) % 360
out: list[str] = []
for offset, saturation, value in ((0, 0.72, 0.95), (30, 0.62, 0.80), (180, 0.70, 0.90)):
r, g, b = colorsys.hsv_to_rgb(((hue + offset) % 360) / 360, saturation, value)
out.append(_hex(int(r * 255), int(g * 255), int(b * 255)))
return out[0], out[1], out[2]
def colors_from_cover(image_data: bytes, album_id: str) -> AlbumColors:
"""Rank the cover's dominant colours, dropping the ones that would read as mud.
Falls back to :func:`colors_from_id` for the slots that do not fill - a black-and-
white cover legitimately has no three usable colours.
"""
try:
from PIL import Image
with Image.open(BytesIO(image_data)) as image:
sample = image.convert("RGB").resize(_SAMPLE_SIZE)
quantized = sample.quantize(colors=_PALETTE_SIZE, method=Image.Quantize.MAXCOVERAGE)
palette: list[int] = quantized.getpalette() or []
# On a quantized ("P" mode) image getcolors() yields (count, palette index).
# Pillow's annotation covers every mode, hence the cast.
counts = cast("list[tuple[int, int]]", quantized.getcolors() or [])
except Exception: # pragma: no cover - defensive around arbitrary embedded art
_log.debug("Cover of %s could not be read; using a synthesised palette", album_id)
return colors_from_id(album_id)
picked: list[str] = []
for _count, index in sorted(counts, reverse=True):
r, g, b = palette[index * 3 : index * 3 + 3]
_, saturation, value = colorsys.rgb_to_hsv(r / 255, g / 255, b / 255)
if saturation < _MIN_SATURATION or value < _MIN_VALUE:
continue
picked.append(_hex(r, g, b))
if len(picked) == 3:
return picked[0], picked[1], picked[2]
fallback = colors_from_id(album_id)
filled = picked + list(fallback[len(picked) :])
return filled[0], filled[1], filled[2]

View File

@@ -0,0 +1,88 @@
"""What the browse API serves: albums of tracks, with the colours to draw them in."""
from __future__ import annotations
import hashlib
from dataclasses import dataclass
from pathlib import Path
from musicmouse.library.analysis import TrackAnalysis
from musicmouse.library.sections import AlbumKind
from musicmouse.media import Playlist, Track
__all__ = ["Album", "AlbumColors", "LibraryTrack", "album_id", "track_key"]
#: Primary, secondary and accent as ``"#rrggbb"`` - the format
#: :func:`musicmouse.color.parse_color` already accepts, so the LED side needs no new
#: parsing and the frontend gets CSS colours for free.
type AlbumColors = tuple[str, str, str]
def album_id(root: Path, folder: Path) -> str:
"""A stable id for an album folder, from its path relative to the library root."""
relative = folder.relative_to(root).as_posix()
return hashlib.sha1(relative.encode()).hexdigest()[:12]
def track_key(path: Path) -> str:
"""Content key for expensive per-track results.
Keyed on the *file*, not the album, so renaming a folder or re-sorting a section
never throws away analysis that took minutes to compute.
"""
stat = path.stat()
material = f"{path.name}:{stat.st_size}:{int(stat.st_mtime)}"
return hashlib.sha1(material.encode()).hexdigest()[:16]
@dataclass(frozen=True, slots=True)
class LibraryTrack:
path: Path
title: str
#: Seconds, from the tags. ``0.0`` when the file carries no duration.
duration: float = 0.0
analysis: TrackAnalysis | None = None
def __repr__(self) -> str:
return f"LibraryTrack({self.title!r}, {self.duration:.0f}s)"
@dataclass(frozen=True, slots=True)
class Album:
id: str
section: str
kind: AlbumKind
title: str
artist: str
#: Grouping key for audiobooks and podcasts. ``None`` for music, which groups by
#: artist instead - the two cases the browse UI's category view distinguishes.
series: str | None
#: Figure name when this folder sits under ``Figuren``, else ``None``.
figure: str | None
colors: AlbumColors
folder: Path
cover: Path | None
tracks: tuple[LibraryTrack, ...]
def __len__(self) -> int:
return len(self.tracks)
@property
def duration(self) -> float:
return sum(track.duration for track in self.tracks)
@property
def category(self) -> str:
"""How the browse view groups this album: series for books, artist for music."""
return self.series or self.artist
def to_playlist(self) -> Playlist:
"""The player's view of this album."""
return Playlist(
name=self.figure or self.title,
tracks=tuple(Track(track.path) for track in self.tracks),
album_id=self.id,
)
def __repr__(self) -> str:
return f"Album({self.title!r}, {self.artist!r}, {len(self.tracks)} tracks)"

View File

@@ -0,0 +1,226 @@
"""Turning folders on disk into :class:`~musicmouse.library.models.Album` objects.
One level under each section folder, every directory holding audio files is one album.
Tags come from mutagen; where a section's tags are known to be useless the folder name
wins instead (see :mod:`musicmouse.library.sections`). Nothing here raises on bad
input - an unreadable file loses its metadata, not the boot.
"""
from __future__ import annotations
import logging
from collections import Counter
from collections.abc import Mapping
from pathlib import Path
from musicmouse.library.cache import Fingerprint, LibraryCache
from musicmouse.library.colors import colors_from_cover, colors_from_id
from musicmouse.library.models import Album, LibraryTrack, album_id
from musicmouse.library.sections import SECTIONS, AlbumKind, Section
_log = logging.getLogger(__name__)
__all__ = ["scan_library"]
#: Checked in order for a cover sitting next to the audio.
_COVER_NAMES = ("cover.jpg", "cover.jpeg", "cover.png", "folder.jpg")
def _audio_files(folder: Path, extensions: frozenset[str]) -> list[Path]:
"""The playable files in ``folder``, alphabetically.
Dotfiles are skipped outright and everything else must match ``audio_extensions``,
which is what keeps a podcast folder's ``archive.json`` and its half-finished
``.podcast-dl-*.download.tmp`` out of the playlist.
"""
return sorted(
(
path
for path in folder.iterdir()
if path.is_file()
and not path.name.startswith(".")
and path.suffix.lower() in extensions
),
key=lambda path: path.name,
)
def _tags(path: Path) -> tuple[dict[str, str], float]:
"""``(tags, duration)`` for one file. Empty and ``0.0`` when it cannot be read."""
try:
import mutagen
audio = mutagen.File(path, easy=True)
if audio is None:
return {}, 0.0
tags = {key: values[0] for key, values in dict(audio).items() if values}
return tags, float(getattr(audio.info, "length", 0.0) or 0.0)
except Exception: # pragma: no cover - mutagen raises freely on damaged files
_log.debug("No readable tags in %s", path)
return {}, 0.0
def _embedded_art(path: Path) -> bytes | None:
try:
import mutagen
raw = mutagen.File(path)
if raw is None or raw.tags is None:
return None
for key in raw.tags:
if key.startswith("APIC"):
data: bytes = raw.tags[key].data
return data
except Exception: # pragma: no cover - defensive
_log.debug("No readable embedded art in %s", path)
return None
def _split_folder_name(name: str) -> tuple[str, str]:
"""``"Conni - Conni in den Bergen"`` -> ``("Conni", "Conni in den Bergen")``."""
artist, separator, title = name.partition(" - ")
return (artist, title) if separator else ("", name)
def _most_common(values: list[str]) -> str:
"""The tag value most files in an album agree on, ignoring blanks."""
counted = Counter(value for value in values if value)
return counted.most_common(1)[0][0] if counted else ""
def _cover_for(
folder: Path, paths: list[Path], identifier: str, cache: LibraryCache
) -> tuple[Path | None, bytes | None]:
for name in _COVER_NAMES:
candidate = folder / name
if candidate.is_file():
return candidate, candidate.read_bytes()
for path in paths[:3]:
# Podcast feeds sometimes art only some episodes; a couple of tries is enough.
art = _embedded_art(path)
if art is not None:
return cache.store_cover(identifier, art), art
return None, None
def scan_album(
folder: Path,
*,
root: Path,
section_name: str,
section: Section,
extensions: frozenset[str],
cache: LibraryCache,
figure_kinds: Mapping[str, AlbumKind] | None = None,
) -> tuple[Album, Fingerprint] | None:
paths = _audio_files(folder, extensions)
if not paths:
_log.debug("No audio files in %s", folder)
return None
if section.order == "newest_first":
paths.reverse()
fingerprint = Fingerprint.of(paths)
identifier = album_id(root, folder)
tracks: list[LibraryTrack] = []
albums: list[str] = []
artists: list[str] = []
for path in paths:
tags, duration = _tags(path)
tracks.append(
LibraryTrack(path=path, title=tags.get("title") or path.stem, duration=duration)
)
albums.append(tags.get("album", ""))
artists.append(tags.get("albumartist") or tags.get("artist", ""))
figure = folder.name if section.figures else None
# Every other shelf is named after what is on it. A figure folder is named after the
# figurine, so its media type has to be declared in the config.
kind = (figure_kinds or {}).get(figure, "music") if figure else section.kind
folder_artist, folder_title = _split_folder_name(folder.name)
if section.title_from == "folder":
# A figure folder is named in lowercase ("fuchs"); it sits next to real album
# titles in the browse grid, so give it a capital.
title = folder.name[:1].upper() + folder.name[1:] if section.figures else folder.name
else:
title = _most_common(albums) or folder_title or folder.name
if section.artist_from == "folder":
artist = folder.name
else:
artist = _most_common(artists) or folder_artist
cover, art = _cover_for(folder, paths, identifier, cache)
colors = colors_from_cover(art, identifier) if art else colors_from_id(identifier)
album = Album(
id=identifier,
section=section_name,
kind=kind,
title=title,
artist=artist,
# Books group by who or what they are about; music groups by artist. The browse
# view's category row is built straight off this, so it takes only the part
# before the first comma - an ``album_artist`` of "Bobo Siebenschlaefer, Markus
# Osterwalder, ..." is a credit list whose first name is the character.
series=artist.split(",")[0].strip() if kind == "book" else None,
figure=figure,
colors=colors,
folder=folder,
cover=cover,
tracks=tuple(tracks),
)
return album, fingerprint
def scan_library(
root: Path,
extensions: frozenset[str],
cache: LibraryCache,
*,
known: dict[str, tuple[Album, Fingerprint]] | None = None,
figure_kinds: Mapping[str, AlbumKind] | None = None,
) -> dict[str, tuple[Album, Fingerprint]]:
"""Scan every section under ``root``.
``known`` is the previously cached index: a folder whose files, sizes and mtimes are
unchanged is taken from it without a single tag being read. ``figure_kinds`` maps a
figure name to what it holds, which is the one thing the folders cannot say.
"""
cache.prepare()
known = known or {}
out: dict[str, tuple[Album, Fingerprint]] = {}
for section_name, section in SECTIONS.items():
section_root = root / section_name
if not section_root.is_dir():
_log.warning("Library section %r has no folder at %s", section_name, section_root)
continue
for folder in sorted(section_root.iterdir(), key=lambda path: path.name):
if not folder.is_dir() or folder.name.startswith("."):
continue
identifier = album_id(root, folder)
cached = known.get(identifier)
if cached is not None:
paths = _audio_files(folder, extensions)
if paths and Fingerprint.of(paths) == cached[1]:
out[identifier] = cached
continue
scanned = scan_album(
folder,
root=root,
section_name=section_name,
section=section,
extensions=extensions,
cache=cache,
figure_kinds=figure_kinds,
)
if scanned is not None:
out[identifier] = scanned
_log.info("Library: %d albums under %s", len(out), root)
return out

View File

@@ -0,0 +1,46 @@
"""The four shelves of the music library, and how each one differs.
These are folder names, not configuration. The library has one root and the layout
underneath it is fixed - a section that needed configuring would be a section whose
quirks are not understood yet.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Final, Literal
__all__ = ["SECTIONS", "AlbumKind", "ArtistSource", "Section", "TitleSource", "TrackOrder"]
type AlbumKind = Literal["music", "book"]
type TrackOrder = Literal["filename", "newest_first"]
type TitleSource = Literal["tags", "folder"]
type ArtistSource = Literal["tags", "folder"]
@dataclass(frozen=True, slots=True)
class Section:
kind: AlbumKind = "music"
#: Subfolders are figure folders: their name is the figure name from the config.
figures: bool = False
order: TrackOrder = "filename"
title_from: TitleSource = "tags"
artist_from: ArtistSource = "tags"
#: ``Kinderpodcasts`` is the odd one out twice over. Its ``artist`` tag is the full
#: presenter list ("Thomas Welling, Sarah Schultes, ...") and its ``album`` tag is the
#: feed name, so neither groups usefully - the folder name is the show. And its files
#: are named ``YYYYMMDD - Title.mp3``, so reversing filename order puts the newest
#: episode first, which is the one anybody wants.
SECTIONS: Final[dict[str, Section]] = {
# A figure folder is named after the figure, and its contents are whatever that
# figure should play - often several albums' worth. The folder name is the honest
# title; the tags still supply a useful artist.
"Figuren": Section(figures=True, title_from="folder"),
"Musik": Section(),
"Hörbücher": Section(kind="book"),
"Kinderpodcasts": Section(
kind="book", order="newest_first", title_from="folder", artist_from="folder"
),
}

View File

@@ -1,18 +1,16 @@
"""Playlist model.
Deliberately a real type rather than a bare ``list[str]``: it is what a future
browse API would serve, and it keeps track metadata in one place.
Deliberately a real type rather than a bare ``list[str]``: it is what the browse API
serves, and it keeps track metadata in one place. Building one from a folder is
:mod:`musicmouse.library`'s job.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from pathlib import Path
_log = logging.getLogger(__name__)
__all__ = ["Playlist", "Track", "build_playlist"]
__all__ = ["Playlist", "Track"]
@dataclass(frozen=True, slots=True)
@@ -31,6 +29,9 @@ class Track:
class Playlist:
name: str
tracks: tuple[Track, ...]
#: Set when this playlist came from a library album. It is how a front-end answers
#: "what is playing?" without keeping its own copy of that state.
album_id: str | None = None
def __len__(self) -> int:
return len(self.tracks)
@@ -43,23 +44,3 @@ class Playlist:
def __repr__(self) -> str:
return f"Playlist({self.name!r}, {len(self.tracks)} tracks)"
def build_playlist(name: str, folder: Path, extensions: tuple[str, ...]) -> Playlist:
"""Collect ``folder``'s audio files into a playlist, ordered alphabetically.
A missing or empty folder yields an empty playlist and a warning rather than an
error: an unfinished playlist should not stop the mouse from booting.
"""
if not folder.is_dir():
_log.warning("Figure %r: no media folder at %s", name, folder)
return Playlist(name=name, tracks=())
suffixes = {ext.lower() for ext in extensions}
paths = sorted(
(p for p in folder.iterdir() if p.is_file() and p.suffix.lower() in suffixes),
key=lambda p: p.name,
)
if not paths:
_log.warning("Figure %r: no audio files in %s", name, folder)
return Playlist(name=name, tracks=tuple(Track(p) for p in paths))

View File

@@ -11,7 +11,11 @@ Publishing button presses to Home Assistant has no reaction of its own: the MQTT
service subscribes to those events directly.
"""
from musicmouse.reactions import lighting, playback # noqa: F401 (import = register)
from musicmouse.reactions import ( # noqa: F401 (import = register)
lighting,
playback,
status,
)
from musicmouse.reactions.registry import Reaction, on, register_all, registered
__all__ = ["Reaction", "on", "register_all", "registered"]

View File

@@ -14,9 +14,10 @@ import logging
from copy import deepcopy
from musicmouse.app import App
from musicmouse.color import ColorRGBW
from musicmouse.color import ColorRGBW, parse_color
from musicmouse.config import FigureColors
from musicmouse.effects import (
EffectCircularConfig,
EffectRandomTwoColorInterpolationConfig,
EffectReverseSwipe,
EffectStaticConfig,
@@ -24,6 +25,7 @@ from musicmouse.effects import (
)
from musicmouse.events import (
ActiveFigureChanged,
PlaybackChanged,
PlaylistFinished,
TouchButtonPressed,
TouchButtonReleased,
@@ -40,6 +42,12 @@ MOUSE_SWIPE_START_DEGREES = 6 / 45 * 360
MOUSE_BELL_CURVE_WIDTH = 16
SWIPE_SPEED = 180
#: The web front-end's animation: a three-quarter arc drifting round twice a minute, in
#: the album's own primary colour. Deliberately unlike the figure swipe - the strips
#: should say *which* way the mouse was started, not just that it was.
WEB_CIRCLE_WIDTH = 270.0 # degrees
WEB_CIRCLE_SPEED = 12.0 # degrees per second
@on(ActiveFigureChanged)
def figure_placed_or_removed(event: ActiveFigureChanged, app: App) -> None:
@@ -52,6 +60,26 @@ def figure_placed_or_removed(event: ActiveFigureChanged, app: App) -> None:
)
@on(PlaybackChanged)
def web_playback(event: PlaybackChanged, app: App) -> None:
"""Light the strips for playback that no figure started.
Reacting to ``PlaybackChanged`` rather than to the play intent covers pause, stop
and playlist-end in one place. The guard is what keeps this off the figure path:
while a figure is on the reader its animation owns all three zones.
"""
if app.mouse.active_figure is not None:
return
if not event.playing:
off_animation(app)
return
album = app.library.get(event.playlist.album_id if event.playlist else None)
if album is None:
return
web_animation(app, parse_color(album.colors[0]))
@on(PlaylistFinished)
def playlist_finished(_event: PlaylistFinished, app: App) -> None:
off_animation(app)
@@ -108,6 +136,16 @@ def start_animation(app: App, colors: FigureColors) -> None:
app.mouse.set_effect(LedZone.MOUSE, mouse, origin="device")
def web_animation(app: App, color: ColorRGBW) -> None:
for zone in LedZone:
app.mouse.set_effect(
zone,
EffectCircularConfig(speed=WEB_CIRCLE_SPEED, width=WEB_CIRCLE_WIDTH, color=color),
origin="device",
)
app.mouse.set_button_brightness(app.config.general.button_leds_brightness, origin="device")
def off_animation(app: App) -> None:
_log.info("Running off animation")
app.mouse.set_effect(LedZone.RING, EffectReverseSwipe(), origin="device")

View File

@@ -15,11 +15,13 @@ from musicmouse.events import (
ButtonEvent,
NextTrackRequested,
PauseRequested,
PlayAlbumRequested,
PlayFigureRequested,
PlaylistFinished,
PlayRequested,
PrevTrackRequested,
RotaryTurned,
SeekRequested,
SetVolumeRequested,
VolumeChangeRequested,
)
@@ -72,6 +74,21 @@ def play_figure(event: PlayFigureRequested, app: App) -> None:
app.player.play_from_start()
@on(PlayAlbumRequested)
def play_album(event: PlayAlbumRequested, app: App) -> None:
"""Play any album in the library. This is the web front-end's way in."""
album = app.library.get(event.album_id)
if album is None:
_log.warning("No album %r in the library", event.album_id)
return
# A figure album keeps the figure's own resume bookkeeping honest: it is the same
# Playlist object either way, because both come from the library index.
playlist = app.playlists.get(album.figure) if album.figure else album.to_playlist()
app.player.set_playlist(playlist or album.to_playlist())
app.player.play_track(event.track_index)
@on(PlaylistFinished)
def playlist_finished(_event: PlaylistFinished, app: App) -> None:
# Nothing was left half-played, so the next placement starts from the top.
@@ -127,6 +144,11 @@ def pause(_event: PauseRequested, app: App) -> None:
app.player.pause()
@on(SeekRequested)
def seek(event: SeekRequested, app: App) -> None:
app.player.seek(event.position)
@on(VolumeChangeRequested)
def change_volume(event: VolumeChangeRequested, app: App) -> None:
app.player.change_volume(event.delta, source=event.source)

View File

@@ -0,0 +1,18 @@
"""Remembering which links are up.
The firmware's state is readable straight off the transport, but a broker's is not:
:class:`~musicmouse.services.mqtt.service.MqttService` announces it and then forgets.
A front-end that wants to show a connection dot needs somewhere to read it from.
"""
from __future__ import annotations
from musicmouse.app import App
from musicmouse.events import ConnectionChanged
from musicmouse.reactions.registry import on
@on(ConnectionChanged)
def connection_changed(event: ConnectionChanged, app: App) -> None:
if event.target == "mqtt":
app.state.mqtt_connected = event.connected

View File

@@ -0,0 +1,6 @@
"""The web front-end: a browse-and-play UI over the same intents the buttons emit."""
from musicmouse.services.web.hub import StateHub
from musicmouse.services.web.service import WebService, build_app
__all__ = ["StateHub", "WebService", "build_app"]

View File

@@ -0,0 +1,192 @@
"""The HTTP surface.
Commands are REST and state is a one-way websocket. That split keeps every control path
testable with ``curl`` and means a command needs no new machinery: it emits the same
intent the MQTT service and the buttons emit, and lands in the same reaction.
Search is not here on purpose. The whole index goes to the browser once and filtering
happens there, which is what makes the design's type-to-search feel instant.
"""
from __future__ import annotations
import asyncio
import logging
from pathlib import Path
from fastapi import APIRouter, HTTPException, Response, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse
from musicmouse.app import App
from musicmouse.events import (
IntentEvent,
NextTrackRequested,
PauseRequested,
PlayAlbumRequested,
PlayRequested,
PrevTrackRequested,
SeekRequested,
SetVolumeRequested,
)
from musicmouse.services.web.hub import StateHub
from musicmouse.services.web.schemas import (
AlbumOut,
BeatsOut,
LibraryOut,
PlayerStateOut,
PlayIn,
SeekIn,
SettingsIn,
SettingsOut,
VolumeIn,
)
from musicmouse.services.web.settings import (
read_settings,
to_device_volume,
to_percent,
write_settings,
)
from musicmouse.services.web.state import snapshot
_log = logging.getLogger(__name__)
__all__ = ["build_router"]
def build_router(app: App, hub: StateHub, config_path: Path) -> APIRouter:
router = APIRouter(prefix="/api")
def emit(intent: IntentEvent) -> Response:
app.bus.emit(intent)
return Response(status_code=204)
# ------------------------------------------------------------------- library
@router.get("/library")
def get_library() -> LibraryOut:
return LibraryOut(albums=[AlbumOut.of(album) for album in app.library.albums])
@router.get("/albums/{album_id}/cover")
def get_cover(album_id: str) -> FileResponse:
album = app.library.get(album_id)
if album is None or album.cover is None:
# Not an error: the client paints the album's own colours instead.
raise HTTPException(status_code=404, detail="no cover")
return FileResponse(
album.cover,
# Cover files are content-addressed by album id and rewritten only by a
# rescan, so a long cache is safe and saves 20 requests per page load.
headers={"Cache-Control": "public, max-age=604800"},
)
@router.get("/tracks/{album_id}/{index}/analysis")
def get_track_analysis(album_id: str, index: int) -> BeatsOut:
grid = app.library.beats(album_id, index)
if grid is None:
raise HTTPException(status_code=404, detail="not analyzed")
return BeatsOut(times=list(grid.times), strengths=list(grid.strengths))
@router.post("/library/refresh", status_code=202)
async def refresh_library() -> Response:
async def rescan() -> None:
await app.library.refresh()
app.playlists.clear()
app.playlists.update(app.library.figure_playlists())
await hub.broadcast_library()
# Returns immediately: a cold rescan reads tags from every file.
asyncio.create_task(rescan()) # noqa: RUF006
return Response(status_code=202)
# --------------------------------------------------------------------- state
@router.get("/state")
def get_state() -> PlayerStateOut:
return snapshot(app)
@router.websocket("/ws")
async def websocket(socket: WebSocket) -> None:
await hub.connect(socket)
try:
while True:
# Push-only. Reading is how we notice the tab closed.
await socket.receive_text()
except WebSocketDisconnect:
pass
finally:
hub.disconnect(socket)
# ------------------------------------------------------------------ commands
@router.post("/play", status_code=204)
def play_album(body: PlayIn) -> Response:
if app.library.get(body.album_id) is None:
raise HTTPException(status_code=404, detail="no such album")
return emit(
PlayAlbumRequested(
album_id=body.album_id, track_index=body.track_index, source="web"
)
)
@router.post("/resume", status_code=204)
def resume() -> Response:
return emit(PlayRequested(source="web"))
@router.post("/pause", status_code=204)
def pause() -> Response:
return emit(PauseRequested(source="web"))
@router.post("/next", status_code=204)
def next_track() -> Response:
return emit(NextTrackRequested(source="web"))
@router.post("/previous", status_code=204)
def previous_track() -> Response:
return emit(PrevTrackRequested(source="web"))
@router.post("/seek", status_code=204)
def seek(body: SeekIn) -> Response:
return emit(SeekRequested(position=body.position, source="web"))
@router.post("/volume", status_code=204)
def set_volume(body: VolumeIn) -> Response:
general = app.config.general
if body.percent is not None:
target = body.percent
elif body.delta_percent is not None:
target = to_percent(app.player.volume, general) + body.delta_percent
else:
raise HTTPException(status_code=422, detail="percent or delta_percent required")
return emit(
SetVolumeRequested(
volume=to_device_volume(max(0, min(100, target)), general), source="web"
)
)
# ------------------------------------------------------------- parent mode
@router.get("/settings")
def get_settings() -> SettingsOut:
return read_settings(app.config.general)
@router.put("/settings")
async def put_settings(body: SettingsIn) -> SettingsOut:
if body.min_volume > body.max_volume:
raise HTTPException(
status_code=422, detail="min_volume must not exceed max_volume"
)
if not body.min_volume <= body.initial_volume <= body.max_volume:
raise HTTPException(
status_code=422, detail="initial_volume must lie between min and max"
)
general = app.config.general
# Applied live as well as saved: a parent lowering the ceiling expects the next
# song to be quieter, not the next boot.
for key, value in body.model_dump().items():
setattr(general, key, value)
app.player.set_volume_limits(general.min_volume, general.max_volume)
await asyncio.to_thread(write_settings, config_path, body)
await hub.broadcast_state()
return read_settings(general)
return router

View File

@@ -0,0 +1,104 @@
"""Getting state out to every open browser tab.
State events only fire on *change*, so a tab that connects halfway through a track
would otherwise sit there knowing nothing. The fix is the one
:class:`~musicmouse.services.mqtt.service.MqttService` already uses for a reconnecting
broker: send a full snapshot on connect, then deltas.
Position is the exception to "everything goes on the bus". A progress bar wants it
twice a second; an event at that rate would flood the queue, the MQTT service and the
log to serve one front-end. So the hub reads it straight off the player on its own
timer, and only while something is playing.
"""
from __future__ import annotations
import asyncio
import contextlib
import json
import logging
from typing import Any
from fastapi import WebSocket
from musicmouse.app import App
from musicmouse.events import Event, StateEvent
from musicmouse.services.web.state import snapshot
_log = logging.getLogger(__name__)
__all__ = ["StateHub"]
#: Twice a second: smooth enough once the client interpolates between frames, cheap
#: enough to leave running.
POSITION_INTERVAL = 0.5
class StateHub:
def __init__(self, app: App) -> None:
self.app = app
self._clients: set[WebSocket] = set()
self._unsubscribe: Any = None
# ------------------------------------------------------------------- lifecycle
def start(self) -> None:
self._unsubscribe = self.app.bus.subscribe(StateEvent, self._on_state)
def stop(self) -> None:
if self._unsubscribe is not None:
self._unsubscribe()
self._unsubscribe = None
async def run(self) -> None:
"""Push the playback position while anything is playing."""
while True:
await asyncio.sleep(POSITION_INTERVAL)
if not self._clients or not self.app.player.is_playing:
continue
await self.broadcast(
{
"type": "position",
"position": self.app.player.position,
"duration": self.app.player.duration,
}
)
# --------------------------------------------------------------------- clients
async def connect(self, socket: WebSocket) -> None:
await socket.accept()
self._clients.add(socket)
await self._send(socket, {"type": "state", "state": snapshot(self.app).model_dump()})
def disconnect(self, socket: WebSocket) -> None:
self._clients.discard(socket)
# ------------------------------------------------------------------ publishing
async def broadcast(self, message: dict[str, Any]) -> None:
if not self._clients:
return
payload = json.dumps(message)
for socket in list(self._clients):
try:
await socket.send_text(payload)
except Exception:
# A tab that closed mid-send is normal, not an error worth logging loudly.
_log.debug("Dropping a websocket client that went away")
self._clients.discard(socket)
async def broadcast_state(self) -> None:
await self.broadcast({"type": "state", "state": snapshot(self.app).model_dump()})
async def broadcast_library(self) -> None:
await self.broadcast({"type": "library"})
# ------------------------------------------------------------------ internals
async def _send(self, socket: WebSocket, message: dict[str, Any]) -> None:
with contextlib.suppress(Exception):
await socket.send_text(json.dumps(message))
async def _on_state(self, _event: Event) -> None:
await self.broadcast_state()

View File

@@ -0,0 +1,156 @@
"""What the browser sees.
Two rules shape these. The whole library ships in one response, because the design's
incremental A-Z search has to be instant and ~25 albums of metadata is under 100 kB -
so no track carries anything it does not need. And volume is a *percentage* here: the
configured ceiling is a parent's business, not a child's, so it never crosses this line.
"""
from __future__ import annotations
from pydantic import BaseModel, Field
from musicmouse.library import Album
from musicmouse.library.analysis import TrackAnalysis
__all__ = [
"AlbumOut",
"BeatsOut",
"LibraryOut",
"PlayIn",
"PlayerStateOut",
"SeekIn",
"SettingsIn",
"SettingsOut",
"TrackOut",
"VolumeIn",
]
class AnalysisOut(BaseModel):
tempo: float | None = None
energy: float | None = None
valence: float | None = None
brightness: float | None = None
beats: bool = False
@classmethod
def of(cls, analysis: TrackAnalysis | None) -> AnalysisOut | None:
if analysis is None or not analysis.is_analyzed:
return None
return cls(
tempo=analysis.tempo,
energy=analysis.energy,
valence=analysis.valence,
brightness=analysis.brightness,
beats=analysis.beats,
)
class TrackOut(BaseModel):
title: str
duration: float
#: Scalars only. The beat grid is fetched per track from ``/api/tracks/...``.
analysis: AnalysisOut | None = None
class AlbumOut(BaseModel):
id: str
section: str
kind: str
title: str
artist: str
series: str | None
figure: str | None
category: str
colors: list[str]
has_cover: bool
duration: float
tracks: list[TrackOut]
@classmethod
def of(cls, album: Album) -> AlbumOut:
return cls(
id=album.id,
section=album.section,
kind=album.kind,
title=album.title,
artist=album.artist,
series=album.series,
figure=album.figure,
category=album.category,
colors=list(album.colors),
has_cover=album.cover is not None,
duration=album.duration,
tracks=[
TrackOut(
title=track.title,
duration=track.duration,
analysis=AnalysisOut.of(track.analysis),
)
for track in album.tracks
],
)
class LibraryOut(BaseModel):
albums: list[AlbumOut]
class BeatsOut(BaseModel):
times: list[float]
strengths: list[float]
class ConnectionOut(BaseModel):
firmware: bool
mqtt: bool
class PlayerStateOut(BaseModel):
playing: bool
album_id: str | None
album_title: str | None
artist: str | None
kind: str | None
track_index: int
track_title: str | None
track_count: int
position: float
duration: float
#: 0..100. The device's own range is deliberately not exposed.
volume: int
active_figure: str | None
connected: ConnectionOut
class PlayIn(BaseModel):
album_id: str
track_index: int = Field(default=0, ge=0)
class SeekIn(BaseModel):
position: float = Field(ge=0)
class VolumeIn(BaseModel):
"""Either an absolute percentage or a relative step; exactly one of the two."""
percent: int | None = Field(default=None, ge=0, le=100)
delta_percent: int | None = Field(default=None, ge=-100, le=100)
class SettingsOut(BaseModel):
min_volume: int
max_volume: int
initial_volume: int
volume_increment: int
button_leds_brightness: float
class SettingsIn(BaseModel):
min_volume: int = Field(ge=0, le=200)
max_volume: int = Field(ge=0, le=200)
initial_volume: int = Field(ge=0, le=200)
volume_increment: int = Field(ge=1, le=100)
button_leds_brightness: float = Field(ge=0, le=1)

View File

@@ -0,0 +1,88 @@
"""The web front-end, as a :class:`~musicmouse.services.base.Service`.
Everything runs on the one event loop the rest of the app already has: uvicorn's
``Server.serve()`` is a coroutine, so there is no second loop and no thread. The
service owns no state of its own - it turns HTTP into intents on the bus, and bus
events into websocket frames.
"""
from __future__ import annotations
import asyncio
import contextlib
import logging
from pathlib import Path
import uvicorn
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from musicmouse.app import App
from musicmouse.config import WebConfig
from musicmouse.services.web.api import build_router
from musicmouse.services.web.hub import StateHub
_log = logging.getLogger(__name__)
__all__ = ["WebService", "build_app"]
def build_app(app: App, config: WebConfig, config_path: Path) -> tuple[FastAPI, StateHub]:
"""Assemble the ASGI app. Separate from the service so tests can drive it directly."""
hub = StateHub(app)
api = FastAPI(title="MusicMouse", docs_url="/api/docs", openapi_url="/api/openapi.json")
api.include_router(build_router(app, hub, config_path))
if config.static_dir is not None:
if config.static_dir.is_dir():
# Mounted last and at the root so every /api route still wins; html=True
# falls back to index.html, which is what a client-side router needs.
api.mount("/", StaticFiles(directory=config.static_dir, html=True), name="web")
else:
_log.warning(
"web.static_dir %s does not exist; serving the API only "
"(run `npm run build` in web/, or drop the setting)",
config.static_dir,
)
return api, hub
class WebService:
name = "web"
def __init__(self, app: App, config: WebConfig, config_path: Path) -> None:
self.config = config
self.api, self.hub = build_app(app, config, config_path)
async def run(self) -> None:
self.hub.start()
server = uvicorn.Server(
uvicorn.Config(
self.api,
host=self.config.host,
port=self.config.port,
# Access logs for a progress-bar poll every 500ms are noise.
access_log=False,
log_level="warning",
)
)
_log.info("Web front-end on http://%s:%d", self.config.host, self.config.port)
# serve() borrows SIGINT for a graceful shutdown and then re-raises it, so
# Ctrl-C still reaches __main__'s KeyboardInterrupt handler afterwards.
position = asyncio.create_task(self.hub.run(), name="web-position")
serving = asyncio.create_task(server.serve(), name="web-serve")
try:
await asyncio.shield(serving)
except asyncio.CancelledError:
# Shutdown reaches us as a cancellation. Cancelling uvicorn mid-accept
# would leave its listening socket to the garbage collector, so ask it to
# wind down and wait for it to let the port go.
server.should_exit = True
with contextlib.suppress(asyncio.CancelledError, Exception):
await asyncio.wait_for(serving, timeout=5.0)
raise
finally:
position.cancel()
serving.cancel()
self.hub.stop()

View File

@@ -0,0 +1,76 @@
"""Volume as a percentage, and the handful of settings parent mode may change.
Two jobs that both come down to "keep the config file's numbers out of the child's UI".
The volume mapping lives here rather than in the frontend so that MQTT, the rotary
encoder and the firmware carry on speaking device units, unaware that a browser is
using a different scale.
"""
from __future__ import annotations
import logging
import os
from pathlib import Path
from ruamel.yaml import YAML
from musicmouse.config import GeneralConfig
from musicmouse.services.web.schemas import SettingsIn, SettingsOut
_log = logging.getLogger(__name__)
__all__ = ["read_settings", "to_device_volume", "to_percent", "write_settings"]
def to_device_volume(percent: int, general: GeneralConfig) -> int:
"""Map 0..100 onto the configured range, so 100 % is exactly ``max_volume``."""
span = general.max_volume - general.min_volume
return general.min_volume + round(span * max(0, min(100, percent)) / 100)
def to_percent(volume: int, general: GeneralConfig) -> int:
"""The inverse, for the state snapshot."""
span = general.max_volume - general.min_volume
if span <= 0:
return 100
return max(0, min(100, round((volume - general.min_volume) / span * 100)))
def read_settings(general: GeneralConfig) -> SettingsOut:
return SettingsOut(
min_volume=general.min_volume,
max_volume=general.max_volume,
initial_volume=general.initial_volume,
volume_increment=general.volume_increment,
button_leds_brightness=general.button_leds_brightness,
)
def write_settings(path: Path, settings: SettingsIn) -> None:
"""Patch the settings into ``config.yml`` in place.
Loaded round-trip rather than with the safe loader :mod:`musicmouse.config` uses,
so the file keeps its comments, ordering and formatting - a config that explains
itself is worth more than one this could rewrite from scratch. Written through a
sibling temp file so an interrupted save cannot truncate the real one.
"""
yaml = YAML(typ="rt")
yaml.preserve_quotes = True
with path.open(encoding="utf-8") as handle:
document = yaml.load(handle)
general = document["general"]
for key, value in settings.model_dump().items():
general[key] = value
temp = path.with_name(f"{path.name}.tmp{os.getpid()}")
try:
with temp.open("w", encoding="utf-8") as handle:
yaml.dump(document, handle)
temp.replace(path)
except BaseException:
temp.unlink(missing_ok=True)
raise
_log.info("Wrote settings to %s", path)

View File

@@ -0,0 +1,47 @@
"""One place that answers "what is the mouse doing right now?".
Assembled on demand rather than accumulated from events. The join between what is
playing and which figure is on the reader is the same one the MQTT player entity makes:
read ``mouse.active_figure``, do not keep a second copy of it that can drift.
"""
from __future__ import annotations
from musicmouse.app import App
from musicmouse.services.web.schemas import ConnectionOut, PlayerStateOut
from musicmouse.services.web.settings import to_percent
__all__ = ["snapshot"]
def snapshot(app: App) -> PlayerStateOut:
player = app.player
playlist = player.playlist
album = app.library.get(playlist.album_id if playlist else None)
index = player.track_index
# The player only knows a file path, and a tagged file's name is rarely its title
# ("01 - So ein schoener Tag.mp3"). The library read the real one at scan time.
indexed = (
album.tracks[index] if album and 0 <= index < len(album.tracks) else None
)
track = player.current_track
return PlayerStateOut(
playing=player.is_playing,
album_id=album.id if album else None,
album_title=album.title if album else (playlist.name if playlist else None),
artist=album.artist if album else None,
kind=album.kind if album else None,
track_index=index,
track_title=indexed.title if indexed else (track.title if track else None),
track_count=len(playlist) if playlist else 0,
position=player.position,
# Likewise the duration: the player only has one once libVLC has opened the file.
duration=(indexed.duration if indexed else 0.0) or player.duration,
volume=to_percent(player.volume, app.config.general),
active_figure=app.mouse.active_figure,
connected=ConnectionOut(
firmware=app.mouse.connected,
mqtt=app.state.mqtt_connected,
),
)

View File

@@ -24,10 +24,12 @@ from musicmouse.events import (
ButtonEvent,
NextTrackRequested,
PauseRequested,
PlayAlbumRequested,
PlayRequested,
PrevTrackRequested,
RfidTokenRead,
RotaryTurned,
SeekRequested,
SetVolumeRequested,
TouchButtonPressed,
TouchButtonReleased,
@@ -169,6 +171,25 @@ class SimulatorDriver:
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:
@@ -213,6 +234,13 @@ class SimulatorDriver:
return self.app.mouse.active_figure or "none"
case "playlist":
return player.playlist.name if player.playlist else "none"
case "album":
album = self.app.library.get(
player.playlist.album_id if player.playlist else None
)
return album.title if album else "none"
case "position":
return f"{player.position:.1f}"
case "track":
return str(player.track_index)
case "title":
@@ -227,8 +255,8 @@ class SimulatorDriver:
return type(effect).__name__ if effect is not None else "none"
case _:
raise ScriptError(
f"unknown property {key!r} (try: playing, figure, playlist, track, "
f"title, volume, brightness, ring, mouse, shelf)"
f"unknown property {key!r} (try: playing, figure, playlist, album, "
f"track, title, position, volume, brightness, ring, mouse, shelf)"
)
# ------------------------------------------------------------------ scripts
@@ -275,6 +303,10 @@ class SimulatorDriver:
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", []):

View File

@@ -14,7 +14,6 @@ import logging
from musicmouse.bus import EventBus
from musicmouse.clock import Clock, RealClock
from musicmouse.devices.player import PlayerBase
from musicmouse.events import EventSource
from musicmouse.media import Playlist
_log = logging.getLogger(__name__)
@@ -45,6 +44,20 @@ class FakePlayer(PlayerBase):
self._timer: asyncio.Task[None] | None = None
self.closed = False
# -------------------------------------------------------------------- state
@property
def position(self) -> float:
"""Derived from the clock, so it is exact under ``FakeClock`` too."""
remaining = self._remaining
if self._started_at is not None:
remaining = max(0.0, remaining - (self._clock.now() - self._started_at))
return max(0.0, self.track_duration - remaining)
@property
def duration(self) -> float:
return self.track_duration if self._playlist else 0.0
# ------------------------------------------------------------------ actions
def set_playlist(self, playlist: Playlist) -> None:
@@ -64,11 +77,14 @@ class FakePlayer(PlayerBase):
self._start_timer()
def play_from_start(self) -> None:
self.play_track(0)
def play_track(self, index: int) -> None:
if self._playlist is None or not self._playlist:
_log.warning("Nothing to play: the playlist is empty")
return
self._cancel_timer()
self._set_index(0)
self._set_index(max(0, min(index, len(self._playlist) - 1)))
self._remaining = self.track_duration
self._set_playing(True)
self._start_timer()
@@ -90,15 +106,12 @@ class FakePlayer(PlayerBase):
def previous_track(self) -> None:
self._skip(-1)
def set_volume(self, volume: int, *, source: EventSource = "system") -> None:
clamped = self._clamp(volume)
if clamped == self._volume:
def seek(self, position: float) -> None:
if self._playlist is None:
return
self._volume = clamped
self._announce_volume(source)
def change_volume(self, delta: int, *, source: EventSource = "system") -> None:
self.set_volume(self._volume + delta, source=source)
self._remaining = max(0.0, self.track_duration - max(0.0, position))
if self._playing:
self._start_timer()
async def run(self) -> None:
return

View File

@@ -12,8 +12,9 @@ from dataclasses import dataclass
from musicmouse.app import App
from musicmouse.bus import EventBus
from musicmouse.clock import Clock, FakeClock
from musicmouse.config import Config, build_playlists
from musicmouse.config import Config
from musicmouse.devices.mouse import MusicMouseDevice
from musicmouse.library import MusicLibrary
from musicmouse.reactions import register_all
from musicmouse.simulator.driver import SimulatorDriver
from musicmouse.simulator.fake_player import DEFAULT_TRACK_DURATION, FakePlayer
@@ -41,6 +42,7 @@ async def build_simulation(
*,
clock: Clock | None = None,
track_duration: float = DEFAULT_TRACK_DURATION,
library: MusicLibrary | None = None,
) -> Simulation:
bus = EventBus()
await bus.start()
@@ -59,12 +61,22 @@ async def build_simulation(
**FakePlayer.volume_kwargs(config.general),
)
if library is None:
library_config = config.general.library
library = await MusicLibrary.build(
library_config.root,
library_config.cache,
frozenset(config.general.audio_extensions),
figure_kinds=config.figure_kinds,
)
app = App(
config=config,
bus=bus,
mouse=mouse,
player=player,
playlists=build_playlists(config),
library=library,
playlists=library.figure_playlists(),
clock=clock,
)
register_all(bus, app)

View File

@@ -5,14 +5,22 @@ description = "Host backend for the MusicMouse RFID music player"
requires-python = ">=3.13"
dependencies = [
"aiomqtt>=2.0",
"fastapi>=0.115",
"mutagen>=1.47",
"pillow>=10.4",
"pydantic>=2.7",
"pyserial-asyncio>=0.6",
"python-vlc>=3.0",
"ruamel.yaml>=0.18",
"uvicorn>=0.30",
# uvicorn ships no websocket implementation of its own; without this the
# state-push upgrade is answered with a 404 and the UI never updates.
"websockets>=13",
]
[project.optional-dependencies]
dev = ["mypy>=1.10", "pytest-asyncio>=0.23", "pytest>=8.0", "ruff>=0.5"]
# httpx2 is what starlette.testclient wants now; plain httpx is deprecated there.
dev = ["httpx2>=2.12", "mypy>=1.10", "pytest-asyncio>=0.23", "pytest>=8.0", "ruff>=0.5"]
[project.scripts]
musicmouse = "musicmouse.__main__:main"

View File

@@ -0,0 +1,52 @@
# The web front-end's path: any album, not just the five with a figure.
# Run it by hand with:
# python -m musicmouse --config <config.yml> --simulate --script scenarios/web_play.txt
expect playing false
expect ring none
# Starting an album that no figure owns lights the strips differently from a figure -
# a slow three-quarter circle in the album's own colour, so the shelf shows which way
# the mouse was started.
album Kinderparty Lieder
expect playing true
expect album Kinderparty Lieder
expect track 0
expect ring EffectCircularConfig
expect mouse EffectCircularConfig
expect shelf EffectCircularConfig
expect brightness 0.50
# Transport works the same whoever asked for it.
next
expect track 1
wait 6s
expect playing false
expect ring EffectReverseSwipe
# A figure still wins: its animation owns the strips while it is on the reader.
place fuchs
expect figure fuchs
expect album Fuchs
expect ring EffectSwipeAndChange
# ...and the web animation does not fight it.
album Eule
expect album Eule
expect ring EffectSwipeAndChange
remove
expect playing false
# Starting a figure's album from the web is the same playlist the figure plays.
album Fuchs
expect playlist fuchs
expect playing true
seek 2s
expect position 2.0
pause
expect playing false
expect ring EffectReverseSwipe

View File

@@ -6,29 +6,93 @@ from typing import Any
import pytest
from ruamel.yaml import YAML
#: Half a second of silence, so ``mutagen`` reports a real duration and tags can be
#: written onto a file that is actually an MP3. Generated once with ffmpeg.
SILENCE = Path(__file__).parent / "data" / "silence.mp3"
VALID_CONFIG: dict[str, Any] = {
"general": {
"figure_folder": "music",
"library": {"root": "music", "cache": ".cache"},
"serial_port": "/dev/ttyUSB0",
"alsa_device": "simulate",
"min_volume": 0,
"max_volume": 60,
"initial_volume": 40,
},
"figures": {
"fuchs": {"id": "04a1b2c3d4", "colors": ["#ff6600", "#ffcc00", "#331100", "wff"]},
"eule": {"id": "04b2c3d4e5", "colors": ["#3355ff", "#66aaff", "#001133", "#ffffff"]},
# One of each, so the figure-kind branch is exercised by every fixture.
"eule": {
"id": "04b2c3d4e5",
"colors": ["#3355ff", "#66aaff", "#001133", "#ffffff"],
"kind": "book",
},
},
}
def write_track(path: Path, **tags: str) -> Path:
"""Copy the silence fixture to ``path`` and stamp the given easy-mode ID3 tags."""
import mutagen
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(SILENCE.read_bytes())
if tags:
audio = mutagen.File(path, easy=True)
if audio.tags is None:
audio.add_tags()
for key, value in tags.items():
audio[key] = value
audio.save()
return path
@pytest.fixture
def config_dir(tmp_path: Path) -> Path:
"""A directory with a valid ``music/`` tree and two figures' worth of tracks."""
"""A directory holding a small but structurally real music library.
Two figure folders (matching ``VALID_CONFIG``), one music album, one audiobook and
one podcast - enough for every branch the scanner has.
"""
root = tmp_path / "music"
for figure, tracks in (("fuchs", 3), ("eule", 2)):
folder = tmp_path / "music" / figure
folder.mkdir(parents=True)
for index in range(tracks):
(folder / f"{index:02d} - track.mp3").write_bytes(b"")
write_track(
root / "Figuren" / figure / f"{index:02d} - track.mp3",
title=f"Track {index}",
album="Waldlieder",
albumartist="Die Tiere",
)
for index in range(2):
write_track(
root / "Musik" / "Kinderparty - Kinderparty Lieder" / f"{index:02d} - lied.mp3",
title=f"Lied {index}",
album="Kinderparty Lieder",
albumartist="Kinderparty",
)
for index in range(2):
write_track(
root / "Hörbücher" / "Conni - Conni in den Bergen" / f"{index:02d} - teil.mp3",
title=f"Teil {index}",
album="Conni in den Bergen",
albumartist="Conni, Julia Boehme",
)
podcast = root / "Kinderpodcasts" / "Wissen macht Ah"
for date, title in (("20240101", "Alt"), ("20260101", "Neu")):
write_track(
podcast / f"{date} - {title}.mp3",
title=title,
album="Wissen macht Ah! - Podcast",
albumartist="Ein Name, Noch Einer, Und Einer",
)
# The two kinds of scratch file a podcast downloader leaves behind.
(podcast / "archive.json").write_text("[]")
(podcast / ".podcast-dl-abc.download.tmp").write_bytes(b"")
return tmp_path

Binary file not shown.

View File

@@ -7,7 +7,7 @@ from typing import Any
import pytest
from musicmouse.color import ColorRGBW, parse_color
from musicmouse.config import ConfigError, build_playlists, load_config
from musicmouse.config import ConfigError, load_config
from tests.conftest import VALID_CONFIG, write_config
@@ -27,57 +27,29 @@ def test_loads_valid_config(config_dir: Path) -> None:
assert config.general.max_volume == 60
def test_figure_folder_resolves_relative_to_the_config_file(config_dir: Path) -> None:
def test_library_root_resolves_relative_to_the_config_file(config_dir: Path) -> None:
config = load_config(write_config(config_dir, VALID_CONFIG))
assert config.general.figure_folder == (config_dir / "music").resolve()
assert config.general.library.root == (config_dir / "music").resolve()
assert config.general.library.figure_folder == (config_dir / "music" / "Figuren").resolve()
assert config.folder_for("fuchs").name == "fuchs"
def test_tag_map_and_playlists(config_dir: Path) -> None:
def test_tag_map(config_dir: Path) -> None:
config = load_config(write_config(config_dir, VALID_CONFIG))
assert config.tag_map == {
bytes.fromhex("04a1b2c3d4"): "fuchs",
bytes.fromhex("04b2c3d4e5"): "eule",
}
playlists = build_playlists(config)
assert [t.path.name for t in playlists["fuchs"].tracks] == [
"00 - track.mp3",
"01 - track.mp3",
"02 - track.mp3",
]
assert len(playlists["eule"]) == 2
def test_playlist_is_alphabetical_regardless_of_creation_order(config_dir: Path) -> None:
folder = config_dir / "music" / "fuchs"
for name in ("zz last.mp3", "aa first.mp3"):
(folder / name).write_bytes(b"")
playlists = build_playlists(load_config(write_config(config_dir, VALID_CONFIG)))
names = [t.path.name for t in playlists["fuchs"].tracks]
assert names == sorted(names)
assert names[0] == "00 - track.mp3"
def test_non_audio_files_are_ignored(config_dir: Path) -> None:
(config_dir / "music" / "eule" / "cover.jpg").write_bytes(b"")
(config_dir / "music" / "eule" / "notes.txt").write_bytes(b"")
playlists = build_playlists(load_config(write_config(config_dir, VALID_CONFIG)))
assert len(playlists["eule"]) == 2
def test_missing_figure_folder_warns_but_does_not_fail(
config_dir: Path, caplog: pytest.LogCaptureFixture
) -> None:
data = copy.deepcopy(VALID_CONFIG)
data["figures"]["neu"] = {"id": "0400000001", "colors": ["#111111"] * 4}
def test_web_section_is_optional(config_dir: Path) -> None:
assert load_config(write_config(config_dir, VALID_CONFIG)).general.web is None
data = _config(web={"port": 9000, "static_dir": "dist"})
config = load_config(write_config(config_dir, data))
playlists = build_playlists(config)
assert len(playlists["neu"]) == 0
assert "no media folder" in caplog.text
assert config.general.web is not None
assert config.general.web.port == 9000
assert config.general.web.static_dir == (config_dir / "dist").resolve()
# --------------------------------------------------------------------- error paths
@@ -147,9 +119,9 @@ def test_initial_volume_must_lie_in_range(config_dir: Path) -> None:
assert "must lie between" in message
def test_missing_figure_folder_is_an_error(config_dir: Path) -> None:
message = _error(config_dir, _config(figure_folder="does-not-exist"))
assert "general.figure_folder" in message
def test_missing_library_root_is_an_error(config_dir: Path) -> None:
message = _error(config_dir, _config(library={"root": "does-not-exist"}))
assert "general.library.root" in message
assert "no such directory" in message

View File

@@ -0,0 +1,309 @@
"""The library scan: what ends up in the index, and what stays out of it."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from musicmouse.config import DEFAULT_AUDIO_EXTENSIONS, load_config
from musicmouse.library import Album, MusicLibrary
from musicmouse.library.analysis import ANALYZER_VERSION, BeatGrid, TrackAnalysis
from musicmouse.library.cache import LibraryCache
from musicmouse.library.colors import colors_from_id
from musicmouse.library.models import album_id
from tests.conftest import VALID_CONFIG, write_config, write_track
EXTENSIONS = frozenset(DEFAULT_AUDIO_EXTENSIONS)
async def build(config_dir: Path) -> MusicLibrary:
config = load_config(write_config(config_dir, VALID_CONFIG))
return await MusicLibrary.build(
config.general.library.root,
config.general.library.cache,
EXTENSIONS,
figure_kinds=config.figure_kinds,
)
def album_named(library: MusicLibrary, title: str) -> Album:
return next(album for album in library.albums if album.title == title)
# ------------------------------------------------------------------------ structure
async def test_every_section_is_scanned(config_dir: Path) -> None:
library = await build(config_dir)
assert {album.title for album in library.albums} == {
"Fuchs",
"Eule",
"Kinderparty Lieder",
"Conni in den Bergen",
"Wissen macht Ah",
}
async def test_figure_playlists_are_keyed_by_figure_name(config_dir: Path) -> None:
library = await build(config_dir)
playlists = library.figure_playlists()
assert set(playlists) == {"fuchs", "eule"}
assert len(playlists["fuchs"]) == 3
assert [track.path.name for track in playlists["fuchs"].tracks] == [
"00 - track.mp3",
"01 - track.mp3",
"02 - track.mp3",
]
# The playlist carries the album it came from, which is how a front-end answers
# "what is playing?" without keeping its own copy of that state.
assert playlists["fuchs"].album_id == album_named(library, "Fuchs").id
async def test_music_takes_its_title_and_artist_from_the_tags(config_dir: Path) -> None:
album = album_named(await build(config_dir), "Kinderparty Lieder")
assert album.kind == "music"
assert album.artist == "Kinderparty"
# Music groups by artist, so it has no series.
assert album.series is None
assert album.category == "Kinderparty"
async def test_audiobooks_group_by_the_name_before_the_comma(config_dir: Path) -> None:
album = album_named(await build(config_dir), "Conni in den Bergen")
assert album.kind == "book"
# album_artist is a credit list; only the first name is the character.
assert album.artist == "Conni, Julia Boehme"
assert album.series == "Conni"
assert album.category == "Conni"
async def test_podcasts_are_named_after_the_folder_not_the_tags(config_dir: Path) -> None:
album = album_named(await build(config_dir), "Wissen macht Ah")
# The tags say album="Wissen macht Ah! - Podcast" and artist=<six presenters>.
assert album.kind == "book"
assert album.artist == "Wissen macht Ah"
assert album.series == "Wissen macht Ah"
async def test_podcast_episodes_are_newest_first(config_dir: Path) -> None:
album = album_named(await build(config_dir), "Wissen macht Ah")
assert [track.title for track in album.tracks] == ["Neu", "Alt"]
async def test_other_sections_keep_filename_order(config_dir: Path) -> None:
album = album_named(await build(config_dir), "Kinderparty Lieder")
assert [track.path.name for track in album.tracks] == [
"00 - lied.mp3",
"01 - lied.mp3",
]
async def test_figures_are_marked_and_others_are_not(config_dir: Path) -> None:
library = await build(config_dir)
assert album_named(library, "Fuchs").figure == "fuchs"
assert album_named(library, "Kinderparty Lieder").figure is None
async def test_durations_come_from_the_files(config_dir: Path) -> None:
album = album_named(await build(config_dir), "Eule")
assert all(track.duration > 0 for track in album.tracks)
assert album.duration == pytest.approx(sum(t.duration for t in album.tracks))
# -------------------------------------------------------------------- what is skipped
async def test_scratch_files_never_reach_a_playlist(config_dir: Path) -> None:
album = album_named(await build(config_dir), "Wissen macht Ah")
names = {track.path.name for track in album.tracks}
assert "archive.json" not in names
assert not any(name.startswith(".") for name in names)
assert len(album.tracks) == 2
async def test_unknown_top_level_folders_are_ignored(config_dir: Path) -> None:
stray = config_dir / "music" / "Sonstiges" / "Irgendwas"
write_track(stray / "01 - x.mp3", title="X")
library = await build(config_dir)
assert not any(album.section == "Sonstiges" for album in library.albums)
async def test_a_missing_section_warns_rather_than_failing(
config_dir: Path, caplog: pytest.LogCaptureFixture
) -> None:
import shutil
shutil.rmtree(config_dir / "music" / "Musik")
library = await build(config_dir)
assert "Musik" in caplog.text
assert len(library.albums) == 4
async def test_an_empty_album_folder_is_not_an_album(config_dir: Path) -> None:
(config_dir / "music" / "Musik" / "Leer").mkdir()
library = await build(config_dir)
assert not any(album.title == "Leer" for album in library.albums)
# ------------------------------------------------------------------------- colours
async def test_every_album_has_three_colours(config_dir: Path) -> None:
library = await build(config_dir)
for album in library.albums:
assert len(album.colors) == 3
assert all(colour.startswith("#") and len(colour) == 7 for colour in album.colors)
async def test_colours_are_synthesised_when_there_is_no_cover(config_dir: Path) -> None:
album = album_named(await build(config_dir), "Fuchs")
assert album.cover is None
assert album.colors == colors_from_id(album.id)
async def test_a_cover_file_is_found_and_used(config_dir: Path) -> None:
from PIL import Image
folder = config_dir / "music" / "Musik" / "Kinderparty - Kinderparty Lieder"
Image.new("RGB", (32, 32), (200, 40, 30)).save(folder / "cover.jpg")
album = album_named(await build(config_dir), "Kinderparty Lieder")
assert album.cover == folder / "cover.jpg"
# A solid red cover has one usable colour; the rest fall back to the synthesised
# palette rather than repeating it.
assert album.colors[0] != colors_from_id(album.id)[0]
# --------------------------------------------------------------------------- cache
async def test_a_second_build_reuses_the_index(config_dir: Path) -> None:
first = await build(config_dir)
second = await build(config_dir)
assert {a.id for a in first.albums} == {a.id for a in second.albums}
assert (config_dir / ".cache" / "index.json").is_file()
async def test_a_changed_folder_is_rescanned(config_dir: Path) -> None:
library = await build(config_dir)
assert len(album_named(library, "Eule").tracks) == 2
write_track(config_dir / "music" / "Figuren" / "eule" / "99 - neu.mp3", title="Neu")
library = await build(config_dir)
assert len(album_named(library, "Eule").tracks) == 3
async def test_a_rescan_leaves_analysis_alone(config_dir: Path) -> None:
"""The whole reason the cache is a directory rather than one file."""
library = await build(config_dir)
album = album_named(library, "Eule")
from musicmouse.library.models import track_key
key = track_key(album.tracks[0].path)
library.cache.store_analysis(key, TrackAnalysis(version=ANALYZER_VERSION, tempo=128.0))
library.cache.store_beats(key, BeatGrid((0.5, 1.0), (1.0, 0.5)))
# Force a full rescan by dropping the cheap part of the cache.
(config_dir / ".cache" / "index.json").unlink()
library = await build(config_dir)
album = album_named(library, "Eule")
assert album.tracks[0].analysis is not None
assert album.tracks[0].analysis.tempo == 128.0
grid = library.beats(album.id, 0)
assert grid is not None
assert grid.times == (0.5, 1.0)
async def test_analysis_survives_an_analyzer_that_grew_a_field(tmp_path: Path) -> None:
cache = LibraryCache(tmp_path / "cache")
cache.prepare()
(cache.analysis / "abc.json").write_text(
json.dumps({"version": 1, "tempo": 90.0, "danceability": 0.7})
)
analysis = cache.load_analysis("abc")
assert analysis is not None
assert analysis.tempo == 90.0
async def test_a_corrupt_index_is_rebuilt_rather_than_fatal(config_dir: Path) -> None:
await build(config_dir)
(config_dir / ".cache" / "index.json").write_text("{ not json")
library = await build(config_dir)
assert len(library.albums) == 5
def test_album_ids_are_stable_and_path_derived(tmp_path: Path) -> None:
first = album_id(tmp_path, tmp_path / "Musik" / "Ein Album")
second = album_id(tmp_path, tmp_path / "Musik" / "Ein Album")
other = album_id(tmp_path, tmp_path / "Musik" / "Ein Anderes")
assert first == second
assert first != other
# -------------------------------------------------------------------- figure kinds
async def test_a_figure_is_music_unless_the_config_says_otherwise(
config_dir: Path,
) -> None:
"""A figure folder is named after the figurine, so nothing on disk says what it is."""
library = await build(config_dir)
assert album_named(library, "Fuchs").kind == "music"
assert album_named(library, "Eule").kind == "book"
async def test_a_book_figure_groups_with_the_audiobooks(config_dir: Path) -> None:
"""The bug this fixed: a figure holding an audiobook landed in its artist's shelf
as music, so the category showed "4 Alben" and was drawn like one."""
for index in range(2):
write_track(
config_dir / "music" / "Figuren" / "eule" / f"{index:02d} - track.mp3",
title=f"Teil {index}",
album="Conni in den Bergen",
albumartist="Conni, Julia Boehme",
)
library = await build(config_dir)
figure = album_named(library, "Eule")
assert figure.kind == "book"
# Books group by series, which a music album does not have at all.
assert figure.series == "Conni"
assert figure.category == "Conni"
conni = [album for album in library.albums if album.category == "Conni"]
assert len(conni) == 2
assert {album.kind for album in conni} == {"book"}
async def test_a_music_figure_has_no_series(config_dir: Path) -> None:
figure = album_named(await build(config_dir), "Fuchs")
assert figure.series is None
assert figure.category == "Die Tiere"
async def test_a_figures_kind_survives_the_cache(config_dir: Path) -> None:
"""It cannot be re-derived on load: the section says "Figuren", not what is in it.
The index used to rebuild `kind` from the section, so a book figure came back as
music on the second boot - and the browse view drew it square.
"""
first = await build(config_dir)
assert album_named(first, "Eule").kind == "book"
assert (config_dir / ".cache" / "index.json").is_file()
# Nothing on disk changed, so this run is served entirely from the index.
second = await build(config_dir)
eule = album_named(second, "Eule")
assert eule.kind == "book"
assert eule.series is not None
assert album_named(second, "Fuchs").kind == "music"
assert album_named(second, "Fuchs").series is None

View File

@@ -1,7 +1,9 @@
"""Tests for the shared player behaviour, exercised through FakePlayer.
VlcPlayer adds only the libVLC bindings on top of PlayerBase; it needs a real audio
device and is covered by the on-device checklist, not here.
device and is covered by the on-device checklist, not here - except for how it works
out which track is playing, which is stubbed out at the bottom of this file because
getting it wrong is invisible until a front-end displays it.
"""
from __future__ import annotations
@@ -13,7 +15,7 @@ import pytest
from musicmouse.bus import EventBus
from musicmouse.clock import FakeClock
from musicmouse.devices.player import Player, VlcPlayer
from musicmouse.devices.player import Player, PlayerBase, VlcPlayer
from musicmouse.events import (
Event,
PlaybackChanged,
@@ -295,3 +297,69 @@ def test_fake_player_satisfies_the_player_protocol(player: FakePlayer) -> None:
def _vlc_player_satisfies_the_player_protocol(real: VlcPlayer) -> Player:
"""Checked by mypy, not at runtime: VlcPlayer needs libVLC to instantiate."""
return real
# --------------------------------------------------------- track index without libVLC
class _FakeMedia:
def __init__(self, mrl: str) -> None:
self._mrl = mrl
def get_mrl(self) -> str:
return self._mrl
class _FakeMediaPlayer:
"""Just enough of libVLC's media player to exercise index syncing."""
def __init__(self) -> None:
self.media: _FakeMedia | None = None
def get_media(self) -> _FakeMedia | None:
return self.media
def _vlc_like(bus: EventBus, playlist: Playlist) -> VlcPlayer:
"""A VlcPlayer with its libVLC parts stubbed out, without calling __init__."""
player = object.__new__(VlcPlayer)
PlayerBase.__init__(player, bus)
player._media_player = _FakeMediaPlayer() # type: ignore[assignment]
player._playlist = playlist
player._mrl_to_index = {f"file://{track.path}": i for i, track in enumerate(playlist.tracks)}
return player
async def test_the_track_index_is_read_back_off_the_player(bus: EventBus) -> None:
"""Not taken from the event payload.
``MediaListPlayerNextItemSet`` carries a bare int rather than a Media on some
libVLC builds, and trusting it left the index pinned at zero: the UI showed track 1
of an album that was audibly on track 3.
"""
playlist = Playlist("test", tuple(Track(Path(f"/music/{i}.mp3")) for i in range(3)))
player = _vlc_like(bus, playlist)
events: list[Event] = []
bus.subscribe(TrackChanged, events.append)
player._media_player.media = _FakeMedia("file:///music/2.mp3") # type: ignore[attr-defined]
player._on_next_item(object())
await bus.drain()
assert player.track_index == 2
assert player.current_track is not None
assert player.current_track.path.name == "2.mp3"
assert len(events) == 1
async def test_an_unknown_media_leaves_the_index_alone(bus: EventBus) -> None:
playlist = Playlist("test", tuple(Track(Path(f"/music/{i}.mp3")) for i in range(3)))
player = _vlc_like(bus, playlist)
player._media_player.media = None # type: ignore[attr-defined]
player._on_next_item(object())
player._media_player.media = _FakeMedia("file:///elsewhere/x.mp3") # type: ignore[attr-defined]
player._on_next_item(object())
await bus.drain()
assert player.track_index == 0

View File

@@ -0,0 +1,176 @@
"""Parent mode: the small set of settings the UI may change, and how they are saved."""
from __future__ import annotations
from collections.abc import AsyncIterator
from pathlib import Path
import httpx2
import pytest
from musicmouse.config import WebConfig, load_config
from musicmouse.services.web.service import build_app
from musicmouse.services.web.settings import to_device_volume, to_percent
from musicmouse.simulator.harness import Simulation, build_simulation
from tests.conftest import VALID_CONFIG, write_config
SETTINGS = {
"min_volume": 0,
"max_volume": 60,
"initial_volume": 40,
"volume_increment": 5,
"button_leds_brightness": 0.5,
}
#: A config file with comments, to prove saving does not flatten them.
COMMENTED = """\
# The MusicMouse config.
general:
library:
root: music # where the shelves live
cache: .cache
serial_port: "/dev/ttyUSB0"
alsa_device: simulate # no sound from the test suite, please
# Volume, 0..100.
min_volume: 0
max_volume: 60
initial_volume: 40
figures:
fuchs:
id: "04a1b2c3d4"
colors: ["#ff6600", "#ffcc00", "#331100", "wff"]
eule:
id: "04b2c3d4e5"
colors: ["#3355ff", "#66aaff", "#001133", "#ffffff"]
"""
@pytest.fixture
async def sim(config_dir: Path) -> AsyncIterator[Simulation]:
config = load_config(write_config(config_dir, VALID_CONFIG))
simulation = await build_simulation(config)
try:
yield simulation
finally:
await simulation.aclose()
@pytest.fixture
async def client(sim: Simulation, config_dir: Path) -> AsyncIterator[httpx2.AsyncClient]:
path = config_dir / "config.yml"
path.write_text(COMMENTED, encoding="utf-8")
api, hub = build_app(sim.app, WebConfig(), path)
hub.start()
transport = httpx2.ASGITransport(app=api)
try:
async with httpx2.AsyncClient(transport=transport, base_url="http://mouse") as http:
yield http
finally:
hub.stop()
# ------------------------------------------------------------------------- mapping
def test_full_percent_is_the_configured_ceiling(config_dir: Path) -> None:
general = load_config(write_config(config_dir, VALID_CONFIG)).general
assert to_device_volume(100, general) == 60
assert to_device_volume(0, general) == 0
assert to_device_volume(50, general) == 30
def test_percent_round_trips(config_dir: Path) -> None:
general = load_config(write_config(config_dir, VALID_CONFIG)).general
for percent in (0, 25, 50, 75, 100):
assert to_percent(to_device_volume(percent, general), general) == percent
def test_a_degenerate_range_reads_as_full(config_dir: Path) -> None:
"""min == max is a legal config; it must not divide by zero."""
general = load_config(write_config(config_dir, VALID_CONFIG)).general
general.min_volume = general.max_volume = 40
assert to_percent(40, general) == 100
assert to_device_volume(50, general) == 40
# -------------------------------------------------------------------------- reading
async def test_settings_expose_only_the_editable_subset(client: httpx2.AsyncClient) -> None:
body = (await client.get("/api/settings")).json()
assert set(body) == set(SETTINGS)
assert body["max_volume"] == 60
# -------------------------------------------------------------------------- writing
async def test_saving_keeps_the_file_readable_and_commented(
client: httpx2.AsyncClient, config_dir: Path
) -> None:
response = await client.put("/api/settings", json={**SETTINGS, "max_volume": 45})
assert response.status_code == 200
text = (config_dir / "config.yml").read_text(encoding="utf-8")
assert "max_volume: 45" in text
assert "# The MusicMouse config." in text
assert "# where the shelves live" in text
assert "# Volume, 0..100." in text
assert "# no sound from the test suite, please" in text
# And it still loads.
reloaded = load_config(config_dir / "config.yml")
assert reloaded.general.max_volume == 45
assert set(reloaded.figures) == {"fuchs", "eule"}
async def test_a_new_ceiling_applies_to_the_running_player(
client: httpx2.AsyncClient, sim: Simulation
) -> None:
"""A parent lowering the ceiling expects the next song to be quieter, not the next boot."""
await client.post("/api/volume", json={"percent": 100})
await sim.bus.drain()
assert sim.player.volume == 60
response = await client.put(
"/api/settings", json={**SETTINGS, "max_volume": 30, "initial_volume": 20}
)
assert response.status_code == 200
await sim.bus.drain()
assert sim.player.volume == 30
assert (await client.get("/api/state")).json()["volume"] == 100
async def test_an_inverted_range_is_rejected(client: httpx2.AsyncClient) -> None:
response = await client.put("/api/settings", json={**SETTINGS, "min_volume": 70})
assert response.status_code == 422
assert "min_volume" in response.json()["detail"]
async def test_an_initial_volume_outside_the_range_is_rejected(
client: httpx2.AsyncClient,
) -> None:
response = await client.put("/api/settings", json={**SETTINGS, "initial_volume": 90})
assert response.status_code == 422
assert "initial_volume" in response.json()["detail"]
async def test_out_of_bounds_values_are_rejected_by_the_schema(
client: httpx2.AsyncClient,
) -> None:
assert (
await client.put("/api/settings", json={**SETTINGS, "button_leds_brightness": 5})
).status_code == 422
assert (
await client.put("/api/settings", json={**SETTINGS, "volume_increment": 0})
).status_code == 422
async def test_a_rejected_save_leaves_the_file_alone(
client: httpx2.AsyncClient, config_dir: Path
) -> None:
before = (config_dir / "config.yml").read_text(encoding="utf-8")
await client.put("/api/settings", json={**SETTINGS, "min_volume": 70})
assert (config_dir / "config.yml").read_text(encoding="utf-8") == before

View File

@@ -0,0 +1,128 @@
"""What the app does when the config leaves the hardware out.
Both of these are deliberately warnings rather than errors: a spare machine running
only the web front-end is a supported way to use this, and it should say so once at
startup rather than fail or - worse - look like it is working when it is not.
"""
from __future__ import annotations
import copy
import logging
from pathlib import Path
from typing import Any
import pytest
from musicmouse.__main__ import _build_player, wants_hardware
from musicmouse.bus import EventBus
from musicmouse.clock import RealClock
from musicmouse.config import SIMULATE, Config, ConfigError, load_config
from musicmouse.devices.null_transport import NullTransport
from musicmouse.simulator.fake_player import FakePlayer
from tests.conftest import VALID_CONFIG, write_config
def _config(config_dir: Path, *, drop: tuple[str, ...] = (), **general: Any) -> Config:
data = copy.deepcopy(VALID_CONFIG)
data["general"].update(general)
for key in drop:
data["general"].pop(key, None)
return load_config(write_config(config_dir, data))
# ------------------------------------------------------------------- serial port
def test_a_missing_serial_port_is_an_error(config_dir: Path) -> None:
"""Simulation has to be asked for. A config that lost the line must not boot."""
with pytest.raises(ConfigError) as excinfo:
_config(config_dir, drop=("serial_port",))
assert "general.serial_port" in str(excinfo.value)
assert "required" in str(excinfo.value)
def test_a_missing_alsa_device_is_an_error(config_dir: Path) -> None:
with pytest.raises(ConfigError) as excinfo:
_config(config_dir, drop=("alsa_device",))
assert "general.alsa_device" in str(excinfo.value)
assert "required" in str(excinfo.value)
def test_both_missing_are_reported_together(config_dir: Path) -> None:
with pytest.raises(ConfigError) as excinfo:
_config(config_dir, drop=("serial_port", "alsa_device"))
message = str(excinfo.value)
assert "2 problems" in message
assert "general.serial_port" in message
assert "general.alsa_device" in message
def test_simulate_switches_off_the_serial_link(
config_dir: Path, caplog: pytest.LogCaptureFixture
) -> None:
config = _config(config_dir, serial_port=SIMULATE)
assert config.general.serial_simulated is True
with caplog.at_level(logging.WARNING):
assert wants_hardware(config, False) is False
assert "running without the mouse" in caplog.text
def test_a_real_port_is_used(config_dir: Path) -> None:
config = _config(config_dir)
assert config.general.serial_simulated is False
assert wants_hardware(config, False) is True
def test_the_flag_wins_over_a_configured_port(config_dir: Path) -> None:
assert wants_hardware(_config(config_dir), True) is False
# ------------------------------------------------------------------ audio device
async def test_simulate_gives_a_silent_player(
config_dir: Path, caplog: pytest.LogCaptureFixture
) -> None:
bus = EventBus()
await bus.start()
try:
general = _config(config_dir, alsa_device=SIMULATE).general
assert general.audio_simulated is True
with caplog.at_level(logging.WARNING):
player = _build_player(bus, general, clock=RealClock())
assert isinstance(player, FakePlayer)
assert "alsa_device" in caplog.text
assert "nothing will be audible" in caplog.text
# The clamps still come from the config, so volume behaves the same either way.
assert player.volume == 40
player.set_volume(999)
assert player.volume == 60
finally:
player.close()
await bus.stop()
async def test_an_alsa_device_asks_for_the_real_player(config_dir: Path) -> None:
"""Only that it tries: instantiating VlcPlayer needs libVLC, which CI has not got."""
bus = EventBus()
await bus.start()
try:
general = _config(config_dir, alsa_device="default").general
assert general.audio_simulated is False
with pytest.raises(Exception): # noqa: B017 - ImportError or an OSError from libVLC
_build_player(bus, general, clock=RealClock())
finally:
await bus.stop()
# ---------------------------------------------------------------- null transport
def test_the_null_transport_drops_what_it_is_handed() -> None:
transport = NullTransport()
transport.write(b"\x00\x01\x02")
assert transport.connected is False

View File

@@ -0,0 +1,357 @@
"""The web front-end, driven against the simulator.
Everything below the HTTP layer is production code: the same bus, the same reactions,
the same player interface - only the serial link and VLC are fake.
"""
from __future__ import annotations
import asyncio
import contextlib
from collections.abc import AsyncIterator, Iterator
from pathlib import Path
import httpx2
import pytest
from fastapi import FastAPI
from musicmouse.config import WebConfig, load_config
from musicmouse.services.web.service import build_app
from musicmouse.simulator.harness import Simulation, build_simulation
from tests.conftest import VALID_CONFIG, write_config
from tests.websocket_harness import websocket_connect
TRACK_SECONDS = 10.0
#: Shorthand: every test takes the same client type.
type Client = httpx2.AsyncClient
@pytest.fixture
async def sim(config_dir: Path) -> AsyncIterator[Simulation]:
config = load_config(write_config(config_dir, VALID_CONFIG))
simulation = await build_simulation(config, track_duration=TRACK_SECONDS)
try:
yield simulation
finally:
await simulation.aclose()
@pytest.fixture
def api(sim: Simulation, config_dir: Path) -> Iterator[FastAPI]:
"""The real ASGI app, on the test's own event loop."""
application, hub = build_app(sim.app, WebConfig(), config_dir / "config.yml")
hub.start()
try:
yield application
finally:
hub.stop()
@pytest.fixture
async def client(api: FastAPI) -> AsyncIterator[httpx2.AsyncClient]:
transport = httpx2.ASGITransport(app=api)
async with httpx2.AsyncClient(transport=transport, base_url="http://mouse") as http:
yield http
async def album_by_title(client: httpx2.AsyncClient, title: str) -> dict:
albums = (await client.get("/api/library")).json()["albums"]
return next(album for album in albums if album["title"] == title)
# ------------------------------------------------------------------------- library
async def test_library_lists_every_album_with_its_tracks(client: Client) -> None:
body = (await client.get("/api/library")).json()
titles = {album["title"] for album in body["albums"]}
assert titles == {
"Fuchs",
"Eule",
"Kinderparty Lieder",
"Conni in den Bergen",
"Wissen macht Ah",
}
album = await album_by_title(client, "Conni in den Bergen")
assert album["kind"] == "book"
assert album["category"] == "Conni"
assert len(album["colors"]) == 3
assert [track["title"] for track in album["tracks"]] == ["Teil 0", "Teil 1"]
async def test_a_missing_cover_is_a_404_not_an_error(client: Client) -> None:
album = await album_by_title(client, "Fuchs")
assert album["has_cover"] is False
assert (await client.get(f"/api/albums/{album['id']}/cover")).status_code == 404
async def test_unanalyzed_tracks_report_no_analysis(client: Client) -> None:
album = await album_by_title(client, "Eule")
assert all(track["analysis"] is None for track in album["tracks"])
assert (await client.get(f"/api/tracks/{album['id']}/0/analysis")).status_code == 404
# -------------------------------------------------------------------------- state
async def test_state_starts_idle(client: Client) -> None:
state = (await client.get("/api/state")).json()
assert state["playing"] is False
assert state["album_id"] is None
assert state["active_figure"] is None
assert state["connected"] == {"firmware": True, "mqtt": False}
async def test_play_loads_the_album_and_starts_it(client: Client, sim: Simulation) -> None:
album = await album_by_title(client, "Kinderparty Lieder")
response = await client.post("/api/play", json={"album_id": album["id"]})
assert response.status_code == 204
await sim.bus.drain()
assert sim.player.is_playing
assert sim.player.playlist is not None
assert sim.player.playlist.album_id == album["id"]
assert (await client.get("/api/state")).json()["album_title"] == "Kinderparty Lieder"
async def test_track_titles_come_from_the_tags_not_the_filename(
client: Client, sim: Simulation
) -> None:
"""``01 - So ein schoener Tag.mp3`` is a filename, not a title."""
album = await album_by_title(client, "Fuchs")
await client.post("/api/play", json={"album_id": album["id"]})
await sim.bus.drain()
state = (await client.get("/api/state")).json()
assert state["track_title"] == "Track 0"
assert state["duration"] > 0
async def test_play_can_start_at_a_track(client: Client, sim: Simulation) -> None:
album = await album_by_title(client, "Fuchs")
await client.post("/api/play", json={"album_id": album["id"], "track_index": 2})
await sim.bus.drain()
assert sim.player.track_index == 2
assert (await client.get("/api/state")).json()["track_index"] == 2
async def test_playing_a_figure_album_reuses_the_figure_playlist(
client: Client, sim: Simulation
) -> None:
"""Identity matters: ``play_figure`` resumes on an ``is`` check."""
album = await album_by_title(client, "Fuchs")
await client.post("/api/play", json={"album_id": album["id"]})
await sim.bus.drain()
assert sim.player.playlist is sim.app.playlists["fuchs"]
async def test_playing_an_unknown_album_is_a_404(client: Client) -> None:
response = await client.post("/api/play", json={"album_id": "nope"})
assert response.status_code == 404
async def test_transport_commands_reach_the_player(client: Client, sim: Simulation) -> None:
album = await album_by_title(client, "Fuchs")
await client.post("/api/play", json={"album_id": album["id"]})
await sim.bus.drain()
await client.post("/api/next")
await sim.bus.drain()
assert sim.player.track_index == 1
await client.post("/api/previous")
await sim.bus.drain()
assert sim.player.track_index == 0
await client.post("/api/pause")
await sim.bus.drain()
assert not sim.player.is_playing
await client.post("/api/resume")
await sim.bus.drain()
assert sim.player.is_playing
async def test_seek_moves_the_position(client: Client, sim: Simulation) -> None:
album = await album_by_title(client, "Fuchs")
await client.post("/api/play", json={"album_id": album["id"]})
await sim.bus.drain()
response = await client.post("/api/seek", json={"position": 4.0})
assert response.status_code == 204
await sim.bus.drain()
assert sim.player.position == pytest.approx(4.0, abs=0.1)
async def test_seeking_backwards_is_rejected(client: Client) -> None:
assert (await client.post("/api/seek", json={"position": -1})).status_code == 422
# ------------------------------------------------------------------------- volume
async def test_full_volume_means_the_configured_ceiling(
client: Client, sim: Simulation
) -> None:
"""The child sees 0..100; the config's max of 60 never crosses the API boundary."""
response = await client.post("/api/volume", json={"percent": 100})
assert response.status_code == 204
await sim.bus.drain()
assert sim.player.volume == 60
assert (await client.get("/api/state")).json()["volume"] == 100
async def test_volume_scales_across_the_allowed_range(
client: Client, sim: Simulation
) -> None:
await client.post("/api/volume", json={"percent": 50})
await sim.bus.drain()
assert sim.player.volume == 30
assert (await client.get("/api/state")).json()["volume"] == 50
async def test_volume_steps_are_relative_to_the_percentage(
client: Client, sim: Simulation
) -> None:
await client.post("/api/volume", json={"percent": 50})
await sim.bus.drain()
await client.post("/api/volume", json={"delta_percent": 10})
await sim.bus.drain()
assert (await client.get("/api/state")).json()["volume"] == 60
assert sim.player.volume == 36
async def test_volume_steps_clamp_at_the_ends(client: Client, sim: Simulation) -> None:
await client.post("/api/volume", json={"percent": 95})
await sim.bus.drain()
await client.post("/api/volume", json={"delta_percent": 20})
await sim.bus.drain()
assert (await client.get("/api/state")).json()["volume"] == 100
async def test_the_device_volume_range_is_never_exposed(client: Client) -> None:
state = (await client.get("/api/state")).json()
assert "volume_min" not in state
assert "volume_max" not in state
async def test_volume_needs_one_of_the_two_fields(client: Client) -> None:
assert (await client.post("/api/volume", json={})).status_code == 422
# ---------------------------------------------------------------------- websocket
async def test_a_new_client_is_sent_a_snapshot_before_any_deltas(api: FastAPI) -> None:
"""State events only fire on change, so a tab that connects mid-track needs this."""
async with websocket_connect(api, "/api/ws") as socket:
message = await socket.next_json()
assert message["type"] == "state"
assert message["state"]["playing"] is False
async def test_a_state_change_reaches_every_client(api: FastAPI, sim: Simulation) -> None:
async with websocket_connect(api, "/api/ws") as first, websocket_connect(
api, "/api/ws"
) as second:
await first.next_json()
await second.next_json()
await sim.driver.place("fuchs")
await sim.bus.drain()
for socket in (first, second):
message = await socket.next_json()
assert message["type"] == "state"
assert message["state"]["active_figure"] == "fuchs"
async def test_a_refresh_tells_the_clients_to_reload_the_library(
api: FastAPI, client: Client, sim: Simulation
) -> None:
async with websocket_connect(api, "/api/ws") as socket:
await socket.next_json()
response = await client.post("/api/library/refresh")
assert response.status_code == 202
message = await socket.next_json(timeout=5.0)
assert message["type"] == "library"
assert set(sim.app.playlists) == {"fuchs", "eule"}
# ------------------------------------------------------------------------ figures
async def test_the_web_ui_follows_a_figure_placed_on_the_reader(
client: Client, sim: Simulation
) -> None:
await sim.driver.place("fuchs")
await sim.bus.drain()
state = (await client.get("/api/state")).json()
assert state["active_figure"] == "fuchs"
assert state["playing"] is True
assert state["album_title"] == "Fuchs"
# ------------------------------------------------- through a real uvicorn, not just ASGI
async def test_websockets_survive_a_real_server(sim: Simulation, config_dir: Path) -> None:
"""Boot the actual service and open an actual websocket.
Everything above drives the ASGI app directly, which cannot see whether uvicorn is
able to answer an upgrade at all - and it is not, unless a websocket implementation
is installed alongside it. That gap answered ``/api/ws`` with a 404 in production
while every other test passed.
"""
import json
import socket
import websockets
from musicmouse.services.web.service import WebService
with socket.socket() as probe:
probe.bind(("127.0.0.1", 0))
port = int(probe.getsockname()[1])
service = WebService(
sim.app,
WebConfig(host="127.0.0.1", port=port),
config_dir / "config.yml",
)
server = asyncio.create_task(service.run(), name="web-service")
try:
await _wait_for_port(port)
async with websockets.connect(f"ws://127.0.0.1:{port}/api/ws") as client:
snapshot = json.loads(await asyncio.wait_for(client.recv(), 5.0))
assert snapshot["type"] == "state"
assert snapshot["state"]["playing"] is False
finally:
server.cancel()
with contextlib.suppress(asyncio.CancelledError):
await server
async def _wait_for_port(port: int, timeout: float = 5.0) -> None:
deadline = asyncio.get_running_loop().time() + timeout
while asyncio.get_running_loop().time() < deadline:
try:
reader, writer = await asyncio.open_connection("127.0.0.1", port)
except OSError:
await asyncio.sleep(0.05)
continue
del reader
writer.close()
with contextlib.suppress(Exception):
await writer.wait_closed()
return
raise AssertionError(f"nothing listening on {port} after {timeout}s")

View File

@@ -0,0 +1,68 @@
"""Drive an ASGI websocket endpoint on the caller's own event loop.
Starlette's ``TestClient`` runs the app in a second thread with its own loop, which
would put the bus and the websockets on different loops - so ``bus.emit`` would take
its thread-safe path and a following ``drain()`` could return before the event was even
queued. In the real app there is only ever one loop, and this harness keeps the tests
that way.
"""
from __future__ import annotations
import asyncio
import contextlib
import json
from collections.abc import AsyncIterator
from typing import Any
class WebSocketSession:
def __init__(self) -> None:
self.to_app: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
self.from_app: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
async def receive(self) -> dict[str, Any]:
return await self.to_app.get()
async def send(self, message: dict[str, Any]) -> None:
await self.from_app.put(message)
async def next_json(self, timeout: float = 2.0) -> dict[str, Any]:
"""The next ``websocket.send`` frame, decoded."""
while True:
message = await asyncio.wait_for(self.from_app.get(), timeout)
if message["type"] == "websocket.send":
text: str = message["text"]
return json.loads(text)
@contextlib.asynccontextmanager
async def websocket_connect(app: Any, path: str) -> AsyncIterator[WebSocketSession]:
session = WebSocketSession()
scope = {
"type": "websocket",
"asgi": {"version": "3.0", "spec_version": "2.3"},
"http_version": "1.1",
"scheme": "ws",
"path": path,
"raw_path": path.encode(),
"query_string": b"",
"root_path": "",
"headers": [(b"host", b"testserver")],
"client": ("testclient", 50000),
"server": ("testserver", 80),
"subprotocols": [],
"state": {},
}
await session.to_app.put({"type": "websocket.connect"})
task = asyncio.create_task(app(scope, session.receive, session.send))
accepted = await asyncio.wait_for(session.from_app.get(), 2.0)
assert accepted["type"] == "websocket.accept", accepted
try:
yield session
finally:
await session.to_app.put({"type": "websocket.disconnect", "code": 1000})
task.cancel()
with contextlib.suppress(asyncio.CancelledError, Exception):
await task