目標(biāo)跟蹤初探（DeepSORT）

import numpy as np from sklearn.utils.linear_assignment_ import linear_assignmentfrom scipy.optimize import linear_sum_assignment

cost_matrix = np.array([    [15,40,45],    [20,60,35],    [20,40,25]])
matches = linear_assignment(cost_matrix)print('sklearn API result:\n', matches)matches = linear_sum_assignment(cost_matrix)print('scipy API result:\n', matches)

"""Outputssklearn API result: [[0 1]  [1 0]  [2 2]]scipy API result: (array([0, 1, 2], dtype=int64), array([1, 0, 2], dtype=int64))"""

#  linear_assignment.pydef min_cost_matching(distance_metric, max_distance, tracks, detections,                       track_indices=None, detection_indices=None):    ...
    # 計(jì)算代價(jià)矩陣    cost_matrix = distance_metric(tracks, detections, track_indices, detection_indices)    cost_matrix[cost_matrix > max_distance] = max_distance + 1e-5
    # 執(zhí)行匈牙利算法，得到匹配成功的索引對(duì)，行索引為tracks的索引，列索引為detections的索引    row_indices, col_indices = linear_assignment(cost_matrix)
    matches, unmatched_tracks, unmatched_detections = [], [], []
    # 找出未匹配的detections    for col, detection_idx in enumerate(detection_indices):        if col not in col_indices:            unmatched_detections.append(detection_idx)
    # 找出未匹配的tracks    for row, track_idx in enumerate(track_indices):        if row not in row_indices:            unmatched_tracks.append(track_idx)
    # 遍歷匹配的(track, detection)索引對(duì)    for row, col in zip(row_indices, col_indices):        track_idx = track_indices[row]        detection_idx = detection_indices[col]        # 如果相應(yīng)的cost大于閾值max_distance，也視為未匹配成功        if cost_matrix[row, col] > max_distance:            unmatched_tracks.append(track_idx)            unmatched_detections.append(detection_idx)        else:            matches.append((track_idx, detection_idx))
    return matches, unmatched_tracks, unmatched_detections

#  kalman_filter.pydef predict(self, mean, covariance):    """Run Kalman filter prediction step.
    Parameters    ----------    mean: ndarray, the 8 dimensional mean vector of the object state at the previous time step.    covariance: ndarray, the 8x8 dimensional covariance matrix of the object state at the previous time step.
    Returns    -------    (ndarray, ndarray), the mean vector and covariance matrix of the predicted state.      Unobserved velocities are initialized to 0 mean.    """    std_pos = [        self._std_weight_position * mean[3],        self._std_weight_position * mean[3],        1e-2,        self._std_weight_position * mean[3]]    std_vel = [        self._std_weight_velocity * mean[3],        self._std_weight_velocity * mean[3],        1e-5,        self._std_weight_velocity * mean[3]]
    motion_cov = np.diag(np.square(np.r_[std_pos, std_vel]))  # 初始化噪聲矩陣Q    mean = np.dot(self._motion_mat, mean)  # x' = Fx    covariance = np.linalg.multi_dot((self._motion_mat, covariance, self._motion_mat.T)) + motion_cov  # P' = FPF(T) + Q
    return mean, covariance

#  demo_yolo3_deepsort.pydef detect(self):    while self.vdo.grab():  ...  bbox_xcycwh, cls_conf, cls_ids = self.yolo3(im)  # 檢測(cè)到的bbox[cx,cy,w,h]，置信度，類別id  if bbox_xcycwh is not None:          # 篩選出人的類別          mask = cls_ids == 0        bbox_xcycwh = bbox_xcycwh[mask]        bbox_xcycwh[:, 3:] *= 1.2         cls_conf = cls_conf[mask]            ...

#  deep_sort.pydef update(self, bbox_xywh, confidences, ori_img):    self.height, self.width = ori_img.shape[:2]    # 提取每個(gè)bbox的feature    features = self._get_features(bbox_xywh, ori_img)    # [cx,cy,w,h] -> [x1,y1,w,h]    bbox_tlwh = self._xywh_to_tlwh(bbox_xywh)    # 過濾掉置信度小于self.min_confidence的bbox，生成detections    detections = [Detection(bbox_tlwh[i], conf, features[i]) for i,conf in enumerate(confidences) if conf > self.min_confidence]    # NMS (這里self.nms_max_overlap的值為1，即保留了所有的detections)    boxes = np.array([d.tlwh for d in detections])    scores = np.array([d.confidence for d in detections])    indices = non_max_suppression(boxes, self.nms_max_overlap, scores)    detections = [detections[i] for i in indices]    ...   

#  track.pydef predict(self, kf):    """Propagate the state distribution to the current time step using a        Kalman filter prediction step.    Parameters    ----------    kf: The Kalman filter.    """    self.mean, self.covariance = kf.predict(self.mean, self.covariance)  # 預(yù)測(cè)    self.age += 1  # 該track自出現(xiàn)以來的總幀數(shù)加1    self.time_since_update += 1  # 該track自最近一次更新以來的總幀數(shù)加1

#  tracker.pydef _match(self, detections):    def gated_metric(racks, dets, track_indices, detection_indices):        """        基于外觀信息和馬氏距離，計(jì)算卡爾曼濾波預(yù)測(cè)的tracks和當(dāng)前時(shí)刻檢測(cè)到的detections的代價(jià)矩陣        """        features = np.array([dets[i].feature for i in detection_indices])        targets = np.array([tracks[i].track_id for i in track_indices]  # 基于外觀信息，計(jì)算tracks和detections的余弦距離代價(jià)矩陣        cost_matrix = self.metric.distance(features, targets)  # 基于馬氏距離，過濾掉代價(jià)矩陣中一些不合適的項(xiàng) (將其設(shè)置為一個(gè)較大的值)        cost_matrix = linear_assignment.gate_cost_matrix(self.kf, cost_matrix, tracks,                       dets, track_indices, detection_indices)        return cost_matrix
    # 區(qū)分開confirmed tracks和unconfirmed tracks    confirmed_tracks = [i for i, t in enumerate(self.tracks) if t.is_confirmed()]    unconfirmed_tracks = [i for i, t in enumerate(self.tracks) if not t.is_confirmed()]
    # 對(duì)confirmd tracks進(jìn)行級(jí)聯(lián)匹配    matches_a, unmatched_tracks_a, unmatched_detections = \        linear_assignment.matching_cascade(            gated_metric, self.metric.matching_threshold, self.max_age,            self.tracks, detections, confirmed_tracks)
    # 對(duì)級(jí)聯(lián)匹配中未匹配的tracks和unconfirmed tracks中time_since_update為1的tracks進(jìn)行IOU匹配    iou_track_candidates = unconfirmed_tracks + [k for k in unmatched_tracks_a if                                                 self.tracks[k].time_since_update == 1]    unmatched_tracks_a = [k for k in unmatched_tracks_a if                          self.tracks[k].time_since_update != 1]    matches_b, unmatched_tracks_b, unmatched_detections = \        linear_assignment.min_cost_matching(            iou_matching.iou_cost, self.max_iou_distance, self.tracks,            detections, iou_track_candidates, unmatched_detections)
    # 整合所有的匹配對(duì)和未匹配的tracks    matches = matches_a + matches_b    unmatched_tracks = list(set(unmatched_tracks_a + unmatched_tracks_b))
    return matches, unmatched_tracks, unmatched_detections

# 級(jí)聯(lián)匹配源碼  linear_assignment.pydef matching_cascade(distance_metric, max_distance, cascade_depth, tracks, detections,                      track_indices=None, detection_indices=None):    ...    unmatched_detections = detection_indice    matches = []    # 由小到大依次對(duì)每個(gè)level的tracks做匹配    for level in range(cascade_depth):  # 如果沒有detections，退出循環(huán)        if len(unmatched_detections) == 0:              break  # 當(dāng)前l(fā)evel的所有tracks索引        track_indices_l = [k for k in track_indices if                            tracks[k].time_since_update == 1 + level]  # 如果當(dāng)前l(fā)evel沒有track，繼續(xù)        if len(track_indices_l) == 0:             continue
  # 匈牙利匹配        matches_l, _, unmatched_detections = min_cost_matching(distance_metric, max_distance, tracks, detections,                                                                track_indices_l, unmatched_detections)
  matches += matches_l  unmatched_tracks = list(set(track_indices) - set(k for k, _ in matches))    return matches, unmatched_tracks, unmatched_detections

#  tracker.pydef update(self, detections):    """Perform measurement update and track management.    Parameters    ----------    detections: List[deep_sort.detection.Detection]                A list of detections at the current time step.    """    # 得到匹配對(duì)、未匹配的tracks、未匹配的dectections    matches, unmatched_tracks, unmatched_detections = self._match(detections)
    # 對(duì)于每個(gè)匹配成功的track，用其對(duì)應(yīng)的detection進(jìn)行更新    for track_idx, detection_idx in matches:        self.tracks[track_idx].update(self.kf, detections[detection_idx])
  # 對(duì)于未匹配的成功的track，將其標(biāo)記為丟失  for track_idx in unmatched_tracks:        self.tracks[track_idx].mark_missed()
    # 對(duì)于未匹配成功的detection，初始化為新的track    for detection_idx in unmatched_detections:        self._initiate_track(detections[detection_idx])
  ...