feat: note generation from alignment + pitch

src/karaoke_maker/notes.py

@@ -0,0 +1,147 @@
+from dataclasses import dataclass
+
+import numpy as np
+import pyphen
+
+CONF_MASK = 0.5          # pitch frames below this are unvoiced
+LOW_SCORE = 0.5          # word alignment score below this -> low_confidence
+SPLIT_SEMITONES = 0.8
+MIN_SEG_SEC = 0.05
+MIN_NOTE_SEC = 0.05      # alignment may hand us zero-length words
+HOP_SEC = 0.01           # pitch stage hop
+
+_PYPHEN = {"english": "en_US", "german": "de_DE", "spanish": "es",
+           "french": "fr", "italian": "it", "portuguese": "pt",
+           "dutch": "nl_NL"}
+_SYLL_SEP = "\0"         # sentinel so literal hyphens survive syllabification
+
+
+@dataclass
+class TimedNote:
+    start: float
+    duration: float
+    pitch: int            # UltraStar semitone (0 = C4 = MIDI 60)
+    text: str             # display text incl. leading space / "~"
+    low_confidence: bool
+
+
+def _midi(hz):
+    return 69 + 12 * np.log2(hz / 440.0)
+
+
+def hz_to_midi_float(hz: float) -> float:
+    return float(_midi(hz))
+
+
+def hz_to_ultrastar(hz: float) -> int:
+    return round(hz_to_midi_float(hz)) - 60
+
+
+def segment_span(times, f0, conf, start: float, end: float
+                 ) -> list[tuple[float, float, float]]:
+    """Voiced pitch segments within [start, end) as (start, end, median_midi)."""
+    mask = (times >= start) & (times < end) & (conf >= CONF_MASK) & (f0 > 0)
+    t = times[mask]
+    if len(t) == 0:
+        return []
+    midi = _midi(f0[mask])
+    segs: list[tuple[float, float, float]] = []
+
+    def emit(a: int, b: int) -> None:
+        if float(t[b - 1]) - float(t[a]) + HOP_SEC >= MIN_SEG_SEC:
+            segs.append((float(t[a]), float(t[b - 1]) + HOP_SEC,
+                         float(np.median(midi[a:b]))))
+
+    seg_start, dev_start = 0, None
+    for i in range(1, len(t)):
+        stable_end = i if dev_start is None else dev_start
+        if abs(midi[i] - np.median(midi[seg_start:stable_end])) >= SPLIT_SEMITONES:
+            # only split once the deviation has held for MIN_SEG_SEC, so that
+            # short tracker glitches don't fragment a sustained note
+            if dev_start is None:
+                dev_start = i
+            elif t[i] - t[dev_start] + HOP_SEC >= MIN_SEG_SEC:
+                emit(seg_start, dev_start)
+                seg_start, dev_start = dev_start, None
+        else:
+            dev_start = None
+    emit(seg_start, len(t))
+    return segs
+
+
+def syllabify(word: str, language: str | None) -> list[str]:
+    code = _PYPHEN.get((language or "english").lower(), "en_US")
+    try:
+        dic = pyphen.Pyphen(lang=code)
+    except KeyError:
+        return [word]
+    parts = [p for p in dic.inserted(word, hyphen=_SYLL_SEP).split(_SYLL_SEP) if p]
+    return parts or [word]
+
+
+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 = None, 0.0
+    for row in rows:
+        if prev_line is not None and row["line_index"] > prev_line:
+            breaks.append((prev_end + row["start"]) / 2)
+        line_initial = row["line_index"] != prev_line
+        prev_line, prev_end = row["line_index"], row["end"]
+        segs = segment_span(times, f0, conf, row["start"], row["end"])
+        if not segs:
+            pitch = notes[-1].pitch if notes else 0
+            notes.append(_note(row["start"], row["end"] - row["start"], pitch,
+                               _text(row["word"], 0, line_initial), True))
+            continue
+        low = row["score"] < LOW_SCORE
+        sylls = syllabify(row["word"], language)
+        if len(segs) < len(sylls):
+            segs = _subdivide(segs, len(sylls))
+        for i, (s, e, midi) in enumerate(segs):
+            text = _text(sylls[i], i, line_initial) if i < len(sylls) else "~"
+            notes.append(_note(s, e - s, round(midi) - 60, text, low))
+    return notes, breaks, _warnings(notes)
+
+
+def _note(start: float, duration: float, pitch: int, text: str,
+          low: bool) -> TimedNote:
+    return TimedNote(float(start), max(float(duration), MIN_NOTE_SEC),
+                     int(pitch), text, low)
+
+
+def _text(syllable: str, index: int, line_initial: bool) -> str:
+    return syllable if index or line_initial else " " + syllable
+
+
+def _subdivide(segs, target: int):
+    """Cut segments into equal slices until there is one per syllable."""
+    parts = [1] * len(segs)
+    while sum(parts) < target:
+        i = max(range(len(segs)),
+                key=lambda j: (segs[j][1] - segs[j][0]) / parts[j])
+        parts[i] += 1
+    out = []
+    for (s, e, midi), n in zip(segs, parts):
+        step = (e - s) / n
+        out.extend((s + k * step, s + (k + 1) * step, midi) for k in range(n))
+    return out
+
+
+def _fmt(t: float) -> str:
+    return f"{int(t // 60)}:{int(t % 60):02d}"
+
+
+def _warnings(notes: list[TimedNote]) -> list[str]:
+    warnings, run = [], []
+    for n in list(notes) + [None]:
+        if n is not None and n.low_confidence:
+            run.append(n)
+        elif run:
+            first, last = run[0], run[-1]
+            warnings.append(
+                f"{len(run)} low-confidence notes near "
+                f"{_fmt(first.start)}–{_fmt(last.start + last.duration)}")
+            run = []
+    return warnings

src/karaoke_maker/stages/__init__.py

@@ -1,10 +1,12 @@
 from ..pipeline import Stage
 from .acquire import acquire_stage
 from .align import align_stage
+from .notes import notes_stage
 from .pitch import pitch_stage
 from .separate import separate_stage
 from .transcribe import transcribe_stage
 
 
 def all_stages() -> list[Stage]:
-    return [acquire_stage(), separate_stage(), transcribe_stage(), align_stage(), pitch_stage()]
+    return [acquire_stage(), separate_stage(), transcribe_stage(), align_stage(),
+            pitch_stage(), notes_stage()]

src/karaoke_maker/stages/notes.py

@@ -0,0 +1,25 @@
+import json
+from dataclasses import asdict
+
+import numpy as np
+
+from ..notes import build_notes
+from ..pipeline import Ctx, Stage
+
+
+def _run(ctx: Ctx) -> None:
+    rows = json.loads((ctx.work / "alignment.json").read_text())
+    data = np.load(ctx.work / "pitch.npz")
+    notes, breaks, warnings = build_notes(
+        rows, data["times"], data["f0"], data["conf"], ctx.inp.language)
+    (ctx.work / "notes_sec.json").write_text(json.dumps({
+        "notes": [asdict(n) for n in notes],
+        "breaks": breaks,
+        "warnings": warnings,
+    }))
+
+
+def notes_stage() -> Stage:
+    return Stage(name="notes",
+                 outputs=lambda ctx: [ctx.work / "notes_sec.json"],
+                 run=_run)

tests/test_notes.py

@@ -0,0 +1,220 @@
+import json
+
+import numpy as np
+from karaoke_maker.config import Config
+from karaoke_maker.notes import (
+    MIN_NOTE_SEC, TimedNote, build_notes, hz_to_ultrastar, segment_span,
+    syllabify,
+)
+from karaoke_maker.pipeline import Ctx, SongInput
+from karaoke_maker.stages import all_stages, notes as notes_stage_mod
+
+
+def flat_pitch(duration=10.0, hz=261.63):   # C4
+    times = np.arange(0, duration, 0.01)
+    return times, np.full_like(times, hz), np.full_like(times, 0.9)
+
+
+def test_hz_to_ultrastar():
+    assert hz_to_ultrastar(261.63) == 0        # C4
+    assert hz_to_ultrastar(440.0) == 9         # A4
+    assert hz_to_ultrastar(220.0) == -3        # A3
+
+
+def test_segment_span_single_flat():
+    times, f0, conf = flat_pitch()
+    segs = segment_span(times, f0, conf, 1.0, 2.0)
+    assert len(segs) == 1
+    s, e, midi = segs[0]
+    assert abs(midi - 60.0) < 0.1
+    assert abs(s - 1.0) < 0.02 and abs(e - 2.0) < 0.02
+
+
+def test_segment_span_splits_on_jump():
+    times = np.arange(0, 2.0, 0.01)
+    f0 = np.where(times < 1.0, 261.63, 329.63)   # C4 -> E4
+    conf = np.full_like(times, 0.9)
+    segs = segment_span(times, f0, conf, 0.0, 2.0)
+    assert len(segs) == 2
+    assert abs(segs[0][2] - 60.0) < 0.2 and abs(segs[1][2] - 64.0) < 0.2
+    assert abs(segs[0][1] - segs[1][0]) < 1e-6      # contiguous
+
+
+def test_segment_span_masks_low_conf():
+    times, f0, conf = flat_pitch()
+    conf[:] = 0.1
+    assert segment_span(times, f0, conf, 0.0, 1.0) == []
+
+
+def test_segment_span_ignores_brief_pitch_glitch():
+    times, f0, conf = flat_pitch(duration=1.0)
+    f0[50:52] = 523.25                          # 2-frame octave blip
+    conf = np.full_like(times, 0.9)
+    segs = segment_span(times, f0, conf, 0.0, 1.0)
+    assert len(segs) == 1
+
+
+def test_segment_span_drops_sub_minimum_segments():
+    times, f0, conf = flat_pitch(duration=1.0)
+    assert segment_span(times, f0, conf, 0.5, 0.52) == []
+
+
+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"]
+
+
+def test_syllabify_preserves_word_characters():
+    assert "".join(syllabify("well-known", "English")) == "well-known"
+    assert "".join(syllabify("beautiful", "English")) == "beautiful"
+
+
+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": "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[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
+    assert warnings == []
+
+
+def test_build_notes_melisma_continuation():
+    times = np.arange(0, 3.0, 0.01)
+    f0 = np.where(times < 1.0, 261.63, 329.63)
+    conf = np.full_like(times, 0.9)
+    rows = [{"word": "mine", "start": 0.2, "end": 2.8, "score": 0.9, "line_index": 0}]
+    notes, _, _ = build_notes(rows, times, f0, conf, "English")
+    assert [n.text for n in notes] == ["mine", "~"]
+    assert notes[0].pitch != notes[1].pitch
+
+
+def test_build_notes_unvoiced_word_flagged():
+    times, f0, conf = flat_pitch()
+    conf[:] = 0.0
+    rows = [{"word": "hey", "start": 1.0, "end": 1.4, "score": 0.9, "line_index": 0}]
+    notes, _, warnings = build_notes(rows, times, f0, conf, "English")
+    assert notes[0].low_confidence and len(warnings) == 1
+    assert warnings[0] == "1 low-confidence notes near 0:01–0:01"
+
+
+def test_build_notes_low_score_word_flagged():
+    times, f0, conf = flat_pitch()
+    rows = [{"word": "hey", "start": 1.0, "end": 1.4, "score": 0.1, "line_index": 0}]
+    notes, _, warnings = build_notes(rows, times, f0, conf, "English")
+    assert all(n.low_confidence for n in notes) and len(warnings) == 1
+
+
+def test_build_notes_unvoiced_word_inherits_previous_pitch():
+    times, f0, conf = flat_pitch()
+    conf[300:] = 0.0
+    rows = [
+        {"word": "one", "start": 1.0, "end": 2.0, "score": 0.9, "line_index": 0},
+        {"word": "two", "start": 4.0, "end": 4.5, "score": 0.9, "line_index": 0},
+    ]
+    notes, _, _ = build_notes(rows, times, f0, conf, "English")
+    assert notes[1].pitch == notes[0].pitch == 0
+    assert notes[1].low_confidence and not notes[0].low_confidence
+
+
+def test_build_notes_unvoiced_first_word_uses_zero_pitch():
+    times, f0, conf = flat_pitch()
+    conf[:] = 0.0
+    rows = [{"word": "hey", "start": 1.0, "end": 1.4, "score": 0.9, "line_index": 0}]
+    notes, _, _ = build_notes(rows, times, f0, conf, "English")
+    assert notes[0].pitch == 0
+
+
+def test_build_notes_multisyllable_word_emits_syllables_only():
+    times, f0, conf = flat_pitch()
+    rows = [
+        {"word": "reminds", "start": 1.0, "end": 1.8, "score": 0.9, "line_index": 0},
+        {"word": "reminds", "start": 2.0, "end": 2.8, "score": 0.9, "line_index": 0},
+    ]
+    notes, _, _ = build_notes(rows, times, f0, conf, "English")
+    assert [n.text for n in notes] == ["re", "minds", " re", "minds"]
+
+
+def test_build_notes_more_syllables_than_segments():
+    times, f0, conf = flat_pitch()
+    rows = [{"word": "beautiful", "start": 1.0, "end": 1.9, "score": 0.9,
+             "line_index": 0}]
+    notes, _, _ = build_notes(rows, times, f0, conf, "English")
+    assert [n.text for n in notes] == ["beau", "ti", "ful"]
+    assert all(abs(n.duration - notes[0].duration) < 1e-6 for n in notes)
+    for a, b in zip(notes, notes[1:]):
+        assert abs(a.start + a.duration - b.start) < 1e-6
+
+
+def test_build_notes_zero_duration_row_gets_minimum_length():
+    times, f0, conf = flat_pitch()
+    rows = [
+        {"word": "...", "start": 1.0, "end": 1.0, "score": 0.0, "line_index": 0},
+        {"word": "go", "start": 1.0, "end": 1.5, "score": 0.9, "line_index": 0},
+    ]
+    notes, _, _ = build_notes(rows, times, f0, conf, "English")
+    assert notes[0].duration >= MIN_NOTE_SEC
+    assert all(n.duration > 0 for n in notes)
+
+
+def test_build_notes_starts_are_non_decreasing():
+    times = np.arange(0, 12.0, 0.01)
+    f0 = 220.0 * 2 ** (np.floor(times * 2) % 5 / 12.0)   # stepped melody
+    conf = np.where((times % 3.0) < 2.6, 0.9, 0.1)       # unvoiced gaps
+    words = ["We", "count", "a", "beautiful", "number", "that",
+             "reminds", "us", "of", "everything", "we", "list"]
+    rows = []
+    for i, w in enumerate(words):
+        rows.append({"word": w, "start": 0.5 + i * 0.9, "end": 0.5 + i * 0.9 + 0.8,
+                     "score": 0.9 if i % 5 else 0.2, "line_index": i // 4})
+    notes, breaks, warnings = build_notes(rows, times, f0, conf, "English")
+    assert len(notes) >= len(words)
+    starts = [n.start for n in notes]
+    assert starts == sorted(starts)
+    assert breaks == sorted(breaks) and len(breaks) == 2
+    assert all(n.duration > 0 for n in notes)
+    assert warnings and all("low-confidence notes near" in w for w in warnings)
+
+
+def test_build_notes_line_initial_after_break_has_no_space():
+    times, f0, conf = flat_pitch()
+    rows = [
+        {"word": "one", "start": 1.0, "end": 1.4, "score": 0.9, "line_index": 0},
+        {"word": "two", "start": 2.0, "end": 2.4, "score": 0.9, "line_index": 1},
+        {"word": "three", "start": 3.0, "end": 3.4, "score": 0.9, "line_index": 1},
+    ]
+    notes, breaks, _ = build_notes(rows, times, f0, conf, "English")
+    assert [n.text for n in notes] == ["one", "two", " three"]
+    assert breaks == [1.7]
+
+
+def test_notes_stage_writes_json(tmp_path):
+    work = tmp_path / "w"
+    work.mkdir()
+    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": "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"),
+              cfg=Config(whisper_url="u"), work=work, log=lambda m: None)
+    stage = notes_stage_mod.notes_stage()
+    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 data["breaks"] == [2.25]
+    assert data["warnings"] == []
+    assert TimedNote(**data["notes"][0]).pitch == 0
+
+
+def test_notes_stage_registered():
+    assert [s.name for s in all_stages()][-1] == "notes"