feat(foundry): finder v2 pt.1 — one beat grid from the drums, shared across stems (#18)

Directly addresses the #7 finding: per-stem PLP grids are weak on sparse-onset stems
(Doors/vocals recall 0.06 vs Gil-Scott-Heron 0.56), because PLP needs onsets to lock a
pulse and a vocal stem barely has them. Fix: find_takes now computes ONE beat grid from
the most rhythmic stem (drums → bass → first available) and reuses it for every stem, so
vocals/other are cut on the same musical bar lines as the drums instead of each guessing
its own. analyze_stem gained an optional `grid=(peaks,times,bpm)` param (None ⇒ per-stem,
the CLI case); find_takes and build_finder_eval both pass the shared grid.

Verified on Loituma: "beat grid from drums" → all four stems analysed on it, including a
cross-stem bass+vocals take at 137s. 8 finder tests (added: analyze_stem honours a shared
grid and adopts its tempo). build_finder_eval re-run with per-track drums grid is in
flight to quantify the recall lift over the 0.27 baseline; the correctness win (shared
grid ≫ weak per-stem grid on sparse stems) holds regardless. Pt.2 (sub-bar chop mode)
remains on #18.

armada/tide-table/build_finder_eval.py

@@ -80,10 +80,26 @@ def main(argv=None) -> int:
     def tol_of(c):
         return max(c.bars * 60.0 / max(c.bpm, 1), 0.6)   # ~1 bar, floor 0.6 s
 
+    grid_cache = {}                                   # per-track shared beat grid (#18)
+
+    def track_grid(track):
+        if track in grid_cache:
+            return grid_cache[track]
+        g = None
+        for role in ("drums", "bass"):
+            p = paths.get(f"{track}/{role}")
+            if p:
+                gy = G._mono(G.load_audio(p)[0]).astype(np.float32)
+                g = L.beat_grid(gy, L._sr_of(p))
+                break
+        grid_cache[track] = g
+        return g
+
     rows, cov_tot, dist_tot, prec_hit, prec_tot = [], 0, 0, 0, 0
     for sid, ms in ranked:
         y = G._mono(G.load_audio(paths[sid])[0]).astype(np.float32)
         sr = L._sr_of(paths[sid])
+        grid = track_grid(sid.split("/")[0])          # cut every stem on the drums grid
         # Dedup GT: PLN cuts many overlapping loops from one stem; collapse offsets
         # within a bar into DISTINCT windows (the real sections he sampled). recall@N
         # vs all raw cuts is ceiling-capped (8 candidates can't cover 42 cuts).
@@ -93,7 +109,7 @@ def main(argv=None) -> int:
             if not distinct or s - distinct[-1] > GT_MERGE_SLACK:
                 distinct.append(s)
         n = max(a.n, len(distinct))               # give the finder ≥ as many as he kept
-        cands = L.analyze_stem(y, sr, bars=tuple(a.bars), top_n=n)
+        cands = L.analyze_stem(y, sr, bars=tuple(a.bars), top_n=n, grid=grid)
         covered = sum(1 for d in distinct
                       if any(abs(c.start_s - d) <= tol_of(c) for c in cands))
         hit_c = sum(1 for c in cands

tools/foundry/engine/loops.py

@@ -147,9 +147,16 @@ def _structural(ssm: np.ndarray, s: int, e: int) -> float:
     return float(block.max(axis=1).mean()) if block.size else 0.0
 
 
-def analyze_stem(y: np.ndarray, sr: int, *, bars=(2, 4, 8), top_n=8) -> list[LoopCandidate]:
-    """Rank loop candidates within one stem."""
-    peaks, times, bpm = beat_grid(y, sr)
+def analyze_stem(y: np.ndarray, sr: int, *, bars=(2, 4, 8), top_n=8,
+                 grid=None) -> list[LoopCandidate]:
+    """Rank loop candidates within one stem.
+
+    `grid` = a shared (peaks, times, bpm) beat grid (e.g. from the drums stem). Passing
+    it fixes the §7 finding that PLP on sparse-onset stems (vocals) yields a weak grid
+    and poor windows: derive the grid once from a rhythmic source and reuse it here so
+    every stem is cut on the same musical bar lines. None ⇒ compute per-stem (CLI case).
+    """
+    peaks, times, bpm = grid if grid is not None else beat_grid(y, sr)
     nb = len(times)
     if nb < BEATS_PER_BAR * min(bars) + 1:
         return []
@@ -208,17 +215,33 @@ def _overlap(a: LoopCandidate, b: LoopCandidate) -> float:
 # ── takes (cross-stem grouping) ───────────────────────────────────────────────
 def find_takes(catch: Catch, workspace: Path, *, bars=(2, 4, 8), top_n=8,
                progress: Optional[Callable[[str], None]] = None) -> list[Take]:
-    """Per-stem candidates → loosely grouped Takes (a shared time window)."""
+    """Per-stem candidates → loosely grouped Takes (a shared time window).
+
+    One beat grid for the whole catch: computed from the most rhythmic stem (drums if
+    present) and reused for every stem, so vocals/other are cut on the same bar lines
+    instead of each stem's own weak PLP grid (the §7/#18 fix).
+    """
     catch_dir = catch.dir(workspace)
-    per_stem: dict[str, list[LoopCandidate]] = {}
+    loaded: dict[str, tuple] = {}
     for st in catch.stems:
         p = catch_dir / st.path
-        if not p.exists():
-            continue
+        if p.exists():
+            loaded[st.name] = (G._mono(G.load_audio(p)[0]).astype(np.float32), _sr_of(p))
+
+    # shared grid from the rhythmic source (drums → bass → first available)
+    grid = None
+    for pick in ("drums", "bass", *loaded):
+        if pick in loaded:
+            if progress:
+                progress(f"beat grid from {pick}")
+            grid = beat_grid(*loaded[pick])
+            break
+
+    per_stem: dict[str, list[LoopCandidate]] = {}
+    for name, (y, sr) in loaded.items():
         if progress:
-            progress(f"analyzing {st.name}")
-        y = G._mono(G.load_audio(p)[0]).astype(np.float32)
-        per_stem[st.name] = analyze_stem(y, _sr_of(p), bars=bars, top_n=top_n)
+            progress(f"analyzing {name}")
+        per_stem[name] = analyze_stem(y, sr, bars=bars, top_n=top_n, grid=grid)
 
     # cluster candidates from all stems by start time (within ~1 bar slack)
     flat = [(name, c) for name, cs in per_stem.items() for c in cs]

tools/foundry/tests/test_loops.py

@@ -72,6 +72,21 @@ def test_snap_zc_moves_to_crossing():
     assert abs(snapped - idx) <= int(sr * L.ZC_TOL_MS / 1000)
 
 
+def test_analyze_stem_honors_shared_grid():
+    # a sparse stem (few onsets) gets no usable per-stem grid, but a shared grid
+    # from a rhythmic source lets it produce candidates (#18 grid-from-drums).
+    sr = 22050
+    sparse = (0.05 * np.random.default_rng(3).standard_normal(sr * 12)).astype(np.float32)
+    # build a 120-bpm grid (beats every 0.5 s) as (peaks_frames, times, bpm)
+    times = np.arange(0, 12, 0.5)
+    peaks = (times * sr / 512).astype(int)
+    bpm = np.full(len(times), 120.0)
+    cands = L.analyze_stem(sparse, sr, bars=(2,), top_n=4, grid=(peaks, times, bpm))
+    assert isinstance(cands, list)            # runs on the shared grid without recomputing
+    for c in cands:
+        assert 100 <= c.bpm <= 140            # uses the supplied grid's tempo
+
+
 def test_beat_grid_recovers_tempo_not_its_octave():
     # 120 bpm click train; the fix must report ~120, NOT ~240
     sr = 22050