CAMERA框架：多模态自适应风险预测——驾驶员注意力熵驱动的动态告警阈值

论文信息

项目	内容
标题	Eyes on the Road, Mind Beyond Vision: Context-Aware Multi-modal Enhanced Risk Anticipation
会议	ACM Multimedia 2025
框架名	CAMERA (Context-Aware Multi-modal Enhanced Risk Anticipation)
数据集	DADA-2000, CCD, DAD, A3D
关键词	accident anticipation, driver attention, gaze entropy, adaptive threshold

核心创新

自适应阈值：从固定50%到场景+认知驱动

传统方法: P(accident) > 50% → 警报
           ↑ 高速公路50% vs 城市路口50% 语义完全不同

CAMERA: threshold = f(scene_complexity, gaze_entropy)
         场景复杂 + 注意力分散 → 降低阈值（更敏感）
         场景简单 + 注意力集中 → 提高阈值（减少误报）

方法	阈值策略	误报率	漏报率
固定50%	静态	高	低
场景自适应	基于交通密度	中	中
CAMERA	场景+注视熵	低	低

方法详解

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

class GazeEntropyCalculator(nn.Module):
    """
    注视熵计算模块
    
    使用Shannon熵量化驾驶员注视分布的不确定性:
    - 低熵: 注视集中在一点（专注或隧道视觉）
    - 高熵: 注视分散（搜索或困惑）
    
    论文Section 3.3: Adaptive Thresholding
    """
    
    def __init__(self, num_zones: int = 16):
        super().__init__()
        self.num_zones = num_zones
        
    def forward(self, gaze_map: torch.Tensor) -> torch.Tensor:
        """
        Args:
            gaze_map: (B, 1, H, W) 注视热力图
            
        Returns:
            entropy: (B,) 注视熵值 [0, log(num_zones)]
        """
        B = gaze_map.size(0)
        
        # 将注视图划分为num_zones个区域
        gz = F.adaptive_avg_pool2d(gaze_map, (4, 4))  # (B, 1, 4, 4)
        gz = gz.flatten(1)  # (B, 16)
        
        # 归一化为概率分布
        prob = F.softmax(gz, dim=1)  # (B, 16)
        
        # Shannon熵
        entropy = -(prob * torch.log(prob + 1e-8)).sum(dim=1)
        
        # 归一化到 [0, 1]
        max_entropy = torch.log(torch.tensor(float(self.num_zones)))
        normalized_entropy = entropy / max_entropy
        
        return normalized_entropy


class SceneComplexityEstimator(nn.Module):
    """
    场景复杂度估计
    
    基于交通密度、天气、道路类型等估计场景复杂度
    """
    
    def __init__(self, embed_dim: int = 128):
        super().__init__()
        # 视觉特征提取
        self.visual_encoder = nn.Sequential(
            nn.Conv2d(3, 32, 3, stride=2, padding=1), nn.ReLU(),
            nn.Conv2d(32, 64, 3, stride=2, padding=1), nn.ReLU(),
            nn.Conv2d(64, 128, 3, stride=2, padding=1), nn.ReLU(),
            nn.AdaptiveAvgPool2d(1),
            nn.Flatten()
        )
        # 文本标注编码（天气/道路类型）
        self.text_encoder = nn.Embedding(20, embed_dim)
        # 融合
        self.fusion = nn.Sequential(
            nn.Linear(128 + embed_dim, 64),
            nn.ReLU(),
            nn.Linear(64, 1),
            nn.Sigmoid()
        )
        
    def forward(self, scene_img: torch.Tensor,
                text_label: torch.Tensor) -> torch.Tensor:
        """
        Returns:
            complexity: (B,) 场景复杂度 [0, 1]
        """
        visual_feat = self.visual_encoder(scene_img)
        text_feat = self.text_encoder(text_label)
        fused = torch.cat([visual_feat, text_feat], dim=1)
        return self.fusion(fused).squeeze(-1)


class AdaptiveThreshold(nn.Module):
    """
    自适应风险阈值
    
    threshold = base - α * scene_complexity - β * gaze_entropy
    
    当场景复杂或注意力分散时，降低阈值（更敏感）
    """
    
    def __init__(self, base_threshold: float = 0.5,
                 alpha: float = 0.15, beta: float = 0.20):
        super().__init__()
        self.base = base_threshold
        self.alpha = alpha
        self.beta = beta
        
    def forward(self, scene_complexity: torch.Tensor,
                gaze_entropy: torch.Tensor) -> torch.Tensor:
        """
        Returns:
            threshold: (B,) 自适应风险阈值
        """
        threshold = self.base - self.alpha * scene_complexity \
                          - self.beta * gaze_entropy
        return torch.clamp(threshold, 0.15, 0.85)


class BiGRUTemporalModel(nn.Module):
    """
    双向GRU时序推理
    
    3-5秒窗口内分析事件序列，识别碰撞前兆
    """
    
    def __init__(self, input_dim: int = 256, hidden_dim: int = 128,
                 num_layers: int = 2, num_classes: int = 1):
        super().__init__()
        self.bigru = nn.GRU(
            input_dim, hidden_dim, num_layers=num_layers,
            batch_first=True, bidirectional=True, dropout=0.2
        )
        self.classifier = nn.Sequential(
            nn.Linear(hidden_dim * 2, 64),
            nn.ReLU(),
            nn.Linear(64, num_classes),
            nn.Sigmoid()
        )
        
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        Args:
            x: (B, T, input_dim) 时序特征序列
            
        Returns:
            risk: (B,) 风险概率 [0, 1]
        """
        out, _ = self.bigru(x)
        # 取最后时刻输出
        risk = self.classifier(out[:, -1, :]).squeeze(-1)
        return risk


class CAMERAFramework(nn.Module):
    """
    CAMERA: 完整框架
    
    架构:
      [Dashcam视频] → [多模态特征提取MFE] → [自适应分层融合AHF]
      [文本标注]   ↗                        ↓
      [注视图]    ↗                        [Bi-GRU时序推理]
                                            ↓
                                    [自适应阈值判定]
                                            ↓
                                    [Geo-Context VLM告警]
    
    论文Section 3: Full Framework
    """
    
    def __init__(self, embed_dim: int = 256):
        super().__init__()
        
        # 多模态特征提取
        self.visual_encoder = nn.Sequential(
            nn.Conv2d(3, 32, 3, stride=2, padding=1), nn.ReLU(),
            nn.Conv2d(32, 64, 3, stride=2, padding=1), nn.ReLU(),
            nn.Conv2d(64, 128, 3, stride=2, padding=1), nn.ReLU(),
            nn.AdaptiveAvgPool2d((4, 4)),
            nn.Flatten(),
            nn.Linear(128 * 16, embed_dim)
        )
        
        # 注意图编码
        self.gaze_encoder = nn.Sequential(
            nn.Conv2d(1, 16, 3, stride=2, padding=1), nn.ReLU(),
            nn.Conv2d(16, 32, 3, stride=2, padding=1), nn.ReLU(),
            nn.AdaptiveAvgPool2d((4, 4)),
            nn.Flatten(),
            nn.Linear(32 * 16, embed_dim)
        )
        
        # 文本编码
        self.text_encoder = nn.Embedding(50, embed_dim)
        
        # 注视熵计算
        self.gaze_entropy = GazeEntropyCalculator()
        
        # 场景复杂度
        self.scene_complexity = SceneComplexityEstimator()
        
        # 自适应阈值
        self.adaptive_threshold = AdaptiveThreshold()
        
        # Bi-GRU时序推理
        self.temporal_model = BiGRUTemporalModel(
            input_dim=embed_dim * 3,
            hidden_dim=128,
            num_layers=2
        )
        
    def forward(self, video_seq: torch.Tensor, gaze_seq: torch.Tensor,
                text_labels: torch.Tensor, scene_imgs: torch.Tensor):
        """
        Args:
            video_seq: (B, T, 3, 224, 224) 视频帧序列
            gaze_seq: (B, T, 1, 56, 56) 注视热力图序列
            text_labels: (B,) 环境文本标签
            scene_imgs: (B, 3, 224, 224) 当前场景图像
            
        Returns:
            dict: {
                'risk': (B,) 风险概率,
                'threshold': (B,) 自适应阈值,
                'alert': (B,) 是否告警,
                'gaze_entropy': (B,) 注视熵,
                'complexity': (B,) 场景复杂度
            }
        """
        B, T = video_seq.shape[:2]
        
        # 多模态特征提取
        visual_feats = []
        gaze_feats = []
        for t in range(T):
            vf = self.visual_encoder(video_seq[:, t])
            gf = self.gaze_encoder(gaze_seq[:, t])
            visual_feats.append(vf)
            gaze_feats.append(gf)
        
        visual_feats = torch.stack(visual_feats, dim=1)  # (B, T, D)
        gaze_feats = torch.stack(gaze_feats, dim=1)  # (B, T, D)
        
        text_feats = self.text_encoder(text_labels)  # (B, D)
        text_feats = text_feats.unsqueeze(1).expand(-1, T, -1)  # (B, T, D)
        
        # 融合
        fused = torch.cat([visual_feats, gaze_feats, text_feats], dim=-1)
        
        # 时序推理
        risk = self.temporal_model(fused)  # (B,)
        
        # 自适应阈值
        gaze_ent = self.gaze_entropy(gaze_seq[:, -1])  # 当前注视熵
        complexity = self.scene_complexity(scene_imgs, text_labels)
        threshold = self.adaptive_threshold(complexity, gaze_ent)
        
        # 告警判定
        alert = (risk > threshold).float()
        
        return {
            'risk': risk,
            'threshold': threshold,
            'alert': alert,
            'gaze_entropy': gaze_ent,
            'complexity': complexity
        }


# ============ 完整测试代码 ============
if __name__ == "__main__":
    import numpy as np
    import time
    from sklearn.metrics import precision_score, recall_score, f1_score
    
    BATCH_SIZE = 16
    SEQ_LEN = 10  # 3-5秒 @ 2-3fps
    EMBED_DIM = 256
    
    model = CAMERAFramework(embed_dim=EMBED_DIM)
    total_params = sum(p.numel() for p in model.parameters())
    print(f"CAMERA参数量: {total_params:,}")
    
    # 模拟数据
    np.random.seed(42)
    
    def generate_scenario(risky: bool, batch=1):
        """生成驾驶场景"""
        if risky:
            # 高风险：复杂场景 + 分散注视
            video = torch.rand(batch, SEQ_LEN, 3, 224, 224) * 0.7 + 0.3
            gaze = torch.zeros(batch, SEQ_LEN, 1, 56, 56)
            gaze[:, :, :, 10:40, 5:50] = 0.5  # 分散注视
            gaze += torch.rand_like(gaze) * 0.3
            text = torch.randint(5, 10, (batch,))  # 复杂场景
        else:
            # 低风险：简单场景 + 集中注视
            video = torch.rand(batch, SEQ_LEN, 3, 224, 224) * 0.5
            gaze = torch.zeros(batch, SEQ_LEN, 1, 56, 56)
            gaze[:, :, :, 20:35, 20:35] = 0.8  # 集中注视
            text = torch.randint(0, 3, (batch,))  # 简单场景
        
        return video, gaze, text
    
    # 生成数据
    print("\n--- 生成数据 ---")
    all_video, all_gaze, all_text, all_labels = [], [], [], []
    for _ in range(10):
        v, g, t = generate_scenario(True, 8)
        all_video.append(v); all_gaze.append(g); all_text.append(t)
        all_labels.extend([1] * 8)
    for _ in range(10):
        v, g, t = generate_scenario(False, 8)
        all_video.append(v); all_gaze.append(g); all_text.append(t)
        all_labels.extend([0] * 8)
    
    video_batch = torch.cat(all_video)
    gaze_batch = torch.cat(all_gaze)
    text_batch = torch.cat(all_text)
    labels = torch.tensor(all_labels)
    
    # 当前场景图
    scene_batch = video_batch[:, -1].clone()
    
    print(f"  Video: {video_batch.shape}")
    print(f"  Gaze: {gaze_batch.shape}")
    print(f"  Labels: {labels.shape} (pos={labels.sum()}, neg={(1-labels).sum()})")
    
    # 训练
    print("\n--- 训练流程 ---")
    criterion = nn.BCELoss()
    optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4)
    
    model.train()
    for epoch in range(3):
        optimizer.zero_grad()
        out = model(video_batch, gaze_batch, text_batch, scene_batch)
        loss = criterion(out['risk'], labels.float())
        loss.backward()
        optimizer.step()
        
        # 评估
        with torch.no_grad():
            preds = (out['risk'] > out['threshold']).float()
            prec = precision_score(labels.numpy(), preds.numpy(), zero_division=0)
            rec = recall_score(labels.numpy(), preds.numpy(), zero_division=0)
            f1 = f1_score(labels.numpy(), preds.numpy(), zero_division=0)
        
        print(f"Epoch {epoch+1}: Loss={loss.item():.4f}, "
              f"Prec={prec:.2%}, Rec={rec:.2%}, F1={f1:.2%}")
    
    # 阈值分析
    print("\n--- 自适应阈值分析 ---")
    model.eval()
    with torch.no_grad():
        out = model(video_batch[:4], gaze_batch[:4], text_batch[:4], scene_batch[:4])
        for i in range(4):
            print(f"  Sample {i}: risk={out['risk'][i]:.3f}, "
                  f"threshold={out['threshold'][i]:.3f}, "
                  f"entropy={out['gaze_entropy'][i]:.3f}, "
                  f"complexity={out['complexity'][i]:.3f}, "
                  f"alert={'YES' if out['alert'][i] else 'NO'}")
    
    # 推理速度
    with torch.no_grad():
        start = time.time()
        for _ in range(20):
            _ = model(video_batch[:1], gaze_batch[:1], text_batch[:1], scene_batch[:1])
        elapsed = (time.time() - start) / 20
        print(f"\n单样本推理延迟: {elapsed*1000:.1f}ms")

实验结果

DADA-2000性能

指标	前SOTA	CAMERA	提升
Average Precision (AP)	75.3%	80.52%	+5.2pp
AUC	84.5%	89.64%	+5.1pp
mTTA (秒)	3.8s	4.48s	+0.68s
数据效率 (50%数据)	~85%	92%+	+7pp

自适应阈值效果

场景	固定阈值误报	自适应误报	改善
高速公路	12.3%	3.1%	-74.8%
城市路口	8.5%	5.2%	-38.8%
复杂天气	22.1%	9.8%	-55.7%
驾驶员分心	18.7%	6.3%	-66.3%

IMS应用启示

1. Euro NCAP DSM + ADAS协同

Euro NCAP 2026 → 2027:
  DSM检测驾驶员状态 → ADAS调整干预策略
  
CAMERA框架:
  注视熵 → 认知状态 → 自适应告警阈值
  场景复杂度 → 环境感知 → 风险预判
  多模态融合 → 综合决策

2. 驾驶员注意力量化

def classify_attention_state(gaze_entropy: float, 
                              complexity: float) -> str:
    """
    基于注视熵和场景复杂度分类驾驶员状态
    """
    if gaze_entropy > 0.7 and complexity > 0.6:
        return "COGNITIVE_OVERLOAD"  # 认知过载
    elif gaze_entropy > 0.6 and complexity < 0.3:
        return "MIND_WANDERING"  # 走神
    elif gaze_entropy < 0.2:
        return "TUNNEL_VISION"  # 隧道视觉
    elif gaze_entropy < 0.4 and complexity > 0.5:
        return "FOCUSED"  # 专注
    else:
        return "NORMAL"  # 正常

3. 对Euro NCAP测试场景的映射

场景	注视熵	场景复杂度	CAMERA响应
D-02 手机使用	高	中	快速检测（阈值降低）
D-05 视线偏离	高	低	走神检测
F-01 PERCLOS	N/A	N/A	由疲劳模块处理
城市路口复杂	中	高	提高敏感度

4. 部署建议

模块	部署位置	计算量	延迟
注视图提取	DMS域控	5ms	实时
场景复杂度	ADAS域控	20ms	实时
Bi-GRU推理	域控融合	30ms	100ms
自适应阈值	DMS域控	<1ms	实时
总延迟	-	-	~60ms

---

论文来源： ACM Multimedia 2025

数据集： DADA-2000 (2000视频, 54种碰撞类型, 同步注视日志)