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
| class FatigueIndicatorExtractor:
"""
多模态疲劳指标提取器
论文核心方法：从6种传感器提取疲劳相关指标
"""
@staticmethod
def extract_hrv_features(ecg: np.ndarray, fs: int = 1000) -> dict:
"""
ECG→HRV特征（论文发现最鲁棒的指标）
"""
from scipy.signal import find_peaks
r_peaks, _ = find_peaks(
ecg, height=np.percentile(ecg, 97),
distance=fs * 0.3
)
rr_intervals = np.diff(r_peaks) / fs * 1000
if len(rr_intervals) < 5:
return {'rmssd': 0, 'sdnn': 0, 'lf_hf': 0, 'mean_hr': 0}
rmssd = np.sqrt(np.mean(np.diff(rr_intervals) ** 2))
sdnn = np.std(rr_intervals)
mean_hr = 60000 / np.mean(rr_intervals)
rr_uniform = np.interp(
np.linspace(0, len(rr_intervals) - 1, 256),
np.arange(len(rr_intervals)),
rr_intervals
)
freqs = np.fft.rfftfreq(256, d=1/4)
power = np.abs(np.fft.rfft(rr_uniform)) ** 2
lf_mask = (freqs >= 0.04) & (freqs < 0.15)
hf_mask = (freqs >= 0.15) & (freqs <= 0.4)
lf_power = np.sum(power[lf_mask])
hf_power = np.sum(power[hf_mask])
return {
'rmssd': rmssd,
'sdnn': sdnn,
'lf_hf': lf_power / (hf_power + 1e-8),
'mean_hr': mean_hr,
'lf_power': lf_power,
'hf_power': hf_power
}
@staticmethod
def extract_eeg_features(eeg: np.ndarray, fs: int = 500) -> dict:
"""
EEG→频段功率（论文发现真实环境不可靠）
"""
from scipy.signal import welch
freqs, psd = welch(eeg, fs=fs, nperseg=fs*4)
bands = {
'theta': (4, 8),
'alpha': (8, 13),
'beta': (13, 30)
}
features = {}
for name, (f_low, f_high) in bands.items():
mask = (freqs >= f_low) & (freqs < f_high)
features[f'{name}_power'] = np.sum(psd[mask])
features['theta_alpha_ratio'] = (
features['theta_power'] / (features['alpha_power'] + 1e-8)
)
return features
@staticmethod
def extract_breathing_features(resp: np.ndarray, fs: int = 25) -> dict:
"""
呼吸带→呼吸率（论文发现第二鲁棒指标）
"""
from scipy.signal import find_peaks
from scipy.signal import butter, filtfilt
b, a = butter(4, [0.1, 0.5], btype='band', fs=fs)
filtered = filtfilt(b, a, resp)
peaks, _ = find_peaks(filtered, distance=fs * 2)
breathing_rate = len(peaks) / (len(resp) / fs / 60)
amplitudes = [filtered[p] for p in peaks]
return {
'breathing_rate': breathing_rate,
'breathing_amp_mean': np.mean(amplitudes) if amplitudes else 0,
'breathing_amp_var': np.std(amplitudes) if amplitudes else 0,
'breathing_interval_var': np.std(np.diff(peaks) / fs) if len(peaks) > 2 else 0
}
|