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
| class ContextAwareSuppression: """上下文感知抑制机制""" def __init__(self): self.suppression_rules = { 'low_speed': { 'condition': lambda ctx: ctx['speed'] < 20, 'suppress': ['DISTRACTION_WARNING'] }, 'reversing': { 'condition': lambda ctx: ctx['gear'] == 'R', 'suppress': ['DISTRACTION_WARNING', 'FATIGUE_WARNING'] }, 'adas_active': { 'condition': lambda ctx: ctx['adas_mode'] in ['L2', 'L3'], 'suppress': ['FATIGUE_WARNING'] }, 'turning': { 'condition': lambda ctx: ctx['steering_angle'] > 30, 'suppress': ['DISTRACTION_WARNING'], 'duration': 3 }, 'voice_interaction': { 'condition': lambda ctx: ctx['voice_active'], 'suppress': ['DISTRACTION_WARNING'] } } self.suppression_state = {} def update_context(self, context): """ 更新上下文 Args: context: { 'speed': float, 'gear': str, 'adas_mode': str, 'steering_angle': float, 'voice_active': bool, 'timestamp': float } """ for rule_name, rule in self.suppression_rules.items(): if rule['condition'](context): self.suppression_state[rule_name] = context['timestamp'] def should_suppress(self, warning_type, current_time): """ 判断是否抑制警告 Args: warning_type: 警告类型 current_time: 当前时间 Returns: suppress: 是否抑制 """ for rule_name, rule in self.suppression_rules.items(): if warning_type in rule['suppress']: if rule_name in self.suppression_state: start_time = self.suppression_state[rule_name] duration = rule.get('duration', float('inf')) if current_time - start_time < duration: return True return False
detector = DistractionDetector() suppressor = ContextAwareSuppression()
def process_frame(frame, vehicle_data): """处理单帧""" suppressor.update_context(vehicle_data) result = detector.detect(frame, vehicle_data['gaze']) if result['detected']: should_suppress = suppressor.should_suppress( 'DISTRACTION_WARNING', vehicle_data['timestamp'] ) if should_suppress: result['detected'] = False result['suppressed'] = True return result
|