Create app.py

app.py

@@ -0,0 +1,150 @@
+
+
+import os
+import joblib
+import gradio as gr
+import matplotlib.pyplot as plt
+import numpy as np
+import pandas as pd
+import shap
+from sklearn.compose import ColumnTransformer
+from sklearn.ensemble import RandomForestClassifier
+from sklearn.pipeline import Pipeline
+from sklearn.preprocessing import OneHotEncoder, StandardScaler
+
+# =====================================================================================
+# PART 1: MODEL CREATION AND LOADING (Self-contained for Hugging Face)
+# This part creates, trains, and saves a mock model if one doesn't exist.
+# This ensures the app is fully reproducible in any environment.
+# =====================================================================================
+
+MODEL_FILE = "machine_failure_model.joblib"
+
+def create_and_train_model():
+    """Creates, trains, and saves a mock model pipeline."""
+    # Mock data that resembles the predictive maintenance dataset
+    mock_features = pd.DataFrame({
+        'Type': ['L', 'M', 'H', 'L', 'M', 'H', 'L', 'M', 'H', 'L'],
+        'Air temperature [K]': [298.1, 298.2, 298.3, 298.4, 299.0, 299.5, 300.1, 301.0, 302.5, 303.0],
+        'Process temperature [K]': [308.6, 308.7, 308.8, 308.9, 309.1, 309.8, 310.5, 311.0, 312.0, 313.5],
+        'Rotational speed [rpm]': [1551, 1428, 1455, 1600, 1750, 2000, 2200, 2500, 2850, 1300],
+        'Torque [Nm]': [42.8, 46.3, 40.0, 50.1, 55.2, 60.0, 65.5, 70.0, 75.0, 35.0],
+        'Tool wear [min]': [0, 5, 10, 15, 25, 50, 80, 120, 180, 210]
+    })
+    # Mock target: 0 = No Failure, 1 = Failure
+    mock_target = np.array([0, 0, 0, 0, 0, 1, 1, 1, 1, 0])
+
+    # Define preprocessing steps for different column types
+    numeric_features = ['Air temperature [K]', 'Process temperature [K]', 'Rotational speed [rpm]', 'Torque [Nm]', 'Tool wear [min]']
+    categorical_features = ['Type']
+
+    preprocessor = ColumnTransformer(
+        transformers=[
+            ('num', StandardScaler(), numeric_features),
+            ('cat', OneHotEncoder(handle_unknown='ignore'), categorical_features)
+        ])
+
+    # Create the full pipeline with preprocessing and a classifier
+    model_pipeline = Pipeline(steps=[
+        ('preprocessor', preprocessor),
+        ('classifier', RandomForestClassifier(n_estimators=50, random_state=42))
+    ])
+
+    # Train the model
+    model_pipeline.fit(mock_features, mock_target)
+
+    # Save the trained model to a file
+    joblib.dump(model_pipeline, MODEL_FILE)
+    print(f"Model trained and saved to {MODEL_FILE}")
+    return model_pipeline
+
+# Check if the model file exists; if not, create it.
+if not os.path.exists(MODEL_FILE):
+    loaded_model = create_and_train_model()
+else:
+    loaded_model = joblib.load(MODEL_FILE)
+    print(f"Model loaded from {MODEL_FILE}")
+
+
+# =====================================================================================
+# PART 2: BACKEND LOGIC (Prediction and SHAP Calculation)
+# =====================================================================================
+
+def predict_failure(Type, air_temperature, process_temperature, rotational_speed, torque, tool_wear):
+    """Predicts machine failure and calculates SHAP values using the loaded model."""
+    input_data = pd.DataFrame({
+        'Type': [Type], 'Air temperature [K]': [air_temperature],
+        'Process temperature [K]': [process_temperature], 'Rotational speed [rpm]': [rotational_speed],
+        'Torque [Nm]': [torque], 'Tool wear [min]': [tool_wear]
+    })
+
+    preprocessor = loaded_model.named_steps['preprocessor']
+    classifier = loaded_model.named_steps['classifier']
+    input_processed = preprocessor.transform(input_data)
+    probability = classifier.predict_proba(input_processed)[:, 1]
+
+    explainer = shap.TreeExplainer(classifier)
+    shap_values = explainer.shap_values(input_processed)
+    feature_names = preprocessor.get_feature_names_out()
+
+    # SHAP values for the "Failure" class (index 1)
+    shap_val_failure = shap_values[1][0]
+    base_val_failure = explainer.expected_value[1]
+
+    return probability[0], shap_val_failure, feature_names, base_val_failure
+
+
+# =====================================================================================
+# PART 3: FRONTEND LOGIC (Plotting and Gradio Interface)
+# =====================================================================================
+
+def generate_shap_plot(shap_values, feature_names, base_value):
+    """Generates a SHAP waterfall plot for the Gradio interface."""
+    plt.close('all')  # Ensure plots don't stack in memory
+    explanation = shap.Explanation(
+        values=shap_values, base_values=base_value, feature_names=feature_names
+    )
+    fig, _ = plt.subplots()
+    shap.waterfall_plot(explanation, max_display=10, show=False)
+    plt.tight_layout()
+    return fig
+
+def predict_and_generate_plot(Type, air_temperature, process_temperature, rotational_speed, torque, tool_wear):
+    """Wrapper function that connects the backend prediction to the frontend plot."""
+    probability, shap_values, feature_names, base_value = predict_failure(
+        Type, air_temperature, process_temperature, rotational_speed, torque, tool_wear
+    )
+    shap_plot = generate_shap_plot(shap_values, feature_names, base_value)
+    return f"{probability:.2%}", shap_plot # Format probability as percentage
+
+# Define the Gradio interface layout and components
+with gr.Blocks(theme=gr.themes.Soft()) as iface_with_shap:
+    gr.Markdown("# Machine Failure Prediction with Live SHAP Analysis")
+    gr.Markdown("Adjust the sliders to see the real-time probability of machine failure and how each feature's value contributes to the prediction.")
+    with gr.Row():
+        with gr.Column(scale=1):
+            gr.Markdown("### Input Features")
+            type_input = gr.Dropdown(label="Type", choices=['L', 'M', 'H'], value='L')
+            air_temp_input = gr.Slider(minimum=295, maximum=305, value=300, label="Air temperature [K]")
+            proc_temp_input = gr.Slider(minimum=305, maximum=315, value=310, label="Process temperature [K]")
+            rpm_input = gr.Slider(minimum=1000, maximum=3000, value=1500, label="Rotational speed [rpm]")
+            torque_input = gr.Slider(minimum=5, maximum=80, value=40, label="Torque [Nm]")
+            wear_input = gr.Slider(minimum=0, maximum=250, value=100, label="Tool wear [min]")
+            inputs = [type_input, air_temp_input, proc_temp_input, rpm_input, torque_input, wear_input]
+
+        with gr.Column(scale=2):
+            gr.Markdown("### Prediction Outputs")
+            probability_output = gr.Textbox(label="Probability of Machine Failure")
+            plot_output = gr.Plot(label="Feature Contribution to Failure (SHAP Waterfall Plot)")
+
+    # Connect the inputs to the function and outputs
+    for input_comp in inputs:
+        input_comp.change(
+            fn=predict_and_generate_plot,
+            inputs=inputs,
+            outputs=[probability_output, plot_output]
+        )
+
+# Launch the application
+if __name__ == "__main__":
+    iface_with_shap.launch(debug=True)