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 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225
| import numpy as np from scipy import signal from scipy.stats import linregress
def extract_hrv_features(rr_intervals: np.ndarray, sampling_rate: float = 4.0) -> dict: """ 从RR间期序列提取HRV时域和频域特征 论文Section 2.3描述的HRV分析方法实现 Args: rr_intervals: RR间期序列(秒),shape=(N,) sampling_rate: 重采样率,默认4 Hz Returns: features: HRV特征字典 Example: >>> rr = np.array([0.8, 0.82, 0.79, 0.85, 0.81, 0.78, 0.83, 0.80]) >>> features = extract_hrv_features(rr) >>> print(f"RMSSD: {features['rmssd']:.4f}s") """ diff_rr = np.diff(rr_intervals) rmssd = np.sqrt(np.mean(diff_rr ** 2)) sdnn = np.std(rr_intervals, ddof=1) nn50 = np.sum(np.abs(diff_rr) > 0.05) pnn50 = (nn50 / len(diff_rr)) * 100 if len(diff_rr) > 0 else 0 t_original = np.cumsum(rr_intervals) t_resampled = np.arange(t_original[0], t_original[-1], 1.0 / sampling_rate) rr_resampled = np.interp(t_resampled, t_original, rr_intervals) rr_detrended = signal.detrend(rr_resampled, type='linear') freqs, psd = signal.welch(rr_detended, fs=sampling_rate, nperseg=256) lf_mask = (freqs >= 0.04) & (freqs < 0.15) hf_mask = (freqs >= 0.15) & (freqs < 0.40) lf_power = np.trapz(psd[lf_mask], freqs[lf_mask]) hf_power = np.trapz(psd[hf_mask], freqs[hf_mask]) lf_hf_ratio = lf_power / (hf_power + 1e-10) total_power = np.trapz(psd, freqs) return { 'rmssd': rmssd, 'sdnn': sdnn, 'pnn50': pnn50, 'lf_power': lf_power, 'hf_power': hf_power, 'lf_hf_ratio': lf_hf_ratio, 'total_power': total_power }
def extract_breathing_rate(ecg_signal: np.ndarray, fs: float = 500.0) -> float: """ 从ECG信号提取呼吸率 基于ECG基线漂移中的呼吸调制成分 Args: ecg_signal: ECG信号,shape=(N,) fs: 采样率 Returns: breathing_rate: 估计呼吸率(次/分钟) """ lowcut = 0.1 highcut = 0.5 nyq = fs / 2 b, a = signal.butter(4, [lowcut/nyq, highcut/nyq], btype='band') breathing_signal = signal.filtfilt(b, a, ecg_signal) freqs = np.fft.rfftfreq(len(breathing_signal), d=1/fs) fft_mag = np.abs(np.fft.rfft(breathing_signal)) mask = (freqs >= 0.1) & (freqs <= 0.5) peak_idx = np.argmax(fft_mag[mask]) peak_freq = freqs[mask][peak_idx] breathing_rate = peak_freq * 60 return breathing_rate
def mf_multimodal_fusion(hrv_features: dict, breathing_rate: float, blink_duration: float, perclos: float, n_back_accuracy: float) -> float: """ 多模态MF融合分类器 论文核心结论:HRV和呼吸率权重最高 Args: hrv_features: HRV特征字典 breathing_rate: 呼吸率(次/分钟) blink_duration: 平均眨眼时长(秒) perclos: PERCLOS百分比 n_back_accuracy: n-back任务准确率 Returns: mf_score: 精神疲劳评分 0-1(越高越疲劳) """ weights = { 'hrv_rmssd': 0.30, 'hrv_lf_hf': 0.25, 'breathing_rate': 0.20, 'blink_duration': 0.10, 'perclos': 0.10, 'n_back_accuracy': 0.05 } rmssd_score = 1.0 / (1.0 + np.exp(hrv_features['rmssd'] * 10 - 3)) lf_hf_score = 1.0 / (1.0 + np.exp(hrv_features['lf_hf_ratio'] - 2)) br_score = 1.0 / (1.0 + np.exp(breathing_rate - 12)) perclos_score = min(perclos / 30.0, 1.0) blink_score = min(blink_duration / 0.5, 1.0) nback_score = 1.0 - n_back_accuracy mf_score = ( weights['hrv_rmssd'] * rmssd_score + weights['hrv_lf_hf'] * lf_hf_score + weights['breathing_rate'] * br_score + weights['blink_duration'] * blink_score + weights['perclos'] * perclos_score + weights['n_back_accuracy'] * nback_score ) return min(mf_score, 1.0)
if __name__ == "__main__": np.random.seed(42) print("=" * 60) print("铁路驾驶员精神疲劳多传感器分析") print("论文复现: Bosch et al., Applied Ergonomics, 2026") print("=" * 60) print("\n--- 基线驾驶(正常状态)---") rr_baseline = np.random.normal(0.85, 0.03, 600) ecg_baseline = np.random.randn(300000) * 0.5 features_baseline = extract_hrv_features(rr_baseline) br_baseline = extract_breathing_rate(ecg_baseline, fs=500) print(f"HRV RMSSD: {features_baseline['rmssd']:.4f}s") print(f"HRV SDNN: {features_baseline['sdnn']:.4f}s") print(f"HRV LF/HF: {features_baseline['lf_hf_ratio']:.2f}") print(f"呼吸率: {br_baseline:.1f} 次/分钟") print("\n--- 1小时n-back任务后(疲劳状态)---") rr_fatigue = np.random.normal(0.90, 0.015, 600) ecg_fatigue = np.random.randn(300000) * 0.3 features_fatigue = extract_hrv_features(rr_fatigue) br_fatigue = extract_breathing_rate(ecg_fatigue, fs=500) print(f"HRV RMSSD: {features_fatigue['rmssd']:.4f}s (↓{((features_baseline['rmssd']-features_fatigue['rmssd'])/features_baseline['rmssd']*100):.1f}%)") print(f"HRV SDNN: {features_fatigue['sdnn']:.4f}s (↓{((features_baseline['sdnn']-features_fatigue['sdnn'])/features_baseline['sdnn']*100):.1f}%)") print(f"HRV LF/HF: {features_fatigue['lf_hf_ratio']:.2f} (↓{((features_baseline['lf_hf_ratio']-features_fatigue['lf_hf_ratio'])/features_baseline['lf_hf_ratio']*100):.1f}%)") print(f"呼吸率: {br_fatigue:.1f} 次/分钟") print("\n--- 多模态MF融合评分 ---") mf_baseline = mf_multimodal_fusion( features_baseline, br_baseline, blink_duration=0.15, perclos=5.0, n_back_accuracy=0.95 ) mf_fatigue = mf_multimodal_fusion( features_fatigue, br_fatigue, blink_duration=0.25, perclos=12.0, n_back_accuracy=0.82 ) print(f"基线 MF评分: {mf_baseline:.3f}") print(f"疲劳 MF评分: {mf_fatigue:.3f}") print(f"评分变化: +{((mf_fatigue-mf_baseline)/mf_baseline*100):.1f}%") print("\n" + "=" * 60) print("仿真器 vs 真实环境 传感器可靠性对比") print("=" * 60) comparison = { "传感器": ["HRV (RMSSD)", "呼吸率", "EEG frontal theta", "EDA", "眨眼时长", "n-back准确率"], "仿真器(n=14)": ["✅ 一致", "✅ 一致", "✅ 一致", "⚠️ 部分", "✅ 一致", "✅ 一致"], "真实铁路(n=6)": ["✅ 一致", "✅ 一致", "❌ 不稳定", "❌ 不稳定", "⚠️ 部分", "✅ 一致"], "推荐用于DMS": ["✅ 是", "✅ 是", "❌ 需验证", "❌ 不推荐", "⚠️ 辅助", "⚠ 辅助"] } for i, sensor in enumerate(comparison["传感器"]): print(f" {sensor:20s} | 仿真器: {comparison['仿真器(n=14)'][i]:8s} | 真实: {comparison['真实铁路(n=6)'][i]:8s} | 推荐: {comparison['推荐用于DMS'][i]}")
|