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)