Merge remote-tracking branch 'origin/feat/drum-parts-loader' into feat/drum-part-picker

CHANGELOG.md

@@ -25,12 +25,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   (primary first; the entry aliasing the song-level `drum_tab:` key is the
   primary and is never loaded twice), the highway WS `song_info` gains a
   `drum_parts` name list, and `?drum_part=<id>` on the WS URL selects which
-  part's tab streams (`drum_tab` messages carry `part_id` when a parts list
-  exists; unknown ids fall back to the primary). Pointer entries are **never**
+  part's tab streams (`drum_tab` messages carry `part_id` when multiple parts
+  exist; unknown ids fall back to the primary). Pointer entries are **never**
   loaded as fretted arrangements — the loader's file/notation gate keeps a drum
   part out of the fretted pipeline (and out of note-detection grading), pinned
   by test. Legacy single-drum packs read exactly as before, as a one-part list.
-  coordinator. Synchronous transforms run after difficulty filtering; host
+- **Chart-transform coordinator.** Synchronous transforms run after difficulty filtering; host
   data is isolated from providers, accepted timelines are time-sorted, and
   failures fall back to the original chart with a fixed public reason.
   Effective chart arrays and metadata are available to 2D/custom renderers

lib/routers/ws_highway.py

@@ -143,6 +143,11 @@ def _sanitize_authors(manifest: dict | None) -> list[dict]:
     return out
 
 
+def _drum_part_id_for_wire(drum_parts: list[dict] | None, selected_id: str | None) -> str | None:
+    """Expose a part id only when the pack genuinely has multiple parts."""
+    return selected_id if selected_id is not None and len(drum_parts or []) > 1 else None
+
+
 @router.websocket("/ws/highway/{filename:path}")
 async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1,
                      naming_mode: str = "legacy", drum_part: str = ""):
@@ -626,10 +631,11 @@ async def highway_ws(websocket: WebSocket, filename: str, arrangement: int = -1,
                 "kit": kit,
                 "total": len(hits_wire),
             }
-            # Which part this stream carries — present only when the pack has
-            # a parts list, so the legacy frame stays byte-identical.
-            if _dt_part_id is not None:
-                _dt_msg["part_id"] = _dt_part_id
+            # Only multi-part packs identify a part on the wire. Legacy packs
+            # synthesize a one-item list internally but keep their old frame.
+            _wire_part_id = _drum_part_id_for_wire(loaded_slop.drum_parts, _dt_part_id)
+            if _wire_part_id is not None:
+                _dt_msg["part_id"] = _wire_part_id
             try:
                 await websocket.send_json(_dt_msg)
                 for i in range(0, len(hits_wire), 500):

lib/sloppak.py

@@ -776,6 +776,72 @@ def _load_drum_tab_file(source_dir: Path, rel: str, label: str) -> dict | None:
     return raw
 
 
+def _resolve_drum_parts(
+    source_dir: Path,
+    drum_tab_rel: object,
+    drum_tab_data: dict | None,
+    drum_pointer_entries: list[dict],
+) -> tuple[dict | None, list[dict] | None]:
+    """Resolve drum pointers into a primary-first list with unique ids."""
+    if drum_tab_data is None and not drum_pointer_entries:
+        return drum_tab_data, None
+
+    primary_id = "drums"
+    primary_name = None
+    extra_parts: list[dict] = []
+    seen_rels: set[str] = set()
+    for entry in drum_pointer_entries:
+        rel = str(entry.get("drum_tab") or "").strip()
+        if not rel or rel in seen_rels:
+            continue
+        seen_rels.add(rel)
+        entry_id = str(entry.get("id") or "").strip()
+        entry_name = str(entry.get("name") or "").strip()
+        if isinstance(drum_tab_rel, str) and rel == drum_tab_rel.strip():
+            if entry_id:
+                primary_id = entry_id
+            if entry_name:
+                primary_name = entry_name
+            continue
+        tab = _load_drum_tab_file(source_dir, rel, f"drum part {entry_id or rel}")
+        if tab is None:
+            continue
+        tab_name = tab.get("name")
+        extra_parts.append({
+            "id": entry_id,
+            "name": entry_name
+                or (tab_name if isinstance(tab_name, str) and tab_name else "Drums"),
+            "drum_tab": tab,
+        })
+
+    parts: list[dict] = []
+    used_ids: set[str] = set()
+    if drum_tab_data is not None:
+        if primary_name is None:
+            tab_name = drum_tab_data.get("name")
+            primary_name = tab_name if isinstance(tab_name, str) and tab_name else "Drums"
+        parts.append({"id": primary_id, "name": primary_name, "drum_tab": drum_tab_data})
+        used_ids.add(primary_id)
+
+    next_generated_id = 2
+    for part in extra_parts:
+        part_id = part["id"]
+        if not part_id or part_id in used_ids:
+            while f"drums-{next_generated_id}" in used_ids:
+                next_generated_id += 1
+            part_id = f"drums-{next_generated_id}"
+            next_generated_id += 1
+        part["id"] = part_id
+        used_ids.add(part_id)
+        parts.append(part)
+
+    if not parts:
+        return drum_tab_data, None
+    if drum_tab_data is None:
+        drum_tab_data = parts[0]["drum_tab"]
+    return drum_tab_data, parts
+
+
 def load_song(
     filename: str,
     dlc_root: Path,
@@ -817,6 +883,11 @@ def load_song(
             _etype = str(entry.get("type") or "").strip().lower()
             if _etype in ("drums", "drum") and isinstance(entry.get("drum_tab"), str):
                 drum_pointer_entries.append(entry)
+            elif isinstance(entry.get("drum_tab"), str):
+                log.warning(
+                    "sloppak: arrangement entry has drum_tab %r but type=%r — ignored",
+                    entry.get("drum_tab"), entry.get("type"),
+                )
             continue
         data = None
         if rel:
@@ -924,59 +995,11 @@ def load_song(
     if isinstance(drum_tab_rel, str) and drum_tab_rel:
         drum_tab_data = _load_drum_tab_file(source_dir, drum_tab_rel, "drum_tab")
 
-    # DRUM PARTS (feedpak 1.17.0 "drums as arrangements"): resolve the
-    # `type: drums` pointer entries collected in the arrangements loop into a
-    # primary-first parts list. The entry whose pointer names the SAME file as
-    # the song-level `drum_tab:` key is the PRIMARY's alias — it contributes
-    # its id/name but is never loaded twice. Extra parts load their own files
-    # with the same permissive posture (a bad part disables that part only).
-    drum_parts: list[dict] | None = None
-    if drum_tab_data is not None or drum_pointer_entries:
-        _primary_id = "drums"
-        _primary_name = None
-        _extra_parts: list[dict] = []
-        _seen_rels: set[str] = set()
-        for _entry in drum_pointer_entries:
-            _rel = str(_entry.get("drum_tab") or "").strip()
-            if not _rel or _rel in _seen_rels:
-                continue
-            _seen_rels.add(_rel)
-            _eid = str(_entry.get("id") or "").strip()
-            _ename = str(_entry.get("name") or "").strip()
-            if isinstance(drum_tab_rel, str) and _rel == drum_tab_rel.strip():
-                # The primary's alias entry — adopt its identity only.
-                if _eid:
-                    _primary_id = _eid
-                if _ename:
-                    _primary_name = _ename
-                continue
-            _tab = _load_drum_tab_file(source_dir, _rel, "drum part")
-            if _tab is None:
-                continue
-            _tab_name = _tab.get("name")
-            _extra_parts.append({
-                "id": _eid or f"drums-{len(_extra_parts) + 2}",
-                "name": _ename
-                    or (_tab_name if isinstance(_tab_name, str) and _tab_name else "Drums"),
-                "drum_tab": _tab,
-            })
-        _parts: list[dict] = []
-        if drum_tab_data is not None:
-            if _primary_name is None:
-                _dt_name = drum_tab_data.get("name")
-                _primary_name = _dt_name if isinstance(_dt_name, str) and _dt_name else "Drums"
-            # The primary's payload IS the song-level tab (same object).
-            _parts.append({"id": _primary_id, "name": _primary_name, "drum_tab": drum_tab_data})
-        _parts.extend(_extra_parts)
-        if _parts:
-            if drum_tab_data is None:
-                # Pointer-only pack (a writer omitted the song-level alias —
-                # spec writers keep it, but a reader must cope): the first
-                # part becomes the primary for every legacy consumer
-                # (has_drum_tab, the default drum_tab stream, the drum-only
-                # placeholder arrangement below).
-                drum_tab_data = _parts[0]["drum_tab"]
-            drum_parts = _parts
+    # Keep the dense compatibility logic independently testable and guarantee
+    # ids are unique before the highway exposes them as selectors.
+    drum_tab_data, drum_parts = _resolve_drum_parts(
+        source_dir, drum_tab_rel, drum_tab_data, drum_pointer_entries,
+    )
 
     # Drum-only sloppak: every GP track was percussion, so it ships a
     # drum_tab but no pitched arrangements. The highway WS rejects an empty

tests/test_highway_ws_drum_parts.py

@@ -0,0 +1,17 @@
+"""Wire-compatibility coverage for selectable drum parts."""
+
+from routers.ws_highway import _drum_part_id_for_wire
+
+
+def test_single_synthesized_part_keeps_legacy_frame_without_part_id():
+    parts = [{"id": "drums", "name": "Drums", "drum_tab": {}}]
+    assert _drum_part_id_for_wire(parts, "drums") is None
+
+
+def test_multiple_parts_expose_selected_part_id():
+    parts = [
+        {"id": "drums", "name": "Drums", "drum_tab": {}},
+        {"id": "drums-2", "name": "Aux", "drum_tab": {}},
+    ]
+    assert _drum_part_id_for_wire(parts, "drums-2") == "drums-2"
+    assert _drum_part_id_for_wire(parts, None) is None

tests/test_sloppak_drum_parts.py

@@ -191,6 +191,33 @@ def test_duplicate_pointer_rels_load_once(tmp_path: Path):
     assert [p["id"] for p in loaded.drum_parts] == ["drums", "drums-2"]
 
 
+def test_duplicate_part_ids_are_made_unique(tmp_path: Path):
+    manifest = _two_part_manifest()
+    manifest["arrangements"][2]["id"] = "drums"
+    manifest["arrangements"].append(
+        {"id": "drums-2", "name": "Aux", "type": "drums",
+         "drum_tab": "drum_tab_aux.json"})
+    pak = _write_pak(tmp_path, manifest, {
+        "drum_tab.json": _tab("Drums"),
+        "drum_tab_drums-2.json": _tab("Drums (Live)"),
+        "drum_tab_aux.json": _tab("Aux"),
+    })
+    loaded = _load(pak, tmp_path)
+    assert [p["id"] for p in loaded.drum_parts] == ["drums", "drums-2", "drums-3"]
+
+
+def test_drum_pointer_with_wrong_type_logs_warning(tmp_path: Path, caplog):
+    pak = _write_pak(tmp_path, {
+        "arrangements": [
+            {"id": "lead", "name": "Lead", "file": "arrangements/lead.json"},
+            {"id": "typo", "type": "druns", "drum_tab": "drum_tab_typo.json"},
+        ],
+    }, {"drum_tab_typo.json": _tab("Typo")})
+    loaded = _load(pak, tmp_path)
+    assert loaded.drum_parts is None
+    assert "has drum_tab" in caplog.text and "type='druns'" in caplog.text
+
+
 # ── Drum-only pack with parts ────────────────────────────────────────────────
 
 def test_drum_only_pack_with_pointer_entries_still_synthesizes_placeholder(tmp_path: Path):