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 import traceback # --- 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}") traceback.print_exc() return None, None, None, None, None # --- RULES & TEMPLATES (no ellipses) --- 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': { 'immediate': "Deploy spot crews...", 'evacuation': "No mass evacuation...", 'containment': "Establish initial fire lines...", 'prevention': "Implement controlled underburning...", 'education': "Inform public on fire watch..." }, 'moderate': { 'immediate': "Dispatch multiple engines...", 'evacuation': "Prepare evacuation zones...", 'containment': "Build substantial fire breaks...", 'prevention': "Initiate fuel reduction...", 'education': "Conduct community emergency drills..." }, 'severe': { 'immediate': "Implement full suppression...", 'evacuation': "Issue mandatory evacuation orders...", 'containment': "Deploy fire retardant lines...", 'prevention': "Plan for reforestation...", 'education': "Conduct comprehensive training..." } } # --- PIPELINE & HELPERS --- 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 Exception: traceback.print_exc() 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 Exception: traceback.print_exc() 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) resp.raise_for_status() df = pd.DataFrame(resp.json().get('daily', {})) except Exception: traceback.print_exc() 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) try: cl = lr_model.predict(feat)[0]; return trend_map.get(cl,'same') except Exception: traceback.print_exc(); return 'same' def generate_recommendations(orig, trend): try: 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']}" except Exception: traceback.print_exc(); return "**Error generating recommendations**" # --- WRAPPER FOR GRADIO --- def safe_pipeline(image): try: return pipeline(image) except Exception as e: tb = traceback.format_exc() return f"Error: {e}\n{tb}", "", "", "" # --- LOAD MODELS GLOBALLY --- vgg_model, xception_model, rf_model, xgb_model, lr_model = load_models() # --- UI LAYOUT & STYLING --- custom_css = "..." # (same as before) with gr.Blocks(css=custom_css, theme=gr.themes.Soft()) as demo: # (UI definition same as before) run_btn.click(fn=safe_pipeline, inputs=image_input, outputs=[last_status, last_severity, last_trend, last_recs]) if __name__ == '__main__': demo.queue(api_open=True).launch()