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 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225
| import numpy as np from scipy import signal from scipy.fft import fft, fftfreq from typing import Tuple, Dict
class VitalSignsExtractor: """生命体征信号提取器 基于60GHz FMCW雷达实现心率/呼吸率提取 """ def __init__(self, sampling_rate: float = 100.0): """ Args: sampling_rate: 采样率 (Hz) """ self.sampling_rate = sampling_rate self.breath_band = (0.1, 0.5) self.heart_band = (0.8, 2.0) def extract_from_radar(self, radar_data: np.ndarray) -> Dict: """ 从雷达数据提取生命体征 Args: radar_data: (N,) 雷达相位信号,N个采样点 Returns: vital_signs: {'heart_rate': bpm, 'respiration_rate': brpm} """ radar_data = signal.detrend(radar_data) radar_data = radar_data - np.mean(radar_data) breath_signal = self._bandpass_filter(radar_data, self.breath_band) heart_signal = self._bandpass_filter(radar_data, self.heart_band) breath_freq, breath_spectrum = self._compute_spectrum(breath_signal) heart_freq, heart_spectrum = self._compute_spectrum(heart_signal) breath_rate = self._find_peak_frequency(breath_freq, breath_spectrum) heart_rate = self._find_peak_frequency(heart_freq, heart_spectrum) respiration_rate = breath_rate * 60 heart_rate_bpm = heart_rate * 60 return { 'heart_rate': heart_rate_bpm, 'respiration_rate': respiration_rate, 'signal_quality': self._assess_signal_quality(radar_data) } def _bandpass_filter(self, data: np.ndarray, band: Tuple[float, float]) -> np.ndarray: """带通滤波""" nyquist = self.sampling_rate / 2 low = band[0] / nyquist high = band[1] / nyquist b, a = signal.butter(4, [low, high], btype='band') return signal.filtfilt(b, a, data) def _compute_spectrum(self, data: np.ndarray) -> Tuple[np.ndarray, np.ndarray]: """计算频谱""" N = len(data) spectrum = np.abs(fft(data))[:N//2] freqs = fftfreq(N, 1/self.sampling_rate)[:N//2] return freqs, spectrum def _find_peak_frequency(self, freqs: np.ndarray, spectrum: np.ndarray) -> float: """找到峰值频率""" peak_idx = np.argmax(spectrum) return freqs[peak_idx] def _assess_signal_quality(self, data: np.ndarray) -> float: """评估信号质量""" signal_power = np.var(data) noise_power = np.var(np.diff(data)) / 12 snr = 10 * np.log10(signal_power / noise_power) if noise_power > 0 else 100 quality = np.clip((snr - 10) / 20, 0, 1) return quality
class CPDDetector: """儿童存在检测器 基于60GHz雷达实现CPD功能 """ def __init__(self): self.thresholds = { 'min_chest_movement': 0.5e-3, 'min_occupancy_time': 30, 'breath_rate_range': (20, 60), } self.detection_history = [] def detect(self, radar_data: np.ndarray, range_profile: np.ndarray, vital_signs: Dict) -> Dict: """ 检测儿童存在 Args: radar_data: 雷达相位信号 range_profile: 距离-强度剖面(用于定位) vital_signs: 已提取的生命体征 Returns: cpd_result: CPD检测结果 """ has_vital_signs = vital_signs['heart_rate'] > 0 and vital_signs['respiration_rate'] > 0 is_infant = self._classify_infant(vital_signs, range_profile) has_micro_movement = self._detect_micro_movement(radar_data) child_detected = has_vital_signs and (is_infant or has_micro_movement) self.detection_history.append(child_detected) if len(self.detection_history) > self.thresholds['min_occupancy_time']: detection_rate = np.mean(self.detection_history[-self.thresholds['min_occupancy_time']:]) child_present = detection_rate > 0.8 else: child_present = False return { 'child_present': child_present, 'confidence': np.mean(self.detection_history[-10:]) if len(self.detection_history) > 10 else 0, 'location': self._estimate_location(range_profile), 'vital_signs': vital_signs } def _classify_infant(self, vital_signs: Dict, range_profile: np.ndarray) -> bool: """判断是否为婴儿""" breath_rate = vital_signs['respiration_rate'] heart_rate = vital_signs['heart_rate'] is_infant_breath = self.thresholds['breath_rate_range'][0] < breath_rate < self.thresholds['breath_rate_range'][1] is_infant_heart = 100 < heart_rate < 160 return is_infant_breath and is_infant_heart def _detect_micro_movement(self, radar_data: np.ndarray) -> bool: """检测微小运动(婴儿特有)""" phase_std = np.std(np.angle(radar_data)) return phase_std > 0.1 def _estimate_location(self, range_profile: np.ndarray) -> Dict: """估计位置""" peak_idx = np.argmax(range_profile) distance = peak_idx * 0.01 return { 'distance': distance, 'angle': 0, 'confidence': range_profile[peak_idx] / np.max(range_profile) }
if __name__ == "__main__": vital_extractor = VitalSignsExtractor(sampling_rate=100.0) cpd_detector = CPDDetector() t = np.linspace(0, 30, 3000) breath_signal = 0.1 * np.sin(2 * np.pi * 0.25 * t) heart_signal = 0.02 * np.sin(2 * np.pi * 1.2 * t) radar_data = breath_signal + heart_signal + np.random.normal(0, 0.005, len(t)) vital_signs = vital_extractor.extract_from_radar(radar_data) print(f"心率: {vital_signs['heart_rate']:.1f} bpm") print(f"呼吸率: {vital_signs['respiration_rate']:.1f} brpm") print(f"信号质量: {vital_signs['signal_quality']:.2f}") range_profile = np.random.rand(100) cpd_result = cpd_detector.detect(radar_data, range_profile, vital_signs) print(f"\n儿童存在: {cpd_result['child_present']}") print(f"置信度: {cpd_result['confidence']:.2f}")
|