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
| import numpy as np from scipy import signal
class CPD_Signal_Processor: """ CPD信号处理器 检测链: 1. Range-Doppler Map生成 2. 静态杂波去除 3. 生命体征提取(呼吸/心跳) 4. 儿童分类 """ def __init__(self, num_range_bins=256, num_doppler_bins=64): self.num_range_bins = num_range_bins self.num_doppler_bins = num_doppler_bins def process_frame(self, adc_data): """ 处理单帧雷达数据 Args: adc_data: (num_chirps, num_samples) ADC数据 Returns: detection_map: (H, W) 检测热力图 vital_signs: {'respiration': float, 'heartbeat': float} """ range_fft = np.fft.fft(adc_data, n=self.num_range_bins, axis=1) rd_map = np.fft.fftshift( np.fft.fft(range_fft, n=self.num_doppler_bins, axis=0), axes=0 ) rd_map_filtered = self._remove_static_clutter(rd_map) vital_signs = self._extract_vital_signs(rd_map) detection_map = np.abs(rd_map_filtered) return detection_map, vital_signs def _remove_static_clutter(self, rd_map): """ 去除静态杂波(座椅/车内饰) 留下动态信号(呼吸运动) """ doppler_axis = 0 b, a = signal.butter(4, 0.1, 'high') filtered = signal.filtfilt(b, a, rd_map, axis=doppler_axis) return filtered def _extract_vital_signs(self, rd_map): """ 从RD图提取生命体征 检测: - 呼吸频率: 0.2-0.5 Hz (12-30次/分钟) - 心跳频率: 1.0-2.0 Hz (60-120次/分钟) """ low_velocity_region = rd_map[ self.num_doppler_bins//2-5:self.num_doppler_bins//2+5, : ] velocity_profile = np.sum(np.abs(low_velocity_region), axis=1) freq_spectrum = np.fft.fft(velocity_profile) freqs = np.fft.fftfreq(len(velocity_profile), d=0.1) resp_band = (freqs > 0.2) & (freqs < 0.5) resp_freq = freqs[resp_band][np.argmax(np.abs(freq_spectrum[resp_band]))] heart_band = (freqs > 1.0) & (freqs < 2.0) heart_freq = freqs[heart_band][np.argmax(np.abs(freq_spectrum[heart_band]))] return { 'respiration': resp_freq * 60, 'heartbeat': heart_freq * 60 }
|