Update app.py

app.py

@@ -1,166 +1,116 @@
-import os
 import json
 import numpy as np
-import gradio as gr
 import tensorflow as tf
+import gradio as gr
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
 
 # ---------- CONFIG ----------
-MODEL_PATH = "best_model.h5"                 # <- you converted & uploaded this
-STATS_PATH = "Means & Std for Excel.json"    # <- keep exact filename
-CLASSES = ["Top", "Mid-Top", "Mid", "Mid-Low", "Low"]  # your group order
+MODEL_PATH = "best_model.h5"                 # or best_model.keras if that’s what you uploaded
+STATS_PATH = "Means & Std for Excel.json"    # exact filename
+CLASSES = ["Top", "Mid-Top", "Mid", "Mid-Low", "Low"]
 # ----------------------------
 
 print("Loading model and stats...")
 model = tf.keras.models.load_model(MODEL_PATH, compile=False)
-
 with open(STATS_PATH, "r") as f:
     stats = json.load(f)
 
-# Feature order: use the order from JSON to remain consistent
 FEATURES = list(stats.keys())
 print("Feature order:", FEATURES)
 
-def _zscore(val: float, mean: float, sd: float) -> float:
-    # Robust z-score (avoid division by zero)
+def _z(val):  # safe z-score
     try:
         v = float(val)
-    except (TypeError, ValueError):
+    except Exception:
         v = 0.0
-    return 0.0 if (sd is None or sd == 0) else (v - mean) / sd
+    return v
 
-def predict_core(ratios: dict):
-    """
-    ratios: dict mapping feature name -> raw numeric ratio.
-    Returns a dict with predicted_state, probabilities, z_scores, missing.
-    """
+def _zscore(val, mean, sd):
+    v = _z(val)
+    return 0.0 if (sd is None or sd == 0) else (v - mean) / sd
 
-    # Validate that all required features are present
-    missing = [f for f in FEATURES if f not in ratios]
+def coral_probs_from_logits(logits_np):
+    import tensorflow as tf
+    logits = tf.convert_to_tensor(logits_np, dtype=tf.float32)  # (1, K-1)
+    sig = tf.math.sigmoid(logits)                               # (1, 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()
 
-    # Build z-score vector (same order as training)
+def predict_core(ratios: dict):
+    # build z vector in fixed order
     zscores = []
-    zscores_dict = {}
+    z_map = {}
     for f in FEATURES:
         mean = stats[f]["mean"]
-        sd = stats[f]["std"]
-        val = ratios.get(f, 0.0)
-        z = _zscore(val, mean, sd)
+        sd   = stats[f]["std"]
+        val  = ratios.get(f, 0.0)
+        z    = _zscore(val, mean, sd)
         zscores.append(z)
-        zscores_dict[f] = z
+        z_map[f] = z
 
-    # Predict logits (shape: (1, K-1))
     X = np.array([zscores], dtype=np.float32)
-    logits = model.predict(X, verbose=0)[0]  # (K-1,)
-
-    # --- CORAL decoding with monotonic enforcement ---
-    # 1. Compute cumulative sigmoids
-    sig = tf.math.sigmoid(logits).numpy()
-
-    # 2. Enforce monotonic decrease: σ1 >= σ2 >= ... >= σ_{K-1}
-    sig_mono = np.minimum.accumulate(sig)
-
-    # 3. Construct boundaries [1, σ1, σ2, ..., σ_{K-1}, 0]
-    edges = np.concatenate(([1.0], sig_mono, [0.0]))
-
-    # 4. Compute adjacent differences -> per-class probabilities
-    probs = edges[:-1] - edges[1:]
-
-    # 5. Normalise just in case of minor floating-point drift
-    s = probs.sum()
-    if s > 0:
-        probs = probs / s
-
-    # Map to class labels
-    probs_dict = {CLASSES[i]: float(probs[i]) for i in range(len(CLASSES))}
-
-    # Determine the predicted state (highest probability)
-    pred_state = max(probs_dict, key=probs_dict.get)
-
-    # Optional: print sanity check for internal testing
-    # print("Sigmoids:", sig)
-    # print("Probabilities:", probs_dict)
-    # print("Sum of probs:", probs.sum())
+    y = model.predict(X, verbose=0)
 
+    # handle either softmax K or CORAL K-1
+    if y.ndim == 2 and y.shape[1] == len(CLASSES):
+        probs = y[0]
+    elif y.ndim == 2 and y.shape[1] == len(CLASSES) - 1:
+        probs = coral_probs_from_logits(y)[0]
+    else:
+        # fallback: normalize positive scores
+        s = y[0].astype(np.float64)
+        if s.ndim == 0:
+            s = np.array([float(s)], dtype=np.float64)
+        s = np.maximum(s, 0.0)
+        probs = s / s.sum() if s.sum() > 0 else np.ones(len(CLASSES)) / len(CLASSES)
+
+    pred_idx = int(np.argmax(probs))
+    pred_state = CLASSES[pred_idx]
     return {
-        "input_ok": len(missing) == 0,
-        "missing": missing,
-        "z_scores": zscores_dict,
-        "probabilities": probs_dict,
+        "input_ok": True,
+        "missing": [f for f in FEATURES if f not in ratios],
+        "z_scores": z_map,
+        "probabilities": {CLASSES[i]: float(probs[i]) for i in range(len(CLASSES))},
         "predicted_state": pred_state
     }
-# ---------- Gradio adapter ----------
-def predict_from_json(payload, x_api_key: str = ""):
-    """
-    Gradio will pass the JSON from the UI or /run/predict.
-    We accept either:
-      { ...feature: value... }
-    or   [ { ...feature: value... } ]  (unwrap common API client shape)
-    """
 
-    # Unwrap if list-of-one was sent
+def predict_from_json(payload):
+    # accept raw dict OR list-of-one dict
     if isinstance(payload, list) and len(payload) == 1 and isinstance(payload[0], dict):
         payload = payload[0]
-
     if not isinstance(payload, dict):
-        return {"error": "Invalid payload: expected a JSON object mapping feature -> value."}
-
+        return {"error": "Invalid payload: send a JSON object mapping feature->value."}
     return predict_core(payload)
 
-# ---------- AUTOLOAD DEMO INPUT ----------
-DEMO_PATH = "example_input.json"  # <-- name of your uploaded file
-demo_data = {}
+# ------------------ FastAPI + Gradio ------------------
+app = FastAPI()
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"], allow_methods=["*"], allow_headers=["*"],
+)
 
-if os.path.exists(DEMO_PATH):
-    try:
-        with open(DEMO_PATH, "r") as f:
-            demo_data = json.load(f)
-        # Unwrap if the file uses {"data":[{...}]} format
-        if isinstance(demo_data, dict) and "data" in demo_data and isinstance(demo_data["data"], list):
-            demo_data = demo_data["data"][0]
-    except Exception as e:
-        print("⚠️ Could not load demo input:", e)
-else:
-    # Default zeros if file not found
-    demo_data = {f: 0.0 for f in FEATURES}
-# -----------------------------------------
+# Plain REST endpoint for Excel/VBA (raw dict)
+@app.post("/predict")
+async def api_predict(req: Request):
+    body = await req.json()
+    if isinstance(body, dict) and "data" in body and isinstance(body["data"], list) and body["data"]:
+        body = body["data"][0]  # unwrap gradio shape
+    return predict_from_json(body)
 
+# Optional health check
+@app.get("/health")
+def health():
+    return {"ok": True}
 
-# ---------- DEFINE GRADIO INTERFACE ----------
-# --- UI (unchanged) ---
-iface = gr.Interface(
+# Mount UI at root
+ui = gr.Interface(
     fn=predict_from_json,
     inputs=gr.JSON(label="ratios JSON (dict of feature -> value)"),
     outputs="json",
     title="Static Fingerprint Model API",
     description="POST your 21 ratios as a JSON dict. Returns probabilities + predicted state."
 )
-
-# --- FastAPI app with a simple REST endpoint ---
-from fastapi import FastAPI, Request
-import gradio as gr
-
-app = FastAPI()
-# Mount the Gradio UI at the root so your Space page still works
-app = gr.mount_gradio_app(app, iface, path="/")
-
-@app.post("/predict")
-async def api_predict(req: Request):
-    """
-    Accepts either:
-      1) raw dict:       {"autosuf_oper":1.2, ...}
-      2) gradio format:  {"data":[{...}]}
-    """
-    body = await req.json()
-    if isinstance(body, dict) and "data" in body and isinstance(body["data"], list) and body["data"]:
-        payload = body["data"][0]    # unwrap Gradio shape
-    elif isinstance(body, dict):
-        payload = body               # raw dict
-    else:
-        return {"error": "Invalid payload. Send a JSON object of feature->value OR {'data':[that_object]}"}
-
-    try:
-        return predict_from_json(payload)  # reuse your existing function
-    except Exception as e:
-        return {"error": f"{type(e).__name__}: {e}"}
-
-# Spaces will auto-run with uvicorn; no need to call launch() here.
+app = gr.mount_gradio_app(app, ui, path="/")