File size: 2,856 Bytes
d40f6d9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 |
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)
|