align: exclude never-sung lines from proportional distribution, record align_meta.json

- Added classify_lyrics_match import to wire lyric classification into align stage
- Changed _assign_lines_to_sections signature to accept line_indices list instead of ref
- Modified _run to classify lyrics match, filter dropped lines, and write align_meta.json
- Sort segments by start time before processing for correct matching of non-chronological transcripts
- Added tests for dropped line detection and missing span flagging
This commit is contained in:
2026-07-27 23:06:02 -07:00
parent cf1c7410ab
commit ebb7fc4433
2 changed files with 37 additions and 9 deletions
+14 -9
View File
@@ -3,7 +3,7 @@ from pathlib import Path
from ..lyrics import RefLyrics, alignment_token, normalize_lyrics from ..lyrics import RefLyrics, alignment_token, normalize_lyrics
from ..pipeline import Ctx, PipelineError, Stage from ..pipeline import Ctx, PipelineError, Stage
from .transcribe import load_transcript, lyrics_similarity, vocal_sections from .transcribe import classify_lyrics_match, load_transcript, lyrics_similarity, vocal_sections
SIMILARITY_THRESHOLD = 0.4 SIMILARITY_THRESHOLD = 0.4
@@ -42,16 +42,15 @@ def force_align_section(vocals_path: Path, start: float, end: float,
sum(t.score for t in s) / len(s)) for s in spans] sum(t.score for t in s) / len(s)) for s in spans]
def _assign_lines_to_sections(ref: RefLyrics, sections) -> list[list[int]]: def _assign_lines_to_sections(line_indices: list[int], sections) -> list[list[int]]:
"""Split lyric lines across vocal sections proportionally to section duration.""" """Split matched lyric line indices across vocal sections proportionally to section duration."""
total = sum(b - a for a, b in sections) or 1.0 total = sum(b - a for a, b in sections) or 1.0
n_lines = len(ref.lines) n = len(line_indices)
groups, elapsed, start = [], 0.0, 0 groups, elapsed, start = [], 0.0, 0
for i, (a, b) in enumerate(sections): for i, (a, b) in enumerate(sections):
elapsed += b - a elapsed += b - a
stop = n_lines if i == len(sections) - 1 else min( stop = n if i == len(sections) - 1 else min(n, round(n * elapsed / total))
n_lines, round(n_lines * elapsed / total)) groups.append(line_indices[start:max(stop, start)])
groups.append(list(range(start, max(stop, start))))
start = max(stop, start) start = max(stop, start)
return groups return groups
@@ -99,7 +98,7 @@ def _align_section(vocals: Path, sec: tuple[float, float], words: list[tuple[str
def _run(ctx: Ctx) -> None: def _run(ctx: Ctx) -> None:
ref = normalize_lyrics(ctx.inp.lyrics_text) ref = normalize_lyrics(ctx.inp.lyrics_text)
segments = load_transcript(ctx) segments = sorted(load_transcript(ctx), key=lambda s: s.start)
sim = lyrics_similarity(segments, ref) sim = lyrics_similarity(segments, ref)
if sim < SIMILARITY_THRESHOLD: if sim < SIMILARITY_THRESHOLD:
raise PipelineError( raise PipelineError(
@@ -107,15 +106,21 @@ def _run(ctx: Ctx) -> None:
"wrong lyrics or wrong recording?") "wrong lyrics or wrong recording?")
# normalised so an inverted span can never collapse a section onto one instant # 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)] sections = [(min(a, b), max(a, b)) for a, b in vocal_sections(segments)]
dropped_lines, missing_spans = classify_lyrics_match(segments, ref, sections)
matched = [i for i in range(len(ref.lines)) if i not in set(dropped_lines)]
vocals = ctx.work / "vocals.wav" vocals = ctx.work / "vocals.wav"
rows = [] rows = []
for sec, line_idxs in zip(sections, _assign_lines_to_sections(ref, sections)): 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]] words = [(w, li) for li in line_idxs for w in ref.lines[li]]
if words: if words:
rows.extend(_align_section(vocals, sec, words, ctx.inp.language, rows.extend(_align_section(vocals, sec, words, ctx.inp.language,
ctx.cfg.device)) ctx.cfg.device))
_enforce_monotonic(rows) _enforce_monotonic(rows)
(ctx.work / "alignment.json").write_text(json.dumps(rows)) (ctx.work / "alignment.json").write_text(json.dumps(rows))
(ctx.work / "align_meta.json").write_text(json.dumps({
"dropped_lines": dropped_lines,
"missing_spans": [list(s) for s in missing_spans],
}))
def align_stage() -> Stage: def align_stage() -> Stage:
+23
View File
@@ -137,3 +137,26 @@ def test_words_split_across_sections(tmp_path, monkeypatch):
check_invariants(rows, LYRICS) check_invariants(rows, LYRICS)
assert rows[3]["end"] <= 5.0 assert rows[3]["end"] <= 5.0
assert rows[4]["start"] >= 29.0 assert rows[4]["start"] >= 29.0
def test_align_drops_lines_never_sung(tmp_path, monkeypatch):
lyrics = "it's one two three\nfour five six seven eight\nnine ten eleven twelve"
transcript = [{"start": 1.0, "end": 4.0,
"text": "it's one two three four five six seven eight"}]
rows = run_align(tmp_path, monkeypatch, lyrics, transcript)
assert {r["line_index"] for r in rows} == {0, 1}
meta = json.loads((tmp_path / "w" / "align_meta.json").read_text())
assert meta["dropped_lines"] == [2]
assert meta["missing_spans"] == []
def test_align_flags_missing_span_for_unmatched_singing(tmp_path, monkeypatch):
lyrics = "it's one two three\nfour five six seven eight"
transcript = [
{"start": 1.0, "end": 4.0, "text": "it's one two three four five six seven eight"},
{"start": 20.0, "end": 23.0, "text": "completely unrelated ad lib bridge"},
]
run_align(tmp_path, monkeypatch, lyrics, transcript)
meta = json.loads((tmp_path / "w" / "align_meta.json").read_text())
assert meta["dropped_lines"] == []
assert meta["missing_spans"] == [[19.0, 24.0]]