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. 核心创新

1.1 问题定义

DMS 边缘部署面临的核心矛盾：

需求	约束	现有方案问题
高精度	≥95% 分类准确率	3D-CNN/ViT 精度高但计算量大
低延迟	≥30 FPS, ≤50ms 端到端	2D-CNN 快但丢失时序信息
连续风险	连续 DRI 而非离散分类	传统分类无法反映渐进风险
边缘算力	Jetson Orin 级别	重模型无法部署

1.2 LTC-CGMN 三大创新

1. CGSM (Cross-Granularity State Mapping Module)：通过交叉注意力将细粒度头部/手部特征与粗粒度躯干特征对齐
2. LTC (Lightweight Temporal Convolution)：深度可分离因果膨胀卷积，低计算成本捕获时序依赖
3. NHTSA-AHP 知识蒸馏：从离散驾驶行为生成连续 Driver Risk Index (DRI)

1.3 性能指标

指标	LTC-CGMN	对比
Top-1 准确率	96.37%	3MDAD 数据集
DRI RMSE	0.058	连续风险指数预测
推理速度	33.6 FPS	NVIDIA Jetson Orin
端到端延迟	< 30ms	满足实时要求

---

2. 方法详解

2.1 整体架构

graph TD
    A[多模态输入<br/>RGB+IR+深度] --> B[特征提取]
    
    B --> B1[细粒度分支<br/>头部+手部特征]
    B --> B2[粗粒度分支<br/>躯干+整体姿态]
    
    B1 --> C[CGSM 跨粒度状态映射<br/>Cross-Attention 对齐]
    B2 --> C
    
    C --> D[LTC 轻量时序卷积<br/>因果膨胀卷积]
    
    D --> E1[离散状态分类<br/>正常/疲劳/分心/危险]
    D --> E2[连续 DRI 回归<br/>NHTSA-AHP 知识蒸馏]

2.2 CGSM 模块

"""
CGSM (Cross-Granularity State Mapping) 模块复现
通过交叉注意力将细粒度特征与粗粒度特征对齐
"""

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

class CGSM(nn.Module):
    """
    跨粒度状态映射模块
    
    论文 Section 3.2:
    - 细粒度分支：头部+手部区域特征（小感受野，高分辨率）
    - 粗粒度分支：躯干+整体姿态特征（大感受野，低分辨率）
    - Cross-Attention：细粒度 query, 粗粒度 key/value
    """
    
    def __init__(self, fine_dim: int = 256, coarse_dim: int = 256,
                 num_heads: int = 4, dropout: float = 0.1):
        super().__init__()
        
        # 多头交叉注意力
        self.cross_attn = nn.MultiheadAttention(
            embed_dim=fine_dim,
            num_heads=num_heads,
            dropout=dropout,
            batch_first=True
        )
        
        # 层归一化
        self.norm_fine = nn.LayerNorm(fine_dim)
        self.norm_coarse = nn.LayerNorm(coarse_dim)
        self.norm_out = nn.LayerNorm(fine_dim)
        
        # FFN
        self.ffn = nn.Sequential(
            nn.Linear(fine_dim, fine_dim * 4),
            nn.GELU(),
            nn.Dropout(dropout),
            nn.Linear(fine_dim * 4, fine_dim),
            nn.Dropout(dropout)
        )
        
        # 维度对齐（如果细粒度和粗粒度维度不同）
        self.proj = nn.Linear(coarse_dim, fine_dim) if fine_dim != coarse_dim else nn.Identity()
        
    def forward(self, fine_feat: torch.Tensor, coarse_feat: torch.Tensor) -> torch.Tensor:
        """
        Args:
            fine_feat: 细粒度特征 (B, N_fine, C_fine)
            coarse_feat: 粗粒度特征 (B, N_coarse, C_coarse)
            
        Returns:
            aligned_feat: 跨粒度对齐特征 (B, N_fine, C_fine)
        """
        # 维度对齐
        coarse_aligned = self.proj(coarse_feat)
        
        # Cross-Attention: fine as query, coarse as key/value
        attn_out, _ = self.cross_attn(
            query=self.norm_fine(fine_feat),
            key=self.norm_coarse(coarse_aligned),
            value=self.norm_coarse(coarse_aligned)
        )
        
        # 残差连接
        out = fine_feat + attn_out
        
        # FFN + 残差
        out = self.norm_out(out + self.ffn(out))
        
        return out


# 测试
if __name__ == "__main__":
    cgsm = CGSM(fine_dim=256, coarse_dim=256, num_heads=4)
    
    # 模拟特征
    fine = torch.randn(2, 16, 256)   # 16个细粒度 token
    coarse = torch.randn(2, 4, 256)   # 4个粗粒度 token
    
    out = cgsm(fine, coarse)
    print(f"输入: fine={fine.shape}, coarse={coarse.shape}")
    print(f"输出: {out.shape}")
    print(f"参数量: {sum(p.numel() for p in cgsm.parameters()):,}")

2.3 LTC 轻量时序卷积

"""
LTC (Lightweight Temporal Convolution) 模块复现
深度可分离因果膨胀卷积
"""

class CausalDepthwiseConv1d(nn.Module):
    """
    因果深度可分离一维卷积
    
    特点：
    1. 因果性：只看当前和过去帧，不看未来（在线推理）
    2. 深度可分离：depthwise + pointwise
    3. 膨胀：扩大感受野而不增加参数
    """
    
    def __init__(self, channels: int, kernel_size: int = 3, 
                 dilation: int = 1):
        super().__init__()
        
        # 因果 padding：只在前面 pad
        self.causal_pad = (kernel_size - 1) * dilation
        
        # Depthwise 卷积（每通道独立）
        self.depthwise = nn.Conv1d(
            channels, channels, kernel_size,
            dilation=dilation,
            padding=0,  # 手动 pad 实现因果性
            groups=channels,
            bias=False
        )
        
        # Pointwise 卷积（通道混合）
        self.pointwise = nn.Conv1d(channels, channels, 1, bias=False)
        
        self.bn = nn.BatchNorm1d(channels)
        self.act = nn.GELU()
        
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        Args:
            x: (B, C, T) 时序特征
        
        Returns:
            out: (B, C, T) 因果卷积输出
        """
        # 因果 padding（只在前面）
        x = F.pad(x, (self.causal_pad, 0))
        
        # Depthwise
        x = self.depthwise(x)
        
        # Pointwise
        x = self.pointwise(x)
        
        x = self.bn(x)
        x = self.act(x)
        
        return x


class LTCBlock(nn.Module):
    """
    LTC 轻量时序卷积块
    
    多层膨胀卷积，指数递增膨胀率
    感受野 = sum(dilation * (kernel_size - 1)) for each layer
    """
    
    def __init__(self, channels: int, num_layers: int = 4,
                 kernel_size: int = 3):
        super().__init__()
        
        self.layers = nn.ModuleList()
        for i in range(num_layers):
            dilation = 2 ** i  # 1, 2, 4, 8
            self.layers.append(CausalDepthwiseConv1d(
                channels=channels,
                kernel_size=kernel_size,
                dilation=dilation
            ))
            
        # 残差连接
        self.residual_proj = nn.Conv1d(channels, channels, 1, bias=False)
        
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        Args:
            x: (B, C, T)
        Returns:
            out: (B, C, T)
        """
        residual = self.residual_proj(x)
        
        for layer in self.layers:
            x = layer(x)
            x = x + residual  # 每层残差
            residual = x
            
        return x


# 测试
if __name__ == "__main__":
    ltc = LTCBlock(channels=256, num_layers=4, kernel_size=3)
    
    # 模拟时序特征：batch=2, channels=256, time=30
    x = torch.randn(2, 256, 30)
    out = ltc(x)
    
    print(f"输入: {x.shape}")
    print(f"输出: {out.shape}")
    
    # 计算感受野
    receptive_field = sum(2**i * 2 for i in range(4))
    print(f"感受野: {receptive_field} 帧")
    
    # 参数量
    params = sum(p.numel() for p in ltc.parameters())
    print(f"参数量: {params:,} ({params/1e3:.1f}K)")
    
    # 推理速度测试
    import time
    ltc.eval()
    with torch.no_grad():
        # 预热
        for _ in range(10):
            _ = ltc(x)
        
        # 计时
        start = time.perf_counter()
        for _ in range(100):
            _ = ltc(x)
        elapsed = time.perf_counter() - start
        fps = 100 / elapsed
        print(f"推理速度: {fps:.1f} FPS (CPU)")

2.4 NHTSA-AHP 连续 DRI

"""
NHTSA-AHP 引导的连续 Driver Risk Index (DRI) 生成
通过层次分析法将离散驾驶行为映射为连续风险值
"""

class NHTSA_AHP_DRI:
    """
    基于 NHTSA 标准 + AHP 层次分析的连续风险指数
    
    NHTSA 驾驶员状态分级：
    0: Normal Driving (安全)
    1: Visual Distraction (视觉分心)
    2: Cognitive Distraction (认知分心)
    3: Manual Distraction (操作分心)
    4: Fatigue/Drunk (疲劳/酒驾)
    5: Emergency (紧急)
    """
    
    # AHP 权重矩阵（经验/专家设定）
    AHP_WEIGHTS = {
        'eye_closure': 0.30,      # 闭眼
        'gaze_off_road': 0.25,    # 视线偏离
        'head_pose': 0.15,        # 头部姿态
        'phone_use': 0.15,        # 手机使用
        'yawning': 0.10,          # 打哈欠
        'body_posture': 0.05      # 身体姿态
    }
    
    # 离散到连续的映射
    STATE_TO_BASE_DRI = {
        0: 0.1,   # Normal
        1: 0.3,   # Visual Distraction
        2: 0.45,  # Cognitive Distraction
        3: 0.55,  # Manual Distraction
        4: 0.75,  # Fatigue
        5: 0.95   # Emergency
    }
    
    def __init__(self):
        # 知识蒸馏目标：从分类 logits 生成连续 DRI
        self.temperature = 2.0  # 蒸馏温度
        
    def compute_dri(self, logits: torch.Tensor, 
                    features: dict = None) -> torch.Tensor:
        """
        从分类 logits 生成连续 DRI
        
        Args:
            logits: 分类输出 (B, num_classes)
            features: 可选的细粒度特征权重
            
        Returns:
            dri: 连续风险指数 (B,) 范围 [0, 1]
        """
        # Softmax with temperature
        probs = F.softmax(logits / self.temperature, dim=1)
        
        # 加权求和：离散类别 -> 连续 DRI
        state_values = torch.tensor(
            list(self.STATE_TO_BASE_DRI.values()),
            device=logits.device,
            dtype=logits.dtype
        )
        
        dri = torch.sum(probs * state_values, dim=1)
        
        # 如果有细粒度特征，进行微调
        if features:
            adjustment = 0.0
            for feat_name, weight in self.AHP_WEIGHTS.items():
                if feat_name in features:
                    # 特征值 [0,1]，越大越危险
                    adjustment += weight * features[feat_name]
            dri = 0.7 * dri + 0.3 * adjustment  # 70% 离散 + 30% 细粒度
        
        return dri
    
    def knowledge_distillation_loss(self, student_dri: torch.Tensor,
                                     teacher_label: int,
                                     soft_label: float = None) -> torch.Tensor:
        """
        知识蒸馏损失
        
        Args:
            student_dri: 学生模型预测的 DRI
            teacher_label: 离散类别标签
            soft_label: 软标签（如果有专家标注）
            
        Returns:
            loss: 蒸馏损失
        """
        # 硬标签目标
        hard_target = self.STATE_TO_BASE_DRI[teacher_label]
        hard_loss = F.mse_loss(student_dri, 
                               torch.full_like(student_dri, hard_target))
        
        if soft_label is not None:
            # 软标签目标（专家连续标注）
            soft_loss = F.mse_loss(student_dri,
                                   torch.full_like(student_dri, soft_label))
            return 0.5 * hard_loss + 0.5 * soft_loss
        
        return hard_loss


# 测试
import torch.nn.functional as F

if __name__ == "__main__":
    dri_generator = NHTSA_AHP_DRI()
    
    # 模拟分类输出
    logits = torch.tensor([
        [5.0, 0.5, 0.3, 0.2, 0.1, 0.0],  # 正常驾驶
        [0.5, 1.0, 0.3, 0.2, 5.0, 0.1],  # 疲劳
        [0.5, 0.3, 0.2, 0.1, 0.0, 5.0],  # 紧急
    ])
    
    dri = dri_generator.compute_dri(logits)
    
    for i, d in enumerate(dri):
        print(f"样本 {i}: DRI = {d.item():.3f}")
    
    # 蒸馏损失
    loss = dri_generator.knowledge_distillation_loss(
        dri[1], teacher_label=4, soft_label=0.78
    )
    print(f"\n蒸馏损失: {loss.item():.4f}")

---

3. 完整 LTC-CGMN 模型

"""
LTC-CGMN 完整模型
"""

class LTC_CGMN(nn.Module):
    """
    LTC-CGMN: Lightweight Temporal Convolution + Cross-Granularity Mapping Network
    
    架构：
    1. Backbone: 轻量 2D-CNN 提取空间特征
    2. CGSM: 跨粒度状态映射
    3. LTC: 轻量时序卷积
    4. 双头输出：离散分类 + 连续 DRI
    """
    
    def __init__(self, num_classes: int = 6, backbone_dim: int = 256):
        super().__init__()
        
        # Backbone: 轻量 CNN（MobileNetV3 风格）
        self.backbone = nn.Sequential(
            nn.Conv2d(3, 32, 3, stride=2, padding=1),
            nn.BatchNorm2d(32), nn.GELU(),
            nn.Conv2d(32, 64, 3, stride=2, padding=1),
            nn.BatchNorm2d(64), nn.GELU(),
            nn.Conv2d(64, 128, 3, stride=2, padding=1),
            nn.BatchNorm2d(128), nn.GELU(),
            nn.Conv2d(128, backbone_dim, 3, stride=2, padding=1),
            nn.BatchNorm2d(backbone_dim), nn.GELU(),
            nn.AdaptiveAvgPool2d((4, 4))
        )
        
        # 区域划分：头部/手部（细粒度）vs 躯干（粗粒度）
        self.fine_proj = nn.Linear(backbone_dim * 4, backbone_dim)
        self.coarse_proj = nn.Linear(backbone_dim * 4, backbone_dim)
        
        # CGSM
        self.cgsm = CGSM(backbone_dim, backbone_dim, num_heads=4)
        
        # 时序维度池化
        self.temporal_pool = nn.AdaptiveAvgPool1d(1)
        
        # LTC 时序卷积
        self.ltc = LTCBlock(backbone_dim, num_layers=4, kernel_size=3)
        
        # 分类头
        self.classifier = nn.Sequential(
            nn.Linear(backbone_dim, backbone_dim // 2),
            nn.GELU(),
            nn.Dropout(0.2),
            nn.Linear(backbone_dim // 2, num_classes)
        )
        
        # DRI 回归头
        self.dri_head = nn.Sequential(
            nn.Linear(backbone_dim, backbone_dim // 2),
            nn.GELU(),
            nn.Dropout(0.2),
            nn.Linear(backbone_dim // 2, 1),
            nn.Sigmoid()
        )
        
    def forward(self, x: torch.Tensor) -> tuple:
        """
        Args:
            x: (B, T, C, H, W) 视频输入
            
        Returns:
            logits: (B, num_classes) 离散分类
            dri: (B,) 连续风险指数
        """
        B, T, C, H, W = x.shape
        
        # Backbone 逐帧提取
        x = x.view(B * T, C, H, W)
        feat = self.backbone(x)  # (B*T, C, h, w)
        
        # 展平空间
        feat = feat.view(B, T, -1)  # (B, T, C*4*4)
        
        # 分为细粒度和粗粒度
        fine_feat = self.fine_proj(feat)   # (B, T, C)
        coarse_feat = self.coarse_proj(feat)  # (B, T, C)
        
        # CGSM 跨粒度映射
        aligned = self.cgsm(fine_feat, coarse_feat)  # (B, T, C)
        
        # 转为 (B, C, T) 供时序卷积
        aligned = aligned.permute(0, 2, 1)  # (B, C, T)
        
        # LTC 时序建模
        temporal_feat = self.ltc(aligned)  # (B, C, T)
        
        # 时序池化
        pooled = self.temporal_pool(temporal_feat).squeeze(-1)  # (B, C)
        
        # 双头输出
        logits = self.classifier(pooled)
        dri = self.dri_head(pooled).squeeze(-1)
        
        return logits, dri


# 测试
if __name__ == "__main__":
    model = LTC_CGMN(num_classes=6, backbone_dim=256)
    
    # 模拟输入：batch=2, time=16, channels=3, H=224, W=224
    x = torch.randn(2, 16, 3, 224, 224)
    
    with torch.no_grad():
        logits, dri = model(x)
    
    print(f"输入: {x.shape}")
    print(f"分类输出: {logits.shape}")
    print(f"DRI 输出: {dri.shape}")
    print(f"DRI 值: {dri.tolist()}")
    
    # 参数量统计
    total_params = sum(p.numel() for p in model.parameters())
    print(f"\n总参数量: {total_params:,} ({total_params/1e6:.2f}M)")
    
    # 推理速度测试
    model.eval()
    import time
    with torch.no_grad():
        for _ in range(5):
            _ = model(x)
        
        start = time.perf_counter()
        for _ in range(20):
            _ = model(x)
        elapsed = time.perf_counter() - start
        fps = 20 * 16 / elapsed  # 每秒处理帧数
        print(f"推理速度: {fps:.1f} FPS (CPU, 估计)")

---

4. 对 IMS 开发的启示

4.1 直接可落地的开发建议

优先级	建议	具体行动
🔴 P0	采用因果膨胀卷积替代 3D-CNN	在现有 DMS pipeline 中用 LTCBlock 替代 3D-CNN，减少 80% 计算量
🔴 P0	引入连续 DRI 替代离散分类	从 NHTSA-AHP 映射生成 DRI，避免风险跳变
🟡 P1	跨粒度特征对齐	头部/手部特征与躯干特征做 Cross-Attention
🟡 P1	Jetson Orin 部署验证	在 Orin 上测试模型推理速度，目标 ≥30 FPS
🟢 P2	3MDAD 数据集评估	下载 3MDAD 数据集，验证模型泛化性

4.2 与现有方案的对比

方案	准确率	FPS	参数量	连续DRI	边缘部署
LTC-CGMN	96.37%	33.6	~5M	✅	✅ Jetson Orin
3D-CNN (I3D)	~95%	8-12	~25M	❌	❌ 太重
2D-CNN+LSTM	~93%	25	~15M	❌	⚠️ 勉强
ViT (TimeSformer)	~96%	5-8	~100M	❌	❌ 太重
TSM+ResNet	~94%	30	~10M	❌	✅

4.3 部署配置建议

配置项	推荐值	说明
平台	NVIDIA Jetson Orin	27 TOPS, 8GB
输入分辨率	224×224	精度与速度平衡
时序长度	16帧	约0.5秒@30fps
Backbone	MobileNetV3-Small	可替换为更轻量backbone
FP16	✅ 启用	TensorRT FP16 加速
Batch Size	1	实时推理

---

5. 参考

• 论文原文
• 3MDAD 数据集
• NHTSA Driver Distraction Guidelines
• NVIDIA Jetson Orin