手把手教你使用 1D 卷积和 LSTM 混合模型做 EEG 信号识别

脑机接口社区

发布于 2022-08-17 09:52:06

6020

发布于 2022-08-17 09:52:06

文章被收录于专栏：脑机接口

本文主要介绍了使用 1D 卷积和 LSTM 混合模型做 EEG 信号识别。感谢Memory逆光！

内容包括：1. 数据集(1.1 数据集下载、1.2 数据集解释）；2. 读取数据；3. 搭建模型；4. 训练模型；5. 展示结果；6. 完整代码。

数据集

1.1 数据集下载：

https://archive.ics.uci.edu/ml/datasets/Epileptic+Seizure+Recognition

打开看一下

1.2 数据集解释：

表头为 X* 的是电信号数据，共有 11500 行，每行有 178 个数据，表示 1s 时间内截取的 178 个电信号；表头为 Y 的一列是该时间段数据的标签，包括 5 个分类：

5-记录大脑的EEG信号时病人睁开了眼睛；

4-记录大脑的EEG信号时患者闭上了眼睛；

3-健康大脑区域的脑电图活动；

2-肿瘤所在区域的脑电图活动；

1-癫痫发作活动；

读取数据

import pandas as pd

data = "data.csv"

df = pd.read_csv(data, header=0, index_col=0)
df1 = df.drop(["y"], axis=1)
lbls = df["y"].values - 1

这里使用 pandas 库读取 data.csv，df1 保存电位数据，lbls 保存标签；

import numpy as np

wave = np.zeros((11500, 178))

z = 0
for index, row in df1.iterrows():
    wave[z, :] = row
    z+=1

mean = wave.mean(axis=0)
wave -= mean
std = wave.std(axis=0)
wave /= std

def one_hot(y):
    lbl = np.zeros(5)
    lbl[y] = 1
    return lbl

target = []
for value in lbls:
    target.append(one_hot(value))
target = np.array(target)
wave = np.expand_dims(wave, axis=-1)

我们将数据保存在数组 wave 和 target 中，将点位数据标准化（减去均值后除以方差），并将标签转换成 one hot 的形式；

搭建模型

我们使用 keras 搭建一个模型，包括 1D 卷积层和几个堆叠的 LSTM 层：

from keras.models import Sequential
from keras import layers

model = Sequential()
model.add(layers.Conv1D(64, 15, strides=2,input_shape=(178, 1), use_bias=False))
model.add(layers.ReLU())
model.add(layers.Conv1D(64, 3))
model.add(layers.Conv1D(64, 3, strides=2))
model.add(layers.ReLU())
model.add(layers.Conv1D(64, 3))
model.add(layers.Conv1D(64, 3, strides=2))  # [None, 54, 64]
model.add(layers.BatchNormalization())
model.add(layers.LSTM(64, dropout=0.5, return_sequences=True))
model.add(layers.LSTM(64, dropout=0.5, return_sequences=True))
model.add(layers.LSTM(32))
model.add(layers.Dense(5, activation="softmax"))
model.summary()

网络结构如图：

即该模型使用 1D 卷积进行特征提取，使用 LSTM 进行时域建模，最后通过一个全连接层预测类别；

训练模型

我们使用 Adam 优化器，并设置学习率衰减来进行训练：

import matplotlib.pyplot as plt
import pandas as pd
from keras.models import Sequential
from keras import layers
from keras import regularizers
import os
import keras

import keras.backend as K

save_path = './keras_model.h5'

if os.path.isfile(save_path):
    model.load_weights(save_path)
    print('reloaded.')

adam = keras.optimizers.adam()

model.compile(optimizer=adam,
              loss="categorical_crossentropy", metrics=["acc"])
# 计算学习率
def lr_scheduler(epoch):
    # 每隔100个epoch，学习率减小为原来的0.5
    if epoch % 100 == 0 and epoch != 0:
        lr = K.get_value(model.optimizer.lr)
        K.set_value(model.optimizer.lr, lr * 0.5)
        print("lr changed to {}".format(lr * 0.5))
    return K.get_value(model.optimizer.lr)

lrate = LearningRateScheduler(lr_scheduler)

history = model.fit(wave, target, epochs=400,
                    batch_size=128, validation_split=0.2,
                    verbose=1, callbacks=[lrate])

model.save_weights(save_path)

这样就可以开始训练啦：

训练的模型参数保存在 sace_path 中；

展示结果

print(history.history.keys())
# summarize history for accuracy
plt.plot(history.history['acc'])
plt.plot(history.history['val_acc'])
plt.title('model accuracy')
plt.ylabel('accuracy')
plt.xlabel('epoch')
plt.legend(['train', 'test'], loc='upper left')
plt.show()
# summarize history for loss
plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.title('model loss')
plt.ylabel('loss')
plt.xlabel('epoch')
plt.legend(['train', 'test'], loc='upper left')
plt.show()

这时我们可以查看训练结果（因为时间有限，我只训练了 100 个 epoch：

完整代码


import matplotlib.pyplot as plt
import pandas as pd
from keras.models import Sequential
from keras import layers
from keras import regularizers
import os
import keras

import keras.backend as K

import numpy as np

from keras.callbacks import LearningRateScheduler

data = "data.csv"

df = pd.read_csv(data, header=0, index_col=0)
df1 = df.drop(["y"], axis=1)
lbls = df["y"].values - 1

wave = np.zeros((11500, 178))

z = 0
for index, row in df1.iterrows():
    wave[z, :] = row
    z+=1

mean = wave.mean(axis=0)
wave -= mean
std = wave.std(axis=0)
wave /= std

def one_hot(y):
    lbl = np.zeros(5)
    lbl[y] = 1
    return lbl

target = []
for value in lbls:
    target.append(one_hot(value))
target = np.array(target)
wave = np.expand_dims(wave, axis=-1)

model = Sequential()
model.add(layers.Conv1D(64, 15, strides=2,
                        input_shape=(178, 1), use_bias=False))
model.add(layers.ReLU())
model.add(layers.Conv1D(64, 3))
model.add(layers.Conv1D(64, 3, strides=2))
model.add(layers.BatchNormalization())
model.add(layers.Dropout(0.5))
model.add(layers.Conv1D(64, 3))
model.add(layers.Conv1D(64, 3, strides=2))
model.add(layers.BatchNormalization())
model.add(layers.LSTM(64, dropout=0.5, return_sequences=True))
model.add(layers.LSTM(64, dropout=0.5, return_sequences=True))
model.add(layers.LSTM(32))
model.add(layers.Dropout(0.5))
model.add(layers.Dense(5, activation="softmax"))
model.summary()

save_path = './keras_model3.h5'

if os.path.isfile(save_path):
    model.load_weights(save_path)
    print('reloaded.')

adam = keras.optimizers.adam()

model.compile(optimizer=adam,
              loss="categorical_crossentropy", metrics=["acc"])
# 计算学习率
def lr_scheduler(epoch):
    # 每隔100个epoch，学习率减小为原来的0.5
    if epoch % 100 == 0 and epoch != 0:
        lr = K.get_value(model.optimizer.lr)
        K.set_value(model.optimizer.lr, lr * 0.5)
        print("lr changed to {}".format(lr * 0.5))
    return K.get_value(model.optimizer.lr)

lrate = LearningRateScheduler(lr_scheduler)

history = model.fit(wave, target, epochs=400,
                    batch_size=128, validation_split=0.2,
                    verbose=2, callbacks=[lrate])

model.save_weights(save_path)

print(history.history.keys())
# summarize history for accuracy
plt.plot(history.history['acc'])
plt.plot(history.history['val_acc'])
plt.title('model accuracy')
plt.ylabel('accuracy')
plt.xlabel('epoch')
plt.legend(['train', 'test'], loc='upper left')
plt.show()
# summarize history for loss
plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.title('model loss')
plt.ylabel('loss')
plt.xlabel('epoch')
plt.legend(['train', 'test'], loc='upper left')
plt.show()

作者博客：

https://blog.csdn.net/weixin_44936889/article/details/105202661

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2021-07-19，如有侵权请联系 cloudcommunity@tencent.com 删除

机器学习