deepsort源码解读(七)

deepsort项目地址

---

目录
deepsort原理及解析汇总

---

tracker.py

# vim: expandtab:ts=4:sw=4
from __future__ import absolute_import
import numpy as np
from . import kalman_filter
from . import linear_assignment
from . import iou_matching
from .track import Track


class Tracker:
    """
    This is the multi-target tracker.

    Parameters
    ----------
    metric : nn_matching.NearestNeighborDistanceMetric
        A distance metric for measurement-to-track association.
    max_age : int
        Maximum number of missed misses before a track is deleted.
    n_init : int
        Number of consecutive detections before the track is confirmed. The
        track state is set to `Deleted` if a miss occurs within the first
        `n_init` frames.

    Attributes
    ----------
    metric : nn_matching.NearestNeighborDistanceMetric
        The distance metric used for measurement to track association.
    max_age : int
        Maximum number of missed misses before a track is deleted.
    n_init : int
        Number of frames that a track remains in initialization phase.
    kf : kalman_filter.KalmanFilter
        A Kalman filter to filter target trajectories in image space.
    tracks : List[Track]
        The list of active tracks at the current time step.

    """

    def __init__(self, metric, max_iou_distance=0.7, max_age=30, n_init=3):
        # metric是一个类，用于计算距离(余弦距离或马氏距离)
        self.metric = metric
        # iou匹配时的最大值
        self.max_iou_distance = max_iou_distance
        # 未被匹配的track最大失配次数，也即直接指定级联匹配的cascade_depth参数
        self.max_age = max_age
        # 从unconfirmed track转为confirmed track的最小匹配次数
        self.n_init = n_init

        # 卡尔曼滤波器
        self.kf = kalman_filter.KalmanFilter()
        # 保存一系列轨迹
        self.tracks = []
        # 下一个分配的轨迹id
        self._next_id = 1

    def predict(self):
        """Propagate track state distributions one time step forward.

        This function should be called once every time step, before `update`.
        """
        # 第一帧，tracks为空，不执行预测
        for track in self.tracks:
            track.predict(self.kf)

    def 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.

        """
        # Run matching cascade.
        matches, unmatched_tracks, unmatched_detections = \
            self._match(detections)

        # Update track set.
        # 对于匹配上的结果，更新track
        for track_idx, detection_idx in matches:
            self.tracks[track_idx].update(
                self.kf, detections[detection_idx])
        # 对于未匹配的track，调用mark_missed进行标记
        for track_idx in unmatched_tracks:
            self.tracks[track_idx].mark_missed()
        # 对于未匹配的detection,建立新的track并进行初始化，放入track列表
        for detection_idx in unmatched_detections:
            self._initiate_track(detections[detection_idx])
        # 遍历得到最新的tracks列表，保存的是confirmed与unconfirmed状态的tracks
        # 包含detection新建的track
        self.tracks = [t for t in self.tracks if not t.is_deleted()]

        # Update distance metric.
        # 获取所有confirmed状态的track id
        active_targets = [t.track_id for t in self.tracks if t.is_confirmed()]
        features, targets = [], []
        for track in self.tracks:
            if not track.is_confirmed():
                continue
            features += track.features
            targets += [track.track_id for _ in track.features]
            track.features = []
        self.metric.partial_fit(
            np.asarray(features), np.asarray(targets), active_targets)

    # 首先对基于外观信息的马氏距离计算tracks和detections的代价矩阵，
    # 然后相继进行级联匹配和IOU匹配，最后得到当前帧的所有匹配对、未匹配的tracks以及未匹配的detections
    def _match(self, detections):

        # track为所有track
        # dets为detections
        # track_indices为
        def gated_metric(tracks, dets, track_indices, detection_indices):
            # 所有检测对象的特征
            features = np.array([dets[i].feature for i in detection_indices])
            # 跟踪对象的id
            targets = np.array([tracks[i].track_id for i in track_indices])
            # 这里返回的是使用余弦距离计算出来的代价矩阵
            cost_matrix = self.metric.distance(features, targets)
            # gate_cost_matrix，计算马氏距离
            cost_matrix = linear_assignment.gate_cost_matrix(
                self.kf, cost_matrix, tracks, dets, track_indices,
                detection_indices)
            # 返回经过余弦距离与马氏距离处理后的代价矩阵
            return cost_matrix

        # Split track set into confirmed and 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()]

        # Associate confirmed tracks using appearance features.
        # 进行级联匹配（特别的，对于track为空时，不执行级联匹配操作）
        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)

        # 进行iou匹配计算
        # Associate remaining tracks together with unconfirmed tracks using IOU.
        # 只计算time_since_update等于1的track（特别的，对于第一帧也不做iou处理）
        iou_track_candidates = unconfirmed_tracks + [
            k for k in unmatched_tracks_a if
            self.tracks[k].time_since_update == 1]
        # 筛选出不符合iou匹配的track
        unmatched_tracks_a = [
            k for k in unmatched_tracks_a if
            self.tracks[k].time_since_update != 1]
        # 进行iou匹配
        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)

        matches = matches_a + matches_b
        unmatched_tracks = list(set(unmatched_tracks_a + unmatched_tracks_b))
        return matches, unmatched_tracks, unmatched_detections

    # 初始化第一帧
    def _initiate_track(self, detection):
        # 由检测目标detection的坐标（(center x, center y, aspect ratio, height)构建
        # 均值向量（1×8的向量，初始化为[center_x,center_y,w/h,h,0,0,0,0]与协方差矩阵(8*8矩阵，根据目标的高度构造的一个对角矩阵）
        mean, covariance = self.kf.initiate(detection.to_xyah())
        # Track()本身是一个类，这里把每个新目标用一个类来刻画，放入tracks列表
        self.tracks.append(Track(
            mean, covariance, self._next_id, self.n_init, self.max_age,
            detection.feature))
        self._next_id += 1