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
| """ 视觉酒驾检测算法
基于瞳孔、眼动、面部表情等多模态特征 """
import numpy as np from typing import Dict, List
class VisualImpairmentDetector: """ 视觉酒驾检测器 综合分析瞳孔、眼动、面部表情特征 """ def __init__(self): self.pupil_diameter_normal = (2.5, 4.0) self.pupil_diameter_dilated = (5.0, 8.0) self.nystagmus_threshold = 0.5 self.ear_fluctuation_threshold = 0.15 self.weights = { 'pupil_dilation': 0.25, 'pupil_response': 0.20, 'nystagmus': 0.25, 'ear_fluctuation': 0.15, 'head_stability': 0.15 } def analyze(self, features: Dict) -> Dict: """ 分析视觉特征 Args: features: { 'pupil_diameter_left': float, 'pupil_diameter_right': float, 'pupil_response_time': float, 'gaze_velocity': List[float], 'ear_sequence': List[float], 'head_pose_sequence': List[tuple] } Returns: result: { 'impairment_score': float, # 0-1 'is_impaired': bool, 'indicators': dict } """ scores = {} pupil_avg = (features['pupil_diameter_left'] + features['pupil_diameter_right']) / 2 if self.pupil_diameter_dilated[0] <= pupil_avg <= self.pupil_diameter_dilated[1]: scores['pupil_dilation'] = 1.0 elif pupil_avg > self.pupil_diameter_normal[1]: scores['pupil_dilation'] = (pupil_avg - self.pupil_diameter_normal[1]) / 2.0 else: scores['pupil_dilation'] = 0.0 response_time = features['pupil_response_time'] if response_time > 0.8: scores['pupil_response'] = 1.0 elif response_time > 0.5: scores['pupil_response'] = (response_time - 0.5) / 0.3 else: scores['pupil_response'] = 0.0 gaze_velocity = np.array(features['gaze_velocity']) gaze_std = np.std(gaze_velocity) scores['nystagmus'] = min(1.0, gaze_std / self.nystagmus_threshold) ear_sequence = np.array(features['ear_sequence']) ear_fluctuation = np.std(ear_sequence) scores['ear_fluctuation'] = min(1.0, ear_fluctuation / self.ear_fluctuation_threshold) head_pose = np.array(features['head_pose_sequence']) head_std = np.std(head_pose, axis=0).mean() scores['head_stability'] = min(1.0, head_std / 5.0) impairment_score = sum( scores[k] * self.weights[k] for k in scores ) return { 'impairment_score': impairment_score, 'is_impaired': impairment_score > 0.6, 'indicators': scores }
if __name__ == "__main__": detector = VisualImpairmentDetector() normal_features = { 'pupil_diameter_left': 3.5, 'pupil_diameter_right': 3.4, 'pupil_response_time': 0.3, 'gaze_velocity': [0.1, 0.2, 0.1, 0.15, 0.1], 'ear_sequence': [0.25, 0.26, 0.25, 0.24, 0.25], 'head_pose_sequence': [(0, 0, 0), (1, 0, 0), (0, 1, 0), (0, 0, 1), (0, 0, 0)] } impaired_features = { 'pupil_diameter_left': 5.5, 'pupil_diameter_right': 5.8, 'pupil_response_time': 0.9, 'gaze_velocity': [0.8, 0.5, 1.2, 0.3, 0.9], 'ear_sequence': [0.20, 0.30, 0.22, 0.35, 0.18], 'head_pose_sequence': [(2, 3, 1), (5, 2, 3), (3, 6, 2), (4, 1, 5), (2, 4, 3)] } print("=== 正常驾驶员 ===") result_normal = detector.analyze(normal_features) print(f"损伤评分: {result_normal['impairment_score']:.2f}") print(f"判断: {'受损' if result_normal['is_impaired'] else '正常'}") print(f"指标: {result_normal['indicators']}") print("\n=== 酒精影响驾驶员 ===") result_impaired = detector.analyze(impaired_features) print(f"损伤评分: {result_impaired['impairment_score']:.2f}") print(f"判断: {'受损' if result_impaired['is_impaired'] else '正常'}") print(f"指标: {result_impaired['indicators']}")
|