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 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272
| """ 风险预测 HMI 评估管道 基于 Liang et al., 2026 实验设计
功能: 1. 模拟 3×2×2 实验设计 2. 生成 fNIRS/ECG/行为数据 3. 统计分析 HMI 效果 4. 可视化结果
依赖: pip install numpy scipy scikit-learn matplotlib """
import numpy as np from scipy import stats from dataclasses import dataclass from typing import List, Dict, Tuple from enum import IntEnum import json
class HMIType(IntEnum): BASELINE = 0 RISK_ALERT = 1 MULTIMODAL = 2
class RiskLevel(IntEnum): LOW = 0 HIGH = 1
@dataclass class TrialResult: """单次试验结果""" hmi: HMIType risk: RiskLevel turn: str takeover: bool takeover_time: float subjective_risk: float situational_trust: float workload: float fnirs_left_ba10: float fnirs_right_ba10: float hrv_sdnn: float eda_peak: float
class RiskPredictiveHMISimulator: """风险预测 HMI 效果仿真器""" def __init__(self, n_participants: int = 30): self.n = n_participants np.random.seed(42) def generate_trial(self, p_id: int, hmi: HMIType, risk: RiskLevel, turn: str) -> TrialResult: """生成单次试验数据""" p_baseline = np.random.normal(0, 0.1) hmi_effects = { HMIType.BASELINE: { 'risk_diff': 1.1, 'trust': 5.5, 'takeover_rate': 0.65, 'fnirs_left': 0.35, 'fnirs_right': 0.28 }, HMIType.RISK_ALERT: { 'risk_diff': 2.5, 'trust': 6.0, 'takeover_rate': 0.45, 'fnirs_left': 0.52, 'fnirs_right': 0.40 }, HMIType.MULTIMODAL: { 'risk_diff': 3.5, 'trust': 7.2, 'takeover_rate': 0.35, 'fnirs_left': 0.68, 'fnirs_right': 0.55 } } eff = hmi_effects[hmi] risk_level = 8.0 if risk == RiskLevel.HIGH else 4.5 subj_risk = risk_level + np.random.normal(0, 1.0) - (eff['risk_diff'] - 1.5) * 0.3 trust = eff['trust'] + np.random.normal(0, 0.8) if risk == RiskLevel.HIGH: trust -= 1.5 if turn == 'right': trust -= 0.5 if risk == RiskLevel.HIGH: takeover_prob = eff['takeover_rate'] + 0.2 else: takeover_prob = eff['takeover_rate'] - 0.2 takeover = np.random.random() < takeover_prob takeover_time = np.random.exponential(2.5) if takeover else 0.0 workload = 5.0 + (1 if risk == RiskLevel.HIGH else 0) * 2 + \ (1 if hmi == HMIType.MULTIMODAL else 0) * 0.5 + \ np.random.normal(0, 0.8) fnirs_left = eff['fnirs_left'] + p_baseline + \ (0.2 if risk == RiskLevel.HIGH else 0) + \ np.random.normal(0, 0.08) fnirs_right = eff['fnirs_right'] + p_baseline + \ (0.15 if risk == RiskLevel.HIGH else 0) + \ np.random.normal(0, 0.08) hrv = 45 + (5 if hmi == HMIType.MULTIMODAL else 0) + \ (-10 if risk == RiskLevel.HIGH else 0) + \ np.random.normal(0, 5) eda = 2.0 + (1.5 if risk == RiskLevel.HIGH else 0) + \ np.random.normal(0, 0.5) return TrialResult( hmi=hmi, risk=risk, turn=turn, takeover=takeover, takeover_time=takeover_time, subjective_risk=np.clip(subj_risk, 1, 10), situational_trust=np.clip(trust, 1, 10), workload=np.clip(workload, 1, 10), fnirs_left_ba10=max(0, fnirs_left), fnirs_right_ba10=max(0, fnirs_right), hrv_sdnn=max(20, hrv), eda_peak=max(0.5, eda) ) def run_experiment(self) -> List[TrialResult]: """运行完整实验""" results = [] for p_id in range(self.n): for hmi in HMIType: for risk in RiskLevel: for turn in ['left', 'right']: results.append(self.generate_trial(p_id, hmi, risk, turn)) return results def analyze(self, results: List[TrialResult]) -> Dict: """统计分析""" analysis = {} for hmi in HMIType: hmi_results = [r for r in results if r.hmi == hmi] high = [r for r in hmi_results if r.risk == RiskLevel.HIGH] low = [r for r in hmi_results if r.risk == RiskLevel.LOW] analysis[hmi.name] = { 'n': len(hmi_results), 'takeover_rate_high': np.mean([r.takeover for r in high]), 'takeover_rate_low': np.mean([r.takeover for r in low]), 'subj_risk_high': np.mean([r.subjective_risk for r in high]), 'subj_risk_low': np.mean([r.subjective_risk for r in low]), 'trust': np.mean([r.situational_trust for r in hmi_results]), 'workload': np.mean([r.workload for r in hmi_results]), 'fnirs_left': np.mean([r.fnirs_left_ba10 for r in high]), 'fnirs_right': np.mean([r.fnirs_right_ba10 for r in high]), 'hrv_high': np.mean([r.hrv_sdnn for r in high]), 'hrv_low': np.mean([r.hrv_sdnn for r in low]), } risk_high = [r.subjective_risk for r in high] risk_low = [r.subjective_risk for r in low] t_stat, p_val = stats.ttest_ind(risk_high, risk_low) analysis[hmi.name]['risk_t_stat'] = round(t_stat, 2) analysis[hmi.name]['risk_p_value'] = round(p_val, 4) return analysis
if __name__ == "__main__": print("=" * 70) print("风险预测 HMI 评估管道") print("基于 Liang et al., 2026") print("=" * 70) sim = RiskPredictiveHMISimulator(n_participants=30) results = sim.run_experiment() print(f"\n实验完成: {len(results)} 次试验 (30被试 × 3 HMI × 2 风险 × 2 转向)") analysis = sim.analyze(results) print(f"\n{'='*70}") print("HMI 效果对比") print(f"{'='*70}") print(f"\n{'指标':<20s} {'Baseline':>10s} {'Risk-alert':>10s} {'Multimodal':>10s}") print("-" * 55) metrics = [ ('高风险接管率', 'takeover_rate_high', '%'), ('低风险接管率', 'takeover_rate_low', '%'), ('高风险主观评分', 'subj_risk_high', '/10'), ('低风险主观评分', 'subj_risk_low', '/10'), ('情境信任', 'trust', '/10'), ('工作量', 'workload', '/10'), ('左前额BA10', 'fnirs_left', 'μM'), ('右前额BA10', 'fnirs_right', 'μM'), ('HRV (高风险)', 'hrv_high', 'ms'), ('HRV (低风险)', 'hrv_low', 'ms'), ] for name, key, unit in metrics: vals = [] for hmi in ['BASELINE', 'RISK_ALERT', 'MULTIMODAL']: v = analysis[hmi][key] if unit == '%': vals.append(f"{v*100:.0f}%") elif unit == '/10': vals.append(f"{v:.1f}") else: vals.append(f"{v:.2f}") print(f"{name:<20s} {vals[0]:>10s} {vals[1]:>10s} {vals[2]:>10s}") print(f"\n{'='*70}") print("高 vs 低风险区分能力 (t检验)") print(f"{'='*70}") print(f"{'HMI类型':<15s} {'t统计量':>10s} {'p值':>10s} {'显著性':>10s}") print("-" * 50) for hmi_name in ['BASELINE', 'RISK_ALERT', 'MULTIMODAL']: t = analysis[hmi_name]['risk_t_stat'] p = analysis[hmi_name]['risk_p_value'] sig = '***' if p < 0.001 else ('**' if p < 0.01 else ('*' if p < 0.05 else 'ns')) print(f"{hmi_name:<15s} {t:>10.2f} {p:>10.4f} {sig:>10s}") print(f"\n{'='*70}") print("IMS 应用启示") print(f"{'='*70}") print(""" 1. 风险预测 HMI 设计: - 视觉提示: HUD 显示风险等级颜色 (绿→黄→红) - 听觉提示: 高风险时增加定向声音警报 - 多模态 > 仅视觉 > 无提示
2. 接管决策支持: - 低风险: 抑制接管提示 (降低不必要接管) - 高风险: 强化接管提示 (确保及时接管) - 效果: 总接管次数 -48%, 低风险接管 -64%
3. AR-HUD 自适应: ┌─────────────────────────────────────────┐ │ 认知状态 AR-HUD 策略 │ │ ─────────────────────────────────── │ │ 低风险+清醒 标准导航+路况信息 │ │ 低风险+疲劳 增加刺激信息密度 │ │ 高风险+清醒 简化显示, 突出风险 │ │ 高风险+过载 仅显示关键警告+声音 │ └─────────────────────────────────────────┘
4. 部署路线: Phase 1 (2026): 视觉风险提示 (HUD颜色编码) Phase 2 (2027): +听觉多模态 (定向声音) Phase 3 (2028): +fNIRS认知状态自适应 Phase 4 (2029): 完全自适应风险预测 HMI """)
|