EmberDeepAI / app.py
AbdullahImran's picture
Update app.py
2530e3b verified
raw
history blame
7.56 kB
import os
import requests
import pandas as pd
import numpy as np
import joblib
import gradio as gr
from datetime import datetime, timedelta
from tensorflow.keras.models import load_model
from tensorflow.keras.preprocessing import image as keras_image
from tensorflow.keras.applications.vgg16 import preprocess_input as vgg_preprocess
from tensorflow.keras.applications.xception import preprocess_input as xce_preprocess
from tensorflow.keras.losses import BinaryFocalCrossentropy
from PIL import Image
# --- CONFIGURATION ---
FOREST_COORDS = {'Pakistan Forest': (34.0, 73.0)}
API_URL = (
"https://archive-api.open-meteo.com/v1/archive"
"?latitude={lat}&longitude={lon}"
"&start_date={start}&end_date={end}"
"&daily=temperature_2m_max,temperature_2m_min,"
"precipitation_sum,windspeed_10m_max,"
"relative_humidity_2m_max,relative_humidity_2m_min"
"&timezone=UTC"
)
# --- LOAD MODELS ---
def load_models():
try:
vgg_model = load_model(
'vgg16_focal_unfreeze_more.keras',
custom_objects={'BinaryFocalCrossentropy': BinaryFocalCrossentropy}
)
def focal_loss_fixed(gamma=2., alpha=.25):
import tensorflow.keras.backend as K
def loss_fn(y_true, y_pred):
eps = K.epsilon(); y_pred = K.clip(y_pred, eps, 1.-eps)
ce = -y_true * K.log(y_pred)
w = alpha * K.pow(1-y_pred, gamma)
return K.mean(w * ce, axis=-1)
return loss_fn
xce_model = load_model(
'severity_post_tta.keras',
custom_objects={'focal_loss_fixed': focal_loss_fixed()}
)
rf_model = joblib.load('ensemble_rf_model.pkl')
xgb_model = joblib.load('ensemble_xgb_model.pkl')
lr_model = joblib.load('wildfire_logistic_model_synthetic.joblib')
return vgg_model, xce_model, rf_model, xgb_model, lr_model
except Exception as e:
print(f"Error loading models: {e}")
return None, None, None, None, None
# --- RULES & TEMPLATES ---
target_map = {0: 'mild', 1: 'moderate', 2: 'severe'}
trend_map = {1: 'increase', 0: 'same', -1: 'decrease'}
task_rules = {
'mild': {'decrease':'mild','same':'mild','increase':'moderate'},
'moderate':{'decrease':'mild','same':'moderate','increase':'severe'},
'severe': {'decrease':'moderate','same':'severe','increase':'severe'}
}
recommendations = {
'mild': {...},
'moderate': {...},
'severe': {...}
}
# --- PIPELINE FUNCTIONS ---
def detect_fire(img):
try:
if vgg_model is None:
return True, 0.85
x = keras_image.img_to_array(img.resize((128,128)))[None]
x = vgg_preprocess(x)
prob = float(vgg_model.predict(x)[0][0])
return prob >= 0.5, prob
except:
return False, 0.0
def classify_severity(img):
try:
if xception_model is None:
return 'moderate'
x = keras_image.img_to_array(img.resize((224,224)))[None]
x = xce_preprocess(x)
preds = xception_model.predict(x)
rf_p = rf_model.predict(preds)[0]
xgb_p = xgb_model.predict(preds)[0]
ensemble = int(round((rf_p + xgb_p)/2))
return target_map.get(ensemble,'moderate')
except:
return 'moderate'
def fetch_weather_trend(lat, lon):
try:
end = datetime.utcnow()
start = end - timedelta(days=1)
url = API_URL.format(lat=lat, lon=lon,
start=start.strftime('%Y-%m-%d'),
end=end.strftime('%Y-%m-%d'))
resp = requests.get(url, timeout=5)
if resp.status_code==200:
df = pd.DataFrame(resp.json().get('daily', {}))
else:
raise Exception()
except:
df = pd.DataFrame({ 'date': ['2025-04-25','2025-04-26'], 'precipitation_sum':[5,2], 'temperature_2m_max':[28,30], 'temperature_2m_min':[18,20], 'relative_humidity_2m_max':[70,65], 'relative_humidity_2m_min':[40,35], 'windspeed_10m_max':[15,18] })
df['temperature'] = (df['temperature_2m_max'] + df['temperature_2m_min'])/2
df['humidity'] = (df['relative_humidity_2m_max'] + df['relative_humidity_2m_min'])/2
df['wind_speed'] = df['windspeed_10m_max']
df['precipitation'] = df['precipitation_sum']
df['fire_risk_score'] = (0.4*(df['temperature']/55) + 0.2*(1-df['humidity']/100) + 0.3*(df['wind_speed']/60) + 0.1*(1-df['precipitation']/50))
feat = df[['temperature','humidity','wind_speed','precipitation','fire_risk_score']].iloc[-1].values.reshape(1,-1)
if lr_model is not None:
cl = lr_model.predict(feat)[0]
return trend_map.get(cl,'same')
return 'same'
def generate_recommendations(orig, trend):
proj = task_rules[orig][trend]
rec = recommendations[proj]
return f"**Original Severity:** {orig.title()} \n**Weather Trend:** {trend.title()} \n**Projected Severity:** {proj.title()}\n\n### Management Recommendations:\n**Immediate:** {rec['immediate']}\n\n**Evacuation:** {rec['evacuation']}\n\n**Containment:** {rec['containment']}\n\n**Prevention:** {rec['prevention']}\n\n**Education:** {rec['education']}"
# --- MAIN PIPELINE ---
def pipeline(image):
if image is None:
return "No image provided","N/A","N/A","**Upload image**"
img = Image.fromarray(image).convert('RGB')
fire, prob = detect_fire(img)
if not fire:
return f"No wildfire detected ({(1-prob)*100:.1f}% sure)","N/A","N/A","**No wildfire.**"
sev = classify_severity(img)
trend = fetch_weather_trend(*FOREST_COORDS['Pakistan Forest'])
recs = generate_recommendations(sev, trend)
return f"Wildfire detected ({prob*100:.1f}%)", sev.title(), trend.title(), recs
# --- LOAD MODELS ---
vgg_model, xception_model, rf_model, xgb_model, lr_model = load_models()
# --- UI STYLING & LAYOUT ---
custom_css = """
.sidebar { background: #2e3440; color: #eceff4; padding: 1rem; border-radius: 1rem; }
#main-title { font-size: 2.5rem; color: #3b4252; }
#sub-title { font-size: 1.125rem; color: #4c566a; }
.card { background: #eceff4; color: #2e3440; border-radius: 0.75rem; padding: 1rem; margin-bottom: 1rem; box-shadow: 0 2px 8px rgba(0,0,0,0.1); }
.gr-button { background: #5e81ac !important; color: white !important; border-radius: 0.5rem; }
.status-badge { padding: 0.25em 0.75em; border-radius: 9999px; font-weight: 600; }
.status-fire { background: #bf616a; color: white; }
.status-no-fire { background: #a3be8c; color: white; }
.gr-markdown { color: #2e3440; }
"""
with gr.Blocks(css=custom_css, theme=gr.themes.Soft()) as demo:
with gr.Row():
with gr.Column(scale=2):
gr.Markdown("# 🔥 Wildfire Command Center", elem_id="main-title")
gr.Markdown(
"Upload a **forest image** to detect wildfire, classify severity, fetch weather trend, and get recommendations.",
elem_id="sub-title"
)
image_input = gr.Image(type="numpy", label="Upload Forest Image")
run_btn = gr.Button("Analyze Now", variant="primary")
with gr.Column(scale=1, elem_classes="sidebar"):
gr.Markdown("## 📊 Last Analysis")
last_status = gr.Markdown("*No analysis yet*")
last_severity = gr.Markdown("---")
last_trend = gr.Markdown("---")
last_recs = gr.Markdown("---")
run_btn.click(
fn=pipeline,
inputs=image_input,
outputs=[last_status, last_severity, last_trend, last_recs]
)
if __name__ == '__main__':
demo.queue(api_open=True).launch()