# -*- 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()