Spaces:
Sleeping
Sleeping
File size: 4,823 Bytes
1b5b1a6 |
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 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 |
import os
import sys
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
import uvicorn
from fastapi import FastAPI, Request, File, UploadFile
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from src.utils import load_pickle, make_prediction, process_label, process_json_csv, output_batch, return_columns
from src.module import Inputs
import pandas as pd
import numpy as np
from typing import List
# Create an instance of FastAPI
app = FastAPI(debug=True)
DIRPATH = os.path.dirname(os.path.realpath(__file__))
model_path = os.path.join(DIRPATH, '..', 'assets', 'ml_components', 'model-1.pkl')
transformer_path = os.path.join(DIRPATH, '..', 'assets', 'ml_components', 'preprocessor.pkl')
properties_path = os.path.join(DIRPATH, '..', 'assets', 'ml_components', 'properties.pkl')
# Load the trained model, pipeline, and other properties
model = load_pickle(model_path)
transformer = load_pickle(transformer_path)
properties = load_pickle(properties_path)
# Configure static and template files
app.mount("/static", StaticFiles(directory="src/app/static"), name="static") # Mount static files
templates = Jinja2Templates(directory="src/app/templates") # Mount templates for HTML
# Root endpoint to serve index.html template
@app.get("/", response_class=HTMLResponse)
async def root(request: Request):
return templates.TemplateResponse("index.html", {'request': request})
# Health check endpoint
@app.get("/health")
def check_health():
return {"status": "ok"}
# Model information endpoint
@app.post('/model-info')
async def model_info():
model_name = model.__class__.__name__
model_params = model.get_params()
features = properties['train features']
print(features)
model_information = {'model info': {
'model name ': model_name,
'model parameters': model_params,
'train feature': features}
}
return model_information
# Prediction endpoint
@app.post('/predict')
async def predict(plasma_glucose: float, blood_work_result_1: float,
blood_pressure: float, blood_work_result_2: float,
blood_work_result_3: float, body_mass_index: float,
blood_work_result_4: float, age: int, insurance: bool):
# Create a dataframe from inputs
data = pd.DataFrame([[plasma_glucose,blood_work_result_1,blood_pressure,
blood_work_result_2,blood_work_result_3,body_mass_index,
blood_work_result_4, age,insurance]], columns=return_columns())
data_copy = data.copy() # Create a copy of the dataframe
label, prob = make_prediction(data, transformer, model) # Get the labels
data_copy['Predicted Label'] = label[0] # Get the labels from making a prediction
data_copy['Predicted Label'] = data_copy.apply(process_label, axis=1)
inputs = data.to_dict('index') # Convert dataframe to dictionary
outputs = data_copy[['Predicted Label']].to_dict('index')
response = {'inputs': inputs,
'outputs': outputs}
return response
# Batch prediction endpoint
@app.post('/predict-batch')
async def predict_batch(inputs: Inputs):
# Create a dataframe from inputs
data = pd.DataFrame(inputs.return_dict_inputs())
data_copy = data.copy() # Create a copy of the data
labels, probs = make_prediction(data, transformer, model) # Get the labels
data_labels = pd.DataFrame(labels, columns=['Predicted Label'])
data_labels['Predicted Label'] = data_labels.apply(process_label, axis=1)
response = output_batch(data, data_labels)
return response
# Upload data endpoint
@app.post("/upload-data")
async def upload_data(file: UploadFile = File(...)):
file_type = file.content_type
print(f'INFO {file_type}')
valid_formats = ['text/csv', 'application/json']
if file_type not in valid_formats:
return JSONResponse(content={"error": f"Invalid file format. Must be one of: {', '.join(valid_formats)}"})
else:
contents = await file.read()
data= process_json_csv(contents=contents,file_type=file_type, valid_formats=valid_formats)
data_copy = data.copy() # Create a copy of the data
labels, probs = make_prediction(data, transformer, model) # Get the labels
data_copy['Predicted Label'] = labels# Create the predicted label column
data_copy['Predicted Label'] = data_copy.apply(process_label, axis=1)
data_dict = data_copy.to_dict('index') # Convert data to a dictionary
# print(data_dict.index)
return {'outputs': data_dict}
# Run the FastAPI application
if __name__ == '__main__':
uvicorn.run('app:app', reload=True)
|