Commit c5963f63 by PLN (Algolia)

feat(foundry): lyric-aware vocal sampler (engine/vox.py) — phrase-level,…

feat(foundry): lyric-aware vocal sampler (engine/vox.py) — phrase-level, hook-ranked, loop-capable vox one-shots

PROBLEM. The Foundry's vocal path (loops.analyze_chops) cuts vocals blind to what
is sung — onset→onset slices scored on seam/zc mechanics. But a vocal one-shot's
value to a livecoder IS the lyric. The freshly-shipped superfreak dub kit had four
tiny 0.35-0.65s onset chops (03-06) with no idea they sat on top of some of the most
iconic lyrics in funk ("she's a super freak", "kinky girl", "the kind you read about").

APPROACH. engine/vox.py: transcribe → phrase-segment → iconicity-rank → cut → grade →
lyric-name, DRY-reusing grade.py sub-scorers and naming.py convention.
 - transcribe: whisper word-level timestamps, shelled out like separate.py→demucs,
   cached at workspace/vox_transcript.json so re-runs are instant. Model escalates
   small→medium empirically when the known hooks don't surface (small garbled the
   patois delivery; medium recovers the real lyrics).
 - segment: group words on inter-word gaps (>=0.45s), cap 6s, split over-long phrases
   at their widest interior gap; keep per-phrase avg ASR confidence.
 - iconicity: feature-engineered rank = repetition (normalized phrase text + content
   n-grams recurring across the track — hooks repeat) + hook keywords (title-derived
   or --hooks) + clarity (ASR prob) + energy (RMS vs stem median) + duration sweet-spot.
 - cut: pre-pad + post-pad (clamped to next phrase), zero-crossing snap both edges,
   3-10ms fades so edge samples are ~0 regardless of where the snap landed; as-cut level.
 - kit-level loopability: report each phrase's duration in BEATS at the kit BPM; flag
   loop-capable phrases (within ±3% of a 1/2/4-bar multiple) and emit a bar-quantized
   _loop variant (tail pulled to the exact beat grid, still zc-snapped) for the best
   hooks, so a vocal can loopAt alongside instrument loops.
 - name: NN_vox_<lyricslug> (naming.lint-clean), dedup identical texts (best-graded
   instance, up to 2 takes of THE hook).

VALIDATION. 17 new mocked-transcription tests (whisper never runs in tests): gap
segmentation, over-long split, iconicity repetition-beats-oneoff + hook-keyword boost,
beat/loop annotation, cut edges ~0 after zc+fade, next-start clamp, exact-beat loop
variant, lyric-slug + lint contract. Full suite 60->77 green. Applied to the superfreak
stem: medium transcript contains every iconic line; shipped 6 curated vox files (S/B
tier) replacing the 4 blind chops, incl. one 2-bar (8.00-beat) loop variant. All
re-graded on disk: no clip (peaks <0.8), no DC (<2e-4), edges exactly 0.0.
parent 791a0d78
"""vox — the lyric-aware vocal sampler (the toaster).
The bar-aligned finder (loops.analyze_stem) and the onset chopper
(loops.analyze_chops) both cut vocals *blind to what is sung* — they slice on
transients and score on seam/zc mechanics. But a vocal one-shot's VALUE to a
livecoder is the LYRIC: "she's a super freak", "kinky girl", a dub ad-lib. This
module cuts vocals into **phrase-level, lyric-named** one-shots.
The pipeline (design fixed; DSP reuses grade.py, naming from naming.py):
1. transcribe — whisper word-level timestamps on the (clean, demucs) vocal
stem. Heavy dep, so shelled out to the `whisper` CLI like separate.py shells
to demucs; the transcription JSON is CACHED in the workspace so re-runs are
instant. Model size escalates empirically (small → medium) when the known
hooks don't surface — dub/patois vocals defeat the small model.
2. segment — group words into phrases on inter-word gaps (≥ GAP_S), capped
at MAX_PHRASE_S; keep per-phrase avg word confidence.
3. rank — feature-engineered ICONICITY: repetition (a hook repeats),
hook-keyword match (from --hooks or the source title), ASR clarity, segment
energy vs stem median, and a duration sweet-spot. Hooks float to the top.
4. cut — small pre-pad, post-pad to the gap, zero-crossing snap at both
edges, short fades → clickless one-shots at as-cut level (no normalize).
5. grade — each cut graded role='one_shot' (grade.py); drop < C tier.
6. name — NN_vox_<lyricslug> (naming.lint-clean); dedup identical texts,
keep the best-graded instance (up to 2 takes of THE hook).
python3 -m engine.vox <vocals.wav> [--workspace DIR] [--hooks super freak] \
[--top N] [--json]
"""
from __future__ import annotations
import json
import re
import subprocess
from pathlib import Path
from typing import Callable, Optional
import numpy as np
import soundfile as sf
from pydantic import BaseModel, Field
from . import grade as G
from . import naming
# ── tunables ────────────────────────────────────────────────────────────────
GAP_S = 0.45 # inter-word gap that starts a new phrase (tuned on real dub)
MAX_PHRASE_S = 6.0 # cap a phrase; a long line splits at the widest interior gap
PRE_PAD_S = 0.040 # lead-in before the first word onset
POST_PAD_S = 0.120 # tail after the last word (or up to the next-word gap)
FADE_MS = 6.0 # in/out fade to guarantee clickless edges (3–10 ms band)
ZC_TOL_MS = 12.0 # zero-crossing snap tolerance at each edge
EXPORT_SR = 44100 # B8: SuperDirt rate
DUR_SWEET = (0.6, 4.0) # iconicity duration sweet-spot (s)
MIN_TIER = "C" # drop cuts below this grade tier
# loopability (kit-level): the Super freak dub is 132.5 BPM. A vocal one-shot that
# happens to span ~an integer beat / 1-2-4-bar multiple can be loopAt'd alongside the
# instrument loops. beats reported for every phrase; the best such hooks also ship a
# bar-quantized _loop variant (edges pulled to the exact grid, still zc-snapped).
DEFAULT_BPM = 132.5
LOOP_TOL = 0.03 # within ±3% of an integer-beat / bar multiple ⇒ loop-capable
# iconicity feature weights (provisional; tuned on the superfreak corpus)
ICON_W = {"repeat": 0.40, "hook": 0.25, "clarity": 0.15, "energy": 0.10, "dur": 0.10}
_TIER_ORDER = {"S": 5, "A": 4, "B": 3, "C": 2, "D": 1}
# stopwords excluded from n-gram/keyword hook matching (function words don't hook)
_STOP = {
"a", "an", "the", "and", "or", "but", "is", "are", "was", "were", "be", "been",
"to", "of", "in", "on", "at", "it", "its", "i", "you", "he", "she", "her", "his",
"we", "they", "them", "my", "me", "your", "that", "this", "with", "for", "so",
"do", "don", "t", "s", "m", "re", "ll", "ve", "oh", "yeah", "na",
}
# ── data model ────────────────────────────────────────────────────────────────
class Word(BaseModel):
text: str
start: float
end: float
prob: float = 1.0
class Phrase(BaseModel):
"""A group of contiguous words = one candidate vocal one-shot."""
text: str
start: float
end: float
conf: float # avg ASR word probability
words: list[Word] = Field(default_factory=list)
# populated by rank()/cut()/grade
iconicity: float = 0.0
repeat: float = 0.0
hook: float = 0.0
energy: float = 0.0
grade: float = 0.0
tier: str = "?"
name: str = ""
# loopability (kit-level, filled by annotate_beats)
beats: float = 0.0 # duration in beats at the kit BPM
loop_capable: bool = False # lands within LOOP_TOL of an integer-beat/bar multiple
loop_beats: int = 0 # nearest musical beat count it snaps to (0 = none)
@property
def dur(self) -> float:
return self.end - self.start
# ── 1. transcribe (whisper shell-out, cached) ──────────────────────────────────
def transcribe(vocals: Path, workspace: Optional[Path] = None, *, model: str = "medium",
language: str = "en", device: Optional[str] = None,
progress: Optional[Callable[[str], None]] = None,
force: bool = False) -> dict:
"""Word-level whisper transcription of `vocals`, cached at workspace/vox_transcript.json.
Shells out to the `whisper` CLI (heavy-dep convention, cf. separate.py→demucs) with
--word_timestamps True --output_format json. The cache makes re-runs instant; pass
force=True to re-transcribe (e.g. after bumping the model). Returns the raw whisper
result dict (segments[].words[] carry start/end/probability).
"""
ws = Path(workspace) if workspace else Path(vocals).parent
cache = ws / "vox_transcript.json"
if cache.exists() and not force:
data = json.loads(cache.read_text())
if data.get("_vox_model") == model: # honour a model change
return data
ws.mkdir(parents=True, exist_ok=True)
cmd = ["whisper", str(vocals), "--model", model, "--language", language,
"--word_timestamps", "True", "--output_format", "json",
"--output_dir", str(ws)]
if device:
cmd += ["--device", device]
if progress:
progress(f"whisper {model} transcribing (cpu, ~2min for medium)…")
subprocess.run(cmd, check=True, capture_output=True, text=True)
# whisper writes <stem>.json into output_dir
out = ws / f"{Path(vocals).stem}.json"
data = json.loads(out.read_text())
data["_vox_model"] = model
cache.write_text(json.dumps(data, indent=2))
if out.resolve() != cache.resolve():
out.unlink(missing_ok=True)
return data
def words_from_transcript(data: dict) -> list[Word]:
"""Flatten whisper segments → a clean word list (start/end/prob), sorted by time."""
words: list[Word] = []
for seg in data.get("segments", []):
for w in seg.get("words", []):
txt = str(w.get("word", "")).strip()
if not txt:
continue
words.append(Word(text=txt, start=float(w["start"]), end=float(w["end"]),
prob=float(w.get("probability", 1.0))))
words.sort(key=lambda x: x.start)
return words
# ── 2. phrase segmentation ──────────────────────────────────────────────────────
def _clean_text(s: str) -> str:
return re.sub(r"\s+", " ", re.sub(r"[^A-Za-z0-9' ]+", " ", s)).strip()
def segment_phrases(words: list[Word], *, gap_s: float = GAP_S,
max_phrase_s: float = MAX_PHRASE_S) -> list[Phrase]:
"""Group words into phrases: a gap ≥ gap_s between consecutive words starts a new
phrase; a phrase exceeding max_phrase_s is split at its widest interior gap.
"""
if not words:
return []
groups: list[list[Word]] = [[words[0]]]
for prev, w in zip(words, words[1:]):
if w.start - prev.end >= gap_s:
groups.append([w])
else:
groups[-1].append(w)
# split over-long phrases at the widest interior gap (recursively)
def _split(g: list[Word]) -> list[list[Word]]:
if len(g) < 2 or (g[-1].end - g[0].start) <= max_phrase_s:
return [g]
gaps = [(g[i + 1].start - g[i].end, i) for i in range(len(g) - 1)]
_, k = max(gaps) # widest interior gap
return _split(g[:k + 1]) + _split(g[k + 1:])
phrases: list[Phrase] = []
for g in groups:
for sub in _split(g):
txt = _clean_text(" ".join(w.text for w in sub))
if not txt:
continue
conf = float(np.mean([w.prob for w in sub]))
phrases.append(Phrase(text=txt, start=sub[0].start, end=sub[-1].end,
conf=round(conf, 4), words=sub))
return phrases
# ── 3. iconicity ranking ────────────────────────────────────────────────────────
def _tokens(text: str) -> list[str]:
return [t for t in re.split(r"\s+", text.lower().strip()) if t]
def _content_tokens(text: str) -> list[str]:
out = []
for t in _tokens(text):
base = t.split("'")[0] # she's→she, don't→don, walkin'→walkin
if base in _STOP or t in _STOP or len(base) <= 1:
continue
out.append(base if base else t)
return out
def _ngrams(tokens: list[str], n: int) -> list[str]:
return [" ".join(tokens[i:i + n]) for i in range(len(tokens) - n + 1)]
def hooks_from_title(title: str) -> list[str]:
"""Default hook keywords = significant (content) words from the source title.
Drops the boilerplate around a YouTube title (artist names, 'meets', 'ft.',
decorative stars) heuristically and keeps content words as hook seeds.
"""
t = re.sub(r"[★☆|•\-–—]+", " ", title.lower())
t = re.sub(r"\b(meets|feat|featuring|ft|remix|cover|official|video|audio|prod)\b", " ", t)
return sorted(set(_content_tokens(t)))
def rank_phrases(phrases: list[Phrase], *, hooks: Optional[list[str]] = None,
energies: Optional[dict[int, float]] = None) -> list[Phrase]:
"""Score each phrase's ICONICITY in place and return sorted (desc).
Features (each 0..1, combined by ICON_W):
• repeat — how often this phrase's normalized text / content n-grams recur
across the track. Hooks repeat; a one-off ad-lib does not.
• hook — fraction of hook keywords present (title-derived or --hooks).
• clarity — avg ASR word confidence.
• energy — segment RMS vs stem median (loud, committed delivery), if provided.
• dur — duration sweet-spot (DUR_SWEET); too-short=click, too-long≠one-shot.
"""
hook_set = {h.lower() for h in (hooks or [])}
# repetition corpus: count normalized phrase texts and content bigrams/trigrams
norm = [" ".join(_content_tokens(p.text)) for p in phrases]
text_ct: dict[str, int] = {}
for s in norm:
if s:
text_ct[s] = text_ct.get(s, 0) + 1
gram_ct: dict[str, int] = {}
for p in phrases:
toks = _content_tokens(p.text)
for n in (2, 3):
for g in set(_ngrams(toks, n)):
gram_ct[g] = gram_ct.get(g, 0) + 1
max_text = max(text_ct.values(), default=1)
max_gram = max(gram_ct.values(), default=1)
for i, p in enumerate(phrases):
toks = _content_tokens(p.text)
n = norm[i]
# repetition: blend exact-phrase recurrence with best recurring n-gram
rep_text = (text_ct.get(n, 1) - 1) / max(max_text - 1, 1) if n else 0.0
best_gram = 0
for gn in (2, 3):
for g in set(_ngrams(toks, gn)):
best_gram = max(best_gram, gram_ct.get(g, 0))
rep_gram = (best_gram - 1) / max(max_gram - 1, 1)
repeat = max(rep_text, rep_gram)
# hook keywords present in the phrase
hook = 0.0
if hook_set:
present = sum(1 for h in hook_set if h in set(toks))
hook = min(1.0, present / max(1, min(len(hook_set), 2)))
clarity = _clamp01(p.conf)
energy = _clamp01(energies.get(i, 0.0)) if energies else 0.0
lo, hi = DUR_SWEET
if lo <= p.dur <= hi:
dur = 1.0
elif p.dur < lo:
dur = _clamp01(p.dur / lo)
else:
dur = _clamp01(1.0 - (p.dur - hi) / hi)
p.repeat = round(repeat, 4)
p.hook = round(hook, 4)
p.energy = round(energy, 4)
w = ICON_W
p.iconicity = round(w["repeat"] * repeat + w["hook"] * hook +
w["clarity"] * clarity + w["energy"] * energy +
w["dur"] * dur, 4)
return sorted(phrases, key=lambda x: -x.iconicity)
def _clamp01(x: float) -> float:
return 0.0 if x < 0 else 1.0 if x > 1 else float(x)
def segment_energies(y_mono: np.ndarray, sr: int, phrases: list[Phrase]) -> dict[int, float]:
"""Per-phrase RMS normalized against the stem's active median → 0..1 energy feature."""
if y_mono.size == 0:
return {}
win = int(0.05 * sr)
frames = np.array([np.sqrt(np.mean(y_mono[i:i + win] ** 2) + 1e-12)
for i in range(0, max(1, y_mono.size - win), win)])
active = frames[frames > np.percentile(frames, 40)] # ignore silence
med = float(np.median(active)) if active.size else 1e-6
out: dict[int, float] = {}
for i, p in enumerate(phrases):
a, b = int(p.start * sr), int(p.end * sr)
seg = y_mono[a:b]
if seg.size < 2:
out[i] = 0.0
continue
rms = float(np.sqrt(np.mean(seg ** 2) + 1e-12))
out[i] = _clamp01((rms / (med + 1e-9)) / 2.0) # med ⇒ 0.5, 2×med ⇒ 1.0
return out
# ── 4. cut (pre/post-pad, zc-snap, fade) ────────────────────────────────────────
def _snap_zc(mono: np.ndarray, idx: int, sr: int) -> int:
"""Snap a boundary to the nearest zero crossing within ZC_TOL_MS (loops._snap_zc rule)."""
tol = int(sr * ZC_TOL_MS / 1000)
lo, hi = max(0, idx - tol), min(len(mono) - 1, idx + tol)
seg = mono[lo:hi]
if seg.size < 2:
return int(min(max(idx, 0), len(mono) - 1))
zc = np.where(np.diff(np.signbit(seg)))[0]
return int(lo + zc[np.argmin(np.abs(zc - (idx - lo)))]) if zc.size else \
int(min(max(idx, 0), len(mono) - 1))
def cut_phrase(y: np.ndarray, sr: int, phrase: Phrase, *, next_start: Optional[float] = None,
pre_pad_s: float = PRE_PAD_S, post_pad_s: float = POST_PAD_S,
fade_ms: float = FADE_MS) -> np.ndarray:
"""Cut a clickless one-shot for `phrase` from stem `y` (n,ch) or (n,).
Small pre-pad before the first word; post-pad after the last word but never into
the next phrase (clamped to next_start if given). Both edges zero-crossing snapped
(on the mono reference), then short in/out fades applied so edge samples ≈ 0 —
guarantees no click regardless of where the snap landed. Level is kept as-cut.
"""
y2 = y if y.ndim == 2 else y[:, None]
mono = y2.mean(axis=1)
n = mono.size
a = int((phrase.start - pre_pad_s) * sr)
post_cap = phrase.end + post_pad_s
if next_start is not None:
post_cap = min(post_cap, next_start - 0.010)
b = int(post_cap * sr)
a = max(0, min(a, n - 2))
b = max(a + 2, min(b, n))
a = _snap_zc(mono, a, sr)
b = _snap_zc(mono, b, sr)
if b <= a:
b = min(n, a + 2)
clip = y2[a:b].astype(np.float32).copy()
f = min(int(fade_ms / 1000 * sr), clip.shape[0] // 2)
if f > 1:
ramp = np.linspace(0.0, 1.0, f, dtype=np.float32)[:, None]
clip[:f] *= ramp
clip[-f:] *= ramp[::-1]
return clip
# ── loopability (kit-level: beats + bar-quantized variant) ──────────────────────
# musical beat counts worth loopAt'ing a vocal to: a beat, a 2-beat, a bar (4),
# 2 bars (8), 4 bars (16). We favour these over arbitrary integer beats.
_LOOP_BEATS = (1, 2, 3, 4, 6, 8, 16)
def annotate_beats(phrases: list[Phrase], *, bpm: float = DEFAULT_BPM,
tol: float = LOOP_TOL) -> None:
"""Fill each phrase's beats / loop_capable / loop_beats (in place).
A phrase is loop-capable if its duration is within ±tol of one of _LOOP_BEATS
at `bpm` — i.e. it can be trimmed/extended a hair onto the grid and loopAt'd.
"""
beat_s = 60.0 / bpm
for p in phrases:
p.beats = round(p.dur / beat_s, 3)
best = min(_LOOP_BEATS, key=lambda n: abs(p.beats - n))
rel = abs(p.beats - best) / best
p.loop_capable = rel <= tol
p.loop_beats = best if p.loop_capable else 0
def cut_phrase_loop(y: np.ndarray, sr: int, phrase: Phrase, *, bpm: float = DEFAULT_BPM,
fade_ms: float = FADE_MS) -> Optional[np.ndarray]:
"""Bar-quantized loop variant of a loop-capable phrase: edges pulled to the EXACT
beat grid (phrase.loop_beats * beat_s), then zc-snapped within ±ZC_TOL_MS and faded.
Anchored at the phrase's first-word onset (minus the small pre-pad so the syllable
isn't clipped); the tail is placed at exactly loop_beats away so dur == an integer
beat multiple, giving a clean loopAt. Returns None if the phrase isn't loop-capable.
"""
if not phrase.loop_capable or phrase.loop_beats <= 0:
return None
y2 = y if y.ndim == 2 else y[:, None]
mono = y2.mean(axis=1)
n = mono.size
beat_s = 60.0 / bpm
target = phrase.loop_beats * beat_s
a = int((phrase.start - PRE_PAD_S) * sr)
a = max(0, min(a, n - 2))
a = _snap_zc(mono, a, sr)
b = a + int(round(target * sr)) # EXACT grid length from the anchor
b = _snap_zc(mono, b, sr) # ±ZC_TOL_MS only — stays on grid
b = max(a + 2, min(b, n))
clip = y2[a:b].astype(np.float32).copy()
f = min(int(fade_ms / 1000 * sr), clip.shape[0] // 2)
if f > 1:
ramp = np.linspace(0.0, 1.0, f, dtype=np.float32)[:, None]
clip[:f] *= ramp
clip[-f:] *= ramp[::-1]
return clip
# ── 5+6. grade + name ────────────────────────────────────────────────────────────
def lyric_slug(text: str, *, max_tokens: int = 3, taken: Optional[set] = None) -> str:
"""`NN_vox_<lyricslug>` body slug from a phrase's most salient content words.
Keeps up to max_tokens content words (falls back to raw tokens if a phrase is all
stopwords, e.g. an ad-lib "oh yeah"). Returns just the <lyricslug> body — the caller
joins it via naming.suggest_name so the NN_ prefix + dedup stay in one place.
"""
toks = _content_tokens(text) or _tokens(text)
toks = toks[:max_tokens]
slug = naming.slug_token("".join(toks)) or "vox"
return slug
def grade_cut(clip: np.ndarray, sr: int, path: str = "<vox>") -> G.LoopGrade:
"""Grade a cut as a one-shot (grade.py, role='one_shot' weighting)."""
return G.grade_array(clip.T if clip.ndim == 2 else clip, sr,
path=path, role="vocals")
# ── orchestration ─────────────────────────────────────────────────────────────────
def analyze_vox(vocals: Path, *, workspace: Optional[Path] = None,
hooks: Optional[list[str]] = None, title: Optional[str] = None,
model: str = "medium", top_n: int = 12, bpm: float = DEFAULT_BPM,
progress: Optional[Callable[[str], None]] = None,
transcript: Optional[dict] = None) -> list[Phrase]:
"""Full read-only pipeline: transcribe → segment → energy → rank. No files written.
`transcript` lets a caller (or a test) inject a whisper-shaped dict and skip the
shell-out entirely. `hooks` overrides the default title-derived hook keywords.
"""
if transcript is None:
transcript = transcribe(vocals, workspace, model=model, progress=progress)
words = words_from_transcript(transcript)
phrases = segment_phrases(words)
y, sr = G.load_audio(vocals)
mono = G._mono(y)
energies = segment_energies(mono, sr, phrases)
if hooks is None:
hooks = hooks_from_title(title) if title else \
hooks_from_title(transcript.get("_vox_title", "")) or None
ranked = rank_phrases(phrases, hooks=hooks, energies=energies)
annotate_beats(ranked, bpm=bpm)
return ranked[:top_n] if top_n else ranked
def grade_and_tier(phrases: list[Phrase], vocals: Path, *, bpm: float = DEFAULT_BPM) -> None:
"""Cut each phrase as a one-shot, grade it, and fill grade/tier in place."""
y, sr = G.load_audio(vocals)
ych = y.T if y.ndim == 2 else y
starts = sorted(p.start for p in phrases)
for p in phrases:
nxt = next((s for s in starts if s > p.end), None)
clip = cut_phrase(ych, sr, p, next_start=nxt)
g = grade_cut(clip, sr)
p.grade, p.tier = g.grade, g.tier
def _tier_ge(tier: str, floor: str) -> bool:
return _TIER_ORDER.get(tier, 0) >= _TIER_ORDER.get(floor, 0)
def select_ship(phrases: list[Phrase], *, max_files: int = 9, min_tier: str = MIN_TIER,
max_loops: int = 2) -> tuple[list[Phrase], list[Phrase]]:
"""Curate what ships: dedup identical texts (keep best-graded), gate on tier, then
pick the top-iconicity phrases as one-shots and up to `max_loops` loop-capable hooks.
Returns (one_shots, loops). Total files = len(one_shots)+len(loops) ≤ max_files.
`phrases` must already be graded+ranked (iconicity desc). THE hook may keep up to 2
takes; every other text dedups to its single best-graded instance.
"""
# dedup by normalized text → keep the best-graded (then most iconic) instance,
# allowing up to 2 takes of the single most-iconic text (THE hook).
hook_norm = " ".join(_content_tokens(phrases[0].text)) if phrases else ""
by_text: dict[str, list[Phrase]] = {}
for p in phrases:
by_text.setdefault(" ".join(_content_tokens(p.text)) or p.text.lower(), []).append(p)
kept: list[Phrase] = []
for norm, group in by_text.items():
group = sorted(group, key=lambda x: (-x.grade, -x.iconicity))
n_keep = 2 if norm == hook_norm else 1
kept.extend(group[:n_keep])
kept = [p for p in kept if _tier_ge(p.tier, min_tier)]
kept.sort(key=lambda x: -x.iconicity)
# loop variants: best loop-capable hooks (≥ B tier so a loopAt is clean)
loops = [p for p in kept if p.loop_capable and _tier_ge(p.tier, "B")][:max_loops]
loop_ids = {id(p) for p in loops}
n_oneshot = max_files - len(loops)
one_shots = [p for p in kept][:n_oneshot]
# ensure any phrase chosen as a loop is also present as a one-shot anchor if room,
# but never exceed max_files
return one_shots, loops
def ship_vox(vocals: Path, kit: str, *, one_shots: list[Phrase], loops: list[Phrase],
start_index: int = 3, bpm: float = DEFAULT_BPM,
samples_root: Optional[Path] = None,
replace_indices: Optional[list[int]] = None) -> list[tuple[Path, Phrase, bool]]:
"""Write the curated vox one-shots (+ loop variants) into Samples/<kit>/, numbered
from start_index. Deletes `replace_indices` old files first. Returns
[(path, phrase, is_loop)]. Re-grades on disk are the caller's job (verify).
"""
from . import publish
root = samples_root or publish.samples_root()
out_dir = root / kit
out_dir.mkdir(parents=True, exist_ok=True)
# remove superseded files (e.g. old 03..06 chops)
if replace_indices:
for f in list(out_dir.glob("*.wav")):
m = re.match(r"^(\d{2})_", f.name)
if m and int(m.group(1)) in replace_indices:
f.unlink()
y, sr = G.load_audio(vocals)
ych = y.T if y.ndim == 2 else y
starts = sorted(p.start for p in [*one_shots, *loops])
taken: set[str] = set()
written: list[tuple[Path, Phrase, bool]] = []
idx = start_index
def _write(clip: np.ndarray, base_slug: str, phrase: Phrase, is_loop: bool):
nonlocal idx
# loop variant → NN_vox_<slug>_loop (slug as section keeps `loop` a distinct token,
# since suggest_name slugs each part separately; a single character would collapse it)
if is_loop:
nm = naming.suggest_name(idx, "vox", section=base_slug, character="loop", taken=taken)
else:
nm = naming.suggest_name(idx, "vox", character=base_slug, taken=taken)
taken.add(nm)
dest = out_dir / f"{nm}.wav"
sf.write(str(dest), clip, EXPORT_SR if sr == EXPORT_SR else sr, subtype="PCM_24")
written.append((dest, phrase, is_loop))
phrase.name = nm
idx += 1
for p in one_shots:
nxt = next((s for s in starts if s > p.end), None)
clip = cut_phrase(ych, sr, p, next_start=nxt)
_write(clip, lyric_slug(p.text), p, False)
for p in loops:
clip = cut_phrase_loop(ych, sr, p, bpm=bpm)
if clip is None:
continue
_write(clip, lyric_slug(p.text), p, True)
if written:
try:
publish.link_kit(kit)
except Exception:
pass
return written
def render_auditions(shipped: list[tuple[Path, Phrase, bool]], audition_dir: Path,
*, gap_s: float = 0.3, bpm: float = DEFAULT_BPM) -> Path:
"""Copy each shipped wav into audition_dir and build vox_medley.wav (all phrases
sequenced with `gap_s` silence between). Returns the medley path.
"""
import shutil
audition_dir.mkdir(parents=True, exist_ok=True)
pieces: list[np.ndarray] = []
sr_ref = EXPORT_SR
for path, _p, _loop in shipped:
shutil.copy2(path, audition_dir / path.name)
clip, sr_ref = sf.read(str(path), always_2d=True, dtype="float32")
pieces.append(clip)
pieces.append(np.zeros((int(gap_s * sr_ref), clip.shape[1]), dtype=np.float32))
medley_path = audition_dir / "vox_medley.wav"
if pieces:
medley = np.concatenate(pieces, axis=0)
sf.write(str(medley_path), medley, sr_ref, subtype="PCM_24")
return medley_path
def _print_table(phrases: list[Phrase]) -> None:
print(f" {'#':>2} {'t0':>6} {'dur':>5} {'beats':>5} {'loop':>5} {'icon':>5} "
f"{'rep':>4} {'hook':>4} {'tier':>4} text")
for i, p in enumerate(phrases):
loop = f"{p.loop_beats}b" if p.loop_capable else "—"
print(f" {i:>2} {p.start:6.1f} {p.dur:5.2f} {p.beats:5.2f} {loop:>5} "
f"{p.iconicity:5.3f} {p.repeat:4.2f} {p.hook:4.2f} {p.tier:>4} {p.text[:56]}")
# ── CLI ─────────────────────────────────────────────────────────────────────────
def _main(argv=None) -> int:
import argparse
ap = argparse.ArgumentParser(description="lyric-aware vocal sampler (transcribe→rank)")
ap.add_argument("vocals")
ap.add_argument("--workspace", default=None)
ap.add_argument("--hooks", nargs="+", default=None)
ap.add_argument("--title", default=None, help="source title for default hook seeds")
ap.add_argument("--model", default="medium")
ap.add_argument("--top", type=int, default=12)
ap.add_argument("--json", action="store_true")
a = ap.parse_args(argv)
ws = Path(a.workspace) if a.workspace else None
ranked = analyze_vox(Path(a.vocals), workspace=ws, hooks=a.hooks, title=a.title,
model=a.model, top_n=a.top,
progress=lambda m: print(f"… {m}", flush=True))
grade_and_tier(ranked, Path(a.vocals)) # cut+grade so the table carries a tier
if a.json:
print(json.dumps([p.model_dump(exclude={"words"}) for p in ranked], indent=2))
else:
_print_table(ranked)
return 0
if __name__ == "__main__":
raise SystemExit(_main())
"""Tests for the lyric-aware vocal sampler (engine/vox.py).
Transcription is ALWAYS mocked — whisper never runs in tests. We inject a
whisper-shaped transcript dict (segments[].words[]) and exercise the pure
pipeline: phrase grouping on gaps, iconicity (repetition + hook keyword),
cut edge properties (zc + fade ⇒ edges ≈ 0), beat/loop annotation, and the
lyric-slug naming / lint contract.
"""
import sys
from pathlib import Path
import numpy as np
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from engine import vox as V
from engine import naming as N
# ── helpers ──────────────────────────────────────────────────────────────────
def _w(text, start, end, prob=0.9):
return {"word": text, "start": start, "end": end, "probability": prob}
def _transcript(words):
"""Wrap a flat word list as a single whisper segment dict."""
return {"segments": [{"words": words}]}
# ── phrase segmentation ────────────────────────────────────────────────────────
def test_segment_splits_on_gap():
# two words tight, then a 0.6s gap, then two more → two phrases
words = V.words_from_transcript(_transcript([
_w(" she's", 0.0, 0.3), _w(" super", 0.3, 0.6),
_w(" freak", 1.5, 1.8), _w(" now", 1.8, 2.0),
]))
phrases = V.segment_phrases(words, gap_s=0.45)
assert len(phrases) == 2
assert phrases[0].text == "she's super"
assert phrases[1].text == "freak now"
def test_segment_no_split_within_gap():
words = V.words_from_transcript(_transcript([
_w(" a", 0.0, 0.2), _w(" b", 0.3, 0.5), _w(" c", 0.6, 0.8),
]))
phrases = V.segment_phrases(words, gap_s=0.45)
assert len(phrases) == 1
assert phrases[0].text == "a b c"
def test_segment_splits_overlong_phrase_at_widest_gap():
# one contiguous run (all gaps < 0.45) but > max_phrase_s → split at widest gap
words = V.words_from_transcript(_transcript([
_w(" one", 0.0, 1.0), _w(" two", 1.2, 2.0),
_w(" three", 2.4, 3.0), _w(" four", 3.1, 4.0),
]))
phrases = V.segment_phrases(words, gap_s=0.45, max_phrase_s=3.0)
assert len(phrases) == 2 # split, widest interior gap is 2.0→2.4
assert phrases[0].text == "one two"
assert phrases[1].text == "three four"
def test_phrase_conf_is_mean_word_prob():
words = V.words_from_transcript(_transcript([
_w(" x", 0.0, 0.2, prob=0.8), _w(" y", 0.2, 0.4, prob=1.0),
]))
p = V.segment_phrases(words)[0]
assert abs(p.conf - 0.9) < 1e-6
# ── iconicity ────────────────────────────────────────────────────────────────
def _phrase(text, start=0.0, dur=1.0, conf=0.9):
return V.Phrase(text=text, start=start, end=start + dur, conf=conf)
def test_iconicity_repetition_beats_oneoff():
# "super freak" recurs 3×; "random mumble" once → repeated phrase ranks higher
phrases = [
_phrase("super freak", 0), _phrase("super freak", 5),
_phrase("super freak", 10), _phrase("random mumble here", 15),
]
ranked = V.rank_phrases(phrases, hooks=None)
assert ranked[0].text == "super freak"
hook = next(p for p in ranked if p.text == "super freak")
oneoff = next(p for p in ranked if p.text == "random mumble here")
assert hook.repeat > oneoff.repeat
assert hook.iconicity > oneoff.iconicity
def test_iconicity_hook_keyword_boost():
a = _phrase("super freak", 0)
b = _phrase("walking down street", 5)
V.rank_phrases([a, b], hooks=["super", "freak"])
assert a.hook > b.hook
assert a.iconicity > b.iconicity
def test_hooks_from_title_drops_boilerplate():
hooks = V.hooks_from_title("Soul Sugar meets Dub Shepherds ft. Jolly Joseph ★ Super freak ★")
assert "super" in hooks and "freak" in hooks
assert "meets" not in hooks and "ft" not in hooks
# ── beat / loop annotation ─────────────────────────────────────────────────────
def test_annotate_beats_flags_loop_capable():
beat_s = 60.0 / 132.5
on_grid = _phrase("kinky girl", 0.0, dur=4 * beat_s) # exactly a bar
off_grid = _phrase("home to mother", 0.0, dur=2.5 * beat_s) # 2.5 beats, off
V.annotate_beats([on_grid, off_grid], bpm=132.5)
assert on_grid.loop_capable and on_grid.loop_beats == 4
assert abs(on_grid.beats - 4.0) < 0.05
assert not off_grid.loop_capable
# ── cut edge properties ─────────────────────────────────────────────────────────
def _tone(sr, dur, freq=220.0, amp=0.5):
t = np.arange(int(sr * dur)) / sr
return (amp * np.sin(2 * np.pi * freq * t)).astype(np.float32)
def test_cut_edges_near_zero_after_fade():
sr = 44100
y = _tone(sr, 3.0) # continuous tone: no natural zero at cut point
p = _phrase("test phrase", start=1.0, dur=0.6)
clip = V.cut_phrase(y, sr, p)
assert clip.shape[0] > 100
# fade guarantees the very edges are ~0 regardless of zc landing
assert abs(float(clip[0, 0])) < 1e-3
assert abs(float(clip[-1, 0])) < 1e-3
def test_cut_respects_next_start():
sr = 44100
y = _tone(sr, 4.0)
p = _phrase("first", start=1.0, dur=0.5) # ends 1.5; post-pad would reach ~1.62
clip = V.cut_phrase(y, sr, p, next_start=1.55)
end_t = 1.0 - V.PRE_PAD_S + clip.shape[0] / sr
assert end_t <= 1.55 + 0.02 # never runs into the next phrase
def test_cut_phrase_loop_is_exact_beat_multiple():
sr, bpm = 44100, 132.5
beat_s = 60.0 / bpm
y = _tone(sr, 6.0)
p = _phrase("super freak", start=1.0, dur=4 * beat_s)
V.annotate_beats([p], bpm=bpm)
assert p.loop_capable
clip = V.cut_phrase_loop(y, sr, p, bpm=bpm)
got_beats = (clip.shape[0] / sr) / beat_s
assert abs(got_beats - p.loop_beats) < 0.06 # within zc-snap slack of exact grid
def test_cut_phrase_loop_none_when_not_capable():
sr = 44100
y = _tone(sr, 3.0)
p = _phrase("off grid", start=0.5, dur=0.37)
V.annotate_beats([p], bpm=132.5)
if not p.loop_capable:
assert V.cut_phrase_loop(y, sr, p) is None
# ── lyric slug naming / lint ─────────────────────────────────────────────────────
def test_lyric_slug_from_content_words():
assert V.lyric_slug("she's a super freak") == "superfreak" # stopwords dropped
assert V.lyric_slug("the kind you don't take home to mother") == "kindtakehome"
def test_lyric_slug_falls_back_on_all_stopwords():
slug = V.lyric_slug("oh yeah")
assert slug and slug.isalnum()
def test_lyric_name_passes_lint():
slug = V.lyric_slug("she's a very kinky girl")
name = N.suggest_name(3, "vox", character=slug)
assert N.lint(name) == []
assert name.startswith("03_vox_")
def test_lyric_name_dedupes():
slug = V.lyric_slug("super freak")
taken = {N.suggest_name(3, "vox", character=slug)}
n2 = N.suggest_name(4, "vox", character=slug, taken=taken)
assert n2 not in taken
assert N.lint(n2) == []
def test_grade_cut_returns_one_shot_for_short():
sr = 44100
clip = _tone(sr, 0.5)[:, None] # 0.5s < one_shot threshold
g = V.grade_cut(clip, sr)
assert g.kind == "one_shot"
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment