1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118
| import numpy as np from scipy.signal import welch from sklearn.preprocessing import StandardScaler
class UnifiedEEGFeatureExtractor: """ 统一EEG特征提取器 目标: 提取跨受试者通用的驾驶员状态特征 核心思路: 1. 相对功率(而非绝对功率) → 消除个体幅度差异 2. 比率特征 → 消除基线差异 3. 非线性特征 → 捕捉跨个体共性 """ def __init__(self, fs=250): self.fs = fs self.bands = { "delta": (0.5, 4), "theta": (4, 8), "alpha": (8, 13), "beta": (13, 30), "gamma": (30, 45) } def extract_features(self, eeg_signal): """ 统一特征提取 Args: eeg_signal: EEG信号 (N_channels, N_samples) Returns: features: 统一特征向量 """ n_channels, n_samples = eeg_signal.shape features = {} for ch in range(n_channels): signal = eeg_signal[ch] freqs, psd = welch(signal, self.fs, nperseg=512) abs_power = {} for band, (f_low, f_high) in self.bands.items(): mask = (freqs >= f_low) & (freqs < f_high) abs_power[band] = np.trapz(psd[mask], freqs[mask]) total_power = sum(abs_power.values()) + 1e-10 for band in self.bands: features[f"ch{ch}_{band}_rel"] = abs_power[band] / total_power features[f"ch{ch}_theta_alpha"] = abs_power["theta"] / (abs_power["alpha"] + 1e-10) features[f"ch{ch}_beta_alpha"] = abs_power["beta"] / (abs_power["alpha"] + 1e-10) features[f"ch{ch}_theta_beta"] = abs_power["theta"] / (abs_power["beta"] + 1e-10) features[f"ch{ch}_alpha_beta"] = abs_power["alpha"] / (abs_power["beta"] + 1e-10) features[f"ch{ch}_mean"] = np.mean(signal) features[f"ch{ch}_std"] = np.std(signal) features[f"ch{ch}_skewness"] = float(np.mean(((signal - np.mean(signal)) / (np.std(signal) + 1e-10))**3)) features[f"ch{ch}_kurtosis"] = float(np.mean(((signal - np.mean(signal)) / (np.std(signal) + 1e-10))**4) - 3) d1 = np.diff(signal) d2 = np.diff(d1) var0 = np.var(signal) var1 = np.var(d1) var2 = np.var(d2) features[f"ch{ch}_activity"] = var0 features[f"ch{ch}_mobility"] = np.sqrt(var1 / (var0 + 1e-10)) features[f"ch{ch}_complexity"] = np.sqrt(var2 / (var1 + 1e-10)) / \ (np.sqrt(var1 / (var0 + 1e-10)) + 1e-10) return np.array(list(features.values())) def normalize_cross_subject(self, features): """跨受试者标准化""" scaler = StandardScaler() return scaler.fit_transform(features.reshape(1, -1)).flatten()
extractor = UnifiedEEGFeatureExtractor(fs=250)
np.random.seed(42) n_channels = 14 n_samples = 2500
eeg_rested = np.random.normal(0, 50, (n_channels, n_samples))
for ch in range(n_channels): t = np.arange(n_samples) / 250 eeg_rested[ch] += 30 * np.sin(2 * np.pi * 10 * t)
eeg_tired = np.random.normal(0, 40, (n_channels, n_samples))
for ch in range(n_channels): t = np.arange(n_samples) / 250 eeg_tired[ch] += 25 * np.sin(2 * np.pi * 5 * t)
feat_rested = extractor.extract_features(eeg_rested) feat_tired = extractor.extract_features(eeg_tired)
print(f"特征维度: {len(feat_rested)}") print(f"休息态 vs 疲劳态差异: {np.mean(np.abs(feat_rested - feat_tired)):.4f}")
|