Update app.py

app.py

@@ -1,6 +1,17 @@
+# -*- coding: utf-8 -*-
+"""
+Created on Fri Jan 31 13:24:37 2025
 
+@author: Ashmitha
+"""
 
-#---------------------------------------------Libraries--------------------------
+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
 import pandas as pd
 import numpy as np
 import gradio as gr
@@ -25,272 +36,113 @@ from sklearn.feature_selection import SelectFromModel
 import tempfile
 import matplotlib.pyplot as plt
 import seaborn as sns
-#------------------------------------------GRUModel-------------------------------------
-
 
-def GRUModel(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
+# 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=100, batch_size=64, learning_rate=0.0001, dropout_rate=0.3):
+
+    # Feature Scaling
     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
-
-#def GRUModel(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, feature_selection=True, top_k=10):
-    
-    
+    # 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 input data
-   # scaler = MinMaxScaler()
-   #trainX_scaled = scaler.fit_transform(trainX)
-   # testX_scaled = scaler.transform(testX) if testX is not None else None
+    # Transformer Blocks
+    for _ in range(3):  
+        x = TransformerBlock(embed_dim, num_heads, ff_dim, dropout_rate)(x)
 
-    # Scale the target variable
-    #target_scaler = MinMaxScaler()
-    #trainy_scaled = target_scaler.fit_transform(trainy.reshape(-1, 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
 
-    # Reshape inputs to (samples, timesteps, features)
-    #trainX = trainX_scaled.reshape((trainX.shape[0], 1, trainX.shape[1]))
-    #if testX is not None:
-       # testX = testX_scaled.reshape((testX.shape[0], 1, testX.shape[1]))
-
-    # Model definition
-    #model = Sequential()
-    #model.add(GRU(512, input_shape=(trainX.shape[1], trainX.shape[2]), return_sequences=False, 
-                  #kernel_regularizer=regularizers.l1_l2(l1=l1_reg, l2=l2_reg)))
-    
-    
-    #model.add(Dense(512, 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(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))
-    
-    #model.add(Dense(1, activation="relu"))  # Output layer
-
-    #model.compile(loss='mse', optimizer=Adam(learning_rate=learning_rate), metrics=['mse'])
+    model = Model(inputs, outputs)
+    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)
-    #]
+    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)
 
-    # Train model
-    #history = model.fit(trainX, trainy_scaled, epochs=epochs, batch_size=batch_size, validation_split=0.1, verbose=1, callbacks=callbacks)
+    # 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)
 
     # Predictions
-    #predicted_train = model.predict(trainX).flatten()
-   # predicted_test = model.predict(testX).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'))
+    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
 
-    # 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())
@@ -310,7 +162,6 @@ 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',
@@ -353,10 +204,8 @@ def NestedKFoldCrossValidation(training_data, training_additive, testing_data, t
         return mse, rmse, r2, corr
 
     models = [
-        ('GRUModel', GRUModel),
-        ('CNNModel', CNNModel),
-        ('RFModel', RFModel),
-        ('XGBoostModel', XGBoostModel)
+        
+        ('TransformerModel', TransformerModel)
     ]
 
     for outer_fold, (outer_train_index, outer_test_index) in enumerate(outer_kf.split(phenotypic_info), 1):
@@ -383,10 +232,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 ['GRUModel', 'CNNModel']:
+            if model_name in ['TransformerModel' ]:
                 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)
 
@@ -547,5 +396,4 @@ with gr.Blocks() as interface:
     )
 
 # Launch the interface
-interface.launch()
-
+interface.launch()