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
| import numpy as np
from typing import List, Tuple
class DrivingAngerDetector:
"""
驾驶员愤怒检测器
多模态融合方案:
1. 面部AU强度 → 愤怒表情评分
2. 语音特征 → 声学愤怒评分
3. 驾驶行为 → 行为愤怒评分
4. 上下文 → 环境压力评分
论文参考:
- Frontiers 2024: 运动活动+面部几何 96%准确率
- IEEE 2024: 上下文感知驾驶员愤怒识别
- arXiv 2025: VLM路怒推理数据集
"""
def __init__(self, config: dict = None):
config = config or {}
self.weights = config.get('weights', {
'facial': 0.35,
'voice': 0.25,
'behavior': 0.25,
'context': 0.15
})
self.anger_threshold = config.get('threshold', 0.6)
self.history = []
def detect(
self,
facial_au: np.ndarray,
voice_features: dict,
driving_metrics: dict,
context: dict
) -> dict:
"""多模态愤怒检测"""
anger_aus = [3, 14, 19, 22]
facial_score = np.mean([facial_au[i] for i in anger_aus])
voice_score = self._voice_anger_score(voice_features)
behavior_score = self._behavior_anger_score(driving_metrics)
context_score = self._context_score(context)
total = (
self.weights['facial'] * facial_score +
self.weights['voice'] * voice_score +
self.weights['behavior'] * behavior_score +
self.weights['context'] * context_score
)
self.history.append(total)
if len(self.history) > 90:
self.history.pop(0)
sustained = len([s for s in self.history if s > self.anger_threshold]) > 60
return {
'anger_score': total,
'is_angry': total > self.anger_threshold and sustained,
'components': {
'facial': facial_score,
'voice': voice_score,
'behavior': behavior_score,
'context': context_score
},
'regulation_trigger': total > 0.8 and sustained
}
def _voice_anger_score(self, features: dict) -> float:
pitch = features.get('pitch_hz', 150)
speech_rate = features.get('speech_rate', 2.5)
volume = features.get('volume_db', -20)
pitch_score = min((pitch - 120) / 100, 1.0)
rate_score = min((speech_rate - 2.0) / 3.0, 1.0)
vol_score = min((volume + 30) / 30, 1.0)
return np.mean([pitch_score, rate_score, vol_score])
def _behavior_anger_score(self, metrics: dict) -> float:
accel = metrics.get('acceleration', 0)
brake = metrics.get('braking', 0)
lane_changes = metrics.get('lane_changes_per_min', 0)
following_dist = metrics.get('following_distance_m', 10)
accel_score = min(abs(accel) / 3.0, 1.0)
brake_score = min(abs(brake) / 4.0, 1.0)
lane_score = min(lane_changes / 4.0, 1.0)
dist_score = min(max(0, (5 - following_dist) / 5), 1.0)
return np.mean([accel_score, brake_score, lane_score, dist_score])
def _context_score(self, ctx: dict) -> float:
traffic = ctx.get('traffic_density', 0.5)
was_cut_off = ctx.get('was_cut_off', False)
slow_lead = ctx.get('slow_lead_vehicle', False)
score = traffic * 0.3
if was_cut_off: score += 0.5
if slow_lead: score += 0.2
return min(score, 1.0)
def get_regulation_action(self, anger_result: dict) -> str:
"""根据愤怒程度选择调节策略"""
score = anger_result['anger_score']
if score > 0.8:
return "强烈建议：深呼吸、靠边停车休息"
elif score > 0.6:
return "温和提醒：播放舒缓音乐、建议调整呼吸"
elif score > 0.4:
return "隐性干预：调低音量、柔和导航语音"
else:
return "无需干预"
if __name__ == "__main__":
detector = DrivingAngerDetector()
normal = detector.detect(
facial_au=np.random.uniform(0.1, 0.3, 17),
voice_features={'pitch_hz': 120, 'speech_rate': 2.0, 'volume_db': -25},
driving_metrics={'acceleration': 0.5, 'braking': -0.3,
'lane_changes_per_min': 0.5, 'following_distance_m': 12},
context={'traffic_density': 0.3, 'was_cut_off': False, 'slow_lead_vehicle': False}
)
angry = detector.detect(
facial_au=np.random.uniform(0.6, 0.9, 17),
voice_features={'pitch_hz': 180, 'speech_rate': 4.0, 'volume_db': -10},
driving_metrics={'acceleration': 2.5, 'braking': -3.5,
'lane_changes_per_min': 3.0, 'following_distance_m': 3},
context={'traffic_density': 0.8, 'was_cut_off': True, 'slow_lead_vehicle': True}
)
print("=== 正常驾驶 ===")
print(f"愤怒评分: {normal['anger_score']:.3f}")
print(f"各维度: {normal['components']}")
print(f"调节建议: {detector.get_regulation_action(normal)}")
print("\n=== 愤怒驾驶 ===")
print(f"愤怒评分: {angry['anger_score']:.3f}")
print(f"各维度: {angry['components']}")
print(f"调节建议: {detector.get_regulation_action(angry)}")
print("\n=== 论文基准 ===")
print("Frontiers 2024: 运动活动+面部几何 → 96%准确率")
print("IEEE 2024: 上下文感知愤怒识别")
print("arXiv 2025: VLM路怒推理数据集")
|
