用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]关键数据结构解析数据层级描述维度信息Data23名被试的主容器(23,)EEG每个被试的脑电记录包含baseline/stimulibaseline静息状态基准信号(7808,) per videostimuli视频刺激期间的脑电信号(25472, 14)ScoreValence效价评分1-5(18,) per subjectScoreArousal唤醒度评分1-5(18,) per subject注意MATLAB的.mat文件在Python中会被转换为numpy的ndarray但保留了MATLAB的1-based索引习惯因此需要通过[0,0]访问最外层数据。2. 数据提取核心代码实现2.1 电极信号提取函数我们设计一个支持批量处理的提取器避免原始代码中的硬编码问题def extract_eeg_signals(raw, participants23, videos18, electrodes14): 三维张量提取参与者×视频×电极信号 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)提示相比MinMaxScalerRobustScaler对脑电信号中的突发噪声更具鲁棒性3.2 数据分割策略将长时序信号切分为适合神经网络的片段def segment_signal(signal, window_size128, overlap0.5): 滑动窗口分割时序信号 segments [] step int(window_size * (1 - overlap)) for start in range(0, len(signal)-window_size, step): segments.append(signal[start:startwindow_size]) return np.array(segments)4. 工程化存储方案4.1 按参与者保存NPY文件def save_participant_npy(eeg_data, labels, output_dir): os.makedirs(output_dir, exist_okTrue) 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_participants23): 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_pickleTrue).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_dtypeTrue, struct_as_recordFalse, squeeze_meTrue, matlab_compatibleFalse, variable_names[DREAMER], appendmatFalse) return mmap[DREAMER]5.2 并行化提取加速使用joblib加速多参与者数据处理from joblib import Parallel, delayed def parallel_extract(raw, n_jobs4): def process_one(p): return extract_eeg_signals(raw, participants1, startp) results Parallel(n_jobsn_jobs)( delayed(process_one)(p) for p in range(23)) return np.concatenate(results, axis0)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_pickleTrue).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_namesNone): if ch_names is None: ch_names [fEEG-{i} for i in range(14)] info mne.create_info(ch_names, sfreq128, ch_typeseeg) evoked mne.EvokedArray(eeg_data.mean(axis1), info) return evoked.plot_topomap(times[0.1, 0.2, 0.3], size3, show_namesTrue)在完成所有数据处理流程后建议建立数据版本控制。我们团队在实际项目中发现为每处理阶段打上git标签能极大方便实验复现。例如data_v1_raw/ data_v2_processed/ data_v3_normalized/