File size: 2,871 Bytes
689a7a4 8ea1efb 689a7a4 |
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 82 83 84 85 86 87 |
from fastapi import FastAPI
from pydantic import BaseModel
import pickle
import pandas as pd
import numpy as np
import uvicorn
# call the app
app = FastAPI(title="API")
# Load the model and scaler
def load_model_and_scaler():
with open("model.pkl", "rb") as f1, open("scaler.pkl", "rb") as f2:
return pickle.load(f1), pickle.load(f2)
model, scaler = load_model_and_scaler()
def predict(df, endpoint="simple"):
# Scaling
scaled_df = scaler.transform(df) # Scale the input data using a pre-defined scaler
# Prediction
prediction = model.predict_proba(scaled_df) # Make predictions using a pre-trained model
highest_proba = prediction.max(axis=1) # Get the highest probability for each prediction
# Assign predicted labels based on the highest probabilities
predicted_labels = ["Patient does not have sepsis" if i == 0 else "Patient has sepsis" for i in highest_proba]
print(f"Predicted labels: {predicted_labels}") # Print the predicted labels to the terminal
print(highest_proba) # Print the highest probabilities to the terminal
response = []
for label, proba in zip(predicted_labels, highest_proba):
# Create a response for each prediction with the predicted label and probability
output = {
"prediction": label,
"probability of prediction": str(round(proba * 100)) + '%' # Convert the probability to a percentage
}
response.append(output) # Add the response to the list of responses
return response # Return the list of responses
class Patient(BaseModel):
Blood_Work_R1: int
Blood_Pressure: int
Blood_Work_R3: int
BMI: float
Blood_Work_R4: float
Patient_age: int
class Patients(BaseModel):
all_patients: list[Patient]
@classmethod
def return_list_of_dict(cls, patients: "Patients"):
patient_list = []
for patient in patients.all_patients: #for each item in all_patients,
patient_dict = patient.dict() #convert to a dictionary
patient_list.append(patient_dict) #add it to the empty list called patient_list
return patient_list
# Endpoints
# Root Endpoint
@app.get("/")
def root():
return {"API": "This is an API for sepsis prediction."}
# Prediction endpoint
@app.post("/predict")
def predict_sepsis(patient: Patient):
# Make prediction
data = pd.DataFrame(patient.dict(), index=[0])
parsed = predict(df=data)
return {"output": parsed}
# Multiple Prediction Endpoint
@app.post("/predict_multiple")
def predict_sepsis_for_multiple_patients(patients: Patients):
"""Make prediction with the passed data"""
data = pd.DataFrame(Patients.return_list_of_dict(patients))
parsed = predict(df=data, endpoint="multi")
return {"output": parsed}
if __name__ == "__main__":
uvicorn.run("main:app", reload=True)
|