From 587fbbea813d8c7bfb7329190abef0b34f5550ee Mon Sep 17 00:00:00 2001 From: Byron Gamatos Date: Sat, 20 Jun 2026 21:35:07 +0200 Subject: [PATCH] feat(core): consume song_timeline tempos + time_signatures + per-chart tempos (feedpak 1.2.0) (#529) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit feedpak 1.2.0 added song-level `tempos` + `time_signatures` to song_timeline.json and a per-chart `tempos` override on arrangements (§6.10). Core stored the raw song_timeline dict but never consumed the maps, and didn't read per-chart tempos. - song.py: shared `sanitize_tempos([{time,bpm}])` (finite non-bool time, finite bpm>0, sorted); `Arrangement.tempos` field wired through arrangement_to_wire (omitted when None/empty per §6.10) / arrangement_from_wire. - sloppak.py: `_sanitize_time_signatures([{time,ts:[num,den]}])`; LoadedSloppak.tempos / .time_signatures, loaded from song_timeline.json INDEPENDENTLY of beats/sections (all are optional in 1.2.0). - server.py: stream `tempos` + `time_signatures` highway-WS messages; the active arrangement's per-chart `tempos` overrides the song-level map for that chart. Renderer/UI surfacing is a thin follow-up; this lands the data plumbing. Codex-reviewed: clean (no findings). +9 tests (sanitizers, per-chart wire round-trip + omit-when-absent, song-level load/sanitize/absent + maps-without- beats). 90 song/sloppak tests pass. (Pre-existing unrelated failure: test_diagnostics_redact, fails on clean main too.) Closes #526. Part of got-feedback/feedback#334. Co-authored-by: Claude Opus 4.8 (1M context) --- lib/sloppak.py | 43 +++++++++++++++++++++++ lib/song.py | 32 +++++++++++++++++ server.py | 14 ++++++++ tests/test_sloppak_song_timeline_load.py | 44 ++++++++++++++++++++++++ tests/test_song.py | 33 ++++++++++++++++++ 5 files changed, 166 insertions(+) diff --git a/lib/sloppak.py b/lib/sloppak.py index bc72dd0..90c41c4 100644 --- a/lib/sloppak.py +++ b/lib/sloppak.py @@ -34,6 +34,7 @@ from song import ( Arrangement, arrangement_from_wire, _finite_float, + sanitize_tempos, ) import drums as drums_mod import notation as notation_mod @@ -284,6 +285,30 @@ def read_cover_bytes( return None +def _sanitize_time_signatures(events) -> list[dict]: + """Clean a time-signature event list (``[{time, ts:[num, den]}]``): keep + entries with a finite non-bool ``time`` and a ``ts`` of two integers >= 1, + sorted by time. Non-list / all-invalid input -> ``[]``.""" + out: list[dict] = [] + if isinstance(events, list): + for ev in events: + if not isinstance(ev, dict): + continue + t = ev.get("time") + ts = ev.get("ts") + if (not isinstance(t, (int, float)) or isinstance(t, bool) + or not math.isfinite(t)): + continue + if not isinstance(ts, list) or len(ts) != 2: + continue + if not all(isinstance(x, int) and not isinstance(x, bool) and x >= 1 + for x in ts): + continue + out.append({"time": float(t), "ts": [int(ts[0]), int(ts[1])]}) + out.sort(key=lambda e: e["time"]) + return out + + @dataclass class LoadedSloppak: """Result of loading a sloppak: the Song object plus stem descriptors.""" @@ -307,6 +332,13 @@ class LoadedSloppak: # absent / unreadable / malformed. Streamed over the highway WS as a # `keys` message; consumers (renderers, plugins) read it from there. keys: dict | None = None + # Sanitized song-level tempo + time-signature maps from `song_timeline.json` + # (feedpak 1.2.0). `tempos`: [{time, bpm}]; `time_signatures`: [{time, ts}]. + # None when absent/empty. Streamed over the highway WS (`tempos` / + # `time_signatures` messages); a per-chart arrangement `tempos` overrides + # `tempos` for that chart (spec §6.10). + tempos: list | None = None + time_signatures: list | 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 @@ -515,6 +547,8 @@ def load_song( # already loaded onto the song object — song_timeline is the authoritative # source for timeline data in sloppaks that carry it. song_timeline_data: dict | None = None + tempos_data: list | None = None + time_sigs_data: list | None = None song_timeline_rel = manifest.get("song_timeline") if isinstance(song_timeline_rel, str) and song_timeline_rel: try: @@ -597,6 +631,13 @@ def load_song( ) continue song_timeline_data = raw + # tempos / time_signatures (feedpak 1.2.0) are independent of the + # beats/sections validation above — all are optional — so load them + # whenever the payload parsed to a dict. + if isinstance(raw, dict): + tempos_data = sanitize_tempos(raw.get("tempos")) or None + time_sigs_data = _sanitize_time_signatures( + raw.get("time_signatures")) or None # Optional shared lyrics file. Same safety posture as the drum_tab # loader above: constrain the manifest-declared path to source_dir @@ -755,6 +796,8 @@ def load_song( manifest=manifest, drum_tab=drum_tab_data, song_timeline=song_timeline_data, + tempos=tempos_data, + time_signatures=time_sigs_data, keys=keys_data, notation_by_id=notation_by_id_data, arrangement_ids=arrangement_ids_acc, diff --git a/lib/song.py b/lib/song.py index 10e05fe..b58f5b6 100644 --- a/lib/song.py +++ b/lib/song.py @@ -151,6 +151,10 @@ class Arrangement: # RS2014 custom song pitch-shift field (cents). Commonly -1200.0 (one octave # down) for extended-range bass arrangements. 0.0 when absent or zero. cent_offset: float = 0.0 + # Per-chart tempo override (§6.10): [{time, bpm}]. None when the chart + # follows the song-level tempo; when present a Reader uses it for this + # chart and ignores the song-level tempo. + tempos: list | None = None @dataclass @@ -585,6 +589,29 @@ def _finite_float(value, default: float = 0.0) -> float: return v if math.isfinite(v) else default +def sanitize_tempos(events) -> list[dict]: + """Clean a tempo-event list (``[{time, bpm}]``): keep entries with a finite + non-bool ``time`` and a finite ``bpm > 0``, coerced to float and sorted by + time. Non-list / all-invalid input -> ``[]``. Shared by the per-chart + arrangement ``tempos`` (§6.10) and the song-level ``song_timeline.tempos``.""" + out: list[dict] = [] + if isinstance(events, list): + for ev in events: + if not isinstance(ev, dict): + continue + t = ev.get("time") + bpm = ev.get("bpm") + if (not isinstance(t, (int, float)) or isinstance(t, bool) + or not math.isfinite(t)): + continue + if (not isinstance(bpm, (int, float)) or isinstance(bpm, bool) + or not math.isfinite(bpm) or bpm <= 0): + continue + out.append({"time": float(t), "bpm": float(bpm)}) + out.sort(key=lambda e: e["time"]) + return out + + def arrangement_to_wire(arr: Arrangement) -> dict: """Serialize an Arrangement into a JSON-ready dict matching the wire format.""" out = { @@ -612,6 +639,10 @@ def arrangement_to_wire(arr: Arrangement) -> dict: # "no tones". if arr.tones: out["tones"] = arr.tones + # Per-chart tempo override (§6.10) — additive; omit when the chart follows + # the song-level tempo (empty/None). + if arr.tempos: + out["tempos"] = list(arr.tempos) return out @@ -622,6 +653,7 @@ def arrangement_from_wire(d: dict) -> Arrangement: tuning=list(d.get("tuning", [0] * 6)), capo=int(d.get("capo", 0)), cent_offset=_finite_float(d.get("centOffset", 0.0)), + tempos=(sanitize_tempos(d.get("tempos")) or None), notes=[note_from_wire(n) for n in d.get("notes", [])], chords=[chord_from_wire(c) for c in d.get("chords", [])], anchors=[ diff --git a/server.py b/server.py index ad69f99..6d36a95 100644 --- a/server.py +++ b/server.py @@ -7032,6 +7032,20 @@ async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1, "data": loaded_slop.keys.get("events") or [], }) + # Song-level tempo + time-signature maps (song_timeline, feedpak 1.2.0), + # plus the per-chart tempo override (§6.10): the active arrangement's own + # `tempos` wins over the song-level map for this chart. Both are + # pre-sanitized by the loader / arrangement_from_wire, so they stream + # directly. Consumers read these rather than the file. + _song_tempos = loaded_slop.tempos if (is_slop and loaded_slop is not None) else None + _tempos_out = getattr(arr, "tempos", None) or _song_tempos + if _tempos_out: + await websocket.send_json({"type": "tempos", "data": _tempos_out}) + _time_sigs = (loaded_slop.time_signatures + if (is_slop and loaded_slop is not None) else None) + if _time_sigs: + await websocket.send_json({"type": "time_signatures", "data": _time_sigs}) + # 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_song_timeline_load.py b/tests/test_sloppak_song_timeline_load.py index 5c66f44..6986261 100644 --- a/tests/test_sloppak_song_timeline_load.py +++ b/tests/test_sloppak_song_timeline_load.py @@ -148,6 +148,50 @@ def test_song_timeline_absent_when_path_escapes_sloppak(tmp_path: Path): assert loaded.song_timeline is None +# ── tempos + time_signatures (feedpak 1.2.0) ───────────────────────────────── + +def test_song_timeline_tempos_and_time_signatures_loaded(tmp_path: Path): + payload = { + "version": 1, "beats": [], "sections": [], + "tempos": [{"time": 0.0, "bpm": 120}, {"time": 4.0, "bpm": 90}], + "time_signatures": [{"time": 0.0, "ts": [4, 4]}, {"time": 8.0, "ts": [6, 8]}], + } + pak = _write_dir_sloppak(tmp_path, {"song_timeline": "song_timeline.json"}, payload) + loaded = _load(pak, tmp_path) + assert loaded.tempos == [{"time": 0.0, "bpm": 120.0}, {"time": 4.0, "bpm": 90.0}] + assert loaded.time_signatures == [{"time": 0.0, "ts": [4, 4]}, + {"time": 8.0, "ts": [6, 8]}] + + +def test_song_timeline_maps_absent_when_not_provided(tmp_path: Path): + payload = {"version": 1, "beats": [], "sections": []} + pak = _write_dir_sloppak(tmp_path, {"song_timeline": "song_timeline.json"}, payload) + loaded = _load(pak, tmp_path) + assert loaded.tempos is None and loaded.time_signatures is None + + +def test_song_timeline_maps_sanitized(tmp_path: Path): + payload = { + "version": 1, "beats": [], "sections": [], + "tempos": [{"time": 1.0, "bpm": 0}, {"time": 0.0, "bpm": 100}], # bpm 0 dropped + sorted + "time_signatures": [{"time": 0.0, "ts": [4, 4, 4]}, # 3-long dropped + {"time": 2.0, "ts": [3, 4]}], + } + pak = _write_dir_sloppak(tmp_path, {"song_timeline": "song_timeline.json"}, payload) + loaded = _load(pak, tmp_path) + assert loaded.tempos == [{"time": 0.0, "bpm": 100.0}] + assert loaded.time_signatures == [{"time": 2.0, "ts": [3, 4]}] + + +def test_song_timeline_maps_load_even_without_beats_or_sections(tmp_path: Path): + # tempos/time_signatures are independent of beats/sections — a payload that + # omits beats (invalid for the override path) must still surface the maps. + payload = {"version": 1, "tempos": [{"time": 0.0, "bpm": 100}]} + pak = _write_dir_sloppak(tmp_path, {"song_timeline": "song_timeline.json"}, payload) + loaded = _load(pak, tmp_path) + assert loaded.tempos == [{"time": 0.0, "bpm": 100.0}] + + def test_song_timeline_absent_when_path_is_absolute(tmp_path: Path): pak = _write_dir_sloppak(tmp_path, {"song_timeline": "/etc/passwd"}, None) loaded = _load(pak, tmp_path) diff --git a/tests/test_song.py b/tests/test_song.py index fcab092..73217fc 100644 --- a/tests/test_song.py +++ b/tests/test_song.py @@ -18,6 +18,7 @@ from song import ( arrangement_to_wire, chord_from_wire, chord_to_wire, + sanitize_tempos, compute_smart_names, note_from_wire, note_to_wire, @@ -927,3 +928,35 @@ def test_smart_names_arrangement_properties_defaults(): assert arr.path_bass is False assert arr.bonus_arr is False assert arr.represent == 0 + + +# ── tempos (per-chart §6.10 + shared sanitizer) ────────────────────────────── + +def test_sanitize_tempos_filters_sorts_and_coerces(): + assert sanitize_tempos([ + {"time": 2.0, "bpm": 90}, + {"time": 0.0, "bpm": 120}, + {"time": 1.0, "bpm": 0}, # bpm <= 0 -> dropped + {"time": float("nan"), "bpm": 100}, # non-finite time -> dropped + {"bpm": 100}, # missing time -> dropped + {"time": 3.0, "bpm": float("inf")}, # non-finite bpm -> dropped + "x", # non-dict -> dropped + ]) == [{"time": 0.0, "bpm": 120.0}, {"time": 2.0, "bpm": 90.0}] + assert sanitize_tempos(None) == [] + assert sanitize_tempos("nope") == [] + + +def test_arrangement_tempos_round_trip_and_omitted_when_absent(): + arr = arrangement_from_wire({ + "name": "Bass", "tuning": [0, 0, 0, 0, 0, 0], "capo": 0, + "tempos": [{"time": 0.0, "bpm": 60}, {"time": 2.0, "bpm": 120}], + }) + assert arr.tempos == [{"time": 0.0, "bpm": 60.0}, {"time": 2.0, "bpm": 120.0}] + assert arrangement_to_wire(arr)["tempos"] == \ + [{"time": 0.0, "bpm": 60.0}, {"time": 2.0, "bpm": 120.0}] + + # Absent per-chart tempos -> None, and the wire key is OMITTED (not []), + # so the chart follows the song-level tempo (spec §6.10). + arr2 = arrangement_from_wire({"name": "Lead", "tuning": [0] * 6, "capo": 0}) + assert arr2.tempos is None + assert "tempos" not in arrangement_to_wire(arr2)