Spaces:

aun09
/

NEXA

Sleeping

NEXA / app.py

Update app.py

c468ecd verified about 1 year ago

13.2 kB

# ------------------------------------------------------------
# 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()

	# ------------------------------------------------------------
	# 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()