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
| import numpy as np
from typing import Tuple, Dict
class HRVFatigueDetector:
"""
基于 HRV 的铁路/汽车驾驶员疲劳检测器
论文发现：
- LF/HF ratio 在疲劳后显著降低（副交感神经主导）
- 呼吸率在疲劳后降低
- 这两个指标在模拟器和真实环境中都鲁棒
适用场景：
- 列车驾驶室（振动环境）
- 汽车驾驶室（已有座椅/方向盘电极）
- 船舶驾驶台（类似振动环境）
硬件配置：
| 组件 | 型号 | 参数 | 部署 |
|------|------|------|------|
| ECG 电极 | 座椅集成 | 3导联, 250Hz | 驾驶员座椅 |
| 呼吸传感器 | 压电带 | 1Hz | 安全带 |
| 处理器 | 车载计算 | ARM Cortex-A78 | 驾驶舱电子柜 |
"""
def __init__(self, sampling_rate: int = 250):
self.fs = sampling_rate
def extract_rr_intervals(self, ecg_signal: np.ndarray) -> np.ndarray:
"""
从 ECG 信号提取 RR 间期
Args:
ecg_signal: ECG 信号, shape=(N,)
Returns:
rr_intervals: RR 间期序列, shape=(M,)
"""
from scipy.signal import find_peaks
peaks, _ = find_peaks(ecg_signal, distance=self.fs*0.5, height=0.6*np.max(ecg_signal))
rr_intervals = np.diff(peaks) / self.fs
return rr_intervals
def compute_hrv_features(self, rr: np.ndarray) -> Dict[str, float]:
"""
计算 HRV 时域和频域特征
Args:
rr: RR 间期序列 (秒)
Returns:
features: HRV 特征字典
"""
rmssd = np.sqrt(np.mean(np.diff(rr) ** 2))
sdnn = np.std(rr)
mean_hr = 60 / np.mean(rr)
from scipy.signal import welch
freqs, psd = welch(rr * 1000, fs=1.0/np.mean(np.diff(rr)), nperseg=len(rr))
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-8)
total_power = lf_power + hf_power
lf_nu = lf_power / (total_power + 1e-8) * 100
hf_nu = hf_power / (total_power + 1e-8) * 100
return {
"RMSSD_ms": rmssd * 1000,
"SDNN_ms": sdnn * 1000,
"mean_HR_bpm": mean_hr,
"LF_HF_ratio": lf_hf_ratio,
"LF_nu": lf_nu,
"HF_nu": hf_nu,
"total_power_ms2": total_power,
}
def detect_fatigue(self, baseline_hrv: Dict, current_hrv: Dict) -> Dict:
"""
疲劳检测（对比基线）
论文关键发现：
- 疲劳后 LF/HF ratio 降低（副交感主导）
- 疲劳后 RMSSD 增加（副交感增强）
- 疲劳后呼吸率降低
Args:
baseline_hrv: 基线 HRV 特征
current_hrv: 当前 HRV 特征
Returns:
result: 疲劳评估结果
"""
lf_hf_change = (current_hrv["LF_HF_ratio"] - baseline_hrv["LF_HF_ratio"]) / baseline_hrv["LF_HF_ratio"]
rmssd_change = (current_hrv["RMSSD_ms"] - baseline_hrv["RMSSD_ms"]) / baseline_hrv["RMSSD_ms"]
fatigue_score = 0.5 * max(0, -lf_hf_change) + 0.5 * max(0, rmssd_change)
fatigue_score = min(1.0, fatigue_score)
if fatigue_score < 0.15:
level = "alert"
elif fatigue_score < 0.35:
level = "mild_fatigue"
else:
level = "severe_fatigue"
return {
"fatigue_score": round(fatigue_score, 3),
"level": level,
"lf_hf_change_pct": round(lf_hf_change * 100, 1),
"rmssd_change_pct": round(rmssd_change * 100, 1),
"confidence": "high" if abs(lf_hf_change) > 0.1 else "medium",
}
if __name__ == "__main__":
detector = HRVFatigueDetector(sampling_rate=250)
np.random.seed(42)
t_alert = np.arange(0, 60, 1/250)
hr_alert = 75
ecg_alert = np.sin(2*np.pi*hr_alert/60*t_alert) + 0.1*np.random.randn(len(t_alert))
rr_alert = detector.extract_rr_intervals(ecg_alert)
hrv_alert = detector.compute_hrv_features(rr_alert)
hr_fatigued = 68
ecg_fatigued = np.sin(2*np.pi*hr_fatigued/60*t_alert) + 0.15*np.random.randn(len(t_alert))
rr_fatigued = detector.extract_rr_intervals(ecg_fatigued)
hrv_fatigued = detector.compute_hrv_features(rr_fatigued)
result = detector.detect_fatigue(hrv_alert, hrv_fatigued)
print("=== HRV 疲劳检测 ===")
print(f"基线 HRV: {hrv_alert}")
print(f"当前 HRV: {hrv_fatigued}")
print(f"\n检测结果: {result}")
|
