app.py

""" Gradio app for biomass and carbon stock estimation. """
from pathlib import Path

import joblib
import numpy as np
import pandas as pd
import gradio as gr

from utils import check_inside_civ, initialize_ee, extract_from_gee

initialize_ee()

def load_pipeline():
    """ Load the best model pipeline. """
    path = Path("data/best_model_c_pipeline.pkl")
    if not path.exists():
        raise RuntimeError("Model not found: data/best_model_c_pipeline.pkl")

    return joblib.load(path)

PIPE = load_pipeline()

def predict_agbd(df_row: pd.DataFrame) -> float:
    """ Predict AGBD from a DataFrame row. """
    y = PIPE.predict(df_row)
    return float(np.asarray(y).ravel()[0])

def carbon_from_agbd(agbd_t_ha: float, cf: float = 0.5) -> float:
    """ Convert AGBD to carbon stock. """
    return float(agbd_t_ha * cf)

def run_all(latitude: float, longitude: float):
    """ Run the app. """
    inside = check_inside_civ(latitude, longitude)
    if not inside:
        return ("n/a", "n/a", "Choose a location inside Côte d'Ivoire.")
    try:
        data, err = extract_from_gee(latitude, longitude)
        if err is not None or data is None:
            return ("n/a", "n/a", "n/a")
        cloud_cov = data["cloud"]
        estimation_date = data["estimation_date"]
        X = data["X"]
        agb = predict_agbd(X)
        carbon = carbon_from_agbd(agb)
        return (f"{estimation_date}",
                f"{cloud_cov:.2f} %",
                f"{agb:.2f} t/ha",
                f"{carbon:.2f} tC/ha")
    except Exception as e:
        return ("error", "", str(e))


with gr.Blocks(theme=gr.themes.Ocean(), fill_height=True) as demo:
    gr.Markdown("🌴 **BCSE** : Biomass and Carbon Stock Estimation using ML and satellite images (Sentinel-1/2, DEM) 🌴")

    with gr.Row():
        with gr.Column(scale=1):
            lat = gr.Number(label="lat", value=5.280498, precision=6)
            lon = gr.Number(label="lon", value=-4.089883, precision=6)
            with gr.Row():
                clear = gr.Button("Clear")
                submit = gr.Button("Submit", variant="primary")
        with gr.Column(scale=1):
            est_date = gr.Textbox(label="Date of estimation")
            cloud = gr.Textbox(label="Cloud coverage")
            agbd  = gr.Textbox(label="Above ground biomass density (AGBD) t/ha")
            cstock = gr.Textbox(label="Carbon stock density tC/ha")

    def _on_submit(la, lo):
        return run_all(la, lo)
    submit.click(_on_submit, [lat, lon], [est_date, cloud, agbd, cstock]) # pylint: disable=no-member

    def _on_clear():
        return None, None, None, None, None
    clear.click(_on_clear, outputs=[lat, lon, est_date, cloud, agbd, cstock]) # pylint: disable=no-member

if __name__ == "__main__":
    demo.launch(share=True)