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 157 158 159 160 161 162 163
| """ 120GHz 雷达生命体征检测算法 利用高微动灵敏度检测呼吸和心跳 """
import numpy as np from scipy.signal import butter, filtfilt, find_peaks
class VitalSignsDetector: """ 120GHz 雷达生命体征检测器 工作原理: 1. 锁定目标距离单元 2. 提取相位变化序列(相位 = 胸壁位移的 2 倍) 3. 带通滤波分离呼吸和心跳 4. 频谱分析确定频率 """ def __init__(self, config: dict): self.breath_band = (0.1, 0.5) self.heart_band = (0.8, 2.5) self.fs = config.get('frame_rate', 20) def extract_phase(self, if_signal: np.ndarray, range_bin: int) -> np.ndarray: """ 从 IF 信号中提取相位序列 Args: if_signal: shape=(num_frames, num_samples) range_bin: 目标距离单元 Returns: phase: 相位序列(弧度) """ range_fft = np.fft.fft(if_signal, axis=1) complex_val = range_fft[:, range_bin] phase = np.angle(complex_val) phase = np.unwrap(phase) phase = phase - np.mean(phase) return phase def separate_breath_heart(self, phase: np.ndarray) -> tuple: """ 分离呼吸和心跳信号 Returns: breath_signal, heart_signal """ nyq = self.fs / 2 b_breath, a_breath = butter( 4, [self.breath_band[0]/nyq, self.breath_band[1]/nyq], btype='band' ) breath_signal = filtfilt(b_breath, a_breath, phase) b_heart, a_heart = butter( 4, [self.heart_band[0]/nyq, self.heart_band[1]/nyq], btype='band' ) heart_signal = filtfilt(b_heart, a_heart, phase) return breath_signal, heart_signal def estimate_rate(self, signal: np.ndarray, freq_range: tuple) -> float: """ 估算频率(呼吸率或心率) """ spectrum = np.abs(np.fft.fft(signal)) freqs = np.fft.fftfreq(len(signal), d=1.0/self.fs) mask = (freqs >= freq_range[0]) & (freqs <= freq_range[1]) if not np.any(mask): return 0.0 peak_idx = np.argmax(spectrum[mask]) peak_freq = freqs[mask][peak_idx] return peak_freq def detect_vital_signs(self, if_signal: np.ndarray, range_bin: int) -> dict: """ 完整生命体征检测 Returns: { 'breathing_rate': float, # Hz 'heart_rate': float, # Hz 'breathing_amplitude_mm': float, 'heart_amplitude_mm': float, 'snr_db': float } """ phase = self.extract_phase(if_signal, range_bin) breath, heart = self.separate_breath_heart(phase) breath_rate = self.estimate_rate(breath, self.breath_band) heart_rate = self.estimate_rate(heart, self.heart_band) wavelength = 2.5e-3 breath_amp_mm = np.std(breath) * wavelength / (4 * np.pi) * 1000 heart_amp_mm = np.std(heart) * wavelength / (4 * np.pi) * 1000 signal_power = np.var(breath) + np.var(heart) noise_power = np.var(phase - breath - heart) snr_db = 10 * np.log10(signal_power / (noise_power + 1e-10)) return { 'breathing_rate': breath_rate, 'heart_rate': heart_rate, 'breathing_rate_bpm': breath_rate * 60, 'heart_rate_bpm': heart_rate * 60, 'breathing_amplitude_mm': breath_amp_mm, 'heart_amplitude_mm': heart_amp_mm, 'snr_db': snr_db }
if __name__ == "__main__": detector = VitalSignsDetector({'frame_rate': 20}) np.random.seed(42) if_sig = np.random.randn(100, 256) * 0.01 t = np.arange(100) breath_phase = 0.5 * np.sin(2 * np.pi * 0.3 * t / 20) heart_phase = 0.1 * np.sin(2 * np.pi * 1.2 * t / 20) for i in range(100): if_sig[i, 10] += np.exp(1j * (breath_phase[i] + heart_phase[i])) results = detector.detect_vital_signs(if_sig, range_bin=10) print("=== 生命体征检测结果 ===") print(f"呼吸率: {results['breathing_rate_bpm']:.1f} breaths/min") print(f"心率: {results['heart_rate_bpm']:.1f} bpm") print(f"呼吸幅度: {results['breathing_amplitude_mm']:.2f} mm") print(f"心跳幅度: {results['heart_amplitude_mm']:.2f} mm") print(f"SNR: {results['snr_db']:.1f} dB")
|