feat(core): consume song_timeline tempos + time_signatures + per-chart tempos (feedpak 1.2.0) (#529)

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) <noreply@anthropic.com>

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,

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=[

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.

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)

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)