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
| import numpy as np from typing import Dict
class CabinSensorFusion: """ 座舱多传感器融合框架 三层融合: 1. 数据级: 时间同步+空间标定 2. 特征级: 多模态特征注意力融合 3. 决策级: 加权投票+冲突解决 """ def __init__(self): self.sensors = { 'ir_camera': {'fps': 30, 'latency_ms': 15}, 'mmwave': {'fps': 30, 'latency_ms': 10}, 'uwb': {'fps': 10, 'latency_ms': 50}, 'pressure': {'fps': 100, 'latency_ms': 2}, 'imu': {'fps': 200, 'latency_ms': 1}, } self.reliability = {k: 1.0 for k in self.sensors} self.sync_buffer = {} self.calibration = self._load_calibration() def _load_calibration(self): """加载传感器间标定参数""" return { 'ir_camera': {'extrinsics': np.eye(4), 'intrinsics': np.eye(3)}, 'mmwave': {'extrinsics': np.eye(4)}, 'uwb': {'position': np.array([0, 0, 0])}, } def data_level_sync(self, sensor_data: Dict, target_time: float): """ 数据级融合:时间同步 将不同帧率的传感器数据同步到目标时间 """ synced = {} for name, data in sensor_data.items(): timestamps = data['timestamps'] values = data['values'] if len(timestamps) > 1: synced_value = np.interp( target_time, timestamps, values ) else: synced_value = values[0] latency = self.sensors[name]['latency_ms'] / 1000 synced[name] = { 'value': synced_value, 'latency_compensated': target_time - latency, 'confidence': self.reliability[name] } return synced def feature_level_fuse(self, features: Dict): """ 特征级融合:注意力机制 各传感器提取特征后,用注意力加权融合 """ total_energy = sum(np.sum(f**2) for f in features.values()) fused = np.zeros_like(list(features.values())[0]) for name, feat in features.items(): weight = np.sum(feat**2) / (total_energy + 1e-8) weight *= self.reliability[name] fused += weight * feat return fused def decision_level_fuse(self, decisions: Dict): """ 决策级融合:加权投票 各传感器给出分类结果后,加权投票 """ combined = {} for name, decision in decisions.items(): weight = self.reliability[name] for cls, prob in decision.items(): if cls in combined: combined[cls] *= (1 - weight * prob) else: combined[cls] = 1 - weight * prob for cls in combined: combined[cls] = 1 - combined[cls] total = sum(combined.values()) if total > 0: combined = {k: v/total for k, v in combined.items()} best = max(combined, key=combined.get) return best, combined[best] def update_reliability(self, sensor: str, quality_score: float): """动态更新传感器可靠性""" self.reliability[sensor] = ( 0.9 * self.reliability[sensor] + 0.1 * quality_score )
if __name__ == "__main__": fusion = CabinSensorFusion() decisions = { 'ir_camera': {'adult_normal': 0.7, 'adult_oop': 0.2, 'child': 0.1}, 'mmwave': {'adult_normal': 0.5, 'adult_oop': 0.4, 'child': 0.1}, 'pressure': {'adult_normal': 0.6, 'child': 0.3, 'adult_oop': 0.1}, } best, conf = fusion.decision_level_fuse(decisions) print(f"融合结果: {best} (置信度: {conf:.1%})")
|