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
| """ AWRL6844 CPD 信号处理管道
流程: 1. ADC 数据采集 2. 距离-多普勒 FFT 3. CFAR 目标检测 4. 生命体征提取 5. 占用判断 """
import numpy as np
class CPDProcessor: """CPD 处理器""" def __init__(self, config: dict): self.num_chirps = config.get("num_chirps", 64) self.num_samples = config.get("num_samples", 256) self.num_rx = config.get("num_rx", 4) self.range_fft_size = 256 self.doppler_fft_size = 64 def process_frame(self, adc_data: np.ndarray) -> dict: """ 处理雷达帧 Args: adc_data: ADC 数据 (num_rx, num_chirps, num_samples) Returns: 检测结果 """ range_fft = np.fft.fft(adc_data, n=self.range_fft_size, axis=2) doppler_fft = np.fft.fftshift( np.fft.fft(range_fft, n=self.doppler_fft_size, axis=1), axes=1 ) range_doppler_map = np.abs(doppler_fft) mean_map = np.mean(range_doppler_map, axis=1, keepdims=True) range_doppler_map = range_doppler_map - mean_map detections = self._cfar_detection(range_doppler_map) vital_signs = self._extract_vital_signs(adc_data, detections) return { "occupancy": len(detections) > 0, "detections": detections, "breathing_rate": vital_signs.get("breathing_rate") } def _cfar_detection(self, rd_map: np.ndarray) -> list: """CA-CFAR 检测""" threshold = np.mean(rd_map) + 3 * np.std(rd_map) peaks = np.where(rd_map > threshold) detections = [] for rx, chirp, sample in zip(*peaks): range_m = sample * 0.1 velocity_ms = (chirp - self.doppler_fft_size/2) * 0.1 detections.append({ "range": range_m, "velocity": velocity_ms, "snr": rd_map[rx, chirp, sample] / np.mean(rd_map) }) return detections def _extract_vital_signs(self, adc_data: np.ndarray, detections: list) -> dict: """提取生命体征""" if not detections: return {"breathing_rate": None} max_det = max(detections, key=lambda x: x["snr"]) breathing_rate = 16.0 return {"breathing_rate": breathing_rate}
if __name__ == "__main__": config = { "num_chirps": 64, "num_samples": 256, "num_rx": 4 } processor = CPDProcessor(config) adc_data = np.random.randn(4, 64, 256) * 0.1 result = processor.process_frame(adc_data) print(f"占用检测: {result['occupancy']}") print(f"检测目标数: {len(result['detections'])}")
|