Merge pull request 'align: keep lines Whisper missed when their singing went untranscribed' (#24) from unheard-lines into main

Reviewed-on: #24
Reviewed-by: branderson <brad@branderson.io>
This commit was merged in pull request #24.
This commit is contained in:
2026-10-10 22:08:59 -07:00
4 changed files with 239 additions and 26 deletions
+72 -9
View File
@@ -1,12 +1,24 @@
import json
from pathlib import Path
import numpy as np
import soundfile as sf
from ..lyrics import alignment_token, normalize_lyrics
from ..models import on_release
from ..pipeline import Ctx, PipelineError, Stage
from .transcribe import classify_lyrics_match, load_transcript, lyrics_similarity, vocal_sections
from ..whisper_client import TranscriptSegment
from .transcribe import (classify_lyrics_match, load_transcript, lyrics_similarity,
recovered_spans, vocal_sections)
SIMILARITY_THRESHOLD = 0.4
# MMS_FA's wildcard token: any audio the transcript does not cover
STAR = "*"
# Singing is audio within this much of the stem's loud level; separation
# bleed sits far below it.
SINGING_BELOW_LOUD_DB = 25.0
# Shorter stretches are breaths, ad-libs or Whisper's loose segment edges.
MIN_UNHEARD_SEC = 3.0
# a stage aligns section by section, so the model is cached across the calls
# within one run and dropped at the end of it
@@ -40,7 +52,7 @@ def force_align_section(vocals_path: Path, start: float, end: float,
waveform[:, lo:hi], sr, bundle.sample_rate)
# MMS_FA only models a-z and apostrophe; anything else becomes the star token.
vocab = bundle.get_dict()
tokens = ["".join(c if c in vocab else "*" for c in t) for t in tokens]
tokens = ["".join(c if c in vocab else STAR for c in t) for t in tokens]
with torch.inference_mode():
emission, _ = _get_model(device)(waveform.to(device))
spans = bundle.get_aligner()(emission[0], bundle.get_tokenizer()(tokens))
@@ -50,6 +62,39 @@ def force_align_section(vocals_path: Path, start: float, end: float,
sum(t.score for t in s) / len(s)) for s in spans]
def untranscribed_singing(vocals_path: Path, segments,
frame: float = 0.05) -> list[tuple[float, float]]:
"""Stretches of the vocal stem with singing in them but no transcript."""
data, sr = sf.read(str(vocals_path), dtype="float32", always_2d=True)
hop = int(sr * frame)
n = len(data) // hop
if n == 0:
return []
rms = np.sqrt((data[:n * hop].mean(1).reshape(n, hop) ** 2).mean(1))
db = 20 * np.log10(rms + 1e-10)
sung = db >= np.quantile(db, 0.9) - SINGING_BELOW_LOUD_DB
times = np.arange(n) * frame
for s in segments:
sung[(times >= min(s.start, s.end)) & (times < max(s.start, s.end))] = False
spans, start = [], None
for t, on in zip(times, sung):
if on and start is None:
start = float(t)
elif not on and start is not None:
spans.append((start, float(t)))
start = None
if start is not None:
spans.append((start, float(times[-1] + frame)))
# a breath or a quiet syllable must not split one unsung passage in two
merged: list[list[float]] = []
for a, b in spans:
if merged and a - merged[-1][1] < 1.0:
merged[-1][1] = b
else:
merged.append([a, b])
return [(a, b) for a, b in merged if b - a >= MIN_UNHEARD_SEC]
def _assign_lines_to_sections(line_indices: list[int], sections) -> list[list[int]]:
"""Split matched lyric line indices across vocal sections proportionally to section duration."""
total = sum(b - a for a, b in sections) or 1.0
@@ -92,9 +137,20 @@ def _enforce_monotonic(rows: list[dict]) -> None:
def _align_section(vocals: Path, sec: tuple[float, float], words: list[tuple[str, int]],
language: str | None, device: str) -> list[dict]:
tokens = [alignment_token(w, language) for w, _ in words]
alignable = [t for t in tokens if t]
aligned = iter(force_align_section(vocals, sec[0], sec[1], alignable, device)
if alignable else [])
# A star before, between and after lines lets speech, laughter or an
# ad-lib there go unexplained rather than claim the next line's letters.
seq, prev_line = [], None
for (_, line_index), token in zip(words, tokens):
if not token:
continue
if line_index != prev_line:
seq.append(STAR)
prev_line = line_index
seq.append(token)
if seq:
seq.append(STAR)
spans = force_align_section(vocals, sec[0], sec[1], seq, device) if seq else []
aligned = iter(span for span, token in zip(spans, seq) if token != STAR)
rows, blank = [], (None, None, 0.0)
for (word, line_index), token in zip(words, tokens):
start, end, score = next(aligned, blank) if token else blank
@@ -112,11 +168,18 @@ def _run(ctx: Ctx) -> None:
raise PipelineError(
f"lyrics don't match audio (similarity {sim:.2f}) — "
"wrong lyrics or wrong recording?")
# normalised so an inverted span can never collapse a section onto one instant
sections = [(min(a, b), max(a, b)) for a, b in vocal_sections(segments)]
dropped_lines, missing_spans = classify_lyrics_match(segments, ref)
matched = [i for i in range(len(ref.lines)) if i not in set(dropped_lines)]
vocals = ctx.work / "vocals.wav"
unheard = untranscribed_singing(vocals, segments)
dropped_lines, missing_spans = classify_lyrics_match(segments, ref,
unheard=unheard)
# lines Whisper missed are aligned against the singing it had no text for;
# only that singing, so an untranscribed ad-lib cannot draw lines into it
recovered = [TranscriptSegment(a, b, "")
for a, b in recovered_spans(segments, ref, unheard)]
# normalised so an inverted span can never collapse a section onto one instant
sections = [(min(a, b), max(a, b))
for a, b in vocal_sections(segments + recovered)]
matched = [i for i in range(len(ref.lines)) if i not in set(dropped_lines)]
rows = []
for sec, line_idxs in zip(sections, _assign_lines_to_sections(matched, sections)):
words = [(w, li) for li in line_idxs for w in ref.lines[li]]
+62 -16
View File
@@ -39,28 +39,74 @@ def lyrics_similarity(segments, ref: RefLyrics) -> float:
return matcher.ratio()
def classify_lyrics_match(segments, ref: RefLyrics, pad: float = 1.0
def _line_times(segments, matcher, seg_of, line_of) -> dict[int, tuple[float, float]]:
"""For each reference line that was heard, the start of the first and the
end of the last transcript segment its words were heard in. Words in a
'replace' block are paired with heard words proportionally."""
times: dict[int, tuple[float, float]] = {}
for tag, ai1, ai2, bj1, bj2 in matcher.get_opcodes():
if tag not in ("equal", "replace"):
continue
for j in range(bj1, bj2):
seg = segments[seg_of[ai1 + (j - bj1) * (ai2 - ai1) // (bj2 - bj1)]]
a, b = times.get(line_of[j], (seg.start, seg.end))
times[line_of[j]] = (min(a, seg.start), max(b, seg.end))
return times
def _unheard_runs(segments, ref: RefLyrics, unheard
) -> tuple[list[list[int]], list[list[tuple[float, float]]]]:
"""Runs of reference lines with nothing heard for them, each with the
untranscribed singing between the heard lines either side of it."""
matcher, seg_of, line_of = match_lyrics(segments, ref)
times = _line_times(segments, matcher, seg_of, line_of)
alignable = set(line_of)
runs, spans, run, prev_end = [], [], [], 0.0
def close(next_start: float) -> None:
if run:
runs.append(list(run))
spans.append([(max(a, prev_end), min(b, next_start)) for a, b in unheard
if a < next_start and b > prev_end])
run.clear()
for li in range(len(ref.lines)):
if li in times:
close(times[li][0])
prev_end = times[li][1]
elif li in alignable:
run.append(li)
close(float("inf"))
return runs, spans
def recovered_spans(segments, ref: RefLyrics, unheard) -> list[tuple[float, float]]:
"""The untranscribed singing that stands in for lines Whisper missed, so
alignment can place them there."""
return _merge_spans([sp for spans in _unheard_runs(segments, ref, unheard)[1]
for sp in spans])
def classify_lyrics_match(segments, ref: RefLyrics, pad: float = 1.0, unheard=()
) -> tuple[list[int], list[tuple[float, float]]]:
"""Split reference lines into matched vs. dropped, and flag sung spans no
reference line explains.
dropped_lines: reference line indices covered only by 'insert' opcodes
(present in the lyrics, never heard). missing_spans: the padded, merged
bounds of the heard segments behind 'delete' opcodes (heard, but matching
no reference word). A segment can also hold matched lines, so a span
marks where to look, not words to discard.
"""
matcher, seg_of, line_of = match_lyrics(segments, ref)
covered = [False] * len(line_of)
for tag, _, _, bj1, bj2 in matcher.get_opcodes():
if tag in ("equal", "replace"):
for j in range(bj1, bj2):
covered[j] = True
by_line = {}
for j, li in enumerate(line_of):
by_line.setdefault(li, []).append(covered[j])
dropped_lines = sorted(li for li, cov in by_line.items() if not any(cov))
(present in the lyrics, never heard), unless `unheard` (sung spans the
transcript has no text for) has singing between the heard lines either
side of them: then Whisper missed them, not the singer.
missing_spans: the padded, merged bounds of the heard segments behind
'delete' opcodes (heard, but matching no reference word). A segment can
also hold matched lines, so a span marks where to look, not words to
discard.
"""
runs, run_spans = _unheard_runs(segments, ref, unheard)
dropped_lines = sorted(li for run, sp in zip(runs, run_spans) if not sp
for li in run)
matcher, seg_of, _ = match_lyrics(segments, ref)
spans = []
for tag, ai1, ai2, _, _ in matcher.get_opcodes():
if tag != "delete":
+75
View File
@@ -1,11 +1,23 @@
import json
import numpy as np
import pytest
import soundfile as sf
from songtrace.config import Config
from songtrace.lyrics import normalize_lyrics
from songtrace.pipeline import Ctx, PipelineError, SongInput
from songtrace.stages import align
from songtrace.whisper_client import TranscriptSegment
REAL_UNTRANSCRIBED_SINGING = align.untranscribed_singing
@pytest.fixture(autouse=True)
def no_untranscribed_singing(monkeypatch):
"""The tests have no vocals.wav; those about missed singing set their own."""
monkeypatch.setattr(align, "untranscribed_singing", lambda vocals, segments: [])
LYRICS = "it's one two three\nfour five six seven eight"
@@ -160,3 +172,66 @@ def test_align_flags_missing_span_for_unmatched_singing(tmp_path, monkeypatch):
meta = json.loads((tmp_path / "w" / "align_meta.json").read_text())
assert meta["dropped_lines"] == []
assert meta["missing_spans"] == [[19.0, 24.0]]
def _tone(seconds, sr, level):
t = np.arange(int(seconds * sr)) / sr
return level * np.sin(2 * np.pi * 220 * t)
def test_untranscribed_singing_finds_sung_audio_the_transcript_skips(
tmp_path):
sr = 16000
rng = np.random.default_rng(0)
audio = np.concatenate([
_tone(2, sr, 0.3), # 0-2s: sung and transcribed
np.zeros(2 * sr), # 2-4s: silence
_tone(5, sr, 0.3), # 4-9s: sung, never transcribed
np.zeros(3 * sr), # 9-12s: silence
_tone(1, sr, 0.3), # 12-13s: too short to count
np.zeros(3 * sr),
]) + 0.001 * rng.standard_normal(16 * sr) # separation bleed
path = tmp_path / "vocals.wav"
sf.write(path, audio.astype(np.float32), sr)
segments = [TranscriptSegment(0.0, 2.0, "one two")]
spans = REAL_UNTRANSCRIBED_SINGING(path, segments)
assert len(spans) == 1
assert spans[0][0] == pytest.approx(4.0, abs=0.1)
assert spans[0][1] == pytest.approx(9.0, abs=0.1)
def test_lines_whisper_missed_are_aligned_in_the_untranscribed_singing(
tmp_path, monkeypatch):
lyrics = "one two three\nfour five six\nseven eight nine"
transcript = [{"start": 1.0, "end": 3.0, "text": "one two three"},
{"start": 40.0, "end": 42.0, "text": "seven eight nine"}]
monkeypatch.setattr(align, "untranscribed_singing",
lambda vocals, segments: [(20.0, 26.0)])
rows = run_align(tmp_path, monkeypatch, lyrics, transcript)
check_invariants(rows, lyrics)
middle = [r for r in rows if r["line_index"] == 1]
assert all(19.0 <= r["start"] and r["end"] <= 27.0 for r in middle)
meta = json.loads((tmp_path / "w" / "align_meta.json").read_text())
assert meta["dropped_lines"] == []
def test_stars_bracket_each_line_and_take_no_words(tmp_path, monkeypatch):
lyrics = "one two\n...\nthree four"
transcript = [{"start": 0.0, "end": 10.0, "text": "one two three four"}]
seen = []
def spy(vocals_path, start, end, tokens, device):
seen.append(tokens)
# each star claims a long stretch, so a word given a star's span shows
spans = [(start + i, start + i + (5.0 if t == "*" else 0.5), 0.9)
for i, t in enumerate(tokens)]
return spans
monkeypatch.setattr(align, "force_align_section", spy)
ctx = make_ctx(tmp_path, lyrics, transcript)
align.align_stage().run(ctx)
rows = json.loads((ctx.work / "alignment.json").read_text())
assert seen == [["*", "one", "two", "*", "three", "four", "*"]]
timed = [r for r in rows if r["word"] != "..."]
assert all(r["end"] - r["start"] == pytest.approx(0.5) for r in timed)
check_invariants(rows, lyrics)
+30 -1
View File
@@ -1,6 +1,7 @@
from songtrace.lyrics import normalize_lyrics
from songtrace.whisper_client import TranscriptSegment
from songtrace.stages.transcribe import classify_lyrics_match, lyrics_similarity, vocal_sections
from songtrace.stages.transcribe import (classify_lyrics_match, lyrics_similarity,
recovered_spans, vocal_sections)
def seg(s, e, t=""):
return TranscriptSegment(start=s, end=e, text=t)
@@ -64,3 +65,31 @@ def test_classify_lyrics_match_scopes_missing_span_to_adlib_not_whole_section():
assert missing == [(5.0, 9.0)]
for a, b in missing:
assert not (a <= 1.0 < b) # matched line's start (1.0) untouched
THREE_LINES = "one two three\nfour five six\nseven eight nine"
def test_lines_whisper_missed_are_kept_when_singing_went_untranscribed():
ref = normalize_lyrics(THREE_LINES)
segs = [seg(0, 2, "one two three"), seg(20, 22, "seven eight nine")]
assert classify_lyrics_match(segs, ref)[0] == [1]
dropped, _ = classify_lyrics_match(segs, ref, unheard=[(8.0, 14.0)])
assert dropped == []
def test_untranscribed_singing_elsewhere_does_not_keep_a_line():
ref = normalize_lyrics(THREE_LINES)
segs = [seg(0, 2, "one two three"), seg(20, 22, "seven eight nine")]
dropped, _ = classify_lyrics_match(segs, ref, unheard=[(30.0, 35.0)])
assert dropped == [1]
def test_recovered_spans_are_the_singing_between_heard_neighbours():
ref = normalize_lyrics(THREE_LINES)
segs = [seg(10, 12, "four five six"), seg(20, 22, "seven eight nine")]
unheard = [(3.0, 9.0), (14.0, 18.0), (30.0, 35.0)]
# line 0 comes before anything heard, so only the singing before line 1
# can stand in for it; the other spans have no missing line around them
assert recovered_spans(segs, ref, unheard) == [(3.0, 9.0)]
assert classify_lyrics_match(segs, ref, unheard=unheard)[0] == []