Commit aa0f4479 authored by Ard Kastrati's avatar Ard Kastrati
Browse files

Implemented new architecture for DeepEye, inspired by Xception and EEGNet

parent 0c27f129
import keras
import tensorflow as tf import tensorflow as tf
from tensorflow.keras.layers import Dense, Activation, Permute, Dropout import numpy as np
from tensorflow.keras.layers import Conv2D, MaxPooling2D, AveragePooling2D import tensorflow.keras as keras
from tensorflow.keras.layers import SeparableConv2D, DepthwiseConv2D
from tensorflow.keras.layers import BatchNormalization
from tensorflow.keras.constraints import max_norm
from keras.callbacks import CSVLogger
from config import config from config import config
from utils.utils import * from utils.utils import *
import seaborn as sns import logging
sns.set_style('darkgrid') from keras.callbacks import CSVLogger
def run(trainX, trainY): def run(trainX, trainY):
classifier = Classifier_DEEPEYE(output_directory=config['root_dir'], input_shape=config['deepeye']['input_shape']) logging.info("Starting DeepEyeX.")
hist = classifier.fit(deepeye_x=trainX, y=trainY) if config['split']:
config['model'] = config['model'] + '_cluster'
classifier = Classifier_DEEPEYEX(output_directory=config['root_dir'],
input_shape=config['inception']['input_shape'])
hist = classifier.fit(trainX, trainY)
plot_loss(hist, config['model_dir'], config['model'], True) plot_loss(hist, config['model_dir'], config['model'], True)
plot_acc(hist, config['model_dir'], config['model'], True) plot_acc(hist, config['model_dir'], config['model'], True)
save_logs(hist, config['model_dir'], config['model'], pytorch=False) save_logs(hist, config['model_dir'], config['model'], pytorch=False)
save_model_param(classifier.model, config['model_dir'], config['model'], pytorch=False) save_model_param(classifier.model, config['model_dir'], config['model'], pytorch=False)
class Classifier_DEEPEYE:
"""
Inputs:
nb_classes : int, number of classes to classify
input_shape : shape of the input tensor, in our case: 128 * 500 * 1
use_bottleneck : use Bottleneck layer to select the most informative channels
use_residual : use a shortcut layer (RNN) to try to avoid vanishing gradient
kernel_size : 41
batch_size : 64
epochs : 1500
output_directory: directory where plot weights and results are stored
depth : 6, number of repetion of the inception module
Outputs: class Classifier_DEEPEYEX:
def __init__(self, output_directory, input_shape, verbose=True, build=True, batch_size=64, nb_filters=32,
y_pred : class (left/right for nb_class=2) of the given input_tensor use_residual=True, use_bottleneck=True, depth=6, kernel_size=40, nb_epochs=1500, X=None):
"""
def __init__(self, output_directory, input_shape, verbose=False, build=True,
batch_size=64, nb_filters=32, use_residual=True, use_bottleneck=True, depth=6,
kernel_size=41, nb_epochs=1500):
self.output_directory = output_directory self.output_directory = output_directory
self.nb_filters = nb_filters self.nb_filters = nb_filters
self.use_residual = use_residual self.use_residual = use_residual
self.use_bottleneck = use_bottleneck self.use_bottleneck = use_bottleneck
self.depth = depth self.depth = depth
self.kernel_size = kernel_size - 1 self.kernel_size = kernel_size
self.callbacks = None self.callbacks = None
self.batch_size = batch_size self.batch_size = batch_size
self.bottleneck_size = 32 self.bottleneck_size = 32
self.nb_epochs = nb_epochs self.nb_epochs = nb_epochs
self.verbose = verbose
if build: if build:
# build model if config['split']:
self.model = self.build_model(input_shape) self.model = self.split_model(input_shape)
if verbose: else:
self.model = self._build_model(input_shape)
if self.verbose:
self.model.summary() self.model.summary()
self.verbose = verbose
# self.model.save_weights(self.output_directory + 'model_init.hdf5') # self.model.save_weights(self.output_directory + 'model_init.hdf5')
@staticmethod def split_model(self, input_shape):
def _eeg_preprocessing(input_tensor, F1=8, D=2, kernLength=250): input_layer = tf.keras.layers.Input(input_shape)
""" output = []
Static method since this function does not receive any reference argument from this class.
""" # run inception over the cluster
# EEGNet feature extraction for c in config['cluster'].keys():
Chans = config['deepeye']['channels'] a = [input_shape[0]]
Samples = config['deepeye']['samples'] a.append(len(config['cluster'][c]))
input_shape = tuple(a)
# Filter slides horizontally
horizontal_tensor = Conv2D(F1, (1, kernLength), padding='same', output.append(self._build_model(input_shape,
input_shape=(Chans, Samples, 1), use_bias=False)(input_tensor) X=tf.transpose(tf.nn.embedding_lookup(tf.transpose(input_layer),
horizontal_tensor = BatchNormalization()(horizontal_tensor) config['cluster'][c]))))
# Filter slides vertically # append the results and perform 1 dense layer with last_channel dimension and the output layer
vertical_tensor = DepthwiseConv2D((Chans, 1), use_bias=False,
depth_multiplier=D, x = tf.keras.layers.Concatenate()(output)
depthwise_constraint=max_norm(1.))(horizontal_tensor) dense = tf.keras.layers.Dense(32, activation='relu')(x)
vertical_tensor = BatchNormalization()(vertical_tensor) output_layer = tf.keras.layers.Dense(1, activation='sigmoid')(dense)
eeg_tensor = Activation('elu')(vertical_tensor) model = tf.keras.models.Model(inputs=input_layer, outputs=output_layer)
model.compile(loss='binary_crossentropy', optimizer=keras.optimizers.Adam(), metrics=['accuracy'])
# Reshape the tensor (129, 500, 1) to (129, 500), and feed into the inception module
output_tensor = eeg_tensor[:, 0, :, :] return model
#output_tensor = tf.transpose(output_tensor, perm=[0, 2, 1])
return output_tensor
def _inception_module(self, input_tensor, nb_filters=32, use_bottleneck=True, kernel_size=40, bottleneck_size=32, def _inception_module(self, input_tensor, nb_filters=32, use_bottleneck=True, kernel_size=40, bottleneck_size=32,
stride=1, activation='linear'): stride=1, activation='linear'):
if use_bottleneck and int(input_tensor.shape[-1]) > 1: if use_bottleneck and int(input_tensor.shape[-1]) > 1:
input_inception = keras.layers.Conv1D(filters=bottleneck_size, kernel_size=1, input_inception = tf.keras.layers.Conv1D(filters=bottleneck_size, kernel_size=1, padding='same',
padding='same', activation=activation, use_bias=False)(input_tensor) activation=activation, use_bias=False)(input_tensor)
else: else:
input_inception = input_tensor input_inception = input_tensor
...@@ -103,38 +85,45 @@ class Classifier_DEEPEYE: ...@@ -103,38 +85,45 @@ class Classifier_DEEPEYE:
conv_list = [] conv_list = []
for i in range(len(kernel_size_s)): for i in range(len(kernel_size_s)):
conv_list.append(keras.layers.Conv1D(filters=nb_filters, kernel_size=kernel_size_s[i], conv_list.append(
strides=stride, padding='same', activation=activation, tf.keras.layers.Conv1D(filters=nb_filters, kernel_size=kernel_size_s[i], strides=stride, padding='same',
use_bias=False)(input_inception)) activation=activation, use_bias=False)(input_inception))
max_pool_1 = keras.layers.MaxPool1D(pool_size=3, strides=stride, padding='same')(input_tensor) max_pool_1 = tf.keras.layers.MaxPool1D(pool_size=3, strides=stride, padding='same')(input_tensor)
conv_6 = keras.layers.Conv1D(filters=nb_filters, kernel_size=1, padding='same', activation=activation, conv_6 = tf.keras.layers.Conv1D(filters=nb_filters, kernel_size=1, padding='same', activation=activation,
use_bias=False)(max_pool_1) use_bias=False)(max_pool_1)
conv_7 = tf.keras.layers.SeparableConv1D(filters=nb_filters, kernel_size=32, padding='same', activation=activation, use_bias=False, depth_multiplier=1)(input_tensor)
conv_list.append(conv_6) conv_list.append(conv_6)
conv_list.append(conv_7)
x = keras.layers.Concatenate(axis=2)(conv_list) x = tf.keras.layers.Concatenate(axis=2)(conv_list)
x = keras.layers.BatchNormalization()(x) x = tf.keras.layers.BatchNormalization()(x)
x = keras.layers.Activation(activation='relu')(x) x = tf.keras.layers.Activation(activation='relu')(x)
return x return x
def _shortcut_layer(self, input_tensor, out_tensor): def _shortcut_layer(self, input_tensor, out_tensor):
shortcut_y = keras.layers.Conv1D(filters=int(out_tensor.shape[-1]), kernel_size=1,
padding='same', use_bias=False)(input_tensor) shortcut_y = tf.keras.layers.Conv1D(filters=int(out_tensor.shape[-1]), kernel_size=1,
shortcut_y = keras.layers.BatchNormalization()(shortcut_y) padding='same', use_bias=False)(input_tensor)
shortcut_y = tf.keras.layers.BatchNormalization()(shortcut_y)
x = keras.layers.Add()([shortcut_y, out_tensor]) x = keras.layers.Add()([shortcut_y, out_tensor])
x = keras.layers.Activation('relu')(x) x = keras.layers.Activation('relu')(x)
return x return x
def build_model(self, input_shape, nb_filters=32, use_residual=True, use_bottleneck=True, depth=6, kernel_size=40, F1=8, def _build_model(self, input_shape, X=[], nb_filters=32, use_residual=True, use_bottleneck=True, depth=20,
D=2, kernLength=125): kernel_size=40):
input_layer = keras.layers.Input((input_shape[0], input_shape[1], 1)) if config['split']:
eeg_tensor = self._eeg_preprocessing(input_layer, F1, D, kernLength) input_layer = X
x = eeg_tensor
input_res = eeg_tensor else:
input_layer = tf.keras.layers.Input(input_shape)
x = input_layer
input_res = input_layer
for d in range(depth): for d in range(depth):
...@@ -145,14 +134,14 @@ class Classifier_DEEPEYE: ...@@ -145,14 +134,14 @@ class Classifier_DEEPEYE:
input_res = x input_res = x
gap_layer = tf.keras.layers.GlobalAveragePooling1D()(x) gap_layer = tf.keras.layers.GlobalAveragePooling1D()(x)
if config['split']:
return gap_layer
output_layer = tf.keras.layers.Dense(1, activation='sigmoid')(gap_layer) output_layer = tf.keras.layers.Dense(1, activation='sigmoid')(gap_layer)
model = tf.keras.models.Model(inputs=input_layer, outputs=output_layer) model = tf.keras.models.Model(inputs=input_layer, outputs=output_layer)
model.compile(loss='binary_crossentropy', optimizer=keras.optimizers.Adam(), metrics=['accuracy'])
return model return model
def fit(self, deepeye_x, y): def fit(self, inception_x, y):
self.model.compile(loss='binary_crossentropy', optimizer=keras.optimizers.Adam(), metrics=['accuracy'])
csv_logger = CSVLogger(config['batches_log'], append=True, separator=';') csv_logger = CSVLogger(config['batches_log'], append=True, separator=';')
hist = self.model.fit(deepeye_x, y, verbose=1, validation_split=0.2, epochs=1, callbacks=[csv_logger]) hist = self.model.fit(inception_x, y, verbose=1, validation_split=0.2, epochs=35, callbacks=[csv_logger])
return hist return hist
...@@ -22,7 +22,7 @@ def run(trainX, trainY): ...@@ -22,7 +22,7 @@ def run(trainX, trainY):
class Classifier_INCEPTION: class Classifier_INCEPTION:
def __init__(self, output_directory, input_shape, verbose=False, build=True, batch_size=64, nb_filters=32, def __init__(self, output_directory, input_shape, verbose=True, build=True, batch_size=64, nb_filters=32,
use_residual=True, use_bottleneck=True, depth=6, kernel_size=40, nb_epochs=1500, X = None): use_residual=True, use_bottleneck=True, depth=6, kernel_size=40, nb_epochs=1500, X = None):
self.output_directory = output_directory self.output_directory = output_directory
......
...@@ -35,7 +35,7 @@ deepeye: Our method ...@@ -35,7 +35,7 @@ deepeye: Our method
""" """
# Choosing model # Choosing model
config['model'] = 'inception' config['model'] = 'deepeye2'
config['downsampled'] = False config['downsampled'] = False
config['split'] = False config['split'] = False
config['cluster'] = clustering() config['cluster'] = clustering()
...@@ -51,12 +51,15 @@ config['cnn'] = {} ...@@ -51,12 +51,15 @@ config['cnn'] = {}
config['inception'] = {} config['inception'] = {}
# DeepEye # DeepEye
config['deepeye'] = {} config['deepeye'] = {}
# DeepEye2
config['deepeye2'] = {}
# EEGNet # EEGNet
config['eegnet'] = {} config['eegnet'] = {}
# LSTM-DeepEye # LSTM-DeepEye
config['deepeye-lstm'] = {} config['deepeye-lstm'] = {}
config['inception']['input_shape'] = (125, 129) if config['downsampled'] else (500, 129) config['inception']['input_shape'] = (125, 129) if config['downsampled'] else (500, 129)
config['deepeye2']['input_shape'] = (125, 129) if config['downsampled'] else (500, 129)
config['eegnet']['channels'] = 129 config['eegnet']['channels'] = 129
config['eegnet']['samples'] = 125 if config['downsampled'] else 500 config['eegnet']['samples'] = 125 if config['downsampled'] else 500
config['deepeye']['input_shape'] = (129, 125) if config['downsampled'] else (129, 500) config['deepeye']['input_shape'] = (129, 125) if config['downsampled'] else (129, 500)
......
...@@ -2,7 +2,7 @@ from config import config ...@@ -2,7 +2,7 @@ from config import config
import time import time
from CNN import CNN from CNN import CNN
from utils import IOHelper from utils import IOHelper
from DeepEye import deepEye from DeepEye import deepEye, RNNdeep, deepEye2
from InceptionTime import inception from InceptionTime import inception
from EEGNet import eegNet from EEGNet import eegNet
import numpy as np import numpy as np
...@@ -40,6 +40,10 @@ def main(): ...@@ -40,6 +40,10 @@ def main():
deepeye_x = np.transpose(trainX, (0, 2, 1)) deepeye_x = np.transpose(trainX, (0, 2, 1))
logging.info(deepeye_x.shape) logging.info(deepeye_x.shape)
deepEye.run(trainX=deepeye_x, trainY=trainY) deepEye.run(trainX=deepeye_x, trainY=trainY)
elif config['model'] == 'deepeye2':
logging.info("Started running DeepEye2. If you want to run other methods please choose another model in the config.py file.")
deepEye2.run(trainX=trainX, trainY=trainY)
elif config['model'] == 'deepeye-lstm': elif config['model'] == 'deepeye-lstm':
logging.info("Started running deepeye-lstm. If you want to run other methods please choose another model in the config.py file.") logging.info("Started running deepeye-lstm. If you want to run other methods please choose another model in the config.py file.")
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment