LTC-CGMN：轻量时序卷积+跨粒度状态映射的DMS连续风险评估（Frontiers 2026 论文解读+代码复现）

LTC-CGMN：轻量时序卷积+跨粒度状态映射的DMS连续风险评估

论文信息

项目 内容
标题 Visual measurement network for intelligent cockpit driving state combining lightweight temporal convolution and cross-granularity state mapping
期刊 Frontiers in Mechanical Engineering
年份 2026
DOI 10.3389/fmech.2026.1972239
链接 https://www.frontiersin.org/journals/mechanical-engineering/articles/10.3389/fmech.2026.1972239/full
数据集 3MDAD 多模态驾驶数据集
部署平台 NVIDIA Jetson Orin

1. 核心创新

LTC-CGMN 解决了 DMS 部署的三个核心痛点：

  1. 跨粒度状态映射（CGSM）：对齐细粒度头部/手部特征与粗粒度躯干特征
  2. 轻量时序卷积（LTC）：深度可分离因果膨胀卷积，低算力下捕获时序依赖
  3. 连续风险指数（DRI）：基于 NHTSA-AHP 的知识蒸馏方案，从离散分类走向连续风险估计

1.1 问题定义

现有 DMS 面临”高精度 vs 低延迟”的二元困境：

方法 精度 延迟 参数量 边缘可行性
2D-CNN（单帧） ⚠️ 中 ✅ 低 小 ✅ 可用
3D-CNN（多帧） ✅ 高 ❌ 高 大 ❌ 不可用
ViT（TimeSformer等） ✅ 高 ❌ 很高 很大 ❌ 不可用
TSM/X3D（轻量） ⚠️ 中 ✅ 低 中 ⚠️ 勉强
LTC-CGMN ✅ 高 ✅ 低 小 ✅ 33.6 FPS @ Jetson Orin

1.2 性能指标

指标 LTC-CGMN 基线（TSM） 基线（X3D）
Top-1 准确率 96.37% 93.2% 94.1%
DRI 预测 RMSE 0.058 — —
推理帧率 33.6 FPS 28.5 FPS 22.3 FPS
端到端延迟 < 30ms ~35ms ~45ms
部署平台 Jetson Orin Jetson Orin Jetson Orin

2. 方法详解

2.1 整体架构

graph TD
    A[输入视频帧序列<br/>T=30帧] --> B[CGSM 跨粒度状态映射]
    
    B --> B1[细粒度分支<br/>头部/手部特征]
    B --> B2[粗粒度分支<br/>躯干/全局特征]
    
    B1 --> C[跨注意力对齐]
    B2 --> C
    
    C --> D[LTC 轻量时序卷积]
    
    D --> D1[深度可分离卷积]
    D --> D2[因果膨胀卷积<br/>膨胀率=2,4,8]
    
    D --> E{双头输出}
    
    E --> F[离散状态分类<br/>正常/疲劳/分心/phone等]
    E --> G[连续DRI回归<br/>0-1风险指数]
    
    G --> H[NHTSA-AHP 知识蒸馏]

2.2 CGSM：跨粒度状态映射

核心思想：驾驶状态需要多尺度特征——面部表情（细粒度）+ 躯干姿态（粗粒度）。

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
"""
CGSM: Cross-Granularity State Mapping Module
跨粒度状态映射模块
"""

import torch
import torch.nn as nn
import torch.nn.functional as F

class CGSMModule(nn.Module):
    """
    跨粒度状态映射模块
    
    将细粒度（头部/手部）和粗粒度（躯干/全局）特征
    通过跨注意力机制对齐融合
    """
    
    def __init__(self, fine_dim: int = 256, coarse_dim: int = 512, 
                 hidden_dim: int = 512, num_heads: int = 4):
        super().__init__()
        
        # 细粒度特征投影
        self.fine_proj = nn.Linear(fine_dim, hidden_dim)
        # 粗粒度特征投影
        self.coarse_proj = nn.Linear(coarse_dim, hidden_dim)
        
        # 跨注意力：fine → coarse
        self.cross_attn_f2c = nn.MultiheadAttention(
            hidden_dim, num_heads, batch_first=True
        )
        # 跨注意力：coarse → fine
        self.cross_attn_c2f = nn.MultiheadAttention(
            hidden_dim, num_heads, batch_first=True
        )
        
        # 层归一化
        self.norm1 = nn.LayerNorm(hidden_dim)
        self.norm2 = nn.LayerNorm(hidden_dim)
        
        # FFN
        self.ffn = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim * 4),
            nn.GELU(),
            nn.Linear(hidden_dim * 4, hidden_dim)
        )
        self.norm3 = nn.LayerNorm(hidden_dim)
        
    def forward(self, fine_feat: torch.Tensor, 
                coarse_feat: torch.Tensor) -> torch.Tensor:
        """
        Args:
            fine_feat: 细粒度特征 (B, T, fine_dim)
            coarse_feat: 粗粒度特征 (B, T, coarse_dim)
            
        Returns:
            fused_feat: 融合特征 (B, T, hidden_dim)
        """
        # 投影到统一维度
        fine = self.fine_proj(fine_feat)    # (B, T, hidden)
        coarse = self.coarse_proj(coarse_feat)  # (B, T, hidden)
        
        # 跨注意力：细粒度查询粗粒度
        attn_f2c, _ = self.cross_attn_f2c(
            query=fine, key=coarse, value=coarse
        )
        fine_enhanced = self.norm1(fine + attn_f2c)
        
        # 跨注意力：粗粒度查询细粒度
        attn_c2f, _ = self.cross_attn_c2f(
            query=coarse, key=fine, value=fine
        )
        coarse_enhanced = self.norm2(coarse + attn_c2f)
        
        # 双向融合
        fused = fine_enhanced + coarse_enhanced
        
        # FFN
        fused = self.norm3(fused + self.ffn(fused))
        
        return fused


# 测试
if __name__ == "__main__":
    cgsm = CGSMModule(fine_dim=256, coarse_dim=512, hidden_dim=512)
    
    # 模拟输入：B=2, T=30帧
    fine = torch.randn(2, 30, 256)   # 头部/手部特征序列
    coarse = torch.randn(2, 30, 512) # 躯干/全局特征序列
    
    output = cgsm(fine, coarse)
    print(f"输入: fine={fine.shape}, coarse={coarse.shape}")
    print(f"输出: {output.shape}")  # (2, 30, 512)
    print(f"参数量: {sum(p.numel() for p in cgsm.parameters()) / 1e6:.2f}M")

2.3 LTC：轻量时序卷积

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
"""
LTC: Lightweight Temporal Convolution
深度可分离因果膨胀卷积
"""

import torch
import torch.nn as nn

class LTCBlock(nn.Module):
    """
    轻量时序卷积块
    
    特点：
    1. 深度可分离：减少参数量
    2. 因果性：只看过去，不看未来（在线推理）
    3. 膨胀：扩大感受野不增加参数
    """
    
    def __init__(self, channels: int, 
                 kernel_size: int = 3,
                 dilation: int = 2,
                 dropout: float = 0.1):
        super().__init__()
        
        # 因果卷积：左padding
        padding = (kernel_size - 1) * dilation
        
        # 深度可分离卷积 = 深度卷积 + 逐点卷积
        self.depthwise_conv = nn.Conv1d(
            channels, channels, kernel_size,
            padding=padding, dilation=dilation,
            groups=channels  # 深度卷积
        )
        self.pointwise_conv = nn.Conv1d(channels, channels, 1)
        
        # 归一化和激活
        self.norm = nn.BatchNorm1d(channels)
        self.act = nn.GELU()
        self.dropout = nn.Dropout(dropout)
        
        # 残差连接
        self.residual = nn.Conv1d(channels, channels, 1) \
            if True else nn.Identity()
        
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        Args:
            x: (B, C, T)
        Returns:
            out: (B, C, T)
        """
        residual = self.residual(x)
        
        # 深度卷积（因果）
        out = self.depthwise_conv(x)
        # 裁剪右侧（因果性）
        out = out[:, :, :x.shape[2]]
        
        # 逐点卷积
        out = self.pointwise_conv(out)
        
        # 归一化
        out = self.norm(out)
        out = self.act(out)
        out = self.dropout(out)
        
        # 残差
        return out + residual


class LTCStack(nn.Module):
    """多层 LTC 堆叠，膨胀率递增"""
    
    def __init__(self, channels: int = 512, 
                 num_layers: int = 4,
                 base_dilation: int = 2):
        super().__init__()
        
        self.layers = nn.ModuleList([
            LTCBlock(
                channels=channels,
                kernel_size=3,
                dilation=base_dilation ** i,  # 2, 4, 8, 16
                dropout=0.1
            ) for i in range(num_layers)
        ])
        
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """x: (B, C, T)"""
        for layer in self.layers:
            x = layer(x)
        return x


# 测试
if __name__ == "__main__":
    ltc = LTCStack(channels=512, num_layers=4, base_dilation=2)
    
    # B=2, C=512, T=30
    x = torch.randn(2, 512, 30)
    out = ltc(x)
    print(f"输入: {x.shape}")
    print(f"输出: {out.shape}")
    print(f"参数量: {sum(p.numel() for p in ltc.parameters()) / 1e6:.2f}M")
    
    # 感受野分析
    receptive_field = 1
    for i in range(4):
        rf = 3 * (2 ** i)  # kernel=3, dilation=2^i
        receptive_field += rf - 1
    print(f"理论感受野: {receptive_field} 帧")

2.4 DRI：连续驾驶员风险指数

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
"""
DRI: Driver Risk Index
基于 NHTSA-AHP 的连续风险指数生成
"""

import torch
import torch.nn as nn
import numpy as np

class DRIHead(nn.Module):
    """
    驾驶员风险指数回归头
    
    基于知识蒸馏：教师模型为 NHTSA-AHP 专家规则
    学生模型为神经网络回归器
    """
    
    def __init__(self, feature_dim: int = 512, 
                 num_classes: int = 8):
        super().__init__()
        
        # 离散分类头
        self.classifier = nn.Linear(feature_dim, num_classes)
        
        # 连续 DRI 回归头
        self.dri_regressor = nn.Sequential(
            nn.Linear(feature_dim, 256),
            nn.GELU(),
            nn.Linear(256, 1),
            nn.Sigmoid()  # 0-1 连续值
        )
        
    def forward(self, features: torch.Tensor) -> tuple:
        """
        Args:
            features: (B, T, feature_dim) 时序特征
            
        Returns:
            logits: (B, num_classes) 离散分类
            dri: (B, 1) 连续风险指数 0-1
        """
        # 时序池化（取最后一帧或平均）
        pooled = features.mean(dim=1)  # (B, feature_dim)
        
        logits = self.classifier(pooled)
        dri = self.dri_regressor(pooled)
        
        return logits, dri


# NHTSA-AHP 风险等级映射
NHTSA_RISK_MAP = {
    # (离散状态, 严重程度) → DRI 范围
    'normal_safe': (0.0, 0.15),
    'normal_distracted_light': (0.15, 0.30),
    'phone_call': (0.30, 0.45),
    'phone_text': (0.45, 0.60),
    'drowsy_light': (0.40, 0.55),
    'drowsy_moderate': (0.55, 0.75),
    'drowsy_severe': (0.75, 0.90),
    'unresponsive': (0.90, 1.00),
}

def nhtsa_ahp_teacher(predictions: np.ndarray, 
                       class_names: list) -> float:
    """
    NHTSA-AHP 教师模型：根据分类结果生成连续 DRI
    
    Args:
        predictions: 分类概率分布
        class_names: 类别名称
        
    Returns:
        dri: 连续风险指数 0-1
    """
    dri = 0.0
    for prob, name in zip(predictions, class_names):
        if name in NHTSA_RISK_MAP:
            low, high = NHTSA_RISK_MAP[name]
            # 概率加权
            dri += prob * np.random.uniform(low, high)
    
    return np.clip(dri, 0.0, 1.0)


# 完整模型测试
if __name__ == "__main__":
    from cgsm_module import CGSMModule  # 假设已定义
    
    # 完整 LTC-CGMN 模型
    class LTC_CGMN(nn.Module):
        def __init__(self):
            super().__init__()
            # 假设已有特征提取器
            self.fine_extractor = nn.Linear(224*224*3, 256)  # 简化
            self.coarse_extractor = nn.Linear(224*224*3, 512)
            
            self.cgsm = CGSMModule(256, 512, 512, num_heads=4)
            self.ltc = LTCStack(512, num_layers=4, base_dilation=2)
            self.head = DRIHead(512, num_classes=8)
            
        def forward(self, fine_input, coarse_input):
            # B, T, ...
            B, T = fine_input.shape[:2]
            
            # 特征提取（逐帧）
            fine_feat = self.fine_extractor(fine_input.view(B*T, -1))
            coarse_feat = self.coarse_extractor(coarse_input.view(B*T, -1))
            
            fine_feat = fine_feat.view(B, T, -1)
            coarse_feat = coarse_feat.view(B, T, -1)
            
            # CGSM 融合
            fused = self.cgsm(fine_feat, coarse_feat)  # (B, T, 512)
            
            # LTC 时序建模
            fused_t = fused.transpose(1, 2)  # (B, 512, T)
            temporal_out = self.ltc(fused_t)  # (B, 512, T)
            temporal_out = temporal_out.transpose(1, 2)  # (B, T, 512)
            
            # 双头输出
            logits, dri = self.head(temporal_out)
            
            return logits, dri
    
    model = LTC_CGMN()
    
    # 模拟输入
    fine_input = torch.randn(2, 30, 3, 224, 224)  # 裁剪后的面部/手部
    coarse_input = torch.randn(2, 30, 3, 224, 224)  # 全身/躯干
    
    logits, dri = model(fine_input, coarse_input)
    
    print(f"分类 logits: {logits.shape}")  # (2, 8)
    print(f"连续 DRI: {dri.shape}")        # (2, 1)
    print(f"DRI 值: {dri.detach().numpy().flatten()}")
    print(f"总参数量: {sum(p.numel() for p in model.parameters()) / 1e6:.2f}M")

3. 实验结果分析

3.1 3MDAD 数据集

属性 数值
模态 多模态（RGB + 深度 + 热成像）
类别数 8（正常、phone、分心、疲劳等）
被试数 20+
场景 模拟器 + 真实驾驶

3.2 性能对比

方法 Top-1 (%) FPS 参数量 连续DRI
2D-CNN (ResNet-18) 89.5 45.2 11.2M ❌
TSM 93.2 28.5 24.8M ❌
X3D-M 94.1 22.3 3.8M ❌
TimeSformer 95.8 8.5 121M ❌
LTC-CGMN 96.37 33.6 ~15M ✅ RMSE=0.058

3.3 Jetson Orin 部署结果

指标 数值
推理帧率 33.6 FPS
端到端延迟 < 30ms
内存占用 < 2GB
功耗 < 15W
精度 FP16 混合精度

4. 对 IMS 开发的启示

4.1 可落地的开发建议

优先级 建议 理由
🔴 P0 从离散分类升级到连续 DRI 离散分类有阶跃跳变，连续 DRI 更平滑、更适合 ADAS 联动
🔴 P0 采用因果时序卷积 在线推理必须因果性，膨胀卷积扩大感受野不增参数
🟡 P1 跨粒度特征对齐 面部细粒度 + 躯干粗粒度的多尺度融合
🟡 P1 NHTSA-AHP 知识蒸馏 用专家规则做教师，解决标注数据不足
🟢 P2 Jetson Orin 部署验证 FP16 量化，30+ FPS 实时推理

4.2 DRI 与 ADAS 联动

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
"""
DRI 驱动的 ADAS 联动策略
"""

class DRIBasedADASControl:
    """
    基于连续 DRI 的 ADAS 控制策略
    """
    
    def __init__(self):
        self.thresholds = {
            'safe': 0.15,        # DRI < 0.15: 正常
            'caution': 0.30,    # 0.15-0.30: 轻微风险
            'warning': 0.50,    # 0.30-0.50: 中等风险
            'alert': 0.70,      # 0.50-0.70: 高风险
            'critical': 0.85,   # 0.70-0.85: 严重风险
            'emergency': 0.90,  # > 0.90: 紧急
        }
        
    def decide_action(self, dri: float, speed: float, 
                      road_type: str) -> dict:
        """
        根据 DRI 决定 ADAS 行动
        
        Returns:
            {
                'hud_alert_level': 0-3,
                'audio_warning': bool,
                'auto_slowdown': bool,
                'takeover_request': bool,
                'pull_over': bool,
            }
        """
        if dri < self.thresholds['caution']:
            return self._no_action()
        
        if dri < self.thresholds['warning']:
            return self._level1_alert(dri)
        
        if dri < self.thresholds['alert']:
            return self._level2_warning(dri, speed)
        
        if dri < self.thresholds['critical']:
            return self._level3_alert(dri, speed, road_type)
        
        if dri < self.thresholds['emergency']:
            return self._takeover_request(dri, speed)
        
        return self._emergency_stop(dri, speed)
    
    def _no_action(self):
        return {'hud_alert_level': 0, 'audio_warning': False,
                'auto_slowdown': False, 'takeover_request': False,
                'pull_over': False}
    
    def _level1_alert(self, dri):
        return {'hud_alert_level': 1, 'audio_warning': False,
                'auto_slowdown': False, 'takeover_request': False,
                'pull_over': False, 'dri': dri}
    
    def _level2_warning(self, dri, speed):
        return {'hud_alert_level': 2, 'audio_warning': True,
                'auto_slowdown': speed > 80, 'takeover_request': False,
                'pull_over': False, 'dri': dri}
    
    def _level3_alert(self, dri, speed, road_type):
        return {'hud_alert_level': 3, 'audio_warning': True,
                'auto_slowdown': True, 'takeover_request': True,
                'pull_over': False, 'dri': dri, 
                'slowdown_target': max(speed * 0.7, 40)}
    
    def _takeover_request(self, dri, speed):
        return {'hud_alert_level': 3, 'audio_warning': True,
                'auto_slowdown': True, 'takeover_request': True,
                'pull_over': False, 'dri': dri,
                'slowdown_target': max(speed * 0.5, 30)}
    
    def _emergency_stop(self, dri, speed):
        return {'hud_alert_level': 3, 'audio_warning': True,
                'auto_slowdown': True, 'takeover_request': True,
                'pull_over': True, 'dri': dri,
                'slowdown_target': 0}


# 测试
if __name__ == "__main__":
    controller = DRIBasedADASControl()
    
    # 场景1：正常驾驶
    action = controller.decide_action(dri=0.10, speed=100, road_type='highway')
    print(f"正常 (DRI=0.10): {action}")
    
    # 场景2：轻度分心
    action = controller.decide_action(dri=0.35, speed=80, road_type='urban')
    print(f"分心 (DRI=0.35): {action}")
    
    # 场景3：严重疲劳
    action = controller.decide_action(dri=0.85, speed=120, road_type='highway')
    print(f"严重疲劳 (DRI=0.85): {action}")
    
    # 场景4：无响应
    action = controller.decide_action(dri=0.95, speed=60, road_type='urban')
    print(f"无响应 (DRI=0.95): {action}")

4.3 部署优化路线

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
Phase 1: 模型训练
├── 数据集：3MDAD + 自有标注数据
├── 模型：LTC-CGMN（CGSM + LTC + DRI Head）
├── 损失：分类 CE + DRI MSE + 知识蒸馏 KL
└── 指标：Top-1 > 95%，DRI RMSE < 0.08

Phase 2: 量化部署
├── FP32 → FP16（Jetson Orin TensorRT）
├── 量化感知训练（QAT）
├── 算子融合优化
└── 指标：FPS > 30，延迟 < 30ms

Phase 3: ADAS 联动
├── DRI → CAN-FD 消息定义
├── ADAS 控制策略映射
├── HIL 仿真验证
└── 指标：联动响应 < 100ms

5. 技术路线判断

5.1 核心洞察

LTC-CGMN 的关键创新在于从离散到连续的风险评估范式转变：

  1. 离散分类的问题：状态切换有阶跃，ADAS 无法平滑响应
  2. 连续 DRI 的优势：0-1 连续值，ADAS 可以设置多级阈值平滑过渡
  3. 知识蒸馏的价值：用专家规则（NHTSA-AHP）解决标注数据不足

5.2 与 Euro NCAP 的关系

Euro NCAP DSM 评估目前基于场景级判定（检测到→通过/未通过）。但连续 DRI 可以：

  • 提供更细粒度的检测时延评估
  • 支持更精细的误报/漏报率统计
  • 为未来的功能准备度评估（DFR）提供基础

5.3 对比其他方案

方案 离散/连续 时序建模 边缘部署 DRI
ResNet-18 单帧 离散 ❌ ✅ 45 FPS ❌
3D-CNN 离散 ✅ ❌ 8 FPS ❌
TSM 离散 ✅ ⚠️ 28 FPS ❌
LTC-CGMN 两者 ✅ ✅ 33 FPS ✅
TimeSformer 离散 ✅ ❌ 8 FPS ❌

6. 参考

论文解读
#DMS #论文解读 #边缘部署 #时序卷积 #跨粒度 #Jetson Orin #连续风险估计
LTC-CGMN：轻量时序卷积+跨粒度状态映射的DMS连续风险评估（Frontiers 2026 论文解读+代码复现）
https://dapalm.com/2026/10/06/2026-10-06-018-ltc-cgmn-lightweight-temporal-convolution-dms-frontiers2026/
作者
Mars
发布于
2026年10月6日
许可协议