Update app.py

app.py

@@ -1,20 +1,18 @@
 import json
+import os
 from typing import Any, Dict
-from fastapi import FastAPI
+
 import numpy as np
 import tensorflow as tf
-from fastapi import FastAPI, HTTPException
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
 
-# -----------------
-# Config
-# -----------------
-MODEL_PATH = "best_model.h5"      # upload alongside app.py
-STATS_PATH = "means_std.json"     # {"feature":{"mean":..., "std":...}, ...}
+# ----------------- CONFIG -----------------
+MODEL_PATH = os.getenv("MODEL_PATH", "best_model.h5")
+STATS_PATH = os.getenv("STATS_PATH", "means_std.json")
 CLASSES = ["Top", "Mid-Top", "Mid", "Mid-Low", "Low"]
+# ------------------------------------------
 
-# -----------------
-# Load artifacts
-# -----------------
 print("Loading model and stats...")
 model = tf.keras.models.load_model(MODEL_PATH, compile=False)
 
@@ -24,76 +22,100 @@ with open(STATS_PATH, "r") as f:
 FEATURES = list(stats.keys())
 print("Feature order:", FEATURES)
 
-# -----------------
-# Helpers
-# -----------------
-def _z(v: Any, mean: float, sd: float) -> float:
+def _z(val: Any, mean: float, sd: float) -> float:
     try:
-        x = float(v)
+        v = float(val)
     except Exception:
         return 0.0
-    if not sd or sd == 0.0:
+    if not sd:
         return 0.0
-    return (x - mean) / sd
+    return (v - mean) / sd
 
-def _coral_probs_from_logits(logits_np: np.ndarray) -> np.ndarray:
-    """
-    For ordinal (K-1 logits): p_k = σ(z_{k-1}) - σ(z_k) with σ(z_0)=1, σ(z_K)=0
-    """
+def coral_probs_from_logits(logits_np: np.ndarray) -> np.ndarray:
+    """(N, K-1) logits -> (N, K) probabilities for CORAL ordinal output."""
     logits = tf.convert_to_tensor(logits_np, dtype=tf.float32)
-    sig = tf.math.sigmoid(logits)                  # (N, K-1)
+    sig = tf.math.sigmoid(logits)  # (N, K-1)
     left  = tf.concat([tf.ones_like(sig[:, :1]), sig], axis=1)
     right = tf.concat([sig, tf.zeros_like(sig[:, :1])], axis=1)
     probs = tf.clip_by_value(left - right, 1e-12, 1.0)
     return probs.numpy()
 
-def _predict_core(ratios: Dict[str, Any]) -> Dict[str, Any]:
-    missing = [f for f in FEATURES if f not in ratios]
-
-    z = [ _z(ratios.get(f, 0.0), stats[f]["mean"], stats[f]["std"]) for f in FEATURES ]
-    X = np.array([z], dtype=np.float32)  # (1, D)
+# ------------- FastAPI app ----------------
+app = FastAPI(title="Static Fingerprint API", version="1.0.0")
 
-    raw = model.predict(X, verbose=0)    # (1, K) or (1, K-1)
-    if raw.ndim != 2:
-        raise ValueError(f"Unexpected model output shape: {raw.shape}")
-
-    if raw.shape[1] == len(CLASSES) - 1:
-        probs = _coral_probs_from_logits(raw)[0]
-    elif raw.shape[1] == len(CLASSES):
-        probs = raw[0]
-    else:
-        raise ValueError(f"Model output width {raw.shape[1]} incompatible with {len(CLASSES)} classes")
+# Allow Excel / local tools to call the API
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=False,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
 
-    probs = np.maximum(probs, 0.0)
-    s = probs.sum()
-    probs = probs / s if s > 0 else np.ones(len(CLASSES)) / len(CLASSES)
-
-    pred_idx = int(np.argmax(probs))
+@app.get("/")
+def root():
     return {
-        "input_ok": len(missing) == 0,
-        "missing": missing,
-        "probabilities": {CLASSES[i]: float(probs[i]) for i in range(len(CLASSES))},
-        "predicted_state": CLASSES[pred_idx],
+        "message": "Static Fingerprint API is running.",
+        "try": ["GET /health", "POST /predict"],
     }
 
-# -----------------
-# FastAPI app
-# -----------------
-app = FastAPI()
-
 @app.get("/health")
 def health():
-    return {"status": "ok", "features": FEATURES, "classes": CLASSES}
+    return {
+        "status": "ok",
+        "features": FEATURES,
+        "classes": CLASSES,
+        "model_file": MODEL_PATH,
+        "stats_file": STATS_PATH,
+    }
 
 @app.post("/predict")
-def predict(payload: Any):
-    # Accept either {feature:value,...} or [ {feature:value,...} ]
-    if isinstance(payload, list) and len(payload) == 1 and isinstance(payload[0], dict):
-        payload = payload[0]
+async def predict(req: Request):
+    """
+    Body: a single JSON dict mapping feature -> numeric value.
+    Example:
+      {
+        "autosuf_oper": 1.0,
+        "cov_improductiva": 0.9,
+        ...
+      }
+    """
+    payload = await req.json()
     if not isinstance(payload, dict):
-        raise HTTPException(status_code=400, detail="Expected JSON object mapping feature -> value.")
+        return {"error": "Expected a JSON object mapping feature -> value."}
+
+    # Build z-scores in strict model order
+    z = []
+    z_detail = {}
+    missing = []
+    for f in FEATURES:
+        val = payload.get(f, 0)
+        mean = stats[f]["mean"]
+        sd   = stats[f]["std"]
+        zf = _z(val, mean, sd)
+        z.append(zf)
+        z_detail[f] = zf
+        if f not in payload:
+            missing.append(f)
+
+    X = np.array([z], dtype=np.float32)
+    raw = model.predict(X, verbose=0)
+
+    # Detect CORAL (K-1) vs softmax (K)
+    if raw.ndim == 2 and raw.shape[1] == (len(CLASSES) - 1):
+        probs = coral_probs_from_logits(raw)[0]
+    else:
+        probs = raw[0]
+        # If it's not normalized, normalize defensively:
+        s = float(np.sum(probs))
+        if s > 0:
+            probs = probs / s
 
-    try:
-        return _predict_core(payload)
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
+    pred_idx = int(np.argmax(probs))
+    return {
+        "input_ok": (len(missing) == 0),
+        "missing": missing,
+        "z_scores": z_detail,
+        "probabilities": {CLASSES[i]: float(probs[i]) for i in range(len(CLASSES))},
+        "predicted_state": CLASSES[pred_idx],
+    }