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 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256
| """ Tesla 车内60GHz雷达系统模拟 基于 TI AWR6843 参数
核心功能: 1. 乘员存在检测 (穿透遮挡) 2. 乘员分类 (成人/儿童/宠物) 3. 生命体征监测 (心率/呼吸) 4. 儿童遗弃检测 (CPD) """
import numpy as np from dataclasses import dataclass from typing import Tuple, Optional
@dataclass class RadarConfig: """TI AWR6843 配置""" freq_min: float = 60e9 freq_max: float = 64e9 bandwidth: float = 4e9 num_tx: int = 3 num_rx: int = 4 range_resolution: float = 0.0375 velocity_resolution: float = 0.1 angular_resolution: float = 15 frame_rate: float = 10 @property def wavelength(self): return 3e8 / ((self.freq_min + self.freq_max) / 2)
class TeslaCabinRadar: """Tesla 车内雷达系统""" def __init__(self): self.config = RadarConfig() self.baseline = None self.hr_buffer = [] self.br_buffer = [] def process_frame(self, radar_cube: np.ndarray) -> dict: """ 处理一帧雷达数据 Args: radar_cube: (num_rx, num_chirps, num_samples) Returns: 检测结果字典 """ range_doppler = self._range_doppler_fft(radar_cube) range_angle = self._angle_estimation(radar_cube) presence = self._detect_presence(range_doppler) category = self._classify_occupant(range_doppler, range_angle) hr, br = self._vital_signs(range_doppler) child_detected = self._detect_child(category, range_angle) return { 'presence': presence, 'category': category, 'heart_rate': hr, 'breathing_rate': br, 'child_detected': child_detected, } def _range_doppler_fft(self, cube): """距离-多普勒 FFT""" rd = np.fft.fft2(cube, axes=(1, 2)) rd = np.abs(rd).mean(axis=0) return rd def _angle_estimation(self, cube): """角度估计 (MUSIC/FFT)""" ra = np.fft.fft(cube, axis=0) return np.abs(ra).mean(axis=(1, 2)) def _detect_presence(self, rd): """存在检测""" if self.baseline is None: return np.max(rd) > 0.5 diff = rd - self.baseline return np.max(diff) > 0.3 def _classify_occupant(self, rd, ra): """乘员分类""" max_range_idx = np.argmax(np.sum(rd, axis=0)) range_val = max_range_idx * self.config.range_resolution rcs = np.max(rd) if rcs < 0.1: return 'empty' elif rcs < 0.3 and range_val < 0.5: return 'child' elif rcs < 0.5: return 'pet' else: return 'adult' def _vital_signs(self, rd): """生命体征 (微多普勒)""" phase = np.angle(rd) hr_fft = np.abs(np.fft.fft(phase, n=256)) hr_freq = np.argmax(hr_fft[8:40]) + 8 hr = hr_freq * 60 / 256 * 10 br_fft = np.abs(np.fft.fft(phase, n=256)) br_freq = np.argmax(br_fft[1:6]) + 1 br = br_freq * 60 / 256 * 10 self.hr_buffer.append(hr) self.br_buffer.append(br) if len(self.hr_buffer) > 30: self.hr_buffer.pop(0) self.br_buffer.pop(0) return ( np.median(self.hr_buffer) if self.hr_buffer else 0, np.median(self.br_buffer) if self.br_buffer else 0 ) def _detect_child(self, category, ra): """儿童遗弃检测""" if category == 'child': return True rear_energy = np.sum(ra[len(ra)//2:]) return rear_energy > 0.2 and category != 'adult'
class TeslaCPDSystem: """Tesla 儿童遗弃检测完整系统""" def __init__(self): self.radar = TeslaCabinRadar() self.alert_sent = False def monitor(self, radar_data, vehicle_state): """ 持续监控 Args: radar_data: 原始雷达帧 vehicle_state: {'locked': bool, 'engine_on': bool, 'interior_temp': float, 'exterior_temp': float} """ result = self.radar.process_frame(radar_data) if result['child_detected'] and not vehicle_state['engine_on']: temp = vehicle_state['interior_temp'] if temp > 35 or temp < -10: self._trigger_alert('critical', temp) else: self._trigger_alert('warning', temp) return result def _trigger_alert(self, level, temp): """触发警报""" if self.alert_sent and level == 'warning': return actions = { 'warning': [ 'App推送通知', '车内灯光闪烁', '空调激活(恒温22°C)', '鸣笛3次' ], 'critical': [ 'App紧急推送', '全部灯光闪烁', '空调最大制冷/加热', '降下车窗5cm', 'eCall紧急呼叫', '鸣笛持续' ] } for action in actions[level]: print(f"[CPD-{level}] {action}") if level == 'critical': self.alert_sent = True
if __name__ == "__main__": print("=" * 60) print("Tesla 车内雷达系统模拟") print("=" * 60) system = TeslaCabinRadar() for i in range(10): radar_cube = np.random.randn(4, 64, 128) * 0.1 radar_cube[:, :, 32] += 0.8 result = system.process_frame(radar_cube) print(f"存在检测: {result['presence']}") print(f"乘员分类: {result['category']}") print(f"心率: {result['heart_rate']:.1f} bpm") print(f"呼吸率: {result['breathing_rate']:.1f} /min") print(f"儿童检测: {result['child_detected']}") print("\n" + "=" * 60) print("CPD 场景: 锁车+引擎关闭+后排儿童+高温") print("=" * 60) cpd = TeslaCPDSystem() radar_cube = np.random.randn(4, 64, 128) * 0.1 radar_cube[:, :, 40] += 0.15 vehicle_state = { 'locked': True, 'engine_on': False, 'interior_temp': 42, 'exterior_temp': 38 } result = cpd.monitor(radar_cube, vehicle_state)
|