app.py

# -*- coding: utf-8 -*-
"""
Created on Tue Feb  4 14:44:33 2025

@author: Ashmitha
"""

#-------------------------------------Libraries-------------------------
import pandas as pd
import numpy as np
import gradio as gr
from sklearn.metrics import mean_squared_error,r2_score
from scipy.stats import pearsonr
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import KFold
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import GRU,Dense,Dropout,BatchNormalization,LeakyReLU
from tensorflow.keras.optimizers import Adam
from tensorflow.keras import regularizers
from tensorflow.keras.callbacks import ReduceLROnPlateau,EarlyStopping
import os
from sklearn.preprocessing import MinMaxScaler
from keras.layers import Conv1D,MaxPooling1D,Dense,Flatten,Dropout,LeakyReLU
from keras.callbacks import ReduceLROnPlateau,EarlyStopping
from sklearn.ensemble import RandomForestRegressor
from xgboost import XGBRegressor
import io
from sklearn.feature_selection import SelectFromModel
import tempfile
import matplotlib.pyplot as plt
import seaborn as sns
#import lightgbm as lgb
import lightgbm as lgb
import numpy as np
from sklearn.model_selection import KFold
from sklearn.preprocessing import StandardScaler
from lightgbm import LGBMRegressor
from sklearn.svm import SVR
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from lightgbm import LGBMRegressor
from sklearn.preprocessing import StandardScaler
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.svm import SVR as SVR_Model 

#--------------------------------------------------FNNModel-----------------------------------
def FNNModel(trainX, trainy, testX=None, testy=None, epochs=1000, batch_size=64, learning_rate=0.0001, 
             l1_reg=0.001, l2_reg=0.001, dropout_rate=0.2):

    # Scale the input data
    scaler = MinMaxScaler()
    trainX_scaled = scaler.fit_transform(trainX)
    testX_scaled = scaler.transform(testX) if testX is not None else None

    # Scale the target variable
    target_scaler = MinMaxScaler()
    trainy_scaled = target_scaler.fit_transform(trainy.reshape(-1, 1))

    # Model definition
    model = Sequential()

    # Input Layer
    model.add(Dense(512, input_shape=(trainX.shape[1],), kernel_initializer='he_normal',
                    kernel_regularizer=regularizers.l1_l2(l1=l1_reg, l2=l2_reg)))
    model.add(BatchNormalization())
    model.add(Dropout(dropout_rate))
    model.add(LeakyReLU(alpha=0.1))

    # Hidden Layers
    model.add(Dense(256, kernel_initializer='he_normal', kernel_regularizer=regularizers.l1_l2(l1=l1_reg, l2=l2_reg)))
    model.add(BatchNormalization())
    model.add(Dropout(dropout_rate))
    model.add(LeakyReLU(alpha=0.1))

    model.add(Dense(128, kernel_initializer='he_normal', kernel_regularizer=regularizers.l1_l2(l1=l1_reg, l2=l2_reg)))
    model.add(BatchNormalization())
    model.add(Dropout(dropout_rate))
    model.add(LeakyReLU(alpha=0.1))

    model.add(Dense(64, kernel_initializer='he_normal', kernel_regularizer=regularizers.l1_l2(l1=l1_reg, l2=l2_reg)))
    model.add(BatchNormalization())
    model.add(Dropout(dropout_rate))
    model.add(LeakyReLU(alpha=0.1))

    model.add(Dense(32, kernel_initializer='he_normal', kernel_regularizer=regularizers.l1_l2(l1=l1_reg, l2=l2_reg)))
    model.add(BatchNormalization())
    model.add(Dropout(dropout_rate))
    model.add(LeakyReLU(alpha=0.1))

    # Output Layer
    model.add(Dense(1, activation="relu"))  

    # Compile Model
    model.compile(loss='mse', optimizer=Adam(learning_rate=learning_rate), metrics=['mse'])

    # Callbacks
    callbacks = [
        ReduceLROnPlateau(monitor='val_loss', patience=10, verbose=1, factor=0.5, min_lr=1e-6),
        EarlyStopping(monitor='val_loss', verbose=1, restore_best_weights=True, patience=10)
    ]

    # Train model
    history = model.fit(trainX_scaled, trainy_scaled, epochs=epochs, batch_size=batch_size, validation_split=0.1, 
                        verbose=1, callbacks=callbacks)

    # Predictions
    predicted_train = model.predict(trainX_scaled).flatten()
    predicted_test = model.predict(testX_scaled).flatten() if testX is not None else None

    # Inverse transform predictions
    predicted_train = target_scaler.inverse_transform(predicted_train.reshape(-1, 1)).flatten()
    if predicted_test is not None:
        predicted_test = target_scaler.inverse_transform(predicted_test.reshape(-1, 1)).flatten()

    return predicted_train, predicted_test, history



#--------------------------------------------------CNNModel-------------------------------------------

# CHANGE TO RNN MODEL OR DNN Model
def CNNModel(trainX, trainy, testX, testy, epochs=1000, batch_size=64, learning_rate=0.0001, l1_reg=0.0001, l2_reg=0.0001, dropout_rate=0.3,feature_selection=True):
    
   
    
    # Scaling the inputs
    scaler = MinMaxScaler()
    trainX_scaled = scaler.fit_transform(trainX)
    if testX is not None:
        testX_scaled = scaler.transform(testX)
    
    # Reshape for CNN input (samples, features, channels)
    trainX = trainX_scaled.reshape((trainX.shape[0], trainX.shape[1], 1))
    if testX is not None:
        testX = testX_scaled.reshape((testX.shape[0], testX.shape[1], 1))
    
    model = Sequential()
    
    # Convolutional layers
    model.add(Conv1D(512, kernel_size=3, activation='relu', input_shape=(trainX.shape[1], 1), kernel_regularizer=regularizers.l1_l2(l1=l1_reg, l2=l2_reg)))
    model.add(MaxPooling1D(pool_size=2))
    model.add(Dropout(dropout_rate))
    
    model.add(Conv1D(256, kernel_size=3, activation='relu', kernel_regularizer=regularizers.l1_l2(l1=l1_reg, l2=l2_reg)))
    model.add(MaxPooling1D(pool_size=2))
    model.add(Dropout(dropout_rate))

    model.add(Conv1D(128, kernel_size=3, activation='relu', kernel_regularizer=regularizers.l1_l2(l1=l1_reg, l2=l2_reg)))
    model.add(MaxPooling1D(pool_size=2))
    model.add(Dropout(dropout_rate))
    
    # Flatten and Dense layers
    model.add(Flatten())
    model.add(Dense(64, kernel_regularizer=regularizers.l1_l2(l1=l1_reg, l2=l2_reg)))
    model.add(LeakyReLU(alpha=0.1))
    model.add(Dropout(dropout_rate))

    model.add(Dense(1, activation='linear'))

    # Compile the model
    model.compile(loss='mse', optimizer=Adam(learning_rate=learning_rate), metrics=['mse'])

    # Callbacks
    learning_rate_reduction = ReduceLROnPlateau(monitor='val_loss', patience=5, verbose=1, factor=0.5, min_lr=1e-6)
    early_stopping = EarlyStopping(monitor='val_loss', verbose=1, restore_best_weights=True, patience=10)
    
    # Train the model
    history = model.fit(trainX, trainy, epochs=epochs, batch_size=batch_size, validation_split=0.1, verbose=1, 
                        callbacks=[learning_rate_reduction, early_stopping])
    
    predicted_train = model.predict(trainX).flatten()
    predicted_test = model.predict(testX).flatten() if testX is not None else None
    
    return predicted_train, predicted_test, history
#-------------------------------------------LGBoost-----------------------------------------------


#def LGBoostModel(trainX, trainy, testX, testy, learning_rate=0.05, num_leaves=31, max_depth=-1, min_child_samples=20, n_estimators=500):
    
    #scaler = StandardScaler()
    #trainX_scaled = scaler.fit_transform(trainX)
    #testX_scaled = scaler.transform(testX)

    # Create and train the model
   # lgbm_model = LGBMRegressor(
      #  n_estimators=n_estimators,
       # learning_rate=learning_rate,
      #  num_leaves=num_leaves,  # More leaves for complex data
       # max_depth=max_depth,  # No limit (-1) allows deeper trees
      #  min_child_samples=min_child_samples,  # Minimum data needed to split
      #  reg_alpha=0.1,  # L1 regularization
      #  reg_lambda=0.1,  # L2 regularization
   # )

   # history = lgbm_model.fit(trainX_scaled, trainy)

    # Predicting the values
  #  predicted_train = lgbm_model.predict(trainX_scaled)
   # predicted_test = lgbm_model.predict(testX_scaled)

   # return predicted_train, predicted_test, history
def LGBoostModel(trainX, trainy, testX, testy, learning_rate=0.05, num_leaves=15, max_depth=5, min_child_samples=10, n_estimators=1000):
    """
    Train a LightGBM model with the given data and parameters.
    """
    print(f"Training LightGBM Model with n_estimators={n_estimators}, learning_rate={learning_rate}, num_leaves={num_leaves}, max_depth={max_depth}")

    # Standardizing the data
    scaler = StandardScaler()
    trainX_scaled = scaler.fit_transform(trainX)
    testX_scaled = scaler.transform(testX)

    # Create and train the model
    lgbm_model = LGBMRegressor(
        n_estimators=n_estimators,
        learning_rate=learning_rate,
        num_leaves=num_leaves,
        max_depth=max_depth,
        min_child_samples=min_child_samples,
        reg_alpha=0.01,  # Reduced L1 regularization
        reg_lambda=0.01,
        verbose=-1# Reduced L2 regularization
    )

    lgbm_model.fit(trainX_scaled, trainy)

    # Predicting the values
    predicted_train = lgbm_model.predict(trainX_scaled)
    predicted_test = lgbm_model.predict(testX_scaled)

    return predicted_train, predicted_test, lgbm_model

#------------------------------------------RFModel---------------------------------------------------
def RFModel(trainX, trainy, testX, testy, n_estimators=100, max_depth=None,feature_selection=True):
    
    
    # Log transformation of the target variable
   
    # Scaling the feature data
    scaler = MinMaxScaler()
    trainX_scaled = scaler.fit_transform(trainX)
    if testX is not None:
        testX_scaled = scaler.transform(testX)
    
    # Define and train the RandomForest model
    rf_model = RandomForestRegressor(n_estimators=n_estimators, max_depth=max_depth, random_state=42)
    history=rf_model.fit(trainX_scaled, trainy)
    
    
    # Predictions
    predicted_train = rf_model.predict(trainX_scaled)
    predicted_test = rf_model.predict(testX_scaled) if testX is not None else None
    
    return predicted_train, predicted_test,history

#--------------------------------------SVR-------------------------------------
 # Avoid function name conflict

def SVR(trainX, trainy, testX, testy, kernel='rbf', C=1.0, epsilon=0.1, gamma='scale'):
    """
    Train a Support Vector Regression (SVR) model with the given data and parameters.
    
    Parameters:
        trainX, trainy: Training data (features & target)
        testX, testy: Testing data (features & target)
        kernel: 'linear', 'poly', 'rbf' (default is 'rbf')
        C: Regularization parameter
        epsilon: Defines a margin of tolerance where predictions don't get penalized
        gamma: Kernel coefficient (used for 'rbf' and 'poly')
    """
    print(f"Training SVR Model with kernel={kernel}, C={C}, epsilon={epsilon}, gamma={gamma}")

    # Create a pipeline with scaling and SVR
    svr_model = Pipeline([
        ('scaler', StandardScaler()),
        ('svr', SVR_Model(kernel=kernel, C=C, epsilon=epsilon, gamma=gamma))
    ])
    
    # Train the model
    svr_model.fit(trainX, trainy)
    
    # Predict values
    predicted_train = svr_model.predict(trainX)
    predicted_test = svr_model.predict(testX)
    
    return predicted_train, predicted_test, svr_model



#------------------------------------------------------------------File--------------------------------------------
def read_csv_file(uploaded_file):
    if uploaded_file is not None:
        if hasattr(uploaded_file, 'data'):  # For NamedBytes
            return pd.read_csv(io.BytesIO(uploaded_file.data))
        elif hasattr(uploaded_file, 'name'):  # For NamedString
            return pd.read_csv(uploaded_file.name)
    return None


#_-------------------------------------------------------------NestedKFold Cross Validation---------------------
def calculate_topsis_score(df):
    # Normalize the data
    norm_df = (df.iloc[:, 1:] - df.iloc[:, 1:].min()) / (df.iloc[:, 1:].max() - df.iloc[:, 1:].min())

    # Calculate the positive and negative ideal solutions
    ideal_positive = norm_df.max(axis=0)
    ideal_negative = norm_df.min(axis=0)

    # Calculate the Euclidean distances
    dist_positive = np.sqrt(((norm_df - ideal_positive) ** 2).sum(axis=1))
    dist_negative = np.sqrt(((norm_df - ideal_negative) ** 2).sum(axis=1))

    # Calculate the TOPSIS score
    topsis_score = dist_negative / (dist_positive + dist_negative)

    # Add the TOPSIS score to the dataframe
    df['TOPSIS_Score'] = topsis_score

    return df
#----------------------------------------------------------NestedKFoldCrossValidation------------
def NestedKFoldCrossValidation(training_data, training_additive, testing_data, testing_additive, 
                                training_dominance, testing_dominance, epochs, learning_rate, min_child_weight, batch_size=64,
                                outer_n_splits=2, kernel='rbf', C=1.0, epsilon=0.1, gamma='scale', output_file='cross_validation_results.csv',
                                predicted_phenotype_file='predicted_phenotype.csv', feature_selection=True):
    
    
    
    
    # Define calculate_topsis_score before using it
    

    # Original function logic continues here
    if 'phenotypes' not in training_data.columns:
        raise ValueError("Training data does not contain the 'phenotypes' column.")
    
    # Remove Sample ID columns from additive and dominance data
    training_additive = training_additive.iloc[:, 1:]
    testing_additive = testing_additive.iloc[:, 1:]
    training_dominance = training_dominance.iloc[:, 1:]
    testing_dominance = testing_dominance.iloc[:, 1:]
    A_square_training=training_additive**2
    D_square_training=training_dominance**2
    A_square_testing=testing_additive**2
    D_square_testing=testing_dominance**2
    additive_dominance_training=training_additive*training_dominance
    additive_dominance_testing=testing_additive*testing_dominance
    training_data_merged=np.concatenate([training_additive,training_dominance,A_square_training,D_square_training,additive_dominance_training], axis=1)
    testing_data_merged=np.concatenate([testing_additive,testing_dominance,A_square_testing,D_square_testing,additive_dominance_testing], axis=1)
    phenotypic_info=training_data['phenotypes'].values
    phenotypic_test_info=testing_data['phenotypes'].values if 'phenotypes' in testing_data.columns else None
    sample_ids=testing_data.iloc[:,0].values
    training_data_merged=pd.DataFrame(training_data_merged)
    testing_data_merged=pd.DataFrame(testing_data_merged)
    training_genotypic_data_merged=training_data_merged.iloc[:,1:].values
    testing_genotypic_data_merged=testing_data_merged.iloc[:,1:].values
    print(training_genotypic_data_merged)
    print(testing_genotypic_data_merged)
    outer_kf=KFold(n_splits=outer_n_splits)
    results=[]
    all_predicted_phenotypes=[]
    def calculate_metrics(true_values,predicted_values):
        mse=mean_squared_error(true_values,predicted_values)
        rmse=np.sqrt(mse)
        r2=r2_score(true_values,predicted_values)
        corr=pearsonr(true_values,predicted_values)[0]
        return mse,rmse,corr,r2
    models=[
        ('FNNModel',FNNModel),
        ('CNNModel', CNNModel),
        ('RFModel',RFModel),
        ('LGBoostModel',LGBoostModel),
        ('SVR',SVR)
    ]
    for outer_fold, (outer_train_index, outer_test_index) in enumerate(outer_kf.split(phenotypic_info), 1):
        outer_trainX = training_genotypic_data_merged[outer_train_index]
        outer_trainy = phenotypic_info[outer_train_index]

        
        if feature_selection:
            rf = RandomForestRegressor(n_estimators=100, random_state=42)
            rf.fit(outer_trainX, outer_trainy)  
            selector = SelectFromModel(rf, threshold="mean", prefit=True)
            outer_trainX = selector.transform(outer_trainX)
            testing_genotypic_data_merged_fold = selector.transform(testing_genotypic_data_merged)  # Transform testing data
        else:
            testing_genotypic_data_merged_fold = testing_genotypic_data_merged

        
        scaler = StandardScaler()
        outer_trainX = scaler.fit_transform(outer_trainX)  # Fit and transform on outer_trainX
        testing_genotypic_data_merged_fold = scaler.transform(testing_genotypic_data_merged_fold)  # Transform testing data
        outer_testX = testing_genotypic_data_merged_fold
        outer_testy = phenotypic_test_info
        for model_name, model_func in models:
            print(f"Running model: {model_name} for fold {outer_fold}")
            if model_name in ['FNNModel', 'CNNModel']:
                predicted_train, predicted_test, history = model_func(outer_trainX, outer_trainy, outer_testX, outer_testy, epochs=epochs, batch_size=batch_size)
            elif model_name in ['RFModel']:
                predicted_train, predicted_test, history = model_func(outer_trainX, outer_trainy, outer_testX, outer_testy)
            elif model_name in ['LGBoostModel']:
                
                predicted_train, predicted_test, history = model_func(outer_trainX, outer_trainy, outer_testX, outer_testy,learning_rate=0.05, num_leaves=31, max_depth=-1, min_child_samples=20, n_estimators=500)
            else:
                predicted_train, predicted_test, svr_model=model_func(outer_trainX,outer_trainy,outer_testX,outer_testy,kernel='rbf', C=1.0, epsilon=0.1, gamma='scale')
               
            # Calculate metrics
            mse_train, rmse_train, r2_train, corr_train = calculate_metrics(outer_trainy, predicted_train)
            mse_test, rmse_test, r2_test, corr_test = calculate_metrics(outer_testy, predicted_test) if outer_testy is not None else (None, None, None, None)

            results.append({
                'Model': model_name,
                'Fold': outer_fold,
                'Train_MSE': mse_train,
                'Train_RMSE': rmse_train,
                'Train_R2': r2_train,
                'Train_Corr': corr_train,
                'Test_MSE': mse_test,
                'Test_RMSE': rmse_test,
                'Test_R2': r2_test,
                'Test_Corr': corr_test
            })

            if predicted_test is not None:
                predicted_test_df = pd.DataFrame({
                    'Sample_ID': sample_ids,
                    'Predicted_Phenotype': predicted_test,
                    'Model': model_name
                })
                all_predicted_phenotypes.append(predicted_test_df)

    # Compile results
    results_df = pd.DataFrame(results)

    # Calculate the average metrics for each model
    if 'phenotypes' in testing_data.columns:
        avg_results_df = results_df.groupby('Model').agg({
           # 'Train_MSE': 'mean',
           # 'Train_RMSE': 'mean',
            'Train_R2': 'mean',
            'Train_Corr': 'mean',
            #'Test_MSE': 'mean',
            #'Test_RMSE': 'mean',
            'Test_R2': 'mean',
            'Test_Corr': 'mean'
        }).reset_index()
    else:
        avg_results_df = results_df.groupby('Model').agg({
            #'Train_MSE': 'mean',
           # 'Train_RMSE': 'mean',
            'Train_R2': 'mean',
            'Train_Corr': 'mean'
        }).reset_index()

    avg_results_df = calculate_topsis_score(avg_results_df)
    print(avg_results_df)

    # Save the results with TOPSIS scores to the file
    avg_results_df.to_csv(output_file, index=False)

    # Save predicted phenotypes 
    if all_predicted_phenotypes:
        predicted_all_df = pd.concat(all_predicted_phenotypes, axis=0, ignore_index=True)
        predicted_all_df.to_csv(predicted_phenotype_file, index=False)

    return avg_results_df, predicted_all_df if all_predicted_phenotypes else None
def visualize_topsis_scores(results_df):
    """
    Function to visualize the TOPSIS scores as a bar chart.
    """
    if 'TOPSIS_Score' not in results_df.columns:
        print("TOPSIS scores are missing in the DataFrame!")
        return None

    plt.figure(figsize=(10, 6))
    sns.barplot(x='Model', y='TOPSIS_Score', data=results_df, palette="viridis")
    plt.xlabel("Models", fontsize=12)
    plt.ylabel("TOPSIS Score", fontsize=12)
    plt.title("Model Performance - TOPSIS Score", fontsize=14)
    plt.xticks(rotation=45)
    plt.tight_layout()

    # Save the figure
    plt.savefig("topsis_scores.png")
    return "topsis_scores.png"
def run_cross_validation(training_file, training_additive_file, testing_file, testing_additive_file, 
                         training_dominance_file, testing_dominance_file, feature_selection, learning_rate, min_child_weight,kernel,C,epsilon,gamma):

    # Default parameters
    epochs = 1000
    batch_size = 64
    outer_n_splits = 2

    # Load datasets
    training_data = pd.read_csv(training_file.name)
    training_additive = pd.read_csv(training_additive_file.name)
    testing_data = pd.read_csv(testing_file.name)
    testing_additive = pd.read_csv(testing_additive_file.name)
    training_dominance = pd.read_csv(training_dominance_file.name)
    testing_dominance = pd.read_csv(testing_dominance_file.name)

    # Call the cross-validation function
    results, predicted_phenotypes = NestedKFoldCrossValidation(
        training_data=training_data,
        training_additive=training_additive,
        testing_data=testing_data,
        testing_additive=testing_additive,
        training_dominance=training_dominance,
        testing_dominance=testing_dominance,
        epochs=epochs,
        batch_size=batch_size,
        outer_n_splits=outer_n_splits,
        learning_rate=learning_rate,
        min_child_weight=min_child_weight,
        feature_selection=feature_selection,
        kernel='rbf',
        C=1.0, 
        epsilon=0.1, 
        gamma='scale'
        
    )

    # Save outputs
    #results_file = "cross_validation_results.csv"
    predicted_file = "predicted_phenotype.csv"
    #results.to_csv(results_file, index=False)
    if predicted_phenotypes is not None:
        predicted_phenotypes.to_csv(predicted_file, index=False)

    # Generate visualization of TOPSIS scores
    topsis_plot = visualize_topsis_scores(results)

    return  predicted_file, topsis_plot

# Gradio interface
with gr.Blocks() as interface:
    gr.Markdown("# DeepMap - An Integrated GUI for Genotype to Phenotype Prediction")

    with gr.Row():
        training_file = gr.File(label="Upload Training Data (CSV)")
        training_additive_file = gr.File(label="Upload Training Additive Data (CSV)")
        training_dominance_file = gr.File(label="Upload Training Dominance Data (CSV)")

    with gr.Row():
        testing_file = gr.File(label="Upload Testing Data (CSV)")
        testing_additive_file = gr.File(label="Upload Testing Additive Data (CSV)")
        testing_dominance_file = gr.File(label="Upload Testing Dominance Data (CSV)")

    with gr.Row():
        feature_selection = gr.Checkbox(label="Enable Feature Selection", value=True)

    #output1 = gr.File(label="Cross-Validation Results (CSV)")
    output2 = gr.File(label="Predicted Phenotypes (CSV)")
    output3 = gr.Image(label="TOPSIS Score Visualization")

    submit_btn = gr.Button("Run DeepMap")
    submit_btn.click(
        run_cross_validation,
        inputs=[
            training_file, training_additive_file, testing_file, 
            testing_additive_file, training_dominance_file, testing_dominance_file, 
            feature_selection
        ],
        outputs=[output2, output3]
    )

# Launch the interface
interface.launch()