Spaces:
Sleeping
Sleeping
File size: 4,157 Bytes
6272ed2 5bd316b 6272ed2 5bd316b 6272ed2 5bd316b 6272ed2 |
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 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 |
from fastapi import FastAPI
import uvicorn
from typing import List, Literal, Optional
from pydantic import BaseModel
import pandas as pd
import pickle
import os
import json
import logging
# logger
logging.basicConfig(format='%(levelname)s:%(message)s', level=logging.DEBUG)
# Util Functions & Classes
def loading(fp):
with open(fp, "rb") as f:
data = pickle.load(f)
print(f"INFO: Loaded data : {data}")
return data
def predict(df, endpoint="simple"):
"""Take a dataframe as input and use it to make predictions"""
print(
f"[Info] 'predict' function has been called through the endpoint '{endpoint}'.\n"
)
logging.info(f" \n{df.to_markdown()}")
# scaling
scaled_df = scaler.transform(df)
logging.info(f" Scaler output is of type {type(scaled_df)}")
# prediction
prediction = model.predict_proba(scaled_df)
print(f"INFO: Prediction output: {prediction}")
# Formatting of the prediction
## extract highest proba
highest_proba = prediction.max(axis=1)
print(f"INFO: Highest probabilities : {highest_proba}")
## extract indexes of the highest proba
highest_proba_idx = prediction.argmax(axis=1)
print(f"INFO: Highest probability indexes : {highest_proba_idx}")
## Maching prediction with classes
predicted_classes = [labels[i] for i in highest_proba_idx]
print(f"INFO: Predicted classes : {predicted_classes}")
# prediction[:, highest_proba_idx]
# save in df
df["predicted proba"] = highest_proba
df["predicted label"] = predicted_classes
print(f"INFO: dataframe filled with prediction\n{df.to_markdown()}\n")
# parsing prediction
# parsed = json.loads(df.to_json(orient="index")) # or
parsed = df.to_dict("records")
return parsed
## INPUT MODELING
class Land(BaseModel):
"""Modeling of one input data in a type-restricted dictionary-like format
column_name : variable type # strictly respect the name in the dataframe header.
eg.:
=========
customer_age : int
gender : Literal['male', 'female', 'other']
"""
N: float
P: float
K: float
temperature: float
humidity: float
ph: float
rainfall: float
class Lands(BaseModel):
inputs: List[Land]
def return_list_of_dict(
cls,
):
# return [land.dict() for land in cls.inputs]
return [i.dict() for i in cls.inputs]
# API Config
app = FastAPI(title="Agri-Tech API",
description="This is a ML API for classification of crop to plant on a land regarding some features")
# ML Config
ml_objects = loading(fp=os.path.join("assets", "ml", "crop_recommandation2.pkl"))
## Extract the ml components
model = ml_objects["model"]
scaler = ml_objects["scaler"].set_output(transform="pandas")
labels = ml_objects["labels"]
# Endpoints
@app.get("/")
def root():
return {"Description": " This is a ML API for classification of crop to plant on a land regarding some features."}
@app.get("/checkup")
def test(a: Optional[int], b: int):
return {"a": a, "b": b}
## ML endpoint
@app.post("/predict")
def make_prediction(
N: float,
P: float,
K: float,
temperature: float,
humidity: float,
ph: float,
rainfall: float,
):
"""Make prediction with the passed data"""
df = pd.DataFrame(
{
"N": [N],
"P": [P],
"K": [K],
"temperature": [temperature],
"humidity": [humidity],
"ph": [ph],
"rainfall": [rainfall],
}
)
parsed = predict(df=df) # df.to_dict('records')
return {
"output": parsed,
}
@app.post("/predict_multi")
def make_multi_prediction(multi_lands: Lands):
"""Make prediction with the passed data"""
print(f"Mutiple inputs passed: {multi_lands}\n")
df = pd.DataFrame(multi_lands.return_list_of_dict())
parsed = predict(df=df, endpoint="multi inputs") # df.to_dict('records')
return {
"output": parsed,
"author": "Stella Archar",
"api_version": ";)",
}
if __name__ == "__main__":
uvicorn.run("main:app", reload=True)
|