Update app.py

app.py

@@ -1,17 +1,10 @@
 # -*- coding: utf-8 -*-
 """
-Created on Fri Jan 31 13:24:37 2025
+Created on Fri Jan 31 14:12:26 2025
 
 @author: Ashmitha
 """
-
-import tensorflow as tf
-from tensorflow.keras.layers import Input, Dense, Dropout, LayerNormalization
-from tensorflow.keras.optimizers import Adam
-from tensorflow.keras.models import Model
-from tensorflow.keras.callbacks import ReduceLROnPlateau, EarlyStopping
-from sklearn.preprocessing import MinMaxScaler
-import pandas as pd
+#-------------------------------------Libraries-------------------------
 import pandas as pd
 import numpy as np
 import gradio as gr
@@ -36,123 +29,195 @@ from sklearn.feature_selection import SelectFromModel
 import tempfile
 import matplotlib.pyplot as plt
 import seaborn as sns
+#--------------------------------------------------------------------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):
 
-import os
-import tempfile
-
-# Set a new temp directory inside /home/user
-os.environ["GRADIO_CACHE"] = "/home/user/tmp"
-tempfile.tempdir = "/home/user/tmp"
-
-# Ensure the directory exists
-os.makedirs(tempfile.tempdir, exist_ok=True)
-
-# Positional Encoding Function
-def positional_encoding(seq_len, d_model):
-    pos = tf.range(seq_len, dtype=tf.float32)[:, tf.newaxis]
-    div_term = tf.exp(tf.range(0, d_model, 2, dtype=tf.float32) * (-tf.math.log(10000.0) / d_model))
-    pos_encoding = tf.concat([tf.sin(pos * div_term), tf.cos(pos * div_term)], axis=-1)
-    return pos_encoding[tf.newaxis, ...]
-
-# Multi-Head Self-Attention Layer
-class MultiHeadSelfAttention(tf.keras.layers.Layer):
-    def __init__(self, embed_dim, num_heads):
-        super().__init__()
-        self.num_heads = num_heads
-        self.embed_dim = embed_dim
-        assert embed_dim % num_heads == 0, "Embedding dimension must be divisible by number of heads"
-
-        self.depth = embed_dim // num_heads
-        self.wq = Dense(embed_dim)
-        self.wk = Dense(embed_dim)
-        self.wv = Dense(embed_dim)
-        self.dense = Dense(embed_dim)
-
-    def split_heads(self, x, batch_size):
-        x = tf.reshape(x, (batch_size, -1, self.num_heads, self.depth))
-        return tf.transpose(x, perm=[0, 2, 1, 3])  # (batch_size, num_heads, seq_length, depth)
-
-    def call(self, inputs):
-        batch_size = tf.shape(inputs)[0]
-        q = self.split_heads(self.wq(inputs), batch_size)
-        k = self.split_heads(self.wk(inputs), batch_size)
-        v = self.split_heads(self.wv(inputs), batch_size)
-
-        attention_scores = tf.matmul(q, k, transpose_b=True) / tf.math.sqrt(float(self.depth))
-        attention_weights = tf.nn.softmax(attention_scores, axis=-1)
-        attention_output = tf.matmul(attention_weights, v)
-
-        attention_output = tf.transpose(attention_output, perm=[0, 2, 1, 3])  
-        concat_attention = tf.reshape(attention_output, (batch_size, -1, self.embed_dim))  
-        output = self.dense(concat_attention)
-        return output
-
-# Transformer Block
-class TransformerBlock(tf.keras.layers.Layer):
-    def __init__(self, embed_dim, num_heads, ff_dim, dropout_rate=0.1):
-        super().__init__()
-        self.att = MultiHeadSelfAttention(embed_dim, num_heads)
-        self.norm1 = LayerNormalization(epsilon=1e-6)
-        self.norm2 = LayerNormalization(epsilon=1e-6)
-        self.ffn = tf.keras.Sequential([
-            Dense(ff_dim, activation="relu"),
-            Dense(embed_dim),
-        ])
-        self.dropout1 = Dropout(dropout_rate)
-        self.dropout2 = Dropout(dropout_rate)
-
-    def call(self, inputs, training):
-        attn_output = self.att(inputs)
-        attn_output = self.dropout1(attn_output, training=training)
-        out1 = self.norm1(inputs + attn_output)
-
-        ffn_output = self.ffn(out1)
-        ffn_output = self.dropout2(ffn_output, training=training)
-        return self.norm2(out1 + ffn_output)
-
-# Transformer Model
-def TransformerModel(trainX, trainy, testX, testy, embed_dim=128, num_heads=8, ff_dim=256, 
-                      epochs=1, batch_size=64, learning_rate=0.0001, dropout_rate=0.3):
-
-    # Feature Scaling
+    # Scale the input data
     scaler = MinMaxScaler()
     trainX_scaled = scaler.fit_transform(trainX)
     testX_scaled = scaler.transform(testX) if testX is not None else None
 
-    # Ensure correct input shape
-    seq_len = trainX.shape[1]
-    
-    # Define Model
-    inputs = Input(shape=(seq_len, 1))  # Input reshaped to (batch, seq_len, 1)
-    x = Dense(embed_dim)(inputs)  # Feature projection
-    pos_encoding = positional_encoding(seq_len, embed_dim)
-    x += tf.broadcast_to(pos_encoding, tf.shape(x))  # Ensure shape compatibility
+    # 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))
 
-    # Transformer Blocks
-    for _ in range(3):  
-        x = TransformerBlock(embed_dim, num_heads, ff_dim, dropout_rate)(x)
+    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))
 
-    x = Dense(64, activation="relu")(x)
-    x = Dropout(dropout_rate)(x)
-    outputs = Dense(1, activation="linear")(tf.reduce_mean(x, axis=1))  # Reduce along sequence length
+    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 = Model(inputs, outputs)
-    model.compile(loss="mse", optimizer=Adam(learning_rate=learning_rate), metrics=["mse"])
+    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
-    lr_reduction = ReduceLROnPlateau(monitor="val_loss", patience=5, factor=0.5, min_lr=1e-6, verbose=1)
-    early_stopping = EarlyStopping(monitor="val_loss", patience=10, restore_best_weights=True, verbose=1)
+    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[..., np.newaxis], trainy, validation_split=0.1, 
-                        epochs=epochs, batch_size=batch_size, callbacks=[lr_reduction, early_stopping], verbose=1)
+    # 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[..., np.newaxis]).flatten()
-    predicted_test = model.predict(testX_scaled[..., np.newaxis]).flatten() if testX is not None else None
+    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-------------------------------------------
+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
+#------------------------------------------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
+#-------------------------------------------------XGBoost--------------------------------------------
+def XGBoostModel(trainX, trainy, testX, testy,learning_rate,min_child_weight,feature_selection=True, n_estimators=100, max_depth=None):
+    
+    
+    
+    # Scale the features
+    scaler = MinMaxScaler()
+    trainX_scaled = scaler.fit_transform(trainX)
+    if testX is not None:
+        testX_scaled = scaler.transform(testX)
+
+    
+    xgb_model=XGBRegressor(objective="reg:squarederror",random_state=42)
+    history=xgb_model.fit(trainX, trainy)
+    #param_grid={
+        #"learning_rate":0.01,
+        #"max_depth" : 10,
+         #"n_estimators": 100,
+         #"min_child_weight": 10
+       # }
+    
+
+    # Predictions
+    predicted_train = xgb_model.predict(trainX_scaled)
+    predicted_test = xgb_model.predict(testX_scaled) if testX is not None else None
+    
+
+    return predicted_train, predicted_test,history
+#------------------------------------------------------------------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())
@@ -172,6 +237,7 @@ def calculate_topsis_score(df):
     df['TOPSIS_Score'] = topsis_score
 
     return df
+#_-------------------------------------------------------------NestedKFold Cross Validation---------------------
 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, output_file='cross_validation_results.csv',
@@ -214,8 +280,10 @@ def NestedKFoldCrossValidation(training_data, training_additive, testing_data, t
         return mse, rmse, r2, corr
 
     models = [
-        
-        ('TransformerModel', TransformerModel)
+        ('FNNModel', FNNModel),
+        ('CNNModel', CNNModel),
+        ('RFModel', RFModel),
+        ('XGBoostModel', XGBoostModel)
     ]
 
     for outer_fold, (outer_train_index, outer_test_index) in enumerate(outer_kf.split(phenotypic_info), 1):
@@ -242,10 +310,10 @@ def NestedKFoldCrossValidation(training_data, training_additive, testing_data, t
 
         for model_name, model_func in models:
             print(f"Running model: {model_name} for fold {outer_fold}")
-            if model_name in ['TransformerModel' ]:
+            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 ['RFModel']:
+                predicted_train, predicted_test, history = model_func(outer_trainX, outer_trainy, outer_testX, outer_testy)
             else:
                 predicted_train, predicted_test, history = model_func(outer_trainX, outer_trainy, outer_testX, outer_testy, learning_rate, min_child_weight)
 
@@ -333,7 +401,7 @@ def run_cross_validation(training_file, training_additive_file, testing_file, te
                          training_dominance_file, testing_dominance_file, feature_selection, learning_rate, min_child_weight):
 
     # Default parameters
-    epochs = 1
+    epochs = 1000
     batch_size = 64
     outer_n_splits = 2