feat(core): consume keys.json — song-level key/scale track (loader + WS)

lib/sloppak.py

@@ -15,6 +15,7 @@
 
 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,13 +687,75 @@ 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,
         source_dir=source_dir,
         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,
     )

server.py

@@ -6976,6 +6976,11 @@ def _evict_audio_cache():
                 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 @@ def _evict_audio_cache():
         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.

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"}]