DriveEmo-FL：mmWave雷达座舱情感感知——隐私保护联邦学习框架论文解读

论文信息

项目 内容
标题 DriveEmo-FL: in-cabin radar-based emotion sensing for autonomous vehicles smart response
期刊 Scientific Reports (Nature)
发表 2026年3月15日
链接 https://www.nature.com/articles/s41598-026-43662-x
雷达型号 TI IWR-1443BOOST（毫米波mmWave）
方法 联邦学习+微多普勒+速度时间剖面
开源 部分开源

核心创新

  1. 首个座舱mmWave雷达情感识别框架：利用紧凑型mmWave雷达捕获上半身情绪性手势，无需摄像头即可识别乘客情绪
  2. 联邦学习架构：跨车辆去中心化训练，保护乘客隐私，解决车队级情感模型部署
  3. EmoNet双流架构：融合微多普勒签名和速度时间剖面（VTP），两种雷达特征互补提升分类精度

问题定义

为什么需要座舱情感感知？

场景 情绪状态 AV响应策略
通勤高峰焦虑 焦虑/紧张 平稳驾驶+舒缓音乐
长途疲劳 疲倦/低落 提醒休息+切换为运动模式
雨天恐惧 恐惧/不安 减速+语音安慰
放松享受 愉悦 风景路线推荐
路怒前兆 愤怒 降速+冷静提示

现有方案的局限

方案 隐私 遮挡 低光 精度 成本
摄像头面部表情 ❌ 隐私担忧 ❌ 遮挡影响 ❌ 需补光 ✅ 高 🟡
可穿戴生理传感器 ✅ ✅ ✅ ✅ 高 ❌ 需佩戴
WiFi CSI ✅ ✅ ✅ ❌ 低分辨率 ✅
mmWave雷达 ✅ ✅ ✅ ✅ 高 ✅

方法详解

1. 系统架构

flowchart TD
    A[IWR-1443BOOST mmWave雷达] --> B[CFAR预处理]
    B --> C[微多普勒签名提取]
    B --> D[速度时间剖面VTP提取]
    C --> E[EmoNet Stream 1: 微多普勒CNN]
    D --> F[EmoNet Stream 2: VTP-BiLSTM]
    E --> G[双流融合层]
    F --> G
    G --> H[情绪分类输出]
    H --> I[AV自适应响应]
    
    J[车辆1 本地训练] --> K[联邦学习服务器]
    L[车辆2 本地训练] --> K
    M[车辆N 本地训练] --> K
    K --> J
    K --> L
    K --> M

2. 雷达硬件配置

参数 数值
型号 TI IWR-1443BOOST
频率 76-81 GHz
扫频带宽 4 GHz
天线 4发4收（4Tx-4Rx）
帧率 30 fps
角度分辨率 ~15°
距离分辨率 ~3.75 cm
功耗 ~3.5W
价格 ~$149（开发板）

3. CFAR预处理与特征提取

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
import numpy as np
from scipy import signal as sig

class RadarPreprocessor:
    """
    mmWave雷达信号预处理管道
    
    论文Section System Overview:
    1. CFAR检测提取目标范围
    2. 微多普勒签名生成
    3. 速度时间剖面（VTP）生成
    """
    
    def __init__(self, n_range_bins: int = 64, 
                 n_doppler_bins: int = 64,
                 fps: int = 30):
        self.n_range_bins = n_range_bins
        self.n_doppler_bins = n_doppler_bins
        self.fps = fps
        
    def cfar_detect(self, range_doppler_map: np.ndarray,
                    guard_cells: int = 2,
                    train_cells: int = 8,
                    pfa: float = 1e-4) -> np.ndarray:
        """
        CA-CFAR目标检测
        
        Args:
            range_doppler_map: [n_range, n_doppler] 功率谱
            guard_cells: 保护单元数
            train_cells: 训练单元数
            pfa: 虚警概率
            
        Returns:
            detection_mask: [n_range, n_doppler] 检测掩码
        """
        n_range, n_doppler = range_doppler_map.shape
        mask = np.zeros_like(range_doppler_map, dtype=bool)
        
        # 阈值因子
        n_train = 4 * train_cells * guard_cells
        alpha = n_train * (pfa ** (-1/n_train) - 1)
        
        for i in range(guard_cells + train_cells, 
                       n_range - guard_cells - train_cells):
            for j in range(guard_cells + train_cells,
                           n_doppler - guard_cells - train_cells):
                # 提取训练窗口
                train_region = range_doppler_map[
                    i-guard_cells-train_cells:i+guard_cells+train_cells+1,
                    j-guard_cells-train_cells:j+guard_cells+train_cells+1
                ]
                # 排除保护区域
                cut = range_doppler_map[
                    i-guard_cells:i+guard_cells+1,
                    j-guard_cells:j+guard_cells+1
                ]
                train_region = np.delete(
                    train_region,
                    slice(train_cells, train_cells+2*guard_cells+1),
                    axis=0
                )
                
                # 均值估计
                noise_level = np.mean(train_region)
                threshold = noise_level * (1 + alpha/n_train)
                
                if range_doppler_map[i, j] > threshold:
                    mask[i, j] = True
                    
        return mask
    
    def extract_micro_doppler(self, radar_cube: np.ndarray,
                               target_range: int,
                               duration_sec: float = 3.0) -> np.ndarray:
        """
        提取微多普勒签名
        
        Args:
            radar_cube: [n_frames, n_range, n_doppler]
            target_range: 目标距离bin
            duration_sec: 采集时长
            
        Returns:
            spectrogram: [n_time, n_doppler] 时频图
        """
        n_frames = int(duration_sec * self.fps)
        
        # 提取目标距离单元的多普勒序列
        doppler_seq = radar_cube[:n_frames, target_range, :]
        
        # STFT生成微多普勒图
        f, t, Sxx = sig.stft(
            doppler_seq,
            fs=self.fps,
            nperseg=64,
            noverlap=56,
            axis=0
        )
        
        # 取幅度
        spectrogram = np.abs(Sxx)
        
        # 归一化
        spectrogram = (spectrogram - spectrogram.min()) / \
                      (spectrogram.max() - spectrogram.min() + 1e-8)
        
        return spectrogram.T  # [n_time, n_doppler]
    
    def extract_vtp(self, radar_cube: np.ndarray,
                    target_range: int,
                    duration_sec: float = 3.0) -> np.ndarray:
        """
        提取速度时间剖面（VTP）
        
        Args:
            radar_cube: [n_frames, n_range, n_doppler]
            target_range: 目标距离bin
            
        Returns:
            vtp: [n_frames] 速度时间序列
        """
        n_frames = int(duration_sec * self.fps)
        
        # 对每帧的多普勒谱取加权平均速度
        doppler_bins = np.arange(self.n_doppler_bins)
        # 速度映射（假设已校准）
        velocities = np.linspace(-3, 3, self.n_doppler_bins)  # m/s
        
        vtp = np.zeros(n_frames)
        for f_idx in range(n_frames):
            doppler = radar_cube[f_idx, target_range, :]
            # 多普勒功率加权平均速度
            vtp[f_idx] = np.sum(doppler * velocities) / \
                         (np.sum(doppler) + 1e-8)
        
        return vtp


# 测试
if __name__ == "__main__":
    proc = RadarPreprocessor()
    
    # 模拟雷达数据
    radar_cube = np.random.randn(90, 64, 64) * 0.1
    # 添加模拟人体运动信号
    for f in range(90):
        radar_cube[f, 32, 30+int(5*np.sin(f*0.3))] += 2.0
    
    spectrogram = proc.extract_micro_doppler(radar_cube, target_range=32)
    vtp = proc.extract_vtp(radar_cube, target_range=32)
    
    print(f"微多普勒图: {spectrogram.shape}")  # ~[90, 33]
    print(f"速度时间剖面: {vtp.shape}")       # [90]
    print(f"平均速度: {np.mean(vtp):.3f} m/s")
    print(f"速度标准差: {np.std(vtp):.3f} m/s")

4. EmoNet双流分类网络

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
import torch
import torch.nn as nn

class EmoNet(nn.Module):
    """
    EmoNet: 双流情感分类网络
    
    Stream 1: 微多普勒图 → CNN提取空间特征
    Stream 2: VTP序列 → BiLSTM提取时序特征
    融合层: 拼接+MLP → 情绪分类
    """
    
    def __init__(self, n_classes: int = 6, 
                 n_doppler_bins: int = 33):
        super().__init__()
        
        # Stream 1: 微多普勒CNN
        self.doppler_cnn = nn.Sequential(
            nn.Conv2d(1, 32, 3, padding=1),
            nn.BatchNorm2d(32),
            nn.ReLU(),
            nn.MaxPool2d(2),
            
            nn.Conv2d(32, 64, 3, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(),
            nn.MaxPool2d(2),
            
            nn.Conv2d(64, 128, 3, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(),
            nn.AdaptiveAvgPool2d(1),
        )
        
        # Stream 2: VTP BiLSTM
        self.vtp_lstm = nn.LSTM(
            input_size=1, hidden_size=64,
            num_layers=2, batch_first=True,
            bidirectional=True, dropout=0.3
        )
        
        # 融合分类头
        self.classifier = nn.Sequential(
            nn.Linear(128 + 128, 64),
            nn.ReLU(),
            nn.Dropout(0.5),
            nn.Linear(64, n_classes)
        )
        
    def forward(self, doppler_spec: torch.Tensor, 
                vtp_seq: torch.Tensor) -> torch.Tensor:
        """
        Args:
            doppler_spec: [B, 1, T, F] 微多普勒图
            vtp_seq: [B, T, 1] 速度时间剖面
            
        Returns:
            logits: [B, n_classes]
        """
        # Stream 1
        dop_feat = self.doppler_cnn(doppler_spec)  # [B, 128, 1, 1]
        dop_feat = dop_feat.flatten(1)  # [B, 128]
        
        # Stream 2
        lstm_out, _ = self.vtp_lstm(vtp_seq)  # [B, T, 128]
        vtp_feat = lstm_out[:, -1, :]  # 取最后时步 [B, 128]
        
        # 融合
        fused = torch.cat([dop_feat, vtp_feat], dim=1)  # [B, 256]
        logits = self.classifier(fused)
        
        return logits


# 测试
if __name__ == "__main__":
    model = EmoNet(n_classes=6)
    
    # 模拟输入：3秒@30fps = 90帧
    doppler_input = torch.randn(4, 1, 90, 33)  # [B, C, T, F]
    vtp_input = torch.randn(4, 90, 1)          # [B, T, 1]
    
    output = model(doppler_input, vtp_input)
    print(f"微多普勒输入: {doppler_input.shape}")
    print(f"VTP输入: {vtp_input.shape}")
    print(f"输出logits: {output.shape}")
    print(f"模型参数量: {sum(p.numel() for p in model.parameters()):,}")
    
# 预期输出:
# 输出logits: torch.Size([4, 6])
# 模型参数量: ~450K

5. 联邦学习框架

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
from typing import List, Dict
import copy

class FederatedEmotionLearning:
    """
    DriveEmo-FL联邦学习框架
    
    论文Section Federated Learning Framework:
    1. 每辆车本地训练EmoNet
    2. 上传梯度到中心服务器
    3. FedAvg聚合全局模型
    4. 下发更新后的模型
    """
    
    def __init__(self, n_clients: int, 
                 global_model: EmoNet,
                 round_interval: int = 100):
        self.n_clients = n_clients
        self.global_model = global_model
        self.round_interval = round_interval
        self.round_count = 0
        
    def client_train(self, client_id: int, 
                     local_data, local_epochs: int = 5) -> dict:
        """
        车端本地训练
        """
        local_model = copy.deepcopy(self.global_model)
        local_model.train()
        
        optimizer = torch.optim.Adam(
            local_model.parameters(), lr=1e-3, weight_decay=1e-4
        )
        criterion = nn.CrossEntropyLoss()
        
        for epoch in range(local_epochs):
            for doppler, vtp, labels in local_data:
                optimizer.zero_grad()
                output = local_model(doppler, vtp)
                loss = criterion(output, labels)
                loss.backward()
                optimizer.step()
        
        # 返回模型参数差值
        global_state = self.global_model.state_dict()
        local_state = local_model.state_dict()
        update = {
            k: local_state[k] - global_state[k]
            for k in global_state.keys()
        }
        return update
    
    def fed_avg(self, client_updates: List[dict], 
                client_weights: List[float]) -> dict:
        """
        FedAvg聚合
        """
        global_state = self.global_model.state_dict()
        aggregated = {k: torch.zeros_like(v) 
                      for k, v in global_state.items()}
        
        for update, weight in zip(client_updates, client_weights):
            for k in aggregated.keys():
                aggregated[k] += update[k] * weight
        
        # 更新全局模型
        for k in global_state.keys():
            global_state[k] += aggregated[k]
        
        self.global_model.load_state_dict(global_state)
        self.round_count += 1
        return global_state

实验结果

数据集

属性 数值
参与者数 30人
情绪类别 6类（高兴、悲伤、愤怒、恐惧、惊讶、中性）
每人每情绪样本 20个×3秒
总样本数 3,600
联邦客户端 5个（模拟5辆车）

性能对比

方法 准确率 F1-Score 延迟 隐私
摄像头FACS 82.3% 0.81 50ms ❌
可穿戴ECG+EDA 78.5% 0.77 100ms ✅
WiFi CSI 45.2% 0.41 200ms ✅
mmWave仅微多普勒 61.8% 0.59 30ms ✅
mmWave仅VTP 54.3% 0.52 30ms ✅
EmoNet（微多普勒+VTP） 68.7% 0.66 35ms ✅
EmoNet + 联邦学习 66.2% 0.64 35ms ✅✅

联邦学习收敛分析

轮次 中心化准确率 联邦准确率 差距
10 58.3% 52.1% 6.2%
50 65.1% 62.8% 2.3%
100 68.7% 66.2% 2.5%
200 69.1% 67.8% 1.3%

联邦学习在100轮后仅损失2.5%准确率，换取完全的数据隐私保护。

情绪类别分析

情绪 准确率 典型上半身手势 微多普勒特征
高兴 78.2% 挥手、拍手 高频多分量
愤怒 72.1% 快速前倾、握拳 突变高频
悲伤 55.3% 缓慢低头、垂肩 低频缓变
恐惧 63.8% 后仰、抱胸 急促短位移
惊讶 71.5% 突然抬头、举手 脉冲式
中性 71.0% 静坐微动 微弱呼吸

IMS开发启示

1. 与CPD/OOP功能共享雷达

功能 雷达类型 频段 共享可行性
CPD儿童检测 60GHz 60-64GHz ⚠️ 需双雷达
生命体征监测 60GHz/77GHz mmWave ✅ 共用77GHz
情感感知 77GHz 76-81GHz ✅ 共用
手势控制 60GHz/77GHz mmWave ✅ 共用
OOP姿态 77GHz 76-81GHz ✅ 共用

建议：77GHz mmWave雷达作为座舱统一感知平台，同时支持CPD+生命体征+情感+手势+OOP。

2. AV响应策略映射

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
class AVEmotionResponse:
    """
    AV情感自适应响应策略
    
    根据DriveEmo-FL输出调整AV行为
    """
    
    response_map = {
        'happy': {
            'driving_style': 'normal',
            'music': 'upbeat',
            'route': 'scenic',
            'hvac': 'comfortable'
        },
        'sad': {
            'driving_style': 'gentle',
            'music': 'soft_instrumental',
            'route': 'fastest',
            'hvac': 'warm',
            'voice_tone': 'comforting'
        },
        'angry': {
            'driving_style': 'conservative',  # 降低激进驾驶
            'music': 'calm_classical',
            'route': 'low_traffic',
            'hvac': 'cool',
            'voice_tone': 'neutral_calm',
            'alert': 'consider_rest_stop'
        },
        'fear': {
            'driving_style': 'extra_cautious',
            'music': 'none',
            'route': 'well_lit',
            'hvac': 'fresh_air',
            'voice_tone': 'reassuring'
        },
        'fatigued': {
            'driving_style': 'conservative',
            'music': 'energetic',
            'route': 'nearest_rest',
            'alert': 'strong_drowsiness_warning'
        }
    }
    
    def get_response(self, emotion: str, 
                    confidence: float) -> dict:
        if confidence < 0.5:
            return self.response_map.get('neutral', {})
        return self.response_map.get(emotion, {})

3. 部署硬件方案

方案 雷达 位置 覆盖范围 成本
经济型 1×IWR1443 顶灯 前排 $150
标准型 2×IWR1443 顶灯+中控 全座舱 $300
高端型 4×IWR6843 四角 全座舱+行李箱 $600

4. 与Euro NCAP的关联

ENCAP功能 雷达情感贡献 优先级
CPD儿童检测 共享雷达硬件 P0
生命体征监测 共享信号处理管道 P1
OOP姿态检测 微多普勒辅助姿态 P2
情感感知 增值功能 P3
驾驶员干预 情绪→干预策略 P3

情感感知是ENCAP不强制但大幅提升用户体验和差异化的增值功能。

5. 隐私优势

对比维度 摄像头 mmWave雷达
面部图像 ❌ 采集 ✅ 不采集
身份识别 ❌ 可被滥用 ✅ 不可识别
数据存储 图像/视频 时序信号<1KB/s
GDPR合规 需明确同意 信号级数据不构成个人数据
车队部署 每车需隐私协议 ✅ 联邦学习天然合规

局限性与未来方向

局限 描述 解决方向
情绪→手势映射假设 假设情绪通过上半身手势表达 叠加呼吸/心率
文化差异 不同文化手势表达不同 多文化数据集
静止情绪 无外显手势时无法检测 叠加微呼吸检测
6类情绪限制 未覆盖复杂情绪 细粒度分类
联邦通信成本 每100轮需上传梯度 梯度压缩+稀疏化

相关工作对比

方法 传感器 情绪类别 准确率 隐私 实时
FACS摄像头 RGB 6-7 82% ❌ ✅
ECG+EDA可穿戴 生理 4-6 78% ✅ ✅
EQ-Radio (RF) 60GHz RF 4 72% ✅ ❌
FERT (FMCW) mmWave 6 99%* ✅ ✅
DriveEmo-FL 77GHz mmWave 6 68.7% ✅✅ ✅

*FERT在受控室内环境达99%，车辆环境中未验证

总结

DriveEmo-FL 开创了座舱雷达情感感知的完整技术路径：

  1. mmWave雷达替代摄像头做情感感知：隐私保护+全天候工作，准确率68.7%
  2. 联邦学习实现车队级部署：100轮后仅损失2.5%准确率，完全数据本地化
  3. 双流EmoNet架构：微多普勒+VTP互补，参数量450K，端侧可运行
  4. 与CPD/生命体征共享77GHz雷达：一套雷达多功能，降低BOM成本

对IMS的启示：情感感知虽非ENCAP强制项，但可作为差异化卖点。77GHz雷达平台化部署——一套雷达同时支持CPD+生命体征+情感+OOP+手势——是降低BOM成本、提升用户体验的最优路径。联邦学习框架确保车队级部署的隐私合规。

https://dapalm.com/2026/09/21/2026-09-21-16-driveemo-fl-mmwave-radar-emotion-sensing-cabin-ims/
作者
Mars
发布于
2026年9月21日
许可协议