Update app.py

app.py

@@ -1,118 +1,74 @@
 from flask import Flask, request, jsonify, make_response
-import joblib
-import pandas as pd
-import numpy as np
-import os
-from typing import Union, List, Dict
+import joblib, pandas as pd, numpy as np, os, time, sys
 
+print("==> [BOOT] Starting app.py", flush=True)
 app = Flask(__name__)
 
-# -----------------------------
-# Config / Model Load
-# -----------------------------
-MODEL_PATH = os.getenv("MODEL_PATH", "best_model_random_forest.joblib")
+MODEL_PATH = os.getenv("MODEL_PATH", "superkart_sales_forecast.joblib")
 PORT = int(os.getenv("PORT", "5000"))
 
+print(f"==> [BOOT] MODEL_PATH={MODEL_PATH}", flush=True)
+
 def load_pipeline(path: str):
+    t0 = time.time()
     if not os.path.exists(path):
-        raise FileNotFoundError(
-            f"Model file not found: {path}. "
-            "Ensure the artifact exists in the Space root and Dockerfile COPY line matches."
-        )
+        raise FileNotFoundError(f"Model file not found: {path}")
+    print(f"==> [LOAD] Loading model from {path} ...", flush=True)
     bundle = joblib.load(path)
-    # Support both a dict bundle {"pipeline": pipe} and a bare fitted pipeline
-    return bundle["pipeline"] if isinstance(bundle, dict) and "pipeline" in bundle else bundle
+    if isinstance(bundle, dict) and "pipeline" in bundle:
+        pipe = bundle["pipeline"]
+        print("==> [LOAD] Loaded dict bundle with 'pipeline'", flush=True)
+    else:
+        pipe = bundle
+        print("==> [LOAD] Loaded pipeline object", flush=True)
+    print(f"==> [LOAD] Done in {time.time()-t0:.2f}s", flush=True)
+    return pipe
 
 try:
     pipe = load_pipeline(MODEL_PATH)
     MODEL_READY = True
     LOAD_ERROR = None
 except Exception as e:
+    print("==> [ERROR] Model load failed:", e, file=sys.stderr, flush=True)
     pipe = None
     MODEL_READY = False
     LOAD_ERROR = str(e)
 
-# -----------------------------
-# CORS (no external dependency)
-# -----------------------------
 @app.after_request
-def add_cors_headers(response):
-    # Allow Streamlit frontend from a different domain to call this API
-    response.headers["Access-Control-Allow-Origin"] = "*"
-    response.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"
-    response.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS"
-    return response
+def cors(resp):
+    resp.headers["Access-Control-Allow-Origin"] = "*"
+    resp.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"
+    resp.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS"
+    return resp
 
-@app.route("/", methods=["GET"])
+@app.get("/")
 def root():
-    return jsonify({
-        "service": "SuperKart Sales Forecast API",
-        "health": "/health",
-        "predict": "/predict",
-        "model_path": MODEL_PATH
-    })
+    return jsonify({"service": "SuperKart API", "health": "/health", "predict": "/predict", "model_path": MODEL_PATH})
 
-@app.route("/health", methods=["GET"])
+@app.get("/health")
 def health():
-    status = "ok" if MODEL_READY else "error"
-    payload = {"status": status, "model_path": MODEL_PATH}
-    if LOAD_ERROR:
-        payload["error"] = LOAD_ERROR
-    return jsonify(payload), (200 if MODEL_READY else 500)
+    return (jsonify({"status": "ok", "model_path": MODEL_PATH}), 200) if MODEL_READY \
+        else (jsonify({"status": "error", "model_path": MODEL_PATH, "error": LOAD_ERROR}), 500)
 
 @app.route("/predict", methods=["OPTIONS"])
 def predict_options():
-    # Handle CORS preflight cleanly
     return make_response(("", 204))
 
-# -----------------------------
-# Helpers
-# -----------------------------
-def _to_frame(payload: Union[Dict, List[Dict]]) -> pd.DataFrame:
-    # Accept a single dict or a list of dicts
-    if isinstance(payload, dict):
-        return pd.DataFrame([payload])
-    if isinstance(payload, list):
-        if not all(isinstance(x, dict) for x in payload):
-            raise ValueError("If sending a list, it must be a list of JSON objects.")
-        return pd.DataFrame(payload)
-    raise ValueError("Input JSON must be an object or a list of objects.")
-
-def _to_pyfloat(x):
-    # Ensure JSON-serializable floats
-    if isinstance(x, (np.floating,)):
-        return float(x)
-    return float(x)
-
-# -----------------------------
-# Prediction
-# -----------------------------
-@app.route("/predict", methods=["POST"])
+@app.post("/predict")
 def predict():
     if not MODEL_READY or pipe is None:
         return jsonify({"error": "Model not loaded", "details": LOAD_ERROR}), 503
+    data = request.get_json(force=True)
+    if isinstance(data, dict):
+        df = pd.DataFrame([data])
+    elif isinstance(data, list):
+        df = pd.DataFrame(data)
+    else:
+        return jsonify({"error": "Payload must be an object or list of objects"}), 400
+    preds = pipe.predict(df)
+    preds = [float(x) if isinstance(x, (np.floating, float, int)) else x for x in preds]
+    return jsonify({"predictions": preds})
 
-    try:
-        data = request.get_json(force=True, silent=False)
-        if data is None:
-            return jsonify({"error": "No JSON payload received"}), 400
-
-        df = _to_frame(data)
-
-        # Let the sklearn pipeline handle all preprocessing (encoders, etc.)
-        preds = pipe.predict(df)
-
-        # Cast to built-in float for clean JSON
-        preds = [_to_pyfloat(p) for p in preds]
-        return jsonify({"predictions": preds})
-
-    except ValueError as ve:
-        return jsonify({"error": "Invalid input", "details": str(ve)}), 400
-    except Exception as e:
-        return jsonify({"error": "Prediction failed", "details": str(e)}), 500
-
-# -----------------------------
-# Local dev entrypoint (prod uses gunicorn)
-# -----------------------------
 if __name__ == "__main__":
-    app.run(host="0.0.0.0", port=PORT)
+    print("==> [RUN] Flask dev server starting on 0.0.0.0:5000", flush=True)
+    app.run(host="0.0.0.0", port=PORT)