Update app.py

app.py

@@ -1,20 +1,34 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel, Field, validator
+from typing import List, Dict, Any
 import os, json, joblib, numpy as np, pandas as pd, threading
 from huggingface_hub import snapshot_download
 import xgboost as xgb
 from pathlib import Path
 
+# -------------------------------
+# Hugging Face repo config
+# -------------------------------
 HF_CACHE_DIR = os.getenv("HF_CACHE_DIR", "/models/hf")
 HF_REPO_ID   = os.getenv("HF_REPO_ID", "ethnmcl/test-score-predictor-xgb")
-HF_TOKEN     = os.getenv("HF_TOKEN", None)  # set as Space secret for private repos
+HF_TOKEN     = os.getenv("HF_TOKEN", None)  # only needed if repo is private
 
-_loaded_lock = threading.Lock()
+# -------------------------------
+# Global state
+# -------------------------------
 _loaded = False
+_loaded_lock = threading.Lock()
 _pre = None
 _weights = None
 _schema = None
 _model = None
 
+# -------------------------------
+# Loader functions
+# -------------------------------
 def repo_snapshot(repo_id: str = None) -> str:
+    """Download model repo snapshot (if not cached)."""
     repo_id = repo_id or HF_REPO_ID
     local_dir = snapshot_download(
         repo_id=repo_id,
@@ -26,6 +40,7 @@ def repo_snapshot(repo_id: str = None) -> str:
     return local_dir
 
 def load_model():
+    """Load preprocessor, weights, schema, and XGB model into memory."""
     global _loaded, _pre, _weights, _schema, _model
     if _loaded:
         return
@@ -46,7 +61,7 @@ def _transform(records):
     df = pd.DataFrame(records, columns=num + cat)
     Xt = _pre.transform(df)
     Xt = Xt.astype(float, copy=False)
-    Xt[:, :len(num)] *= _weights  # post-transform numeric weighting
+    Xt[:, :len(num)] *= _weights
     return Xt
 
 def predict_one(record: dict) -> float:
@@ -54,7 +69,7 @@ def predict_one(record: dict) -> float:
         load_model()
     Xt = _transform([record])
     pred = float(_model.predict(Xt)[0])
-    return max(50.0, min(100.0, pred))  # optional clamp to match dataset range
+    return max(50.0, min(100.0, pred))  # clamp to dataset range
 
 def predict_batch(records: list) -> np.ndarray:
     if not _loaded:
@@ -63,3 +78,56 @@ def predict_batch(records: list) -> np.ndarray:
     preds = _model.predict(Xt)
     return np.clip(preds, 50.0, 100.0)
 
+# -------------------------------
+# FastAPI app
+# -------------------------------
+app = FastAPI(title="Test Score Predictor API", version="1.0.0")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"], allow_credentials=True,
+    allow_methods=["*"], allow_headers=["*"],
+)
+
+@app.on_event("startup")
+def _startup():
+    repo_snapshot(HF_REPO_ID)
+    load_model()
+
+# -------------------------------
+# Request schemas
+# -------------------------------
+class Record(BaseModel):
+    Subject: str = Field(..., examples=["Mathematics"])
+    Current_Grade: int = Field(..., ge=60, le=98)
+    Max_Test_Percentage: int = Field(..., ge=65, le=100)
+    Days_Preparing: int = Field(..., ge=1, le=14)
+    Hours_Studied: int = Field(..., ge=2, le=50)
+    Study_Session_Average: float = Field(..., ge=0.1, le=10.0)
+    Avg_Previous_Tests: int = Field(..., ge=55, le=95)
+    Test_Difficulty: str = Field(..., examples=["Easy (20)", "Medium (30)", "Hard (50)"])
+
+    @validator("Study_Session_Average", always=True)
+    def recompute_session_avg(cls, v, values):
+        if "Hours_Studied" in values and "Days_Preparing" in values:
+            return round(values["Hours_Studied"] / values["Days_Preparing"], 1)
+        return v
+
+class PredictRequest(BaseModel):
+    data: List[Record]
+
+# -------------------------------
+# Routes
+# -------------------------------
+@app.get("/health")
+def health() -> Dict[str, Any]:
+    return {"status": "ok", "repo": HF_REPO_ID}
+
+@app.post("/predict")
+def predict(req: Record) -> Dict[str, Any]:
+    return {"predicted_score": predict_one(req.dict())}
+
+@app.post("/predict-batch")
+def predict_many(req: PredictRequest) -> Dict[str, Any]:
+    recs = [r.dict() for r in req.data]
+    return {"predicted_scores": predict_batch(recs).tolist(), "count": len(recs)}