feat(drums): import Rock Band / Clone Hero / YARG drum charts

CHANGELOG.md

@@ -8,6 +8,29 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- **Import Rock Band / Clone Hero / YARG drum charts (feedpak 1.20.0).**
+  `scripts/ch_to_sloppak.py` turns a CH/YARG song folder (`notes.mid` PART
+  DRUMS or `notes.chart`, `song.ini`, audio) — or an `.sng` package — into a
+  drums pack: a `type: drums` arrangement whose `drum_tab.json` carries the
+  Expert part as 4-lane pro piece hits plus the chart's gameplay data in the
+  1.20.0 "drum gameplay metadata" shape — accents, double kicks (`x2`, so the
+  Expert / Expert+ split survives), star power, fills, solos, and the chart's
+  own Easy/Medium/Hard parts as `levels` — and a `song_timeline.json` with its
+  tempo map, beats and sections. song.ini `delay` is applied, so the hits line
+  up with the audio as YARG plays it. A chart with star power but no fills gets
+  YARG's generated activation fills, and a missing lower level is reduced from
+  Expert and marked `generated` so a player can tell it wasn't charted by hand.
+  `scripts/sloppak_to_ch.py` goes the other way for a pack's drum part (or
+  `--part <id>`), writing `notes.mid` + `song.ini` + OGG stems that YARG and
+  Clone Hero read, and the two round-trip. The parsing rules — tom markers,
+  5-lane → pro conversion, disco flip, dynamics, delay — are ported from
+  YARG.Core (LGPL-3.0, attributed in `lib/drumchart.py`); piece-ids come from
+  `lib/drums.py`, so the importer can't drift from the vocabulary the drums
+  plugin renders. This is the writer side only: the loader already plays the
+  tab's hits, and streaming the gameplay keys over the highway WS to a drums
+  renderer is a separate follow-up.
+  Implementation in `lib/drumchart.py`, `lib/drumchart_gen.py`, `lib/sngfile.py`;
+  tests in `tests/test_drumchart.py` and `tests/test_sloppak_to_ch.py`.
 - **Core reader for source rigs (feedpak 1.18.0).** A pack can declare what a
   MIDI part should sound like by binding a rig; core now reads that binding and
   hands it to the client instead of dropping it. Three parts: the

lib/drumchart_gen.py

@@ -0,0 +1,160 @@
+"""Generated drum gameplay data for charts that lack it.
+
+Rock Band / Clone Hero charts usually carry their own Easy/Medium/Hard parts and
+fills; not every source does. `ensure_levels` fills missing lower difficulties by
+reducing Expert (feedBack's own rules, `reduce_level`), and `auto_fills` places
+star-power activation fills the way YARG does for charts without them. Results are
+drumchart.DrumHit lists / (start, end) spans, encoded by `drumchart.drum_tab`.
+
+`auto_fills` is ported from YARG.Core SongChart.AutoGeneration
+(ParseForActivationPhrases) — Copyright (c) YARC (YARG) contributors, LGPL-3.0
+(https://github.com/YARC-Official/YARG.Core/blob/master/LICENSE) — and is
+distributed as part of feedBack under AGPL-3.0, as described in lib/drumchart.py.
+"""
+
+from __future__ import annotations
+
+import bisect
+
+from drumchart import LOWER_LEVELS, DrumChart, DrumHit
+
+
+# ── difficulty reduction (for sources without hand-charted lower levels) ───
+
+# Fitted on 136 hand-authored Rock Band / Clone Hero charts (their Easy/Medium/Hard vs
+# their Expert): F1 of matching notes vs the human levels Easy 0.67, Medium 0.78, Hard 0.89,
+# note counts 1.12x / 0.98x / 1.05x the human charts. Easy never pairs a kick with hands.
+REDUCE = {   # level: (grid in beats, max gems per chord, kick together with hands?)
+    "hard": (0.5, 3, True),
+    "medium": (1.0, 2, True),
+    "easy": (1.0, 2, False),
+}
+_HAND_PRIORITY = {("red", False): 0, ("yellow", True): 1, ("green", True): 2, ("blue", True): 3,
+                  ("yellow", False): 4, ("blue", False): 5, ("green", False): 6}
+
+
+def _beat_pos(bt, t):
+    """Fractional beat index of time t on beat grid bt."""
+    if len(bt) < 2:
+        return t / 0.5
+    i = min(max(bisect.bisect_right(bt, t) - 1, 0), len(bt) - 2)
+    return i + (t - bt[i]) / (bt[i + 1] - bt[i])
+
+
+def reduce_level(hits, beats, level: str) -> list[DrumHit]:
+    """feedBack's own reduction of an Expert chart for sources without hand-charted
+    levels (not from YARG/Harmonix; rules fitted to authored charts, see REDUCE). Drops
+    2x kicks and ghost notes; keeps chords on the level's grid (8ths / quarters) first,
+    then off-grid ones that still leave a full grid step to both neighbours; caps gems
+    per chord (snare and cymbals before toms; a kick is kept with one hand gem except on
+    Easy). ``beats``: [(time, measure)]."""
+    grid, max_gems, kick_with_hands = REDUCE[level]
+    # strictly increasing grid: a beat map can repeat a time (zero-length beat)
+    bt = []
+    for t in sorted(t for t, _ in beats or []):
+        if not bt or t - bt[-1] > 1e-6:
+            bt.append(t)
+    chords: dict[float, list] = {}
+    for h in hits:
+        if h.kick2x or h.dyn == "ghost":
+            continue
+        chords.setdefault(round(h.time, 3), []).append(h)
+    ts = sorted(chords)
+    pos = {t: _beat_pos(bt, t) for t in ts}
+    on = {t for t in ts if abs(pos[t] / grid - round(pos[t] / grid)) < 0.12}
+    kept = sorted(on)
+    for t in ts:                       # off-grid chords where they don't crowd the grid
+        if t in on:
+            continue
+        i = bisect.bisect_left(kept, t)
+        prev_ok = i == 0 or pos[t] - pos[kept[i - 1]] >= grid - 0.05
+        next_ok = i == len(kept) or pos[kept[i]] - pos[t] >= grid - 0.05
+        if prev_ok and next_ok:
+            kept.insert(i, t)
+    out = []
+    for t in kept:
+        hs = sorted({(h.pad, h.cymbal): h for h in chords[t] if h.pad != "kick"}.values(),
+                    key=lambda h: _HAND_PRIORITY.get((h.pad, h.cymbal), 9))
+        kick = [h for h in chords[t] if h.pad == "kick"][:1]
+        if kick and hs and kick_with_hands:
+            out.extend(kick + hs[:max_gems - 1])
+        elif hs:
+            out.extend(hs[:max_gems])
+        else:
+            out.extend(kick)
+    return out
+
+
+def ensure_levels(chart: DrumChart, beats=None) -> DrumChart:
+    """Fill in every missing lower level by reducing Expert (marked as generated)."""
+    if beats is None:
+        beats = [(b["time"], b["measure"]) for b in chart.beat_list()] if chart.hits else []
+    for lv in LOWER_LEVELS:
+        if not chart.levels.get(lv):
+            chart.levels[lv] = reduce_level(chart.hits, beats, lv)
+            if lv not in chart.levels_generated:
+                chart.levels_generated.append(lv)
+    return chart
+
+
+# ── fill (star power activation) generation ────────────────────────────────
+
+def _lower_bound(xs, v) -> int:
+    """Index of the last element <= v, -1 if none (YARG ChartEventExtensions.LowerBound)."""
+    return bisect.bisect_right(xs, v + 1e-9) - 1
+
+
+def auto_fills(note_times, beats, star_power, solos=(), sections=()) -> list[tuple[float, float]]:
+    """Fill (star power activation) phrases, ported from YARG.Core
+    SongChart.AutoGeneration.ParseForActivationPhrases: every 4 measures (2 if that's more
+    than 10 s away), snapped to section starts, at least 2 s after the last SP/solo/fill,
+    only where 16+ notes follow within 4 measures; the fill spans the measure before the
+    activation bar line. (Time-signature snapping is skipped: the beat list carries no
+    time signatures.) Returns nothing when there is no star power, like YARG."""
+    MIN_SPACING, MAX_SPACING, SP_MIN_NOTES = 2.0, 10.0, 16
+    measures = [t for t, m in beats if m != -1]
+    sp = sorted(star_power)
+    if not sp or len(measures) < 6:
+        return []
+    solos = sorted(solos)
+    sections = sorted(sections)
+    sp_starts = [a for a, _ in sp]
+    solo_starts = [a for a, _ in solos]
+    last_solo = solos[-1][1] if solos else 0.0
+    nt = sorted(note_times)
+    spacing_ref = sp[0][1]
+    cur = max(0, _lower_bound(measures, spacing_ref))
+    sec_i = _lower_bound(sections, spacing_ref)
+    sp_i, solo_i = 0, _lower_bound(solo_starts, spacing_ref)
+    out = []
+    total = len(measures)
+    while cur < total - 4:
+        per = 4
+        if measures[cur + per] - spacing_ref > MAX_SPACING:
+            per = 2
+        cur += per
+        mt = measures[cur]
+        new_sec = _lower_bound(sections, mt)
+        if new_sec > sec_i:
+            sec_i = new_sec
+            cur = max(0, _lower_bound(measures, sections[sec_i]))
+            mt = measures[cur]
+        new_sp = _lower_bound(sp_starts, mt)
+        if new_sp > sp_i:
+            sp_i = new_sp
+            spacing_ref = max(sp[sp_i][1], spacing_ref)
+        if solos and mt < last_solo:
+            new_solo = _lower_bound(solo_starts, mt)
+            if new_solo > solo_i:
+                solo_i = new_solo
+                spacing_ref = max(solos[solo_i][1], spacing_ref)
+        if mt - spacing_ref < MIN_SPACING:
+            continue
+        end = measures[min(cur + 4, total - 1)]
+        n = bisect.bisect_right(nt, end + 1e-6) - bisect.bisect_right(nt, mt + 1e-6)
+        if n < SP_MIN_NOTES:
+            continue
+        spacing_ref = mt
+        if cur >= 1:
+            out.append((round(measures[cur - 1], 4), round(mt, 4)))
+    return out

lib/sngfile.py

@@ -0,0 +1,115 @@
+"""Read Clone Hero / YARG ``.sng`` packages (a song folder in one masked file).
+
+Layout (SngFileFormat, github.com/mdsitton/SngFileFormat, version 1, little endian):
+  "SNGPKG" | uint32 version | 16-byte xor mask
+  uint64 metadata_len | uint64 count | count x (int32 klen, key, int32 vlen, value)   - song.ini pairs
+  uint64 filemeta_len | uint64 count | count x (uint8 nlen, name, uint64 size, uint64 offset)
+  uint64 data_len | file data, each byte masked: b ^ mask[i % 16] ^ (i & 0xFF), i = index in its file
+"""
+
+from __future__ import annotations
+
+import struct
+from pathlib import Path
+
+# Never unpack these: CH/YARG background videos are large and nothing here plays them.
+_VIDEO_EXTS = (".mp4", ".webm", ".avi", ".mkv", ".vp8")
+
+
+class SngError(ValueError):
+    pass
+
+
+class NeedMore(SngError):
+    """The buffer ends inside the header; fetch at least ``need`` bytes."""
+
+    def __init__(self, need):
+        super().__init__(f"need {need} bytes")
+        self.need = need
+
+
+def parse_header(buf: bytes) -> tuple[dict, dict, int]:
+    """(metadata incl. "mask", {filename: (offset, size)}, header_end) from the start of an .sng."""
+    pos = 0
+
+    def take(n):
+        nonlocal pos
+        if pos + n > len(buf):
+            raise NeedMore(max(pos + n, 2 * len(buf)))
+        b = buf[pos:pos + n]
+        pos += n
+        return b
+
+    if take(6) != b"SNGPKG":
+        raise SngError("not an SNG package")
+    (version,) = struct.unpack("<I", take(4))
+    if version != 1:
+        raise SngError(f"unsupported SNG version {version}")
+    mask = take(16)
+    _, n = struct.unpack("<QQ", take(16))
+    meta = {"mask": mask}
+    for _ in range(n):
+        (kl,) = struct.unpack("<i", take(4))
+        k = take(kl).decode("utf-8", "replace")
+        (vl,) = struct.unpack("<i", take(4))
+        meta[k] = take(vl).decode("utf-8", "replace")
+    _, n = struct.unpack("<QQ", take(16))
+    files = {}
+    for _ in range(n):
+        (nl,) = struct.unpack("<B", take(1))
+        name = take(nl).decode("utf-8", "replace")
+        size, off = struct.unpack("<QQ", take(16))
+        files[name] = (off, size)
+    take(8)          # data length
+    return meta, files, pos
+
+
+def read_sng(path) -> tuple[dict, dict]:
+    """(metadata incl. "mask", {filename: (offset, size)}) without reading file contents."""
+    with open(path, "rb") as f:
+        buf = f.read(65536)
+        while True:
+            try:
+                meta, files, _ = parse_header(buf)
+                return meta, files
+            except NeedMore as e:
+                f.seek(0)
+                more = f.read(e.need)
+                if len(more) <= len(buf):
+                    raise SngError(f"{Path(path).name}: truncated header") from None
+                buf = more
+
+
+def unmask(data: bytes, mask: bytes) -> bytes:
+    """Undo the per-file xor mask. The key stream repeats every 256 bytes
+    (lcm of the 16-byte mask and the ``i & 0xFF`` counter), so xor the whole
+    buffer against that period as one big integer instead of byte by byte."""
+    if not data:
+        return b""
+    period = bytes(mask[i % 16] ^ i for i in range(256))
+    key = (period * (len(data) // 256 + 1))[:len(data)]
+    n = int.from_bytes(data, "little") ^ int.from_bytes(key, "little")
+    return n.to_bytes(len(data), "little")
+
+
+def extract_sng(path, out_dir, names=None) -> Path:
+    """Unpack an .sng into a normal song folder (song.ini written from its metadata).
+    ``names``: only these files; default everything but video."""
+    meta, files = read_sng(path)
+    mask = meta.pop("mask")
+    out = Path(out_dir)
+    out.mkdir(parents=True, exist_ok=True)
+    with open(path, "rb") as f:
+        for name, (off, size) in files.items():
+            if names is not None and name not in names:
+                continue
+            if names is None and Path(name).suffix.lower() in _VIDEO_EXTS:
+                continue
+            safe = Path(name.replace("\\", "/")).name      # no directories from the package
+            if not safe or safe in (".", ".."):
+                continue
+            f.seek(off)
+            (out / safe).write_bytes(unmask(f.read(size), mask))
+    ini = "".join(f"{k} = {v}\n" for k, v in meta.items() if "\n" not in k + v)
+    (out / "song.ini").write_text("[song]\n" + ini, encoding="utf-8")
+    return out