app.py

import gradio as gr
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
import tensorflow as tf
import pandas as pd
import numpy as np
import joblib
from load import *
from helper import *
from matplotlib import pyplot as plt

def predict_fire(temp, temp_unit, humidity, wind, wind_unit, veg, elev, elev_unit, use_trust):
    input_data = {
        "temperature": convert_temperature(temp, temp_unit),
        "humidity": humidity,
        "wind_speed": convert_wind_speed(wind, wind_unit),
        "vegetation_index": veg,
        "elevation": convert_elevation(elev, elev_unit)
    }

    input_df = pd.DataFrame([input_data])
    base_prob = FireNet.predict(input_df)[0][0]
    if use_trust:
        trust_score = FireTrustNet.predict(FireScaler.transform(input_df))[0][0]
        final = np.clip(base_prob * trust_score, 0, 1)
    else:
        final = base_prob
    if final > 0.49:
        verdict = "🔥 FIRE LIKELY"
    elif final > 0.43 and final < 0.50:
        verdict = "⚠️ Fire Possible"
    else:
        verdict = "🌿 Fire Unlikely"
    return f"{verdict} ({final:.2f})"

def predict_flood(rainfall_val, rainfall_unit, water_level_val, elevation_val, elev_unit,
                  slope_val, distance_val, distance_unit, use_trustnet):
    # Unit conversion
    rainfall = convert_rainfall(rainfall_val, rainfall_unit)
    elevation = convert_elevation(elevation_val, elev_unit)
    distance = convert_distance(distance_val, distance_unit)

    # Construct input for FloodNet
    base_df = pd.DataFrame([{
        "Rainfall": rainfall,
        "Water Level": water_level_val,
        "Elevation": elevation,
        "Slope": slope_val,
        "Distance from River": distance
    }])

    base_prob = FloodNet.predict(base_df)[0][0]

    if use_trustnet:
        trust_df = pd.DataFrame([{
            "rainfall": rainfall,
            "water_level": water_level_val,
            "elevation": elevation,
            "slope": slope_val,
            "distance_from_river": distance
        }])
        trust_score = FloodTrustNet.predict(FloodScaler.transform(trust_df))[0][0]
        final = np.clip(base_prob * trust_score, 0, 1)
    else:
        final = base_prob

    if final > 0.49:
        verdict = "🏞️ FV-FLOOD LIKELY"
    elif final > 0.43 and final < 0.50:
        verdict = "⚠️ FV-Flood Possible"
    else:
        verdict = "🌿 FV-Flood Unlikely"
    return f"{verdict} ({final:.2f})"

def generate_plot(axis, use_trustnet):
    sweep_values = np.linspace({
        "temperature": (280, 320),
        "humidity": (0, 100),
        "wind_speed": (0, 50),
        "vegetation_index": (0.0, 2.0),
        "elevation": (0, 3000)
    }[axis][0], {
        "temperature": (280, 320),
        "humidity": (0, 100),
        "wind_speed": (0, 50),
        "vegetation_index": (0.0, 2.0),
        "elevation": (0, 3000)
    }[axis][1], 100)

    base_input = {
        "temperature": 300.0,
        "humidity": 30.0,
        "wind_speed": 10.0,
        "vegetation_index": 1.0,
        "elevation": 500.0
    }

    sweep_df = pd.DataFrame([{
        **base_input,
        axis: val
    } for val in sweep_values])

    raw_probs = FireNet.predict(sweep_df).flatten()
    if use_trustnet:
        trust_mods = FireTrustNet.predict(FireScaler.transform(sweep_df)).flatten()
        adjusted_probs = np.clip(raw_probs * trust_mods, 0, 1)
    else:
        adjusted_probs = raw_probs

    fig, ax = plt.subplots()
    ax.plot(sweep_values, raw_probs, "--", color="gray", label="Base Model")
    if use_trustnet:
        ax.plot(sweep_values, adjusted_probs, color="orangered", label="With FireTrustNet")
    ax.set_xlabel(axis.replace("_", " ").title())
    ax.set_ylabel("Fire Probability")
    ax.set_title(f"Fire Probability vs. {axis.replace('_', ' ').title()}")
    ax.legend()
    ax.grid(True)
    return fig

def generate_flood_plot(axis, use_trustnet):
    sweep_range = {
        "rainfall": (0, 150),
        "water_level": (0, 8000),
        "elevation": (0, 20),
        "slope": (0, 20),
        "distance_from_river": (0, 2000)
    }

    values = np.linspace(*sweep_range[axis], 100)

    base_example = {
        "rainfall": 50.0,
        "water_level": 3000.0,
        "elevation": 5.0,
        "slope": 2.0,
        "distance_from_river": 100.0
    }

    # Build test cases by sweeping one input
    inputs = pd.DataFrame([
        {**base_example, axis: v} for v in values
    ])

    # Predict with FloodNet
    floodnet_inputs = inputs.rename(columns={
        "rainfall": "Rainfall",
        "water_level": "Water Level",
        "elevation": "Elevation",
        "slope": "Slope",
        "distance_from_river": "Distance from River"
    })

    base_probs = FloodNet.predict(floodnet_inputs).flatten()

    if use_trustnet:
        trust_inputs = inputs.copy()
        trust_scores = FloodTrustNet.predict(FloodScaler.transform(trust_inputs)).flatten()
        modulated_probs = np.clip(base_probs * trust_scores, 0, 1)
    else:
        modulated_probs = base_probs

    # Plotting
    fig, ax = plt.subplots()
    ax.plot(values, base_probs, "--", color="gray", label="FloodNet")
    if use_trustnet:
        ax.plot(values, modulated_probs, color="blue", label="With FloodTrustNet")
    ax.set_xlabel(axis.replace("_", " ").title())
    ax.set_ylabel("Flood Probability")
    ax.set_title(f"Flood Probability vs. {axis.replace('_', ' ').title()}")
    ax.grid(True)
    ax.legend()
    return fig


# Launch the app
with gr.Blocks(theme=gr.themes.Default(), css=".tab-nav-button { font-size: 1.1rem !important; padding: 0.8em; } ") as demo:
    gr.Markdown("# ClimateNet - A family of tabular classification models to predict natural disasters")

    with gr.Tab("🔥 FireNet"):
        with gr.Row():
            with gr.Column():
                with gr.Row():
                    temp = gr.Slider(280, 330, value=300, label="Temperature (K)")
                    temp_unit = gr.Dropdown(["K", "°C", "°F"], value="K", label="", scale=0.2)

                temp_unit.change(fn=update_temp_slider, inputs=temp_unit, outputs=temp)

                with gr.Row():
                    humidity = gr.Slider(0, 100, value=30, label="Humidity (%)")
                    gr.Dropdown(["%"], value="%", label="", scale=0.1)

                with gr.Row():
                    wind_speed = gr.Slider(0, 50, value=10, label="Wind Speed (m/s)")
                    wind_unit = gr.Dropdown(["m/s", "km/h", "mp/h"], value="m/s", label="", scale=0.2)

                wind_unit.change(update_wind_slider, inputs=wind_unit, outputs=wind_speed)

                with gr.Row():
                    elevation = gr.Slider(0, 3000, value=500, label="Elevation (m)")
                    elev_unit = gr.Dropdown(["m", "ft"], value="m", label="", scale=0.2)

                elev_unit.change(update_elevation_slider, inputs=elev_unit, outputs=elevation)

                with gr.Row():
                    vegetation_index = gr.Slider(0.0, 2.0, value=1.0, label="Vegetation Index (NDVI)")
                    gr.Dropdown(["NDVI"], value="NDVI", label="", scale=0.2)
                use_trust = gr.Checkbox(label="Use FireTrustNet", value=True)
                sweep_axis = gr.Radio(["temperature", "humidity", "wind_speed", "vegetation_index", "elevation"], 
                                      label="Sweep Axis", value="temperature")
                predict_btn = gr.Button("Predict")
            with gr.Column():
                with gr.Accordion("ℹ️ Feature Definitions", open=False):
                    gr.Markdown("""
                **Temperaure:** Current Temperature

                **Humidity:** Current Humidity

                **Wind Speed:** Current Wind Speed

                **Elevation:** Current Elevation Relative to Sea Level

                **Vegitation Index:** Your area's NDVI score.
                    """)
                output = gr.Textbox(label="Result")
                plot_output = gr.Plot(label="Trust Modulation Plot")

    predict_btn.click(
        fn=lambda t, tu, h, w, wu, v, e, eu, trust, axis: (
            predict_fire(t, tu, h, w, wu, v, e, eu, trust),
            generate_plot(axis, trust)
        ),
        inputs=[
            temp, temp_unit,
            humidity,
            wind_speed, wind_unit,
            vegetation_index,
            elevation, elev_unit,
            use_trust,
            sweep_axis
        ],
        outputs=[output, plot_output]
    )

    with gr.Tab("🏞️ FV-FloodNet"):
        with gr.Row():
            with gr.Column():
                with gr.Row():
                    rainfall = gr.Slider(0, 200, value=50, label="Rainfall (mm)")
                    rainfall_unit = gr.Dropdown(["mm", "in"], value="mm", label="", scale=0.2)

                with gr.Row():
                    water_level = gr.Slider(0, 8000, value=3000, label="Relative Water Level (mm)")
                    gr.Dropdown(["mm"], value="mm", label="", scale=0.2)

                with gr.Row():
                    elevation_flood = gr.Slider(0, 20, value=5, label="Relative Elevation (m)")
                    elev_flood_unit = gr.Dropdown(["m", "ft"], value="m", label="", scale=0.2)

                with gr.Row():
                    slope = gr.Slider(0.0, 20.0, value=2.0, label="Slope (°)")
                    gr.Dropdown(["°"], label="",scale=0.2)
                with gr.Row():
                    distance = gr.Slider(0, 2000, value=100, label="Distance from River (m)")
                    distance_unit = gr.Dropdown(["m", "ft"], value="m", label="", scale=0.2)

                elev_flood_unit.change(update_flood_elevation_slider, inputs=elev_flood_unit, outputs=elevation_flood)
                distance_unit.change(update_flood_distance_slider, inputs=distance_unit, outputs=distance)
                rainfall_unit.change(update_flood_rainfall_slider, inputs=rainfall_unit, outputs=rainfall)
                use_trust_flood = gr.Checkbox(label="Use FV-FloodTrustNet", value=True)

                flood_sweep_axis = gr.Radio(
                    ["rainfall", "water_level", "elevation", "slope", "distance_from_river"],
                    label="Sweep Axis", value="rainfall"
                )

                predict_btn_flood = gr.Button("Predict")
            
            with gr.Column():
                with gr.Accordion("ℹ️ Feature Definitions", open=False):
                    gr.Markdown("""
                **Rainfall:** Total recent precipitation - Last 24 hours.

                **Relative Water Level:** Height of river assuming river is 2.5m (8.202 ft) deep. Adjust accordingly.

                **Relative Elevation:** Ground height relative to nearest body of water (river).

                **Slope:** Terrain gradient measured in degrees.

                **Distance from River:** Horizontal distance from riverbed in meters. This does not account for levees or terrain barriers.
                    """)

                flood_output = gr.Textbox(label="Flood Risk")
                flood_plot = gr.Plot(label="Trust Modulation Plot")

    predict_btn_flood.click(
    fn=lambda r, ru, wl, e, eu, s, d, du, trust, axis: (
        predict_flood(r, ru, wl, e, eu, s, d, du, trust),
        generate_flood_plot(axis, trust)
    ),
    inputs=[
        rainfall, rainfall_unit,
        water_level,
        elevation_flood, elev_flood_unit,
        slope,
        distance, distance_unit,
        use_trust_flood,
        flood_sweep_axis
    ],
    outputs=[flood_output, flood_plot]
    )

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

app = gr.mount_gradio_app(app, demo, path="")

@app.get("/api/status")
def hello():
    return JSONResponse({"status": "ok"})