Handle short hums with F0 fallback and initial beat anchor

README.md

@@ -29,6 +29,12 @@ follows voice pitch and amplitude. The sampler uses full-size FreePats electric
 guitar and fingered bass SoundFonts, with FFmpeg compression, filtering,
 oversampled soft clipping, and a short guitar room echo in processed versions.
 
+If SheetSage2 returns an empty vocal melody, a conservative pYIN fallback
+extracts source-timed notes from sustained voiced pitch and supplies a simple
+instrumental ABC for YuE2. It rejects unpitched noise. Fallback note boundaries
+are approximate, and the run details identify which transcription was used.
+SheetSage2's beat grid remains the source for sampled drums when available.
+
 YuE2 is on by default, so the same click also starts three experiments from
 SheetSage2 ABC: solo, melody-with-drums, and composition. Uncheck its option
 when only the faster sampler previews are wanted. The drums in these

f0_fallback.py

@@ -0,0 +1,64 @@
+"""Source-timed monophonic fallback for a hum SheetSage cannot notate."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+
+def transcribe_f0(source: Path, target: Path) -> int:
+    """Write MIDI from voiced F0 spans; reject inputs without a sustained pitch."""
+    import librosa
+    import numpy as np
+    import pretty_midi
+    from scipy.ndimage import binary_closing, median_filter
+    import soundfile as sf
+
+    audio, rate = sf.read(source, dtype="float32")
+    if audio.ndim != 1 or rate != 24000:
+        raise ValueError("Для резервной транскрипции нужен моно PCM 24 кГц")
+    reduced = librosa.resample(audio, orig_sr=rate, target_sr=16000)
+    f0, voiced, _ = librosa.pyin(reduced, sr=16000, fmin=55., fmax=1000.,
+                                 frame_length=1024, hop_length=160, center=True)
+    rms = librosa.feature.rms(y=reduced, frame_length=1024,
+                              hop_length=160, center=True)[0][:len(f0)]
+    threshold = max(.001, float(np.percentile(rms, 95)) * 10 ** (-35 / 20))
+    active = voiced & np.isfinite(f0) & (rms >= threshold)
+    # Bridge only pitch-tracker dropouts shorter than a consonant or breath.
+    active = binary_closing(active, structure=np.ones(4), border_value=0)
+    boundaries = np.flatnonzero(np.diff(np.r_[False, active, False]))
+    score = pretty_midi.PrettyMIDI(initial_tempo=120)
+    track = pretty_midi.Instrument(program=0, name="F0 fallback")
+    for first, last in zip(boundaries[::2], boundaries[1::2]):
+        if last - first < 18:  # Reject short accidental pitched patches in noise.
+            continue
+        pitch = median_filter(69 + 12 * np.log2(np.maximum(f0[first:last], 55.) / 440),
+                              size=9, mode="nearest")
+        # pYIN may be unvoiced for a few interpolated frames.
+        good = np.isfinite(pitch)
+        if not good.any():
+            continue
+        pitch = np.interp(np.arange(len(pitch)), np.flatnonzero(good), pitch[good])
+        rounded = median_filter(np.rint(pitch), size=11, mode="nearest").astype(int)
+        changes = np.r_[0, np.flatnonzero(np.diff(rounded)) + 1, len(rounded)]
+        cuts = [0]
+        for index in range(1, len(changes) - 1):
+            point, next_point = int(changes[index]), int(changes[index + 1])
+            if (point - cuts[-1] >= 12 and next_point - point >= 12
+                    and abs(np.median(pitch[cuts[-1]:point])
+                            - np.median(pitch[point:next_point])) >= 1.5):
+                cuts.append(point)
+        cuts.append(len(rounded))
+        for start, end in zip(cuts, cuts[1:]):
+            if end - start < 8:
+                continue
+            begin = max(0., (first + start) * .01 - .005)
+            finish = min(len(audio) / rate, (first + end) * .01 - .005)
+            if finish - begin < .075:
+                continue
+            note = int(np.clip(np.rint(np.median(pitch[start:end])), 0, 127))
+            track.notes.append(pretty_midi.Note(90, note, begin, finish))
+    if not track.notes:
+        raise ValueError("Не удалось выделить устойчивую высоту напева")
+    score.instruments.append(track)
+    score.write(str(target))
+    return len(track.notes)

logic.py

@@ -28,7 +28,12 @@ def drum_hits(beat_lab: str, melody_spans: list[tuple[float, float]],
               duration: float) -> list[tuple[float, int, int]]:
     """Place a small GM kit on SheetSage's source-time beat grid."""
     beats = []
-    for line in beat_lab.splitlines():
+    lines = beat_lab.splitlines()
+    if (len(lines) > 1 and lines[0].split() == ["0.0", "1", "1", "8"]
+            and lines[1].split()[1:2] == ["1"]):
+        # SheetSage sometimes exports a 1/8 time-zero anchor before its real grid.
+        lines = lines[1:]
+    for line in lines:
         fields = line.split()
         if len(fields) != 4:
             raise ValueError("SheetSage2 вернул некорректную сетку долей")
@@ -81,6 +86,8 @@ def drum_hits(beat_lab: str, melody_spans: list[tuple[float, float]],
 
 def instrumental_abc(native: str, keep_chords: bool = False) -> str:
     """Move the native Vocal melody into Ins, failing on unknown ABC layout."""
+    if "V: Vocal" not in native and 'V: Ins clef=treble name="Instrumental Melody"' in native:
+        return native
     lines = native.splitlines()
     header = []
     body = []
@@ -123,6 +130,54 @@ def instrumental_abc(native: str, keep_chords: bool = False) -> str:
     return result
 
 
+def notes_to_abc(notes: list[tuple[float, float, int]], duration: float) -> str:
+    """Make a coarse instrumental ABC solely for YuE when SheetSage has no notes."""
+    if not notes:
+        raise ValueError("ABC requires notes")
+    unit = .0625  # A 1/32 ABC note at 120 quarter-notes per minute.
+    events = []
+    previous = 0
+    for start, end, pitch in sorted(notes):
+        first = max(previous, round(start / unit))
+        last = max(first + 1, round(end / unit))
+        events.append((first, last, pitch))
+        previous = last
+    total = max(32, round(duration / unit), events[-1][1])
+    names = ("C", "^C", "D", "^D", "E", "F", "^F", "G", "^G", "A", "^A", "B")
+
+    def name(pitch: int) -> str:
+        octave = pitch // 12 - 1
+        base = names[pitch % 12]
+        if octave >= 5:
+            return base.lower() + "'" * (octave - 5)
+        return base + "," * (4 - octave)
+
+    body = []
+    cursor = 0
+    for first, last, pitch in events:
+        for event_start, event_end, symbol in ((cursor, first, "z"),
+                                               (first, last, name(pitch))):
+            while event_start < event_end:
+                chunk = min(event_end - event_start, 32 - event_start % 32)
+                more = symbol != "z" and event_start + chunk < event_end
+                body.append(f"{symbol}{chunk}{'-' if more else ''}")
+                event_start += chunk
+                if event_start % 32 == 0:
+                    body.append("|")
+        cursor = last
+    while cursor < total:
+        chunk = min(total - cursor, 32 - cursor % 32)
+        body.append(f"z{chunk}")
+        cursor += chunk
+        if cursor % 32 == 0:
+            body.append("|")
+    if body[-1] != "|":
+        body.append("|")
+    return ('X:1\nT:Hum F0 fallback\nM:4/4\nL:1/32\nQ:1/4=120\n'
+            'V: Ins clef=treble name="Instrumental Melody" snm="Ins"\n'
+            'K:C\nV: Ins\n' + " ".join(body) + "\n")
+
+
 def validate_duration(seconds: float) -> None:
     if not 0.4 <= seconds <= MAX_SECONDS:
         raise ValueError(f"Нужен напев от 0,4 до {MAX_SECONDS:g} секунд; сейчас {seconds:.1f} с.")

pipeline.py

@@ -14,7 +14,7 @@ from typing import Iterator
 from datetime import datetime, timezone
 from uuid import uuid4
 
-from logic import INSTRUMENTS, drum_hits, instrumental_abc, validate_duration
+from logic import INSTRUMENTS, drum_hits, instrumental_abc, notes_to_abc, validate_duration
 
 
 ROOT = Path(__file__).resolve().parent
@@ -387,6 +387,7 @@ def generate_all(source: str, instrument: str, styles: dict[str, str], seed: int
     drums_midi = None
     bass_shift = None
     bass_source = None
+    transcription_source = "SheetSage2"
     archive = {"requested": save_recording, "saved": False}
 
     def deliver(name: str, raw: Path, *, normalize: bool = True,
@@ -405,6 +406,7 @@ def generate_all(source: str, instrument: str, styles: dict[str, str], seed: int
                     "source_timing_preserved": list(SAMPLER_VARIANTS),
                     "bass_transpose_semitones": bass_shift,
                     "bass_source": bass_source,
+                    "transcription_source": transcription_source,
                     "recording_archive": dict(archive),
                     "yue_vocal_absence_guaranteed": False,
                     "elapsed_seconds": round(time.monotonic() - began, 2)}
@@ -423,7 +425,27 @@ def generate_all(source: str, instrument: str, styles: dict[str, str], seed: int
             errors["recording_archive"] = str(error)
     if progress:
         progress(.12, desc="SheetSage2 извлекает мелодию и сетку долей")
-    midi, melody_abc, full_abc, beat_lab = transcribe(pcm24, folder, log)
+    try:
+        midi, melody_abc, full_abc, beat_lab = transcribe(pcm24, folder, log)
+    except RuntimeError as error:
+        if "could not export ABC" not in str(error):
+            raise
+        errors["sheet_sage"] = "SheetSage2 не смог записать ноты в ABC"
+        midi, melody_abc, full_abc, beat_lab = (
+            folder / "notes.mid", folder / "melody.abc", folder / "full.abc", folder / "beat.lab")
+    import pretty_midi
+    if not midi.is_file() or not any(track.notes for track in
+                                     pretty_midi.PrettyMIDI(str(midi)).instruments):
+        from f0_fallback import transcribe_f0
+        transcribe_f0(pcm24, midi)
+        notes = [(note.start, note.end, note.pitch)
+                 for track in pretty_midi.PrettyMIDI(str(midi)).instruments
+                 for note in track.notes]
+        abc = notes_to_abc(notes, duration)
+        melody_abc.write_text(abc)
+        full_abc.write_text(abc)
+        transcription_source = "pYIN fallback"
+        errors["sheet_sage"] = "SheetSage2 не нашёл нот; использован резервный F0"
     bass_shift, bass_font, bass_program, bass_source = choose_bass_route(midi)
     if progress:
         progress(.30, desc="Синтезирую инструмент и барабаны")

test_f0_fallback.py

@@ -0,0 +1,40 @@
+"""Source-time fallback contracts on a known hum-like signal and plain noise."""
+
+from __future__ import annotations
+
+from pathlib import Path
+import tempfile
+import unittest
+
+
+class F0FallbackTest(unittest.TestCase):
+    def test_short_stable_note_survives_sheet_sage_empty_melody(self):
+        import numpy as np
+        import pretty_midi
+        import soundfile as sf
+        from f0_fallback import transcribe_f0
+
+        rate = 24000
+        wave = np.zeros(round(.6 * rate), dtype="float32")
+        t = np.arange(round(.48 * rate)) / rate
+        wave[round(.08 * rate):round(.56 * rate)] = .12 * np.sin(2 * np.pi * 261.626 * t)
+        with tempfile.TemporaryDirectory() as directory:
+            source, target = (Path(directory) / name for name in ("hum.wav", "notes.mid"))
+            sf.write(source, wave, rate)
+            self.assertEqual(transcribe_f0(source, target), 1)
+            note = pretty_midi.PrettyMIDI(str(target)).instruments[0].notes[0]
+            self.assertEqual(note.pitch, 60)
+            self.assertAlmostEqual(note.start, .08, delta=.05)
+            self.assertAlmostEqual(note.end, .56, delta=.05)
+
+    def test_unpitched_noise_does_not_create_midi(self):
+        import numpy as np
+        import soundfile as sf
+        from f0_fallback import transcribe_f0
+
+        with tempfile.TemporaryDirectory() as directory:
+            source, target = (Path(directory) / name for name in ("noise.wav", "notes.mid"))
+            sf.write(source, np.random.default_rng(20260926).normal(0, .02, 72000), 24000)
+            with self.assertRaisesRegex(ValueError, "устойчивую высоту"):
+                transcribe_f0(source, target)
+            self.assertFalse(target.exists())

test_logic.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import unittest
 
-from logic import drum_hits, instrumental_abc, validate_duration
+from logic import drum_hits, instrumental_abc, notes_to_abc, validate_duration
 
 
 NATIVE = '''X:1
@@ -44,6 +44,15 @@ class LogicTest(unittest.TestCase):
         with self.assertRaises(ValueError):
             instrumental_abc(NATIVE.replace("Z1|", "z16|", 1))
 
+    def test_fallback_abc_is_instrumental_and_has_source_order(self):
+        abc = notes_to_abc([(.08, .52, 60), (.68, 1.1, 60),
+                            (1.3, 1.7, 62)], 2.)
+        self.assertEqual(instrumental_abc(abc), abc)
+        self.assertNotIn("Vocal", abc)
+        self.assertIn("V: Ins", abc)
+        self.assertIn("C7", abc)
+        self.assertIn("D6", abc)
+
     def test_duration_bounds(self):
         validate_duration(0.4)
         validate_duration(35)
@@ -74,6 +83,14 @@ class LogicTest(unittest.TestCase):
         self.assertFalse(any(time == 0 for time, _, _ in hits))
         self.assertIn((.44, 36, 90), hits)
 
+    def test_drums_ignore_sheet_sage_initial_one_eighth_anchor(self):
+        # Seen in HumTrans F05/M01: 1/8 at zero precedes a real 3/4 or 4/4 grid.
+        beats = ("0.0\t1\t1\t8\n.19\t1\t3\t4\n.67\t2\t3\t4\n"
+                 "1.16\t3\t3\t4\n1.67\t1\t3\t4\n")
+        hits = drum_hits(beats, [(.2, 1.8)], 2.0)
+        self.assertFalse(any(time == 0 for time, _, _ in hits))
+        self.assertIn((.19, 36, 90), hits)
+
 
 if __name__ == "__main__":
     unittest.main()

test_sampler.py

@@ -10,6 +10,43 @@ from unittest.mock import patch
 
 
 class SamplerTest(unittest.TestCase):
+    def test_empty_sheet_sage_melody_uses_f0_for_sampler(self):
+        import numpy as np
+        import pretty_midi
+        import soundfile as sf
+        from pipeline import generate_all
+
+        with tempfile.TemporaryDirectory() as directory:
+            folder = Path(directory)
+            wave = folder / "hum.wav"
+            t = np.arange(24000) / 24000
+            sf.write(wave, .12 * np.sin(2 * np.pi * 261.626 * t), 24000)
+            empty, abc, beat = (folder / name for name in
+                                ("empty.mid", "empty.abc", "beat.lab"))
+            pretty_midi.PrettyMIDI().write(str(empty))
+            abc.write_text("unusable empty melody")
+            beat.write_text("0 1 4 4\n.5 2 4 4\n1 3 4 4\n")
+
+            def fake_mp3(_, target, __, **___):
+                target.write_bytes(b"fake mp3")
+
+            with (patch("pipeline.decode", return_value=(wave, wave, 1.)),
+                  patch("pipeline.transcribe", return_value=(empty, abc, abc, beat)),
+                  patch("pipeline.choose_bass_route", return_value=(-24, folder / "bass.sf2", 0, "test")),
+                  patch("pipeline.render_instrument", return_value=(wave, empty)),
+                  patch("pipeline.render_drums", return_value=(wave, empty, 3)),
+                  patch("pipeline.pair_with_drums", return_value=(wave, wave)),
+                  patch("pipeline.render_contour", return_value=wave),
+                  patch("pipeline.render_preset", return_value=(wave, empty)),
+                  patch("pipeline.apply_effect", return_value=wave),
+                  patch("pipeline.encode_mp3", side_effect=fake_mp3)):
+                result = list(generate_all(str(wave), "Флейта", {}, 4096, False))[-1]
+            self.assertTrue(result[-1]["complete"])
+            self.assertEqual(result[-1]["transcription_source"], "pYIN fallback")
+            self.assertIn("SheetSage2 не нашёл нот", result[-1]["errors"]["sheet_sage"])
+            self.assertEqual(len(pretty_midi.PrettyMIDI(str(empty)).instruments[0].notes), 1)
+            self.assertNotIn("Vocal", abc.read_text())
+
     def test_archive_keeps_original_bytes_and_uses_opaque_name(self):
         from pipeline import archive_recording, source_sha256
 
@@ -62,6 +99,7 @@ class SamplerTest(unittest.TestCase):
 
     def test_optional_yue_does_not_reclassify_sampler_outputs_as_missing(self):
         import numpy as np
+        import pretty_midi
         import soundfile as sf
         from pipeline import generate_all
 
@@ -75,7 +113,11 @@ class SamplerTest(unittest.TestCase):
             sf.write(wave, np.full(48000, .1, dtype="float32"), 48000)
             midi, abc, beat = (folder / name for name in
                                ("notes.mid", "melody.abc", "beat.lab"))
-            midi.write_bytes(b"fixture")
+            score = pretty_midi.PrettyMIDI()
+            track = pretty_midi.Instrument(0)
+            track.notes.append(pretty_midi.Note(90, 60, .1, .8))
+            score.instruments.append(track)
+            score.write(str(midi))
             abc.write_text(abc_text)
             beat.write_text("0 1 4 4\n.5 2 4 4\n1 3 4 4\n")