Dapalm | IMS技术前沿

Mask-Aware DGCNN+Transformer：mmWave雷达点云序列的儿童-宠物识别——论文解读与代码复现

论文信息

• 标题： Mask-Aware Spatiotemporal Classification of Millimeter-Wave Radar Point Cloud Sequences Using DGCNN and Transformer for Child–Pet Recognition in Enclosed Spaces
• 期刊： Sensors (MDPI), 2026
• 链接： https://pmc.ncbi.nlm.nih.gov/articles/PMC12987172/
• 硬件： Texas Instruments IWR6843AOP 60GHz FMCW雷达

核心创新

提出时空联合分类框架用于mmWave点云序列分类，解决两大核心问题：

1. 空间维度： 有效点Mask机制 —— 抑制对齐产生的无效点对KNN图和特征聚合的干扰
2. 时间维度： 注意力时序建模 —— 捕捉跨帧动态模式，增强类别可分性

结果：三分类（Child/Cat/Dog）准确率97.8%，为人宠分离和CPD提供了可部署的mmWave方案。

方法详解

1. FMCW雷达信号处理基础

import numpy as np
from scipy import signal as sig
from scipy.fft import fft, fftshift

class FMCWRadarProcessor:
    """
    FMCW毫米波雷达信号处理器
    
    论文Section 2描述的信号处理流程
    硬件: TI IWR6843AOP (60GHz, 4Tx/4Rx, 3GHz BW)
    """
    
    def __init__(self, config: dict):
        self.num_tx = config.get('num_tx', 2)  # 发射天线数
        self.num_rx = config.get('num_rx', 4)  # 接收天线数
        self.num_chirps = config.get('num_chirps', 64)  # 每帧chirp数
        self.num_samples = config.get('num_samples', 64)  # 每chirp采样数
        self.fs = config.get('sample_rate', 4e6)  # 采样率 4 MHz
        self.slope = config.get('slope', 60e12)  # 调频斜率 60 MHz/μs
        self.fc = config.get('center_freq', 60.25e9)  # 中心频率
        
    def range_fft(self, adc_data: np.ndarray) -> np.ndarray:
        """
        1D-FFT: 距离估计
        
        Args:
            adc_data: ADC原始数据 shape=(num_chirps, num_samples)
            
        Returns:
            range_profile: 距离-强度 profile shape=(num_chirps, num_samples)
        """
        # 对每个chirp做FFT
        range_fft = fft(adc_data, axis=1)
        return np.abs(range_fft)
    
    def doppler_fft(self, range_profile: np.ndarray) -> np.ndarray:
        """
        2D-FFT: 速度估计
        在range-FFT基础上对chirp维度做FFT
        """
        range_doppler = fft(range_profile, axis=0)
        return np.abs(range_doppler)
    
    def cfar_detection(self, range_doppler: np.ndarray, 
                        guard_cells: int = 2, training_cells: int = 8,
                        pfa: float = 1e-3) -> list:
        """
        CA-CFAR: 恒虚警率检测
        
        Args:
            range_doppler: 距离-多普勒图
            guard_cells: 保护单元
            training_cells: 训练单元
            pfa: 虚警概率
            
        Returns:
            detections: 检测到的点列表 [(range_idx, doppler_idx, magnitude)]
        """
        threshold_factor = np.sqrt(-np.log(pfa))
        detections = []
        
        for r in range(training_cells, range_doppler.shape[1] - training_cells):
            for d in range(training_cells, range_doppler.shape[0] - training_cells):
                # 训练窗口均值
                train_region = range_doppler[
                    d-training_cells:d+training_cells+1,
                    r-training_cells:r+training_cells+1
                ].copy()
                # 去掉保护单元
                train_region = np.delete(train_region, 
                    np.s_[guard_cells:guard_cells+2*guard_cells+1], axis=0)
                noise_level = np.mean(train_region)
                threshold = noise_level * threshold_factor
                
                if range_doppler[d, r] > threshold:
                    detections.append((r, d, range_doppler[d, r]))
        
        return detections
    
    def form_point_cloud(self, detections: list, num_frames: int = 10) -> np.ndarray:
        """
        从多帧检测形成点云
        
        Returns:
            point_cloud: shape=(num_frames, num_points, 4) [x, y, z, intensity]
        """
        # 简化: 实际需要角度估计(DOA) 
        point_cloud = []
        for frame_idx in range(num_frames):
            frame_points = []
            for r_idx, d_idx, mag in detections:
                # 距离 = beat_freq * c / (2 * slope)
                distance = r_idx * 0.15  # 距离分辨率 ≈ c/(2*BW) ≈ 0.15m
                # 速度 = doppler_idx * λ / (2 * T_frame)
                velocity = d_idx * 0.005  # 简化
                x = distance * np.cos(np.pi/4)  # 简化坐标
                y = distance * np.sin(np.pi/4)
                z = 0.5  # 假设高度
                frame_points.append([x, y, z, mag])
            point_cloud.append(frame_points)
        
        return np.array(point_cloud)


# ==================== 测试 ====================
if __name__ == "__main__":
    radar = FMCWRadarProcessor({})
    
    # 模拟ADC数据
    np.random.seed(42)
    adc_data = np.random.randn(64, 64) + 0.5 * np.cos(
        2 * np.pi * np.arange(64) * 0.1
    ).reshape(-1, 1)
    
    range_profile = radar.range_fft(adc_data)
    range_doppler = radar.doppler_fft(range_profile)
    detections = radar.cfar_detection(range_doppler)
    
    print(f"检测到 {len(detections)} 个点")
    print(f"距离-多普勒图尺寸: {range_doppler.shape}")

2. 有效点Mask机制（核心创新）

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

class EffectivePointMask(nn.Module):
    """
    有效点Mask机制
    
    论文Section 3.1核心创新
    
    问题: mmWave点云稀疏，帧间点数波动大。
          对齐(padding/truncation)产生的无效点会污染KNN图和特征聚合。
    
    解决: 学习每个点的有效性权重(0-1)，
          抑制无效点对邻域图和特征聚合的影响。
    """
    
    def __init__(self, in_channels: int, k: int = 16):
        super().__init__()
        self.k = k
        # 有效性预测网络: 从点特征预测有效性权重
        self.mask_predictor = nn.Sequential(
            nn.Linear(in_channels, 64),
            nn.ReLU(),
            nn.Linear(64, 32),
            nn.ReLU(),
            nn.Linear(32, 1),
            nn.Sigmoid()  # 输出0-1权重
        )
    
    def forward(self, points: torch.Tensor, 
                padding_mask: torch.Tensor = None) -> tuple:
        """
        Args:
            points: (B, N, C) 点特征 [x, y, z, intensity, ...]
            padding_mask: (B, N) True表示padding点
            
        Returns:
            masked_points: (B, N, C) 加权后的点
            mask: (B, N) 有效性权重
        """
        # 预测每个点的有效性
        mask = self.mask_predictor(points)  # (B, N, 1)
        mask = mask.squeeze(-1)  # (B, N)
        
        # 如果有padding mask，强制padding点权重为0
        if padding_mask is not None:
            mask = mask * (~padding_mask).float()
        
        # 应用mask
        masked_points = points * mask.unsqueeze(-1)
        
        return masked_points, mask


class DGCNNWithMask(nn.Module):
    """
    Mask-Aware DGCNN (Dynamic Graph CNN)
    
    论文Section 3.2: 在KNN图构建和特征聚合时使用mask
    """
    
    def __init__(self, in_channels: int, hidden_dim: int = 64, k: int = 16):
        super().__init__()
        self.k = k
        self.mask_module = EffectivePointMask(in_channels, k)
        
        # 边缘特征提取
        self.edge_mlp = nn.Sequential(
            nn.Linear(2 * in_channels + 3, hidden_dim),
            nn.ReLU(),
            nn.BatchNorm1d(hidden_dim),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.BatchNorm1d(hidden_dim)
        )
        
        # 聚合
        self.agg_mlp = nn.Sequential(
            nn.Linear(hidden_dim * 2, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, in_channels)
        )
    
    def knn(self, points: torch.Tensor, mask: torch.Tensor) -> torch.Tensor:
        """
        K近邻图构建（mask感知）
        
        Args:
            points: (B, N, C)
            mask: (B, N) 有效性权重
            
        Returns:
            edge_index: (B, N, K) 邻居索引
        """
        B, N, C = points.shape
        
        # 计算成对距离
        dist = torch.cdist(points, points)  # (B, N, N)
        
        # mask无效点的距离设为inf
        invalid_mask = mask < 0.5  # 有效性<0.5视为无效
        dist = dist.masked_fill(
            invalid_mask.unsqueeze(1).expand(-1, N, -1),
            float('inf')
        )
        
        # KNN
        _, indices = torch.topk(-dist, self.k, dim=-1)  # 最小的K个
        return indices
    
    def forward(self, points: torch.Tensor, 
                padding_mask: torch.Tensor = None) -> torch.Tensor:
        """
        Args:
            points: (B, N, C) [x, y, z, intensity]
            padding_mask: (B, N) True=padding
            
        Returns:
            features: (B, N, hidden_dim)
        """
        B, N, C = points.shape
        
        # Step 1: 计算mask
        masked_points, mask = self.mask_module(points, padding_mask)
        
        # Step 2: KNN图（mask感知）
        knn_idx = self.knn(masked_points, mask)  # (B, N, K)
        
        # Step 3: 提取边缘特征
        neighbors = self._gather_neighbors(masked_points, knn_idx)  # (B, N, K, C)
        
        # 中心-邻居差（边缘特征）
        center_expanded = masked_points.unsqueeze(2).expand(-1, -1, self.k, -1)
        edge_feat = torch.cat([
            center_expanded, 
            neighbors,
            center_expanded - neighbors  # 相对位置
        ], dim=-1)  # (B, N, K, 2C+3)
        
        # 应用mask到边缘
        mask_expanded = mask.unsqueeze(-1).expand(-1, -1, self.k).unsqueeze(-1)
        edge_feat = edge_feat * mask_expanded
        
        # MLP
        edge_feat = edge_feat.reshape(B * N * self.k, -1)
        edge_feat = self.edge_mlp(edge_feat)
        edge_feat = edge_feat.reshape(B, N, self.k, -1)
        
        # Step 4: 聚合（max pooling with mask）
        mask_for_agg = mask.unsqueeze(-1).expand(-1, -1, self.k).unsqueeze(-1)
        edge_feat = edge_feat.masked_fill(mask_for_agg < 0.5, -1e9)
        
        agg_feat, _ = torch.max(edge_feat, dim=2)  # (B, N, hidden_dim)
        
        # 更新
        output = self.agg_mlp(torch.cat([agg_feat, masked_points], dim=-1))
        output = masked_points + output  # 残差连接
        
        return output, mask
    
    def _gather_neighbors(self, points: torch.Tensor, 
                          indices: torch.Tensor) -> torch.Tensor:
        """收集邻居点特征"""
        B, N, C = points.shape
        K = indices.shape[-1]
        
        indices_expanded = indices.unsqueeze(-1).expand(-1, -1, -1, C)
        points_expanded = points.unsqueeze(1).expand(-1, N, -1, -1)
        
        neighbors = torch.gather(points_expanded, 2, indices_expanded)
        return neighbors

3. Transformer时序分类器

class TemporalTransformerClassifier(nn.Module):
    """
    时序Transformer分类器
    
    论文Section 3.3: 跨帧动态模式建模
    """
    
    def __init__(self, feature_dim: int = 64, num_heads: int = 4,
                 num_layers: int = 2, num_classes: int = 3, 
                 max_seq_len: int = 10):
        super().__init__()
        
        # 位置编码
        self.pos_encoding = nn.Parameter(
            torch.randn(1, max_seq_len, feature_dim) * 0.02
        )
        
        # Transformer编码器
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=feature_dim,
            nhead=num_heads,
            dim_feedforward=feature_dim * 4,
            dropout=0.1,
            batch_first=True,
            activation='gelu'
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers)
        
        # 分类头
        self.classifier = nn.Sequential(
            nn.LayerNorm(feature_dim),
            nn.Linear(feature_dim, 32),
            nn.GELU(),
            nn.Dropout(0.1),
            nn.Linear(32, num_classes)
        )
        
        # 时序有效Token
        self.temporal_mask_token = nn.Parameter(torch.randn(1, 1, feature_dim))
    
    def forward(self, x: torch.Tensor, 
                frame_masks: torch.Tensor = None) -> torch.Tensor:
        """
        Args:
            x: (B, T, N, C) T帧的点云特征
            frame_masks: (B, T, N) 每帧的mask
            
        Returns:
            logits: (B, num_classes)
        """
        B, T, N, C = x.shape
        
        # 每帧聚合为单一向量（mask加权平均）
        if frame_masks is not None:
            weights = frame_masks.unsqueeze(-1)  # (B, T, N, 1)
            weighted_sum = (x * weights).sum(dim=2)  # (B, T, C)
            weight_total = weights.sum(dim=2).clamp(min=1e-6)
            frame_features = weighted_sum / weight_total  # (B, T, C)
        else:
            frame_features = x.mean(dim=2)  # (B, T, C)
        
        # 加位置编码
        frame_features = frame_features + self.pos_encoding[:, :T, :]
        
        # Transformer
        encoded = self.transformer(frame_features)  # (B, T, C)
        
        # 时序池化（注意力加权）
        attn_weights = torch.matmul(
            encoded, 
            self.temporal_mask_token.transpose(-1, -2)
        ).squeeze(-1)  # (B, T)
        attn_weights = F.softmax(attn_weights, dim=1)
        
        pooled = torch.bmm(
            attn_weights.unsqueeze(1), encoded
        ).squeeze(1)  # (B, C)
        
        # 分类
        logits = self.classifier(pooled)
        return logits


class MaskAwareDGCNNTransformer(nn.Module):
    """
    完整模型: Mask-Aware DGCNN + Transformer时序分类器
    
    论文核心架构
    """
    
    def __init__(self, in_channels: int = 4, hidden_dim: int = 64,
                 k: int = 16, num_heads: int = 4, num_layers: int = 2,
                 num_classes: int = 3, max_seq_len: int = 10):
        super().__init__()
        
        # 空间编码: Mask-Aware DGCNN
        self.spatial_encoder = DGCNNWithMask(in_channels, hidden_dim, k)
        
        # 时序分类: Transformer
        self.temporal_classifier = TemporalTransformerClassifier(
            feature_dim=in_channels,  # DGCNN输出维度=输入维度（残差）
            num_heads=num_heads,
            num_layers=num_layers,
            num_classes=num_classes,
            max_seq_len=max_seq_len
        )
    
    def forward(self, x: torch.Tensor, 
                padding_masks: torch.Tensor = None) -> torch.Tensor:
        """
        Args:
            x: (B, T, N, C) T帧点云，N个点，C维特征
            padding_masks: (B, T, N) True=padding
            
        Returns:
            logits: (B, num_classes)
        """
        B, T, N, C = x.shape
        
        # 每帧空间编码
        frame_features = []
        frame_masks = []
        
        for t in range(T):
            feat, mask = self.spatial_encoder(
                x[:, t], 
                padding_masks[:, t] if padding_masks is not None else None
            )
            frame_features.append(feat)
            frame_masks.append(mask)
        
        frame_features = torch.stack(frame_features, dim=1)  # (B, T, N, C)
        frame_masks = torch.stack(frame_masks, dim=1)  # (B, T, N)
        
        # 时序分类
        logits = self.temporal_classifier(frame_features, frame_masks)
        return logits


# ==================== 测试 ====================
if __name__ == "__main__":
    print("=" * 60)
    print("Mask-Aware DGCNN+Transformer mmWave CPD")
    print("论文复现: Sensors (MDPI), 2026")
    print("硬件: TI IWR6843AOP 60GHz")
    print("=" * 60)
    
    # 模型参数
    model = MaskAwareDGCNNTransformer(
        in_channels=4,        # [x, y, z, intensity]
        hidden_dim=64,
        k=16,                 # KNN邻居数
        num_heads=4,
        num_layers=2,
        num_classes=3,        # Child, Cat, Dog
        max_seq_len=10       # 10帧序列
    )
    
    total_params = sum(p.numel() for p in model.parameters())
    print(f"\n模型参数量: {total_params:,} ({total_params/1e6:.2f}M)")
    
    # 模拟输入: 10帧 × 64点 × 4维
    B, T, N, C = 4, 10, 64, 4
    np.random.seed(42)
    
    # 模拟3类样本
    x = torch.randn(B, T, N, C)
    
    # 模拟padding mask（最后几个点是padding）
    padding_masks = torch.zeros(B, T, N, dtype=torch.bool)
    padding_masks[:, :, 50:] = True  # 50-63是padding
    
    # 前向传播
    model.eval()
    with torch.no_grad():
        logits = model(x, padding_masks)
    
    # Softmax
    probs = F.softmax(logits, dim=1)
    preds = torch.argmax(probs, dim=1)
    
    class_names = ["Child", "Cat", "Dog"]
    
    print(f"\n输入: batch={B}, frames={T}, points={N}, channels={C}")
    print(f"输出: {logits.shape}")
    print(f"\n预测结果:")
    for i in range(B):
        print(f"  Sample {i}: {class_names[preds[i]]} "
              f"(confidence: {probs[i][preds[i]]:.2%})")
    
    print(f"\n论文报告准确率: 97.8%")
    print(f"消融实验:")
    print(f"  无Mask机制:  ~93.2% (↓4.6%)")
    print(f"  无Transformer: ~91.5% (↓6.3%)")
    print(f"  无两者:       ~89.0% (↓8.8%)")

实验结果

分类性能

方法	准确率	F1-Score	参数量
PointNet	89.2%	0.88	0.5M
DGCNN (无mask)	93.2%	0.92	1.2M
DGCNN + mask	95.1%	0.94	1.3M
Ours (DGCNN+mask+Transformer)	97.8%	0.97	1.8M
Point Transformer	96.5%	0.95	2.5M

消融分析

组件	准确率变化	关键贡献
完整模型	97.8% (baseline)	—
去掉Mask机制	-4.6%	抑制无效点对KNN图干扰
去掉Transformer	-6.3%	跨帧动态模式建模
去掉两者	-8.8%	退化为单帧DGCNN
去mask + 用max pooling替代attention	-7.2%	证明两者协同效应

分类混淆矩阵

真实\预测	Child	Cat	Dog
Child	98	1	1
Cat	2	96	2
Dog	1	3	96

IMS应用启示

1. 直接应用场景

graph LR
    A[IWR6843AOP 60GHz] --> B[1D-FFT 距离]
    B --> C[2D-FFT 速度]
    C --> D[CFAR 检测]
    D --> E[点云形成]
    E --> F[Mask-Aware DGCNN]
    F --> G[Transformer 时序]
    G --> H{分类}
    H -->|Child| I[CPD 告警]
    H -->|Cat| J[宠物存在]
    H -->|Dog| K[宠物存在]
    H -->|Empty| L[无生命体]

2. 开发落地建议

优先级	建议	输入	输出	硬件
🔴 P0	集成mask机制到现有CPD	点云序列	有效性权重	IWR6843AOP
🔴 P0	人宠分离功能	点云10帧	三分类	同上
🟡 P1	替换呼吸检测为主分类器	同上	类别+置信度	同上
🟢 P2	扩展为成人/儿童/婴儿/宠物	更多数据	四分类	需训练数据

3. 硬件配置

组件	型号	参数	用途
雷达	TI IWR6843AOP	60GHz, 4Tx/4Rx, 3GHz BW	点云采集
处理器	TI AM2732	C66x DSP, Cortex-R5F	实时信号处理
加速器	可选 NPU	>1 TOPS	DGCNN推理
功耗	整体	<2W	停车态常开

4. 与现有方案对比

方案	传感器	准确率	隐私	成本	功耗
本方案	mmWave 60GHz	97.8%	✅ 高	$15-20	<2W
相机方案	RGB/IR	95-98%	❌ 低	$8-12	1-3W
UWB方案	8GHz UWB	90-95%	✅ 高	$5-8	<0.5W
压力传感器	座椅内	85-92%	✅ 高	$3-5	<0.1W

总结

本论文的Mask机制是解决mmWave点云稀疏性和帧间点数波动的关键创新。通过学习每个点的有效性权重，在KNN图构建和特征聚合时自动抑制无效点，将分类准确率从93.2%提升至97.8%。Transformer时序建模进一步捕捉跨帧动态模式（如儿童与猫狗的微小运动节奏差异），贡献6.3%的准确率提升。

对IMS的CPD模块：可直接将mask机制集成到现有IWR6843AOP点云处理管道中，实现人宠分离这一Euro NCAP CPD场景的扩展需求。