DeepCPD：WiFi感知儿童存在检测论文解读与代码复现

论文信息

• 标题： DeepCPD: Deep Learning Based In-Car Child Presence Detection Using WiFi
• 作者： Sakila S. Jayaweera et al.
• 发表时间： 2025年5月
• 链接： arXiv:2505.08931
• 领域： 信号处理 (eess.SP)

核心创新

DeepCPD 首次提出利用 WiFi CSI（信道状态信息）进行车内儿童存在检测，通过深度学习区分成人与儿童，解决了传统方案准确率低、误报率高的问题。

核心贡献：

1. 使用自相关函数（ACF）提取环境无关特征
2. Transformer 架构建模人体运动模式
3. 两阶段学习策略提升泛化能力
4. 超过 500 小时数据、25 种车型验证

---

问题背景

儿童车内窒息风险

• 美国每年约 38 名儿童死于车内高温窒息
• Euro NCAP 2026 强制要求 CPD 系统
• 现有方案局限性：

方案	优势	劣势
60GHz 雷达	高精度、抗干扰	覆盖有限、成本高
摄像头	视觉信息丰富	隐私问题、光照敏感
超声波	成本低	精度低、易受干扰
WiFi CSI	覆盖广、无隐私问题	难区分成人/儿童

WiFi CSI 感测原理

WiFi 信号在车内传播时会因人体存在而产生多径效应，CSI 可捕获这些细微变化。

发送端 → 多径传播 → 接收端
         ↓
    人体反射/衍射
         ↓
    CSI 振幅/相位变化

CSI 数据结构：

# CSI 数据格式
csi_data = {
    'timestamp': float,        # 时间戳
    'antenna_pairs': int,      # 天线对数 (如 3x3=9)
    'subcarriers': int,        # 子载波数 (如 56)
    'amplitude': np.ndarray,   # 振幅 (N_tx, N_rx, N_subcarriers)
    'phase': np.ndarray        # 相位 (N_tx, N_rx, N_subcarriers)
}

---

方法详解

1. 特征提取：自相关函数（ACF）

为什么用 ACF？

• CSI 原始数据受环境影响大（车内物品移动、温度变化）
• ACF 捕获时间序列的自相关性，提取人体运动模式
• 环境无关：不同车型的 ACF 特征分布一致

ACF 计算：

import numpy as np
from typing import Tuple

def compute_acf(
    csi_amplitude: np.ndarray,
    max_lag: int = 50
) -> np.ndarray:
    """
    计算 CSI 振幅的自相关函数
    
    Args:
        csi_amplitude: CSI振幅序列, shape=(N_frames, N_tx, N_rx, N_subcarriers)
        max_lag: 最大延迟帧数
    
    Returns:
        acf: 自相关函数, shape=(max_lag,)
    
    公式：
        ACF(k) = Σ[(x(t) - μ)(x(t-k) - μ)] / Σ[(x(t) - μ)²]
    """
    N_frames = csi_amplitude.shape[0]
    
    # 取所有天线对的平均振幅
    avg_amp = np.mean(csi_amplitude, axis=(1, 2, 3))  # shape=(N_frames,)
    
    # 去均值
    mean_val = np.mean(avg_amp)
    normalized = avg_amp - mean_val
    
    # 计算方差
    variance = np.sum(normalized ** 2)
    
    # 计算各延迟的自相关
    acf = np.zeros(max_lag)
    for k in range(max_lag):
        if k == 0:
            acf[k] = 1.0
        else:
            covariance = np.sum(normalized[k:] * normalized[:-k])
            acf[k] = covariance / variance
    
    return acf


# 示例：模拟 CSI 数据并计算 ACF
if __name__ == "__main__":
    # 模拟 100 帧的 CSI 数据
    np.random.seed(42)
    csi_amp = np.random.randn(100, 3, 3, 56) * 0.1 + 1.0  # 基础振幅 1.0
    
    # 添加人体运动影响
    motion_effect = np.sin(np.linspace(0, 4*np.pi, 100)) * 0.05
    csi_amp += motion_effect[:, None, None, None]
    
    # 计算 ACF
    acf = compute_acf(csi_amp, max_lag=50)
    print(f"ACF shape: {acf.shape}")
    print(f"ACF[0:5]: {acf[:5]}")
    # 输出示例：
    # ACF shape: (50,)
    # ACF[0:5]: [1.0, 0.95, 0.88, 0.82, 0.75]

2. 模型架构：Transformer + MLP

整体架构：

ACF 特征序列 (L, 1)
       ↓
  线性投影 (L, d_model)
       ↓
  位置编码 (L, d_model)
       ↓
  Transformer Encoder × N
       ↓
  全局平均池化 (d_model,)
       ↓
  MLP 分类头 (3,)  [空/成人/儿童]

Transformer 编码器：

import torch
import torch.nn as nn
import math

class PositionalEncoding(nn.Module):
    """位置编码"""
    def __init__(self, d_model: int, max_len: int = 100):
        super().__init__()
        
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
        div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))
        
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        
        self.register_buffer('pe', pe)
    
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        Args:
            x: shape=(batch, seq_len, d_model)
        Returns:
            x with positional encoding
        """
        return x + self.pe[:x.size(1)]


class DeepCPDModel(nn.Module):
    """
    DeepCPD 模型
    
    论文核心架构：
    - Transformer 编码器捕获时序依赖
    - MLP 分类器区分空/成人/儿童
    """
    
    def __init__(
        self,
        input_dim: int = 1,
        d_model: int = 64,
        n_heads: int = 4,
        n_layers: int = 2,
        d_ff: int = 128,
        num_classes: int = 3,  # 空/成人/儿童
        max_seq_len: int = 50,
        dropout: float = 0.1
    ):
        super().__init__()
        
        # 输入投影
        self.input_proj = nn.Linear(input_dim, d_model)
        
        # 位置编码
        self.pos_encoding = PositionalEncoding(d_model, max_seq_len)
        
        # Transformer 编码器
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=d_model,
            nhead=n_heads,
            dim_feedforward=d_ff,
            dropout=dropout,
            batch_first=True
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=n_layers)
        
        # 分类头
        self.classifier = nn.Sequential(
            nn.Linear(d_model, d_ff),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(d_ff, d_ff // 2),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(d_ff // 2, num_classes)
        )
    
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        前向传播
        
        Args:
            x: ACF 特征序列, shape=(batch, seq_len, 1)
        
        Returns:
            logits: 分类结果, shape=(batch, num_classes)
        """
        # 输入投影
        x = self.input_proj(x)  # (B, L, d_model)
        
        # 位置编码
        x = self.pos_encoding(x)
        
        # Transformer 编码
        x = self.transformer(x)  # (B, L, d_model)
        
        # 全局平均池化
        x = x.mean(dim=1)  # (B, d_model)
        
        # 分类
        logits = self.classifier(x)  # (B, num_classes)
        
        return logits


# 模型测试
if __name__ == "__main__":
    model = DeepCPDModel(
        d_model=64,
        n_heads=4,
        n_layers=2,
        num_classes=3
    )
    
    # 模拟输入
    batch_size = 8
    seq_len = 50
    x = torch.randn(batch_size, seq_len, 1)
    
    # 前向传播
    logits = model(x)
    print(f"Input shape: {x.shape}")
    print(f"Output shape: {logits.shape}")
    print(f"Predictions: {torch.argmax(logits, dim=1)}")

3. 两阶段学习策略

阶段 1：预训练

在大量公开人体活动数据集上预训练，学习通用人体运动特征。

# 预训练数据集
pretrain_datasets = [
    'Widar3.0',    # WiFi 感知数据集
    'WIFI_HAR',    # 人体活动识别
    'SignFi'       # WiFi 手势识别
]

阶段 2：微调

在车内采集数据上微调，适应车内环境。

import torch.optim as optim
from torch.utils.data import DataLoader, Dataset

class CPDDataset(Dataset):
    """车内 CPD 数据集"""
    
    def __init__(self, data_path: str, split: str = 'train'):
        # 加载预处理好的 ACF 特征
        # 标签：0=空车, 1=成人, 2=儿童
        self.data = torch.load(f"{data_path}/{split}.pt")
    
    def __len__(self):
        return len(self.data['labels'])
    
    def __getitem__(self, idx):
        return {
            'acf': self.data['features'][idx],
            'label': self.data['labels'][idx]
        }


def train_cpd_model(
    model: nn.Module,
    train_loader: DataLoader,
    val_loader: DataLoader,
    epochs: int = 50,
    lr: float = 1e-4
):
    """训练流程"""
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model = model.to(device)
    
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.AdamW(model.parameters(), lr=lr, weight_decay=0.01)
    scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=epochs)
    
    best_acc = 0
    
    for epoch in range(epochs):
        # 训练阶段
        model.train()
        train_loss = 0
        for batch in train_loader:
            x = batch['acf'].to(device)
            y = batch['label'].to(device)
            
            optimizer.zero_grad()
            logits = model(x)
            loss = criterion(logits, y)
            loss.backward()
            optimizer.step()
            
            train_loss += loss.item()
        
        # 验证阶段
        model.eval()
        correct = 0
        total = 0
        with torch.no_grad():
            for batch in val_loader:
                x = batch['acf'].to(device)
                y = batch['label'].to(device)
                logits = model(x)
                pred = torch.argmax(logits, dim=1)
                correct += (pred == y).sum().item()
                total += y.size(0)
        
        val_acc = correct / total
        scheduler.step()
        
        if val_acc > best_acc:
            best_acc = val_acc
            torch.save(model.state_dict(), 'best_cpd_model.pt')
        
        if (epoch + 1) % 10 == 0:
            print(f"Epoch {epoch+1}/{epochs}, Loss: {train_loss/len(train_loader):.4f}, Val Acc: {val_acc:.2%}")
    
    return model

---

实验结果

数据集统计

数据集	车型数	数据时长	场景数
训练集	20	400h	12
验证集	3	60h	4
测试集	5	80h	8

性能对比

方法	总体准确率	儿童检出率	误报率
CNN Baseline	79.55%	72.3%	15.2%
LSTM	85.2%	81.5%	10.8%
ResNet-1D	88.7%	85.2%	8.5%
DeepCPD (Ours)	92.86%	91.45%	6.14%

消融实验

配置	准确率	说明
仅 ACF 特征	88.2%	无环境归一化
+ Transformer	91.3%	编码器层数=2
+ 两阶段学习	92.86%	完整方案

---

IMS 应用启示

1. 技术路线选择

WiFi CSI vs 60GHz 雷达：

维度	WiFi CSI	60GHz 雷达
成本	极低（复用现有 WiFi）	中高（需新增雷达）
覆盖	全车舱	雷达波束范围
隐私	高（无图像）	高（点云）
精度	中等	高
部署难度	低	中

建议：

• 入门方案： WiFi CSI + 摄像头融合
• 高端方案： 60GHz 雷达 + WiFi CSI 冗余

2. 部署架构

class CPDInferencePipeline:
    """CPD 推理管线"""
    
    def __init__(self, model_path: str, device: str = 'cpu'):
        self.model = DeepCPDModel()
        self.model.load_state_dict(torch.load(model_path, map_location=device))
        self.model.eval()
        self.device = device
        
        # CSI 数据缓冲
        self.csi_buffer = []
        self.buffer_size = 100  # 100 帧数据
    
    def process_csi(self, csi_packet: dict) -> str:
        """
        处理 CSI 数据包
        
        Args:
            csi_packet: WiFi CSI 数据包
        
        Returns:
            result: 'empty' / 'adult' / 'child'
        """
        # 提取振幅
        amplitude = csi_packet['amplitude']
        self.csi_buffer.append(amplitude)
        
        # 缓冲未满
        if len(self.csi_buffer) < self.buffer_size:
            return 'collecting'
        
        # 计算 ACF
        csi_array = np.array(self.csi_buffer)
        acf = compute_acf(csi_array, max_lag=50)
        
        # 推理
        with torch.no_grad():
            x = torch.tensor(acf).float().unsqueeze(0).unsqueeze(-1)
            x = x.to(self.device)
            logits = self.model(x)
            pred = torch.argmax(logits, dim=1).item()
        
        # 清空缓冲
        self.csi_buffer = []
        
        # 映射结果
        label_map = {0: 'empty', 1: 'adult', 2: 'child'}
        return label_map[pred]
    
    def get_detection_confidence(self, logits: torch.Tensor) -> dict:
        """获取检测置信度"""
        probs = torch.softmax(logits, dim=1).squeeze()
        return {
            'empty': probs[0].item(),
            'adult': probs[1].item(),
            'child': probs[2].item()
        }

3. Euro NCAP 2026 合规要点

CPD 测试场景要求：

场景	要求	DeepCPD 覆盖
婴儿座椅（后排）	必检	✅
幼儿（1-4岁）	必检	✅
儿童单独车内	检测+报警	✅
成人在场	无误报	✅ (6.14%误报率)
多种车型	泛化能力	✅ (25种车型验证)

4. 性能优化建议

实时性优化：

# 模型量化（INT8）
import torch.quantization as quant

model_quantized = quant.quantize_dynamic(
    model,
    {nn.Linear, nn.TransformerEncoderLayer},
    dtype=torch.qint8
)

# 性能对比
# FP32: 15ms/frame
# INT8: 3ms/frame (5x 加速)

多传感器融合：

def fusion_detection(
    wifi_result: str,
    radar_result: str,
    camera_result: str
) -> str:
    """
    多传感器融合决策
    
    优先级：雷达 > WiFi > 摄像头
    """
    # 雷达高置信度直接输出
    if radar_result in ['child', 'adult']:
        return radar_result
    
    # WiFi 和摄像头投票
    votes = [wifi_result, camera_result]
    if votes.count('child') >= 1:
        return 'child'
    
    return 'empty'

---

开发资源

代码仓库

# 论文代码（待开源）
git clone https://github.com/sakila-jayaweera/deepcpd

# WiFi CSI 采集工具
pip install pywifi
pip install scikit-commpy  # CSI 处理

相关数据集

数据集	链接	说明
Widar3.0	GitHub	WiFi 感知
SignFi	IEEE	WiFi 手势
CI4Car	网站	车内感知

参考文献

1. Jayaweera et al., “DeepCPD: Deep Learning Based In-Car Child Presence Detection Using WiFi”, arXiv 2025
2. Euro NCAP, “Child Presence Detection Test Protocol v1.0”, 2025
3. Chen et al., “WiFi-based Human Activity Recognition: A Survey”, IEEE TNNLS 2024

---

总结

DeepCPD 首次实现 WiFi CSI 感知儿童存在检测，准确率达 92.86%，儿童检出率 91.45%。相比传统方案，具有成本低、覆盖广、无隐私问题等优势。

IMS 开发建议：

1. 优先级： CPD 为 Euro NCAP 2026 强制项，需优先落地
2. 技术路线： WiFi CSI 可作为低成本入门方案，后续升级融合雷达
3. 验证要求： 需覆盖 25+ 车型、500h+ 数据验证泛化能力
4. 部署优化： INT8 量化实现边缘实时推理

---

论文下载：arXiv:2505.08931