app.py

import os
import torch
import shutil
import pickle
import zipfile
import matplotlib
import numpy as np
import gradio as gr
import pandas as pd
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from matplotlib import pyplot as plt
from sklearn.impute import SimpleImputer
from pandas.plotting import andrews_curves
from sklearn.metrics import accuracy_score, confusion_matrix, ConfusionMatrixDisplay, precision_score, recall_score, f1_score
from sklearn.preprocessing import StandardScaler
from torch.utils.data import DataLoader, TensorDataset, Dataset

###################################################### Preprocessing #####################################################################
def preprocess_dataframe(df, target_column=None, fill_method='mean', drop_na=True, sequence_length=32, test_size=0.2, batch_size = 128):
    """

    Loads a DataFrame from a file, preprocesses it, prepares it for LSTM data.

    If a target_column is provided, that column is used as the target (y).

    Otherwise, it prepares data for an autoencoder (no separate y).

    1. Loads file and checks for the target columns

    2. Drops any NaN rows and non numeric columns.

    3. Fills the NaN values with given method.

    4. After preprocessing, data is transformed to fit in lstm.



    Args:

        file_path (str): Path to the data file (e.g., CSV, Excel).

        target_column (str, optional): Name of the target column. If provided, use this as target. Otherwise, treats as autoencoder.  Defaults to None.

        fill_method (str, optional): Method for filling NaNs: 'mean', 'median', 'most_frequent', or 'constant'.

                                       Defaults to 'mean'. If 'constant', `fill_value` must be set.

        drop_na (bool, optional): Whether to drop rows with any NaN values. Defaults to True.

        sequence_length (int): The length of the sequence to create (e.g., number of features to treat as a sequence).

        test_size (float): The proportion of data to use for testing.



    Returns:

        tuple: (train_loader, test_loader, input_size) if no target_column.

               (train_loader, test_loader, input_size, target_column_name) if target_column provided

        A tuple containing:

            - train_loader (DataLoader): DataLoader for training data.

            - test_loader (DataLoader): DataLoader for test data.

            - input_size (int): Number of features.

            - target_column_name (str): The name of the target column only when there is target column.

    """
    # 1. Target Column Check
    target_col = None
    if target_column:
        if target_column in df.columns:
            target_col = target_column
            print(f"Target column '{target_column}' found.")
        else:
            target_column = None  # Reset target_column so we treat as autoencoder
    else:
        print("No target column specified. Treating as autoencoder.")

    #2. Drop Rows with NaNs before Fill
    if drop_na:
        print("Dropping rows with any NaN values...")
        df = df.dropna()


    # 3. Drop Non-Numeric Columns (Except Target)
    columns_to_drop = []
    for col in df.columns:
        if col != target_col and not pd.api.types.is_numeric_dtype(df[col]):
             columns_to_drop.append(col) #exclude the target column if target column is not numeric
    if columns_to_drop:
        print(f"Dropping non-numeric columns: {columns_to_drop}")
        df = df.drop(columns=columns_to_drop)
    else:
        print("No non-numeric columns found.")


    # 4. Handle Missing Values (Only in Numeric Columns After Dropping)
    numeric_cols = df.select_dtypes(include=np.number).columns #select numeric columns after non-numeric columsn removed
    if df[numeric_cols].isnull().any().any():  # Check if any NaN values exist (in numeric columns)
        print("Handling missing values...")
        if fill_method in ['mean', 'median', 'most_frequent', 'constant']:
            imputer = SimpleImputer(strategy=fill_method)

            if fill_method == 'constant':
                 imputer = SimpleImputer(strategy=fill_method, fill_value=0) #only with constant filling value must be provided

            df[numeric_cols] = imputer.fit_transform(df[numeric_cols])  # Apply only to numeric columns

        else:
            raise ValueError("Invalid fill_method. Choose 'mean', 'median', 'most_frequent', or 'constant'.")

    # Droping NaN and inf
    df.replace([np.inf, -np.inf], np.nan, inplace=True)
    df.dropna(inplace=True)

    if target_col:
        inputdf = df.drop(columns=[target_col])
        outputdf = df[target_col].apply(lambda x: 0 if x.lower() == 'benign' else 1)
        malinputdf = inputdf[outputdf == 1]
        beninputdf = inputdf[outputdf == 0]
        sample_size = min(len(beninputdf), len(malinputdf), 500)
        bensample = beninputdf.sample(n=sample_size, random_state=42)
        bensample['Label'] = 'Benign'
        malsample = malinputdf.sample(n=sample_size, random_state=42)
        malsample['Label'] = 'Malicious'
        sample = pd.concat([bensample, malsample])
        data = beninputdf.values
    else:
        inputdf = df
        sample_size = min(len(inputdf), 500)
        sample = df.sample(n=sample_size, random_state=42)
        data = inputdf.values

    scaler = StandardScaler()
    data = scaler.fit_transform(data)

    if target_col:
      X_train = data
      data = malinputdf.values
      data = scaler.transform(data)
      X_test = data
    else:
      X_train = data

    class TabularDatasetTest(Dataset):
      def __init__(self, data):
          self.data = data.clone().detach()

      def __len__(self):
          return len(self.data)

      def __getitem__(self, idx):
          return self.data[idx], self.data[idx]

    class TabularDatasetTrain(Dataset):
      def __init__(self, data, sequence_length):
          self.data = data.clone().detach()
          self.sequence_length = sequence_length

      def __len__(self):
          return len(self.data) - self.sequence_length + 1

      def __getitem__(self, idx):
          return self.data[idx:idx + self.sequence_length], self.data[idx:idx + self.sequence_length]

    if target_column:
        X_train = torch.tensor(X_train, dtype=torch.float32)
        X_test = torch.tensor(X_test, dtype=torch.float32)
        train_dataset = TabularDatasetTrain(X_train, sequence_length = sequence_length)
        test_dataset = TabularDatasetTest(X_test)
        train_DataLoader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
        test_Dataloader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False)
        return {
            'train_loader': train_DataLoader,
            'test_loader': test_Dataloader,
            'input_df': inputdf,
            'target_df': outputdf,
            'malinput_df': malinputdf,
            'beninput_df': beninputdf,
            'target_col': target_col,
            'scaler': scaler,
            'sample': sample
        }
    else:
        X_train = torch.tensor(X_train, dtype=torch.float32)
        train_dataset = TabularDatasetTrain(X_train, sequence_length = sequence_length)
        train_DataLoader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
        sample["Label"] = "dummy_class"
        return {
            'train_loader': train_DataLoader,
            'test_loader': None,
            'input_df': inputdf,
            'malinput_df': None,
            'beninput_df': None,
            'target_df': None,
            'target_col': None,
            'scaler': scaler,
            'sample': sample
        }
################################################## Model #############################################################################
class EncoderRNN(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers, isCuda):
        super(EncoderRNN, self).__init__()
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        self.bottleneck_size = int(input_size/2)

        self.isCuda = isCuda
        self.lstm1 = nn.LSTM(input_size, int(hidden_size/2), num_layers, batch_first=True, bidirectional = True)
        self.relu = nn.ReLU()
        self.dropout = nn.Dropout(0.2)
        self.lstm2 = nn.LSTM(hidden_size, self.bottleneck_size, num_layers, batch_first=True)


    def forward(self, inputs):
        intermediate_state, hidden = self.lstm1(inputs)#, (h0_1, c0_1))
        intermediate_state = self.relu(self.dropout(intermediate_state))
        encoded_input, hidden = self.lstm2(intermediate_state)#, (h0_2, c0_2))
        return encoded_input, intermediate_state

class DecoderRNN(nn.Module):
    def __init__(self, hidden_size, output_size, num_layers, isCuda):
        super(DecoderRNN, self).__init__()
        self.hidden_size = hidden_size
        self.output_size = output_size
        self.num_layers = num_layers
        self.bottleneck_size = int(output_size/2)

        self.isCuda = isCuda
        self.lstm2 = nn.LSTM(self.bottleneck_size, hidden_size, num_layers, batch_first=True)
        self.relu = nn.ReLU()
        self.dropout = nn.Dropout(0.2)
        self.lstm1 = nn.LSTM(2*hidden_size, output_size, num_layers, batch_first=True)

    def forward(self, encoded_input, intermediate_state):
        encoded_input, hidden = self.lstm2(encoded_input)#, (h0_2, c0_2))
        inputs = torch.cat((self.dropout(encoded_input), intermediate_state), dim=2)
        inputs = self.relu(inputs)
        decoded_output, hidden = self.lstm1(inputs)#, (h0_1, c0_1))
        # print(f"output: {decoded_output}")
        return decoded_output

class LSTMAE(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers=1, isCuda="cuda" if torch.cuda.is_available() else "cpu"):
        super(LSTMAE, self).__init__()
        hidden_size = hidden_size if hidden_size%2==0 else hidden_size+1
        self.encoder = EncoderRNN(input_size, hidden_size, num_layers, isCuda)
        self.decoder = DecoderRNN(hidden_size, input_size, num_layers, isCuda)
        self.initialize_weights()

    def initialize_weights(self):
        """

        Initializes the weights of the linear, LSTM, and convolutional layers

        using appropriate initialization schemes.

        """
        for m in self.modules():  # Iterate through all modules in the network
            if isinstance(m, nn.LSTM):
                for name, param in m.named_parameters():
                    if 'weight' in name:
                        if 'ih' or 'hh' in name:
                            nn.init.xavier_uniform_(param.data)  # Input-to-hidden
                    elif 'bias' in name:
                        nn.init.zeros_(param.data)

    def forward(self, input):
        encoded_input, intermediate_state = self.encoder(input)
        decoded_output = self.decoder(encoded_input, intermediate_state)
        return decoded_output



############################################## Andrews Curves ###########################################################################
def make_better_andrews_curves(df, class_column, colors=None, plot_title="Andrews Curves",

                               line_width=0.8, transparency=0.5,  sample_size=None, legend_loc='best',

                               custom_labels=None,  x_axis_ticks=None, x_axis_labels=None,

                               figsize=(10, 6), dpi=300, name = "andrews_curves"):
    """

    Generates an Andrews Curves plot with enhanced styling.



    Args:

        df: pandas DataFrame containing the data.

        class_column: Name of the column containing class labels.

        colors: List of colors to use for each class (e.g., ['blue', 'red']).  Defaults to matplotlib's defaults if None.

        plot_title: Title of the plot.

        line_width: Width of the lines.

        transparency: Alpha value (transparency) of the lines.

        sample_size: If an integer is provided, a random sample of the data will be used.  Useful for large datasets.

        legend_loc: Location of the legend (e.g., 'best', 'upper right', 'lower left').

        custom_labels: A dictionary mapping original class labels to more descriptive labels for the legend.

        x_axis_ticks: A list of tick positions for the x-axis. If None, default ticks are used.

        x_axis_labels: A list of labels for the x-axis ticks. Must be the same length as x_axis_ticks.

        figsize: Tuple specifying the figure size (width, height) in inches.

    """

    if sample_size and sample_size < len(df):
        df = df.sample(n=sample_size, random_state=42)  # Sample for faster plotting

    plt.figure(figsize=figsize)  # Set the figure size before plotting

    ax = andrews_curves(df, class_column, color=colors)  # Store the Axes object

    plt.title(plot_title, fontsize=16)
    plt.xlabel("t", fontsize=12)  # Added x-axis label
    plt.ylabel("f(t)", fontsize=12) # Added y-axis label

    for line in ax.get_lines():
        line.set_linewidth(line_width)
        line.set_alpha(transparency)

    # Customize Legend
    if custom_labels:
        handles, labels = ax.get_legend_handles_labels()
        new_labels = [custom_labels.get(label, label) for label in labels] # Use .get() to handle missing labels
        ax.legend(handles, new_labels, loc=legend_loc, fontsize=10)
    else:
        plt.legend(loc=legend_loc, fontsize=10)


    # Customize X-axis ticks and labels
    if x_axis_ticks:
        plt.xticks(x_axis_ticks, x_axis_labels)

    plt.grid(False)  # Add a grid
    plt.tight_layout()  # Adjust layout to prevent labels from overlapping
    plt.savefig(f"{name}.png", dpi=dpi)
################################################# Model Training ######################################################################
def train_model(model, train_loader, test_loader = None, learning_rate=0.001, epochs=10):
    criterion = nn.MSELoss()
    info = ""
    optimizer = optim.Adam(model.parameters(), lr=learning_rate)
    train_loss_data = {}
    for epoch in range(epochs):
        model.train()
        train_loss = 0.0
        epoch_train_losses = []
        mse_losses = []
        for i,(inputs, targets) in enumerate(train_loader):
            inputs = inputs.to(device)
            targets = targets.to(device)
            outputs = model(inputs)
            # l1_lambda = 0.001
            # l2_lambda = 0.0001
            # l1_norm = sum(p.abs().sum() for p in model.parameters())  # L1 norm
            # l2_norm = sum(p.pow(2.0).sum() for p in model.parameters())  # L2 norm
            loss = criterion(outputs, targets)# + l2_lambda * l2_norm + l1_lambda * l1_norm
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            if epoch == epochs-1:
              mse_loss = F.mse_loss(targets, outputs, reduction='none')
              mse_loss_per_data_point = mse_loss.mean(dim=-1)
              mse_losses.extend(mse_loss_per_data_point.tolist())
            epoch_train_losses.append(loss.item())
            train_loss += loss.item()
        train_loss /= len(train_loader)

        # Validation
        if test_loader and epoch%1==0:
          model.eval()
          test_loss = 0.0
          with torch.no_grad():
              for i,(inputs, targets) in enumerate(test_loader):
                  inputs = inputs.to(device)
                  targets = targets.to(device)
                  outputs = model(inputs.unsqueeze(1))
                  loss = criterion(outputs.squeeze(1), targets)
                  test_loss += loss.item()

          test_loss /= len(test_loader)
        else:
          test_loss = 0.0
        info += f"Epoch {epoch+1}/{epochs}, Train Loss: {train_loss:.4f}, Test Loss: {test_loss:.4f}\n"
        print(f"Epoch {epoch+1}/{epochs}, Train Loss: {train_loss:.4f}, Test Loss: {test_loss:.4f}")
        train_loss_data[f'Epoch {epoch + 1}'] = epoch_train_losses
    train_loss_df = pd.DataFrame(dict([(k,pd.Series(v)) for k,v in train_loss_data.items()]))
    return model, train_loss_df, mse_losses, info

#########################################################################################################################################
def detect_anomalies(csv_file, sample_choice="Custom Data", data_slicing_percentage=80, epochs=3, threshold_factor=1.0):
  images = []
  anomaly_summary = ""
  device = "cuda" if torch.cuda.is_available() else "cpu"
  if os.path.exists("Results"):
    shutil.rmtree("Results")
  os.mkdir("Results")
  if sample_choice == "Custom Data":
    anomaly_summary += f"[INFO] Loading Custom Dataset {data_slicing_percentage}%...\n"
    dataframe = pd.read_csv(csv_file.name).sample(frac=data_slicing_percentage/100, random_state=42).reset_index(drop=True)
    anomaly_summary += f"[INFO] Preprocessing Dataset...\n"
    if dataframe.get('Label') is not None:
        processed_data = preprocess_dataframe(dataframe, target_column="Label")
    else:
        processed_data = preprocess_dataframe(dataframe)
        anomaly_summary += f"[WARNING] No Label Column Found, Using Unsupervised Learning...\n"
    anomaly_summary += f"[INFO] Generating Andrews Curves...\n"
    make_better_andrews_curves(processed_data['sample'], 'Label',
                              colors=['Blue', 'Red'],
                              plot_title="Dataset Andrews Curves",
                              line_width=1.2,
                              transparency=0.7,
                              legend_loc='upper right',
                              figsize=(12, 7),
                              name = "Results/Dataset_andrews_curves")
    images.append("Results/Dataset_andrews_curves.png")
    model = LSTMAE(len(processed_data["input_df"].columns),128).to(device)
    model.to(device)
    anomaly_summary += f"[INFO] Training Model...\n"
    _, train_loss_df, mse_losses, info = train_model(model, processed_data['train_loader'], processed_data['test_loader'],epochs=epochs)
    anomaly_summary += info
    anomaly_summary += f"[INFO] Saving model, scaler, Dataset Used...\n"
    dataframe.to_csv('Results/Original_dataset.csv', columns=dataframe.columns, index=False)
    pickle.dump(processed_data['scaler'], open('Results/scaler.pkl', 'wb'))
    torch.save(model, 'Results/model.pth')
    anomaly_summary += f"[INFO] Generating Loss Curves...\n"
    plt.figure(figsize=(12, 6))  # Adjust figure size as needed
    for column in train_loss_df.columns:
        plt.plot(train_loss_df[column], label=column)
    plt.xlabel("Batch")
    plt.ylabel("Loss")
    plt.title("Training Loss per Epoch")
    plt.legend()  # Show the legend to identify each epoch
    plt.grid(True)  # Add a grid for easier reading
    plt.tight_layout()  # Adjust layout to prevent labels from overlapping
    plt.savefig("Results/loss_curves.png", dpi=300)
    images.append("Results/loss_curves.png")
    Q1, Q3 = np.percentile(mse_losses, [25, 75])
    Dict = {"Q1": Q1, "Q3": Q3}
    pickle.dump(Dict, open('Results/INFO.pkl', 'wb'))

  else:
    Q1, Q3 = 0.19226229563355446, 0.7454282641410828
    IQR = Q3 - Q1
    lower_bound = Q1 - threshold_factor * IQR
    upper_bound = Q3 + threshold_factor * IQR
    # print(lower_bound, upper_bound)
    data_path = os.path.join(os.path.abspath('Data'),sample_choice)
    dataframe = pd.read_csv(data_path).sample(frac=data_slicing_percentage/100, random_state=42).reset_index(drop=True)
    anomaly_summary += f"[INFO] Saving model, scaler, Dataset Used...\n"
    dataframe.to_csv('Results/Scaled_dataset.csv', columns=dataframe.columns, index=False)
    scaler = pickle.load(open('scaler.pkl', 'rb'))
    original_df = scaler.inverse_transform(dataframe.iloc[:,:-1])
    original_df = pd.DataFrame(original_df, columns=dataframe.columns[:-1])
    original_df['Label'] = dataframe['Label']
    original_df.to_csv('Results/Original_dataset.csv', columns=dataframe.columns, index=False)
    shutil.copy('scaler.pkl', 'Results/scaler.pkl')
    shutil.copy('model.pth', 'Results/model.pth')
    # andrew curve of dataset
    anomaly_summary += f"[INFO] Generating Andrews Curves...\n"
    make_better_andrews_curves(dataframe, 'Label',
                               colors=['Blue', 'Red'],
                               plot_title="Dataset Andrews Curves",
                               line_width=1.2,
                               transparency=0.7,
                               legend_loc='upper right',
                               figsize=(12, 7),
                               name = "Results/Dataset_andrews_curves")
    images.append("Results/Dataset_andrews_curves.png")
    inputdf = torch.tensor(dataframe.iloc[:,:-1].to_numpy(), dtype=torch.float32, device=device)
    outputdf = dataframe['Label']
    model = torch.load("model.pth",weights_only = False, map_location=device)
    model.eval()
    outputs = model(inputdf.unsqueeze(1)).squeeze(1)
    mse_loss = F.mse_loss(outputs, inputdf, reduction='none')
    mse_loss_per_data_point = mse_loss.mean(dim=-1)
    anomaly_scores = pd.DataFrame({'Loss': mse_loss_per_data_point.detach().cpu().numpy(), 'Label': outputdf})
    anomaly_scores['Anomaly'] = anomaly_scores['Loss'].apply(lambda x: 1 if x > upper_bound else 0)
    anomaly_scores['Label'] = anomaly_scores['Label'].apply(lambda x: 1 if x == "Malicious" else 0)
    out_confusion_matrix = confusion_matrix(anomaly_scores['Label'], anomaly_scores['Anomaly'])
    disp = ConfusionMatrixDisplay(confusion_matrix=out_confusion_matrix, display_labels=["Benign","Malignant"])
    disp.plot(cmap=plt.cm.Blues)
    plt.title('Confusion Matrix')
    plt.savefig(f"Results/confusion_matrix.png", dpi=300)
    images.append("Results/confusion_matrix.png")
    accuracy = accuracy_score(anomaly_scores['Label'], anomaly_scores['Anomaly'])
    precision = precision_score(anomaly_scores['Label'], anomaly_scores['Anomaly'])
    recall = recall_score(anomaly_scores['Label'], anomaly_scores['Anomaly'])
    f1 = f1_score(anomaly_scores['Label'], anomaly_scores['Anomaly'])
    # print(f"Accuracy: {accuracy:.4f}, Precision: {precision:.4f}, Recall: {recall:.4f}, F1: {f1:.4f}")
    anomaly_summary += f"[RESULT] Accuracy: {accuracy:.4f}, Precision: {precision:.4f}, Recall: {recall:.4f}, F1: {f1:.4f}"
    anomaly_summary = anomaly_summary + f"Confusion Matrix:\n{out_confusion_matrix}\n"

  folder_path = "Results"
  with zipfile.ZipFile("Results.zip", 'w', zipfile.ZIP_DEFLATED) as zipf:
    for root, _, files in os.walk(folder_path):
        for file in files:
            file_path = os.path.join(root, file)
            relative_path = os.path.relpath(file_path, folder_path)
            zipf.write(file_path, relative_path)

  return anomaly_summary, images, "Results.zip"

iface = gr.Interface(
    fn=detect_anomalies,
    inputs=[
        gr.File(file_types=[".csv"], label="Upload CSV File"),
        gr.Radio(["Benign500.csv", "Malignant500.csv", "Balance1000.csv", "Custom Data"], value="Custom Data", label="Choose Samples or CustomData"),
        gr.Slider(minimum=10, maximum=100, step=10, value=80, label="Data Usage Percentage (Training or Detection)"),
        gr.Slider(minimum=1, maximum=20, step=1, value=3, label="Training Epochs (Default value is 3)"),
        gr.Slider(minimum=0, maximum=5, step=0.5, value=1.5, label="Loss Threshold (x, higher x means high threshold) = Q3 + x*IQR"),
    ],
    outputs=[
        gr.Textbox(label="Anomaly Summary"),
        gr.Gallery(label="Anomaly Plots"),
        "file",
    ],
    title="Your own Anomaly Detector",
    description="""

    ### Fully Unsupervised Anomaly Detection Tool (uses Bidirectional based Autoencoder with skip conn. and Dropout Layers)

    ##### Download *"Result.zip"* (contains model.pkl, dataset images, output images) to download the results from Right Bottom.

    Upload a *CSV file* (Custom Anomalies Detection: Use Output Column: "Label" or ), or Use *our trained model*.

    """
)

if __name__ == "__main__":
    iface.launch(debug=False)