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
| import torch
import torch.nn as nn
import torch.nn.functional as F
class CCGF(nn.Module):
"""
Causal Context-Gated Forecaster (CCGF)
论文: arXiv:2609.12374
功能: 在追踪丢失期间因果预测驾驶员注视方向
输入:
- 60帧历史注视+头部姿态
- DINOv3场景特征（Live或Frozen模式）
输出:
- 预测注视坐标 (2D)
核心创新:
1. 可靠性门控：随丢失时间增长，动态调整历史vs场景权重
2. DINOv3场景特征：自监督场景理解
3. 因果预测：仅用过去数据，不依赖未来
"""
def __init__(self, gaze_dim=2, head_dim=6,
hist_feat_dim=256, scene_feat_dim=768,
num_heads=8, num_layers=3):
super().__init__()
self.history_encoder = nn.LSTM(
input_size=gaze_dim + head_dim,
hidden_size=hist_feat_dim,
num_layers=2,
batch_first=True,
bidirectional=False
)
self.scene_proj = nn.Sequential(
nn.Linear(scene_feat_dim, hist_feat_dim),
nn.GELU(),
nn.Linear(hist_feat_dim, hist_feat_dim)
)
self.reliability_gate = ReliabilityGate(
hist_dim=hist_feat_dim,
scene_dim=hist_feat_dim
)
self.predictor = nn.Sequential(
nn.Linear(hist_feat_dim, hist_feat_dim),
nn.GELU(),
nn.Linear(hist_feat_dim, gaze_dim)
)
def forward(self, history_gaze_head, scene_feat, dropout_progress):
"""
Args:
history_gaze_head: (B, 60, gaze_dim+head_dim) 历史60帧
scene_feat: (B, scene_feat_dim) DINOv3场景特征
dropout_progress: (B,) 丢失进度 0-1 (0=刚丢失, 1=长时间丢失)
Returns:
pred_gaze: (B, gaze_dim) 预测注视坐标
gate_weights: (B, 2) 历史/场景权重
"""
_, hist_feat = self.history_encoder(history_gaze_head)
hist_feat = hist_feat[-1]
scene_feat = self.scene_proj(scene_feat)
fused_feat, gate_weights = self.reliability_gate(
hist_feat, scene_feat, dropout_progress
)
pred_gaze = self.predictor(fused_feat)
return pred_gaze, gate_weights
class ReliabilityGate(nn.Module):
"""
可靠性门控模块
核心思想：
- 丢失初期：历史信息可靠（短期记忆有效）
- 丢失后期：场景信息更可靠（长期记忆衰减）
"""
def __init__(self, hist_dim, scene_dim):
super().__init__()
self.hist_decay = nn.Sequential(
nn.Linear(1, 32),
nn.ReLU(),
nn.Linear(32, hist_dim),
nn.Sigmoid()
)
self.scene_growth = nn.Sequential(
nn.Linear(1, 32),
nn.ReLU(),
nn.Linear(32, scene_dim),
nn.Sigmoid()
)
self.fusion = nn.Sequential(
nn.Linear(hist_dim + scene_dim, hist_dim),
nn.GELU(),
nn.Linear(hist_dim, hist_dim)
)
def forward(self, hist_feat, scene_feat, dropout_progress):
"""
Args:
hist_feat: (B, hist_dim)
scene_feat: (B, scene_dim)
dropout_progress: (B,) 0-1
Returns:
fused: (B, hist_dim)
weights: (B, 2) [hist_weight, scene_weight]
"""
progress = dropout_progress.unsqueeze(-1)
hist_weight = self.hist_decay(progress)
hist_weighted = hist_feat * hist_weight
scene_weight = self.scene_growth(progress)
scene_weighted = scene_feat * scene_weight
fused = self.fusion(torch.cat([hist_weighted, scene_weighted], dim=-1))
weights = torch.stack([
hist_weight.mean(dim=-1),
scene_weight.mean(dim=-1)
], dim=-1)
return fused, weights
if __name__ == "__main__":
model = CCGF(gaze_dim=2, head_dim=6, hist_feat_dim=256, scene_feat_dim=768)
history = torch.randn(4, 60, 8)
scene = torch.randn(4, 768)
progress = torch.rand(4)
pred, gates = model(history, scene, progress)
print(f"History input: {history.shape}")
print(f"Scene feature: {scene.shape}")
print(f"Predicted gaze: {pred.shape}")
print(f"Gate weights [hist, scene]: {gates.shape}")
print(f"Parameters: {sum(p.numel() for p in model.parameters()):,}")
progress_long = torch.ones(4)
_, gates_long = model(history, scene, progress_long)
progress_short = torch.zeros(4)
_, gates_short = model(history, scene, progress_short)
print(f"\n刚丢失 - 历史权重: {gates_short[:, 0].mean():.3f}")
print(f"长时间 - 历史权重: {gates_long[:, 0].mean():.3f}")
print("✅ 历史权重应随时间下降")
|
