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
| class TrunkMotionDetector: """ 后备箱运动检测器 基于纯时域特征区分呼吸和后备箱运动 """ def __init__(self, fps: int = 20, window_sec: int = 10): self.fps = fps self.window_sec = window_sec self.window_size = fps * window_sec def detect(self, radar_signal: np.ndarray) -> dict: """ 检测信号中是否包含后备箱运动 Args: radar_signal: 雷达回波信号 (N,) Returns: result: 检测结果 """ windows = self._split_windows(radar_signal) results = [] for window in windows: features = self._extract_time_domain_features(window) is_trunk = self._classify(features) results.append({ 'features': features, 'is_trunk_motion': is_trunk, 'confidence': features['regularity_score'] }) trunk_count = sum(1 for r in results if r['is_trunk_motion']) is_trunk_present = trunk_count > len(results) / 2 return { 'is_trunk_motion': is_trunk_present, 'confidence': trunk_count / len(results) if results else 0, 'window_results': results, } def _extract_time_domain_features(self, window: np.ndarray) -> dict: """ 提取时域特征(纯时域,无需FFT) """ features = {} features['amplitude'] = np.max(window) - np.min(window) features['rms'] = np.sqrt(np.mean(window**2)) features['regularity_score'] = self._calc_regularity(window) zero_crossings = np.sum(np.diff(np.sign(window)) != 0) features['zero_crossing_rate'] = zero_crossings / len(window) from scipy.signal import find_peaks peaks, _ = find_peaks(window, height=0.5 * np.max(window)) features['peak_count'] = len(peaks) if len(peaks) > 1: peak_intervals = np.diff(peaks) / self.fps features['peak_interval_mean'] = np.mean(peak_intervals) features['peak_interval_std'] = np.std(peak_intervals) features['peak_interval_cv'] = (features['peak_interval_std'] / (features['peak_interval_mean'] + 1e-6)) else: features['peak_interval_mean'] = 0 features['peak_interval_std'] = 0 features['peak_interval_cv'] = 0 features['energy'] = np.sum(window**2) return features def _calc_regularity(self, window: np.ndarray) -> float: """ 计算规则性评分 后备箱运动更规则 → 评分高 呼吸运动有更多变化 → 评分低 方法:自相关函数的峰值锐利度 """ autocorr = np.correlate(window, window, mode='full') autocorr = autocorr[len(autocorr)//2:] autocorr = autocorr / (autocorr[0] + 1e-6) from scipy.signal import find_peaks peaks, properties = find_peaks(autocorr[1:], height=0.3) if len(peaks) == 0: return 0.0 peak_idx = peaks[0] + 1 peak_height = autocorr[peak_idx] half_max = peak_height / 2 left = peak_idx right = peak_idx while left > 0 and autocorr[left] > half_max: left -= 1 while right < len(autocorr) - 1 and autocorr[right] > half_max: right += 1 fwhm = right - left regularity = peak_height / (fwhm + 1e-6) return float(regularity) def _classify(self, features: dict) -> bool: """ 分类:是否为后备箱运动 规则:高规则性 + 大振幅 + 低变异系数 → 后备箱运动 """ score = 0 if features['regularity_score'] > 0.5: score += 1 if features['amplitude'] > 2.0: score += 1 if features['peak_interval_cv'] < 0.2: score += 1 if 0.1 < features['zero_crossing_rate'] < 0.3: score += 1 return score >= 3 def _split_windows(self, signal: np.ndarray) -> list: """分窗处理""" windows = [] step = self.window_size // 2 for i in range(0, len(signal) - self.window_size + 1, step): windows.append(signal[i:i + self.window_size]) return windows
if __name__ == "__main__": sim = TrunkMotionInterference(fps=20) detector = TrunkMotionDetector(fps=20, window_sec=10) breathing = sim.simulate_breathing(60, 20) result_breathing = detector.detect(breathing) print(f"纯呼吸信号 - 后备箱运动: {result_breathing['is_trunk_motion']}, " f"置信度: {result_breathing['confidence']:.2f}") trunk = sim.simulate_trunk_motion(60, 15) result_trunk = detector.detect(trunk) print(f"纯后备箱运动 - 后备箱运动: {result_trunk['is_trunk_motion']}, " f"置信度: {result_trunk['confidence']:.2f}") combined = sim.simulate_combined(60) result_combined = detector.detect(combined) print(f"混合信号 - 后备箱运动: {result_combined['is_trunk_motion']}, " f"置信度: {result_combined['confidence']:.2f}")
|