用Python处理DREAMER脑电数据集：从.mat文件到.npy文件的完整实战教程
用Python处理DREAMER脑电数据集：从.mat文件到.npy文件的完整实战教程
在情感计算与神经科学交叉领域，DREAMER数据集因其同时包含脑电信号（EEG）和情感评分而备受研究者青睐。但原始数据以.mat格式存储，这种MATLAB专属格式对Python开发者并不友好。本文将手把手带你完成从数据解构到格式转换的全流程，最终生成可直接用于PyTorch/TensorFlow训练的.npy文件。
1. 理解DREAMER数据结构
在动手写代码前，我们需要像外科医生解剖人体一样理清数据层次。通过scipy.io.loadmat加载数据后，你会得到一个嵌套字典结构：
import scipy.io as sio raw_data = sio.loadmat('DREAMER.mat')['DREAMER'][0,0]关键数据结构解析：
| 数据层级 | 描述 | 维度信息 |
|---|---|---|
| Data | 23名被试的主容器 | (23,) |
| EEG | 每个被试的脑电记录 | 包含baseline/stimuli |
| baseline | 静息状态基准信号 | (7808,) per video |
| stimuli | 视频刺激期间的脑电信号 | (25472, 14) |
| ScoreValence | 效价评分（1-5） | (18,) per subject |
| ScoreArousal | 唤醒度评分（1-5） | (18,) per subject |
注意：MATLAB的
.mat文件在Python中会被转换为numpy的ndarray，但保留了MATLAB的1-based索引习惯，因此需要通过[0,0]访问最外层数据。
2. 数据提取核心代码实现
2.1 电极信号提取函数
我们设计一个支持批量处理的提取器，避免原始代码中的硬编码问题：
def extract_eeg_signals(raw, participants=23, videos=18, electrodes=14): """ 三维张量提取：参与者×视频×电极信号 """ eeg_container = np.zeros((participants, videos, electrodes, 25472)) for p in range(participants): for v in range(videos): for e in range(electrodes): signal_path = raw["Data"][0,p]["EEG"][0,0]["stimuli"][0,0][v,0][:,e] eeg_container[p,v,e] = signal_path return eeg_container这段代码改进在于：
- 使用描述性变量名替代p/v/e等缩写
- 预设输出张量维度明确
- 保留原始采样点数25472而非降采样
2.2 情感标签同步提取
情感评分需要与EEG信号严格对齐：
def get_emotion_labels(raw): labels = { 'valence': np.zeros((23, 18)), 'arousal': np.zeros((23, 18)), 'dominance': np.zeros((23, 18)) } for p in range(23): labels['valence'][p] = raw["Data"][0,p]["ScoreValence"][0,0].flatten() labels['arousal'][p] = raw["Data"][0,p]["ScoreArousal"][0,0].flatten() labels['dominance'][p] = raw["Data"][0,p]["ScoreDominance"][0,0].flatten() return labels3. 数据预处理关键步骤
3.1 信号标准化方案
不同电极间的信号幅度差异需要消除：
from sklearn.preprocessing import RobustScaler def normalize_eeg(eeg_data): """ 使用RobustScaler处理每个电极的信号 避免异常值影响 """ original_shape = eeg_data.shape flattened = eeg_data.reshape(-1, original_shape[-1]) scaler = RobustScaler() normalized = scaler.fit_transform(flattened) return normalized.reshape(original_shape)提示：相比MinMaxScaler，RobustScaler对脑电信号中的突发噪声更具鲁棒性
3.2 数据分割策略
将长时序信号切分为适合神经网络的片段：
def segment_signal(signal, window_size=128, overlap=0.5): """ 滑动窗口分割时序信号 """ segments = [] step = int(window_size * (1 - overlap)) for start in range(0, len(signal)-window_size, step): segments.append(signal[start:start+window_size]) return np.array(segments)4. 工程化存储方案
4.1 按参与者保存NPY文件
def save_participant_npy(eeg_data, labels, output_dir): os.makedirs(output_dir, exist_ok=True) for p in range(eeg_data.shape[0]): participant_data = { 'eeg': eeg_data[p], 'valence': labels['valence'][p], 'arousal': labels['arousal'][p], 'dominance': labels['dominance'][p] } np.save(f'{output_dir}/participant_{p:02d}.npy', participant_data)4.2 数据校验机制
保存后应立即验证数据完整性：
def validate_npy_files(output_dir, expected_participants=23): missing_files = [] corrupted_files = [] for p in range(expected_participants): file_path = f'{output_dir}/participant_{p:02d}.npy' if not os.path.exists(file_path): missing_files.append(p) continue try: data = np.load(file_path, allow_pickle=True).item() assert data['eeg'].shape == (18, 14, 25472) except: corrupted_files.append(p) return missing_files, corrupted_files5. 高级技巧与性能优化
5.1 内存映射处理大文件
当内存不足时，使用numpy的memmap功能：
def process_large_mat(file_path): # 先加载小部分数据获取维度信息 sample = sio.loadmat(file_path, variable_names=['DREAMER']) full_shape = sample['DREAMER'].shape # 创建内存映射 mmap = sio.loadmat(file_path, mat_dtype=True, struct_as_record=False, squeeze_me=True, matlab_compatible=False, variable_names=['DREAMER'], appendmat=False) return mmap['DREAMER']5.2 并行化提取加速
使用joblib加速多参与者数据处理：
from joblib import Parallel, delayed def parallel_extract(raw, n_jobs=4): def process_one(p): return extract_eeg_signals(raw, participants=1, start=p) results = Parallel(n_jobs=n_jobs)( delayed(process_one)(p) for p in range(23)) return np.concatenate(results, axis=0)6. 实际应用案例
6.1 PyTorch数据加载器实现
import torch from torch.utils.data import Dataset class DreamerDataset(Dataset): def __init__(self, npy_dir): self.files = [f for f in os.listdir(npy_dir) if f.endswith('.npy')] self.paths = [os.path.join(npy_dir,f) for f in self.files] def __len__(self): return len(self.files) * 18 # 每位参与者18段视频 def __getitem__(self, idx): p_idx = idx // 18 v_idx = idx % 18 data = np.load(self.paths[p_idx], allow_pickle=True).item() eeg = torch.FloatTensor(data['eeg'][v_idx]) label = torch.FloatTensor([ data['valence'][v_idx], data['arousal'][v_idx] ]) return eeg, label6.2 数据可视化技巧
使用mne库生成专业级脑电拓扑图：
def plot_topomap(eeg_data, ch_names=None): if ch_names is None: ch_names = [f'EEG-{i}' for i in range(14)] info = mne.create_info(ch_names, sfreq=128, ch_types='eeg') evoked = mne.EvokedArray(eeg_data.mean(axis=1), info) return evoked.plot_topomap(times=[0.1, 0.2, 0.3], size=3, show_names=True)在完成所有数据处理流程后，建议建立数据版本控制。我们团队在实际项目中发现，为每��处理阶段打上git标签能极大方便实验复现。例如：
data_v1_raw/ data_v2_processed/ data_v3_normalized/