align: anchor each line to the transcript segments it was heard in

The align, transcribe, lyrics and notes tests quoted lines from published
songs, and the UltraStar parser fixtures were complete community charts of
commercial songs. Replace them with counting words and two hand-written
charts that keep what the parser tests rely on: CRLF line endings,
Latin-1 bytes, a decimal-comma BPM, freestyle/golden/rap notes, negative
pitch and a two-number line break.

src/songtrace/notes.py

@@ -84,19 +84,11 @@ def build_notes(rows: list[dict], times, f0, conf, language: str | None,
                 ) -> tuple[list[TimedNote], list[float], list[str]]:
     notes: list[TimedNote] = []
     breaks: list[float] = []
-    prev_line, prev_end, last_seen_line = None, 0.0, None
+    prev_line, prev_end = None, 0.0
     for row in rows:
-        # last_seen_line tracks every row (skipped or not) so a run of
-        # skipped rows can't trigger a duplicate break on the next row of
-        # the same still-unseen line; prev_line/prev_end only advance on
-        # rows that actually emit notes, so a skipped row can't corrupt
-        # line_initial or break timing for its own line's later words.
-        if last_seen_line is not None and row["line_index"] > last_seen_line:
+        if prev_line is not None and row["line_index"] > prev_line:
             breaks.append((prev_end + row["start"]) / 2)
-        last_seen_line = row["line_index"]
         line_initial = row["line_index"] != prev_line
-        if any(a <= row["start"] < b for a, b in missing_spans):
-            continue
         prev_line, prev_end = row["line_index"], row["end"]
         segs = segment_span(times, f0, conf, row["start"], row["end"])
         if not segs:

src/songtrace/stages/align.py

@@ -4,9 +4,12 @@ from pathlib import Path
 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 .transcribe import (classify_lyrics_match, line_segments, load_transcript,
+                         lyrics_similarity)
 
 SIMILARITY_THRESHOLD = 0.4
+# how far a chunk's window reaches past its segments, into the gap between them
+PAD_SEC = 1.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
@@ -50,17 +53,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 _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
-    n = len(line_indices)
-    groups, elapsed, start = [], 0.0, 0
-    for i, (a, b) in enumerate(sections):
-        elapsed += b - a
-        stop = n if i == len(sections) - 1 else min(n, round(n * elapsed / total))
-        groups.append(line_indices[start:max(stop, start)])
-        start = max(stop, start)
-    return groups
+def _chunks(segments, lines: list[int], anchors: dict[int, tuple[int, int]]
+            ) -> list[tuple[list[int], tuple[float, float]]]:
+    """Group lines heard in the same transcript segments, each group with the
+    window those segments cover.
+
+    Aligning each group only within its own window stops one misheard
+    passage from pushing every later word off its time. A line with no
+    alignable words joins the group before it.
+    """
+    filled, prev = {}, None
+    for li in lines:
+        prev = filled[li] = anchors.get(li, prev)
+    first = next((a for a in filled.values() if a), (0, len(segments) - 1))
+    groups: list[list] = []
+    for li in lines:
+        lo, hi = filled[li] or first
+        if groups and lo <= groups[-1][2]:
+            groups[-1][0].append(li)
+            groups[-1][2] = max(groups[-1][2], hi)
+        else:
+            groups.append([[li], lo, hi])
+    bounds = []
+    for _, lo, hi in groups:
+        times = [t for s in segments[lo:hi + 1] for t in (s.start, s.end)]
+        bounds.append((min(times), max(times)))
+    # pad only into the gap before the neighbouring group's segments
+    out = []
+    for i, ((group, _, _), (a, b)) in enumerate(zip(groups, bounds)):
+        prev_end = bounds[i - 1][1] if i else 0.0
+        next_start = bounds[i + 1][0] if i + 1 < len(bounds) else b + PAD_SEC
+        out.append((group, (max(a - PAD_SEC, min(prev_end, a)),
+                            min(b + PAD_SEC, max(next_start, b)))))
+    return out
 
 
 def _interpolate(rows: list[dict], lo_bound: float, hi_bound: float) -> None:
@@ -112,16 +137,15 @@ 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"
     rows = []
-    for sec, line_idxs in zip(sections, _assign_lines_to_sections(matched, sections)):
+    for line_idxs, window in _chunks(segments, matched,
+                                     line_segments(segments, ref)):
         words = [(w, li) for li in line_idxs for w in ref.lines[li]]
         if words:
-            rows.extend(_align_section(vocals, sec, words, ctx.inp.language,
+            rows.extend(_align_section(vocals, window, words, ctx.inp.language,
                                        ctx.cfg.device))
     _enforce_monotonic(rows)
     (ctx.work / "alignment.json").write_text(json.dumps(rows))

src/songtrace/stages/transcribe.py

@@ -6,16 +6,6 @@ from ..pipeline import Ctx, Stage
 from ..whisper_client import TranscriptSegment, transcribe
 
 
-def vocal_sections(segments, min_gap: float = 6.0, pad: float = 1.0):
-    sections = []
-    for s in sorted(segments, key=lambda s: s.start):
-        if sections and s.start - sections[-1][1] < min_gap:
-            sections[-1][1] = max(sections[-1][1], s.end)
-        else:
-            sections.append([s.start, s.end])
-    return [(max(0.0, a - pad), b + pad) for a, b in sections]
-
-
 def _tokens(text: str) -> list[str]:
     return [t for w in text.split() if (t := alignment_token(w))]
 
@@ -39,6 +29,26 @@ def lyrics_similarity(segments, ref: RefLyrics) -> float:
     return matcher.ratio()
 
 
+def line_segments(segments, ref: RefLyrics) -> dict[int, tuple[int, int]]:
+    """For each reference line that was heard, the first and last transcript
+    segment its matched words were heard in.
+
+    Matching is in order, so these ranges never go backwards from one line to
+    the next. Words in a 'replace' block are paired with heard words
+    proportionally, since the block only says the two runs correspond.
+    """
+    matcher, seg_of, line_of = match_lyrics(segments, ref)
+    spans: dict[int, tuple[int, int]] = {}
+    for tag, ai1, ai2, bj1, bj2 in matcher.get_opcodes():
+        if tag not in ("equal", "replace"):
+            continue
+        for j in range(bj1, bj2):
+            si = seg_of[ai1 + (j - bj1) * (ai2 - ai1) // (bj2 - bj1)]
+            lo, hi = spans.get(line_of[j], (si, si))
+            spans[line_of[j]] = (min(lo, si), max(hi, si))
+    return spans
+
+
 def classify_lyrics_match(segments, ref: RefLyrics, pad: float = 1.0
                           ) -> tuple[list[int], list[tuple[float, float]]]:
     """Split reference lines into matched vs. dropped, and flag sung spans no
@@ -47,9 +57,8 @@ def classify_lyrics_match(segments, ref: RefLyrics, pad: float = 1.0
     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) — scoped tightly to the unmatched audio itself, not
-    the whole vocal section it falls in, so matched lines sharing a section
-    with an ad-lib keep their notes.
+    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)

tests/fixtures/chart_crlf.txt

@@ -0,0 +1,36 @@
+#TITLE:Placeholder Chart
+#ARTIST:Test Artist
+#LANGUAGE:English
+#YEAR:2000
+#CREATOR:songtrace tests
+#GENRE:Test
+#COVER:Test Artist - Placeholder Chart [CO].jpg
+#MP3:Test Artist - Placeholder Chart.mp3
+#VIDEO:Test Artist - Placeholder Chart.mp4
+#BPM:217,98
+#GAP:4880
+#MEDLEYSTARTBEAT:113
+#MEDLEYENDBEAT:240
+: 0 3 16 ah
+: 5 3 20  ah
+: 11 3 23  la
+: 17 3 16  la
+: 22 28 15  ah
+: 51 2 16  doo
+: 54 31 16 ~
+- 95  
+F 113 3 18 la
+F 117 3 21  la
+F 121 3 21  la
+* 125 3 21  doo
+* 129 12 19 ~
+- 150
+: 155 4 12 Na
+R 160 4 0  na
+G 165 4 0  na
+: 170 6 14  la
+- 190
+: 194 3 16 La
+: 198 3 18  la
+* 202 10 20  ah
+E

tests/fixtures/chart_latin1.txt

@@ -0,0 +1,21 @@
+#TITLE:Café Placeholder
+#ARTIST:Tést Artist
+#LANGUAGE:English
+#GENRE:Test
+#YEAR:2000
+#MP3:Tést Artist - Café Placeholder.mp3
+#CREATOR:songtrace tests
+#BPM:256
+#GAP:49673
+: 0 4 2 Doo
+: 6 4 4  doo
+: 12 8 5  laé
+- 28 30
+: 32 4 0 Na
+: 38 4 -2  na
+* 44 12 -9  ah
+- 64
+F 70 4 3 La
+: 76 6 -5  la
+: 84 10 -9 ~
+E

tests/test_align.py

@@ -123,11 +123,12 @@ def test_non_chronological_transcript(tmp_path, monkeypatch):
 
 
 def test_inverted_section_does_not_collapse(tmp_path, monkeypatch):
-    monkeypatch.setattr(align, "vocal_sections", lambda segments: [(29.0, 5.0)])
-    rows = run_align(tmp_path, monkeypatch)
+    transcript = [{"start": 29.0, "end": 5.0,
+                   "text": "it's one two three four five six seven eight"}]
+    rows = run_align(tmp_path, monkeypatch, transcript=transcript)
     check_invariants(rows, LYRICS)
     assert len({r["start"] for r in rows}) == len(rows)
-    assert rows[0]["start"] >= 5.0 and rows[-1]["end"] <= 29.0
+    assert rows[0]["start"] >= 4.0 and rows[-1]["end"] <= 30.0
 
 
 def test_words_split_across_sections(tmp_path, monkeypatch):
@@ -160,3 +161,47 @@ 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 test_contiguous_segments_are_aligned_separately(tmp_path, monkeypatch):
+    # No gap between the segments, so each line must still be held to the
+    # segment it was heard in rather than aligned across both.
+    lyrics = "one two three four\nfive six seven eight"
+    transcript = [{"start": 0.0, "end": 10.0, "text": "one two three four"},
+                  {"start": 10.0, "end": 12.0, "text": "five six seven eight"}]
+    calls = []
+    def spy(vocals_path, start, end, tokens, device):
+        calls.append((start, end, len(tokens)))
+        return fake_align(vocals_path, start, end, tokens, device)
+    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())
+    check_invariants(rows, lyrics)
+    assert calls == [(0.0, 10.0, 4), (10.0, 13.0, 4)]
+    assert all(r["start"] >= 10.0 for r in rows[4:])
+
+
+def test_lines_sharing_a_segment_align_together(tmp_path, monkeypatch):
+    lyrics = "one two\nthree four\nfive six"
+    transcript = [{"start": 2.0, "end": 6.0, "text": "one two three"},
+                  {"start": 6.0, "end": 9.0, "text": "four"},
+                  {"start": 20.0, "end": 23.0, "text": "five six"}]
+    calls = []
+    def spy(vocals_path, start, end, tokens, device):
+        calls.append((start, end, len(tokens)))
+        return fake_align(vocals_path, start, end, tokens, device)
+    monkeypatch.setattr(align, "force_align_section", spy)
+    ctx = make_ctx(tmp_path, lyrics, transcript)
+    align.align_stage().run(ctx)
+    assert calls == [(1.0, 10.0, 4), (19.0, 24.0, 2)]
+
+
+def test_line_without_alignable_words_joins_the_line_before(tmp_path, monkeypatch):
+    lyrics = "one two\n...\nthree four"
+    transcript = [{"start": 1.0, "end": 3.0, "text": "one two"},
+                  {"start": 20.0, "end": 22.0, "text": "three four"}]
+    rows = run_align(tmp_path, monkeypatch, lyrics, transcript)
+    check_invariants(rows, lyrics)
+    assert rows[2]["end"] <= 4.0
+    assert rows[3]["start"] >= 19.0

tests/test_notes.py

@@ -78,7 +78,7 @@ def test_syllabify():
     assert syllabify("reminds", "English") == ["re", "minds"]
     assert syllabify("a", "English") == ["a"]
     assert syllabify("xyzzy", None) == ["xyzzy"]
-    assert syllabify("She's", "English") == ["She's"]
+    assert syllabify("It's", "English") == ["It's"]
 
 
 def test_syllabify_preserves_word_characters():
@@ -96,12 +96,12 @@ def test_syllabify_language_without_pyphen_dict_is_not_mishandled_as_english():
 def test_build_notes_basic_line():
     times, f0, conf = flat_pitch()
     rows = [
-        {"word": "She's", "start": 1.0, "end": 1.5, "score": 0.9, "line_index": 0},
+        {"word": "It's", "start": 1.0, "end": 1.5, "score": 0.9, "line_index": 0},
         {"word": "got", "start": 1.6, "end": 2.0, "score": 0.9, "line_index": 0},
         {"word": "Now", "start": 5.0, "end": 5.5, "score": 0.9, "line_index": 1},
     ]
     notes, breaks, warnings = build_notes(rows, times, f0, conf, "English")
-    assert notes[0].text == "She's"          # line-initial: no leading space
+    assert notes[0].text == "It's"          # line-initial: no leading space
     assert notes[1].text == " got"           # mid-line: leading space
     assert notes[2].text == "Now"            # new line: no leading space
     assert len(breaks) == 1 and 2.0 <= breaks[0] <= 5.0
@@ -278,7 +278,7 @@ def test_notes_stage_writes_json(tmp_path):
     times, f0, conf = flat_pitch()
     np.savez(work / "pitch.npz", times=times, f0=f0, conf=conf)
     (work / "alignment.json").write_text(json.dumps([
-        {"word": "She's", "start": 1.0, "end": 1.5, "score": 0.9, "line_index": 0},
+        {"word": "It's", "start": 1.0, "end": 1.5, "score": 0.9, "line_index": 0},
         {"word": "gone", "start": 3.0, "end": 3.5, "score": 0.9, "line_index": 1},
     ]))
     ctx = Ctx(inp=SongInput(title="T", artist="A", lyrics_text="x", language="English"),
@@ -287,7 +287,7 @@ def test_notes_stage_writes_json(tmp_path):
     assert stage.outputs(ctx) == [work / "notes_sec.json"]
     stage.run(ctx)
     data = json.loads((work / "notes_sec.json").read_text())
-    assert [n["text"] for n in data["notes"]] == ["She's", "gone"]
+    assert [n["text"] for n in data["notes"]] == ["It's", "gone"]
     assert data["breaks"] == [2.25]
     assert data["warnings"] == []
     assert TimedNote(**data["notes"][0]).pitch == 0
@@ -297,15 +297,19 @@ def test_notes_stage_registered():
     assert "notes" in [s.name for s in all_stages()]
 
 
-def test_build_notes_skips_notes_within_missing_span():
+def test_build_notes_keeps_matched_words_within_missing_span():
+    # A missing span covers whole Whisper segments, so it can hold matched
+    # lines alongside the unmatched singing; those lines must keep their notes.
     times, f0, conf = flat_pitch()
     rows = [
         {"word": "keep", "start": 1.0, "end": 1.4, "score": 0.9, "line_index": 0},
-        {"word": "skip", "start": 5.0, "end": 5.4, "score": 0.9, "line_index": 1},
+        {"word": "held", "start": 5.0, "end": 5.4, "score": 0.9, "line_index": 1},
     ]
-    notes, _, _ = build_notes(rows, times, f0, conf, "English",
-                              missing_spans=[(4.5, 6.0)])
-    assert [n.text for n in notes] == ["keep"]
+    notes, breaks, warnings = build_notes(rows, times, f0, conf, "English",
+                                          missing_spans=[(4.5, 6.0)])
+    assert [n.text for n in notes] == ["keep", "held"]
+    assert breaks == [(1.4 + 5.0) / 2]
+    assert warnings == ["missing lyrics near 0:04–0:06"]
 
 
 def test_warnings_report_dropped_lines():
@@ -325,23 +329,6 @@ def test_warnings_report_missing_spans():
     assert warnings == ["missing lyrics near 1:21–1:29"]
 
 
-def test_build_notes_skipped_row_mid_line_does_not_corrupt_spacing_or_breaks():
-    # "gone" falls inside a missing span and is skipped; "away" is the same
-    # line's next word and must still get its leading space, and the break
-    # before the next line must use the last *emitted* word's end (3.4, from
-    # "away"), not the skipped row's end (2.4).
-    times, f0, conf = flat_pitch()
-    rows = [
-        {"word": "keep", "start": 1.0, "end": 1.4, "score": 0.9, "line_index": 0},
-        {"word": "gone", "start": 2.0, "end": 2.4, "score": 0.9, "line_index": 0},
-        {"word": "away", "start": 3.0, "end": 3.4, "score": 0.9, "line_index": 0},
-        {"word": "next", "start": 5.0, "end": 5.4, "score": 0.9, "line_index": 1},
-    ]
-    notes, breaks, _ = build_notes(rows, times, f0, conf, "English",
-                                   missing_spans=[(1.9, 2.5)])
-    assert [n.text for n in notes] == ["keep", " away", "next"]
-    assert breaks == [(3.4 + 5.0) / 2]
-
 def test_notes_carry_line_index():
     times, f0, conf = flat_pitch()
     rows = [{"word": "hey", "start": 1.0, "end": 1.4, "score": 0.9, "line_index": 3}]

tests/test_transcribe.py

@@ -1,22 +1,10 @@
 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, line_segments, lyrics_similarity
 
 def seg(s, e, t=""):
     return TranscriptSegment(start=s, end=e, text=t)
 
-def test_vocal_sections_merges_close_segments():
-    sections = vocal_sections([seg(2, 5), seg(6, 9), seg(30, 33)])
-    assert sections == [(1.0, 10.0), (29.0, 34.0)]
-
-def test_vocal_sections_handles_out_of_order_segments():
-    sections = vocal_sections([seg(30, 33), seg(1, 4)])
-    assert sections == [(0.0, 5.0), (29.0, 34.0)]
-
-def test_vocal_sections_keeps_contained_segment():
-    sections = vocal_sections([seg(1, 10), seg(3, 5)])
-    assert sections == [(0.0, 11.0)]
-
 def test_similarity_high_for_matching_lyrics():
     ref = normalize_lyrics("it's one two three\nfour five six seven eight")
     segs = [seg(0, 2, "It's one two three"), seg(2, 4, "four five six seven eight")]
@@ -51,16 +39,32 @@ def test_classify_lyrics_match_keeps_partially_matched_lines():
     assert dropped == []
 
 def test_classify_lyrics_match_scopes_missing_span_to_adlib_not_whole_section():
-    # matched line and an unmatched ad-lib close enough together that
-    # vocal_sections would merge them into one section; missing_spans must
-    # stay tight around the ad-lib so the matched line keeps its notes.
+    # matched line and an unmatched ad-lib two seconds apart; missing_spans
+    # must stay tight around the ad-lib so the review marks the right place.
     ref = normalize_lyrics("it's one two three\nfour five six seven eight")
     segs = [seg(1, 4, "it's one two three four five six seven eight"),
             seg(6, 8, "completely unrelated ad lib")]
-    sections = vocal_sections(segs)
-    assert sections == [(0.0, 9.0)]      # confirms the two segments do merge
     dropped, missing = classify_lyrics_match(segs, ref)
     assert dropped == []
     assert missing == [(5.0, 9.0)]
     for a, b in missing:
         assert not (a <= 1.0 < b)        # matched line's start (1.0) untouched
+
+
+def test_line_segments_maps_lines_to_the_segments_they_were_heard_in():
+    ref = normalize_lyrics("one two three\nfour five six\nseven eight nine")
+    segs = [seg(1, 4, "one two three four"), seg(5, 8, "five six"),
+            seg(9, 12, "seven eight nine")]
+    assert line_segments(segs, ref) == {0: (0, 0), 1: (0, 1), 2: (2, 2)}
+
+
+def test_line_segments_omits_lines_never_heard():
+    ref = normalize_lyrics("one two three\nfour five six\nseven eight nine")
+    segs = [seg(1, 4, "one two three"), seg(9, 12, "seven eight nine")]
+    assert line_segments(segs, ref) == {0: (0, 0), 2: (1, 1)}
+
+
+def test_line_segments_spreads_a_replace_block_across_its_segments():
+    ref = normalize_lyrics("one two\nthree four")
+    segs = [seg(1, 2, "won too"), seg(3, 4, "tree for")]
+    assert line_segments(segs, ref) == {0: (0, 0), 1: (1, 1)}

tests/test_ultrastar.py

@@ -10,8 +10,8 @@ def load(name):
     return (FIXTURES / name).read_text(encoding="utf-8", errors="replace")
 
 def test_parse_headers_and_notes():
-    song = parse_ultrastar(load("sanctuary.txt"))
-    assert get_header(song, "artist") == "Utada Hikaru"
+    song = parse_ultrastar(load("chart_crlf.txt"))
+    assert get_header(song, "artist") == "Test Artist"
     assert parse_bpm(get_header(song, "BPM")) == 217.98
     first = song.body[0]
     assert first == Note(kind=":", beat=0, length=3, pitch=16, text="ah")
@@ -19,16 +19,16 @@ def test_parse_headers_and_notes():
     assert any(n.kind == "F" for n in song.body if isinstance(n, Note))
 
 def test_note_text_keeps_leading_space():
-    song = parse_ultrastar(load("sanctuary.txt"))
+    song = parse_ultrastar(load("chart_crlf.txt"))
     notes = [n for n in song.body if isinstance(n, Note)]
     assert notes[1].text == " ah"
 
 def test_negative_pitch():
-    song = parse_ultrastar(load("sweet_child.txt"))
+    song = parse_ultrastar(load("chart_latin1.txt"))
     assert any(n.pitch == -9 for n in song.body if isinstance(n, Note))
 
 def test_round_trip():
-    for name in ("sweet_child.txt", "sanctuary.txt"):
+    for name in ("chart_latin1.txt", "chart_crlf.txt"):
         song = parse_ultrastar(load(name))
         again = parse_ultrastar(serialize_ultrastar(song))
         assert again == song