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
| import numpy as np from scipy import signal
class FMCWRadarProcessor: """ FMCW(调频连续波)雷达信号处理 工作原理: 1. 发射线性调频信号 2. 接收反射信号 3. 混频得到中频信号 4. FFT得到距离-多普勒图 """ def __init__(self, config): self.fs = config['sample_rate'] self.slope = config['chirp_slope'] self.num_chirps = config['num_chirps'] self.num_samples = config['samples_per_chirp'] def process_frame(self, adc_data): """ 处理一帧数据 Args: adc_data: (num_chirps, num_samples) ADC数据 Returns: range_doppler_map: 距离-多普勒图 """ range_fft = np.fft.fft(adc_data, axis=1) range_doppler_map = np.fft.fftshift( np.fft.fft(range_fft, axis=0), axes=0 ) range_doppler_map = np.abs(range_doppler_map) return range_doppler_map def extract_phase(self, adc_data, range_bin): """ 提取相位信息 用于生命体征检测 """ range_fft = np.fft.fft(adc_data, axis=1) phase = np.angle(range_fft[:, range_bin]) return phase def detect_breathing(self, phase): """ 从相位检测呼吸 呼吸频率:0.1-0.5 Hz (6-30 BPM) """ phase_unwrapped = np.unwrap(phase) phase_detrended = signal.detrend(phase_unwrapped) fs = self.num_chirps b, a = signal.butter(4, [0.1/(fs/2), 0.5/(fs/2)], btype='band') breathing_signal = signal.filtfilt(b, a, phase_detrended) freq = np.fft.fftfreq(len(breathing_signal), 1/fs) spectrum = np.fft.fft(breathing_signal) resp_mask = (freq >= 0.1) & (freq <= 0.5) resp_spectrum = np.abs(spectrum[resp_mask]) resp_freq = freq[resp_mask] peak_freq = resp_freq[np.argmax(resp_spectrum)] breathing_rate = peak_freq * 60 return breathing_rate, breathing_signal
|