import os import xgboost as xgb import pandas as pd import numpy as np import gradio as gr # ----- config ----- FEATURES = [ "aspect","clay", "contextual_slope_1000m","contextual_slope_300m","contextual_slope_5000m", "curvature","elevation","hand", "ndvi_change","ndvi_post","ndvi_pre", "organicC","rainfall","relief","ruggedness", "sand","silt","twi" ] # map predicted class -> human label (edit to match your training) LABEL_MAP = {0: "Low", 1: "Medium", 2: "High", 3: "Extreme"} # ----- load model safely (json) ----- MODEL_PATH = os.path.join(os.path.dirname(__file__), "rockfall_xgb.json") model = xgb.XGBClassifier() model.load_model(MODEL_PATH) # ----- helpers ----- def _predict_df(df: pd.DataFrame) -> pd.DataFrame: # ensure feature order and types missing = [c for c in FEATURES if c not in df.columns] if missing: raise ValueError(f"Missing required columns: {missing}") X = df[FEATURES].astype(float) preds = model.predict(X) probs = model.predict_proba(X) out = df.copy() out["predicted_class"] = preds.astype(int) out["predicted_label"] = [LABEL_MAP.get(int(c), str(int(c))) for c in preds] # add per-class probabilities for i in range(probs.shape[1]): label = LABEL_MAP.get(i, str(i)) out[f"prob_{label}"] = probs[:, i] # if ground truth exists, mark correctness if "target" in out.columns: out["correct"] = (out["target"].astype(int) == out["predicted_class"].astype(int)) return out # ----- Gradio functions ----- def predict_single(*vals): row = pd.DataFrame([vals], columns=FEATURES) scored = _predict_df(row) label = scored.loc[0, "predicted_label"] # build a tidy prob table for display prob_cols = [c for c in scored.columns if c.startswith("prob_")] probs = scored.loc[0, prob_cols].rename_axis("class").reset_index(name="probability") probs["class"] = probs["class"].str.replace("prob_", "", regex=False) probs = probs.sort_values("probability", ascending=False) return label, probs def predict_batch(file): # accept CSV or Excel name = os.path.basename(file.name) if name.lower().endswith((".xls", ".xlsx")): df = pd.read_excel(file.name) else: df = pd.read_csv(file.name) scored = _predict_df(df) # write downloadable CSV out_path = "/tmp/rockfall_predictions.csv" scored.to_csv(out_path, index=False) return scored, out_path # ----- UI ----- with gr.Blocks(title="⛏️ Rockfall Risk Prediction") as demo: gr.Markdown("## ⛏️ Rockfall Risk Prediction\nEnter features or upload a file to get class and probabilities.") with gr.Tab("Single prediction"): inputs = [gr.Number(label=col) for col in FEATURES] pred_label = gr.Label(label="Predicted Risk") prob_table = gr.Dataframe(headers=["class","probability"], label="Probabilities (descending)") gr.Button("Predict").click(predict_single, inputs=inputs, outputs=[pred_label, prob_table]) with gr.Tab("Batch (CSV/Excel)"): up = gr.File(label="Upload CSV/Excel with required columns", file_count="single", type="filepath") out_df = gr.Dataframe(label="Scored Data") out_file = gr.File(label="Download predictions CSV") gr.Button("Run batch").click(predict_batch, inputs=[up], outputs=[out_df, out_file]) # Spaces will call demo.launch() automatically if __name__ == "__main__": demo.launch()