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
| import numpy as np from dataclasses import dataclass, field from typing import List, Dict, Any, Tuple, Optional from enum import Enum import time
class PostureType(Enum): """姿态类型""" NORMAL = "normal" FEET_ON_DASHBOARD = "feet_on_dashboard" UPPER_BODY_FORWARD = "upper_body_forward" RECLINED = "reclined" CHILD_SEAT_REAR_FACING = "child_seat_rear_facing" CHILD_SEAT_FORWARD_FACING = "child_seat_forward_facing"
class RiskLevel(Enum): """风险等级""" LOW = "low" MEDIUM = "medium" HIGH = "high" CRITICAL = "critical"
@dataclass class PostureState: """姿态状态""" timestamp: float posture_type: PostureType risk_level: RiskLevel distance_to_dashboard: float feet_detected: bool = False upper_body_lean: float = 0.0 confidence: float = 1.0
@dataclass class SensorData: """传感器数据""" timestamp: float camera_depth: Optional[np.ndarray] = None seat_pressure: Optional[np.ndarray] = None seatbelt_status: Optional[Dict[str, Any]] = None seat_position: Optional[Tuple[float, float]] = None
class OOPDetector: """OOP检测器""" def __init__(self, dashboard_distance_threshold: float = 20.0, feet_detection_threshold: float = 30.0, alert_cooldown: float = 900.0): self.dashboard_distance_threshold = dashboard_distance_threshold self.feet_detection_threshold = feet_detection_threshold self.alert_cooldown = alert_cooldown self.state_history: List[PostureState] = [] self.last_alert_time: float = 0 def process_sensor_data(self, sensor_data: SensorData) -> PostureState: """处理传感器数据""" posture_type = PostureType.NORMAL risk_level = RiskLevel.LOW distance_to_dashboard = 50.0 feet_detected = False upper_body_lean = 0.0 if sensor_data.camera_depth is not None: depth_analysis = self._analyze_depth(sensor_data.camera_depth) distance_to_dashboard = depth_analysis.get('min_distance', 50.0) feet_detected = depth_analysis.get('feet_detected', False) upper_body_lean = depth_analysis.get('upper_body_lean', 0.0) if sensor_data.seat_pressure is not None: pressure_analysis = self._analyze_pressure(sensor_data.seat_pressure) if feet_detected: posture_type = PostureType.FEET_ON_DASHBOARD risk_level = RiskLevel.HIGH elif distance_to_dashboard < self.dashboard_distance_threshold: posture_type = PostureType.UPPER_BODY_FORWARD risk_level = RiskLevel.HIGH elif sensor_data.seatbelt_status and not sensor_data.seatbelt_status.get('buckled', True): if sensor_data.seat_position: posture_type = PostureType.CHILD_SEAT_REAR_FACING risk_level = RiskLevel.MEDIUM state = PostureState( timestamp=sensor_data.timestamp, posture_type=posture_type, risk_level=risk_level, distance_to_dashboard=distance_to_dashboard, feet_detected=feet_detected, upper_body_lean=upper_body_lean ) self.state_history.append(state) return state def _analyze_depth(self, depth_image: np.ndarray) -> Dict[str, Any]: """分析深度图像""" min_distance = np.min(depth_image[depth_image > 0]) if depth_image.any() else 50.0 feet_detected = False if depth_image.shape[0] > 0: upper_region = depth_image[:int(depth_image.shape[0]/3), :] if np.any(upper_region < self.feet_detection_threshold): feet_detected = True upper_body_lean = 0.0 return { 'min_distance': min_distance, 'feet_detected': feet_detected, 'upper_body_lean': upper_body_lean } def _analyze_pressure(self, pressure_matrix: np.ndarray) -> Dict[str, Any]: """分析压力矩阵""" total_pressure = np.sum(pressure_matrix) if total_pressure == 0: return {'center': (0, 0)} y_coords, x_coords = np.mgrid[0:pressure_matrix.shape[0], 0:pressure_matrix.shape[1]] center_x = np.sum(x_coords * pressure_matrix) / total_pressure center_y = np.sum(y_coords * pressure_matrix) / total_pressure return { 'center': (center_x, center_y), 'total_pressure': total_pressure } def should_alert(self, state: PostureState) -> bool: """判断是否需要报警""" if state.risk_level not in [RiskLevel.HIGH, RiskLevel.CRITICAL]: return False current_time = state.timestamp if current_time - self.last_alert_time < self.alert_cooldown: return False return True def generate_alert(self, state: PostureState) -> Dict[str, Any]: """生成报警信息""" if not self.should_alert(state): return None self.last_alert_time = state.timestamp alert = { 'timestamp': state.timestamp, 'type': 'oop_detected', 'posture_type': state.posture_type.value, 'risk_level': state.risk_level.value, 'message': self._get_alert_message(state.posture_type), 'action': 'visual_and_audible' } return alert def _get_alert_message(self, posture_type: PostureType) -> str: """获取报警消息""" messages = { PostureType.FEET_ON_DASHBOARD: "请将脚放回地面,仪表板附近有安全气囊", PostureType.UPPER_BODY_FORWARD: "请调整坐姿,保持与仪表板20cm以上距离", PostureType.RECLINED: "请调整座椅靠背角度,确保安全带正确贴合", PostureType.CHILD_SEAT_REAR_FACING: "检测到后向式儿童座椅,请确认安全气囊已关闭" } return messages.get(posture_type, "请调整坐姿")
def test_oop_detector(): """测试OOP检测器""" detector = OOPDetector() print("测试正常姿态:") normal_depth = np.full((480, 640), 50.0) sensor_data = SensorData( timestamp=time.time(), camera_depth=normal_depth, seat_pressure=np.random.rand(20, 20) * 100 ) state = detector.process_sensor_data(sensor_data) print(f" 姿态: {state.posture_type.value}") print(f" 风险: {state.risk_level.value}") print("\n测试脚踩仪表板:") abnormal_depth = np.full((480, 640), 50.0) abnormal_depth[:160, :] = 15.0 sensor_data = SensorData( timestamp=time.time(), camera_depth=abnormal_depth ) state = detector.process_sensor_data(sensor_data) print(f" 姿态: {state.posture_type.value}") print(f" 风险: {state.risk_level.value}") alert = detector.generate_alert(state) if alert: print(f" 报警: {alert['message']}")
if __name__ == "__main__": test_oop_detector()
|