Upload structure.py with huggingface_hub

structure.py

@@ -0,0 +1,272 @@
+"""Map a canonical lyric sheet onto what the recording actually sings.
+
+A lyric sheet is *canonical*: the chorus is written once, repeats are collapsed,
+and unsung extra verses sometimes ride along. Karaoke needs the *performance*
+sequence — the real order, with the chorus appearing as many times as it is sung.
+Forced alignment cannot invent that: it consumes the reference in order, so a
+chorus written once but sung three times leaves two thirds of the vocal to be
+absorbed by whatever line happens to be adjacent (measured: 1.37 s mean line
+error, worst 3.7 s, when a reference carried lines the recording never sang).
+
+The two inputs have exactly complementary strengths:
+
+    sheet       right words, wrong structure
+    transcript  right structure, wrong words
+
+So we use the transcript only to decide *which sheet line is being sung when*,
+never for its words. That works even when the transcript is poor — measured CER
+on real songs is 0.64, but token-overlap similarity still identifies the correct
+sheet line, because picking one line out of ~30 needs far less signal than
+reading it. This is why the mapper is worth more than a better ASR.
+
+No network, no API key, no LLM: it is a similarity matrix plus a Viterbi pass
+with a continuation bonus. See `resolve_with_llm` for where a model genuinely
+helps (ambiguous sheets), which is a much smaller job than this one.
+"""
+from __future__ import annotations
+
+import re
+from typing import List, Tuple
+
+# A sheet line only counts as "sung here" above this token-overlap score. Below
+# it the transcript segment is an ad-lib, an instrumental mis-fire, or a line the
+# sheet simply does not contain.
+MIN_MATCH = 0.34
+# Reward for continuing to the next sheet line, which disambiguates the common
+# case of near-identical lines (a chorus whose lines differ by one word) without
+# forbidding the backward jump that a chorus repeat *is*.
+CONTINUE_BONUS = 0.22
+# Cost of jumping backwards in the sheet, i.e. claiming a line is sung again.
+# A *penalty*, not a reward: a repeat has to be earned by the similarity, because
+# sheets legitimately contain the same chorus text twice and a second chorus
+# reads as a backward jump otherwise. Swept against ground truth — at 0.0 two
+# fixtures gained phantom repeats; at -0.10 both are exact and the real repeat is
+# still found.
+REPEAT_BONUS = -0.10
+
+
+def _norm(s: str) -> str:
+    s = s.lower().replace("ё", "е")
+    s = re.sub(r"[^\w\s]|_", " ", s, flags=re.UNICODE)
+    return re.sub(r"\s+", " ", s).strip()
+
+
+def _tokens(s: str) -> List[str]:
+    return _norm(s).split()
+
+
+def _bigrams(word: str) -> set:
+    w = f" {word} "
+    return {w[i:i + 2] for i in range(len(w) - 1)}
+
+
+def word_similarity(a: str, b: str) -> float:
+    """Dice coefficient over character bigrams — tolerant of the one- or
+    two-character errors that dominate sung ASR output."""
+    if a == b:
+        return 1.0
+    ga, gb = _bigrams(a), _bigrams(b)
+    if not ga or not gb:
+        return 0.0
+    return 2 * len(ga & gb) / (len(ga) + len(gb))
+
+
+def line_similarity(hyp: str, ref: str) -> float:
+    """Greedy token matching between two lines, 0…1.
+
+    Token-level rather than character-level so that a transcript which gets a
+    word wrong still scores the line it belongs to. Length-normalized against
+    the *reference* so a long transcript run doesn't out-score a short line.
+    """
+    ht, rt = _tokens(hyp), _tokens(ref)
+    if not ht or not rt:
+        return 0.0
+    used = [False] * len(ht)
+    score = 0.0
+    for rw in rt:
+        best, bi = 0.0, -1
+        for i, hw in enumerate(ht):
+            if used[i]:
+                continue
+            s = word_similarity(rw, hw)
+            if s > best:
+                best, bi = s, i
+        if bi >= 0 and best >= 0.5:
+            used[bi] = True
+            score += best
+    return score / len(rt)
+
+
+def map_performance(sheet: List[str], hyp_lines: List[dict],
+                    min_match: float = MIN_MATCH) -> List[dict]:
+    """Decide which sheet line each transcript segment is singing.
+
+    `hyp_lines` are the transcript's timed lines ({startMs, endMs, text}).
+    Returns one entry per transcript segment: the matched sheet index (or None),
+    its score, and the segment's timing.
+
+    Viterbi over sheet index, so the choice is made for the sequence as a whole
+    rather than greedily per line — that is what lets a repeated chorus win over
+    a locally-similar verse line.
+    """
+    n, m = len(hyp_lines), len(sheet)
+    if not n or not m:
+        return []
+
+    sim = [[line_similarity(h["text"], s) for s in sheet] for h in hyp_lines]
+
+    NONE = m                                  # an extra state: "matches nothing"
+    best = [[float("-inf")] * (m + 1) for _ in range(n)]
+    back = [[-1] * (m + 1) for _ in range(n)]
+    for j in range(m):
+        best[0][j] = sim[0][j]
+    best[0][NONE] = min_match * 0.999          # ...just under any real match
+
+    for i in range(1, n):
+        for j in range(m + 1):
+            emit = min_match * 0.999 if j == NONE else sim[i][j]
+            for pj in range(m + 1):
+                if best[i - 1][pj] == float("-inf"):
+                    continue
+                bonus = 0.0
+                if j != NONE and pj != NONE:
+                    if j == pj + 1:
+                        bonus = CONTINUE_BONUS      # running through a section
+                    elif j < pj:
+                        bonus = REPEAT_BONUS        # jumped back: a repeat
+                v = best[i - 1][pj] + emit + bonus
+                if v > best[i][j]:
+                    best[i][j] = v
+                    back[i][j] = pj
+
+    j = max(range(m + 1), key=lambda k: best[n - 1][k])
+    path = [j]
+    for i in range(n - 1, 0, -1):
+        j = back[i][j]
+        path.append(j)
+    path.reverse()
+
+    out = []
+    for i, j in enumerate(path):
+        matched = j != NONE and sim[i][j] >= min_match
+        out.append({
+            "startMs": hyp_lines[i]["startMs"],
+            "endMs": hyp_lines[i]["endMs"],
+            "sheetIdx": j if matched else None,
+            "score": round(sim[i][j], 3) if j != NONE else 0.0,
+            "hyp": hyp_lines[i]["text"],
+        })
+    return out
+
+
+def expand_reference(sheet: List[str], hyp_lines: List[dict],
+                     min_match: float = MIN_MATCH) -> Tuple[List[str], List[dict]]:
+    """Build the reference the aligner should actually be given.
+
+    Returns `(lines, plan)` where `lines` is the sheet rewritten in performance
+    order — a chorus sung twice appears twice — and `plan` is the mapping detail.
+
+    **Strictly additive: no sheet line is ever dropped.** The mapper's recall is
+    bounded by the transcript's, and the transcript is poor — on a fixture where
+    all 16 sheet lines are sung, the ASR produced 12 usable segments, so a
+    "drop what wasn't matched" rule deleted 8 lines that really were sung. Adding
+    a repeat that isn't there costs a little alignment drift; deleting a line the
+    singer sings loses it from the karaoke entirely. So the sheet is the backbone
+    and the transcript may only *insert* into it.
+
+    Consecutive transcript segments matching the *same* sheet line collapse into
+    one: the transcript often splits a sung line in two, which is an artefact
+    rather than a repeat.
+    """
+    plan = map_performance(sheet, hyp_lines, min_match)
+
+    # Collapse ASR-split duplicates, keeping the matched entries in time order.
+    matched: List[dict] = []
+    for p in plan:
+        j = p["sheetIdx"]
+        if j is None:
+            continue
+        if matched and j == matched[-1]["sheetIdx"] and \
+                p["startMs"] - matched[-1]["endMs"] < 1500:
+            matched[-1]["endMs"] = p["endMs"]
+            continue
+        matched.append({"sheetIdx": j, "startMs": p["startMs"],
+                        "endMs": p["endMs"], "score": p["score"]})
+
+    lines: List[str] = []
+    order: List[dict] = []
+
+    def emit(j: int, repeat: bool, hit: dict = None) -> None:
+        lines.append(sheet[j])
+        order.append({
+            "sheetIdx": j, "repeat": repeat,
+            "startMs": (hit or {}).get("startMs"),
+            "endMs": (hit or {}).get("endMs"),
+            "score": (hit or {}).get("score", 0.0),
+        })
+
+    # Walk the matched entries one at a time against a high-water mark. Grouping
+    # them into runs first was wrong twice over: a run that began with a repeat
+    # but then ran forward got classified as a repeat *whole*, and the high-water
+    # mark wasn't advanced on that branch, so the tail re-emitted the entire
+    # sheet — 16 lines came out as 28.
+    emitted = -1
+    for e in matched:
+        j = e["sheetIdx"]
+        if j > emitted:
+            # Forward progress. Emit any sheet lines the transcript skipped over
+            # (it has poor recall) so they are never lost, then this one.
+            for k in range(emitted + 1, j):
+                emit(k, False)
+            emit(j, False, e)
+            emitted = j
+        else:
+            # Already past this line, so the recording is singing it again.
+            emit(j, True, e)
+    for j in range(emitted + 1, len(sheet)):   # tail the transcript never reached
+        emit(j, False)
+    return lines, order
+
+
+def coverage(sheet: List[str], order: List[dict]) -> dict:
+    """How much of the sheet the performance used, and how much it repeated."""
+    return {
+        "sheetLines": len(sheet),
+        "performanceLines": len(order),
+        "repeatsInserted": sum(1 for o in order if o.get("repeat")),
+        "linesWithEvidence": sum(1 for o in order if o.get("startMs") is not None),
+    }
+
+
+def resolve_with_llm(sheet: List[str], hyp_lines: List[dict], call) -> List[str]:
+    """Optional escape hatch for sheets the matcher can't resolve.
+
+    `call(prompt) -> str` is supplied by the caller so this module stays free of
+    any SDK or API key. Only worth reaching for when `coverage()` looks wrong —
+    a sheet in the wrong order, interleaved with a translation, or carrying a
+    second song. For the ordinary "chorus written once, sung twice" case the
+    deterministic path above is cheaper, faster and does not invent lines.
+
+    The model is asked to *reorder and repeat the given lines only*; any line it
+    returns that is not in the sheet is dropped, because an LLM inventing lyrics
+    is the one failure this whole pipeline exists to avoid.
+    """
+    numbered = "\n".join(f"{i}: {l}" for i, l in enumerate(sheet))
+    heard = "\n".join(f"{h['startMs']/1000:.1f}s: {h['text']}" for h in hyp_lines)
+    prompt = (
+        "A lyric sheet is written in canonical form (chorus once). A rough "
+        "machine transcript shows what the recording actually sings, in order, "
+        "with timings. The transcript has many wrong words — trust it only for "
+        "ORDER and REPETITION.\n\n"
+        f"SHEET (numbered):\n{numbered}\n\nTRANSCRIPT:\n{heard}\n\n"
+        "Output the sheet line numbers in the order they are actually sung, one "
+        "per line, repeating a number when its line is sung again. Output "
+        "nothing but numbers."
+    )
+    raw = call(prompt)
+    out = []
+    for tok in re.findall(r"\d+", raw or ""):
+        i = int(tok)
+        if 0 <= i < len(sheet):          # never accept a line not in the sheet
+            out.append(sheet[i])
+    return out