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 226 227 228 229 230 231 232 233 234 235 236 237
| """ 标签抖动 (Label Jitter) 分析
核心问题: 分心事件 T=5s, 生理响应 T=10s 导致: 使用 T=5s 的标签训练, 但 T=5-10s 的生理信号仍为正常 """
import numpy as np from dataclasses import dataclass from typing import Tuple, List
@dataclass class LabelJitterConfig: """标签抖动配置""" event_onset: float = 5.0 physiological_delay: float = 10.0 visual_delay: float = 5.5 window_size: float = 5.0
def simulate_label_jitter() -> dict: """ 模拟标签抖动对分类的影响 场景: T=5s 开始分心, 使用 5s 窗口分类 """ config = LabelJitterConfig() t = np.arange(0, 60, 0.1) true_labels = (t >= config.event_onset).astype(int) physio_labels = (t >= config.physiological_delay).astype(int) visual_labels = (t >= config.visual_delay).astype(int) n_windows = 60 // 5 window_true = [] window_physio = [] window_visual = [] for w in range(n_windows): start = w * 5 end = (w + 1) * 5 true_ratio = true_labels[int(start*10):int(end*10)].mean() physio_ratio = physio_labels[int(start*10):int(end*10)].mean() visual_ratio = visual_labels[int(start*10):int(end*10)].mean() window_true.append(1 if true_ratio > 0.5 else 0) window_physio.append(1 if physio_ratio > 0.5 else 0) window_visual.append(1 if visual_ratio > 0.5 else 0) mismatch_physio = sum(1 for t, p in zip(window_true, window_physio) if t != p) mismatch_visual = sum(1 for t, v in zip(window_true, window_visual) if t != v) print("=== 标签抖动分析 ===") print(f"事件发生: T={config.event_onset}s") print(f"生理延迟: {config.physiological_delay - config.event_onset}s") print(f"视觉延迟: {config.visual_delay - config.event_onset}s") print(f"\n窗口标签对比 (12 个 5s 窗口):") print(f"{'窗口':<8} {'真实':<6} {'生理':<6} {'视觉':<6} {'生理错':<6} {'视觉错'}") for w in range(n_windows): print(f"{w+1:<8} {window_true[w]:<6} {window_physio[w]:<6} " f"{window_visual[w]:<6} {int(window_true[w]!=window_physio[w]):<6} " f"{int(window_true[w]!=window_visual[w])}") print(f"\n生理标签错配: {mismatch_physio}/{n_windows} ({mismatch_physio/n_windows*100:.0f}%)") print(f"视觉标签错配: {mismatch_visual}/{n_windows} ({mismatch_visual/n_windows*100:.0f}%)") print(f"\n→ 生理信号标签抖动导致 {mismatch_physio/n_windows*100:.0f}% 窗口训练错误") return { 'physio_mismatch': mismatch_physio / n_windows, 'visual_mismatch': mismatch_visual / n_windows }
def model_comparison(): """模型对比""" results = { 'XGBoost (AU 60s窗口)': { 'window_f1': 0.79, 'session_f1': 0.94, 'training_time': '分钟级', 'interpretability': '✅ 高', 'data_requirement': '~1K 样本即可' }, 'STRNet (频谱时序 ResNet)': { 'window_f1': 0.75, 'session_f1': 0.87, 'training_time': '小时级', 'interpretability': '❌ 低', 'data_requirement': '~100K+ 样本' }, '标准 CNN': { 'window_f1': 0.71, 'session_f1': 0.82, 'training_time': '小时级', 'interpretability': '❌ 低', 'data_requirement': '~50K+ 样本' }, 'LSTM': { 'window_f1': 0.68, 'session_f1': 0.80, 'training_time': '小时级', 'interpretability': '⚠️ 中', 'data_requirement': '~50K+ 样本' } } print("\n=== 模型性能对比 ===") print(f"{'模型':<30} {'窗口F1':<10} {'会话F1':<10} {'可解释'}") for name, metrics in results.items(): print(f"{name:<30} {metrics['window_f1']:.2f}{'':<5} " f"{metrics['session_f1']:.2f}{'':<5} {metrics['interpretability']}") print(f"\n关键发现:") print(f"1. XGBoost 在 ~20K 数据量上胜出 (0.79 vs 0.75)") print(f"2. 深度学习需要 100K+ 数据才能发挥优势") print(f"3. 会话级分类远优于窗口级 (0.94 vs 0.79)") print(f"4. 面部 AU 是最强单一模态") return results
class DistractionDetector: """ IMS 分心检测器 基于论文发现设计的实用方案 策略: 1. 主特征: 面部 AU + 眼动 (DMS 摄像头) 2. 辅助: rPPG 心率 (延迟补偿后) 3. 模型: XGBoost (中等数据量) → 深度学习 (大数据量) 4. 标签策略: 延迟标注 + 时序平滑 """ def __init__(self): self.features = { 'visual': ['eye_openness', 'gaze_direction', 'blink_rate', 'pupil_diameter', 'au_intensity', 'head_pose'], 'physio': ['heart_rate', 'hrv', 'breathing_rate'], 'context': ['speed', 'steering_angle', 'time_of_day'] } self.delays = { 'visual': 0.5, 'physio': 5.0, 'context': 0.0, } def delay_compensated_label(self, event_time: float, modality: str, window: float = 5.0) -> float: """ 延迟补偿标注 对于生理信号, 标签延迟 delay 秒 对于视觉信号, 标签延迟 0.5s Args: event_time: 分心事件发生时间 modality: 'visual' 或 'physio' window: 分类窗口大小 Returns: compensated_time: 补偿后的标签时间 """ delay = self.delays.get(modality, 0) return event_time + delay def feature_importance(self) -> dict: """特征重要性排序""" importance = { 'au12_lip_corner_pull': 0.085, 'au14_dimpler': 0.072, 'au04_brow_lower': 0.068, 'au15_lip_corner_depress': 0.061, 'gaze_off_center': 0.058, 'blink_rate': 0.045, 'pupil_diameter': 0.038, 'head_yaw': 0.035, 'heart_rate': 0.022, 'hrv_rmssd': 0.018, 'breathing_rate': 0.015, 'palm_eda': 0.012, 'steering_var': 0.025, 'speed_var': 0.020, } print("=== 特征重要性 (Top 10) ===") for i, (feat, imp) in enumerate(sorted(importance.items(), key=lambda x: -x[1])[:10]): print(f" {i+1}. {feat}: {imp:.3f}") visual_sum = sum(v for k, v in importance.items() if not k.startswith(('heart', 'hrv', 'breath', 'palm', 'steer', 'speed'))) physio_sum = sum(v for k, v in importance.items() if k.startswith(('heart', 'hrv', 'breath', 'palm'))) print(f"\n视觉特征总重要性: {visual_sum:.3f} ({visual_sum/(visual_sum+physio_sum)*100:.0f}%)") print(f"生理特征总重要性: {physio_sum:.3f} ({physio_sum/(visual_sum+physio_sum)*100:.0f}%)") return importance
if __name__ == "__main__": jitter_result = simulate_label_jitter() model_comparison() detector = DistractionDetector() importance = detector.feature_importance() print("\n=== IMS 分心检测推荐方案 ===") print("1. 主特征: 面部 AU (30ch) + 眼动 (6ch) — DMS 摄像头可提取") print("2. 辅助: rPPG 心率 — 延迟 5s 补偿后使用") print("3. 模型: XGBoost (<50K 样本) → 深度学习 (100K+ 样本)") print("4. 标签: 延迟标注 + 时序平滑 (减少标签抖动)") print("5. 窗口: 60s 会话级 > 20s 窗口级 (F1 0.94 vs 0.79)")
|