多传感器融合综述：摄像头+雷达+压力的DMS架构

论文信息

核心发现

多模态传感器融合是DMS鲁棒性的关键。单一传感器存在固有局限：

传感器 优势 局限 补偿方案
摄像头 视觉信息丰富 光照敏感、遮挡 +NIR +雷达
雷达 穿透遮挡 分辨率低 +摄像头
压力垫 无遮挡 信息单一 +摄像头

融合策略三层次：

融合架构详解

1. 数据级融合（Early Fusion）

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
class EarlyFusion:
    """
    数据级融合
    
    在特征提取前融合原始数据
    """
    
    def __init__(self):
        # 传感器配置
        self.sensors = {
            'camera': RGBCamera(),
            'nir': NIRCamera(),
            'radar': MmWaveRadar()
        }
    
    def fuse_data(self, rgb_frame, nir_frame, radar_data):
        """
        原始数据融合
        
        Args:
            rgb_frame: RGB图像 (H, W, 3)
            nir_frame: NIR图像 (H, W, 1)
            radar_data: 雷达点云 (N, 4)
        
        Returns:
            fused_tensor: 融合后的tensor (H, W, 4+radar_channels)
        """
        # 1. RGB+NIR拼接
        rgb_nir = np.concatenate([rgb_frame, nir_frame], axis=-1)
        
        # 2. 雷达点云投影到图像平面
        radar_projection = self.project_radar_to_image(
            radar_data,
            intrinsic_matrix=self.camera_intrinsic
        )
        
        # 3. 拼接
        fused_tensor = np.concatenate([
            rgb_nir,
            radar_projection
        ], axis=-1)
        
        return fused_tensor
    
    def project_radar_to_image(self, radar_data, intrinsic_matrix):
        """
        将雷达点云投影到图像平面
        """
        # 雷达点云：(x, y, z, velocity)
        # 转换到相机坐标系
        points_camera = self.transform_radar_to_camera(radar_data)
        
        # 投影到图像
        points_2d = intrinsic_matrix @ points_camera[:3]
        points_2d = points_2d[:2] / points_2d[2]  # 归一化
        
        # 创建投影图
        projection = np.zeros((self.height, self.width, 1))
        
        for i, (u, v) in enumerate(points_2d.T):
            if 0 <= u < self.width and 0 <= v < self.height:
                projection[int(v), int(u), 0] = radar_data[i, 3]  # velocity
        
        return projection

2. 特征级融合（Feature Fusion）

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
class FeatureFusion(nn.Module):
    """
    特征级融合
    
    各传感器独立提取特征，然后融合
    """
    
    def __init__(self):
        super().__init__()
        
        # 各传感器特征提取器
        self.camera_encoder = ResNet50(pretrained=True)
        self.nir_encoder = ResNet18(pretrained=True)
        self.radar_encoder = PointNetEncoder()
        self.pressure_encoder = PressureEncoder()
        
        # 特征融合模块
        self.fusion = CrossAttentionFusion(
            feature_dims=[2048, 512, 256, 128]
        )
    
    def forward(self, rgb, nir, radar, pressure):
        """
        特征融合
        
        Args:
            rgb: RGB图像
            nir: NIR图像
            radar: 雷达点云
            pressure: 压力分布
        
        Returns:
            fused_features: 融合特征
        """
        # 1. 各传感器特征提取
        cam_features = self.camera_encoder(rgb)
        nir_features = self.nir_encoder(nir)
        radar_features = self.radar_encoder(radar)
        pressure_features = self.pressure_encoder(pressure)
        
        # 2. 交叉注意力融合
        fused_features = self.fusion([
            cam_features,
            nir_features,
            radar_features,
            pressure_features
        ])
        
        return fused_features


class CrossAttentionFusion(nn.Module):
    """
    交叉注意力融合模块
    
    动态加权不同传感器特征
    """
    
    def __init__(self, feature_dims):
        super().__init__()
        
        self.attention = nn.MultiheadAttention(
            embed_dim=sum(feature_dims),
            num_heads=8
        )
    
    def forward(self, features_list):
        """
        交叉注意力融合
        """
        # 拼接所有特征
        concat_features = torch.cat(features_list, dim=-1)
        
        # 自注意力
        fused, _ = self.attention(
            concat_features,
            concat_features,
            concat_features
        )
        
        return fused

3. 决策级融合（Decision Fusion）

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
class DecisionFusion:
    """
    决策级融合
    
    各传感器独立检测，结果融合
    """
    
    def __init__(self):
        # 各传感器检测器
        self.camera_detector = CameraDetector()
        self.radar_detector = RadarDetector()
        self.pressure_detector = PressureDetector()
        
        # 融合策略
        self.fusion_strategy = 'weighted_vote'
    
    def fuse_detections(self, camera_result, radar_result, pressure_result):
        """
        决策融合
        
        Args:
            camera_result: {'fatigue': 0.7, 'distraction': 0.3}
            radar_result: {'presence': True, 'position': (x, y)}
            pressure_result: {'occupant_weight': 70, 'posture': 'normal'}
        
        Returns:
            final_decision: 融合后的决策
        """
        if self.fusion_strategy == 'weighted_vote':
            return self.weighted_vote(camera_result, radar_result, pressure_result)
        elif self.fusion_strategy == 'bayesian':
            return self.bayesian_fusion(camera_result, radar_result, pressure_result)
    
    def weighted_vote(self, *results):
        """
        加权投票
        
        根据场景动态调整权重
        """
        # 场景自适应权重
        weights = self.adaptive_weights(results)
        
        # 加权融合
        final = {}
        for key in ['fatigue', 'distraction']:
            scores = [r.get(key, 0) for r in results]
            final[key] = sum(s * w for s, w in zip(scores, weights))
        
        return final
    
    def adaptive_weights(self, results):
        """
        自适应权重
        
        根据传感器可靠性动态调整
        """
        # 示例：夜间提高NIR权重，降低RGB权重
        weights = [0.4, 0.4, 0.2]  # camera, radar, pressure
        
        # 如果NIR有效，提高其权重
        if results[0].get('nir_valid', False):
            weights[0] = 0.5
        
        return weights

DMS应用场景

1. 疲劳检测融合

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
class FatigueFusion:
    """
    疲劳检测多传感器融合
    
    摄像头+NIR+压力
    """
    
    def __init__(self):
        # 指标权重
        self.indicator_weights = {
            'perclos': 0.35,  # 摄像头/NIR
            'blink_rate': 0.20,  # 摄像头/NIR
            'pupil_size': 0.15,  # NIR
            'steering_variance': 0.15,  # CAN
            'seat_pressure_variance': 0.15  # 压力
        }
    
    def detect_fatigue(self, sensor_data):
        """
        多传感器疲劳检测
        
        Returns:
            fatigue_level: 0-3
        """
        indicators = {}
        
        # 1. 摄像头/NIR：眼动指标
        indicators['perclos'] = self.compute_perclos(sensor_data['eye_features'])
        indicators['blink_rate'] = self.compute_blink_rate(sensor_data['eye_features'])
        indicators['pupil_size'] = sensor_data['pupil_size']
        
        # 2. CAN：驾驶行为
        indicators['steering_variance'] = self.compute_steering_variance(
            sensor_data['steering_angle']
        )
        
        # 3. 压力：姿态变化
        indicators['seat_pressure_variance'] = self.compute_pressure_variance(
            sensor_data['pressure_map']
        )
        
        # 4. 加权融合
        fatigue_score = sum(
            indicators[key] * self.indicator_weights[key]
            for key in self.indicator_weights
        )
        
        # 5. 分级
        if fatigue_score < 0.25:
            return 0
        elif fatigue_score < 0.5:
            return 1
        elif fatigue_score < 0.75:
            return 2
        else:
            return 3

2. CPD儿童检测融合

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
class CPDFusion:
    """
    儿童存在检测融合
    
    雷达+NIR+压力
    """
    
    def __init__(self):
        self.radar = MmWaveRadar()
        self.nir = NIRCamera()
        self.pressure = PressureMat()
    
    def detect_child(self, sensor_data):
        """
        融合检测儿童
        
        Returns:
            result: {
                'child_detected': bool,
                'position': (x, y, z),
                'confidence': float
            }
        """
        # 1. 雷达：穿透遮挡检测
        radar_result = self.radar.detect_presence(sensor_data['radar'])
        
        # 2. NIR：视觉验证
        nir_result = self.nir.detect_child(sensor_data['nir'])
        
        # 3. 压力：重量辅助判断
        pressure_result = self.pressure.detect_weight(sensor_data['pressure'])
        
        # 4. 融合逻辑
        # 雷达穿透性强，权重高
        # NIR提供视觉确认
        # 压力提供重量参考
        
        child_detected = False
        confidence = 0.0
        
        # 雷达检测到运动
        if radar_result['presence']:
            child_detected = True
            confidence = 0.5
            
            # NIR确认
            if nir_result['child_detected']:
                confidence += 0.3
            
            # 压力辅助（儿童重量轻）
            if pressure_result['weight'] < 30:  # kg
                confidence += 0.2
        
        return {
            'child_detected': child_detected,
            'position': radar_result.get('position', None),
            'confidence': min(confidence, 1.0)
        }

IMS开发启示

1. 融合策略选择

场景 推荐融合策略 原因
疲劳检测 特征级融合 多模态特征互补
CPD检测 决策级融合 雷达穿透+视觉确认
OOP检测 数据级融合 压力+视觉原始数据

2. 硬件配置

传感器 型号 接口 成本
RGB摄像头 AR0231 MIPI $20
NIR摄像头 OV2311 MIPI $30
mmWave雷达 IWR6843 SPI $40
压力垫 定制 I2C $50
处理器 QCS8255 - $500

3. 性能对比

配置 准确率 成本 适用场景
仅摄像头 85% $50 白天良好光照
摄像头+NIR 92% $80 全天候
+雷达 95% $120 穿透遮挡
+压力 97% $170 完整方案

4. 测试验证要点

测试项 方法 通过标准
单传感器失效 熔断测试 性能下降<10%
光照变化 0-100k lux 波动<5%
遮挡测试 各类遮挡 检测率>90%
融合延迟 时序测试 <50ms

总结

多传感器融合是DMS鲁棒性的核心：

融合层次 优势 挑战 推荐场景
数据级 信息完整 计算量大 数据充足
特征级 平衡性能 需对齐 推荐
决策级 实现简单 信息损失 快速集成

IMS开发建议：

  • 阶段1：摄像头+NIR（特征级融合）
  • 阶段2：+雷达CPD（决策级融合）
  • 阶段3：+压力OOP（完整方案）

参考论文：

  1. Sensors, “A Review of Multi-Sensor Fusion in Autonomous Driving”, 2025