Add beat snap feature

cleave/analyse.py

@@ -10,6 +10,7 @@
 
 from cleave.extract import (
     extract_bass,
+    extract_drums_beats,
     extract_drums_onset,
     extract_mix_onset,
     extract_mix_rms,
@@ -37,16 +38,24 @@ def run_analyse(project_dir: Path, *, high_quality: bool) -> Path:
     )
 
     drums_onset = extract_drums_onset(paths["drums"])
+    drums_beats = extract_drums_beats(paths["drums"])
     bass = extract_bass(paths["bass"])
     vocals = extract_vocals(paths["vocals"], high_quality=high_quality)
     other = extract_other(paths["other"])
     mix_onset = extract_mix_onset(mix)
     mix_rms = extract_mix_rms(mix)
 
+    if len(drums_beats) == 0:
+        print("drum stem beat detection produced no useful data")
+        beat_times = []
+    else:
+        beat_times = [float(t) for t in drums_beats]
+
     output: dict = {
         "version": 2,
         "sample_rate_hz": int(TARGET_HZ),
         "duration_sec": duration_sec,
+        "beat_times": beat_times,
         "drums": {
             "onset_strength": resample_to_100hz(
                 *drums_onset, duration_sec

cleave/extract.py

@@ -100,6 +100,13 @@ def extract_drums_onset(path: Path | str) -> tuple[np.ndarray, np.ndarray]:
     return values, times
 
 
+def extract_drums_beats(path: Path | str) -> np.ndarray:
+    """Beat times in seconds from the drum stem."""
+    y, sr = _load(path)
+    _tempo, frames = librosa.beat.beat_track(y=y, sr=sr, hop_length=HOP_LENGTH)
+    return librosa.frames_to_time(frames, sr=sr, hop_length=HOP_LENGTH)
+
+
 def extract_mix_onset(path: Path | str) -> tuple[np.ndarray, np.ndarray]:
     """Onset strength envelope from the mixed source track."""
     return extract_drums_onset(path)

cleave/signals.py

@@ -8,7 +8,7 @@
 
 import numpy as np
 
-_META_KEYS = frozenset({"version", "sample_rate_hz", "duration_sec"})
+_META_KEYS = frozenset({"version", "sample_rate_hz", "duration_sec", "beat_times"})
 SIGNALS_VERSION = 2
 _EXPECTED_STEMS = frozenset({"drums", "bass", "vocals", "other", "full_mix"})
 _FULL_MIX_KEYS = frozenset({"onset_strength", "rms"})
@@ -32,6 +32,7 @@ class Signals:
     duration_sec: float
     path: Path
     stems: dict[str, dict[str, np.ndarray]] = field(repr=False)
+    beat_times: tuple[float, ...] = ()
     _normalized_cache: dict[tuple[str, str, float], np.ndarray] = field(
         default_factory=dict, init=False, repr=False
     )
@@ -120,9 +121,13 @@ def load_signals(path: Path) -> Signals:
 
     _validate_signals_data(data, stems)
 
+    raw_beats = data.get("beat_times", [])
+    beat_times = tuple(float(t) for t in raw_beats)
+
     return Signals(
         sample_rate_hz=float(data["sample_rate_hz"]),
         duration_sec=float(data["duration_sec"]),
         path=signals_path,
         stems=stems,
+        beat_times=beat_times,
     )

cleave/timeline.py

@@ -2,8 +2,11 @@
 
 from __future__ import annotations
 
+from collections.abc import Sequence
 from dataclasses import dataclass
 
+import numpy as np
+
 from cleave.extract import STEM_SOURCES, StemSource
 
 RECORD_DEBOUNCE_SEC = 0.08
@@ -121,3 +124,55 @@ def should_accept_toggle(last_toggle_t: float | None, t_sec: float) -> bool:
     if last_toggle_t is None:
         return True
     return t_sec - last_toggle_t >= RECORD_DEBOUNCE_SEC
+
+
+def _nearest_with_earlier_tie(t: float, candidates: Sequence[float]) -> float:
+    return min(candidates, key=lambda c: (abs(c - t), c))
+
+
+def snap_cues_to_beats(
+    cues: Sequence[TimelineCue],
+    beat_times: Sequence[float],
+) -> list[TimelineCue]:
+    """Rewrite cue times to the nearest beat; merge collisions at the same ``t``."""
+    if not cues or not beat_times:
+        return list(cues)
+
+    beats = np.asarray(beat_times, dtype=np.float64)
+    if beats.size == 1:
+        sole = float(beats[0])
+        snapped = [
+            TimelineCue(t=sole, layers=dict(cue.layers), show_tick=cue.show_tick)
+            for cue in cues
+        ]
+        return _merge_cues_at_same_t(snapped)
+
+    first = float(beats[0])
+    last = float(beats[-1])
+    interval = float(np.median(np.diff(beats)))
+
+    def snap_t(t: float) -> float:
+        if first <= t <= last:
+            idx = int(np.searchsorted(beats, t))
+            candidates: list[float] = []
+            if idx > 0:
+                candidates.append(float(beats[idx - 1]))
+            if idx < len(beats):
+                candidates.append(float(beats[idx]))
+            return _nearest_with_earlier_tie(t, candidates)
+        raw = (t - first) / interval
+        lo = int(np.floor(raw))
+        return _nearest_with_earlier_tie(
+            t,
+            (first + lo * interval, first + (lo + 1) * interval),
+        )
+
+    snapped = [
+        TimelineCue(
+            t=snap_t(cue.t),
+            layers=dict(cue.layers),
+            show_tick=cue.show_tick,
+        )
+        for cue in cues
+    ]
+    return _merge_cues_at_same_t(snapped)

cleave/viz/controls.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import shutil
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
 from pathlib import Path
 from typing import TYPE_CHECKING
 
@@ -26,6 +26,7 @@
 from cleave.viz.render_post_fx_controls import RenderPostFxControls
 from cleave.viz.settings_controls import SettingsControls
 from cleave.viz.timeline_preset_controls import TimelinePresetController
+from cleave.viz.timeline_snap_controls import TimelineSnapController
 from cleave.viz.user_presets import resolve_user_preset_dest, user_preset_item_display_name
 from cleave.viz.focus_nav import (
     FocusCursor,
@@ -81,6 +82,7 @@ def __init__(
         repo_root_example: Path | None = None,
         modal_host: ModalHost | None = None,
         layer_manager: LayerManager | None = None,
+        beat_times: Sequence[float] = (),
     ) -> None:
         self.session = session
         self.cfg = cfg
@@ -124,6 +126,12 @@ def __init__(
             self._modal_host,
             on_notification=self.show_notification,
         )
+        self._timeline_snap = TimelineSnapController(
+            session,
+            self._modal_host,
+            beat_times,
+            on_notification=self.show_notification,
+        )
         self._view_state = TuningViewStateBuilder(
             session,
             playback,
@@ -377,6 +385,9 @@ def handle_keydown(self, event: pygame.event.Event) -> bool:
             if kind == RowKind.TIMELINE_PRESETS:
                 self._timeline_presets.prompt(self.duration_sec)
                 return True
+            if kind == RowKind.TIMELINE_SNAP_TO_BEATS:
+                self._timeline_snap.prompt()
+                return True
             if kind == RowKind.TRACK_PRESET_DIR:
                 slot = self.focus_descriptor.slot
                 if slot is not None:

cleave/viz/row_fields.py

@@ -1263,6 +1263,10 @@ def _apply_transport(
         panel_label="apply a preset",
         present_style=RowPresentStyle.FULL_LINE,
     ),
+    RowKind.TIMELINE_SNAP_TO_BEATS: RowFieldDef(
+        panel_label="snap to beats",
+        present_style=RowPresentStyle.FULL_LINE,
+    ),
     RowKind.PANEL_NOTIFICATION: RowFieldDef(
         panel_label="",
         present_style=RowPresentStyle.FULL_LINE,

cleave/viz/row_sections.py

@@ -683,6 +683,7 @@ def _build_row_tree_indent_depth() -> dict[RowKind, int]:
     depths[RowKind.TRACK_USER_PRESET_ITEM] = 7
     depths[RowKind.TRACK_USER_PRESET_ADD] = 3
     depths[RowKind.TIMELINE_PRESETS] = 1
+    depths[RowKind.TIMELINE_SNAP_TO_BEATS] = 1
     return depths
 
 
@@ -797,6 +798,7 @@ def append_render_section_rows(
                 and state.render_timeline.expanded
             ):
                 row_list.append(RowDescriptor(RowKind.TIMELINE_PRESETS))
+                row_list.append(RowDescriptor(RowKind.TIMELINE_SNAP_TO_BEATS))
 
 
 def append_track_section_rows(

cleave/viz/row_semantics.py

@@ -75,6 +75,7 @@ class RowKind(Enum):
     RENDER_POST_FX_CHROMA_BOOST_AMOUNT = auto()
     RENDER_TIMELINE_HEADER = auto()
     TIMELINE_PRESETS = auto()
+    TIMELINE_SNAP_TO_BEATS = auto()
     SETTINGS_HEADER = auto()
     SETTINGS_PREVIEW_QUALITY = auto()
     SETTINGS_UI_HEADER = auto()
@@ -691,6 +692,16 @@ class RowBehavior:
         ),
         help_mode_entries=TIMELINE_PRESET_HELP_ENTRIES,
     ),
+    RowKind.TIMELINE_SNAP_TO_BEATS: RowBehavior(
+        RowAffordance.ACTION,
+        navigable=True,
+        help_title="Snap to beats",
+        help_entries=(("Enter", "snap cues to beats"),),
+        help_description=(
+            "Snap all committed timeline cues to the nearest beat",
+            "(irreversible).",
+        ),
+    ),
     RowKind.SETTINGS_HEADER: RowBehavior(
         RowAffordance.EXPAND,
         is_header=True,

cleave/viz/timeline_snap_controls.py

@@ -0,0 +1,51 @@
+"""Timeline beat-snap confirm modal and cue rewrite."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Sequence
+
+from cleave.timeline import snap_cues_to_beats
+from cleave.viz.modal import ModalHost
+from cleave.viz.session import TuningSession
+
+
+class TimelineSnapController:
+    """Prompt for and apply beat snapping to committed timeline cues."""
+
+    def __init__(
+        self,
+        session: TuningSession,
+        modal_host: ModalHost,
+        beat_times: Sequence[float],
+        *,
+        on_notification: Callable[[str], None] | None = None,
+    ) -> None:
+        self.session = session
+        self._modal = modal_host
+        self._beat_times = tuple(beat_times)
+        self._on_notification = on_notification
+
+    def prompt(self) -> None:
+        tl = self.session.timeline
+        if tl.recording:
+            return
+        if not tl.cues:
+            self._notify("No timeline cues to snap")
+            return
+        if not self._beat_times:
+            self._notify("No beats available; re-run separate")
+            return
+        self._modal.prompt_yes_no(
+            "Do you want to snap all timeline cues to nearest beat?",
+            on_confirm=self._snap,
+            cancel_label="CANCEL",
+        )
+
+    def _snap(self) -> None:
+        tl = self.session.timeline
+        tl.cues = snap_cues_to_beats(tl.cues, self._beat_times)
+        self._notify("Snapped timeline cues to beats")
+
+    def _notify(self, message: str) -> None:
+        if self._on_notification is not None:
+            self._on_notification(message)

cleave/viz/tuning_panel_draw.py

@@ -538,6 +538,7 @@ def _row_value_color(state: TuningViewState, index: int) -> tuple[int, int, int]
         RowKind.LAYER_MANAGEMENT_DELETE,
         RowKind.TRACK_USER_PRESET_ADD,
         RowKind.TIMELINE_PRESETS,
+        RowKind.TIMELINE_SNAP_TO_BEATS,
     }:
         if kind == RowKind.CONFIG_HEADER and state.solo_active:
             return DISABLED
@@ -904,6 +905,7 @@ def _estimate_row_content_width(
             RowKind.LAYER_MANAGEMENT_DELETE,
             RowKind.TRACK_USER_PRESET_ADD,
             RowKind.TIMELINE_PRESETS,
+            RowKind.TIMELINE_SNAP_TO_BEATS,
         }
     ):
         label = _row_text(state, index)
@@ -1389,6 +1391,7 @@ def _build_row_at_index(
                 RowKind.LAYER_MANAGEMENT_DELETE,
                 RowKind.TRACK_USER_PRESET_ADD,
                 RowKind.TIMELINE_PRESETS,
+                RowKind.TIMELINE_SNAP_TO_BEATS,
             }
         ):
             label = _row_text(state, index)

cleave/viz/wiring.py

@@ -430,6 +430,7 @@ def on_chroma_boost_apply_mode_change(old_mode: str, new_mode: str) -> None:
         "layer_bindings": layer_bindings,
         "render_post_fx_bindings": render_post_fx_bindings,
         "layer_manager": layer_manager,
+        "beat_times": list(signals.beat_times) if signals is not None else [],
     }
     if modal_host is not None:
         kwargs["modal_host"] = modal_host

docs/roadmap.md

@@ -5,9 +5,12 @@ Aspirational ideas. Not scheduled; revisit when the core workflow feels solid.
 
 ## Timeline v2
 - Fade in/out on layer transitions
-- Beat snap for cue placement
 - External timeline file for very long cue lists
 
+### Beat detection (cue snap) v2
+
+Investigate [madmom](https://github.com/CPJKU/madmom): trained beat and downbeat models for stronger bar alignment and tempo-map quality than librosa alone. Same persisted grid feeds timeline snap and MIDI out; heavier analyse step and new dependency. (v1 librosa snap is in [todos.md](todos.md).)
+
 ## MIDI out
 
 Emit MIDI notes or CC from drum onsets (and other signals in `signals.json`) to drive hardware lighting, drum pads, or synths during playback or export.

docs/todos.md

@@ -15,6 +15,18 @@ Outstanding bugs and issues.
 
 ---
 
+## Features
+
+### Timeline beat snap (v1)
+
+Done. Batch-quantize committed timeline cues to a librosa beat grid from the drums stem (`beat_times` in `signals.json`). Green **snap to beats** ACTION row under Render: TIMELINE.
+
+**v1.1**
+
+- Optional bar snap: every Nth beat (default N=4, assume 4/4 phase).
+
+---
+
 ## Architecture
 
 ### projectM