Homec4science
Diffusion DeepHealth_UC13_seizure_detection (pyeddl_0.9.0) Browse

R9868/src_keras/utils.py
e684d635256apyeddl_0.9.0

Last Change
Raw File

utils.py
View Options

import numpy as np
import os
from matplotlib import pyplot as plt
#sklearn
from sklearn.metrics import confusion_matrix, classification_report, roc_curve,roc_auc_score
#Tensorflow
#os.environ["CUDA_VISIBLE_DEVICES"]="0"
os.environ["CUDA_VISIBLE_DEVICES"]=""
import tensorflow as tf
#gpu_options = tf.GPUOptions(per_process_gpu_memory_fraction=0.95)
#sess = tf.Session(config=tf.ConfigProto(gpu_options=gpu_options))
sess = tf.Session(config=tf.ConfigProto(intra_op_parallelism_threads=4, inter_op_parallelism_threads=4))
#keras
from keras.models import model_from_json
from keras.models import Sequential
from keras.layers import Dense, Dropout, Activation, Flatten
from keras.layers import Conv2D, BatchNormalization, Reshape, MaxPooling2D, ZeroPadding2D
from keras.utils import to_categorical
from keras.regularizers import l2
from keras.optimizers import SGD, Adam
from keras.callbacks import EarlyStopping
NB_CHNS = 4
l2_k, l2_b, l2_a = 0.1, 0.1, 0
dropout_rate = 0.5
def getModel():
model = Sequential()
model.add(ZeroPadding2D((1,2), input_shape=(NB_CHNS,1280,1,)))
model.add(Conv2D(16, (3, 5),data_format="channels_last", kernel_regularizer=l2(l2_k)))
model.add(BatchNormalization())
model.add(Activation("relu"))
model.add(MaxPooling2D((1,2)))
model.add(ZeroPadding2D((1,1)))
model.add(Conv2D(32, (3, 3), strides=(1,1),data_format="channels_last", kernel_regularizer=l2(l2_k)))
model.add(BatchNormalization())
model.add(Activation("relu"))
model.add(MaxPooling2D((1,2)))
model.add(ZeroPadding2D((1,1)))
model.add(Conv2D(32, (3, 3), strides=(2,2),data_format="channels_last", kernel_regularizer=l2(l2_k)))
model.add(BatchNormalization())
model.add(Activation("relu"))
model.add(Dropout(dropout_rate))
model.add(Flatten())
model.add(Dense(64, activation="relu", kernel_regularizer=l2(l2_k)))
model.add(Dropout(dropout_rate))
model.add(Dense(2, activation="softmax"))
model.compile(optimizer=SGD(),
loss="categorical_crossentropy",
metrics=['categorical_accuracy'])
#print(model.summary())
return model
#Train_set and test_set are pandas dataframes where the signal is in the column labeled "signal" and the label "label"
def cut_signal_data(train_set, test_set, size_train = 1280, stride_train = 640, size_test=1280, stride_test=1280):
x_train = []
y_train = []
x_test = []
y_test = []
#For each training seizure/pre-ictal signals
for row in train_set.itertuples(index = False):
signal = row.signal
label = row.label
#Cut signal in chunks of 5 seconds
signals = sliding_window(signal, size_train, stride_train)
labels = np.ones(len(signals))*label
x_train+=list(signals)
y_train+=list(labels)
#For each testing seizure/pre-ictal signals
for row in test_set.itertuples(index = False):
signal = row.signal
label = row.label
#Cut signal in chunks of 5 seconds
signals = sliding_window(signal, size_test, stride_test)
labels = np.ones(len(signals))*label
x_test+=list(signals)
y_test+=list(labels)
return x_train, y_train, x_test, y_test
#input: eeg signal, size: size of the window, stride: step Default : 1280 long windows with 50% overlap output : array of signal cuts according to the window size
def sliding_window(signal,size = 1280, stride = 640):
out = []
numOfChunks = int(((signal.shape[1]-size)/stride)+1)
for i in range(0,numOfChunks*stride,stride):
out.append(signal[:,i:i+size])
return out
#train data according to keras.fit function
def train(model, x_train, y_train, val_data = None, verb=1, epochs = 50, batch_size = 32, cb=None):
#If no pre split validation set is provided, done autamatically here. But may result in class imbalance for small datasets
if(val_data==None):
history = model.fit(x_train, y_train, verbose = verb, epochs=epochs, shuffle=True, batch_size = batch_size, callbacks = cb, validation_split=0.15)
else:
val_data = [val_data[0], val_data[1]]
history = model.fit(x_train, y_train, verbose = verb, epochs=epochs, shuffle=True, batch_size = batch_size, callbacks = cb, validation_data=val_data)
return history
def prepare_standardplot(title, xlabel):
fig, (ax1, ax2) = plt.subplots(1, 2)
fig.set_size_inches(12,6)
fig.suptitle(title)
ax1.set_ylabel('binary cross entropy')
ax1.set_xlabel(xlabel)
ax1.set_yscale('log')
ax2.set_ylabel('accuracy [% correct]')
ax2.set_xlabel(xlabel)
return fig, ax1, ax2
def finalize_standardplot(fig, ax1, ax2):
ax1handles, ax1labels = ax1.get_legend_handles_labels()
if len(ax1labels) > 0:
ax1.legend(ax1handles, ax1labels)
ax2handles, ax2labels = ax2.get_legend_handles_labels()
if len(ax2labels) > 0:
ax2.legend(ax2handles, ax2labels)
fig.tight_layout()
plt.subplots_adjust(top=0.9)
def plot_history(history, title):
fig, ax1, ax2 = prepare_standardplot(title, 'epoch')
ax1.plot(history.history['loss'], label = "training")
ax1.plot(history.history['val_loss'], label = "validation")
ax2.plot(history.history['acc'], label = "training")
ax2.plot(history.history['val_acc'], label = "validation")
finalize_standardplot(fig, ax1, ax2)
return fig
#Display model performance on test set
def show_confusion_matrix(y_true_, y_pred_, title):
cm = confusion_matrix(y_pred=np.max(np.round(y_pred_), axis=1), y_true=np.argmax(y_true_, axis=1))
print(cm)
#plt.imshow(cm, cmap="inferno_r") plt.title(title) plt.show()
print(classification_report(y_pred=np.max(np.round(y_pred_), axis=1), y_true=np.argmax(y_true_, axis=1)))
#Calculate accuracy on test set
def compute_accuracy(y_pred_, y_true_):
return 1 - np.sum(np.abs(np.max(np.round(y_pred_), axis=1) - np.argmax(y_true_, axis=1)))/len(y_true_)
def false_positive_rate(y_test, y_pred, detect_rule):
time_hr = len(y_pred)*1280/(256*60*60)
preds = np.max(np.rint(y_pred), axis=1)
fp=0
x_ = np.zeros(detect_rule[1])
alarm_triggered = False
counter_alarm = 23 #Needs 1mn seconds between seizure onsets
false_alarms = []
for idx, x in enumerate(preds):
if(counter_alarm == 0):
alarm_triggered = False
else:
counter_alarm -= 1
if (alarm_triggered == False):
for j, y in enumerate(x_[::-1]):
if(j == len(x_)-1):
x_[1] = x_[0]
x_[0] = 0
else:
x_[len(x_)-1-j] = x_[len(x_)-2-j]
if(x==1):
x_[0]=1
if(np.sum(x_) >= detect_rule[0]):
fp+=1
alarm_triggered = True
counter_alarm = 23
false_alarms.append(idx)
fpr = fp/time_hr
return fpr, fp, false_alarms
#Compute detection time: first time to have 2 out of 3 segments being classified as ictal
def compute_detect_time(preds, detect_rule):
x_ = np.zeros(detect_rule[1])
for idx, x in enumerate(np.max(np.round(preds), axis=1)):
for j, y in enumerate(x_[::-1]):
if(j == len(x_)-1):
x_[1] = x_[0]
x_[0] = 0
else:
x_[len(x_)-1-j] = x_[len(x_)-2-j]
if(x==1):
x_[0]=1
if(np.sum(x_) >= detect_rule[0]):
return idx
return -1
c4science · Help