Upload 4 files

Dockerfile

@@ -0,0 +1,42 @@
+# ─────────────────────────────────────────────────────────────────────────────
+# Voice-to-MIDI  ·  Hugging Face Spaces Dockerfile
+# Base: official Python slim – keeps image small for faster cold starts
+# ─────────────────────────────────────────────────────────────────────────────
+FROM python:3.10-slim
+
+# HF Spaces runs as a non-root user; create it early
+RUN useradd -m -u 1000 appuser
+
+WORKDIR /app
+
+# System deps: ffmpeg for broad audio codec support, libsndfile for soundfile
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    ffmpeg \
+    libsndfile1 \
+    libgomp1 \
+    git \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy and install Python deps as root (writes to site-packages)
+COPY requirements.txt .
+RUN pip install --no-cache-dir --upgrade pip \
+    && pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY app.py .
+
+# Switch to non-root user (required by HF Spaces)
+USER appuser
+
+# HF Spaces expects the service on port 7860
+EXPOSE 7860
+
+# Pre-warm the Basic Pitch model on container start so the first request
+# doesn't pay the cold-start cost.  The one-worker uvicorn picks it up.
+ENV PYTHONUNBUFFERED=1
+ENV OMP_NUM_THREADS=2
+ENV TF_NUM_INTEROP_THREADS=2
+ENV TF_NUM_INTRAOP_THREADS=2
+
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860", \
+     "--workers", "1", "--log-level", "info"]

app.py

@@ -0,0 +1,521 @@
+"""
+Voice-to-MIDI FastAPI Service
+Powered by Spotify Basic Pitch
+Optimized for Hugging Face Spaces CPU deployment
+"""
+
+import os
+import uuid
+import time
+import asyncio
+import logging
+import tempfile
+import threading
+from pathlib import Path
+from typing import Optional
+from contextlib import asynccontextmanager
+from collections import defaultdict
+
+import numpy as np
+import mido
+import soundfile as sf
+import librosa
+import uvicorn
+from fastapi import FastAPI, File, UploadFile, Form, HTTPException, BackgroundTasks
+from fastapi.responses import FileResponse, JSONResponse
+from fastapi.middleware.cors import CORSMiddleware
+
+# ─────────────────────────────────────────────
+# Logging
+# ─────────────────────────────────────────────
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+log = logging.getLogger("voice-to-midi")
+
+# ─────────────────────────────────────────────
+# Constants
+# ─────────────────────────────────────────────
+SUPPORTED_TYPES = {
+    "audio/wav", "audio/x-wav", "audio/mpeg", "audio/mp3",
+    "audio/ogg", "audio/x-m4a", "audio/mp4", "audio/aac",
+    "audio/flac", "application/octet-stream",
+}
+SUPPORTED_EXTENSIONS = {".wav", ".mp3", ".ogg", ".m4a", ".flac", ".aac"}
+OUTPUT_DIR = Path(tempfile.gettempdir()) / "voice_midi_outputs"
+OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
+MAX_FILE_AGE_SECONDS = 1800          # 30 min – auto-cleanup
+MAX_UPLOAD_MB = 50
+MIDI_NOTE_MIN = 21                   # A0
+MIDI_NOTE_MAX = 108                  # C8
+DEFAULT_SAMPLE_RATE = 22050
+
+# ─────────────────────────────────────────────
+# Global model state (loaded once at startup)
+# ─────────────────────────────────────────────
+_model_lock = threading.Lock()
+_model_loaded = False
+
+
+def _ensure_model_loaded() -> None:
+    """Import basic_pitch and warm up TensorFlow graph on first call."""
+    global _model_loaded
+    if _model_loaded:
+        return
+    with _model_lock:
+        if _model_loaded:
+            return
+        log.info("Loading Basic Pitch model – this happens once at cold start …")
+        # Importing triggers TF/tflite graph compilation
+        from basic_pitch.inference import predict  # noqa: F401  (warm up)
+        _model_loaded = True
+        log.info("Basic Pitch model ready.")
+
+
+# ─────────────────────────────────────────────
+# App lifespan (startup / shutdown)
+# ─────────────────────────────────────────────
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Warm the model in a thread so the event-loop isn't blocked
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(None, _ensure_model_loaded)
+    # Schedule background cleanup
+    cleanup_task = asyncio.create_task(_periodic_cleanup())
+    yield
+    cleanup_task.cancel()
+
+
+app = FastAPI(
+    title="Voice-to-MIDI",
+    description="Convert voice / humming / whistling to MIDI via Spotify Basic Pitch",
+    version="1.0.0",
+    lifespan=lifespan,
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+# ─────────────────────────────────────────────
+# Background cleanup
+# ─────────────────────────────────────────────
+async def _periodic_cleanup():
+    """Delete MIDI files older than MAX_FILE_AGE_SECONDS every 10 minutes."""
+    while True:
+        await asyncio.sleep(600)
+        _cleanup_old_files()
+
+
+def _cleanup_old_files():
+    now = time.time()
+    removed = 0
+    for f in OUTPUT_DIR.glob("*.mid"):
+        if now - f.stat().st_mtime > MAX_FILE_AGE_SECONDS:
+            try:
+                f.unlink()
+                removed += 1
+            except OSError:
+                pass
+    if removed:
+        log.info(f"Cleaned up {removed} old MIDI file(s).")
+
+
+# ─────────────────────────────────────────────
+# Utility helpers
+# ─────────────────────────────────────────────
+
+NOTE_NAMES = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"]
+
+
+def midi_to_note_name(midi_pitch: int) -> str:
+    """Convert MIDI pitch integer to scientific notation, e.g. 60 → C4."""
+    octave = (midi_pitch // 12) - 1
+    name = NOTE_NAMES[midi_pitch % 12]
+    return f"{name}{octave}"
+
+
+def load_audio(path: Path) -> tuple[np.ndarray, int]:
+    """Load any audio file to mono float32, resampled to DEFAULT_SAMPLE_RATE."""
+    try:
+        audio, sr = librosa.load(str(path), sr=DEFAULT_SAMPLE_RATE, mono=True)
+        return audio, sr
+    except Exception as exc:
+        raise HTTPException(status_code=422, detail=f"Cannot decode audio: {exc}")
+
+
+def estimate_bpm(audio: np.ndarray, sr: int) -> Optional[float]:
+    """Use librosa onset/beat tracking to estimate tempo."""
+    try:
+        tempo, _ = librosa.beat.beat_track(y=audio, sr=sr)
+        val = float(np.atleast_1d(tempo)[0])
+        return round(val, 1) if 30 < val < 300 else None
+    except Exception:
+        return None
+
+
+# ─────────────────────────────────────────────
+# MIDI quality post-processing
+# ─────────────────────────────────────────────
+
+def clamp_pitch(pitch: int) -> int:
+    return max(MIDI_NOTE_MIN, min(MIDI_NOTE_MAX, pitch))
+
+
+def remove_duplicate_overlaps(notes: list[dict]) -> list[dict]:
+    """
+    Remove notes that are exact duplicates or where one note completely
+    contains another at the same pitch.
+    """
+    # Sort by pitch then start time
+    notes = sorted(notes, key=lambda n: (n["pitch"], n["start"]))
+    result = []
+    by_pitch: dict[int, list[dict]] = defaultdict(list)
+    for n in notes:
+        by_pitch[n["pitch"]].append(n)
+
+    for pitch, group in by_pitch.items():
+        group = sorted(group, key=lambda n: n["start"])
+        merged = [group[0]]
+        for note in group[1:]:
+            prev = merged[-1]
+            # If new note starts before previous ends – merge or skip
+            if note["start"] < prev["end"]:
+                # Extend previous note if new one reaches further
+                if note["end"] > prev["end"]:
+                    prev["end"] = note["end"]
+                    prev["velocity"] = max(prev["velocity"], note["velocity"])
+                # else: new note is fully contained – skip it
+            else:
+                merged.append(note)
+        result.extend(merged)
+
+    return sorted(result, key=lambda n: n["start"])
+
+
+def merge_tiny_notes(notes: list[dict], min_ms: float = 40) -> list[dict]:
+    """Drop notes shorter than min_ms milliseconds."""
+    min_sec = min_ms / 1000.0
+    kept = []
+    for n in notes:
+        if (n["end"] - n["start"]) >= min_sec:
+            kept.append(n)
+    return kept
+
+
+def quantize_notes(notes: list[dict], bpm: float, subdivisions: int = 16) -> list[dict]:
+    """Snap note start/end times to the nearest subdivision grid."""
+    if bpm <= 0:
+        return notes
+    beat_sec = 60.0 / bpm
+    grid = beat_sec / (subdivisions / 4)  # e.g. 16th note
+
+    def snap(t: float) -> float:
+        return round(round(t / grid) * grid, 4)
+
+    for n in notes:
+        n["start"] = snap(n["start"])
+        n["end"] = snap(n["end"])
+        if n["end"] <= n["start"]:
+            n["end"] = n["start"] + grid
+    return notes
+
+
+def notes_to_midi(
+    notes: list[dict],
+    bpm: float,
+    instrument_name: str = "Acoustic Grand Piano",
+) -> mido.MidiFile:
+    """Convert list of note dicts to a mido MidiFile object."""
+    mid = mido.MidiFile(type=0, ticks_per_beat=480)
+    track = mido.MidiTrack()
+    mid.tracks.append(track)
+
+    # Tempo
+    tempo = mido.bpm2tempo(bpm)
+    track.append(mido.MetaMessage("set_tempo", tempo=tempo, time=0))
+
+    # Program change (General MIDI instrument)
+    gm_programs = {
+        "Acoustic Grand Piano": 0, "Electric Piano": 4,
+        "Violin": 40, "Flute": 73, "Synth Lead": 80,
+    }
+    program = gm_programs.get(instrument_name, 0)
+    track.append(mido.Message("program_change", program=program, time=0))
+
+    # Build flat event list: (abs_time_sec, type, pitch, velocity)
+    events = []
+    for n in notes:
+        events.append((n["start"], "note_on",  n["pitch"], n["velocity"]))
+        events.append((n["end"],   "note_off", n["pitch"], 0))
+
+    events.sort(key=lambda e: (e[0], 0 if e[1] == "note_off" else 1))
+
+    def sec_to_ticks(t: float) -> int:
+        return int(mido.second2tick(t, mid.ticks_per_beat, tempo))
+
+    prev_ticks = 0
+    for abs_sec, msg_type, pitch, vel in events:
+        abs_ticks = sec_to_ticks(abs_sec)
+        delta = max(0, abs_ticks - prev_ticks)
+        track.append(mido.Message(msg_type, note=pitch, velocity=vel, time=delta))
+        prev_ticks = abs_ticks
+
+    return mid
+
+
+def group_chords(notes: list[dict], window_sec: float = 0.05) -> list[list[dict]]:
+    """Group notes that start within window_sec of each other into chords."""
+    if not notes:
+        return []
+    sorted_notes = sorted(notes, key=lambda n: n["start"])
+    groups = [[sorted_notes[0]]]
+    for note in sorted_notes[1:]:
+        if abs(note["start"] - groups[-1][0]["start"]) <= window_sec:
+            groups[-1].append(note)
+        else:
+            groups.append([note])
+    return groups
+
+
+# ─────────────────────────────────────────────
+# Core transcription logic
+# ─────────────────────────────────────────────
+
+def transcribe_audio(
+    audio: np.ndarray,
+    sr: int,
+    bpm_hint: Optional[float],
+    quantize: bool,
+    min_note_length_ms: float,
+    onset_sensitivity: float,
+) -> tuple[list[dict], float, Optional[float]]:
+    """
+    Run Basic Pitch inference and return:
+      - list of note dicts
+      - audio duration (seconds)
+      - detected bpm (or None)
+    """
+    from basic_pitch.inference import predict
+    from basic_pitch import ICASSP_2022_MODEL_PATH
+
+    duration = len(audio) / sr
+
+    # Write temp wav for basic_pitch (it expects a file path)
+    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
+        tmp_path = tmp.name
+    try:
+        sf.write(tmp_path, audio, sr)
+
+        model_output, midi_data, note_events = predict(
+            tmp_path,
+            onset_threshold=onset_sensitivity,
+            frame_threshold=0.3,
+            minimum_note_length=min_note_length_ms / 1000.0,
+            minimum_frequency=librosa.midi_to_hz(MIDI_NOTE_MIN),
+            maximum_frequency=librosa.midi_to_hz(MIDI_NOTE_MAX),
+            melodia_trick=True,
+            midi_tempo=bpm_hint or 120,
+        )
+    finally:
+        os.unlink(tmp_path)
+
+    # note_events: list of (start_time, end_time, pitch_midi, amplitude, pitch_bends)
+    notes = []
+    for start, end, pitch, amplitude, _ in note_events:
+        pitch = clamp_pitch(int(round(pitch)))
+        velocity = int(np.clip(amplitude * 127, 1, 127))
+        notes.append({
+            "pitch": pitch,
+            "note_name": midi_to_note_name(pitch),
+            "start": round(float(start), 4),
+            "end": round(float(end), 4),
+            "velocity": velocity,
+            "confidence": round(float(amplitude), 4),
+        })
+
+    # Post-process
+    notes = merge_tiny_notes(notes, min_ms=min_note_length_ms)
+    notes = remove_duplicate_overlaps(notes)
+
+    # BPM
+    detected_bpm = bpm_hint or estimate_bpm(audio, sr)
+    if quantize and detected_bpm:
+        notes = quantize_notes(notes, detected_bpm)
+
+    notes = sorted(notes, key=lambda n: n["start"])
+    return notes, duration, detected_bpm
+
+
+# ─────────────────────────────────────────────
+# Routes
+# ─────────────────────────────────────────────
+
+@app.get("/", summary="Health check")
+async def root():
+    return {
+        "status": "online",
+        "service": "voice-to-midi",
+        "engine": "basic-pitch",
+        "model_ready": _model_loaded,
+    }
+
+
+@app.get("/health")
+async def health():
+    return {"status": "ok"}
+
+
+@app.post("/transcribe", summary="Convert audio to MIDI")
+async def transcribe(
+    background_tasks: BackgroundTasks,
+    file: UploadFile = File(..., description="Audio file (wav/mp3/ogg/m4a/flac)"),
+    bpm: Optional[float] = Form(None, description="Hint BPM (auto-detected if omitted)"),
+    quantize: bool = Form(False, description="Snap notes to nearest beat grid"),
+    instrument_name: str = Form("Acoustic Grand Piano"),
+    min_note_length_ms: float = Form(40.0, description="Minimum note duration in ms"),
+    onset_sensitivity: float = Form(0.5, description="Onset detection threshold 0–1"),
+):
+    # ── Validate file ──────────────────────────────────────────────────────
+    if not file.filename:
+        raise HTTPException(status_code=400, detail="No filename provided.")
+
+    suffix = Path(file.filename).suffix.lower()
+    if suffix not in SUPPORTED_EXTENSIONS:
+        raise HTTPException(
+            status_code=415,
+            detail=f"Unsupported file type '{suffix}'. Supported: {sorted(SUPPORTED_EXTENSIONS)}",
+        )
+
+    content_type = (file.content_type or "").split(";")[0].strip().lower()
+    if content_type and content_type not in SUPPORTED_TYPES:
+        log.warning(f"Unexpected content-type: {content_type} – proceeding anyway.")
+
+    onset_sensitivity = float(np.clip(onset_sensitivity, 0.1, 0.9))
+    min_note_length_ms = max(10.0, min(500.0, min_note_length_ms))
+
+    # ── Read & size-check upload ──────────────────────────────────────────���
+    raw = await file.read()
+    if len(raw) > MAX_UPLOAD_MB * 1_000_000:
+        raise HTTPException(status_code=413, detail=f"File exceeds {MAX_UPLOAD_MB} MB limit.")
+    if len(raw) < 1024:
+        raise HTTPException(status_code=400, detail="File too small – likely empty or corrupt.")
+
+    # ── Save to temp, load audio ───────────────────────────────────────────
+    with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
+        tmp.write(raw)
+        tmp_input_path = Path(tmp.name)
+
+    try:
+        audio, sr = load_audio(tmp_input_path)
+    finally:
+        tmp_input_path.unlink(missing_ok=True)
+
+    if len(audio) / sr < 0.2:
+        raise HTTPException(status_code=422, detail="Audio too short (< 0.2 s).")
+
+    # ── Ensure model loaded ────────────────────────────────────────────────
+    _ensure_model_loaded()
+
+    # ── Run inference ──────────────────────────────────────────────────────
+    t0 = time.perf_counter()
+    try:
+        loop = asyncio.get_event_loop()
+        notes, duration, detected_bpm = await loop.run_in_executor(
+            None,
+            transcribe_audio,
+            audio, sr, bpm, quantize, min_note_length_ms, onset_sensitivity,
+        )
+    except HTTPException:
+        raise
+    except Exception as exc:
+        log.exception("Inference failed")
+        raise HTTPException(status_code=500, detail=f"Inference error: {exc}")
+
+    elapsed = time.perf_counter() - t0
+    log.info(f"Transcribed {duration:.1f}s audio → {len(notes)} notes in {elapsed:.2f}s")
+
+    if not notes:
+        raise HTTPException(
+            status_code=422,
+            detail="No notes detected. Try adjusting onset_sensitivity or check audio quality.",
+        )
+
+    # ── Build MIDI ─────────────────────────────────────────────────────────
+    effective_bpm = detected_bpm or 120.0
+    midi_obj = notes_to_midi(notes, effective_bpm, instrument_name)
+    midi_id = uuid.uuid4().hex
+    midi_path = OUTPUT_DIR / f"{midi_id}.mid"
+    midi_obj.save(str(midi_path))
+
+    # ── Chord grouping meta ────────────────────────────────────────────────
+    chords = group_chords(notes)
+    chord_count = sum(1 for g in chords if len(g) > 1)
+
+    return JSONResponse({
+        "success": True,
+        "notes": notes,
+        "midi_url": f"/download/{midi_id}.mid",
+        "note_count": len(notes),
+        "duration": round(duration, 3),
+        "bpm": effective_bpm,
+        "bpm_source": "hint" if bpm else ("detected" if detected_bpm else "default"),
+        "chord_count": chord_count,
+        "processing_time_sec": round(elapsed, 3),
+        "quantized": quantize,
+        "instrument": instrument_name,
+    })
+
+
+@app.get("/download/{filename}", summary="Download generated MIDI file")
+async def download(filename: str):
+    # Security: disallow path traversal
+    if "/" in filename or "\\" in filename or ".." in filename:
+        raise HTTPException(status_code=400, detail="Invalid filename.")
+    if not filename.endswith(".mid"):
+        raise HTTPException(status_code=400, detail="Only .mid files available.")
+
+    path = OUTPUT_DIR / filename
+    if not path.exists():
+        raise HTTPException(status_code=404, detail="File not found or expired.")
+
+    return FileResponse(
+        path=str(path),
+        media_type="audio/midi",
+        filename=filename,
+        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
+    )
+
+
+@app.get("/info", summary="Service info & supported formats")
+async def info():
+    return {
+        "supported_formats": sorted(SUPPORTED_EXTENSIONS),
+        "max_upload_mb": MAX_UPLOAD_MB,
+        "midi_note_range": {"min": MIDI_NOTE_MIN, "max": MIDI_NOTE_MAX},
+        "default_sample_rate": DEFAULT_SAMPLE_RATE,
+        "file_expiry_minutes": MAX_FILE_AGE_SECONDS // 60,
+        "endpoints": [
+            {"path": "/", "method": "GET", "description": "Health check"},
+            {"path": "/transcribe", "method": "POST", "description": "Audio → MIDI"},
+            {"path": "/download/{filename}", "method": "GET", "description": "MIDI download"},
+        ],
+    }
+
+
+# ─────────────────────────────────────────────
+# Entry point (local dev)
+# ─────────────────────────────────────────────
+if __name__ == "__main__":
+    uvicorn.run(
+        "app:app",
+        host="0.0.0.0",
+        port=7860,
+        workers=1,          # Single worker on HF Spaces CPU; model is global
+        log_level="info",
+    )

daw_companion.py

@@ -0,0 +1,224 @@
+#!/usr/bin/env python3
+"""
+daw_companion.py
+────────────────
+Local DAW integration script for Voice-to-MIDI service.
+
+Usage:
+    python daw_companion.py record           # Record mic → transcribe → save MIDI
+    python daw_companion.py file audio.wav   # Transcribe existing file → save MIDI
+    python daw_companion.py watch ./inbox    # Watch folder for new audio, auto-convert
+
+Requirements:
+    pip install requests sounddevice scipy
+
+Configure SERVICE_URL below to point at your Hugging Face Space.
+"""
+
+import sys
+import time
+import os
+import argparse
+import tempfile
+from pathlib import Path
+
+import requests
+
+# ─── Configuration ─────────────────────────────────────────────────────────────
+SERVICE_URL = os.getenv(
+    "VOICE_MIDI_URL",
+    "https://YOUR-USERNAME-voice-to-midi.hf.space",   # ← change this
+)
+OUTPUT_DIR = Path("./midi_output")
+OUTPUT_DIR.mkdir(exist_ok=True)
+
+DEFAULT_PARAMS = {
+    "bpm": None,           # None = auto-detect
+    "quantize": False,
+    "instrument_name": "Acoustic Grand Piano",
+    "min_note_length_ms": 40,
+    "onset_sensitivity": 0.5,
+}
+# ───────────────────────────────────────────────────────────────────────────────
+
+
+def check_service():
+    try:
+        r = requests.get(f"{SERVICE_URL}/", timeout=10)
+        r.raise_for_status()
+        data = r.json()
+        print(f"✅  Service online  |  engine: {data.get('engine')}  |  model_ready: {data.get('model_ready')}")
+        return True
+    except Exception as exc:
+        print(f"❌  Cannot reach service: {exc}")
+        return False
+
+
+def transcribe_file(audio_path: Path, params: dict = None, output_name: str = None) -> Path | None:
+    """Send an audio file to the service and save the returned MIDI."""
+    params = {**DEFAULT_PARAMS, **(params or {})}
+
+    print(f"\n📤  Uploading {audio_path.name} ({audio_path.stat().st_size // 1024} KB) …")
+    t0 = time.perf_counter()
+
+    form_data = {k: str(v) for k, v in params.items() if v is not None}
+
+    with audio_path.open("rb") as fh:
+        resp = requests.post(
+            f"{SERVICE_URL}/transcribe",
+            files={"file": (audio_path.name, fh, "audio/wav")},
+            data=form_data,
+            timeout=120,
+        )
+
+    if resp.status_code != 200:
+        print(f"❌  Error {resp.status_code}: {resp.text}")
+        return None
+
+    result = resp.json()
+    elapsed = time.perf_counter() - t0
+
+    print(f"✅  Transcribed in {elapsed:.1f}s")
+    print(f"   Notes      : {result['note_count']}")
+    print(f"   Duration   : {result['duration']:.2f}s")
+    print(f"   BPM        : {result['bpm']} ({result['bpm_source']})")
+    print(f"   Chords     : {result['chord_count']}")
+    print(f"   Processing : {result['processing_time_sec']}s")
+
+    # Download MIDI
+    midi_url = f"{SERVICE_URL}{result['midi_url']}"
+    midi_resp = requests.get(midi_url, timeout=30)
+    midi_resp.raise_for_status()
+
+    stem = output_name or audio_path.stem
+    midi_path = OUTPUT_DIR / f"{stem}_{int(time.time())}.mid"
+    midi_path.write_bytes(midi_resp.content)
+    print(f"💾  Saved MIDI → {midi_path}")
+
+    # Print note table
+    if result["notes"]:
+        print("\n   Pitch  Note  Start    End      Vel  Conf")
+        print("   ─────  ────  ───────  ───────  ───  ────")
+        for n in result["notes"][:20]:
+            print(
+                f"   {n['pitch']:>5}  {n['note_name']:<4}  "
+                f"{n['start']:>7.3f}  {n['end']:>7.3f}  "
+                f"{n['velocity']:>3}  {n['confidence']:.2f}"
+            )
+        if len(result["notes"]) > 20:
+            print(f"   … and {len(result['notes']) - 20} more notes")
+
+    return midi_path
+
+
+def record_and_transcribe(duration_sec: float = 5.0, params: dict = None) -> Path | None:
+    """Record from default microphone and transcribe."""
+    try:
+        import sounddevice as sd
+        from scipy.io.wavfile import write as wav_write
+        import numpy as np
+    except ImportError:
+        print("❌  Install sounddevice and scipy: pip install sounddevice scipy")
+        return None
+
+    SR = 44100
+    print(f"\n🎙️  Recording for {duration_sec:.0f}s … (press Ctrl+C to stop early)")
+    try:
+        audio = sd.rec(int(duration_sec * SR), samplerate=SR, channels=1, dtype="float32")
+        sd.wait()
+    except KeyboardInterrupt:
+        sd.stop()
+        audio = audio[:sd.get_stream().time * SR]  # trim
+        print("  (recording stopped early)")
+
+    # Write to temp wav
+    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
+        wav_path = Path(tmp.name)
+    wav_write(str(wav_path), SR, (audio * 32767).astype("int16"))
+
+    try:
+        return transcribe_file(wav_path, params=params, output_name="recording")
+    finally:
+        wav_path.unlink(missing_ok=True)
+
+
+def watch_folder(folder: Path, params: dict = None):
+    """Watch a folder for new audio files and auto-transcribe them."""
+    print(f"\n👀  Watching {folder.resolve()} for new audio files …")
+    print("    Drop .wav / .mp3 / .ogg files here. Press Ctrl+C to stop.\n")
+    seen = {f for f in folder.iterdir() if f.is_file()}
+    try:
+        while True:
+            time.sleep(1)
+            current = {f for f in folder.iterdir() if f.is_file()}
+            new_files = current - seen
+            seen = current
+            for f in sorted(new_files):
+                if f.suffix.lower() in {".wav", ".mp3", ".ogg", ".m4a", ".flac"}:
+                    print(f"\n🆕  New file detected: {f.name}")
+                    transcribe_file(f, params=params)
+    except KeyboardInterrupt:
+        print("\n👋  Stopped watching.")
+
+
+# ─── CLI ───────────────────────────────────────────────────────────────────────
+
+def main():
+    parser = argparse.ArgumentParser(description="Voice-to-MIDI DAW Companion")
+    sub = parser.add_subparsers(dest="cmd")
+
+    # record
+    rec = sub.add_parser("record", help="Record mic audio then transcribe")
+    rec.add_argument("--duration", type=float, default=5.0)
+    rec.add_argument("--bpm", type=float, default=None)
+    rec.add_argument("--quantize", action="store_true")
+
+    # file
+    fil = sub.add_parser("file", help="Transcribe an existing audio file")
+    fil.add_argument("path", type=Path)
+    fil.add_argument("--bpm", type=float, default=None)
+    fil.add_argument("--quantize", action="store_true")
+    fil.add_argument("--onset", type=float, default=0.5)
+
+    # watch
+    wat = sub.add_parser("watch", help="Watch folder for new audio files")
+    wat.add_argument("folder", type=Path)
+    wat.add_argument("--bpm", type=float, default=None)
+    wat.add_argument("--quantize", action="store_true")
+
+    # check
+    sub.add_parser("check", help="Check service health")
+
+    args = parser.parse_args()
+
+    if not check_service():
+        sys.exit(1)
+
+    if args.cmd == "check":
+        return
+
+    params = {}
+    if hasattr(args, "bpm") and args.bpm:
+        params["bpm"] = args.bpm
+    if hasattr(args, "quantize") and args.quantize:
+        params["quantize"] = True
+    if hasattr(args, "onset"):
+        params["onset_sensitivity"] = args.onset
+
+    if args.cmd == "record":
+        record_and_transcribe(duration_sec=args.duration, params=params)
+    elif args.cmd == "file":
+        if not args.path.exists():
+            print(f"❌  File not found: {args.path}")
+            sys.exit(1)
+        transcribe_file(args.path, params=params)
+    elif args.cmd == "watch":
+        if not args.folder.exists():
+            args.folder.mkdir(parents=True)
+        watch_folder(args.folder, params=params)
+    else:
+        parser.print_help()
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,30 @@
+# ─── Web framework ────────────────────────────────────────────────────────────
+fastapi==0.111.0
+uvicorn[standard]==0.29.0
+python-multipart==0.0.9
+
+# ─── Audio I/O ────────────────────────────────────────────────────────────────
+librosa==0.10.2
+soundfile==0.12.1
+audioread==3.0.1
+
+# ─── Spotify Basic Pitch ──────────────────────────────────────────────────────
+# Pinned to stable CPU-friendly release; uses tflite-runtime under the hood
+basic-pitch==0.3.3
+
+# ─── MIDI ─────────────────────────────────────────────────────────────────────
+mido==1.3.2
+
+# ─── Numerics ─────────────────────────────────────────────────────────────────
+numpy==1.26.4
+scipy==1.13.0
+
+# ─── TensorFlow Lite runtime (CPU, smaller than full TF) ──────────────────────
+# basic-pitch will pull tflite-runtime automatically; pin here for reproducibility
+tflite-runtime==2.14.0; platform_machine == "x86_64"
+# Fallback for other arches (HF Spaces uses x86_64, so this is usually unused)
+tensorflow-cpu==2.15.0; platform_machine != "x86_64"
+
+# ─── HTTP client (for DAW companion script example) ───────────────────────────
+httpx==0.27.0
+requests==2.31.0