Bar snap

cleave/timeline.py

@@ -201,6 +201,105 @@ def _nearest_with_earlier_tie(t: float, candidates: Sequence[float]) -> float:
     return min(candidates, key=lambda c: (abs(c - t), c))
 
 
+def bar_phase_from_beats(
+    beat_times: Sequence[float],
+    onset_at_beats: Sequence[float],
+    *,
+    beats_per_bar: int = 4,
+) -> int | None:
+    """Pick 4/4 bar phase by maximizing summed onset at candidate downbeats."""
+    if len(beat_times) != len(onset_at_beats):
+        raise ValueError(
+            "beat_times and onset_at_beats must have the same length "
+            f"(got {len(beat_times)} and {len(onset_at_beats)})"
+        )
+    if beats_per_bar < 1:
+        raise ValueError(f"beats_per_bar must be >= 1 (got {beats_per_bar})")
+    n = (len(beat_times) // beats_per_bar) * beats_per_bar
+    if n < beats_per_bar:
+        return None
+    onset = np.asarray(onset_at_beats[:n], dtype=np.float64).reshape(
+        -1, beats_per_bar
+    )
+    return int(np.argmax(onset.sum(axis=0)))
+
+
+def bar_times_at_phase(
+    beat_times: Sequence[float],
+    phase: int,
+    *,
+    beats_per_bar: int = 4,
+) -> tuple[float, ...]:
+    """Return every ``beats_per_bar``-th beat starting at ``phase``."""
+    if beats_per_bar < 1:
+        raise ValueError(f"beats_per_bar must be >= 1 (got {beats_per_bar})")
+    if not beat_times:
+        return ()
+    phase = phase % beats_per_bar
+    return tuple(beat_times[phase::beats_per_bar])
+
+
+def bar_times_from_beats(
+    beat_times: Sequence[float],
+    onset_at_beats: Sequence[float],
+    *,
+    beats_per_bar: int = 4,
+) -> tuple[float, ...]:
+    """Bar grid at the onset-strongest phase (see ``bar_phase_from_beats``)."""
+    k = bar_phase_from_beats(
+        beat_times, onset_at_beats, beats_per_bar=beats_per_bar
+    )
+    if k is None:
+        return ()
+    return bar_times_at_phase(beat_times, k, beats_per_bar=beats_per_bar)
+
+
+def grid_period(times: Sequence[float]) -> float | None:
+    """Median spacing between successive times, or None if fewer than two."""
+    if len(times) < 2:
+        return None
+    return float(np.median(np.diff(np.asarray(times, dtype=np.float64))))
+
+
+def bar_period_sec(
+    bar_times: Sequence[float],
+    beat_times: Sequence[float] = (),
+    *,
+    beats_per_bar: int = 4,
+) -> float | None:
+    """Bar period from bar-grid median spacing, else ``beats_per_bar`` beat medians."""
+    period = grid_period(bar_times)
+    if period is not None:
+        return period
+    beat_period = grid_period(beat_times)
+    if beat_period is not None:
+        return float(beats_per_bar) * beat_period
+    return None
+
+
+def shift_lane_times(
+    lane: TimelineLane,
+    delta_sec: float,
+    *,
+    t_min: float = 0.0,
+    t_max: float,
+) -> TimelineLane:
+    """Shift cue times by ``delta_sec``, clamp to ``[t_min, t_max]``, canonicalize."""
+    if not lane.cues:
+        return TimelineLane(baseline=lane.baseline, cues=[])
+    shifted = [
+        SlotCue(
+            t=max(t_min, min(t_max, cue.t + delta_sec)),
+            visible=cue.visible,
+        )
+        for cue in lane.cues
+    ]
+    return TimelineLane(
+        baseline=lane.baseline,
+        cues=canonicalize(lane.baseline, shifted),
+    )
+
+
 def snap_lane_to_beats(
     lane: TimelineLane,
     beat_times: Sequence[float],

cleave/viz/controls.py

@@ -83,6 +83,7 @@ def __init__(
         modal_host: ModalHost | None = None,
         layer_manager: LayerManager | None = None,
         beat_times: Sequence[float] = (),
+        bar_times: Sequence[float] = (),
     ) -> None:
         self.session = session
         self.cfg = cfg
@@ -130,6 +131,7 @@ def __init__(
             session,
             self._modal_host,
             beat_times,
+            bar_times,
             on_notification=self.show_notification,
         )
         self._view_state = TuningViewStateBuilder(
@@ -388,6 +390,9 @@ def handle_keydown(self, event: pygame.event.Event) -> bool:
             if kind == RowKind.TIMELINE_SNAP_TO_BEATS:
                 self._timeline_snap.prompt()
                 return True
+            if kind == RowKind.TIMELINE_SNAP_TO_BARS:
+                self._timeline_snap.prompt_bars(self.duration_sec)
+                return True
             if kind == RowKind.TRACK_PRESET_DIR:
                 slot = self.focus_descriptor.slot
                 if slot is not None:

cleave/viz/modal.py

@@ -23,6 +23,13 @@ def modal_options_vertical(option_count: int) -> bool:
     return option_count > MODAL_VERTICAL_OPTION_THRESHOLD
 
 
+def clamp_modal_focus_index(index: int, option_count: int) -> int:
+    """Return ``index`` when in range, otherwise ``0`` (first option)."""
+    if option_count <= 0 or index < 0 or index >= option_count:
+        return 0
+    return index
+
+
 @dataclass
 class ModalOption:
     label: str
@@ -35,6 +42,7 @@ class ModalRequest:
     message: str | None
     options: list[ModalOption]
     on_dismiss: Callable[[], None] | None = None
+    initial_focus_index: int = 0
 
 
 @dataclass(frozen=True)
@@ -76,7 +84,10 @@ def view_state(self) -> ModalViewState | None:
 
     def prompt(self, request: ModalRequest) -> None:
         self._request = request
-        self._focus_index = 0
+        self._focus_index = clamp_modal_focus_index(
+            request.initial_focus_index,
+            len(request.options),
+        )
 
     def prompt_yes_no(
         self,
@@ -107,13 +118,16 @@ def prompt_choice(
         message: str,
         options: list[ModalOption],
         on_dismiss: Callable[[], None] | None = None,
+        *,
+        initial_focus_index: int = 0,
     ) -> None:
         self.prompt(
             ModalRequest(
                 kind=ModalKind.CHOICE,
                 message=message,
                 options=options,
                 on_dismiss=on_dismiss,
+                initial_focus_index=initial_focus_index,
             )
         )
 

cleave/viz/row_fields.py

@@ -1267,6 +1267,10 @@ def _apply_transport(
         panel_label="snap to beats",
         present_style=RowPresentStyle.FULL_LINE,
     ),
+    RowKind.TIMELINE_SNAP_TO_BARS: RowFieldDef(
+        panel_label="snap to bars",
+        present_style=RowPresentStyle.FULL_LINE,
+    ),
     RowKind.PANEL_NOTIFICATION: RowFieldDef(
         panel_label="",
         present_style=RowPresentStyle.FULL_LINE,

cleave/viz/row_sections.py

@@ -684,6 +684,7 @@ def _build_row_tree_indent_depth() -> dict[RowKind, int]:
     depths[RowKind.TRACK_USER_PRESET_ADD] = 3
     depths[RowKind.TIMELINE_PRESETS] = 1
     depths[RowKind.TIMELINE_SNAP_TO_BEATS] = 1
+    depths[RowKind.TIMELINE_SNAP_TO_BARS] = 1
     return depths
 
 
@@ -799,6 +800,7 @@ def append_render_section_rows(
             ):
                 row_list.append(RowDescriptor(RowKind.TIMELINE_PRESETS))
                 row_list.append(RowDescriptor(RowKind.TIMELINE_SNAP_TO_BEATS))
+                row_list.append(RowDescriptor(RowKind.TIMELINE_SNAP_TO_BARS))
 
 
 def append_track_section_rows(

cleave/viz/row_semantics.py

@@ -76,6 +76,7 @@ class RowKind(Enum):
     RENDER_TIMELINE_HEADER = auto()
     TIMELINE_PRESETS = auto()
     TIMELINE_SNAP_TO_BEATS = auto()
+    TIMELINE_SNAP_TO_BARS = auto()
     SETTINGS_HEADER = auto()
     SETTINGS_PREVIEW_QUALITY = auto()
     SETTINGS_UI_HEADER = auto()
@@ -702,6 +703,16 @@ class RowBehavior:
             "(irreversible).",
         ),
     ),
+    RowKind.TIMELINE_SNAP_TO_BARS: RowBehavior(
+        RowAffordance.ACTION,
+        navigable=True,
+        help_title="Snap to bars",
+        help_entries=(("Enter", "choose bar phase offset"),),
+        help_description=(
+            "Snap all committed timeline cues to the nearest bar;",
+            "Enter opens a phase-offset choice (irreversible).",
+        ),
+    ),
     RowKind.SETTINGS_HEADER: RowBehavior(
         RowAffordance.EXPAND,
         is_header=True,

cleave/viz/timeline_snap_controls.py

@@ -1,31 +1,51 @@
-"""Timeline beat-snap confirm modal and cue rewrite."""
+"""Timeline beat/bar-snap confirm modal and cue rewrite."""
 
 from __future__ import annotations
 
 from collections.abc import Callable, Sequence
 
-from cleave.timeline import empty_lane, snap_lane_to_beats
-from cleave.viz.modal import ModalHost
+from cleave.timeline import (
+    bar_period_sec,
+    empty_lane,
+    shift_lane_times,
+    snap_lane_to_beats,
+)
+from cleave.viz.modal import ModalHost, ModalOption
 from cleave.viz.session import TuningSession
 
+_CANCEL_LABEL = "Cancel"
+_BEATS_PER_BAR = 4
+_BAR_OFFSETS = (-3, -2, -1, 0, 1, 2, 3)
+_BAR_SNAP_MESSAGE = "Snap timeline cues to nearest bar (choose bar offset)"
+
 
 class TimelineSnapController:
-    """Prompt for and apply beat snapping to committed timeline cues."""
+    """Prompt for and apply beat/bar snapping to committed timeline cues."""
 
     def __init__(
         self,
         session: TuningSession,
         modal_host: ModalHost,
         beat_times: Sequence[float],
+        bar_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._bar_times = tuple(bar_times)
         self._on_notification = on_notification
 
     def prompt(self) -> None:
+        self._prompt(
+            self._beat_times,
+            empty_grid_msg="No beats available; re-run separate",
+            confirm_msg="Do you want to snap all timeline cues to nearest beat?",
+            done_msg="Snapped timeline cues to beats",
+        )
+
+    def prompt_bars(self, duration_sec: float) -> None:
         tl = self.session.timeline
         if tl.recording:
             return
@@ -35,20 +55,83 @@ def prompt(self) -> None:
         if not self._beat_times:
             self._notify("No beats available; re-run separate")
             return
+        if not self._bar_times:
+            self._notify("No bars available; re-run separate")
+            return
+        dismiss = lambda: None
+        options: list[ModalOption] = [
+            ModalOption(
+                f"{offset:+d}",
+                lambda o=offset: self._snap_bars_at_offset(o, duration_sec),
+            )
+            for offset in _BAR_OFFSETS
+        ]
+        options.append(ModalOption(_CANCEL_LABEL, dismiss))
+        self._modal.prompt_choice(
+            _BAR_SNAP_MESSAGE,
+            options,
+            on_dismiss=dismiss,
+            initial_focus_index=_BAR_OFFSETS.index(0),
+        )
+
+    def _snap_bars_at_offset(self, offset: int, duration_sec: float) -> None:
+        grid = self._bar_times
+        label = f"{offset:+d}"
+        notify_msg = f"Snapped timeline cues to bars ({label})"
+        period = (
+            bar_period_sec(
+                grid, self._beat_times, beats_per_bar=_BEATS_PER_BAR
+            )
+            if offset != 0
+            else None
+        )
+        tl = self.session.timeline
+        for slot in list(tl.lanes):
+            lane = snap_lane_to_beats(
+                tl.lanes.get(slot) or empty_lane(),
+                grid,
+            )
+            if offset != 0 and period is not None:
+                lane = shift_lane_times(
+                    lane,
+                    offset * period,
+                    t_min=0.0,
+                    t_max=duration_sec,
+                )
+            tl.lanes[slot] = lane
+        self._notify(notify_msg)
+
+    def _prompt(
+        self,
+        grid: Sequence[float],
+        *,
+        empty_grid_msg: str,
+        confirm_msg: str,
+        done_msg: str,
+    ) -> None:
+        tl = self.session.timeline
+        if tl.recording:
+            return
+        if not any(lane.cues for lane in tl.lanes.values()):
+            self._notify("No timeline cues to snap")
+            return
+        if not grid:
+            self._notify(empty_grid_msg)
+            return
         self._modal.prompt_yes_no(
-            "Do you want to snap all timeline cues to nearest beat?",
-            on_confirm=self._snap,
+            confirm_msg,
+            on_confirm=lambda: self._snap_to(grid, done_msg),
             cancel_label="CANCEL",
         )
 
-    def _snap(self) -> None:
+    def _snap_to(self, grid: Sequence[float], notify_msg: str) -> None:
         tl = self.session.timeline
         for slot in list(tl.lanes):
             tl.lanes[slot] = snap_lane_to_beats(
                 tl.lanes.get(slot) or empty_lane(),
-                self._beat_times,
+                grid,
             )
-        self._notify("Snapped timeline cues to beats")
+        self._notify(notify_msg)
 
     def _notify(self, message: str) -> None:
         if self._on_notification is not None: