app.py

import gradio as gr
import numpy as np
import pandas as pd
from io import BytesIO
import matplotlib.pyplot as plt
from scipy.stats import skew, kurtosis
import joblib
import os

SAMPLING_RATE = 125
WINDOW_SIZE = 125
SEQUENCE_LENGTH = 10

scaler = joblib.load("models/scaler/scaler.joblib")

HYBRID_MODEL = [
    ("LSTM", "RandomForest", "LSTM + Random Forest", "models/LSTM/rf_lstm_optuna_10122025_122323.joblib", "models/RandomForest/rf_lstm_optuna_10122025_123350.joblib"),
    ("LSTM", "SVM", "LSTM + SVM", "models/LSTM/SVM_LSTM_optuna_10122025_124318.joblib", "models/SVM/svm_LSTM_optuna_10122025_125033.joblib"),
    ("LSTM", "XGBOOST", "LSTM + XGBoost", "models/LSTM/xgboost_LSTM_optuna_10122025_113333.joblib", "models/XGBOOST/xgboost_LSTM_optuna_10122025_113701.joblib"),
    ("TRANSFORMER", "RandomForest", "Transformer + Random Forest", "models/TRANSFORMER/rf_transformer_optuna_10122025_045920.joblib", "models/RandomForest/rf_transformer_optuna_10122025_045552.joblib"),
    ("TRANSFORMER", "SVM", "Transformer + SVM", "models/TRANSFORMER/svm_transformer_optuna_10122025_050213.joblib", "models/SVM/svm_transformer_optuna_10122025_045226.joblib"),
    ("TRANSFORMER", "XGBOOST", "Transformer + XGBoost", "models/TRANSFORMER/xgboost_transformer_optuna_10122025_050346.joblib", "models/XGBOOST/xgboost_transformer_optuna_10122025_044957.joblib"),
]

SINGLE_MODELS = [
    ("Single LSTM", "models/LSTM/single_LSTM_default_10122025_135523.joblib"),
    ("Single Transformer", "models/TRANSFORMER/single_transformer_grid_10122025_115452.joblib"),
]

MODEL_PAIRS = [m for m in HYBRID_MODEL if 'lstm' in m[3].lower() or 'lstm' in m[4].lower() or 'transformer' in m[3].lower() or 'transformer' in m[4].lower()]

def ekstraksi_fitur_statistik(w):
    return np.array([skew(w), kurtosis(w), np.min(w), np.max(w), np.std(w)])

def ekstraksi_fitur_sinyal(ecg):
    n = len(ecg) // WINDOW_SIZE
    fitur = []
    for i in range(n):
        w = ecg[i*WINDOW_SIZE:(i+1)*WINDOW_SIZE]
        fitur.append(ekstraksi_fitur_statistik(w))
    return np.array(fitur)

def buat_sequence(fitur):
    x = []
    for i in range(len(fitur) - SEQUENCE_LENGTH + 1):
        x.append(fitur[i:i+SEQUENCE_LENGTH])
    return np.array(x)

def preprocessing_sinyal(ecg):
    fitur = ekstraksi_fitur_sinyal(ecg)
    fitur_scaled = scaler.transform(fitur)
    seq = buat_sequence(fitur_scaled)
    return seq

def load_hybrid_models(p1, p2):
    m1 = joblib.load(p1)
    m2 = joblib.load(p2)
    return m1, m2

def analisis_sinyal(file, model_pair_label):

    single = next((s for s in SINGLE_MODELS if s[0] == model_pair_label), None)
    if single:
        model_path = os.path.join(os.path.dirname(__file__), single[1])
        model = joblib.load(model_path)
        df = pd.read_csv(file.name)
        sinyal = df.values.flatten()
        seq = preprocessing_sinyal(sinyal)
        try:
            pred = model.predict(seq)
            if hasattr(model, 'predict_proba'):
                pred_proba = model.predict_proba(seq)
                label = int(np.argmax(pred_proba[0]))
            else:
                label = int(pred[0])
        except Exception as e:
            return f"Failed to predict with single learning model: {e}", None
    else:
        selected = next((m for m in MODEL_PAIRS if m[2] == model_pair_label), None)
        if not selected:
            return "Model not found", None
        p1 = selected[3]
        p2 = selected[4]
        p1 = os.path.join(os.path.dirname(__file__), p1)
        p2 = os.path.join(os.path.dirname(__file__), p2)
        model_dl, model_clf = load_hybrid_models(p1, p2)
        df = pd.read_csv(file.name)
        sinyal = df.values.flatten()
        seq = preprocessing_sinyal(sinyal)
        try:
            fitur = model_dl.predict(seq)
            debug_info = f"Type model_dl: {type(model_dl)}, Output predict: {type(fitur)}, Shape: {getattr(fitur, 'shape', None)}"
        except Exception as e:
            return f"Failed to predict with feature extraction model: {e}", None
        
        n_features_model = getattr(model_clf, 'n_features_in_', None)
        if hasattr(fitur, 'shape') and n_features_model is not None and fitur.shape[0] >= n_features_model:
            fitur = fitur[-n_features_model:].reshape(1, n_features_model)

        if n_features_model is not None and (not hasattr(fitur, 'shape') or fitur.shape[1] != n_features_model):
            return debug_info + f"\nNumber of extracted features ({fitur.shape[1] if hasattr(fitur, 'shape') else '?'}) does not match the model's expected number ({n_features_model}). Ensure the feature extraction model and classifier are compatible.", None
        if hasattr(model_clf, "predict_proba"):
            pred = model_clf.predict_proba(fitur)[0]
            label = int(np.argmax(pred))
        else:
            label = int(model_clf.predict(fitur)[0])

    fig, ax = plt.subplots(figsize=(8, 3))
    ax.plot(sinyal, label="Raw ECG", color="#2196f3", linewidth=1)
 
    if len(sinyal) > 25:
        ma = pd.Series(sinyal).rolling(window=25, min_periods=1, center=True).mean()
        ax.plot(ma, label="Moving Average", color="#ff9800", linewidth=2, alpha=0.7)
    ax.set_title("ECG Signal (Raw & Smoothed)")
    ax.set_xlabel("Sample")
    ax.set_ylabel("Amplitude")
    ax.legend()
    ax.grid(True, linestyle='--', alpha=0.5)
    fig.tight_layout()
    buf = BytesIO()
    fig.savefig(buf, format="png")
    buf.seek(0)
    import PIL.Image
    img = PIL.Image.open(buf)
    return str(label), img

css = """
body {background-color: #181818; color: #f5f5f5;}
.gradio-container, .gradio-app {background-color: #181818 !important;}
#title {text-align:center; font-size:32px; font-weight:700; margin-bottom:20px; color:#f5f5f5;}
#subtitle {text-align:center; font-size:18px; margin-bottom:40px; color:#bbbbbb;}
input, select, textarea, .gr-button, .gr-input, .gr-textbox, .gr-dropdown, .gr-file, .gr-image {
    background-color: #232323 !important;
    color: #f5f5f5 !important;
    border-color: #444 !important;
}
.gr-button {border-radius: 6px;}
.license-box {
    border: 2px solid #fff;
    border-radius: 10px;
    padding: 18px;
    margin-top: 18px;
    background: #111;
    color: #fff;
}
"""



with gr.Blocks() as demo:
    gr.HTML('<div id="title">Atrial Fibrillation Detection</div>')
    gr.HTML('<div id="subtitle">Analyze ECG data for Atrial Fibrillation presence</div>')
    file_upload = gr.File(label="Upload Dataset/Signal (CSV)", file_types=[".csv"])
    shared_file_path = gr.State()
    clear_btn = gr.Button("Clear File")

    def store_file(file):
        return file.name if file is not None else None

    file_upload.change(store_file, inputs=file_upload, outputs=shared_file_path)

    def clear_file():
        return None, None
    clear_btn.click(clear_file, inputs=None, outputs=[file_upload, shared_file_path])

    with gr.Tab("Dataset Info"):
        gr.Markdown("""
    ## Dataset Files
    The dataset files below are the same as those used for model testing in this application. Please download using the buttons below:
            """)
        with gr.Row():
            gr.File(value="Data/mimic_perform_af_001_data.csv", label="Download Atrial Fibrillation Data", interactive=False)
            gr.File(value="Data/mimic_perform_non_af_001_data.csv", label="Download Non-Atrial Fibrillation Data", interactive=False)
        gr.Markdown(
            """
<div class="license-box">
<b>Dataset License</b>
<br>This dataset is licensed under the Open Data Commons Open Database License v1.0 (ODbL 1.0 license).<br>
Further details: <a href="https://opendatacommons.org/licenses/odbl/summary/" target="_blank">ODbL 1.0</a><br><br>
This dataset is derived from the MIMIC III Waveform Database:<br>
Moody, B., Moody, G., Villarroel, M., Clifford, G. D., & Silva, I. (2020). MIMIC-III Waveform Database (version 1.0). PhysioNet. <a href="https://doi.org/10.13026/c2607m" target="_blank">https://doi.org/10.13026/c2607m</a><br><br>
The MIMIC III Waveform Database is licensed under the ODbL 1.0 license.<br><br>
The MIMIC-III database is described in:<br>
Johnson, A. E. W., Pollard, T. J., Shen, L., Lehman, L. H., Feng, M., Ghassemi, M., Moody, B., Szolovits, P., Celi, L. A., & Mark, R. G. (2016). MIMIC-III, a freely accessible critical care database. Scientific Data, 3, 160035. <a href="https://doi.org/10.1038/sdata.2016.35" target="_blank">https://doi.org/10.1038/sdata.2016.35</a><br><br>
It is available on PhysioNet: <a href="https://physionet.org/" target="_blank">https://physionet.org/</a><br>
Goldberger, A., Amaral, L., Glass, L., Hausdorff, J., Ivanov, P. C., Mark, R., ... & Stanley, H. E. (2000). PhysioBank, PhysioToolkit, and PhysioNet: Components of a new research resource for complex physiologic signals. Circulation [Online]. 101 (23), pp. e215–e220.<br><br>
The following annotations of AF and non-AF were used to create the dataset:<br>
Bashar, Syed Khairul (2020): Atrial Fibrillation annotations of electrocardiogram from MIMIC III matched subset. figshare. Dataset. <a href="https://doi.org/10.6084/m9.figshare.12149091.v1" target="_blank">https://doi.org/10.6084/m9.figshare.12149091.v1</a><br><br>
Bashar, S.K., Ding, E., Walkey, A.J., McManus, D.D. and Chon, K.H., 2019. Noise Detection in Electrocardiogram Signals for Intensive Care Unit Patients. IEEE Access, 7, pp.88357-88368. <a href="https://doi.org/10.1109/ACCESS.2019.2926199" target="_blank">https://doi.org/10.1109/ACCESS.2019.2926199</a><br><br>
This annotation information is reproduced under the terms of the <a href="https://creativecommons.org/licenses/by/4.0/" target="_blank">CC BY 4.0 licence</a>
</div> """,elem_id=None)

    with gr.Tab("Analyze Dataset"):
        info_ds = gr.Textbox(label="Dataset Info", interactive=False, lines=5)
        preview_ds = gr.Dataframe(label="Data Preview", interactive=False)
        plot_ds = gr.Plot(label="ECG Signal Plot")
        btn_ds = gr.Button("Analyze Dataset")

        def analyze_dataset(file_path):
            import pandas as pd
            import matplotlib.pyplot as plt
            if file_path is None:
                return "No file uploaded. Please upload a CSV file first.", None, None
            try:
                df = pd.read_csv(file_path)
            except Exception as e:
                return f"Failed to read file. Make sure the file is a valid CSV. Error: {e}", None, None

            info_lines = []
            info_lines.append(f"Shape: {df.shape}")
            info_lines.append(f"Columns: {list(df.columns)}")
            info_lines.append(f"Missing: {df.isnull().sum().to_dict()}")

            duration = None
            sampling_rate = 125  
            n_samples = len(df)
            duration = n_samples / sampling_rate
            info_lines.append(f"Sampling rate: {sampling_rate} Hz")
            info_lines.append(f"Data duration: {duration:.2f} seconds ({duration/60:.2f} minutes)")
            preview = df.head(10)
            fig = None
            ecg_col = None
            for col in df.columns:
                if 'ecg' in col.lower():
                    ecg_col = col
                    break
            if ecg_col is None:
                return "The uploaded CSV does not contain an 'ecg' column. Please upload a CSV file with an 'ecg' feature/column.", preview, None
            if not pd.api.types.is_numeric_dtype(df[ecg_col]):
                return f"The selected signal column ('{ecg_col}') is not numeric. Please upload a valid ECG CSV.", preview, None

            plot_samples = min(sampling_rate*10, len(df))
            try:
                fig, ax = plt.subplots()
                ax.plot(df[ecg_col].values[:plot_samples])
                ax.set_title(f"First 10 Seconds Signal Plot: {ecg_col}")
                ax.set_xlabel("Sample")
                ax.set_ylabel("Amplitude")
            except Exception as e:
                return f"Failed to plot ECG signal: {e}", preview, None
            return "\n".join(info_lines), preview, fig

        btn_ds.click(analyze_dataset, inputs=[shared_file_path], outputs=[info_ds, preview_ds, plot_ds])

    with gr.Tab("Analyze Model"):
        all_model_labels = [m[2] for m in MODEL_PAIRS] + [s[0] for s in SINGLE_MODELS]
        pilih_model = gr.Dropdown(all_model_labels, label="Select Model", value=all_model_labels[0])
        hasil = gr.Textbox(label="Prediction Result", interactive=False)
        tombol = gr.Button("Predict")

        def handle_predict(file_path, model_label):
            if file_path is None or str(file_path).strip() == "":
                return "No file uploaded. Please upload a CSV file first."
            if not model_label:
                return "No model selected. Please select a model."
            try:
                import pandas as pd
                df = pd.read_csv(file_path)
                ecg_col = None
                for col in df.columns:
                    if 'ecg' in col.lower():
                        ecg_col = col
                        break
                if ecg_col is None:
                    return "The uploaded CSV does not contain an 'ecg' column. Please upload a CSV file with an 'ecg' feature/column for prediction."
                if not pd.api.types.is_numeric_dtype(df[ecg_col]):
                    return f"The selected signal column ('{ecg_col}') is not numeric. Please upload a valid ECG CSV."
                class DummyFile:
                    def __init__(self, name):
                        self.name = name
                dummy_file = DummyFile(file_path)
                result, _ = analisis_sinyal(dummy_file, model_label)
                if str(result).strip() == '0':
                    return 'Non-AF'
                elif str(result).strip() == '1':
                    return 'AF'
                else:
                    return str(result)
            except Exception as e:
                return f"Prediction failed: {e}"

        tombol.click(handle_predict, inputs=[shared_file_path, pilih_model], outputs=[hasil])

        def clear_file():
            return None, None, None
        clear_btn.click(clear_file, inputs=None, outputs=[file_upload, shared_file_path, hasil])

    demo.launch(css=css)