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
| """
论文：Detecting Phone-Induced Pedestrian Distraction via MFT
arXiv: 2609.23507
核心方法：多模态融合Transformer
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
class MultimodalFusionTransformer(nn.Module):
"""
MFT: 融合姿态关键点和视觉外观的分心检测Transformer
可迁移到驾驶员分心检测
"""
def __init__(self,
num_keypoints: int = 17,
keypoint_dim: int = 2,
visual_dim: int = 512,
hidden_dim: int = 256,
num_heads: int = 8,
num_layers: int = 4,
num_classes: int = 4):
super().__init__()
self.pose_encoder = nn.Sequential(
nn.Linear(num_keypoints * keypoint_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, hidden_dim)
)
self.visual_encoder = nn.Sequential(
nn.Linear(visual_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, hidden_dim)
)
self.cross_modal_attn = nn.MultiheadAttention(
embed_dim=hidden_dim,
num_heads=num_heads,
batch_first=True
)
self.norm1 = nn.LayerNorm(hidden_dim)
encoder_layer = nn.TransformerEncoderLayer(
d_model=hidden_dim,
nhead=num_heads,
dim_feedforward=hidden_dim * 4,
dropout=0.1,
batch_first=True
)
self.temporal_encoder = nn.TransformerEncoder(
encoder_layer, num_layers=num_layers
)
self.classifier = nn.Sequential(
nn.Linear(hidden_dim, hidden_dim // 2),
nn.ReLU(),
nn.Dropout(0.3),
nn.Linear(hidden_dim // 2, num_classes)
)
def forward(self, pose_seq, visual_seq):
"""
Args:
pose_seq: (B, T, num_keypoints, keypoint_dim) 姿态序列
visual_seq: (B, T, visual_dim) 视觉特征序列
Returns:
logits: (B, num_classes) 分心分类
"""
B, T, K, D = pose_seq.shape
pose_flat = pose_seq.reshape(B, T, K * D)
pose_features = self.pose_encoder(pose_flat)
visual_features = self.visual_encoder(visual_seq)
attn_out, _ = self.cross_modal_attn(
query=pose_features,
key=visual_features,
value=visual_features
)
fused = self.norm1(pose_features + attn_out)
temporal_out = self.temporal_encoder(fused)
last_output = temporal_out[:, -1, :]
logits = self.classifier(last_output)
return logits
class DriverDistractionMFT(nn.Module):
"""
将MFT从行人分心迁移到驾驶员分心检测
关键变更:
- 输入: 面部关键点 + 座舱图像特征 (替代行人关键点+街景)
- 输出: 驾驶员分心类型 (替代行人分心类型)
- 场景: 车内固定视角 (替代多视角)
"""
def __init__(self, num_classes=5):
super().__init__()
self.mft = MultimodalFusionTransformer(
num_keypoints=98,
keypoint_dim=2,
visual_dim=256,
hidden_dim=256,
num_heads=8,
num_layers=4,
num_classes=num_classes
)
def forward(self, face_keypoints_seq, dms_visual_seq):
"""
Args:
face_keypoints_seq: (B, T, 98, 2) 面部关键点序列
dms_visual_seq: (B, T, 256) DMS摄像头视觉特征
"""
return self.mft(face_keypoints_seq, dms_visual_seq)
if __name__ == "__main__":
B, T = 4, 10
pose_seq = torch.randn(B, T, 17, 2)
visual_seq = torch.randn(B, T, 512)
mft = MultimodalFusionTransformer(num_classes=4)
logits = mft(pose_seq, visual_seq)
print("=== 行人分心检测MFT ===")
print(f"输入姿态: {pose_seq.shape}")
print(f"输入视觉: {visual_seq.shape}")
print(f"输出: {logits.shape}")
face_kp = torch.randn(B, T, 98, 2)
dms_vis = torch.randn(B, T, 256)
driver_mft = DriverDistractionMFT(num_classes=5)
d_logits = driver_mft(face_kp, dms_vis)
print("\n=== 驾驶员分心检测迁移 ===")
print(f"输入面部关键点: {face_kp.shape}")
print(f"输入DMS特征: {dms_vis.shape}")
print(f"输出: {d_logits.shape}")
distraction_types = ["正常驾驶", "手机通话", "手机打字", "看屏幕", "注意力分散"]
print(f"\n分类类型: {distraction_types}")
print("\n=== 论文实验结果 ===")
print(f"整体准确率: 95%")
print(f"超越6个基线: +6%")
print(f"数据集: 287行人, 20,741图像")
print(f"跨模态注意力: 关键贡献模块")
|
