28、ResNet50处理STEW数据集：从EEG信号到三分类情感识别的完整实践

1. STEW数据集与EEG信号基础STEW数据集全称为SIMULTANEOUS TASK EEG WORKLOAD DATASET是由48名受试者参与多任务工作负荷实验时采集的原始EEG数据。这个数据集特别之处在于它同时记录了受试者在休息状态和任务执行时的大脑活动为研究认知负荷和情感状态提供了宝贵资源。每个受试者的数据文件命名遵循subno_task.txt的格式比如sub01_lo.txt表示1号受试者在低负荷任务时的数据。数据集包含2.5分钟的EEG记录采样频率为128Hz通过Emotiv EPOC设备采集14个通道的信号通道位置包括AF3, F7, F3, FC5, T7, P7, O1, O2, P8, T8, FC6, F4, F8, AF4。EEG信号本质上是大脑神经元活动产生的微弱电信号幅度通常在微伏级别。原始EEG数据呈现为时间序列格式每一行代表一个时间点的采样每一列对应一个电极通道。这种时序信号直接反映了大脑在不同认知状态下的电活动模式但同时也包含大量噪声和干扰需要进行专业的预处理才能用于机器学习任务。在实际处理中我发现EEG数据有几个关键特点需要注意信号非常微弱容易受到眼动、肌电等生理伪迹干扰不同受试者间的个体差异较大情感状态的变化通常体现在特定频段的能量变化上2. EEG信号预处理实战2.1 基础预处理流程拿到原始EEG数据后第一步是进行必要的预处理。我通常会按照以下步骤操作import numpy as np import mne from scipy import signal def preprocess_eeg(raw_data, sfreq128): # 1. 去除直流偏移 mean_val np.mean(raw_data, axis0) data raw_data - mean_val # 2. 带通滤波 (0.5-45Hz) b, a signal.butter(4, [0.5, 45], btypebandpass, fssfreq) filtered_data signal.filtfilt(b, a, data, axis0) # 3. 去除眼电伪迹 (使用ICA) info mne.create_info(ch_namesch_names, sfreqsfreq, ch_typeseeg) raw mne.io.RawArray(filtered_data.T, info) ica mne.preprocessing.ICA(n_components14, random_state97) ica.fit(raw) ica.apply(raw) # 4. 重参考 (平均参考) data raw.get_data().T ref np.mean(data, axis1, keepdimsTrue) data data - ref return data这个预处理流程包含了EEG处理的几个关键步骤去除直流偏移可以消除设备本身的基线漂移带通滤波保留了最有价值的脑电频段ICA能有效去除眼动等伪迹重参考则使信号更加稳定。2.2 时频分析转换为了将EEG时序信号转换为适合ResNet50处理的图像格式时频分析是关键。我最常用的是短时傅里叶变换(STFT)def create_spectrogram(eeg_data, window_size256, overlap128): spectrograms [] for channel in range(eeg_data.shape[1]): f, t, Sxx signal.spectrogram( eeg_data[:, channel], fs128, windowhann, npersegwindow_size, noverlapoverlap ) # 转换为dB单位 Sxx 10 * np.log10(Sxx 1e-12) spectrograms.append(Sxx) # 合并所有通道的频谱图 spectrogram np.stack(spectrograms, axis-1) return spectrogram这样得到的频谱图是三维的(频率×时间×通道)可以直接视为类似RGB图像的格式。我通常会选择5个关键频带δ(0.5-4Hz)、θ(4-8Hz)、α(8-13Hz)、β(13-30Hz)和γ(30-45Hz)这些频带与不同情感状态密切相关。3. ResNet50模型适配与调优3.1 标准ResNet50的调整原始的ResNet50是为ImageNet设计的直接用于EEG频谱图需要做一些调整import torch import torch.nn as nn from torchvision.models import resnet50 class EEGResNet(nn.Module): def __init__(self, num_classes3): super(EEGResNet, self).__init__() # 加载预训练的ResNet50 self.base_model resnet50(pretrainedTrue) # 修改第一层卷积适应单通道输入 original_conv1 self.base_model.conv1 self.base_model.conv1 nn.Conv2d( 1, 64, kernel_sizeoriginal_conv1.kernel_size, strideoriginal_conv1.stride, paddingoriginal_conv1.padding, biasFalse ) # 修改最后的全连接层 in_features self.base_model.fc.in_features self.base_model.fc nn.Linear(in_features, num_classes) def forward(self, x): return self.base_model(x)这里的关键修改点有三个将输入通道从3(RGB)改为1(灰度)保持其他卷积层的预训练权重调整最后的全连接层输出为3类(对应情感分类)3.2 针对EEG数据的特殊优化在实战中我发现以下几个优化策略特别有效通道注意力机制EEG不同通道的重要性差异很大加入SE模块能提升性能class SEBlock(nn.Module): def __init__(self, channel, reduction16): super(SEBlock, self).__init__() self.avg_pool nn.AdaptiveAvgPool2d(1) self.fc nn.Sequential( nn.Linear(channel, channel // reduction), nn.ReLU(inplaceTrue), nn.Linear(channel // reduction, channel), nn.Sigmoid() ) def forward(self, x): b, c, _, _ x.size() y self.avg_pool(x).view(b, c) y self.fc(y).view(b, c, 1, 1) return x * y.expand_as(x)频带特异性卷积不同频带应该有不同的特征提取方式class BandSpecificConv(nn.Module): def __init__(self, in_channels, out_channels): super(BandSpecificConv, self).__init__() self.delta_conv nn.Conv2d(1, out_channels//5, kernel_size3, padding1) self.theta_conv nn.Conv2d(1, out_channels//5, kernel_size3, padding1) self.alpha_conv nn.Conv2d(1, out_channels//5, kernel_size3, padding1) self.beta_conv nn.Conv2d(1, out_channels//5, kernel_size3, padding1) self.gamma_conv nn.Conv2d(1, out_channels//5, kernel_size3, padding1) def forward(self, x): # x shape: [batch, 5, height, width] (5频带) delta self.delta_conv(x[:, 0:1, :, :]) theta self.theta_conv(x[:, 1:2, :, :]) alpha self.alpha_conv(x[:, 2:3, :, :]) beta self.beta_conv(x[:, 3:4, :, :]) gamma self.gamma_conv(x[:, 4:5, :, :]) return torch.cat([delta, theta, alpha, beta, gamma], dim1)4. 完整训练流程与评估4.1 数据准备与增强针对EEG数据我设计了一套专门的数据增强策略from torch.utils.data import Dataset, DataLoader from sklearn.model_selection import train_test_split class EEGDataset(Dataset): def __init__(self, data, labels, transformNone): self.data data self.labels labels self.transform transform def __len__(self): return len(self.data) def __getitem__(self, idx): sample self.data[idx] label self.labels[idx] if self.transform: sample self.transform(sample) return sample, label # 数据增强变换 class RandomTimeShift: def __init__(self, max_shift10): self.max_shift max_shift def __call__(self, x): shift np.random.randint(-self.max_shift, self.max_shift1) if shift 0: return np.concatenate([x[shift:], x[-shift:]], axis0) elif shift 0: return np.concatenate([x[:shift], x[:-shift]], axis0) return x # 划分训练测试集 X_train, X_test, y_train, y_test train_test_split( all_spectrograms, all_labels, test_size0.2, random_state42 ) train_dataset EEGDataset(X_train, y_train, transformRandomTimeShift()) test_dataset EEGDataset(X_test, y_test) train_loader DataLoader(train_dataset, batch_size32, shuffleTrue) test_loader DataLoader(test_dataset, batch_size32, shuffleFalse)4.2 模型训练与验证完整的训练流程包含以下几个关键组件import torch.optim as optim from torch.optim.lr_scheduler import ReduceLROnPlateau device torch.device(cuda if torch.cuda.is_available() else cpu) model EEGResNet(num_classes3).to(device) criterion nn.CrossEntropyLoss() optimizer optim.AdamW(model.parameters(), lr1e-4, weight_decay1e-3) scheduler ReduceLROnPlateau(optimizer, max, patience3, factor0.5) def train_epoch(model, loader, criterion, optimizer): model.train() running_loss 0.0 correct 0 total 0 for inputs, labels in loader: inputs inputs.float().to(device) labels labels.long().to(device) optimizer.zero_grad() outputs model(inputs) loss criterion(outputs, labels) loss.backward() optimizer.step() running_loss loss.item() _, predicted outputs.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() return running_loss/len(loader), 100.*correct/total def evaluate(model, loader, criterion): model.eval() running_loss 0.0 correct 0 total 0 with torch.no_grad(): for inputs, labels in loader: inputs inputs.float().to(device) labels labels.long().to(device) outputs model(inputs) loss criterion(outputs, labels) running_loss loss.item() _, predicted outputs.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() return running_loss/len(loader), 100.*correct/total # 训练循环 best_acc 0 for epoch in range(50): train_loss, train_acc train_epoch(model, train_loader, criterion, optimizer) val_loss, val_acc evaluate(model, val_loader, criterion) scheduler.step(val_acc) print(fEpoch {epoch1}: Train Loss: {train_loss:.4f} Acc: {train_acc:.2f}% | fVal Loss: {val_loss:.4f} Acc: {val_acc:.2f}%) if val_acc best_acc: best_acc val_acc torch.save(model.state_dict(), best_model.pth)4.3 结果分析与模型解释训练完成后我们不仅要看准确率还要分析模型在不同情感类别上的表现。混淆矩阵和分类报告是很好的工具from sklearn.metrics import confusion_matrix, classification_report import seaborn as sns import matplotlib.pyplot as plt def plot_confusion_matrix(true_labels, pred_labels, classes): cm confusion_matrix(true_labels, pred_labels) plt.figure(figsize(8,6)) sns.heatmap(cm, annotTrue, fmtd, cmapBlues, xticklabelsclasses, yticklabelsclasses) plt.xlabel(Predicted) plt.ylabel(True) plt.show() # 获取测试集预测结果 all_preds [] all_labels [] with torch.no_grad(): for inputs, labels in test_loader: inputs inputs.float().to(device) outputs model(inputs) _, preds outputs.max(1) all_preds.extend(preds.cpu().numpy()) all_labels.extend(labels.numpy()) # 绘制混淆矩阵 plot_confusion_matrix(all_labels, all_preds, classes[Negative, Neutral, Positive]) # 打印分类报告 print(classification_report(all_labels, all_preds, target_names[Negative, Neutral, Positive]))在实际项目中我发现模型对中性情感的识别准确率通常较低这可能是因为中性状态的大脑活动模式不如极端情感那么明显。通过增加难样本挖掘和数据增强可以显著改善这一情况。5. 部署优化与实用技巧5.1 模型轻量化为了在实际应用中部署我们需要考虑模型大小和推理速度。这里有几个有效的轻量化策略知识蒸馏使用更大的教师模型来训练小型学生模型量化将模型参数从FP32转换为INT8剪枝移除不重要的神经元连接# 量化示例 quantized_model torch.quantization.quantize_dynamic( model, {nn.Linear, nn.Conv2d}, dtypetorch.qint8 ) # 剪枝示例 from torch.nn.utils import prune parameters_to_prune ( (model.base_model.conv1, weight), (model.base_model.layer1[0].conv1, weight), ) prune.global_unstructured( parameters_to_prune, pruning_methodprune.L1Unstructured, amount0.2, )5.2 跨受试者泛化EEG数据的一个主要挑战是不同受试者间的差异很大。我通常采用以下方法提升模型的泛化能力受试者无关训练在训练集中排除测试受试者的数据特征标准化使用受试者特定的Z-score标准化域适应技术如对抗训练来减小域间差异# 对抗训练示例 class DomainClassifier(nn.Module): def __init__(self, input_dim): super(DomainClassifier, self).__init__() self.fc nn.Sequential( nn.Linear(input_dim, 256), nn.ReLU(), nn.Linear(256, 128), nn.ReLU(), nn.Linear(128, 1), nn.Sigmoid() ) def forward(self, x): return self.fc(x) # 在特征提取器后添加对抗损失 feature_extractor model.base_model[:-1] # 去掉最后的全连接层 domain_classifier DomainClassifier(2048).to(device) # 对抗训练循环 for epoch in range(10): for inputs, labels, domains in train_loader: # 特征提取 features feature_extractor(inputs) # 1. 主任务训练 outputs model.fc(features) main_loss criterion(outputs, labels) # 2. 域分类器训练 domain_preds domain_classifier(features.detach()) domain_loss F.binary_cross_entropy(domain_preds, domains.float()) # 3. 特征提取器对抗训练 domain_preds domain_classifier(features) adversarial_loss -F.binary_cross_entropy(domain_preds, domains.float()) # 组合损失 total_loss main_loss 0.1 * domain_loss 0.1 * adversarial_loss optimizer.zero_grad() total_loss.backward() optimizer.step()5.3 实际部署注意事项在实际部署EEG情感识别系统时有几个关键点需要注意实时性要求EEG信号处理需要满足实时性通常要求延迟小于200ms设备差异不同EEG设备的通道配置和信号特性可能不同环境噪声实际使用环境中的电磁干扰比实验室更复杂用户适应性模型需要适应用户的长期脑电模式变化针对这些问题我建议使用滑动窗口处理实时数据流开发设备特定的校准程序实现在线学习能力让模型能持续适应用户添加信号质量检测模块在信号质量差时暂停预测# 实时处理示例 class RealTimeEEGProcessor: def __init__(self, model_path, window_size256, stride64): self.model load_model(model_path) self.buffer np.zeros((window_size, 14)) # 14个通道 self.window_size window_size self.stride stride self.counter 0 def process_sample(self, new_sample): # 更新缓冲区 self.buffer np.roll(self.buffer, -1, axis0) self.buffer[-1] new_sample self.counter 1 if self.counter % self.stride 0: # 预处理 processed preprocess_eeg(self.buffer) # 频谱图转换 spec create_spectrogram(processed) # 预测 with torch.no_grad(): pred self.model(torch.from_numpy(spec).float().unsqueeze(0)) return pred.argmax().item() return None