405. Lyft Motion Prediction for Autonomous Vehicles | lyft-motion-prediction-autonomous-vehicles
这是一个很不错的数据集,玩起来很有意思。它就像HPC(高性能计算)模拟一样,你必须放松心态,让程序长时间运行。令我惊讶的是,利用简单的概念和已发表的研究成果就能走得很远。我的方案基本上就是实现了Uber的两篇论文:
以下是我认为对我表现最好的模型有贡献的一些细节:
我只在间隔为 60+10+1 的帧上进行训练。我想避免数据泄露,所以我保留了一段间隔,长度为历史长度(10) + 未来预测长度(50) + 额外(1)。此外,如果一帧太拥挤(即智能体太多),我会对智能体进行子集采样(在我的下述代码中为20个)。
def get_dataset_masks(
zarr_path,
zarr_dt,
th_agent_prob: float,
min_frame_history: int,
min_frame_future: int,
scene_mask=None,
chop_data=False,
chop_idx_list=[100, 200],
chop_agents=False):
"""
Modified from create_chopped_dataset
Returns:
mask: numpy array of bool to pass to AgentDatasetExtended
"""
if scene_mask is None:
scene_mask = np.ones(len(zarr_dt.scenes), dtype=np.bool)
else:
assert len(zarr_dt.scenes) == len(scene_mask), "mask should be equal length"
agents_mask_path = Path(zarr_path) / f"agents_mask/{th_agent_prob}"
if not agents_mask_path.exists(): # don't check in root but check for the path
assert 0
agents_mask_original = np.asarray(convenience.load(str(agents_mask_path)))
agents_mask = np.zeros(len(zarr_dt.agents), dtype=np.bool)
# for scene in zarr_dt.scenes.get_mask_selection(scene_mask):
for scene_idx in range(len(zarr_dt.scenes)):
if scene_mask[scene_idx] == 0:
continue
scene = zarr_dt.scenes[scene_idx]
if chop_data:
for num_frame_to_copy in chop_idx_list:
kept_frame = zarr_dt.frames[scene["frame_index_interval"][0] + num_frame_to_copy - 1]
agents_slice = get_agents_slice_from_frames(kept_frame)
# In create_chopped_dataset: no mask for min_frame_history
mask = agents_mask_original[agents_slice][:, 1] >= min_frame_future
num_agents_per_frame = mask.sum()
max_agents_per_frame = 20
if chop_agents and num_agents_per_frame > max_agents_per_frame: # or 10
removed_indices = np.random.choice(
np.where(mask)[0],
num_agents_per_frame - max_agents_per_frame,
replace=False)
mask[removed_indices] = False
agents_mask[agents_slice] = mask.copy()
else:
first_frame = zarr_dt.frames[scene["frame_index_interval"][0]]
last_frame = zarr_dt.frames[scene["frame_index_interval"][1] - 1]
agents_slice = get_agents_slice_from_frames(first_frame, last_frame)
past_mask = agents_mask_original[agents_slice][:, 0] >= min_frame_history
future_mask = agents_mask_original[agents_slice][:, 1] >= min_frame_future
mask = past_mask * future_mask
agents_mask[agents_slice] = mask.copy()
return agents_mask
我对多轨迹预测(MTP)损失有一个清晰的实现。我在 MoN 损失上进行训练,并在 nll 损失上进行验证。
def uber_like_loss_new(gt, pred, confidences, avails):
batch_size, num_modes, future_len, num_coords = pred.shape
# ensure that your model outputs logits
gt = gt[:, None, :, :] # add modes
avails = avails[:, None, :, None] # add modes and cords
l2_error = torch.sum(((gt - pred) * avails) **