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
| import numpy as np from typing import Dict
class SensorFusionEngine: """多传感器融合引擎""" def __init__(self): self.camera_detector = CameraDetector() self.radar_detector = RadarDetector() self.pressure_detector = PressureDetector() def fuse(self, camera_data, radar_data, pressure_data) -> Dict: """融合三传感器数据""" cam_result = self.camera_detector.detect(camera_data) radar_result = self.radar_detector.detect(radar_data) pressure_result = self.pressure_detector.detect(pressure_data) fused = { 'occupant_present': self._fuse_presence( cam_result, radar_result, pressure_result), 'posture': self._fuse_posture( cam_result, pressure_result), 'child_detected': self._fuse_cpd( radar_result, pressure_result), 'fatigue_level': cam_result.get('fatigue', 0), 'distraction_level': cam_result.get('distraction', 0), 'belt_status': cam_result.get('belt', 'normal'), 'weight_estimate': pressure_result.get('weight', 0), } return fused def _fuse_presence(self, cam, radar, pressure): """融合乘员存在检测""" votes = { 'camera': cam.get('occupant', False), 'radar': radar.get('occupant', False), 'pressure': pressure.get('occupant', False), } return sum(votes.values()) >= 2 def _fuse_posture(self, cam, pressure): """融合姿态检测""" upper = cam.get('upper_body_pose', None) lower = pressure.get('lower_body_pose', None) if upper and lower: return {'upper': upper, 'lower': lower, 'complete': True} elif upper: return {'upper': upper, 'lower': None, 'complete': False} else: return {'upper': None, 'lower': lower, 'complete': False} def _fuse_cpd(self, radar, pressure): """融合CPD检测""" radar_vital = radar.get('vital_signs', False) weight = pressure.get('weight', 0) if radar_vital and weight < 15: return {'child': True, 'confidence': 0.95} elif radar_vital and weight < 30: return {'child': True, 'confidence': 0.80} else: return {'child': False, 'confidence': 0.90}
|