From e64378da78311a9827f0732b147123c74fd75074 Mon Sep 17 00:00:00 2001 From: Byron Gamatos Date: Sat, 20 Jun 2026 21:23:48 +0200 Subject: [PATCH] =?UTF-8?q?feat(core):=20consume=20keys.json=20=E2=80=94?= =?UTF-8?q?=20song-level=20key/scale=20track=20(loader=20+=20WS)=20(#528)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The feedpak spec defines keys.json (instrument-independent key/scale-change track, §7.7) but core never loaded it. Add it, mirroring the song_timeline / drum_tab side-file pattern: - lib/sloppak.py: LoadedSloppak gains a `keys` field; a permissive loader reads the manifest `keys:` key, path-safety-checks it, and stores a SANITIZED {version, events:[{t, key, scale?}]} — finite non-bool t (bad-t events dropped, not rewritten to 0), non-empty string key, optional string scale, sorted. Missing / unreadable / malformed -> None, never fatal. int-only version (a float/NaN version can't abort the load). - server.py: stream a `keys` highway-WS message when present + a `has_keys` song_info flag so a consumer can light up a key/scale display. Renderer/HUD surfacing is a thin follow-up; this lands the data plumbing so the highway, plugins, and the upcoming scale-degree (`sd`) annotation can read the active key/mode from the WS. Codex-reviewed (2 rounds: version-int-coercion + bad-t-drop hardening); clean. +7 loader tests (happy path, absent/permissive variants, sanitize/sort, non-int-version no-abort). 150 sloppak/load tests pass. Closes #525. Part of got-feedback/feedback#334. Co-authored-by: Claude Opus 4.8 (1M context) --- lib/sloppak.py | 68 +++++++++++++++++ server.py | 16 ++++ tests/test_sloppak_keys_load.py | 127 ++++++++++++++++++++++++++++++++ 3 files changed, 211 insertions(+) create mode 100644 tests/test_sloppak_keys_load.py diff --git a/lib/sloppak.py b/lib/sloppak.py index ac50785..bc72dd0 100644 --- a/lib/sloppak.py +++ b/lib/sloppak.py @@ -15,6 +15,7 @@ from __future__ import annotations import json import logging +import math import shutil import threading import zipfile @@ -301,6 +302,11 @@ class LoadedSloppak: # When present, its beats/sections take priority over any beats/sections # embedded in the arrangement JSONs. song_timeline: dict | None = None + # Parsed `keys.json` payload (manifest `keys:` key) — a song-level, + # instrument-independent key/scale-change track (spec §7.7). None when + # absent / unreadable / malformed. Streamed over the highway WS as a + # `keys` message; consumers (renderers, plugins) read it from there. + keys: dict | None = None # Maps arrangement id → validated notation payload. None when no # arrangement passed schema validation; a non-empty dict only when at least # one arrangement carried a `notation:` sub-key whose file loaded and passed @@ -681,6 +687,67 @@ def load_song( default_on = bool(default_val) stems.append({"id": sid, "file": sfile, "default": default_on}) + # Optional keys.json — song-level, instrument-independent key/scale track + # (manifest `keys:` key, spec §7.7). Permissive like the other side-files: + # missing / unreadable / malformed -> None, never fatal. Stored as a + # sanitized {version, events:[{t, key, scale?}]} (finite t, non-empty string + # key, sorted) so the highway WS can stream it without re-validating. + keys_data: dict | None = None + keys_rel = manifest.get("keys") + if isinstance(keys_rel, str) and keys_rel: + try: + k_path = (source_dir / keys_rel).resolve() + k_path.relative_to(source_dir.resolve()) + except ValueError: + log.warning("sloppak: keys path %r escapes source_dir — skipped", keys_rel) + k_path = None + except OSError as e: + log.warning("sloppak: keys path resolution failed (%s) — skipped", e) + k_path = None + if k_path is not None and k_path.exists(): + try: + raw = json.loads(k_path.read_text(encoding="utf-8")) + except Exception as e: + log.warning("sloppak: failed to parse keys %r: %s", keys_rel, e) + raw = None + if raw is not None and not isinstance(raw, dict): + log.warning("sloppak: keys %r ignored — expected dict, got %s", + keys_rel, type(raw).__name__) + elif isinstance(raw, dict): + if not isinstance(raw.get("events"), list): + log.warning("sloppak: keys %r ignored — 'events' must be a list", keys_rel) + else: + clean_events: list[dict] = [] + for ev in raw["events"]: + if not isinstance(ev, dict): + continue + # Drop events with a missing / non-numeric / non-finite + # time rather than silently rewriting them to 0.0 — a + # bad `t` makes the whole event meaningless. + t = ev.get("t") + if (not isinstance(t, (int, float)) or isinstance(t, bool) + or not math.isfinite(t)): + continue + t = float(t) + key = ev.get("key") + if not isinstance(key, str) or not key: + continue + entry = {"t": t, "key": key} + scale = ev.get("scale") + if isinstance(scale, str) and scale: + entry["scale"] = scale + clean_events.append(entry) + clean_events.sort(key=lambda e: e["t"]) + # int only — a float version (incl. NaN/Inf, which json.loads + # accepts) would raise on int(); default rather than abort the + # load of an optional side-file. + _ver = raw.get("version") + keys_data = { + "version": _ver if isinstance(_ver, int) + and not isinstance(_ver, bool) else 1, + "events": clean_events, + } + return LoadedSloppak( song=song, stems=stems, @@ -688,6 +755,7 @@ def load_song( manifest=manifest, drum_tab=drum_tab_data, song_timeline=song_timeline_data, + keys=keys_data, notation_by_id=notation_by_id_data, arrangement_ids=arrangement_ids_acc, ) diff --git a/server.py b/server.py index 49e2b08..ad69f99 100644 --- a/server.py +++ b/server.py @@ -6976,6 +6976,11 @@ async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1, and _notation_arr_id is not None and _notation_arr_id in loaded_slop.notation_by_id ), + # Song-level key/scale track presence (keys.json, spec §7.7) so a + # consumer can light up a key/scale display without parsing the pack. + "has_keys": bool( + is_slop and loaded_slop is not None and loaded_slop.keys is not None + ), }) # Send drum_tab when the sloppak ships one (manifest `drum_tab:` key, @@ -7016,6 +7021,17 @@ async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1, sections = [{"name": s.name, "time": s.start_time} for s in song.sections] await websocket.send_json({"type": "sections", "data": sections}) + # Send the song-level key/scale track (keys.json, spec §7.7) when the + # sloppak ships one. Consumers read it from the WS rather than the file, + # like drum_tab/beats/sections. The loader already sanitized the events + # (finite t, non-empty string key, sorted), so this is a direct send. + if is_slop and loaded_slop is not None and loaded_slop.keys is not None: + await websocket.send_json({ + "type": "keys", + "version": int(loaded_slop.keys.get("version", 1)), + "data": loaded_slop.keys.get("events") or [], + }) + # Send notation data when the sloppak ships it for the active arrangement. # Slots after sections (cursor sync depends on beats, which precede sections) # and before anchors — per docs/sloppak-spec.md §5.3. diff --git a/tests/test_sloppak_keys_load.py b/tests/test_sloppak_keys_load.py new file mode 100644 index 0000000..1852ac3 --- /dev/null +++ b/tests/test_sloppak_keys_load.py @@ -0,0 +1,127 @@ +"""End-to-end test for the sloppak loader recognising a `keys:` manifest key +(keys.json — the song-level, instrument-independent key/scale track, spec §7.7) +and surfacing the sanitized payload on the LoadedSloppak.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import yaml + +import sloppak as sloppak_mod + + +def _write_dir_sloppak(root: Path, manifest_extras: dict, keys_payload) -> Path: + """Minimal directory-form sloppak; writes keys.json when a payload is given. + + Unique filename per test (tmp_path leaf) so the module-level + resolve_source_dir cache isn't poisoned across tests.""" + pak = root / f"{root.name}.sloppak" + pak.mkdir() + arr_dir = pak / "arrangements" + arr_dir.mkdir() + arr = { + "name": "Lead", "tuning": [0, 0, 0, 0, 0, 0], "capo": 0, + "notes": [], "chords": [], "anchors": [], "handshapes": [], + "templates": [], "beats": [], "sections": [], + } + (arr_dir / "lead.json").write_text(json.dumps(arr)) + + manifest = { + "title": "Test", "artist": "Tester", "album": "", "year": 2026, + "duration": 10.0, + "arrangements": [{"id": "lead", "name": "Lead", "file": "arrangements/lead.json"}], + "stems": [{"id": "full", "file": "stems/full.ogg", "default": True}], + } + manifest.update(manifest_extras) + (pak / "manifest.yaml").write_text(yaml.safe_dump(manifest, sort_keys=False)) + + if keys_payload is not None: + (pak / "keys.json").write_text(json.dumps(keys_payload)) + return pak + + +def _load(pak_path: Path, tmp_path: Path): + dlc_root = pak_path.parent + cache = tmp_path / "cache" + cache.mkdir() + return sloppak_mod.load_song(pak_path.name, dlc_root, cache) + + +# ── Happy path ─────────────────────────────────────────────────────────────── + +def test_load_song_attaches_keys_when_manifest_opts_in(tmp_path: Path): + payload = { + "version": 1, + "events": [ + {"t": 0.0, "key": "Em", "scale": "natural_minor"}, + {"t": 2.0, "key": "G", "scale": "major"}, + ], + } + pak = _write_dir_sloppak(tmp_path, {"keys": "keys.json"}, payload) + loaded = _load(pak, tmp_path) + assert loaded.keys is not None + assert loaded.keys["version"] == 1 + evs = loaded.keys["events"] + assert len(evs) == 2 + assert evs[0] == {"t": 0.0, "key": "Em", "scale": "natural_minor"} + assert evs[1] == {"t": 2.0, "key": "G", "scale": "major"} + + +# ── Absent / permissive ────────────────────────────────────────────────────── + +def test_load_song_keys_absent_when_manifest_silent(tmp_path: Path): + pak = _write_dir_sloppak(tmp_path, {}, None) + assert _load(pak, tmp_path).keys is None + + +def test_load_song_keys_absent_when_file_missing(tmp_path: Path): + pak = _write_dir_sloppak(tmp_path, {"keys": "nope.json"}, None) + assert _load(pak, tmp_path).keys is None + + +def test_load_song_keys_absent_when_invalid_json(tmp_path: Path): + pak = _write_dir_sloppak(tmp_path, {"keys": "keys.json"}, None) + (pak / "keys.json").write_text("not json {{{") + assert _load(pak, tmp_path).keys is None + + +def test_load_song_keys_ignored_when_events_not_a_list(tmp_path: Path): + pak = _write_dir_sloppak(tmp_path, {"keys": "keys.json"}, + {"version": 1, "events": "nope"}) + assert _load(pak, tmp_path).keys is None + + +# ── Sanitization ───────────────────────────────────────────────────────────── + +def test_load_song_keys_sanitizes_and_sorts(tmp_path: Path): + payload = { + "version": 1, + "events": [ + {"t": 2.0, "key": "G"}, # no scale -> omitted + {"t": 0.0, "key": "Em", "scale": "major"}, # out of order + {"t": 1.0}, # no key -> dropped + {"foo": "bar"}, # not an event -> dropped + {"t": 3.0, "key": ""}, # empty key -> dropped + {"t": "bad", "key": "X"}, # non-numeric t -> dropped + "garbage", # non-dict -> dropped + ], + } + pak = _write_dir_sloppak(tmp_path, {"keys": "keys.json"}, payload) + evs = _load(pak, tmp_path).keys["events"] + assert evs == [ + {"t": 0.0, "key": "Em", "scale": "major"}, + {"t": 2.0, "key": "G"}, # scale absent, not null + ] + + +def test_load_song_keys_nonint_version_does_not_abort_load(tmp_path: Path): + # json.loads accepts NaN; a float/NaN version must not raise int(NaN) and + # abort the load of an OPTIONAL side-file — it falls back to version 1. + payload = {"version": float("nan"), "events": [{"t": 0.0, "key": "C"}]} + pak = _write_dir_sloppak(tmp_path, {"keys": "keys.json"}, payload) + loaded = _load(pak, tmp_path) + assert loaded.keys is not None + assert loaded.keys["version"] == 1 + assert loaded.keys["events"] == [{"t": 0.0, "key": "C"}]