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 164 165 166 167 168
| import numpy as np from scipy.signal import butter, filtfilt, find_peaks, stft
class InterferometricVitalSign: """ 干涉雷达生命体征检测算法 原理: 1. 发射连续波/步进频率信号 2. 接收信号与参考信号混频 → 干涉 3. 相位差 → 胸壁位移 (亚毫米精度) 4. 带通滤波 → 分离呼吸和心率 参考: MDPI Sensors 2026, 26(18):5724 """ def __init__(self, config: dict = None): config = config or {} self.fs = config.get('sample_rate', 100) self.freq = config.get('carrier_freq', 60e9) self.breath_band = [0.1, 0.5] self.heart_band = [0.8, 2.0] def phase_to_displacement(self, phase: np.ndarray) -> np.ndarray: """ 相位 → 胸壁位移 d = λ * Δφ / (4π) λ = c / f Args: phase: 相位序列 (弧度) Returns: displacement: 位移 (米) """ wavelength = 3e8 / self.freq displacement = wavelength * phase / (4 * np.pi) return displacement def extract_vital_signs(self, displacement: np.ndarray) -> dict: """ 从胸壁位移提取呼吸和心率 Args: displacement: 胸壁位移序列 (米) Returns: vitals: 生命体征字典 """ breath_signal = self._bandpass(displacement, self.breath_band) heart_signal = self._bandpass(displacement, self.heart_band) freqs = np.fft.rfftfreq(len(displacement), 1/self.fs) breath_fft = np.abs(np.fft.rfft(breath_signal)) heart_fft = np.abs(np.fft.rfft(heart_signal)) breath_mask = (freqs >= self.breath_band[0]) & (freqs <= self.breath_band[1]) heart_mask = (freqs >= self.heart_band[0]) & (freqs <= self.heart_band[1]) breath_freq = freqs[breath_mask][np.argmax(breath_fft[breath_mask])] heart_freq = freqs[heart_mask][np.argmax(heart_fft[heart_mask])] heart_rate_clean = self._suppress_breath_harmonics( heart_fft, heart_freq, breath_freq, freqs ) return { 'breathing_rate': breath_freq * 60, 'heart_rate': heart_rate_clean * 60, 'breath_amplitude': np.max(np.abs(breath_signal)), 'heart_amplitude': np.max(np.abs(heart_signal)), 'snr': self._compute_snr(heart_fft, heart_freq, freqs), } def _suppress_breath_harmonics(self, heart_fft, heart_freq, breath_freq, freqs): """ 呼吸谐波抑制 呼吸的高次谐波 (2x, 3x, 4x...) 可能落入心率频段 需要检测并抑制这些谐波分量 参考: MDPI Sensors 2026, 26(17):5587 """ harmonics = [] n = 2 while True: harmonic_freq = breath_freq * n if harmonic_freq > 2.0: break if 0.8 <= harmonic_freq <= 2.0: harmonics.append(harmonic_freq) n += 1 clean_fft = heart_fft.copy() for hf in harmonics: mask = np.abs(freqs - hf) < 0.05 clean_fft[mask] *= 0.1 heart_mask = (freqs >= self.heart_band[0]) & (freqs <= self.heart_band[1]) clean_heart_freq = freqs[heart_mask][np.argmax(clean_fft[heart_mask])] return clean_heart_freq def _bandpass(self, signal, band): nyq = self.fs / 2 low = band[0] / nyq high = band[1] / nyq b, a = butter(4, [low, high], btype='band') return filtfilt(b, a, signal) def _compute_snr(self, fft, peak_freq, freqs, window=0.05): peak_mask = np.abs(freqs - peak_freq) < window signal_power = np.sum(fft[peak_mask]**2) noise_power = np.sum(fft[~peak_mask]**2) return 10 * np.log10(signal_power / max(noise_power, 1e-10))
if __name__ == "__main__": np.random.seed(42) detector = InterferometricVitalSign({'sample_rate': 100, 'carrier_freq': 60e9}) t = np.arange(1000) / 100.0 breath_phase = 4e-3 * np.sin(2 * np.pi * 0.3 * t) heart_phase = 0.3e-3 * np.sin(2 * np.pi * 1.2 * t) harmonic_phase = 0.5e-3 * np.sin(2 * np.pi * 0.6 * t) noise = 0.1e-3 * np.random.randn(1000) total_displacement = breath_phase + heart_phase + harmonic_phase + noise total_phase = total_displacement * 4 * np.pi / (3e8 / 60e9) displacement = detector.phase_to_displacement(total_phase) vitals = detector.extract_vital_signs(displacement) print("=== 干涉雷达生命体征检测 ===") print(f"呼吸频率: {vitals['breathing_rate']:.1f} 次/分 (真实: 18.0)") print(f"心率: {vitals['heart_rate']:.1f} 次/分 (真实: 72.0)") print(f"呼吸幅度: {vitals['breath_amplitude']*1000:.2f} mm") print(f"心脏幅度: {vitals['heart_amplitude']*1000:.2f} mm") print(f"信噪比: {vitals['snr']:.1f} dB")
|