HEAR：mmWave雷达心率检测的可观测性评估与选择性估计（arXiv 2610.03570 论文解读+代码复现）

HEAR：mmWave雷达心率检测的可观测性评估与选择性估计

论文信息

项目	内容
标题	Learning to Assess Heartbeat Observability for mmWave Heart-Rate Sensing
作者	Yuxuan Hu 等
年份	2026
arXiv	2610.03570
链接	https://arxiv.org/abs/2610.03570
项目页	https://yuxuanhu9.github.io/HEAR/
领域	cs.AI

1. 核心创新

1.1 问题定义

mmWave 雷达非接触心率检测的核心挑战：并非所有测量都支持可靠的心率估计。

相干散射叠加可能抑制心跳分量，即使宏观观测几何相似。这意味着：

相同距离 + 相同姿态 → 不同的心跳可观测性
                                    ↓
                    有时心跳信号清晰，有时被淹没

1.2 核心方法

提出 HEAR (Heartbeat Estimation with Assessed Reliability)：

1. 可观测性评估：学习一个评分，判断当前测量是否支持可靠心率估计
2. 选择性估计：只在高可观测性帧上估计心率
3. 双任务 Transformer：同时预测可观测性分数和心率
4. FMCW 仿真器：生成可控多散射体训练数据
5. 零样本迁移：仅用仿真数据训练，直接迁移到 60GHz 和 120GHz 真实数据

1.3 关键结果

指标	数值
120GHz 全覆盖 MAE	17.9 BPM
120GHz 50%覆盖 MAE	1.6 BPM（选择性估计后）
端到端延迟	50.8 ms（边缘设备）
迁移频率	60GHz + 120GHz
被试数	134人（两个公开数据集）
训练数据	仅仿真数据

---

2. 方法详解

2.1 系统架构

graph TD
    A[FMCW 仿真器<br/>多散射体模拟] -->|生成训练数据| B[训练阶段]
    
    B --> C[HEAR 双任务 Transformer]
    C --> C1[可观测性评分头]
    C --> C2[心率估计头]
    
    C1 -->|可观测性分数| D[选择性估计]
    C2 -->|心率估计| D
    
    D -->|高可观测性帧| E[可靠心率输出]
    D -->|低可观测性帧| F[跳过该帧]
    
    G[真实 60GHz/120GHz 数据] -->|零样本迁移| C

2.2 FMCW 多散射体仿真器

"""
HEAR 论文的 FMCW 多散射体仿真器
生成可控可观测性标签的训练数据
"""

import numpy as np
from dataclasses import dataclass
from typing import List, Tuple

@dataclass
class Scatterer:
    """单个散射体"""
    range_m: float        # 距离
    phase_rad: float      # 相位
    amplitude: float      # 振幅
    has_heartbeat: bool   # 是否包含心跳分量

class FMCWSimulator:
    """
    FMCW 多散射体仿真器
    
    模拟多个散射体的相干叠加
    通过控制各散射体的参数生成不同可观测性条件
    """
    
    def __init__(self, config: dict):
        self.center_freq = config.get('center_freq', 60e9)  # 60 or 120 GHz
        self.bandwidth = config.get('bandwidth', 4e9)
        self.num_samples = config.get('num_samples', 128)
        self.sample_rate = config.get('sample_rate', 1e6)
        self.c = 3e8
        
    def generate_measurement(self, scatterers: List[Scatterer],
                              heart_rate_hz: float,
                              duration_frames: int = 100) -> Tuple[np.ndarray, float]:
        """
        生成一次测量数据
        
        Args:
            scatterers: 散射体列表
            heart_rate_hz: 真实心率 (Hz)
            duration_frames: 帧数
            
        Returns:
            phase_spectrum: 相位谱 (num_frames, num_range_bins)
            observability_label: 可观测性标签 (0-1)
        """
        c = self.c
        wavelength = c / self.center_freq
        
        # 距离分辨率
        range_resolution = c / (2 * self.bandwidth)
        
        phase_spectrum = np.zeros((duration_frames, self.num_samples), dtype=complex)
        
        for frame_idx in range(duration_frames):
            t = frame_idx / 20.0  # 假设帧率 20 Hz
            
            # 心跳位移（正弦）
            heartbeat_displacement = 0.5e-3 * np.sin(2 * np.pi * heart_rate_hz * t)  # 0.5mm
            
            for scatterer in scatterers:
                # 距离bin
                range_bin = int(scatterer.range_m / range_resolution)
                
                if range_bin >= self.num_samples:
                    continue
                
                # 相位 = 4 * pi * distance / lambda
                if scatterer.has_heartbeat:
                    effective_range = scatterer.range_m + heartbeat_displacement
                else:
                    effective_range = scatterer.range_m
                
                phase = 4 * np.pi * effective_range / wavelength + scatterer.phase_rad
                
                # 叠加到距离bin
                phase_spectrum[frame_idx, range_bin] += (
                    scatterer.amplitude * np.exp(1j * phase)
                )
        
        # 计算可观测性标签
        # 检查心跳频带是否有峰值
        # 选择包含心跳散射体的距离bin
        heartbeat_bins = []
        for i, s in enumerate(scatterers):
            if s.has_heartbeat:
                heartbeat_bins.append(int(s.range_m / range_resolution))
        
        if not heartbeat_bins:
            return phase_spectrum, 0.0
        
        # 在心跳bin上做FFT，检查心率峰值
        observability_scores = []
        for bin_idx in heartbeat_bins:
            phase_series = np.unwrap(np.angle(phase_spectrum[:, bin_idx]))
            phase_series = phase_series - np.mean(phase_series)
            
            # FFT
            spectrum = np.abs(np.fft.fft(phase_series))
            freqs = np.fft.fftfreq(len(phase_series), d=1/20)
            
            # 检查心率频率处的峰值
            hr_idx = np.argmin(np.abs(freqs - heart_rate_hz))
            peak_power = spectrum[hr_idx]
            total_power = np.sum(spectrum) + 1e-10
            
            observability_scores.append(peak_power / total_power)
        
        # 可观测性 = 心跳峰值的显著性
        observability = np.mean(observability_scores)
        
        # 归一化到 0-1
        observability = min(observability / 0.1, 1.0)
        
        return phase_spectrum, observability


# 使用示例
if __name__ == "__main__":
    sim = FMCWSimulator({
        'center_freq': 60e9,
        'bandwidth': 4e9,
        'num_samples': 128,
        'sample_rate': 1e6
    })
    
    # 场景1：心跳散射体主导（高可观测性）
    scatterers_good = [
        Scatterer(range_m=0.5, phase_rad=0, amplitude=1.0, has_heartbeat=True),
        Scatterer(range_m=0.5, phase_rad=0.5, amplitude=0.1, has_heartbeat=False),
    ]
    spec, obs = sim.generate_measurement(scatterers_good, heart_rate_hz=1.2, duration_frames=100)
    print(f"高可观测性场景: obs={obs:.3f}")
    
    # 场景2：多个非心跳散射体叠加（低可观测性）
    scatterers_bad = [
        Scatterer(range_m=0.5, phase_rad=0, amplitude=0.3, has_heartbeat=True),
        Scatterer(range_m=0.5, phase_rad=2.0, amplitude=0.8, has_heartbeat=False),
        Scatterer(range_m=0.51, phase_rad=1.5, amplitude=0.7, has_heartbeat=False),
    ]
    spec, obs = sim.generate_measurement(scatterers_bad, heart_rate_hz=1.2, duration_frames=100)
    print(f"低可观测性场景: obs={obs:.3f}")

2.3 HEAR 双任务 Transformer

"""
HEAR 双任务 Transformer 模型
同时预测可观测性分数和心率
"""

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

class HEARTransformer(nn.Module):
    """
    HEAR 双任务 Transformer
    
    输入：相位谱幅度 + 相对频率（相对于呼吸基频）
    输出：可观测性分数 + 心率估计
    
    设计要点：
    1. 输入结合频谱幅度和呼吸谐波上下文
    2. 紧凑模型（适合边缘部署）
    3. 双任务共享编码器
    """
    
    def __init__(self, config: dict):
        super().__init__()
        input_dim = config.get('input_dim', 128)  # 频率bin数
        d_model = config.get('d_model', 64)       # 模型维度
        nhead = config.get('nhead', 4)            # 注意力头数
        num_layers = config.get('num_layers', 2)  # Transformer层数
        
        # 输入编码
        self.input_proj = nn.Linear(input_dim, d_model)
        
        # 位置编码
        self.pos_encoding = nn.Parameter(torch.randn(1, input_dim, d_model) * 0.02)
        
        # Transformer 编码器
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=d_model,
            nhead=nhead,
            dim_feedforward=d_model * 4,
            dropout=0.1,
            batch_first=True
        )
        self.encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
        
        # 可观测性评分头
        self.obs_head = nn.Sequential(
            nn.LayerNorm(d_model),
            nn.Linear(d_model, d_model // 2),
            nn.GELU(),
            nn.Linear(d_model // 2, 1),
            nn.Sigmoid()
        )
        
        # 心率估计头
        self.hr_head = nn.Sequential(
            nn.LayerNorm(d_model),
            nn.Linear(d_model, d_model // 2),
            nn.GELU(),
            nn.Linear(d_model // 2, 1),
            nn.Softplus()  # 心率 > 0
        )
        
    def forward(self, spectral_magnitude: torch.Tensor, 
                rel_freq: torch.Tensor) -> tuple:
        """
        前向传播
        
        Args:
            spectral_magnitude: 相位谱幅度, (B, F)
            rel_freq: 相对呼吸基频的频率, (B, F)
            
        Returns:
            (observability_score, heart_rate)
        """
        # 拼接输入
        x = torch.stack([spectral_magnitude, rel_freq], dim=-1)  # (B, F, 2)
        x = x.mean(dim=-1)  # 简化：(B, F)
        
        # 投影到模型维度
        x = self.input_proj(x)  # (B, F, d_model)
        
        # 位置编码
        x = x + self.pos_encoding
        
        # Transformer 编码
        x = self.encoder(x)  # (B, F, d_model)
        
        # 全局池化
        x = x.mean(dim=1)  # (B, d_model)
        
        # 双任务输出
        obs_score = self.obs_head(x).squeeze(-1)  # (B,)
        heart_rate = self.hr_head(x).squeeze(-1)  # (B,)
        
        return obs_score, heart_rate


# 测试
if __name__ == "__main__":
    model = HEARTransformer({
        'input_dim': 128,
        'd_model': 64,
        'nhead': 4,
        'num_layers': 2
    })
    
    # 模拟输入
    B = 4  # batch size
    spectral = torch.randn(B, 128)
    rel_freq = torch.randn(B, 128)
    
    obs, hr = model(spectral, rel_freq)
    
    print(f"可观测性分数: {obs.detach().numpy()}")
    print(f"心率估计: {hr.detach().numpy()} BPM")
    print(f"模型参数量: {sum(p.numel() for p in model.parameters()):,}")
    print(f"模型大小: {sum(p.numel() for p in model.parameters()) * 4 / 1024:.1f} KB")

2.4 选择性心率估计流程

"""
选择性心率估计：只在高可观测性帧上估计
"""

class SelectiveHREstimator:
    """
    基于可观测性分数的选择性心率估计
    
    策略：
    1. 对每帧预测可观测性分数
    2. 设定覆盖阈值（如 50%）
    3. 选择可观测性最高的前 N% 帧
    4. 在这些帧上估计心率
    """
    
    def __init__(self, model: HEARTransformer, coverage: float = 0.5):
        self.model = model
        self.coverage = coverage  # 选择多少比例的帧
        
    def estimate(self, spectral: torch.Tensor, rel_freq: torch.Tensor) -> dict:
        """
        选择性心率估计
        
        Returns:
            {
                'heart_rate': float,        # 估计心率
                'confidence': float,        # 置信度
                'selected_ratio': float,    # 选择的帧比例
                'mean_observability': float # 平均可观测性
            }
        """
        with torch.no_grad():
            obs_scores, hr_estimates = self.model(spectral, rel_freq)
        
        obs_scores = obs_scores.numpy()
        hr_estimates = hr_estimates.numpy()
        
        # 按可观测性排序
        n_total = len(obs_scores)
        n_select = int(n_total * self.coverage)
        
        sorted_idx = np.argsort(obs_scores)[::-1]  # 降序
        selected_idx = sorted_idx[:n_select]
        
        # 只在选择帧上取心率
        selected_hr = hr_estimates[selected_idx]
        selected_obs = obs_scores[selected_idx]
        
        # 加权平均（按可观测性加权）
        weights = selected_obs / (np.sum(selected_obs) + 1e-10)
        final_hr = np.sum(selected_hr * weights)
        
        return {
            'heart_rate': float(final_hr),
            'confidence': float(np.mean(selected_obs)),
            'selected_ratio': float(n_select / n_total),
            'mean_observability': float(np.mean(obs_scores))
        }


# 测试
if __name__ == "__main__":
    model = HEARTransformer({'input_dim': 128, 'd_model': 64, 'nhead': 4, 'num_layers': 2})
    estimator = SelectiveHREstimator(model, coverage=0.5)
    
    # 模拟 20 帧数据
    spectral = torch.randn(20, 128)
    rel_freq = torch.randn(20, 128)
    
    result = estimator.estimate(spectral, rel_freq)
    print(f"心率: {result['heart_rate']:.1f} BPM")
    print(f"置信度: {result['confidence']:.3f}")
    print(f"选择帧比例: {result['selected_ratio']:.0%}")
    print(f"平均可观测性: {result['mean_observability']:.3f}")

---

3. 对 IMS 开发的启示

3.1 核心价值

启示	描述
不是所有帧都可用	mmWave 心率检测必须做帧级质量控制
仿真训练可行	仅用仿真数据训练，零样本迁移到真实数据
边缘部署友好	50.8ms 延迟，模型紧凑
双频率兼容	60GHz 和 120GHz 都适用

3.2 落地建议

优先级	建议
🔴 P0	在 mmWave 心率检测中引入可观测性评分机制
🔴 P0	用 FMCW 仿真器生成训练数据，解决数据不足问题
🟡 P1	部署选择性估计策略，50% 覆盖率下 MAE 从 17.9→1.6 BPM
🟡 P1	评估 60GHz vs 120GHz 的可观测性差异

3.3 与 IMS 系统集成

"""
IMS 系统中 mmWave 心率监测模块集成
"""

class IMSHealthMonitor:
    """IMS 健康监测模块"""
    
    def __init__(self):
        self.hr_estimator = SelectiveHREstimator(
            HEARTransformer({'input_dim': 128, 'd_model': 64}),
            coverage=0.5
        )
        self.hr_history = []
        self.alert_thresholds = {
            'hr_low': 40,       # 心动过缓
            'hr_high': 120,     # 心动过速
            'hr_variability_low': 10  # HRV 过低
        }
    
    def update(self, radar_data: np.ndarray) -> dict:
        """更新健康状态"""
        spectral = torch.tensor(radar_data['magnitude']).float()
        rel_freq = torch.tensor(radar_data['rel_freq']).float()
        
        result = self.hr_estimator.estimate(spectral, rel_freq)
        
        self.hr_history.append(result['heart_rate'])
        if len(self.hr_history) > 300:  # 保留5分钟数据
            self.hr_history.pop(0)
        
        # 异常检测
        alerts = []
        hr = result['heart_rate']
        if hr < self.alert_thresholds['hr_low']:
            alerts.append({'type': 'bradycardia', 'value': hr, 'severity': 'high'})
        if hr > self.alert_thresholds['hr_high']:
            alerts.append({'type': 'tachycardia', 'value': hr, 'severity': 'high'})
        
        return {
            'heart_rate': hr,
            'confidence': result['confidence'],
            'alerts': alerts
        }

---

4. 参考

• arXiv 2610.03570
• 项目页
• 60GHz 和 120GHz mmWave 心率数据集