# ------------------------------------------------------------ # app.py — Gradio wrapped version of predict_explain.py # ------------------------------------------------------------ import os import numpy as np import joblib import pandas as pd import gradio as gr from catboost import CatBoostClassifier, Pool from pytorch_tabnet.tab_model import TabNetClassifier # ------------------------------------------------------------ # Keep the same OUTPUT_DIR as your original script, but resolve it # (so "../models" in your repo will be found when this file runs). # ------------------------------------------------------------ OUTPUT_DIR = os.path.join(os.path.dirname(__file__), "models") BASE_DIR = os.path.dirname(os.path.abspath(__file__)) OUTPUT_DIR = os.path.join(BASE_DIR, OUTPUT_DIR) # Ordered feature lists (kept exactly from your script) numeric_cols = [ 'period', 'duration', 'depth', 'radius', 'eqt', 'insol', 'st_teff', 'st_logg', 'st_rad', 'ra', 'dec' ] categorical_cols = ['source_id', 'mission_code'] # Feature descriptions (kept identical) feature_info = { 'period': ("Orbital period of the exoplanet (in days)", "Time for the planet to complete one orbit around its star."), 'duration': ("Transit duration (in hours)", "Time taken by the planet to pass in front of the star."), 'depth': ("Transit depth (ppm or relative flux)", "Brightness decrease during transit, linked to planet size."), 'radius': ("Planet radius (in Earth radii, R⊕)", "The physical size of the exoplanet."), 'eqt': ("Equilibrium temperature (Kelvin)", "Estimated temperature assuming blackbody radiation."), 'insol': ("Insolation flux (relative to Earth, S⊕)", "Stellar energy received compared to Earth."), 'st_teff': ("Stellar effective temperature (Kelvin)", "Surface temperature of the host star."), 'st_logg': ("Stellar surface gravity (log10 cm/s²)", "Describes the star’s gravity and structure."), 'st_rad': ("Stellar radius (in Solar radii, R☉)", "Size of the star compared to the Sun."), 'ra': ("Right Ascension (degrees)", "Celestial coordinate for the star’s sky position."), 'dec': ("Declination (degrees)", "Celestial coordinate for the star’s sky position."), 'source_id': ("Source ID (e.g., 'Kepler-22', 'TIC12345')", "Identifier of the object in its survey dataset."), 'mission_code': ("Mission code ('KEPLER', 'TESS', or 'K2')", "Survey mission that observed the target.") } # ------------------------------------------------------------ # Load models, encoders, and scaler (same file names as your script) # ------------------------------------------------------------ print("🚀 Loading trained models and encoders from:", OUTPUT_DIR) # Helper to load with clear error messages def _safe_load_catboost(path): model = CatBoostClassifier() model.load_model(path) return model def _path(p): return os.path.join(OUTPUT_DIR, p) # Attempt to load required files (will raise if missing — errors caught in predict) try: cat_model = _safe_load_catboost(_path("catboost_model.cbm")) print("Loaded CatBoost model.") except Exception as e: cat_model = None print("⚠️ Could not load CatBoost model:", e) try: tabnet_model = joblib.load(_path("tabnet_model.pkl")) print("Loaded TabNet model.") except Exception as e: tabnet_model = None print("⚠️ Could not load TabNet model:", e) try: scaler = joblib.load(_path("tabnet_scaler.pkl")) print("Loaded TabNet scaler.") except Exception as e: scaler = None print("⚠️ Could not load scaler:", e) try: label_encoder = joblib.load(_path("label_encoder.pkl")) print("Loaded label encoder.") except Exception as e: label_encoder = None print("⚠️ Could not load label encoder:", e) try: source_encoder = joblib.load(_path("source_id_encoder.pkl")) print("Loaded source_id encoder.") except Exception as e: source_encoder = None print("⚠️ Could not load source_id encoder:", e) try: mission_encoder = joblib.load(_path("mission_code_encoder.pkl")) print("Loaded mission_code encoder.") except Exception as e: mission_encoder = None print("⚠️ Could not load mission_code encoder:", e) # ------------------------------------------------------------ # Utility to safely align arrays (pad with zeros or trim) # ------------------------------------------------------------ def _align_array(arr, target_len): arr = np.asarray(arr, dtype=float) if arr.ndim == 0: arr = np.array([arr]) if len(arr) < target_len: return np.concatenate([arr, np.zeros(target_len - len(arr))]) else: return arr[:target_len] # ------------------------------------------------------------ # Prediction function (follows your script's logic closely) # ------------------------------------------------------------ def predict_exoplanet(*inputs): """ inputs order: numeric_cols + categorical_cols Returns: (markdown_text, shap_dataframe) """ try: # Map inputs to names names = numeric_cols + categorical_cols if len(inputs) != len(names): return f"⚠️ Error: Expected {len(names)} inputs but got {len(inputs)}.", pd.DataFrame() user_data = {} # Convert numeric inputs to float (maintain your user prompt behaviour) for i, col in enumerate(numeric_cols): val = inputs[i] if val is None or str(val).strip() == "": return f"⚠️ Error: Missing numeric input for '{col}'.", pd.DataFrame() try: user_data[col] = float(val) except Exception as e: return f"⚠️ Error: Invalid numeric value for '{col}': {val}", pd.DataFrame() # Categorical inputs for j, col in enumerate(categorical_cols, start=len(numeric_cols)): val = inputs[j] if val is None: val = "" user_data[col] = str(val).strip() # Create DataFrame (same as original) df = pd.DataFrame([user_data]) # Encode categorical features using provided encoders (match your script) if source_encoder is None or mission_encoder is None: # If encoders not available, try using simple factorize fallback (keeps logic robust) df['source_id'] = pd.factorize(df['source_id'].astype(str))[0] df['mission_code'] = pd.factorize(df['mission_code'].astype(str))[0] else: df['source_id'] = source_encoder.transform(df['source_id'].astype(str)) df['mission_code'] = mission_encoder.transform(df['mission_code'].astype(str)) # Prepare feature matrices X_input = df[numeric_cols + categorical_cols] if scaler is not None: X_input_scaled = X_input.copy() X_input_scaled[numeric_cols] = scaler.transform(X_input[numeric_cols]) else: X_input_scaled = X_input.copy() # Predictions (CatBoost & TabNet) if cat_model is None: # fallback zero-proba shape (3 classes) cat_pred_proba = np.zeros((1, 3)) else: cat_pred_proba = cat_model.predict_proba( Pool(X_input, cat_features=[X_input.columns.get_loc(c) for c in categorical_cols]) ) if tabnet_model is None: tabnet_pred_proba = np.zeros((1, 3)) else: tabnet_pred_proba = tabnet_model.predict_proba(X_input_scaled.values) # Ensemble (average) ensemble_proba = (cat_pred_proba + tabnet_pred_proba) / 2.0 ensemble_pred_idx = np.argmax(ensemble_proba, axis=1) ensemble_conf = np.max(ensemble_proba, axis=1) # Inverse label mapping (keep same call as your script) if label_encoder is not None: ensemble_pred_labels = label_encoder.inverse_transform(ensemble_pred_idx) else: # fallback mapping if label encoder missing fallback_map = {0: "candidate", 1: "false_positive", 2: "planet"} ensemble_pred_labels = [fallback_map.get(int(i), str(int(i))) for i in ensemble_pred_idx] # Explainability (CatBoost SHAP + TabNet global importances) # CatBoost SHAP values: same call as your script, but with safe fallback n_features = X_input.shape[1] try: if cat_model is not None: cat_shap_values = cat_model.get_feature_importance( Pool(X_input, cat_features=[X_input.columns.get_loc(c) for c in categorical_cols]), type='ShapValues' ) # original: shap_row = cat_shap_values[0, :-1, pred_class_idx] pred_class_idx = int(ensemble_pred_idx[0]) raw_shap = cat_shap_values[0, :-1, pred_class_idx] shap_row = _align_array(raw_shap, n_features) else: shap_row = np.zeros(n_features) except Exception as e: # If CatBoost SHAP retrieval fails, provide zeros shap_row = np.zeros(n_features) # TabNet global feature importances (safe align) try: if tabnet_model is not None: tabnet_feat_importances = getattr(tabnet_model, "feature_importances_", None) if tabnet_feat_importances is None: tabnet_feat_importances = np.zeros(n_features) tabnet_feat_importances = _align_array(tabnet_feat_importances, n_features) else: tabnet_feat_importances = np.zeros(n_features) except Exception: tabnet_feat_importances = np.zeros(n_features) # Construct results DataFrame similar to original printed outputs feat_names = list(X_input.columns) values = X_input.iloc[0].values shap_df = pd.DataFrame({ "Feature": feat_names, "Value": values, "CatBoost_SHAP": shap_row, "TabNet_Importance": tabnet_feat_importances }) # Sort for display by absolute SHAP importance (descending) shap_df["abs_shap"] = np.abs(shap_df["CatBoost_SHAP"]) shap_df = shap_df.sort_values(by="abs_shap", ascending=False).drop(columns=["abs_shap"]).reset_index(drop=True) # Prepare markdown result following your original display md = ( "============================== 🌌 Prediction Result ==============================\n\n" f"**Predicted Classification** : `{ensemble_pred_labels[0]}` \n" f"**Confidence Score** : `{ensemble_conf[0]:.4f}`\n\n" "==============================================================================\n\n" "**📗 Label meanings:** \n" " • `planet` → Confirmed exoplanet \n" " • `candidate` → Possible exoplanet candidate \n" " • `false_positive` → Not a real planet (noise/artifact)\n\n" "**🔬 Feature Contributions (CatBoost SHAP):**\n" ) # append per-feature lines (keeps same info as your printed loop) for f, shap_val, actual in zip(feat_names, shap_row, values): title = feature_info.get(f, (f, ""))[0] md += f"- {title:45s} → Value={float(actual):.4f}, SHAP={shap_val:+.4f}\n" md += "\n**📊 TabNet Global Feature Importances:**\n" for f, imp_val, actual in zip(feat_names, tabnet_feat_importances, values): title = feature_info.get(f, (f, ""))[0] md += f"- {title:45s} → Value={float(actual):.4f}, Importance={float(imp_val):.4f}\n" md += "\n✅ Prediction complete." # Return the markdown string and the dataframe (for Gradio table) return md, shap_df except Exception as e: # Return error message and empty DF; this prevents app crash in Gradio return f"⚠️ Unexpected error: {e}", pd.DataFrame() # ------------------------------------------------------------ # Build Gradio interface: single-page inputs (no tabs) — keeps exact feature ordering # ------------------------------------------------------------ inputs = [] for col in numeric_cols: title, desc = feature_info[col] inputs.append(gr.Number(label=title, info=desc, precision=6)) for col in categorical_cols: title, desc = feature_info[col] inputs.append(gr.Textbox(label=title, placeholder=desc)) outputs = [ gr.Markdown(label="Prediction Result"), gr.Dataframe(label="Feature Contributions (CatBoost SHAP + TabNet)") ] app = gr.Interface( fn=predict_exoplanet, inputs=inputs, outputs=outputs, title="🔭 Exoplanet Classification Explorer", description="This app is a Gradio UI built directly from your original `predict_explain.py`. Enter numeric & categorical features and get the ensemble prediction plus SHAP & TabNet importances.", allow_flagging="never", theme="soft" ) if __name__ == "__main__": app.launch()