ByteTrack
1. ByteTrack算法步骤:
- 目标检测:ByteTrack算法首先会对视频数据进行目标检测。
- 检测结果划分:根据得分(score)对检测结果进行划分。通过设定一个阈值,将检测结果分为高分(high score)和低分(low score)两个部分。高分部分通常对应着较为准确的检测结果,而低分部分可能包含一些误检或检测不准的目标。
- 初次匹配:将高分中的检测结果与已有的跟踪轨迹进行匹配追踪。这一步通常使用如SORT等算法进行。在这个过程中,会有部分轨迹成功匹配到检测结果,但也有部分轨迹可能没有被匹配成功。
- 再次匹配:对于初次匹配中未能成功匹配的轨迹,ByteTrack算法会将其与低分中的检测结果进行再次匹配。这样做的目的是为了尽可能利用所有的检测结果,提高跟踪的准确性和鲁棒性。
- 新建和保留轨迹:对于未能匹配到任何检测结果的跟踪轨迹,ByteTrack算法会保留这些轨迹m帧(m是一个预设的帧数),等待这些轨迹再次出现时再进行匹配。对于没有匹配上跟踪轨迹但得分又足够高的检测框,ByteTrack算法会新建一个跟踪轨迹。
2. 算法解释
2.1 模型初始化
def __init__(self, args, frame_rate=30):
# 初始化一个空列表,用于存储正在被跟踪的轨迹
self.tracked_stracks = [] # type: list[STrack]
# 初始化一个空列表,用于存储丢失的轨迹(即当前帧中未检测到的轨迹)
self.lost_stracks = [] # type: list[STrack]
# 初始化一个空列表,用于存储已经从跟踪列表中移除的轨迹
self.removed_stracks = [] # type: list[STrack]
# 当前处理的帧的ID,初始化为0
self.frame_id = 0
# 存储算法的参数
self.args = args
#self.det_thresh = args.track_thresh
# 初始化检测阈值,这个值用于判断检测到的物体是否应该被追踪
# 初始值设置为args中的track_thresh值加上0.1
self.det_thresh = args.track_thresh + 0.1
# 根据帧率和args中的track_buffer来计算buffer的大小
self.buffer_size = int(frame_rate / 30.0 * args.track_buffer)
# 最大时间丢失阈值,用于判断一个轨迹何时应该被视为丢失
self.max_time_lost = self.buffer_size
# 初始化Kalman滤波器,用于预测物体的位置
self.kalman_filter = KalmanFilter()
2.2 模型更新算法流程
2.2.1 检测结果划分,划分为高分和较低分段
self.frame_id += 1
# 初始化几个列表,用于存储不同类型的轨迹
activated_starcks = []
refind_stracks = []
lost_stracks = []
removed_stracks = []
# 根据output_results的形状决定如何处理检测结果
if output_results.shape[1] == 5:
# 如果输出结果的列数是5,那么只包含边界框和分数
scores = output_results[:, 4] # 获取分数列
bboxes = output_results[:, :4] # 获取边界框列
else:
# 否则,需要将output_results从tensor转换为numpy数组
output_results = output_results.cpu().numpy()
scores = output_results[:, 4] # 获取分数列
# 这一行被注释掉了,如果取消注释,则会将分数与另一列相乘
# scores = output_results[:, 4] * output_results[:, 5]
bboxes = output_results[:, :4] # 获取边界框列,格式为x1y1x2y2(左上角和右下角坐标)
# 以下是关于边界框缩放的代码
# scale = min(img_size[0] / float(img_h), img_size[1] / float(img_w))
# bboxes /= scale
# 过滤掉分数低于跟踪阈值的检测结果
remain_inds = scores > self.args.track_thresh
# 分割出分数较高的和较低的检测结果
inds_low = scores > 0.1
inds_high = scores < self.args.track_thresh
# 找到同时满足上述两个条件的索引
inds_second = np.logical_and(inds_low, inds_high)
# 获取对应的边界框和分数
dets = bboxes[remain_inds]
scores_keep = scores[remain_inds]
dets_second = bboxes[inds_second]
scores_second = scores[inds_second]
2.2.2 高分段处理手段
if len(dets) > 0:
'''Detections'''
# 创建新的轨迹对象,并添加到detections列表中
detections = [STrack(STrack.tlbr_to_tlwh(tlbr), s) for
(tlbr, s) in zip(dets, scores_keep)]
else:
# 如果没有符合条件的检测结果,则detections列表为空
detections = []
''' Add newly detected tracklets to tracked_stracks'''
# 将未激活的轨迹添加到unconfirmed列表中,已激活的轨迹添加到tracked_stracks列表中
unconfirmed = []
tracked_stracks = [] # type: list[STrack]
for track in self.tracked_stracks:
if not track.is_activated:
unconfirmed.append(track)
else:
tracked_stracks.append(track)
2.2.3 最优匹配与未匹配划分
1.通过计算IoU距离来评估轨迹与检测框之间的匹配程度,并使用匈牙利算法来找到最优匹配。
2.匹配成功后,轨迹会根据新的检测框信息进行更新,或者被重新激活。
3.未匹配的轨迹和检测框则分别存储在u_track和u_detection中,
轨迹和检测框可能需要在后续步骤中进行进一步处理,例如将长时间未匹配的轨迹标记为丢失或移除。
# 第二步:首先与分数较高的检测结果进行关联将当前正在跟踪的轨迹和已丢失的轨迹合并成一个列表,准备进行匹配
strack_pool = joint_stracks(tracked_stracks, self.lost_stracks)
# Predict the current location with KF
# 使用Kalman滤波器预测strack_pool中每个轨迹在当前帧的位置
STrack.multi_predict(strack_pool)
# 计算strack_pool中每个轨迹的预测位置与当前帧检测框之间的IoU距离
dists = matching.iou_distance(strack_pool, detections)
# 如果不是使用MOT20格式的数据集,则根据检测框的分数调整IoU距离
if not self.args.mot20:
dists = matching.fuse_score(dists, detections)
# 使用线性分配算法(也称为匈牙利算法)进行轨迹与检测框的匹配
# 返回匹配成功的轨迹和检测框索引对,未匹配的轨迹索引和未匹配的检测框索引
matches, u_track, u_detection = matching.linear_assignment(dists, thresh=self.args.match_thresh)
# 遍历匹配结果
for itracked, idet in matches:
# 获取匹配的轨迹和检测框
track = strack_pool[itracked]
det = detections[idet]
if track.state == TrackState.Tracked:
# 将激活的轨迹添加到activated_starcks列表中
track.update(detections[idet], self.frame_id)
activated_starcks.append(track)
else:
# 如果轨迹是处于丢失状态(Lost) 重新激活轨迹,使用新的检测框信息,将重新找到的轨迹添加到refind_stracks列表中
track.re_activate(det, self.frame_id, new_id=False)
refind_stracks.append(track)
2.2.4 低分框再匹配
1.如果存在分数较低的检测框,将它们转换成STrack对象列表。
1.1 从未匹配的轨迹中筛选出状态为Tracked的轨迹。
1.2 计算这些轨迹与分数较低的检测框之间的IoU距离。
1.3使用线性分配算法进行二次匹配,阈值设为0.
匹配成功的轨迹根据检测框信息更新或重新激活。
对于剩余未匹配的轨迹,如果它们不是已标记为Lost的状态,则将它们标记为Lost并添加到lost_stracks列表中。
# 第三步:使用分数较低的检测框进行二次关联
# 如果存在分数较低的检测框
if len(dets_second) > 0:
# 创建一个新的轨迹列表,用于存储分数较低的检测框
'''Detections'''
detections_second = [STrack(STrack.tlbr_to_tlwh(tlbr), s) for (tlbr, s) in zip(dets_second, scores_second)]
else:
# 如果没有分数较低的检测框,则创建一个空列表
detections_second = []
# 从未匹配的轨迹中筛选出状态为Tracked的轨迹
r_tracked_stracks = [strack_pool[i] for i in u_track if strack_pool[i].state == TrackState.Tracked]
# 计算筛选后的轨迹与分数较低的检测框之间的IoU距离
dists = matching.iou_distance(r_tracked_stracks, detections_second)
# 使用线性分配算法进行二次匹配,匹配阈值设为0.5
matches, u_track, u_detection_second = matching.linear_assignment(dists, thresh=0.5)
# 遍历匹配结果
for itracked, idet in matches:
# 获取匹配的轨迹和检测框
track = r_tracked_stracks[itracked]
det = detections_second[idet]
# 如果轨迹是处于跟踪状态(Tracked)
if track.state == TrackState.Tracked:
# 使用检测框的信息更新轨迹
track.update(det, self.frame_id)
# 将激活的轨迹添加到activated_starcks列表中
activated_starcks.append(track)
# 如果轨迹不是处于跟踪状态(可能是Lost)
else:
# 重新激活轨迹,使用新的检测框信息
track.re_activate(det, self.frame_id, new_id=False)
# 将重新找到的轨迹添加到refind_stracks列表中
refind_stracks.append(track)
# 遍历剩余未匹配的轨迹
for it in u_track:
# 获取轨迹
track = r_tracked_stracks[it]
# 如果轨迹不是已标记为Lost的状态
if not track.state == TrackState.Lost:
# 标记轨迹为Lost
track.mark_lost()
# 将Lost状态的轨迹添加到lost_stracks列表中
lost_stracks.append(track)
2.2.5 未确认轨迹处理
1. 从所有检测框中筛选出与未确认轨迹相关的检测框。
2. 计算未确认轨迹与这些检测框之间的IoU距离。
3. 如果不是使用MOT20标准,则将轨迹的分数与IoU距离融合,以得到一个综合的匹配距离。
4. 使用线性分配算法进行匹配,阈值设为0.7。
5. 对于匹配成功的未确认轨迹,使用检测框的信息进行更新,并将其标记为已激活,添加到activated_starcks列表中。
6. 对于剩余的未确认轨迹,即未能与任何检测框匹配的轨迹,将其标记为已移除,并添加到removed_stracks列表中。
# 处理未确认的轨迹,这些轨迹通常只有起始帧
'''Deal with unconfirmed tracks, usually tracks with only one beginning frame'''
# 从所有检测框中筛选出与未确认轨迹相关的检测框
detections = [detections[i] for i in u_detection]
# 计算未确认轨迹与筛选后的检测框之间的IoU距离
dists = matching.iou_distance(unconfirmed, detections)
# 如果不是使用MOT20标准,则将轨迹的分数与IoU距离融合
if not self.args.mot20:
dists = matching.fuse_score(dists, detections)
# 使用线性分配算法进行匹配,匹配阈值设为0.7
matches, u_unconfirmed, u_detection = matching.linear_assignment(dists, thresh=0.7)
# 遍历匹配结果
for itracked, idet in matches:
# 获取匹配的未确认轨迹和检测框
track = unconfirmed[itracked]
det = detections[idet]
# 使用检测框的信息更新未确认轨迹
track.update(det, self.frame_id)
# 将已激活的轨迹添加到activated_starcks列表中
activated_starcks.append(track)
# 遍历剩余的未确认轨迹
for it in u_unconfirmed:
# 获取轨迹
track = unconfirmed[it]
# 标记轨迹为已移除
track.mark_removed()
# 将已移除的轨迹添加到removed_stracks列表中
removed_stracks.append(track)
2.2.6 更新状态
'''
检查并移除那些超过最大丢失时间阈值的丢失轨迹。
更新已跟踪的轨迹列表,只保留状态为“Tracked”的轨迹。
将新激活的轨迹和重新找到的轨迹添加到已跟踪的轨迹列表中。
更新丢失的轨迹列表,移除那些已经被跟踪的轨迹,并添加新标记为丢失的轨迹。
从丢失的轨迹列表中移除那些已经被移除的轨迹。
将已移除的轨迹添加到移除的轨迹列表中。
移除已跟踪和丢失轨迹列表中的重复轨迹。
返回所有已激活的跟踪轨迹列表。
'''
# 遍历所有标记为丢失的轨迹
for track in self.lost_stracks:
# 如果轨迹丢失的时间超过了最大允许丢失时间
if self.frame_id - track.end_frame > self.max_time_lost:
# 标记轨迹为已移除
track.mark_removed()
# 将移除的轨迹添加到removed_stracks列表中
removed_stracks.append(track)
# 更新已跟踪的轨迹列表,只保留状态为Tracked的轨迹
self.tracked_stracks = [t for t in self.tracked_stracks if t.state == TrackState.Tracked]
# 将新激活的轨迹(activated_starcks)添加到已跟踪的轨迹列表中
self.tracked_stracks = joint_stracks(self.tracked_stracks, activated_starcks)
# 将重新找到的轨迹(refind_stracks)也添加到已跟踪的轨迹列表中
self.tracked_stracks = joint_stracks(self.tracked_stracks, refind_stracks)
# 从丢失的轨迹列表中移除那些已经被跟踪的轨迹
self.lost_stracks = sub_stracks(self.lost_stracks, self.tracked_stracks)
# 将新标记为丢失的轨迹(lost_stracks)添加到丢失的轨迹列表中
self.lost_stracks.extend(lost_stracks)
# 从丢失的轨迹列表中移除那些已经被移除的轨迹
self.lost_stracks = sub_stracks(self.lost_stracks, self.removed_stracks)
# 将已移除的轨迹(removed_stracks)添加到移除的轨迹列表中
self.removed_stracks.extend(removed_stracks)
# 移除已跟踪和丢失轨迹列表中的重复轨迹
self.tracked_stracks, self.lost_stracks = remove_duplicate_stracks(self.tracked_stracks, self.lost_stracks)
# 获取所有已激活的跟踪轨迹
output_stracks = [track for track in self.tracked_stracks if track.is_activated]
# 返回已激活的跟踪轨迹列表
return output_stracks
2.3 匈牙利匹配算法(线性分配)
在 cost, x, y = lap.lapjv(cost_matrix, extend_cost=True, cost_limit=thresh)
这行代码中,lapjv
函数来自 SciPy 的 scipy.optimize
模块,用于解决线性分配问题(也称为匈牙利算法)。这个函数会找到一种最优的配对方式,使得两组元素之间的配对总成本最小。
函数的参数解释如下:
cost_matrix
: 这是一个二维数组(或矩阵),表示不同元素配对之间的成本。cost_matrix[i][j]
表示将第一组中的第i
个元素与第二组中的第j
个元素配对的成本。extend_cost
: 这是一个布尔值,当设置为True
时,lapjv
函数会扩展成本矩阵,使其成为一个方阵(即行数和列数相等),并在扩展的部分填充一个足够大的值,以确保原始成本矩阵中的配对是最优的。这对于非方阵的成本矩阵是必要的,因为匈牙利算法要求成本矩阵是方阵。cost_limit
: 这是一个阈值,用于限制考虑的配对成本。只有当配对的成本小于或等于此阈值时,该配对才会被考虑在内。这有助于排除那些成本过高的配对,从而加速计算过程或找到满足特定条件的解。
函数的返回值解释如下:
cost
: 这是一个标量值,表示找到的最优配对方案的总成本。x
: 这是一个整数数组,表示第一组元素与第二组元素的配对关系。如果x[i]
的值是j
(且j
是非负的),那么表示第一组中的第i
个元素与第二组中的第j
个元素被配对。如果x[i]
的值是负数,那么表示第一组中的第i
个元素没有被配对。y
: 这是一个整数数组,与x
类似,表示第二组元素与第一组元素的配对关系。如果y[j]
的值是i
(且i
是非负的),那么表示第二组中的第j
个元素与第一组中的第i
个元素被配对。如果y[j]
的值是负数,那么表示第二组中的第j
个元素没有被配对。
def linear_assignment(cost_matrix, thresh):
'''
线性分配,匈牙利算法
Args:
cost_matrix:
thresh:
Returns:
'''
if cost_matrix.size == 0:
return np.empty((0, 2), dtype=int), tuple(range(cost_matrix.shape[0])), tuple(range(cost_matrix.shape[1]))
matches, unmatched_a, unmatched_b = [], [], []
cost, x, y = lap.lapjv(cost_matrix, extend_cost=True, cost_limit=thresh)
for ix, mx in enumerate(x):
if mx >= 0:
matches.append([ix, mx])
unmatched_a = np.where(x < 0)[0]
unmatched_b = np.where(y < 0)[0]
matches = np.asarray(matches)
return matches, unmatched_a, unmatched_b
, [], []
cost, x, y = lap.lapjv(cost_matrix, extend_cost=True, cost_limit=thresh)
for ix, mx in enumerate(x):
if mx >= 0:
matches.append([ix, mx])
unmatched_a = np.where(x < 0)[0]
unmatched_b = np.where(y < 0)[0]
matches = np.asarray(matches)
return matches, unmatched_a, unmatched_b