基于脑电信号的疲劳驾驶状态识别深度学习模型，告别疲劳驾驶：基于EEG信号与深度学习的脑电疲劳状态识别系统

目录

引言：当AI读懂你的困意

第一部分：背景与相关技术

1.1 为什么选择脑电信号（EEG）？

1.2 深度学习的优势

1.3 本文技术栈

第二部分：数据集介绍与预处理

2.1 公开数据集选择

2.2 预处理流水线

2.3 数据增强策略

第三部分：深度学习模型架构

3.1 基线模型：EEGNet

3.2 现代架构：EEG-Conformer（CNN + Transformer）

3.3 创新架构：图动态注意力网络（DG-Transformer）

3.4 半监督对比学习框架（SimCLR for EEG）

第四部分：完整训练流程与评估

4.1 数据加载与划分

4.2 训练函数（支持混合精度）

4.3 评估与分析

第五部分：实时检测系统部署

5.1 流式推理架构

引言：当AI读懂你的困意

你有没有经历过这样的时刻——在高速公路上，眼皮越来越重，意识逐渐模糊，突然一个激灵惊醒，发现自己差点撞上前方的车辆？疲劳驾驶，这个被称为“隐形杀手”的马路危机，每年在全球范围内造成数十万起交通事故。根据世界卫生组织的统计，约有20%的重大交通事故与疲劳驾驶直接相关。而在中国，这一比例甚至更高，尤其是在长途货运、客运领域。

那么，有没有一种技术，可以在驾驶员的大脑发出“我要睡着了”信号的第一时间就做出预警？答案就是——基于脑电信号（EEG）的疲劳驾驶状态识别深度学习模型。

本文将带你从零开始，完整实现一个端到端的疲劳驾驶检测系统。我们将结合最新的Transformer模型、图神经网络（GCN）以及对比学习等前沿技术，构建一个高精度、低延迟的疲劳状态识别方案。全文超过5000字，包含完整的代码实现、数据处理流程、模型训练策略以及部署方案。

第一部分：背景与相关技术

1.1 为什么选择脑电信号（EEG）？

在疲劳检测领域，已有的方法包括：

• 基于视觉的检测：通过摄像头捕捉眼睑闭合、打哈欠、头部姿态等。缺点是受光照、遮挡影响大，且只能检测到“已经出现”的疲劳表现，无法预测。

• 基于车辆行为：检测方向盘微调频率、车道偏离等。缺点是依赖于车辆动力学，反应滞后。

• 基于生理信号：包括心电（ECG）、肌电（EMG）、皮电反应（GSR）等。其中EEG直接反映大脑皮层的电活动，是疲劳状态的最直接、最敏感的生物标志物。

研究表明，当人从清醒状态进入疲劳状态时，EEG信号中的alpha频段（8-13Hz）功率增加，theta频段（4-8Hz）功率显著增加，而beta频段（13-30Hz）功率下降。这些变化往往出现在驾驶员表现出行为学疲劳前5-10分钟，为主动预警提供了宝贵的时间窗口。

1.2 深度学习的优势

传统的疲劳识别方法依赖于人工提取的频域特征（如功率谱密度、小波熵等），然后输入SVM、随机森林等分类器。这类方法存在明显局限：

• 人工特征设计高度依赖先验知识，无法适应个体差异。

• 忽略EEG信号中丰富的时空非线性关系。

而深度学习模型，尤其是卷积神经网络（CNN）、循环神经网络（RNN）、Transformer和图神经网络（GNN），可以从原始EEG信号中端到端地学习到与疲劳状态高度相关的表示。

1.3 本文技术栈

我们将在本项目中用到以下技术：

• EEG预处理：带通滤波、ICA去伪迹、重参考。

• 特征提取：短时傅里叶变换（STFT）生成时频谱图，或直接使用原始时序。

• 模型架构：

  • EEG-Conformer（融合CNN与Transformer）

  • DG-Transformer（动态图Transformer）

  • 基于对比学习的半监督模型（解决标注难问题）

• 评估指标：准确率、F1-score、ROC-AUC、实时性（延迟<1秒）。

第二部分：数据集介绍与预处理

2.1 公开数据集选择

我们使用目前最广泛认可的疲劳驾驶EEG数据集之一：SEED-VIG(SJTU Emotion EEG Dataset for Vigilance Estimation) 或DROZY数据集。为了方便演示，本文采用模拟数据集结构编写代码，但保证逻辑与实际完全一致。

一般情况下，数据集包含：

• 32或64通道的EEG信号（采用国际10-20系统）

• 采样率：200Hz或500Hz

• 标签：0=清醒，1=轻度疲劳，2=重度疲劳（或连续值反应时间）

2.2 预处理流水线

下面是完整的预处理代码，包括加载、滤波、去伪迹和分割。

python

import numpy as np import mne from scipy.signal import butter, filtfilt from sklearn.preprocessing import StandardScaler import torch from torch.utils.data import Dataset, DataLoader class EEGPreprocessor: def __init__(self, sampling_rate=200, lowcut=0.5, highcut=45, epoch_duration=2): """ 初始化EEG预处理器 :param sampling_rate: 采样率(Hz) :param lowcut: 带通下限 :param highcut: 带通上限 :param epoch_duration: 单段时长(秒) """ self.sampling_rate = sampling_rate self.lowcut = lowcut self.highcut = highcut self.epoch_samples = int(epoch_duration * sampling_rate) def butter_bandpass(self, data): """带通滤波器""" nyquist = 0.5 * self.sampling_rate low = self.lowcut / nyquist high = self.highcut / nyquist b, a = butter(4, [low, high], btype='band') return filtfilt(b, a, data, axis=-1) def remove_artifacts_ica(self, raw_data, n_components=20): """使用ICA去除眼电和肌电伪迹（简化版，实际需重构图）""" # 注意：此为标准流程示意，实际使用mne.preprocessing.ICA from mne.preprocessing import ICA ica = ICA(n_components=n_components, random_state=42) ica.fit(raw_data) # 假设人工标记EOG成分，这里略去具体索引 # raw_data = ica.apply(raw_data, exclude=[0]) return raw_data def segment_epochs(self, data, labels): """将连续信号切分为固定长度的epochs""" num_samples = data.shape[1] num_epochs = num_samples // self.epoch_samples epochs_data = [] epochs_labels = [] for i in range(num_epochs): start = i * self.epoch_samples end = start + self.epoch_samples epoch = data[:, start:end] # 标签采用多数投票 label_segment = labels[start:end] majority_label = np.bincount(label_segment.astype(int)).argmax() epochs_data.append(epoch) epochs_labels.append(majority_label) return np.array(epochs_data), np.array(epochs_labels) # 使用示例 preprocessor = EEGPreprocessor(sampling_rate=200, epoch_duration=2) # 模拟原始数据：64通道，10分钟时长，200Hz采样 fake_eeg = np.random.randn(64, 10*60*200) fake_labels = np.random.choice([0,1,2], size=10*60*200) # 模拟标签 # 滤波 filtered = preprocessor.butter_bandpass(fake_eeg) # 切片 epochs, epoch_labels = preprocessor.segment_epochs(filtered, fake_labels) print(f"切分后样本数: {epochs.shape}") # (300, 64, 400) 300个epochs, 64通道, 400时间点

2.3 数据增强策略

为了防止过拟合并提高泛化能力，我们采用EEG领域特有的数据增强方法：

python

class EEGAugmentation: @staticmethod def gaussian_noise(eeg, std=0.05): """添加高斯噪声""" noise = np.random.normal(0, std, eeg.shape) return eeg + noise @staticmethod def channel_dropout(eeg, drop_prob=0.1): """随机丢弃部分通道""" mask = np.random.binomial(1, 1-drop_prob, eeg.shape[0]) return eeg * mask[:, np.newaxis] @staticmethod def time_shift(eeg, max_shift=50): """时间轴随机偏移""" shift = np.random.randint(-max_shift, max_shift) if shift > 0: return np.concatenate([eeg[:, shift:], eeg[:, :shift]], axis=-1) else: return np.concatenate([eeg[:, shift:], eeg[:, :shift]], axis=-1) @staticmethod def mixup(eeg1, eeg2, label1, label2, alpha=0.2): """Mixup增强""" lam = np.random.beta(alpha, alpha) mixed_eeg = lam * eeg1 + (1 - lam) * eeg2 mixed_label = lam * label1 + (1 - lam) * label2 return mixed_eeg, mixed_label

第三部分：深度学习模型架构

3.1 基线模型：EEGNet

EEGNet是EEG领域最经典的紧凑型CNN模型，适合嵌入式部署。

python

import torch.nn as nn import torch.nn.functional as F class EEGNet(nn.Module): def __init__(self, n_channels=64, n_classes=3, time_points=400): super().__init__() # 第一层：时间卷积 self.conv1 = nn.Conv2d(1, 16, (1, 64), padding=(0, 32)) # 核大小为1x64 self.batchnorm1 = nn.BatchNorm2d(16) # 第二层：深度卷积（每个通道独立） self.depthwise = nn.Conv2d(16, 32, (n_channels, 1), groups=16) self.batchnorm2 = nn.BatchNorm2d(32) self.avgpool = nn.AvgPool2d((1, 4)) self.dropout = nn.Dropout(0.25) # 第三层：可分离卷积 self.separable = nn.Conv2d(32, 64, (1, 16)) self.batchnorm3 = nn.BatchNorm2d(64) self.avgpool2 = nn.AvgPool2d((1, 8)) # 分类层 self.fc = nn.Linear(64 * (time_points // 32), n_classes) def forward(self, x): # x shape: (batch, 1, channels, time) x = F.elu(self.conv1(x)) x = self.batchnorm1(x) x = F.elu(self.depthwise(x)) x = self.batchnorm2(x) x = self.avgpool(x) x = self.dropout(x) x = F.elu(self.separable(x)) x = self.batchnorm3(x) x = self.avgpool2(x) x = x.view(x.size(0), -1) x = self.fc(x) return x

3.2 现代架构：EEG-Conformer（CNN + Transformer）

EEG-Conformer结合了CNN的局部特征提取能力和Transformer的全局依赖建模能力，在多个EEG任务上达到SOTA。

python

class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len=1000): super().__init__() pe = torch.zeros(max_len, d_model) position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-np.log(10000.0) / d_model)) pe[:, 0::2] = torch.sin(position * div_term) pe[:, 1::2] = torch.cos(position * div_term) self.register_buffer('pe', pe) def forward(self, x): return x + self.pe[:x.size(1), :] class MultiHeadAttention(nn.Module): def __init__(self, d_model, n_heads): super().__init__() self.attention = nn.MultiheadAttention(d_model, n_heads, batch_first=True) def forward(self, x): attn_output, _ = self.attention(x, x, x) return attn_output class EEGConformer(nn.Module): def __init__(self, n_channels=64, n_classes=3, time_points=400, d_model=128, n_heads=8, depth=4): super().__init__() # CNN前端 self.cnn_block = nn.Sequential( nn.Conv2d(1, 32, (1, 25), padding=(0, 12)), nn.BatchNorm2d(32), nn.ELU(), nn.Conv2d(32, 64, (n_channels, 1)), nn.BatchNorm2d(64), nn.ELU(), nn.AvgPool2d((1, 8)), nn.Dropout(0.3) ) # 投影层将CNN输出映射到d_model self.projection = nn.Linear(64 * (time_points // 8), d_model) # Transformer编码器 self.pos_encoder = PositionalEncoding(d_model) encoder_layer = nn.TransformerEncoderLayer(d_model=d_model, nhead=n_heads, dim_feedforward=256, dropout=0.1, batch_first=True) self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=depth) # 全局平均池化 + 分类头 self.classifier = nn.Linear(d_model, n_classes) def forward(self, x): # x: (batch, 1, channels, time) cnn_out = self.cnn_block(x) # (batch, 64, 1, time_compressed) batch_size = cnn_out.size(0) cnn_out = cnn_out.view(batch_size, 64, -1) # (batch, 64, time_compressed) cnn_out = cnn_out.permute(0, 2, 1) # (batch, time_compressed, 64) proj = self.projection(cnn_out) # (batch, time_compressed, d_model) proj = self.pos_encoder(proj) trans_out = self.transformer(proj) # (batch, time_compressed, d_model) # 全局平均池化 global_feat = trans_out.mean(dim=1) # (batch, d_model) logits = self.classifier(global_feat) return logits

3.3 创新架构：图动态注意力网络（DG-Transformer）

EEG通道天然具有空间拓扑结构，使用图神经网络可以显式建模通道间的功能连接。

python

import torch_geometric.nn as pyg_nn from torch_geometric.data import Data, Batch class GraphStructure: """构建EEG通道的邻接矩阵（基于10-20系统的空间距离）""" @staticmethod def get_adjacency(channel_names, threshold=0.3): """简化版：假设已知通道间欧氏距离归一化后大于threshold的连接""" # 实际应用中需提供真实坐标 num_channels = len(channel_names) adj = np.random.rand(num_channels, num_channels) # 模拟距离 adj = (adj < threshold).astype(float) np.fill_diagonal(adj, 0) return adj class DynamicGraphTransformer(nn.Module): def __init__(self, n_channels=64, n_classes=3, d_model=128, n_heads=4): super().__init__() self.node_embedding = nn.Linear(200, d_model) # 时间点特征嵌入 # 图卷积层 self.gcn1 = pyg_nn.GCNConv(d_model, d_model) self.gcn2 = pyg_nn.GCNConv(d_model, d_model) # 跨时间步的Transformer self.temporal_transformer = nn.TransformerEncoder( nn.TransformerEncoderLayer(d_model=d_model, nhead=n_heads, batch_first=True), num_layers=3 ) self.classifier = nn.Linear(d_model, n_classes) def forward(self, x, edge_index): # x: (batch, time_points, channels) -> 需转为图序列 batch_size, time_points, n_channels = x.shape # 对每个时间步独立做图卷积 node_features = x.permute(0, 2, 1) # (batch, channels, time_points) # 这里简化为：将每个通道的完整时间序列作为节点特征 node_emb = self.node_embedding(node_features) # (batch, channels, d_model) # 对每个样本构建图 graph_embeddings = [] for b in range(batch_size): # 构建PyG图数据 graph_data = Data(x=node_emb[b], edge_index=edge_index) g_out = self.gcn1(graph_data.x, graph_data.edge_index) g_out = F.elu(g_out) g_out = self.gcn2(g_out, graph_data.edge_index) graph_embeddings.append(g_out) graph_batch = torch.stack(graph_embeddings, dim=0) # (batch, channels, d_model) # 时间维度转换：把通道看作序列 temporal_out = graph_batch.permute(0, 2, 1) # (batch, d_model, channels) temporal_out = self.temporal_transformer(temporal_out) # 全局池化 global_feat = temporal_out.mean(dim=1) logits = self.classifier(global_feat) return logits

3.4 半监督对比学习框架（SimCLR for EEG）

由于标注疲劳标签成本高昂，我们可以采用对比学习预训练，再用少量标注数据微调。

python

class ContrastiveEEGEncoder(nn.Module): def __init__(self, base_encoder=EEGConformer, projection_dim=128): super().__init__() self.encoder = base_encoder(n_classes=projection_dim) # 输出128维特征 self.projection_head = nn.Sequential( nn.Linear(projection_dim, projection_dim), nn.ReLU(), nn.Linear(projection_dim, projection_dim) ) def forward(self, x): features = self.encoder(x) # 前向传播得到特征 projection = self.projection_head(features) return F.normalize(projection, dim=-1) def contrastive_loss(z_i, z_j, temperature=0.5): """NT-Xent loss""" batch_size = z_i.size(0) z = torch.cat([z_i, z_j], dim=0) similarity = torch.matmul(z, z.T) / temperature mask = torch.eye(2*batch_size, dtype=torch.bool).to(z.device) similarity = similarity.masked_fill(mask, -1e9) positive_sim = torch.cat([torch.diag(similarity, batch_size), torch.diag(similarity, -batch_size)], dim=0) numerator = torch.exp(positive_sim) denominator = torch.exp(similarity).sum(dim=1) loss = -torch.log(numerator / denominator).mean() return loss # 使用示例：同一epoch的两种增强视图 def train_contrastive(model, dataloader, optimizer, epochs=50): for epoch in range(epochs): for batch in dataloader: x, _ = batch # 不需要标签 # 生成两个增强视图 x_aug1 = EEGAugmentation.gaussian_noise(x) x_aug2 = EEGAugmentation.time_shift(x) z1 = model(x_aug1) z2 = model(x_aug2) loss = contrastive_loss(z1, z2) optimizer.zero_grad() loss.backward() optimizer.step() print(f"Epoch {epoch}, Loss: {loss.item():.4f}")

第四部分：完整训练流程与评估

4.1 数据加载与划分

python

class FatigueDataset(Dataset): def __init__(self, eeg_data, labels, transform=None): self.eeg_data = torch.FloatTensor(eeg_data) # (n_samples, channels, time) self.labels = torch.LongTensor(labels) self.transform = transform def __len__(self): return len(self.labels) def __getitem__(self, idx): x = self.eeg_data[idx] y = self.labels[idx] if self.transform: x = self.transform(x) # 增加通道维度: (channels, time) -> (1, channels, time) x = x.unsqueeze(0) return x, y # 假设已有处理好的epochs和labels # train_loader, val_loader, test_loader def create_dataloaders(epochs, labels, batch_size=64, val_ratio=0.2, test_ratio=0.1): total = len(epochs) indices = np.random.permutation(total) test_end = int(total * test_ratio) val_end = test_end + int(total * val_ratio) test_indices = indices[:test_end] val_indices = indices[test_end:val_end] train_indices = indices[val_end:] train_dataset = FatigueDataset(epochs[train_indices], labels[train_indices]) val_dataset = FatigueDataset(epochs[val_indices], labels[val_indices]) test_dataset = FatigueDataset(epochs[test_indices], labels[test_indices]) train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True) val_loader = DataLoader(val_dataset, batch_size=batch_size) test_loader = DataLoader(test_dataset, batch_size=batch_size) return train_loader, val_loader, test_loader

4.2 训练函数（支持混合精度）

python

from torch.cuda.amp import autocast, GradScaler def train_model(model, train_loader, val_loader, epochs=50, lr=1e-3, device='cuda'): model = model.to(device) criterion = nn.CrossEntropyLoss(label_smoothing=0.1) optimizer = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=1e-4) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=epochs) scaler = GradScaler() best_val_acc = 0.0 for epoch in range(epochs): # 训练阶段 model.train() train_loss = 0.0 train_correct = 0 for x, y in train_loader: x, y = x.to(device), y.to(device) optimizer.zero_grad() with autocast(): outputs = model(x) loss = criterion(outputs, y) scaler.scale(loss).backward() scaler.step(optimizer) scaler.update() train_loss += loss.item() * x.size(0) _, preds = torch.max(outputs, 1) train_correct += torch.sum(preds == y).item() train_acc = train_correct / len(train_loader.dataset) train_loss = train_loss / len(train_loader.dataset) # 验证阶段 model.eval() val_correct = 0 val_loss = 0.0 with torch.no_grad(): for x, y in val_loader: x, y = x.to(device), y.to(device) outputs = model(x) loss = criterion(outputs, y) val_loss += loss.item() * x.size(0) _, preds = torch.max(outputs, 1) val_correct += torch.sum(preds == y).item() val_acc = val_correct / len(val_loader.dataset) val_loss = val_loss / len(val_loader.dataset) scheduler.step() print(f"Epoch {epoch+1}/{epochs} - Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.4f}, " f"Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.4f}") if val_acc > best_val_acc: best_val_acc = val_acc torch.save(model.state_dict(), 'best_fatigue_model.pth') return model

4.3 评估与分析

python

from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score import matplotlib.pyplot as plt import seaborn as sns def evaluate(model, test_loader, device='cuda'): model.eval() all_preds = [] all_labels = [] all_probs = [] with torch.no_grad(): for x, y in test_loader: x = x.to(device) outputs = model(x) probs = F.softmax(outputs, dim=1) _, preds = torch.max(outputs, 1) all_preds.extend(preds.cpu().numpy()) all_labels.extend(y.numpy()) all_probs.extend(probs.cpu().numpy()) print(classification_report(all_labels, all_preds, target_names=['清醒', '轻度疲劳', '重度疲劳'])) # 混淆矩阵 cm = confusion_matrix(all_labels, all_preds) plt.figure(figsize=(8,6)) sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=['清醒','轻度疲劳','重度疲劳'], yticklabels=['清醒','轻度疲劳','重度疲劳']) plt.title('Confusion Matrix') plt.ylabel('True Label') plt.xlabel('Predicted Label') plt.show() # ROC曲线 (一对多) from sklearn.preprocessing import label_binarize y_onehot = label_binarize(all_labels, classes=[0,1,2]) roc_auc = roc_auc_score(y_onehot, np.array(all_probs), multi_class='ovr') print(f"Macro ROC-AUC: {roc_auc:.4f}") return all_preds, all_labels

第五部分：实时检测系统部署

5.1 流式推理架构

在实际车辆中，我们需要一个低延迟的流式处理系统。下面的代码实现了一个滑动窗口实时推理器。

python

import collections import time class RealTimeFatigueDetector: def __init__(self, model, preprocessor, window_seconds=2, step_seconds=1, buffer_seconds=30): self.model = model self.preprocessor = preprocessor self.window_samples = window_seconds * preprocessor.sampling_rate self.step_samples = step_seconds * preprocessor.sampling_rate self.buffer = collections.deque(maxlen=buffer_seconds * preprocessor.sampling_rate) self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') self.model.to(self.device) self.model.eval() def update(self, new_samples): """接收新的EEG数据块（形状: channels x new_samples）""" self.buffer.extend(new_samples.T) # 假设输入按时间排列 def predict(self): """对当前buffer中的最新窗口预测疲劳状态""" if len(self.buffer) < self.window_samples: return None # 取最后window_samples个样本 window = np.array(list(self.buffer))[-self.window_samples:] window = window.T # 变为 (channels, time) # 预处理 window_filtered = self.preprocessor.butter_bandpass(window) # 添加batch和channel维度 input_tensor = torch.FloatTensor(window_filtered).unsqueeze(0).unsqueeze(0) # (1,1,channels,time) input_tensor = input_tensor.to(self.device) with torch.no_grad(): logits = self.model(input_tensor) probs = F.softmax(logits, dim=1) pred = torch.argmax(probs, dim=1).item() confidence = probs[0][pred].item() return {'state': pred, 'confidence': confidence, 'state_name': ['清醒','轻度疲劳','重度疲劳'][pred]} def run_realtime_loop(self, data_stream_generator): """模拟实时数据流循环""" for chunk in data_stream_generator: self.update(chunk) result = self.predict() if result and result['state'] >= 1: # 轻度或重度疲劳 print(f"[{time.time()}] 警告！疲劳状态: {result['state_name']}, 置信度: {result['confidence']:.2f}") # 触发报警 self.trigger_alert(result) def trigger_alert(self, result): """集成车辆警报系统（蜂鸣器、座椅振动等）""" # 实际项目中可调用硬件接口 if result['state'] == 2: # 重度疲劳 print(">>>>>>>> 立即停车提醒！<<<<<<<<")