Update app.py

app.py

@@ -0,0 +1,118 @@
+import os
+import pickle
+import tempfile
+import numpy as np
+import librosa
+import tensorflow as tf
+from fastapi import FastAPI, UploadFile, File, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+
+# ── Constants (from model_config.json) ────────────────────────────────────────
+SR          = 22050
+DURATION    = 3
+HOP_LENGTH  = 512
+TIME_FRAMES = 130
+N_BINS      = 84
+N_MFCC      = 40
+CLASSES     = ["Muwa", "east", "loyal", "poems", "rai"]
+
+# ── App ───────────────────────────────────────────────────────────────────────
+app = FastAPI(title="Audio Classifier API")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# ── Load artifacts on startup ──────────────────────────────────────────────────
+@app.on_event("startup")
+async def load_artifacts():
+    global model, scaler_cqt, scaler_mfcc, label_encoder
+
+    model = tf.keras.models.load_model("model_artifacts/best_cross_attn.keras", compile=False)
+    print("✅ Model loaded")
+
+    with open("model_artifacts/scaler_cqt.pkl", "rb") as f:
+        scaler_cqt = pickle.load(f)
+
+    with open("model_artifacts/scaler_mfcc.pkl", "rb") as f:
+        scaler_mfcc = pickle.load(f)
+
+    with open("model_artifacts/label_encoder.pkl", "rb") as f:
+        label_encoder = pickle.load(f)
+
+    print("✅ Scalers and label encoder loaded")
+
+# ── Feature extraction ─────────────────────────────────────────────────────────
+def extract_features(audio_path: str):
+    y, sr = librosa.load(audio_path, sr=SR, duration=DURATION)
+
+    # Pad if audio is shorter than expected
+    expected_samples = SR * DURATION
+    if len(y) < expected_samples:
+        y = np.pad(y, (0, expected_samples - len(y)))
+
+    # CQT → (TIME_FRAMES, N_BINS)
+    cqt = np.abs(librosa.cqt(y, sr=sr, hop_length=HOP_LENGTH, n_bins=N_BINS))
+    cqt = librosa.util.fix_length(cqt, size=TIME_FRAMES, axis=1).T  # (130, 84)
+
+    # Apply scaler (fit was on 2D, so flatten → scale → reshape)
+    cqt_shape = cqt.shape
+    cqt = scaler_cqt.transform(cqt.reshape(-1, cqt_shape[-1])).reshape(cqt_shape)
+
+    # MFCC → (TIME_FRAMES, N_MFCC)
+    mfcc = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=N_MFCC, hop_length=HOP_LENGTH)
+    mfcc = librosa.util.fix_length(mfcc, size=TIME_FRAMES, axis=1).T  # (130, 40)
+
+    # Apply scaler
+    mfcc_shape = mfcc.shape
+    mfcc = scaler_mfcc.transform(mfcc.reshape(-1, mfcc_shape[-1])).reshape(mfcc_shape)
+
+    # Add batch dimension → (1, 130, 84) and (1, 130, 40)
+    return cqt[np.newaxis], mfcc[np.newaxis]
+
+# ── Predict endpoint ───────────────────────────────────────────────────────────
+@app.post("/predict")
+async def predict(file: UploadFile = File(...)):
+    if not file.filename.endswith((".wav", ".mp3", ".flac", ".ogg")):
+        raise HTTPException(status_code=400, detail="Unsupported format. Use wav, mp3, flac, or ogg.")
+
+    with tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(file.filename)[1]) as tmp:
+        tmp.write(await file.read())
+        tmp_path = tmp.name
+
+    try:
+        cqt, mfcc = extract_features(tmp_path)
+        preds     = model.predict([cqt, mfcc])[0]          # (5,)
+        top_idx   = int(np.argmax(preds))
+
+        # Use label_encoder if sklearn LabelEncoder, else fall back to CLASSES list
+        try:
+            label = label_encoder.inverse_transform([top_idx])[0]
+        except Exception:
+            label = CLASSES[top_idx]
+
+        return {
+            "label":      label,
+            "confidence": round(float(preds[top_idx]), 4),
+            "all_scores": {
+                CLASSES[i]: round(float(p), 4) for i, p in enumerate(preds)
+            }
+        }
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Inference error: {str(e)}")
+
+    finally:
+        os.remove(tmp_path)
+
+# ── Health endpoints ───────────────────────────────────────────────────────────
+@app.get("/")
+def root():
+    return {"status": "ok", "message": "Audio Classifier is running"}
+
+@app.get("/health")
+def health():
+    return {"status": "healthy", "model": "CrossAttnModel", "classes": CLASSES}