Upload app.py with huggingface_hub

app.py

@@ -0,0 +1,80 @@
+import joblib
+import pandas as pd
+import numpy as np
+from fastapi import FastAPI
+from pydantic import BaseModel
+from typing import List, Optional
+from huggingface_hub import hf_hub_download
+from sklearn.impute import SimpleImputer
+
+# --- Constants ---
+REPO_ID = "DP1110/mlp-accessibility-model"
+MODEL_FILENAME = 'mlp_regressor_model.joblib'
+IMPUTER_FILENAME = 'simple_imputer.joblib'
+
+FEATURE_COLUMNS = ['% ASF (Euclidean)', '% Built-Up Area', '% ASF (Network)', '% ASF from Bus Stops ', '% ASF from Bus Stops', '% ASF (Network) ']
+
+# --- Load Model and Imputer ---
+loaded_mlp_model = None
+loaded_imputer = None
+
+try:
+    model_path = hf_hub_download(repo_id=REPO_ID, filename=MODEL_FILENAME)
+    imputer_path = hf_hub_download(repo_id=REPO_ID, filename=IMPUTER_FILENAME)
+
+    loaded_mlp_model = joblib.load(model_path)
+    loaded_imputer = joblib.load(imputer_path)
+    print("Model and imputer loaded successfully!")
+except Exception as e:
+    print(f"Error loading model or imputer: {e}")
+
+# --- FastAPI Application ---
+app = FastAPI()
+
+# --- Pydantic Input Data Model ---
+class InputData(BaseModel):
+    perc_ASF_Euclidean: Optional[float] = None # Example: 0.5
+    perc_Built_Up_Area: Optional[float] = None # Example: 0.5
+    perc_ASF_Network: Optional[float] = None # Example: 0.5
+    perc_ASF_from_Bus_Stops_: Optional[float] = None # Example: 0.5
+    perc_ASF_from_Bus_Stops: Optional[float] = None # Example: 0.5
+    perc_ASF_Network_: Optional[float] = None # Example: 0.5
+
+# --- Prediction Endpoint ---
+@app.post("/predict")
+async def predict(data: InputData):
+    if loaded_mlp_model is None or loaded_imputer is None:
+        return {"error": "Model or imputer not loaded."}
+
+    # Convert input data to pandas DataFrame
+    input_dict = data.dict()
+
+    # Reconstruct input_row with original feature names for imputer/model input
+    input_row = {}
+    for col in FEATURE_COLUMNS:
+        sanitized_col = col.replace(' ', '_').replace('-', '_').replace('%', 'perc').replace('(', '').replace(')', '')
+        input_row[col] = input_dict[sanitized_col]
+
+    input_df = pd.DataFrame([input_row])
+
+    # Ensure column order matches training features
+    input_df = input_df[FEATURE_COLUMNS]
+
+    # Impute missing values
+    input_imputed = loaded_imputer.transform(input_df)
+    input_imputed_df = pd.DataFrame(input_imputed, columns=FEATURE_COLUMNS)
+
+    # Make prediction
+    prediction = loaded_mlp_model.predict(input_imputed_df)[0]
+
+    return {"predicted_overall_accessibility_score": prediction}
+
+# --- Health Check Endpoint ---
+@app.get("/health")
+async def health_check():
+    return {"status": "ok", "model_loaded": loaded_mlp_model is not None, "imputer_loaded": loaded_imputer is not None}
+
+if __name__ == '__main__':
+    import uvicorn
+    # Run the FastAPI application using Uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)