Beat this detection

README.md

@@ -203,6 +203,7 @@ git submodule update --init --recursive
 
 * [Milkdrop / projectM](https://github.com/projectM-visualizer/projectM) - visualizer engine
 * [Demucs](https://github.com/facebookresearch/demucs) - audio separation
+* [Beat This!](https://github.com/CPJKU/beat_this) - beat and downbeat detection
 * [FFmpeg](https://ffmpeg.org) - video encoding
 * [pygame](https://www.pygame.org/) - window, input, overlay UI, and SDL2 audio
 * [OpenGL](https://www.opengl.org/) / [PyOpenGL](https://pyopengl.sourceforge.io/) - layer compositing and rendering

cleave/analyse.py

@@ -10,7 +10,7 @@
 
 from cleave.extract import (
     extract_bass,
-    extract_drums_beats,
+    extract_beats_downbeats,
     extract_drums_onset,
     extract_mix_onset,
     extract_mix_rms,
@@ -38,24 +38,31 @@ def run_analyse(project_dir: Path, *, high_quality: bool) -> Path:
     )
 
     drums_onset = extract_drums_onset(paths["drums"])
-    drums_beats = extract_drums_beats(paths["drums"])
+    beats, downbeats = extract_beats_downbeats(mix)
     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")
+    if len(beats) == 0:
+        print("mix beat detection produced no useful data")
         beat_times = []
     else:
-        beat_times = [float(t) for t in drums_beats]
+        beat_times = [float(t) for t in beats]
+
+    if len(downbeats) == 0:
+        print("mix downbeat detection produced no useful data")
+        downbeat_times = []
+    else:
+        downbeat_times = [float(t) for t in downbeats]
 
     output: dict = {
-        "version": 2,
+        "version": 3,
         "sample_rate_hz": int(TARGET_HZ),
         "duration_sec": duration_sec,
         "beat_times": beat_times,
+        "downbeat_times": downbeat_times,
         "drums": {
             "onset_strength": resample_to_100hz(
                 *drums_onset, duration_sec

cleave/extract.py

@@ -1,4 +1,4 @@
-"""Per-stem librosa feature extraction at native analysis rates."""
+"""Per-stem feature extraction at native analysis rates."""
 
 from __future__ import annotations
 
@@ -7,6 +7,8 @@
 
 import librosa
 import numpy as np
+import torch
+from beat_this.inference import File2Beats
 
 HOP_LENGTH = 512
 BASS_RMS_HOP_MS = 0.02
@@ -100,37 +102,18 @@ 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.
+def extract_beats_downbeats(path: Path | str) -> tuple[np.ndarray, np.ndarray]:
+    """Beat and downbeat times in seconds from the mixed source track.
 
-    Uses a per-frame tempo curve so the DP tracker can follow tempo changes.
-    Falls back to a global-tempo track if the dynamic run fails or finds no beats.
+    Runs Beat This! (`File2Beats`) on the mix path.
     """
-    y, sr = _load(path)
-    onset_env = librosa.onset.onset_strength(y=y, sr=sr, hop_length=HOP_LENGTH)
-    frames: np.ndarray
-    try:
-        dtempo = librosa.feature.tempo(
-            onset_envelope=onset_env,
-            sr=sr,
-            hop_length=HOP_LENGTH,
-            aggregate=None,
-        )
-        _tempo, frames = librosa.beat.beat_track(
-            onset_envelope=onset_env,
-            sr=sr,
-            hop_length=HOP_LENGTH,
-            bpm=dtempo,
-        )
-        if len(frames) == 0:
-            raise ValueError("dynamic beat_track returned no beats")
-    except Exception:
-        _tempo, frames = librosa.beat.beat_track(
-            onset_envelope=onset_env,
-            sr=sr,
-            hop_length=HOP_LENGTH,
-        )
-    return librosa.frames_to_time(frames, sr=sr, hop_length=HOP_LENGTH)
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    file2beats = File2Beats(checkpoint_path="final0", device=device, dbn=False)
+    beats, downbeats = file2beats(str(path))
+    return (
+        np.asarray(beats, dtype=np.float64),
+        np.asarray(downbeats, dtype=np.float64),
+    )
 
 
 def extract_mix_onset(path: Path | str) -> tuple[np.ndarray, np.ndarray]:

cleave/separate.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 import shutil
 import subprocess
 import sys
@@ -13,6 +14,7 @@
 from cleave.extract import stem_paths, stems_dir
 from cleave.paths import project_dir, project_slug, resolve_project
 from cleave.project import load_manifest, manifest_path, mix_path, write_manifest
+from cleave.signals import SIGNALS_VERSION
 
 
 def project_stems_complete(project_dir: Path) -> bool:
@@ -22,8 +24,18 @@ def project_stems_complete(project_dir: Path) -> bool:
 
 
 def signals_complete(project_dir: Path) -> bool:
-    """Return True when ``signals.json`` exists in *project_dir*."""
-    return (project_dir / "signals.json").is_file()
+    """Return True when ``signals.json`` exists at the current schema version."""
+    path = project_dir / "signals.json"
+    if not path.is_file():
+        return False
+    try:
+        with path.open(encoding="utf-8") as handle:
+            data = json.load(handle)
+    except (OSError, json.JSONDecodeError, UnicodeDecodeError):
+        return False
+    if not isinstance(data, dict):
+        return False
+    return data.get("version") == SIGNALS_VERSION
 
 
 def resolve_separate_target(path_or_slug: Path | str) -> tuple[Path, Path]:

cleave/signals.py

@@ -8,8 +8,10 @@
 
 import numpy as np
 
-_META_KEYS = frozenset({"version", "sample_rate_hz", "duration_sec", "beat_times"})
-SIGNALS_VERSION = 2
+_META_KEYS = frozenset(
+    {"version", "sample_rate_hz", "duration_sec", "beat_times", "downbeat_times"}
+)
+SIGNALS_VERSION = 3
 _EXPECTED_STEMS = frozenset({"drums", "bass", "vocals", "other", "full_mix"})
 _FULL_MIX_KEYS = frozenset({"onset_strength", "rms"})
 
@@ -33,6 +35,7 @@ class Signals:
     path: Path
     stems: dict[str, dict[str, np.ndarray]] = field(repr=False)
     beat_times: tuple[float, ...] = ()
+    downbeat_times: tuple[float, ...] = ()
     _normalized_cache: dict[tuple[str, str, float], np.ndarray] = field(
         default_factory=dict, init=False, repr=False
     )
@@ -123,11 +126,14 @@ def load_signals(path: Path) -> Signals:
 
     raw_beats = data.get("beat_times", [])
     beat_times = tuple(float(t) for t in raw_beats)
+    raw_downbeats = data.get("downbeat_times", [])
+    downbeat_times = tuple(float(t) for t in raw_downbeats)
 
     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,
+        downbeat_times=downbeat_times,
     )

cleave/timeline.py

@@ -201,78 +201,35 @@ 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(
+def shift_bars_by_beats(
+    downbeat_times: Sequence[float],
     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,
+    offset: int,
 ) -> 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])
+    """Map each downbeat to the beat ``offset`` positions away (clamped).
 
-
-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:
+    Each downbeat is matched to the nearest beat (earlier on a tie), then the
+    beat index is shifted by ``offset`` and clamped to the beat grid.
+    """
+    if not downbeat_times or not beat_times:
         return ()
-    return bar_times_at_phase(beat_times, k, beats_per_bar=beats_per_bar)
-
-
-def bar_phase_matching(
-    beat_times: Sequence[float],
-    bar_times: Sequence[float],
-    *,
-    beats_per_bar: int = 4,
-) -> int | None:
-    """Return phase ``k`` where ``bar_times_at_phase`` equals ``bar_times``."""
-    if beats_per_bar < 1:
-        raise ValueError(f"beats_per_bar must be >= 1 (got {beats_per_bar})")
-    if not beat_times or not bar_times:
-        return None
-    target = tuple(float(t) for t in bar_times)
-    for k in range(beats_per_bar):
-        if (
-            bar_times_at_phase(beat_times, k, beats_per_bar=beats_per_bar)
-            == target
-        ):
-            return k
-    return None
+    beats = np.asarray(beat_times, dtype=np.float64)
+    last = len(beats) - 1
+    result: list[float] = []
+    for t in downbeat_times:
+        idx = int(np.searchsorted(beats, float(t)))
+        candidates: list[int] = []
+        if idx > 0:
+            candidates.append(idx - 1)
+        if idx <= last:
+            candidates.append(idx)
+        nearest = min(
+            candidates,
+            key=lambda i: (abs(float(beats[i]) - float(t)), float(beats[i])),
+        )
+        shifted = max(0, min(last, nearest + offset))
+        result.append(float(beats[shifted]))
+    return tuple(result)
 
 
 def snap_lane_to_beats(

cleave/viz/timeline_snap_controls.py

@@ -5,17 +5,15 @@
 from collections.abc import Callable, Sequence
 
 from cleave.timeline import (
-    bar_phase_matching,
-    bar_times_at_phase,
     empty_lane,
+    shift_bars_by_beats,
     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_PHASE_OFFSETS = (0, 1, 2, 3)
+_BAR_BEAT_OFFSETS = (0, 1, 2, 3)
 _BAR_SNAP_MESSAGE = "Snap timeline cues to nearest bar (choose bar phase)"
 
 
@@ -35,12 +33,6 @@ def __init__(
         self._modal = modal_host
         self._beat_times = tuple(beat_times)
         self._bar_times = tuple(bar_times)
-        matched = bar_phase_matching(
-            self._beat_times,
-            self._bar_times,
-            beats_per_bar=_BEATS_PER_BAR,
-        )
-        self._bar_phase = 0 if matched is None else matched
         self._on_notification = on_notification
 
     def prompt(self) -> None:
@@ -70,22 +62,21 @@ def prompt_bars(self) -> None:
                 f"{offset:+d}",
                 lambda o=offset: self._snap_bars_at_offset(o),
             )
-            for offset in _BAR_PHASE_OFFSETS
+            for offset in _BAR_BEAT_OFFSETS
         ]
         options.append(ModalOption(_CANCEL_LABEL, dismiss))
         self._modal.prompt_choice(
             _BAR_SNAP_MESSAGE,
             options,
             on_dismiss=dismiss,
-            initial_focus_index=_BAR_PHASE_OFFSETS.index(0),
+            initial_focus_index=_BAR_BEAT_OFFSETS.index(0),
         )
 
     def _snap_bars_at_offset(self, offset: int) -> None:
-        phase = (self._bar_phase + offset) % _BEATS_PER_BAR
-        grid = bar_times_at_phase(
+        grid = shift_bars_by_beats(
+            self._bar_times,
             self._beat_times,
-            phase,
-            beats_per_bar=_BEATS_PER_BAR,
+            offset,
         )
         label = f"{offset:+d}"
         notify_msg = f"Snapped timeline cues to bars ({label})"

cleave/viz/wiring.py

@@ -46,7 +46,6 @@
     sync_manual_browse_with_user_defined_rotation,
 )
 from cleave.stem_pcm import StemPcmBank
-from cleave.timeline import bar_times_from_beats
 from cleave.viz.playback import current_sec, seek
 from cleave.config import VIZ_CONFIG_FILENAME
 
@@ -423,11 +422,7 @@ def on_chroma_boost_apply_mode_change(old_mode: str, new_mode: str) -> None:
     bar_times: list[float] = []
     if signals is not None:
         beat_times = list(signals.beat_times)
-        if beat_times:
-            onset_at_beats = [
-                signals.sample("drums", "onset_strength", t) for t in beat_times
-            ]
-            bar_times = list(bar_times_from_beats(beat_times, onset_at_beats))
+        bar_times = list(signals.downbeat_times)
 
     kwargs: dict = {
         "session": session,