SONICS detect API (FastAPI wrapper, SpecTTTra-alpha-120s)

Dockerfile

@@ -12,11 +12,12 @@ ENV HOME=/home/user \
 WORKDIR /home/user/app
 
 RUN pip install --no-cache-dir --user torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu && \
-    pip install --no-cache-dir --user librosa soundfile numpy fastapi "uvicorn[standard]" python-multipart huggingface_hub timm pandas matplotlib tqdm scikit-learn fvcore && \
+    pip install --no-cache-dir --user librosa soundfile numpy fastapi "uvicorn[standard]" python-multipart huggingface_hub timm pandas matplotlib tqdm scikit-learn fvcore transformers && \
     pip install --no-cache-dir --user --no-deps git+https://github.com/awsaf49/sonics.git
 
 # Bake model weights into the image so cold starts skip the download.
 RUN python -c "from sonics import HFAudioClassifier; HFAudioClassifier.from_pretrained('awsaf49/sonics-spectttra-alpha-120s')"
+RUN python -c "from transformers import AutoFeatureExtractor, AutoModelForAudioClassification; m='MattyB95/AST-ASVspoof5-Synthetic-Voice-Detection'; AutoFeatureExtractor.from_pretrained(m); AutoModelForAudioClassification.from_pretrained(m)"
 
 COPY --chown=user app.py .
 

app.py

@@ -7,6 +7,7 @@ import torch
 from fastapi import FastAPI, File, Header, HTTPException, UploadFile
 
 MODEL_ID = os.environ.get("MODEL_ID", "awsaf49/sonics-spectttra-alpha-120s")
+VOICE_MODEL_ID = os.environ.get("VOICE_MODEL_ID", "MattyB95/AST-ASVspoof5-Synthetic-Voice-Detection")
 API_KEY = os.environ.get("DETECT_API_KEY", "")
 MAX_BYTES = 25 * 1024 * 1024
 MODEL_SR = 16000
@@ -15,21 +16,161 @@ torch.set_num_threads(2)
 
 app = FastAPI(title="SONICS Detect API")
 model = None
+voice_model = None
+voice_extractor = None
 
 
 @app.on_event("startup")
 def load_model():
-    global model
+    global model, voice_model, voice_extractor
     from sonics import HFAudioClassifier
 
     m = HFAudioClassifier.from_pretrained(MODEL_ID)
     m.eval()
     model = m
 
+    try:
+        from transformers import AutoFeatureExtractor, AutoModelForAudioClassification
+
+        voice_extractor = AutoFeatureExtractor.from_pretrained(VOICE_MODEL_ID)
+        vm = AutoModelForAudioClassification.from_pretrained(VOICE_MODEL_ID)
+        vm.eval()
+        voice_model = vm
+    except Exception:
+        voice_model = None
+        voice_extractor = None
+
 
 @app.get("/")
 def health():
-    return {"ok": True, "model": MODEL_ID, "loaded": model is not None}
+    return {
+        "ok": True,
+        "model": MODEL_ID,
+        "loaded": model is not None,
+        "voice_model": VOICE_MODEL_ID,
+        "voice_loaded": voice_model is not None,
+    }
+
+
+# Krumhansl-Schmuckler key profiles (major/minor).
+KS_MAJOR = np.array([6.35, 2.23, 3.48, 2.33, 4.38, 4.09, 2.52, 5.19, 2.39, 3.66, 2.29, 2.88])
+KS_MINOR = np.array([6.33, 2.68, 3.52, 5.38, 2.60, 3.53, 2.54, 4.75, 3.98, 2.69, 3.34, 3.17])
+PITCHES = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"]
+
+
+def estimate_key(y: np.ndarray, sr: int) -> dict:
+    chroma = librosa.feature.chroma_cqt(y=y, sr=sr).mean(axis=1)
+    if chroma.sum() <= 0:
+        return {"key": None, "mode": None, "confidence": None}
+    scores = []
+    for shift in range(12):
+        rolled = np.roll(chroma, -shift)
+        for mode, profile in (("major", KS_MAJOR), ("minor", KS_MINOR)):
+            r = float(np.corrcoef(rolled, profile)[0, 1])
+            scores.append((r, PITCHES[shift], mode))
+    scores.sort(reverse=True)
+    best, second = scores[0], scores[1]
+    confidence = max(0.0, min(1.0, (best[0] - second[0]) * 5 + 0.5))
+    return {"key": best[1], "mode": best[2], "confidence": round(confidence, 2)}
+
+
+def load_upload(audio: UploadFile, sr: int | None = None):
+    data = audio.file.read()
+    if not data:
+        raise HTTPException(status_code=400, detail="Empty file")
+    if len(data) > MAX_BYTES:
+        raise HTTPException(status_code=413, detail="File too large (max 25MB)")
+    suffix = os.path.splitext(audio.filename or "")[1] or ".mp3"
+    try:
+        with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
+            tmp.write(data)
+            tmp_path = tmp.name
+        try:
+            y, sr_out = librosa.load(tmp_path, sr=sr, mono=True)
+        finally:
+            os.unlink(tmp_path)
+    except HTTPException:
+        raise
+    except Exception:
+        raise HTTPException(status_code=400, detail="Could not decode audio")
+    return y, sr_out
+
+
+@app.post("/analyze")
+def analyze(audio: UploadFile = File(...), x_detect_key: str = Header(default="")):
+    """Music utilities: BPM + musical key + basic facts. No AI model involved."""
+    if API_KEY and x_detect_key != API_KEY:
+        raise HTTPException(status_code=401, detail="Invalid detect key")
+
+    y, sr = load_upload(audio, sr=None)
+    if y.size < sr * 3:
+        raise HTTPException(status_code=400, detail="Audio too short (min 3s)")
+
+    tempo_bpm = None
+    tempo_alt = None
+    try:
+        tempo, beats = librosa.beat.beat_track(y=y, sr=sr, hop_length=512)
+        t = float(np.atleast_1d(tempo)[0]) if tempo is not None else 0.0
+        if t > 0:
+            tempo_bpm = round(t, 1)
+            # Common octave error alternative (half/double time).
+            tempo_alt = round(t * 2, 1) if t < 90 else round(t / 2, 1)
+    except Exception:
+        pass
+
+    key = estimate_key(y, sr)
+
+    return {
+        "bpm": tempo_bpm,
+        "bpm_alternative": tempo_alt,
+        "key": key["key"],
+        "mode": key["mode"],
+        "key_confidence": key["confidence"],
+        "duration_s": round(len(y) / sr, 1),
+        "sample_rate": int(sr),
+    }
+
+
+@app.post("/detect-voice")
+def detect_voice(audio: UploadFile = File(...), x_detect_key: str = Header(default="")):
+    """AI voice / speech deepfake detection (AST fine-tuned on ASVspoof 5)."""
+    if API_KEY and x_detect_key != API_KEY:
+        raise HTTPException(status_code=401, detail="Invalid detect key")
+    if voice_model is None or voice_extractor is None:
+        raise HTTPException(status_code=503, detail="Voice model unavailable")
+
+    y, sr = load_upload(audio, sr=16000)
+    if y.size < sr * 2:
+        raise HTTPException(status_code=400, detail="Audio too short (min 2s)")
+
+    # Score up to three 10s windows (start / middle / end) and average.
+    win = sr * 10
+    starts = [0]
+    if len(y) > win * 2:
+        starts.append((len(y) - win) // 2)
+    if len(y) > win:
+        starts.append(max(0, len(y) - win))
+    probs = []
+    id2label = voice_model.config.id2label
+    for s in dict.fromkeys(starts):
+        chunk = y[s : s + win]
+        inputs = voice_extractor(chunk, sampling_rate=sr, return_tensors="pt")
+        with torch.no_grad():
+            logits = voice_model(**inputs).logits
+        p = torch.softmax(logits, dim=-1)[0]
+        spoof_idx = next(
+            (i for i, lbl in id2label.items() if "spoof" in lbl.lower() or "fake" in lbl.lower()),
+            1,
+        )
+        probs.append(float(p[int(spoof_idx)]))
+
+    ai_prob = float(np.mean(probs))
+    return {
+        "ai_prob": round(ai_prob, 4),
+        "windows": len(probs),
+        "duration_s": round(len(y) / sr, 1),
+        "model": VOICE_MODEL_ID,
+    }
 
 
 def compute_signals(y: np.ndarray, sr: int) -> dict: