title electroencephalogram signal classification for action identification suvaditya mukherjeehttps github comsuvadityamuk date created 20221103 last modified 20221105 description training a convolutional model to classify eeg signals produced by exposure to certain stimuli accelerator gpu introduction the following example explores how we can make a convolutionbased neural network to perform classification on electroencephalogram signals captured when subjects were exposed to different stimuli we train a model from scratch since such signalclassification models are fairly scarce in pretrained format the data we use is sourced from the uc berkeleybiosense lab where the data was collected from 15 subjects at the same time our process is as follows load the uc berkeleybiosense synchronized brainwave datasethttps www kaggle comdatasetsberkeleybiosensesynchronizedbrainwavedataset visualize random samples from the data preprocess collate and scale the data to finally make a tf data dataset prepare class weights in order to tackle major imbalances create a conv1d and densebased model to perform classification define callbacks and hyperparameters train the model plot metrics from history and perform evaluation this example needs the following external dependencies gdown scikitlearn pandas numpy matplotlib you can install it via the following commands gdown is an external package used to download large files from google drive to know more you can refer to its pypi page herehttps pypi orgprojectgdown setup and data downloads first lets install our dependencies shell pip install gdown q pip install sklearn q pip install pandas q pip install numpy q pip install matplotlib q next lets download our dataset the gdown package makes it easy to download the data from google drive shell gdown 1v5b7bt6ajm0uhbr7crkbek8jx7lypvux gdown will download eegdata csv onto the local drive for use total size of eegdata csv is 105 7 mb read data from eegdata csv we use the pandas library to read the eegdata csv file and display the first 5 rows using the head command we remove unlabeled samples from our dataset as they do not contribute to the model we also perform a drop operation on the columns that are not required for training data preparation in the data the samples recorded are given a score from 0 to 128 based on how wellcalibrated the sensor was 0 being best 200 being worst we filter the values based on an arbitrary cutoff limit of 128 visualize one random sample from the data we visualize one sample from the data to understand how the stimulusinduced signal looks like preprocess and collate data there are a total of 67 different labels present in the data where there are numbered sublabels we collate them under a single label as per their numbering and replace them in the data itself following this process we perform simple label encoding to get them in an integer format we extract the number of unique classes present in the data we now visualize the number of samples present in each class using a bar plot scale and split data we perform a simple minmax scaling to bring the valuerange between 0 and 1 we do not use standard scaling as the data does not follow a gaussian distribution we now create a traintest split with a 15 holdout set following this we reshape the data to create a sequence of length 512 we also convert the labels from their current labelencoded form to a onehot encoding to enable use of several different keras metrics functions prepare tf data dataset we now create a tf data dataset from this data to prepare it for training we also shuffle and batch the data for use later make class weights using naive method as we can see from the plot of number of samples per class the dataset is imbalanced hence we calculate weights for each class to make sure that the model is trained in a fair manner without preference to any specific class due to greater number of samples we use a naive method to calculate these weights finding an inverse proportion of each class and using that as the weight formula used naive method where weight 1 no of samples present total no of samples so more the samples lower the weight define simple function to plot all the metrics present in a keras callbacks history object define function to generate convolutional model get model summary define callbacks optimizer loss and metrics we set the number of epochs at 30 after performing extensive experimentation it was seen that this was the optimal number after performing earlystopping analysis as well we define a model checkpoint callback to make sure that we only get the best model weights we also define a reducelronplateau as there were several cases found during experimentation where the loss stagnated after a certain point on the other hand a direct lrscheduler was found to be too aggressive in its decay compile model and call model fit we use the adam optimizer since it is commonly considered the best choice for preliminary training and was found to be the best optimizer we use categoricalcrossentropy as the loss as our labels are in a onehotencoded form we define the topkcategoricalaccuracyk3 auc precision and recall metrics to further aid in understanding the model better visualize model metrics during training we use the function defined above to see model metrics during training evaluate model on test data introduction the following example explores how we can make a convolution based neural network to perform classification on electroencephalogram signals captured when subjects were exposed to different stimuli we train a model from scratch since such signal classification models are fairly scarce in pre trained format the data we use is sourced from the uc berkeley biosense lab where the data was collected from 15 subjects at the same time our process is as follows load the uc berkeley biosense synchronized brainwave dataset https www kaggle com datasets berkeley biosense synchronized brainwave dataset visualize random samples from the data pre process collate and scale the data to finally make a tf data dataset prepare class weights in order to tackle major imbalances create a conv1d and dense based model to perform classification define callbacks and hyperparameters train the model plot metrics from history and perform evaluation this example needs the following external dependencies gdown scikit learn pandas numpy matplotlib you can install it via the following commands gdown is an external package used to download large files from google drive to know more you can refer to its pypi page here https pypi org project gdown setup and data downloads first lets install our dependencies shell pip install gdown q pip install sklearn q pip install pandas q pip install numpy q pip install matplotlib q next lets download our dataset the gdown package makes it easy to download the data from google drive shell gdown 1v5b7bt6ajm0uhbr7crkbek8jx7lypvux gdown will download eeg data csv onto the local drive for use total size of eeg data csv is 105 7 mb read data from eeg data csv we use the pandas library to read the eeg data csv file and display the first 5 rows using the head command we remove unlabeled samples from our dataset as they do not contribute to the model we also perform a drop operation on the columns that are not required for training data preparation in the data the samples recorded are given a score from 0 to 128 based on how well calibrated the sensor was 0 being best 200 being worst we filter the values based on an arbitrary cutoff limit of 128 visualize one random sample from the data we visualize one sample from the data to understand how the stimulus induced signal looks like pre process and collate data there are a total of 67 different labels present in the data where there are numbered sub labels we collate them under a single label as per their numbering and replace them in the data itself following this process we perform simple label encoding to get them in an integer format generates a look up table we extract the number of unique classes present in the data we now visualize the number of samples present in each class using a bar plot scale and split data we perform a simple min max scaling to bring the value range between 0 and 1 we do not use standard scaling as the data does not follow a gaussian distribution we now create a train test split with a 15 holdout set following this we reshape the data to create a sequence of length 512 we also convert the labels from their current label encoded form to a one hot encoding to enable use of several different keras metrics functions prepare tf data dataset we now create a tf data dataset from this data to prepare it for training we also shuffle and batch the data for use later make class weights using naive method as we can see from the plot of number of samples per class the dataset is imbalanced hence we calculate weights for each class to make sure that the model is trained in a fair manner without preference to any specific class due to greater number of samples we use a naive method to calculate these weights finding an inverse proportion of each class and using that as the weight formula used naive method where weight 1 no of samples present total no of samples so more the samples lower the weight define simple function to plot all the metrics present in a keras callbacks history object define function to generate convolutional model get model summary define callbacks optimizer loss and metrics we set the number of epochs at 30 after performing extensive experimentation it was seen that this was the optimal number after performing early stopping analysis as well we define a model checkpoint callback to make sure that we only get the best model weights we also define a reducelronplateau as there were several cases found during experimentation where the loss stagnated after a certain point on the other hand a direct lrscheduler was found to be too aggressive in its decay compile model and call model fit we use the adam optimizer since it is commonly considered the best choice for preliminary training and was found to be the best optimizer we use categoricalcrossentropy as the loss as our labels are in a one hot encoded form we define the topkcategoricalaccuracy k 3 auc precision and recall metrics to further aid in understanding the model better visualize model metrics during training we use the function defined above to see model metrics during training evaluate model on test data
import pandas as pd
import matplotlib.pyplot as plt
import json
import numpy as np
import keras
from keras import layers
import tensorflow as tf
from sklearn import preprocessing, model_selection
import random

QUALITY_THRESHOLD = 128
BATCH_SIZE = 64
SHUFFLE_BUFFER_SIZE = BATCH_SIZE * 2



eeg = pd.read_csv("eeg-data.csv")



unlabeled_eeg = eeg[eeg["label"] == "unlabeled"]
eeg = eeg.loc[eeg["label"] != "unlabeled"]
eeg = eeg.loc[eeg["label"] != "everyone paired"]

eeg.drop(
    [
        "indra_time",
        "Unnamed: 0",
        "browser_latency",
        "reading_time",
        "attention_esense",
        "meditation_esense",
        "updatedAt",
        "createdAt",
    ],
    axis=1,
    inplace=True,
)

eeg.reset_index(drop=True, inplace=True)
eeg.head()




def convert_string_data_to_values(value_string):
    str_list = json.loads(value_string)
    return str_list


eeg["raw_values"] = eeg["raw_values"].apply(convert_string_data_to_values)

eeg = eeg.loc[eeg["signal_quality"] < QUALITY_THRESHOLD]
eeg.head()






def view_eeg_plot(idx):
    data = eeg.loc[idx, "raw_values"]
    plt.plot(data)
    plt.title(f"Sample random plot")
    plt.show()


view_eeg_plot(7)





print("Before replacing labels")
print(eeg["label"].unique(), "\n")
print(len(eeg["label"].unique()), "\n")


eeg.replace(
    {
        "label": {
            "blink1": "blink",
            "blink2": "blink",
            "blink3": "blink",
            "blink4": "blink",
            "blink5": "blink",
            "math1": "math",
            "math2": "math",
            "math3": "math",
            "math4": "math",
            "math5": "math",
            "math6": "math",
            "math7": "math",
            "math8": "math",
            "math9": "math",
            "math10": "math",
            "math11": "math",
            "math12": "math",
            "thinkOfItems-ver1": "thinkOfItems",
            "thinkOfItems-ver2": "thinkOfItems",
            "video-ver1": "video",
            "video-ver2": "video",
            "thinkOfItemsInstruction-ver1": "thinkOfItemsInstruction",
            "thinkOfItemsInstruction-ver2": "thinkOfItemsInstruction",
            "colorRound1-1": "colorRound1",
            "colorRound1-2": "colorRound1",
            "colorRound1-3": "colorRound1",
            "colorRound1-4": "colorRound1",
            "colorRound1-5": "colorRound1",
            "colorRound1-6": "colorRound1",
            "colorRound2-1": "colorRound2",
            "colorRound2-2": "colorRound2",
            "colorRound2-3": "colorRound2",
            "colorRound2-4": "colorRound2",
            "colorRound2-5": "colorRound2",
            "colorRound2-6": "colorRound2",
            "colorRound3-1": "colorRound3",
            "colorRound3-2": "colorRound3",
            "colorRound3-3": "colorRound3",
            "colorRound3-4": "colorRound3",
            "colorRound3-5": "colorRound3",
            "colorRound3-6": "colorRound3",
            "colorRound4-1": "colorRound4",
            "colorRound4-2": "colorRound4",
            "colorRound4-3": "colorRound4",
            "colorRound4-4": "colorRound4",
            "colorRound4-5": "colorRound4",
            "colorRound4-6": "colorRound4",
            "colorRound5-1": "colorRound5",
            "colorRound5-2": "colorRound5",
            "colorRound5-3": "colorRound5",
            "colorRound5-4": "colorRound5",
            "colorRound5-5": "colorRound5",
            "colorRound5-6": "colorRound5",
            "colorInstruction1": "colorInstruction",
            "colorInstruction2": "colorInstruction",
            "readyRound1": "readyRound",
            "readyRound2": "readyRound",
            "readyRound3": "readyRound",
            "readyRound4": "readyRound",
            "readyRound5": "readyRound",
            "colorRound1": "colorRound",
            "colorRound2": "colorRound",
            "colorRound3": "colorRound",
            "colorRound4": "colorRound",
            "colorRound5": "colorRound",
        }
    },
    inplace=True,
)

print("After replacing labels")
print(eeg["label"].unique())
print(len(eeg["label"].unique()))

le = preprocessing.LabelEncoder()  
le.fit(eeg["label"])
eeg["label"] = le.transform(eeg["label"])



num_classes = len(eeg["label"].unique())
print(num_classes)



plt.bar(range(num_classes), eeg["label"].value_counts())
plt.title("Number of samples per class")
plt.show()





scaler = preprocessing.MinMaxScaler()
series_list = [
    scaler.fit_transform(np.asarray(i).reshape(-1, 1))
    for i in eeg["raw_values"]
]

labels_list = [i for i in eeg["label"]]



x_train, x_test, y_train, y_test = model_selection.train_test_split(
    series_list, labels_list, test_size=0.15, random_state=42, shuffle=True
)

print(
    f"Length of x_train : {len(x_train)}\nLength of x_test : {len(x_test)}\nLength of y_train : {len(y_train)}\nLength of y_test : {len(y_test)}"
)

x_train = np.asarray(x_train).astype(np.float32).reshape(-1, 512, 1)
y_train = np.asarray(y_train).astype(np.float32).reshape(-1, 1)
y_train = keras.utils.to_categorical(y_train)

x_test = np.asarray(x_test).astype(np.float32).reshape(-1, 512, 1)
y_test = np.asarray(y_test).astype(np.float32).reshape(-1, 1)
y_test = keras.utils.to_categorical(y_test)





train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train))
test_dataset = tf.data.Dataset.from_tensor_slices((x_test, y_test))

train_dataset = train_dataset.shuffle(SHUFFLE_BUFFER_SIZE).batch(BATCH_SIZE)
test_dataset = test_dataset.batch(BATCH_SIZE)





vals_dict = {}
for i in eeg["label"]:
    if i in vals_dict.keys():
        vals_dict[i] += 1
    else:
        vals_dict[i] = 1
total = sum(vals_dict.values())





weight_dict = {k: (1 - (v / total)) for k, v in vals_dict.items()}
print(weight_dict)




def plot_history_metrics(history: keras.callbacks.History):
    total_plots = len(history.history)
    cols = total_plots // 2

    rows = total_plots // cols

    if total_plots % cols != 0:
        rows += 1

    pos = range(1, total_plots + 1)
    plt.figure(figsize=(15, 10))
    for i, (key, value) in enumerate(history.history.items()):
        plt.subplot(rows, cols, pos[i])
        plt.plot(range(len(value)), value)
        plt.title(str(key))
    plt.show()





def create_model():
    input_layer = keras.Input(shape=(512, 1))

    x = layers.Conv1D(
        filters=32, kernel_size=3, strides=2, activation="relu", padding="same"
    )(input_layer)
    x = layers.BatchNormalization()(x)

    x = layers.Conv1D(
        filters=64, kernel_size=3, strides=2, activation="relu", padding="same"
    )(x)
    x = layers.BatchNormalization()(x)

    x = layers.Conv1D(
        filters=128, kernel_size=5, strides=2, activation="relu", padding="same"
    )(x)
    x = layers.BatchNormalization()(x)

    x = layers.Conv1D(
        filters=256, kernel_size=5, strides=2, activation="relu", padding="same"
    )(x)
    x = layers.BatchNormalization()(x)

    x = layers.Conv1D(
        filters=512, kernel_size=7, strides=2, activation="relu", padding="same"
    )(x)
    x = layers.BatchNormalization()(x)

    x = layers.Conv1D(
        filters=1024,
        kernel_size=7,
        strides=2,
        activation="relu",
        padding="same",
    )(x)
    x = layers.BatchNormalization()(x)

    x = layers.Dropout(0.2)(x)

    x = layers.Flatten()(x)

    x = layers.Dense(4096, activation="relu")(x)
    x = layers.Dropout(0.2)(x)

    x = layers.Dense(
        2048, activation="relu", kernel_regularizer=keras.regularizers.L2()
    )(x)
    x = layers.Dropout(0.2)(x)

    x = layers.Dense(
        1024, activation="relu", kernel_regularizer=keras.regularizers.L2()
    )(x)
    x = layers.Dropout(0.2)(x)
    x = layers.Dense(
        128, activation="relu", kernel_regularizer=keras.regularizers.L2()
    )(x)
    output_layer = layers.Dense(num_classes, activation="softmax")(x)

    return keras.Model(inputs=input_layer, outputs=output_layer)




conv_model = create_model()
conv_model.summary()





epochs = 30

callbacks = [
    keras.callbacks.ModelCheckpoint(
        "best_model.keras", save_best_only=True, monitor="loss"
    ),
    keras.callbacks.ReduceLROnPlateau(
        monitor="val_top_k_categorical_accuracy",
        factor=0.2,
        patience=2,
        min_lr=0.000001,
    ),
]

optimizer = keras.optimizers.Adam(amsgrad=True, learning_rate=0.001)
loss = keras.losses.CategoricalCrossentropy()





conv_model.compile(
    optimizer=optimizer,
    loss=loss,
    metrics=[
        keras.metrics.TopKCategoricalAccuracy(k=3),
        keras.metrics.AUC(),
        keras.metrics.Precision(),
        keras.metrics.Recall(),
    ],
)

conv_model_history = conv_model.fit(
    train_dataset,
    epochs=epochs,
    callbacks=callbacks,
    validation_data=test_dataset,
    class_weight=weight_dict,
)





plot_history_metrics(conv_model_history)



loss, accuracy, auc, precision, recall = conv_model.evaluate(test_dataset)
print(f"Loss : {loss}")
print(f"Top 3 Categorical Accuracy : {accuracy}")
print(f"Area under the Curve (ROC) : {auc}")
print(f"Precision : {precision}")
print(f"Recall : {recall}")


def view_evaluated_eeg_plots(model):
    start_index = random.randint(10, len(eeg))
    end_index = start_index + 11
    data = eeg.loc[start_index:end_index, "raw_values"]
    data_array = [
        scaler.fit_transform(np.asarray(i).reshape(-1, 1)) for i in data
    ]
    data_array = [np.asarray(data_array).astype(np.float32).reshape(-1, 512, 1)]
    original_labels = eeg.loc[start_index:end_index, "label"]
    predicted_labels = np.argmax(model.predict(data_array, verbose=0), axis=1)
    original_labels = [
        le.inverse_transform(np.array(label).reshape(-1))[0]
        for label in original_labels
    ]
    predicted_labels = [
        le.inverse_transform(np.array(label).reshape(-1))[0]
        for label in predicted_labels
    ]
    total_plots = 12
    cols = total_plots // 3
    rows = total_plots // cols
    if total_plots % cols != 0:
        rows += 1
    pos = range(1, total_plots + 1)
    fig = plt.figure(figsize=(20, 10))
    for i, (plot_data, og_label, pred_label) in enumerate(
        zip(data, original_labels, predicted_labels)
    ):
        plt.subplot(rows, cols, pos[i])
        plt.plot(plot_data)
        plt.title(f"Actual Label : {og_label}\nPredicted Label : {pred_label}")
        fig.subplots_adjust(hspace=0.5)
    plt.show()


view_evaluated_eeg_plots(conv_model)