app.py

# -*- coding: utf-8 -*-
"""DashBoard.ipynb

Automatically generated by Colab.

Original file is located at
    https://colab.research.google.com/drive/101_T8xMCWHcohzgIs8NvdK3wYK7h7lwO
"""


import gradio as gr
import pandas as pd
import pickle
import numpy as np
from tensorflow.keras.models import load_model

# تحميل النماذج من المجلد المحلي
with open("Models/Feature_Scaler.pkl", "rb") as f:
    scaler = pickle.load(f)

with open("Models/target_scaler.pkl", "rb") as f:
    target_scaler = pickle.load(f)

with open("Models/LR_model.pkl", "rb") as f:
    linear_model = pickle.load(f)

with open("Models/DT_model.pkl", "rb") as f:
    dt_model = pickle.load(f)

with open("Models/RF_model.pkl", "rb") as f:
    rf_model = pickle.load(f)

lstm_model = load_model("Models/best_model.h5")

# Create sequences for LSTM
def create_sequences(data, window_size=11):
    sequences = []
    for i in range(len(data) - window_size + 1):
        seq = data[i:i+window_size]
        sequences.append(seq)
    return np.array(sequences).astype('float32')

# Data processing and alert function
def process_and_alert(file):
    try:
        df = pd.read_csv(file.name)
        df["timestamp"] = pd.to_datetime(df["timestamp"])
        df = df.sort_values("timestamp").reset_index(drop=True)
        df["fault_flag"] = df["status"].apply(lambda x: 1 if x == "fault" else 0)
        fault_indices = df[df["fault_flag"] == 1].index.tolist()
        time_to_failure = []
        for i in range(len(df)):
            next_faults = [j for j in fault_indices if j >= i]
            if next_faults:
                seconds = (df.loc[next_faults[0], "timestamp"] - df.loc[i, "timestamp"]).total_seconds()
            else:
                seconds = None
            time_to_failure.append(seconds)
        df["time_to_failure"] = time_to_failure
        df.dropna(inplace=True)
        X = df.drop(columns=['time_to_failure', 'fault_flag', 'status', 'timestamp'])
        X_scaled = scaler.transform(X)

        # LSTM processing
        window_size = 11
        if len(X_scaled) < window_size:
            raise ValueError(f"Model requires at least {window_size} samples!")
        X_seq = create_sequences(X_scaled, window_size)

        # Predictions
        pred_linear = linear_model.predict(X_scaled)
        pred_dt = dt_model.predict(X_scaled)
        pred_rf = rf_model.predict(X_scaled)
        pred_lstm = lstm_model.predict(X_seq)

        # Align lengths
        min_length = min(len(pred_linear), len(pred_dt), len(pred_rf), len(pred_lstm))
        pred_linear = pred_linear[:min_length]
        pred_dt = pred_dt[:min_length]
        pred_rf = pred_rf[:min_length]
        pred_lstm = pred_lstm[:min_length]

        # Inverse transform
        pred_lstm = target_scaler.inverse_transform(pred_lstm.reshape(-1, 1))
        pred_linear = target_scaler.inverse_transform(pred_linear.reshape(-1, 1))
        pred_dt = target_scaler.inverse_transform(pred_dt.reshape(-1, 1))
        pred_rf = target_scaler.inverse_transform(pred_rf.reshape(-1, 1))

        def format_value(val):
            return f'<span style="color:red;font-weight:bold;">{val:.2f} (Fault)</span>' if val < 0 else f'{val:.2f}'

        html_rows = ""
        for i in range(min_length):
            html_rows += "<tr>" + "".join([
                f"<td>{format_value(pred_linear[i][0])}-second</td>",
                f"<td>{format_value(pred_dt[i][0])}-second</td>",
                f"<td>{format_value(pred_rf[i][0])}-second</td>",
                f"<td>{format_value(pred_lstm[i][0])}-second</td>"
            ]) + "</tr>"

        html_table = f"""
        <table border="1" style="border-collapse:collapse; width:100%; text-align:center;">
            <thead>
                <tr style="background-color:#f0f0f0;">
                    <th>Linear Regression</th>
                    <th>Decision Tree</th>
                    <th>Random Forest</th>
                    <th>LSTM</th>
                </tr>
            </thead>
            <tbody>
                {html_rows}
            </tbody>
        </table>
        """

        # Alert System
        preds = {
            "Linear Regression": pred_linear,
            "Decision Tree": pred_dt,
            "Random Forest": pred_rf,
            "LSTM": pred_lstm
        }
        alerts = []
        for model_name, values in preds.items():
            positives = values[values > 0]
            if positives.size > 0:
                min_pos = np.min(positives)
                alerts.append((model_name, min_pos))
        if not alerts:
            alert_msg = "<h3 style='color:red;'>❌ No positive predictions found! Failure may have already occurred.</h3>"
        else:
            alerts.sort(key=lambda x: x[1])
            best_model, time_left = alerts[0]
            minutes = int(time_left // 60)
            seconds = int(time_left % 60)
            color = "red" if time_left < 60 else "orange" if time_left < 180 else "green"
            msg = f"{minutes} minute(s) and {seconds} second(s)" if minutes > 0 else f"{seconds} second(s)"
            alert_msg = f"""
            <div style="padding:20px; border:2px solid {color}; border-radius:10px;">
                <h3>🔔 Failure Alert</h3>
                <p><strong>Model:</strong> <span style="color:blue;">{best_model}</span></p>
                <p><strong>Estimated time to failure:</strong> <span style="color:{color}; font-weight:bold;">{msg}</span></p>
                {"<p style='color:red; font-weight:bold;'>⚠️ Imminent failure!</p>" if time_left < 60 else ""}
            </div>
            """
        return html_table, alert_msg
    except Exception as e:
        error_msg = f"<div style='color:red; padding:20px; border:2px solid red;'><h3>❌ Error:</h3><p>{str(e)}</p></div>"
        return error_msg, ""

# Load comparison tables
def load_metrics():
    return pd.read_csv("Data/Models_Metrices.csv")

def load_comparison():
    return pd.read_csv("Data/model_comparison_20250419_0955.csv")

# Gradio UI
with gr.Blocks(title="📊 Model Comparison Dashboard") as interface:
    with gr.Tab("📈 Model Comparison"):
        gr.Markdown("## 🔍 Actual vs Predicted")
        with gr.Row():
            gr.Image(value="Images/Liner_Regresion.png", label="Linear Regression")
            gr.Image(value="Images/DT.png", label="Decision Tree")
        with gr.Row():
            gr.Image(value="Images/Random_Forest.png", label="Random Forest")
            gr.Image(value="Images/LSTM.png", label="LSTM")
        gr.Markdown("### 🧮 Data Distribution")
        gr.Image(value="Images/Data_Dis.png", label="Data Distribution")
        gr.Markdown("### 📋 Model Metrics Table")
        gr.Dataframe(load_metrics, interactive=False)
        gr.Markdown("### 🆕 Best and Worst Models Table")
        gr.Dataframe(load_comparison, interactive=False)

    with gr.Tab("📁 Upload Data"):
        gr.Markdown("## 📥 Upload a new CSV file to analyze and detect failure")
        file_input = gr.File(label="Choose CSV File")
        output_html = gr.HTML(label="Prediction Results")
        alert_output = gr.HTML(label="🔔 Alert")
        file_input.change(fn=process_and_alert, inputs=file_input, outputs=[output_html, alert_output])

    with gr.Tab("🧮 Manual Input"):
        gr.Markdown("""
            <div style="display:flex; align-items:center; gap:10px;">
                <span style="font-size:30px;">🧾</span>
                <h3 style="margin:0;">Enter 3 features to get a prediction and failure alert</h3>
            </div>
        """)
        with gr.Row():
            f1 = gr.Number(label="vibration")
            f2 = gr.Number(label="temperature")
            f3 = gr.Number(label="pressure")
        result_output = gr.Textbox(label="🔍 Predicted Time (seconds)", interactive=False)
        alert_output_ready = gr.HTML(label="🚨 Alert")

        def predict_and_alert_ready(x1, x2, x3):
            try:
                X_input = np.array([[x1, x2, x3]])
                X_scaled = scaler.transform(X_input)
                pred = rf_model.predict(X_scaled).reshape(-1, 1)
                pred_original = target_scaler.inverse_transform(pred)[0][0]
                minutes = int(pred_original // 60)
                seconds = int(pred_original % 60)
                color = "red" if pred_original < 60 else "orange" if pred_original < 180 else "green"
                msg = f"{minutes} minute(s) and {seconds} second(s)" if minutes > 0 else f"{seconds} second(s)"
                alert_html = f"""
                <div style="padding:15px; border:2px solid {color}; border-radius:10px;">
                    <h3>📢 Advanced Alert</h3>
                    <p><strong>Estimated time to failure:</strong> <span style="color:{color}; font-weight:bold;">{msg}</span></p>
                    {"<p style='color:red; font-weight:bold;'>⚠️ Imminent failure!</p>" if pred_original < 60 else ""}
                </div>
                """
                return f"{pred_original:.2f} seconds", alert_html
            except Exception as e:
                return "Input Error", f"<p style='color:red;'>❌ {str(e)}</p>"

        btn = gr.Button("🔍 Predict Now")
        btn.click(predict_and_alert_ready, inputs=[f1, f2, f3], outputs=[result_output, alert_output_ready])

interface.launch(share=True)