app

app.py

@@ -0,0 +1,248 @@
+import gradio as gr
+import pandas as pd
+import os
+import pickle
+from datetime import datetime
+from sklearn import set_config
+set_config(transform_output="pandas")
+
+# PAGE CONFIG
+# page_icon = "💐"
+
+# Setup variables and constants
+# datetime.now().strftime('%d-%m-%Y _ %Hh %Mm %Ss')
+DIRPATH = os.path.dirname(os.path.realpath(__file__))
+tmp_dir = os.path.join(DIRPATH, "src", "assets", "tmp",)
+os.system(f'rm -vf {tmp_dir}/*')
+tmp_df_fp = os.path.join(
+    tmp_dir, f"history_{datetime.now().strftime('%d-%m-%Y _ %Hh %Mm %Ss')}.csv")
+ml_core_fp = os.path.join(DIRPATH, "src", "assets",
+                          "ml", "crop_recommandation2.pkl")
+init_df = pd.DataFrame(
+    {
+        "N": [],
+        "P": [],
+        "K": [],
+        "temperature": [],
+        "humidity": [],
+        "ph": [],
+        "rainfall": [],
+    }
+)
+
+# FUNCTIONS
+
+
+def load_ml_components(fp):
+    "Load the ml component to re-use in app"
+    with open(fp, "rb") as f:
+        object = pickle.load(f)
+    return object
+
+
+def setup(fp):
+    "Setup the required elements like files, models, global variables, etc"
+
+    # history frame
+    if not os.path.exists(fp):
+        df_history = init_df.copy()
+    else:
+        df_history = pd.read_csv(fp)
+
+    df_history.to_csv(fp, index=False)
+
+    return df_history
+
+
+def select_categorical_widget(col_index, col_name, encoder):
+    """This function will return the right widget to use for each categorical feature
+    """
+
+    categories = encoder.categories_[col_index].tolist()
+    n_unique = len(categories)
+
+    # print(
+    #     f"[Info] unique categories for feature {col_name}  ({type(categories)}) are : {categories}")
+
+    if n_unique == 2:
+        print(
+            f"[Info] unique categories for feature '{col_index}' {col_name}  ({type(categories)}) are : {categories}")
+
+        widget = gr.Checkbox(label=f"Enter {col_name}", value=categories)
+    elif n_unique <= 5:
+        widget = gr.Radio(label=f"Enter {col_name}", choices=categories)
+    else:
+        widget = gr.Dropdown(label=f"Enter {col_name}", choices=categories)
+
+    return widget
+
+
+def make_prediction(*args):
+    """Function that takes values from fields to make 1-by-1 prediction
+    """
+    print(
+        f"[Info] input args of the function {args} ")
+    raw = {k: [val if not isinstance(val, list) else val[0]]
+           for val, k in zip(args, num_cols+cat_cols)}
+    print(
+        f"[Info] input modified a bit {raw}\n")
+
+    df_input = pd.DataFrame(raw)
+    global df_history
+
+    # print(f"\n[Info] Input information as dataframe: \n{df_input.to_string()}")
+    df_input.drop_duplicates(inplace=True, ignore_index=True)
+    print(f"\n[Info] Input with duplicated rows: \n{df_input.to_string()}")
+
+    df_input_num, df_input_cat = None, None
+
+    if len(cat_cols) > 0:
+        df_input_cat = df_input[cat_cols].copy()
+        if cat_imputer:
+            df_input_cat = cat_imputer.transform(df_input_cat)
+        if encoder:
+            df_input_cat = encoder.transform(df_input_cat)
+
+    if len(num_cols) > 0:
+        df_input_num = df_input[num_cols].copy()
+        if num_imputer:
+            df_input_num = num_imputer.transform(df_input_num)
+        if scaler:
+            df_input_num = scaler.transform(df_input_num)
+
+    df_input_ok = pd.concat([df_input_num, df_input_cat], axis=1)
+
+    prediction_output = model.predict_proba(df_input_ok)
+
+    output = model.predict_proba(df_input_ok)
+
+    # store confidence score/ probability for the predicted class
+    confidence_score = output.max(axis=-1)
+    df_input["confidence score"] = confidence_score
+
+    # get index of the predicted class
+    predicted_idx = output.argmax(axis=-1)
+
+    # store index then replace by the matching label
+    df_input["predicted crop"] = predicted_idx
+    predicted_label = df_input["predicted crop"].replace(idx_to_labels)
+    df_input["predicted crop"] = predicted_label
+
+    print(
+        f"[Info] Prediction output (of type '{type(prediction_output)}') from passed input: {prediction_output} of shape {prediction_output.shape}")
+
+    # print(f"[Info] Prediction: {prediction_output}")
+    # df_input['prediction'] = prediction_output
+
+    print(
+        f"\n[Info] output information as dataframe: \n{df_input.to_string()}")
+    df_history = pd.concat([df_history, df_input], ignore_index=True).drop_duplicates(
+        ignore_index=True, keep='last')
+    df_history.to_csv(tmp_df_fp, index=False, )
+
+    return df_input
+
+
+def download():
+    return gr.File.update(label="History File",
+                          visible=True,
+                          value=tmp_df_fp)
+
+
+def hide_download():
+    return gr.File.update(label="History File",
+                          visible=False)
+
+
+# Setup execution
+ml_components_dict = load_ml_components(fp=ml_core_fp)
+
+num_cols = [
+    "N",
+    "P",
+    "K",
+    "temperature",
+    "humidity",
+    "ph",
+    "rainfall",
+]
+cat_cols = ml_components_dict['cat_cols'] if 'cat_cols' in ml_components_dict else [
+]
+num_imputer = ml_components_dict['num_imputer'].set_output(transform="pandas") if (
+    'num_cols' in ml_components_dict and 'num_imputer' in ml_components_dict) else None
+cat_imputer = ml_components_dict['cat_imputer'].set_output(transform="pandas") if (
+    'cat_cols' in ml_components_dict and 'cat_imputer' in ml_components_dict) else None
+scaler = ml_components_dict['scaler'].set_output(
+    transform="pandas") if 'scaler' in ml_components_dict else None
+encoder = ml_components_dict['encoder'] if 'encoder' in ml_components_dict else None
+model = ml_components_dict['model']
+labels = ml_components_dict['labels'] if 'labels' in ml_components_dict else []
+idx_to_labels = {i: l for (i, l) in enumerate(labels)}
+end2end_pipeline = ml_components_dict['pipeline']
+print(f"\n[Info] ML components loaded: {list(ml_components_dict.keys())}")
+
+df_history = setup(tmp_df_fp)
+
+
+# APP Interface
+
+# Main page
+demo_inputs = []
+with gr.Blocks() as demo:
+    gr.Markdown('''<img class="center" src="https://www.verdict.co.uk/wp-content/uploads/2018/12/Agri-tech.jpg"  width="60%" height="60%">
+    <style>
+    .center {
+    display: block;
+    margin-left: auto;
+    margin-right: auto;
+    width: 50%;
+    }
+    </style>''')
+    gr.Markdown('''<center><h1> 🚀 Agri-Tech App 🚀 </h1><center>''')
+    gr.Markdown('''
+    This is a ML API for classification of crop to plant on a land regarding some features
+    ''')
+
+    multiple_of = 7
+    print(f"\n[INFO] {init_df.shape[1]}\n")
+
+    with gr.Row():
+        for i in range(0, init_df.shape[1],):
+            demo_inputs.append(gr.Number(label=f"Enter {num_cols[i]}"))
+
+    # with gr.Row():
+    #     for i in range(0,init_df.shape[1], multiple_of):
+    #         gr.Number(label=f"Enter {num_cols[i]}")
+    #         gr.Number(label=f"Enter {num_cols[i+1]}")
+    #         # gr.Number(label=f"Enter {num_cols[i+2]}")
+    output = gr.Dataframe(df_history)
+
+    btn_predict = gr.Button("Predict")
+    btn_predict.click(fn=make_prediction, inputs=demo_inputs, outputs=output)
+
+    file_obj = gr.File(label="History File",
+                       visible=False
+                       )
+
+    btn_download = gr.Button("Download")
+    btn_download.click(fn=download, inputs=[], outputs=file_obj)
+    output.change(fn=hide_download, inputs=[], outputs=file_obj)
+
+# second demo
+# num_inputs = [gr.Number(label=f"Enter {col}") for col in num_cols]
+# cat_inputs = [select_categorical_widget(col_index=i, col_name=col, encoder=encoder) for i, col in enumerate(cat_cols)]
+
+# inputs = num_inputs + cat_inputs
+
+# demo = gr.Interface(
+#     make_prediction,
+#     inputs,
+#     "dataframe", # "number"
+#     examples=[
+#         # [2, "cat", ["Japan", "Pakistan"], "park", ["ate", "swam"], True],
+#     ]
+# )
+
+
+if __name__ == "__main__":
+    demo.launch()