Spaces:
Sleeping
Sleeping
| 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 fire detection model | |
| vgg_model = load_model( | |
| 'vgg16_focal_unfreeze_more.keras', | |
| custom_objects={'BinaryFocalCrossentropy': BinaryFocalCrossentropy} | |
| ) | |
| # Xception severity model | |
| 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()} | |
| ) | |
| # Reload ensemble models from .pkl | |
| 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 | |
| # Load models once | |
| vgg_model, xce_model, rf_model, xgb_model, lr_model = load_models() | |
| 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 = { ... } # (your existing recommendations dict) | |
| # --- 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 Exception as e: | |
| print(f"Error in fire detection: {e}") | |
| return False, 0.0 | |
| def classify_severity(img): | |
| try: | |
| if xce_model is None or rf_model is None or xgb_model is None: | |
| return 'moderate' | |
| x = keras_image.img_to_array(img.resize((224,224)))[None] | |
| x = xce_preprocess(x) | |
| preds = xce_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 as e: | |
| print(f"Error in severity classification: {e}") | |
| 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')) | |
| response = requests.get(url, timeout=5) | |
| response.raise_for_status() | |
| df = pd.DataFrame(response.json().get('daily', {})) | |
| except Exception: | |
| # fallback dummy data | |
| df = pd.DataFrame({ | |
| 'date': [(datetime.utcnow() - timedelta(days=i)).strftime('%Y-%m-%d') for i in range(1,-1,-1)], | |
| '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] | |
| }) | |
| # compute features | |
| 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: | |
| trend_cl = lr_model.predict(feat)[0] | |
| return trend_map.get(trend_cl, 'same') | |
| return 'same' | |
| def generate_recommendations(original_severity, weather_trend): | |
| projected = task_rules[original_severity][weather_trend] | |
| rec = recommendations[projected] | |
| return ( | |
| f"**Original Severity:** {original_severity.title()}\n" | |
| f"**Weather Trend:** {weather_trend.title()}\n" | |
| f"**Projected Severity:** {projected.title()}\n\n" | |
| "### Management Recommendations:\n" | |
| f"**Immediate:** {rec['immediate']}\n\n" | |
| f"**Evacuation:** {rec['evacuation']}\n\n" | |
| f"**Containment:** {rec['containment']}\n\n" | |
| f"**Prevention:** {rec['prevention']}\n\n" | |
| f"**Education:** {rec['education']}" | |
| ) | |
| def pipeline(image): | |
| if image is None: | |
| return "No image provided", "N/A", "N/A", "**Please upload an image to analyze**" | |
| img = Image.fromarray(image).convert('RGB') | |
| fire, prob = detect_fire(img) | |
| if not fire: | |
| return ( | |
| f"No wildfire detected (confidence: {(1-prob)*100:.1f}%)", | |
| "N/A", "N/A", | |
| "**No wildfire detected. Stay alert.**" | |
| ) | |
| sev = classify_severity(img) | |
| trend = fetch_weather_trend(*FOREST_COORDS['Pakistan Forest']) | |
| recs = generate_recommendations(sev, trend) | |
| return ( | |
| f"**Wildfire detected** (confidence: {prob*100:.1f}%)", | |
| f"**{sev.title()}**", | |
| f"**{trend.title()}**", | |
| recs | |
| ) | |
| def safe_pipeline(image): | |
| try: | |
| return pipeline(image) | |
| except Exception as e: | |
| print(f"Error in pipeline: {e}") | |
| return "Error during analysis", "N/A", "N/A", f"**Error: {e}**" | |
| # --- GRADIO UI --- | |
| custom_css = ''' | |
| #header { text-align: center; margin-bottom: 1rem; } | |
| ''' | |
| with gr.Blocks(css=custom_css, theme=gr.themes.Soft()) as demo: | |
| with gr.Row(elem_id="header"): | |
| try: | |
| gr.Image(value="logo.png", show_label=False) | |
| except: | |
| pass | |
| with gr.Column(): | |
| gr.Markdown("# 🔥 Wildfire Command Center") | |
| gr.Markdown("Upload a forest image to detect wildfire, classify severity, and get actionable recommendations.") | |
| with gr.Tabs(): | |
| with gr.TabItem("Analyze 🔍"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| # use ImageEditor if in-browser annotation is needed, otherwise simple Image | |
| image_input = gr.Image(type="numpy", label="Forest Image") | |
| run_btn = gr.Button("Analyze Now", variant="primary") | |
| with gr.Column(scale=1): | |
| status_out = gr.Markdown("*Status will appear here*", label="Status") | |
| severity_out = gr.Markdown("---", label="Severity") | |
| trend_out = gr.Markdown("---", label="Weather Trend") | |
| recs_out = gr.Markdown("---", label="Recommendations") | |
| with gr.TabItem("Last Analysis 📊"): | |
| last_status = gr.Markdown("*No analysis yet*", elem_classes="output-card") | |
| last_severity = gr.Markdown("---", elem_classes="output-card") | |
| last_trend = gr.Markdown("---", elem_classes="output-card") | |
| last_recs = gr.Markdown("---", elem_classes="output-card") | |
| run_btn.click( | |
| fn=safe_pipeline, | |
| inputs=image_input, | |
| outputs=[status_out, severity_out, trend_out, recs_out] | |
| ).then( | |
| fn=lambda s,sv,tr,rc: (s,sv,tr,rc), | |
| inputs=[status_out, severity_out, trend_out, recs_out], | |
| outputs=[last_status, last_severity, last_trend, last_recs] | |
| ) | |
| if __name__ == '__main__': | |
| demo.queue(api_open=True).launch() |