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
| class XpengMultiModalDMS: """ 小鹏多模态DMS融合系统 接触式(安全带+方向盘+扶手)+ 非接触式(摄像头) 融合策略: 1. 接触式提供生理指标(心率/HRV/GSR/SpO2) 2. 非接触式提供行为指标(PERCLOS/视线/姿态) 3. 互补:脱手时摄像头补充,闭眼时生理补充 """ def __init__(self): self.contact_sensors = { 'seatbelt': SeatbeltSensorArray(), 'steering_wheel': SteeringWheelSensorArray(), 'armrest': SeatArmrestSensorArray() } self.camera_dms = CameraDMS() def assess_driver_state(self, all_data: dict) -> dict: """ 综合评估驾驶员状态 Args: all_data: { 'ecg': 心电数据, 'respiration': 呼吸数据, 'gsr': 皮肤电导, 'ppg': 脉搏, 'spo2': 血氧, 'camera': 摄像头数据 } Returns: 综合状态评估 """ hrv = self._compute_hrv(all_data['ecg']) heart_rate = self._compute_hr(all_data['ppg']) perclos = self.camera_dms.compute_perclos() gaze_offset = self.camera_dms.get_gaze_offset() stress_level = self._assess_stress(all_data['gsr'], hrv) health_score = self._assess_health( heart_rate, hrv, all_data['spo2'] ) fatigue_score = self._fuse_fatigue( perclos, hrv, heart_rate, respiration ) distraction_score = self._fuse_distraction( gaze_offset, steering_wheel_grip, head_pose ) return { 'fatigue': fatigue_score, 'distraction': distraction_score, 'stress': stress_level, 'health': health_score, 'heart_rate': heart_rate, 'hrv_rmssd': hrv, 'spo2': all_data['spo2'], 'confidence': self._compute_confidence(all_data) } def _fuse_fatigue(self, perclos, hrv, hr, resp): """ 融合疲劳评分 权重分配: - PERCLOS (摄像头): 35% → 行为疲劳 - HRV (接触式): 30% → 自主神经疲劳 - 心率变化: 15% → 生理负荷 - 呼吸变化: 20% → 呼吸频率下降→疲劳 """ return (0.35 * perclos + 0.30 * (1 - hrv/50) + 0.15 * max(0, (hr - 60) / 60) + 0.20 * max(0, (16 - resp) / 16)) * 100
|