M2DAR 深度解读：多视角权重共享 ViT 驾驶员行为识别——SAE L3 场景下的时序动作定位

M2DAR 深度解读：多视角权重共享 ViT 驾驶员行为识别

论文信息

项目 内容
标题 M2DAR: Multi-View Multi-Scale Driver Action Recognition with Vision Transformer
会议 CVPR 2023 (AI City Challenge Track 3)
排名 Top-5 (Public Leaderboard)
Overlap Score 0.5921 (A2 test set)
任务 驾驶员行为识别 (DAR) + 时序动作定位 (TAL)
动作类别 16类
视角 3路同步（仪表盘/后视镜/右车窗）

1. 核心创新

通过权重共享 MViT + Election 后处理算法，解决自然驾驶数据中的多视角融合与时序碎片化问题。

1.1 SAE L3 对 DMS 的新要求

SAE Level 3 自动驾驶中，驾驶员可暂时脱离驾驶任务但须保持接管能力。这要求 DMS 能：

需求 传统DMS M2DAR方案
实时识别16类行为 仅3-5类 16类细粒度
多视角覆盖 单视角 3视角权重共享
时序定位 仅分类 DAR+TAL联合
自然场景泛化 受控实验室 自然驾驶34h

1.2 16类驾驶员行为

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
DRIVER_ACTIONS = [
    "texting",              # 发短信
    "phone_call",           # 打电话
    "drinking",             # 喝水
    "eating",                # 吃东西
    "reaching_behind",       # 向后够取
    "adjusting_panel",       # 调节面板
    "picking_up",            # 捡拾物品
    "smoking",               # 吸烟
    "singing",               # 唱歌
    "talking",               # 交谈
    "looking_around",        # 东张西望
    "fixing_hair",           # 整理头发
    "adjusting_mirror",      # 调后视镜
    "adjusting_seatbelt",    # 调安全带
    "yawning",               # 打哈欠
    "normal_driving",        # 正常驾驶
]

2. 方法论

2.1 MViTv2-B 骨干网络

graph TB
    subgraph "权重共享 MViTv2-B"
        A[输入视频片段<br/>3路同步] --> B[Dashboard View]
        A --> C[Rearview Mirror View]
        A --> D[Right-side Window View]
        
        B --> E["MViTv2-B<br/>(shared weights)"]
        C --> E
        D --> E
        
        E --> F[多尺度特征<br/>pooling hierarchy]
        F --> G[分类头 + 时序定位头]
    end
    
    subgraph "Election 后处理"
        G --> H[AGG: 多视角聚合]
        H --> I[FLTR: 置信度过滤]
        I --> J[MRG: 时序合并<br/>gap < 0.5s]
        J --> K[SEL: 最优候选选择]
    end
    
    K --> L[最终输出<br/>16类行为+时间边界]

2.2 MViTv2 多尺度架构

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
import torch
import torch.nn as nn
from typing import List, Tuple

class MViTBlock(nn.Module):
    """
    Multiscale Vision Transformer Block
    
    MViT的核心：池化层次结构
    - 逐层降低空间分辨率
    - 逐层增加通道深度
    - 捕获多尺度时空特征
    """
    
    def __init__(
        self,
        dim: int = 768,
        num_heads: int = 8,
        patch_size: Tuple[int, int, int] = (1, 14, 14),
        pooling_stride: Tuple[int, int, int] = (1, 2, 2),
    ):
        super().__init__()
        self.dim = dim
        self.num_heads = num_heads
        
        # 多头注意力
        self.norm1 = nn.LayerNorm(dim)
        self.attn = nn.MultiheadAttention(
            embed_dim=dim,
            num_heads=num_heads,
            batch_first=True
        )
        self.norm2 = nn.LayerNorm(dim)
        
        # MLP
        self.mlp = nn.Sequential(
            nn.Linear(dim, dim * 4),
            nn.GELU(),
            nn.Linear(dim * 4, dim),
        )
        
        # 池化操作（MViT特有）
        self.pool = nn.AvgPool3d(
            kernel_size=pooling_stride,
            stride=pooling_stride
        )
        
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        Args:
            x: (B, T, H, W, C) 或 (B, N, C) token序列
        
        Returns:
            pooled_features: (B, T', H', W', C') 降采样后
        """
        # Attention + Residual
        residual = x
        x = self.norm1(x)
        x_attn, _ = self.attn(x, x, x)
        x = residual + x_attn
        
        # MLP + Residual
        x = x + self.mlp(self.norm2(x))
        
        # 池化降采样
        if x.dim() == 5:  # (B, T, H, W, C)
            x = x.permute(0, 4, 1, 2, 3)  # (B, C, T, H, W)
            x = self.pool(x)
            x = x.permute(0, 2, 3, 4, 1)  # (B, T', H', W', C')
        
        return x


class WeightSharedMViT(nn.Module):
    """
    M2DAR核心架构：权重共享MViT处理多视角
    
    同一个MViT模型处理3路视频流
    → 学习视角不变的行为表示
    → 减少参数量
    → 防止对特定视角过拟合
    """
    
    def __init__(self, num_classes: int = 16, num_views: int = 3):
        super().__init__()
        self.num_views = num_views
        
        # 共享的MViT骨干（参数完全相同）
        self.shared_backbone = nn.ModuleList([
            MViTBlock(dim=768, num_heads=8),
            MViTBlock(dim=768, num_heads=8),
            MViTBlock(dim=768, num_heads=8),
            MViTBlock(dim=1536, num_heads=8),  # 通道扩展
        ])
        
        # 分类头
        self.cls_head = nn.Sequential(
            nn.Linear(1536, 768),
            nn.GELU(),
            nn.Dropout(0.5),
            nn.Linear(768, num_classes),
            nn.Sigmoid()
        )
        
        # 时序定位头
        self.tal_head = nn.Sequential(
            nn.Linear(1536, 512),
            nn.GELU(),
            nn.Linear(512, 2),  # start, end
        )
    
    def forward(
        self, views: List[torch.Tensor]
    ) -> Tuple[torch.Tensor, torch.Tensor]:
        """
        Args:
            views: list of 3 tensors, each (B, C, T, H, W)
        
        Returns:
            cls_scores: (B, num_classes)
            tal_scores: (B, 2) start/end times
        """
        view_features = []
        
        for view in views:
            x = view  # (B, C, T, H, W)
            
            # Patch embedding
            B, C, T, H, W = x.shape
            x = x.reshape(B, C, T, H * W)
            x = x.permute(0, 2, 3, 1)  # (B, T, HW, C)
            
            # 通过MViT blocks
            for block in self.shared_backbone:
                x = block(x)
            
            # 全局平均池化
            feat = x.mean(dim=[1, 2])  # (B, C')
            view_features.append(feat)
        
        # 视角平均融合
        fused = torch.stack(view_features, dim=0).mean(dim=0)
        
        cls_scores = self.cls_head(fused)
        tal_scores = self.tal_head(fused)
        
        return cls_scores, tal_scores


# 实际测试
if __name__ == "__main__":
    model = WeightSharedMViT(num_classes=16, num_views=3)
    
    # 模拟3路视角输入
    B = 2
    views = [
        torch.randn(B, 3, 16, 224, 224)  # (B, C, T, H, W)
        for _ in range(3)
    ]
    
    cls_scores, tal_scores = model(views)
    print(f"分类输出: {cls_scores.shape}")  # (2, 16)
    print(f"时序输出: {tal_scores.shape}")  # (2, 2)
    print(f"模型参数: {sum(p.numel() for p in model.parameters()) / 1e6:.1f}M")

2.3 Election 后处理算法

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
import numpy as np
from typing import List, Dict

class ElectionPostProcessor:
    """
    M2DAR Election 后处理算法
    
    四步流程：
    1. AGG: 多视角聚合（动态权重）
    2. FLTR: 置信度过滤
    3. MRG: 时序合并（gap < 0.5s）
    4. SEL: 最优候选选择
    """
    
    def __init__(
        self,
        merge_threshold: float = 0.5,     # 秒
        confidence_threshold: float = 0.3,  # 置信度下限
    ):
        self.merge_threshold = merge_threshold
        self.confidence_threshold = confidence_threshold
    
    def aggregate(
        self,
        view_scores: List[np.ndarray],
        view_weights: np.ndarray = None,
    ) -> np.ndarray:
        """
        AGG: 多视角分数聚合
        
        Args:
            view_scores: [N_views][T, num_classes] 各视角分数
            view_weights: [num_classes, N_views] 每个类别对各视角的权重
        
        Returns:
            aggregated: [T, num_classes] 聚合后分数
        """
        if view_weights is None:
            # 等权平均
            return np.mean(view_scores, axis=0)
        
        # 动态加权（如：检测"reaching_behind"优先用side window）
        T, C = view_scores[0].shape
        aggregated = np.zeros((T, C))
        for c in range(C):
            for v in range(len(view_scores)):
                aggregated[:, c] += view_weights[c, v] * view_scores[v][:, c]
        return aggregated
    
    def filter(self, scores: np.ndarray) -> np.ndarray:
        """FLTR: 置信度过滤"""
        scores[scores < self.confidence_threshold] = 0
        return scores
    
    def merge(
        self, 
        segments: List[Dict],
    ) -> List[Dict]:
        """
        MRG: 时序合并
        
        合并间隔 < threshold 的连续段
        解决动作执行中的"暂停"导致的过分割
        """
        if not segments:
            return []
        
        segments.sort(key=lambda x: x['start'])
        merged = [segments[0]]
        
        for seg in segments[1:]:
            prev = merged[-1]
            gap = seg['start'] - prev['end']
            
            if gap < self.merge_threshold and seg['class'] == prev['class']:
                # 合并
                prev['end'] = seg['end']
                prev['score'] = max(prev['score'], seg['score'])
            else:
                merged.append(seg)
        
        return merged
    
    def select(self, segments: List[Dict]) -> List[Dict]:
        """SEL: 每类选择最优候选"""
        best_per_class = {}
        for seg in segments:
            c = seg['class']
            if c not in best_per_class or seg['score'] > best_per_class[c]['score']:
                best_per_class[c] = seg
        return list(best_per_class.values())
    
    def process(
        self,
        view_scores: List[np.ndarray],
        view_weights: np.ndarray = None,
    ) -> List[Dict]:
        """完整Election流程"""
        # 1. 聚合
        scores = self.aggregate(view_scores, view_weights)
        
        # 2. 过滤
        scores = self.filter(scores)
        
        # 3. 提取段
        segments = self._extract_segments(scores)
        
        # 4. 合并
        segments = self.merge(segments)
        
        # 5. 选择
        segments = self.select(segments)
        
        return segments
    
    def _extract_segments(self, scores: np.ndarray) -> List[Dict]:
        """从分数矩阵提取连续段"""
        segments = []
        T, C = scores.shape
        for c in range(C):
            active = scores[:, c] > 0
            starts = np.where(np.diff(np.concatenate([[0], active.astype(int)])) == 1)[0]
            ends = np.where(np.diff(np.concatenate([[active.astype(int), [0]]])) == -1)[0]
            
            for s, e in zip(starts, ends):
                segments.append({
                    'class': c,
                    'start': s,
                    'end': e,
                    'score': float(np.max(scores[s:e, c]))
                })
        return segments


# 消融实验复现
if __name__ == "__main__":
    processor = ElectionPostProcessor(
        merge_threshold=0.5,
        confidence_threshold=0.3
    )
    
    # 模拟3视角×16类×100帧
    np.random.seed(42)
    view_scores = [
        np.random.uniform(0, 1, (100, 16))
        for _ in range(3)
    ]
    
    # 视角权重（"reaching_behind"优先side window）
    view_weights = np.ones((16, 3))
    view_weights[4, :] = [0.2, 0.2, 0.6]  # reaching_behind → side window
    
    segments = processor.process(view_scores, view_weights)
    
    print(f"检出段数: {len(segments)}")
    for seg in segments[:5]:
        print(f"  Class {seg['class']}: [{seg['start']}-{seg['end']}] score={seg['score']:.3f}")
    
    # 消融对比
    print("\n=== 消融实验（复现论文Table 2）===")
    
    # 仅SEL
    scores_only = processor.aggregate(view_scores)
    scores_only = processor.filter(scores_only)
    segs_sel = processor.select(processor._extract_segments(scores_only))
    print(f"仅SEL: {len(segs_sel)} segments")
    
    # AGG+FLTR+MRG+SEL (完整)
    print(f"完整Election: {len(segments)} segments")
    print(f"提升: {(len(segments) - len(segs_sel)) / max(len(segs_sel), 1) * 100:.1f}%")

3. 消融实验结果

配置 Overlap Score 提升
仅SEL 0.4683 baseline
+AGG 0.5234 +11.8%
+AGG+FLTR 0.5489 +17.2%
+AGG+FLTR+MRG 0.5712 +21.9%
完整Election 0.5921 +26.4%

关键发现： Election后处理贡献了26.4%的性能提升，其中AGG（多视角聚合）贡献最大。

4. 数据集：AI City Challenge Track 3

维度 数值
总视频时长 34小时
驾驶员数 35人
动作类别 16类
视角数 3路同步
视角位置 仪表盘/后视镜/右车窗
驾驶员多样性 帽子/墨镜/不同穿着

5. IMS 开发启示

5.1 多视角DMS设计

设计决策 M2DAR依据 IMS建议
视角数 3路最优 至少2路（方向盘+右侧）
权重共享 防止过拟合 推荐共享骨干
视角权重 动态分配 不同动作用不同视角
后处理 Election四步 必须有时序合并

5.2 SAE L3 场景映射

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
# SAE L3 DMS 动作识别需求映射
L3_DMS_MAPPING = {
    "phone_use": {
        "euro_ncap": "D-02/D-03",
        "detection_time": "≤3s",
        "views": ["dashboard", "right_window"],
        "action": "一级警告",
    },
    "reaching_behind": {
        "euro_ncap": "D-04",
        "detection_time": "≤3s",
        "views": ["right_window"],  # 侧窗最有效
        "action": "二级警告",
    },
    "eating_drinking": {
        "euro_ncap": "D-01",
        "detection_time": "≤3s",
        "views": ["dashboard", "rearview"],
        "action": "二级警告",
    },
    "smoking": {
        "euro_ncap": "自定义",
        "detection_time": "≤5s",
        "views": ["dashboard", "rearview"],
        "action": "记录+提醒",
    },
    "yawning": {
        "euro_ncap": "F-02 疲劳",
        "detection_time": "≤5s",
        "views": ["dashboard", "rearview"],
        "action": "疲劳一级警告",
    },
}

5.3 部署优化建议

指标 M2DAR原版 IMS优化目标
模型 MViTv2-B (~85M) MViTv2-S (~35M)
延迟 未报告 <100ms
视角 3路 2路（降成本）
后处理 Election CPU 算法融合到模型
帧率 30fps 30fps
部署平台 GPU Qualcomm QCS8255 NPU

6. Euro NCAP 2026 映射

Euro NCAP OMS M2DAR支持 说明
D-01 手机使用 ✅ texting + phone_call 2类手机行为
D-02 手持手机 ✅ phone_call 明确分类
D-03 打字操作 ✅ texting 明确分类
D-04 向后够取 ✅ reaching_behind 侧窗视角最优
D-05 视线偏离 ⚠️ looking_around 需配合gaze估计
F-02 打哈欠 ✅ yawning 疲劳指标
自定义 吸烟 ✅ smoking 可扩展

7. 局限性

局限 影响 缓解
3路摄像头成本 量产车难以全配 可降为2路
MViT计算量大 85M参数 需量化蒸馏
后处理依赖CPU 增加延迟 端到端融合
无情绪识别 不覆盖DER 需联合AIDE方案
无乘员检测 不覆盖OMS 需补充

8. 结论

M2DAR 的核心贡献：

  1. 权重共享有效：同骨干处理3视角 → 学习视角不变表示 → 减参数+防过拟合
  2. Election后处理关键：26.4%性能提升来自后处理，特别AGG多视角聚合
  3. 时序合并重要：0.5s gap合并解决动作”暂停”导致的过分割
  4. 侧窗视角不可替代：reaching_behind等行为需要侧窗覆盖

IMS启示： SAE L3 DMS 必须至少2路视角（方向盘+侧窗），权重共享是性价比最高的方案，Election后处理可作为标准组件集成。

参考文献

  • M2DAR: CVPR 2023 AI City Challenge Track 3
  • MViTv2: Li et al., CVPR 2022
  • AI City Challenge: NVIDIA, 2023
  • Euro NCAP 2026 OMS Protocol (v1.0)
DMS研究
#DMS #驾驶员行为识别 #M2DAR #Vision Transformer #MViT #SAE L3 #多视角融合
M2DAR 深度解读：多视角权重共享 ViT 驾驶员行为识别——SAE L3 场景下的时序动作定位
https://dapalm.com/2026/09/15/2026-09-15-m2dar-weight-shared-mvit-multi-view-driver-action-ims/
作者
Mars
发布于
2026年9月15日
许可协议