Add transcribe.py (app pack entry point)

transcribe.py

@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+"""Egystormy Tools — portable local MuScriptor transcription pack.
+
+Usage:
+    runtime/python transcribe.py <input_audio> <output.mid>
+
+Fully offline and self-contained: the interpreter is ./runtime, all
+third-party libraries live in ./libs, and the model weights in ./weights.
+No Python installation, no setup, no network, no account.
+
+Design note — no beat detection, on purpose: this entry point uses
+``TranscriptionModel.transcribe`` + ``events_to_midi_bytes`` directly and
+never calls the beat-grid path (``transcribe_and_postprocess`` /
+``detect_tempo``). MuScriptor's beat detection hard-imports
+beat_this -> torchaudio at call time, and a mismatched torchaudio build
+kills the whole transcription. Skipping it costs nothing that matters
+here: note timings stay true (seconds-accurate); the MIDI simply carries
+no tempo map. (In this pack torch is 2.15.0+cpu while the newest
+torchaudio on the CPU index is 2.11.1+cpu — no matched 2.15 pair exists,
+which is exactly the hazard avoided.)
+"""
+
+import os
+import sys
+import time
+from pathlib import Path
+
+PACK = Path(__file__).resolve().parent
+sys.path.insert(0, str(PACK / "libs"))
+# Hard-offline: never let a library try to phone home.
+os.environ.setdefault("HF_HUB_OFFLINE", "1")
+os.environ.setdefault("TRANSFORMERS_OFFLINE", "1")
+
+T0 = time.perf_counter()
+
+
+def stamp(label: str) -> None:
+    print(f"[pack] {label}: {time.perf_counter() - T0:.2f}s elapsed", flush=True)
+
+
+def main() -> int:
+    if len(sys.argv) != 3:
+        print("usage: runtime/python transcribe.py <input_audio> <output.mid>",
+              file=sys.stderr)
+        return 2
+    src = Path(sys.argv[1])
+    dst = Path(sys.argv[2])
+    weights = PACK / "weights" / "model.safetensors"
+    if not src.exists():
+        print(f"input audio not found: {src}", file=sys.stderr)
+        return 2
+    if not weights.exists():
+        print(f"weights not found: {weights}", file=sys.stderr)
+        return 2
+
+    from muscriptor import TranscriptionModel
+    from muscriptor.events import NoteStartEvent, ProgressEvent
+
+    stamp("imports done")
+
+    # libsndfile cannot decode m4a/AAC and some other containers; if an
+    # ffmpeg happens to be available, transcode those to a temp WAV first.
+    # (mp3/wav/flac/ogg decode natively and never reach this branch.)
+    audio_arg = src
+    tmp_wav = None
+    if src.suffix.lower() in {".m4a", ".aac", ".opus", ".wma"}:
+        import shutil
+        import subprocess
+        import tempfile
+
+        ffmpeg = shutil.which("ffmpeg")
+        if ffmpeg:
+            fd, tmp_wav = tempfile.mkstemp(suffix=".wav")
+            os.close(fd)
+            subprocess.run(
+                [ffmpeg, "-y", "-v", "error", "-i", str(src), tmp_wav],
+                check=True,
+            )
+            audio_arg = Path(tmp_wav)
+
+    model = TranscriptionModel.load_model(weights_path=weights, device="cpu")
+    stamp("model loaded (cold start)")
+
+    t_transcribe = time.perf_counter()
+    events = []
+    notes = 0
+    for ev in model.transcribe(str(audio_arg)):
+        if isinstance(ev, ProgressEvent):
+            print(f"[pack] progress: chunk {ev.completed}/{ev.total}", flush=True)
+        else:
+            events.append(ev)
+            if isinstance(ev, NoteStartEvent):
+                notes += 1
+
+    midi_bytes = model.events_to_midi_bytes(iter(events), beat_grid=None)
+    dst.parent.mkdir(parents=True, exist_ok=True)
+    dst.write_bytes(midi_bytes)
+    if tmp_wav:
+        os.unlink(tmp_wav)
+
+    print(f"[pack] transcribe wall time: {time.perf_counter() - t_transcribe:.2f}s",
+          flush=True)
+    print(f"DONE notes={notes} output={dst} bytes={len(midi_bytes)}", flush=True)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())