Update app.py

app.py

@@ -1,186 +1,74 @@
-# import numpy as np
-# import gradio as gr
-# from tensorflow.keras.models import load_model, Model
-# from hpelm import ELM
-# import joblib
-
-# # 🔄 Load scaler and ELM model
-# scaler = joblib.load("elm_scaler1.pkl")
-# elm = ELM(128, 1, classification='c')
-# elm.load("elm_model1.txt")
-
-# # ⚙️ Load trained MobileNet1D and create feature extractor
-# mobilenet = load_model("physio_model2.h5")
-# feature_extractor = Model(inputs=mobilenet.input, outputs=mobilenet.get_layer("penultimate_dense").output)
-
-# # 🔍 Preprocessing
-# def preprocess_signal(ecg, eda, temp):
-#     ecg = np.array(ecg).reshape(-1, 1)
-#     eda = np.array(eda).reshape(-1, 1)
-#     temp = np.array(temp).reshape(-1, 1)
-#     return np.concatenate([ecg, eda, temp], axis=1)
-
-# # 🔮 Prediction
-# def predict_stress(ecg, eda, temp):
-#     try:
-#         signal = preprocess_signal(ecg, eda, temp).reshape(1, 1280, 3)
-#         features = feature_extractor.predict(signal)
-#         features_scaled = scaler.transform(features)
-#         raw_pred = elm.predict(features_scaled)
-#         raw_pred = np.clip(raw_pred, -20, 20)
-#         prob = 1 / (1 + np.exp(-raw_pred))
-#         label = "Stress" if prob > 0.5 else "No Stress"
-#         return f"{label} (Confidence: {prob[0][0]:.2f})"
-#     except Exception as e:
-#         return f"❌ Error: {str(e)}"
-
-# # 🧪 Parser for string inputs
-# def parse_and_predict(ecg_str, eda_str, temp_str):
-#     try:
-#         ecg = list(map(float, ecg_str.strip().split(',')))
-#         eda = list(map(float, eda_str.strip().split(',')))
-#         temp = list(map(float, temp_str.strip().split(',')))
-#         if len(ecg) != 1280 or len(eda) != 1280 or len(temp) != 1280:
-#             return "❌ Each signal must be exactly 1280 samples long."
-#         return predict_stress(ecg, eda, temp)
-#     except:
-#         return "❌ Invalid input format. Use comma-separated float values."
-
-# # 📊 Demo stress data
-# def generate_stress_sample():
-#     t = np.linspace(0, 10, 1280)
-#     ecg = 0.6 * np.sin(2 * np.pi * 1.8 * t) + 0.1 * np.random.randn(1280)
-#     eda = np.linspace(0.5, 1.2, 1280) + 0.05 * np.random.randn(1280)
-#     eda += np.random.choice([0, 0.3], size=1280, p=[0.95, 0.05])
-#     temp = 36.2 + 0.01 * np.random.randn(1280) - 0.05 * np.linspace(0, 1, 1280)
-#     return ecg, eda, temp
-
-# # 👇 Separate into text strings
-# ecg_demo, eda_demo, temp_demo = generate_stress_sample()
-# demo_example = [
-#     ','.join([str(round(v, 4)) for v in ecg_demo]),
-#     ','.join([str(round(v, 4)) for v in eda_demo]),
-#     ','.join([str(round(v, 4)) for v in temp_demo])
-# ]
-
-# # 🎛️ Gradio Interface
-# iface = gr.Interface(
-#     fn=parse_and_predict,
-#     inputs=[
-#         gr.Textbox(label="ECG Signal (1280 values, comma-separated)", lines=4),
-#         gr.Textbox(label="EDA Signal (1280 values, comma-separated)", lines=4),
-#         gr.Textbox(label="Temperature Signal (1280 values, comma-separated)", lines=4)
-#     ],
-#     outputs=gr.Textbox(label="Prediction"),
-#     examples=[demo_example],
-#     title="🧠 Stress Detection from Physiological Signals",
-#     description="Enter 1280 comma-separated ECG, EDA, and Temp signal values. The model will classify stress vs no stress."
-# )
-
-# # 🏁 Launch
-# if __name__ == "__main__":
-#     iface.launch()
-
-
 import numpy as np
-import pandas as pd
 import gradio as gr
+import matplotlib.pyplot as plt
 from tensorflow.keras.models import load_model, Model
 from hpelm import ELM
 import joblib
-import tempfile
-
-# 🔄 Load scaler and ELM model
-scaler = joblib.load("elm_scaler1.pkl")
-elm = ELM(128, 1, classification='c')
-elm.load("elm_model1.txt")
-
-# ⚙️ Load trained MobileNet1D and create feature extractor
-mobilenet = load_model("physio_model2.h5")
-feature_extractor = Model(inputs=mobilenet.input, outputs=mobilenet.get_layer("penultimate_dense").output)
-
-def preprocess_signal(ecg, eda, temp):
-    ecg = np.array(ecg).reshape(-1, 1)
-    eda = np.array(eda).reshape(-1, 1)
-    temp = np.array(temp).reshape(-1, 1)
-    return np.concatenate([ecg, eda, temp], axis=1)
+import io
+import base64
 
-def predict_stress(ecg, eda, temp):
-    try:
-        signal = preprocess_signal(ecg, eda, temp)  # Shape: (1280, 3)
-        signal = signal.reshape(1, 1280, 3)
+# === Load Models ===
+mobilenet_model = load_model("physio_model2.h5")
+feature_model = Model(inputs=mobilenet_model.input, outputs=mobilenet_model.get_layer("penultimate_dense").output)
 
-        # 🔍 Feature Extraction
-        features = feature_extractor.predict(signal)
-
-        # 🔄 Scaling for ELM
-        features_scaled = scaler.transform(features)
-
-        # 🔮 ELM Inference
-        raw_pred = elm.predict(features_scaled)
-        raw_pred = np.clip(raw_pred, -20, 20)
-        prob = 1 / (1 + np.exp(-raw_pred))
-        label = "Stress" if prob > 0.5 else "No Stress"
-
-        return f"{label} (Confidence: {prob[0][0]:.2f})"
-    
-    except Exception as e:
-        return f"❌ Error: {str(e)}"
+scaler = joblib.load("elm_scaler1.pkl")
 
-# ✅ Predict from CSV File
-def predict_from_csv(file_obj):
-    try:
-        df = pd.read_csv(file_obj.name)
-        if df.shape != (1280, 3) or list(df.columns) != ["ECG", "EDA", "Temp"]:
-            return "❌ CSV must have shape (1280, 3) with columns: ECG, EDA, Temp"
-        ecg = df["ECG"].values
-        eda = df["EDA"].values
-        temp = df["Temp"].values
-        return predict_stress(ecg, eda, temp)
-    except Exception as e:
-        return f"❌ Failed to read CSV: {str(e)}"
+elm = ELM(128, 1, classification="c")
+elm.load("elm_model1.txt")
 
-# ✅ Generate deterministic synthetic stress signal (not random)
-def generate_stress_sample():
+# === Generate stress physiological signal ===
+def generate_stress_signal():
     timesteps = 1280
     t = np.linspace(0, 10, timesteps)
-
-    # Simulated stress ECG: High-frequency sinusoid
-    ecg = 0.6 * np.sin(2 * np.pi * 1.8 * t)
-
-    # Simulated EDA: Linearly increasing with occasional burst
-    eda = 0.5 + 0.7 * (t / t.max())
-    eda[::256] += 0.3  # Small spikes
-
-    # Simulated Temp: Slight downward trend
-    temp = 36.4 - 0.05 * (t / t.max())
-
+    ecg = 0.6 * np.sin(2 * np.pi * 1.8 * t) + 0.1 * np.random.randn(timesteps)
+    eda = np.linspace(0.5, 1.2, timesteps) + 0.1 * np.random.randn(timesteps)
+    eda += np.random.choice([0, 0.5], size=timesteps, p=[0.9, 0.1])
+    temp = 36.2 + 0.02 * np.random.randn(timesteps) - 0.05 * np.linspace(0, 1, timesteps)
     return np.stack([ecg, eda, temp], axis=1)
 
-def generate_stress_sample_csv():
-    sample = generate_stress_sample()
-    df = pd.DataFrame(sample, columns=["ECG", "EDA", "Temp"])
-    tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".csv")
-    df.to_csv(tmp.name, index=False)
-    return tmp.name
-
-# ✅ Gradio Interface
-with gr.Blocks() as demo:
-    gr.Markdown("# 🧠 Stress Detection from Physiological Signals")
-    gr.Markdown("Upload a CSV file with 1280 rows and 3 columns: **ECG**, **EDA**, **Temp**")
-
-    with gr.Row():
-        file_input = gr.File(label="📁 Upload CSV file", file_types=[".csv"])
-        output = gr.Textbox(label="Prediction")
-
-    file_input.change(fn=predict_from_csv, inputs=file_input, outputs=output)
-
-    gr.Markdown("### Or generate a sample stressed input:")
-    generate_btn = gr.Button("Generate Stress Sample CSV")
-    demo_file = gr.File(label="⬇️ Download sample and re-upload above")
-
-    generate_btn.click(fn=generate_stress_sample_csv, outputs=demo_file)
-
-# 🏁 Launch
-if __name__ == "__main__":
-    demo.launch(server_name="0.0.0.0", server_port=7860, share=True, show_api=True)
+# === Prediction Function ===
+def predict_from_generated():
+    stress_input = generate_stress_signal()
+    model_input = stress_input[np.newaxis, ...]  # (1, 1280, 3)
+
+    features = feature_model.predict(model_input)
+    features_scaled = scaler.transform(features)
+    raw_pred = elm.predict(features_scaled)
+    raw_pred_clipped = np.clip(raw_pred, -20, 20)
+    prob_pred = 1 / (1 + np.exp(-raw_pred_clipped))
+    label_pred = "Stress" if prob_pred > 0.5 else "No Stress"
+
+    # Plot
+    fig, axs = plt.subplots(1, 3, figsize=(12, 3))
+    axs[0].plot(stress_input[:, 0])
+    axs[0].set_title("ECG")
+    axs[1].plot(stress_input[:, 1])
+    axs[1].set_title("EDA")
+    axs[2].plot(stress_input[:, 2])
+    axs[2].set_title("Temp")
+    plt.tight_layout()
+
+    buf = io.BytesIO()
+    plt.savefig(buf, format="png")
+    plt.close(fig)
+    buf.seek(0)
+    img_str = base64.b64encode(buf.read()).decode("utf-8")
+    img_data = f"data:image/png;base64,{img_str}"
+
+    prob_display = float(prob_pred[0][0])
+    return label_pred, f"{prob_display:.3f}", img_data
+
+# === Gradio UI ===
+iface = gr.Interface(
+    fn=predict_from_generated,
+    inputs=[],
+    outputs=[
+        gr.Textbox(label="Predicted Label"),
+        gr.Textbox(label="Stress Probability"),
+        gr.Image(label="Signal Plot")
+    ],
+    title="🧠 Stress Detection from Simulated Physiological Signal",
+    description="This demo generates synthetic stressed signals (ECG, EDA, Temp), extracts deep features using a pretrained MobileNet1D model, and classifies stress using ELM."
+)
+
+iface.launch()