Bytetrack module

core

ByteTrack

Initialize the ByteTrack object.

Parameters:

Name	Type	Description	Default
track_activation_threshold	float	Detection confidence threshold for track activation. Increasing track_activation_threshold improves accuracy and stability but might miss true detections. Decreasing it increases completeness but risks introducing noise and instability.	0.25
lost_track_buffer	int	Number of frames to buffer when a track is lost. Increasing lost_track_buffer enhances occlusion handling, significantly reducing the likelihood of track fragmentation or disappearance caused by brief detection gaps.	30
minimum_matching_threshold	float	Threshold for matching tracks with detections. Increasing minimum_matching_threshold improves accuracy but risks fragmentation. Decreasing it improves completeness but risks false positives and drift.	0.8
frame_rate	int	The frame rate of the video.	30
minimum_consecutive_frames	int	Number of consecutive frames that an object must be tracked before it is considered a 'valid' track. Increasing minimum_consecutive_frames prevents the creation of accidental tracks from false detection or double detection, but risks missing shorter tracks.	1

Source code in supertracker/bytetrack/core.py

class ByteTrack:
    """
    Initialize the ByteTrack object.

    Parameters:
        track_activation_threshold (float): Detection confidence threshold
            for track activation. Increasing track_activation_threshold improves accuracy
            and stability but might miss true detections. Decreasing it increases
            completeness but risks introducing noise and instability.
        lost_track_buffer (int): Number of frames to buffer when a track is lost.
            Increasing lost_track_buffer enhances occlusion handling, significantly
            reducing the likelihood of track fragmentation or disappearance caused
            by brief detection gaps.
        minimum_matching_threshold (float): Threshold for matching tracks with detections.
            Increasing minimum_matching_threshold improves accuracy but risks fragmentation.
            Decreasing it improves completeness but risks false positives and drift.
        frame_rate (int): The frame rate of the video.
        minimum_consecutive_frames (int): Number of consecutive frames that an object must
            be tracked before it is considered a 'valid' track.
            Increasing minimum_consecutive_frames prevents the creation of accidental tracks from
            false detection or double detection, but risks missing shorter tracks.
    """ 

    def __init__(
        self,
        track_activation_threshold: float = 0.25,
        lost_track_buffer: int = 30,
        minimum_matching_threshold: float = 0.8,
        frame_rate: int = 30,
        minimum_consecutive_frames: int = 1,
    ):
        self.track_activation_threshold = track_activation_threshold
        self.minimum_matching_threshold = minimum_matching_threshold

        self.frame_id = 0
        self.det_thresh = self.track_activation_threshold + 0.1
        self.max_time_lost = int(frame_rate / 30.0 * lost_track_buffer)
        self.minimum_consecutive_frames = minimum_consecutive_frames
        self.kalman_filter = KalmanFilter()
        self.shared_kalman = KalmanFilter()

        self.tracked_tracks: List[STrack] = []
        self.lost_tracks: List[STrack] = []
        self.removed_tracks: List[STrack] = []

        # Warning, possible bug: If you also set internal_id to start at 1,
        # all traces will be connected across objects.
        self.internal_id_counter = IdCounter()
        self.external_id_counter = IdCounter(start_id=1)

    def update_with_detections(self, detections: Detections) -> Detections:
        """
        Updates the tracker with the provided detections and returns the updated
        detection results.

        Args:
            detections (Detections): The detections to pass through the tracker.

        """
        tensors = np.hstack(
            (
                detections.xyxy,
                detections.confidence[:, np.newaxis],
            )
        )
        tracks = self.update_with_tensors(tensors=tensors)

        if len(tracks) > 0:
            detection_bounding_boxes = np.asarray([det[:4] for det in tensors])
            track_bounding_boxes = np.asarray([track.tlbr for track in tracks])

            ious = box_iou_batch(detection_bounding_boxes, track_bounding_boxes)

            iou_costs = 1 - ious

            matches, _, _ = matching.linear_assignment(iou_costs, 0.5)
            detections.tracker_id = np.full(len(detections), -1, dtype=int)
            for i_detection, i_track in matches:
                detections.tracker_id[i_detection] = int(
                    tracks[i_track].external_track_id
                )

            return detections[detections.tracker_id != -1]

        else:
            detections = Detections.empty()
            detections.tracker_id = np.array([], dtype=int)

            return detections

    def reset(self) -> None:
        """
        Resets the internal state of the ByteTrack tracker.

        This method clears the tracking data, including tracked, lost,
        and removed tracks, as well as resetting the frame counter. It's
        particularly useful when processing multiple videos sequentially,
        ensuring the tracker starts with a clean state for each new video.
        """
        self.frame_id = 0
        self.internal_id_counter.reset()
        self.external_id_counter.reset()
        self.tracked_tracks = []
        self.lost_tracks = []
        self.removed_tracks = []

    def update_with_tensors(self, tensors: np.ndarray) -> List[STrack]:
        """
        Updates the tracker with the provided tensors and returns the updated tracks.

        Parameters:
            tensors: The new tensors to update with.

        Returns:
            List[STrack]: Updated tracks.
        """
        self.frame_id += 1
        activated_starcks = []
        refind_stracks = []
        lost_stracks = []
        removed_stracks = []

        scores = tensors[:, 4]
        bboxes = tensors[:, :4]

        remain_inds = scores > self.track_activation_threshold
        inds_low = scores > 0.1
        inds_high = scores < self.track_activation_threshold

        inds_second = np.logical_and(inds_low, inds_high)
        dets_second = bboxes[inds_second]
        dets = bboxes[remain_inds]
        scores_keep = scores[remain_inds]
        scores_second = scores[inds_second]

        if len(dets) > 0:
            """Detections"""
            detections = [
                STrack(
                    STrack.tlbr_to_tlwh(tlbr),
                    score_keep,
                    self.minimum_consecutive_frames,
                    self.shared_kalman,
                    self.internal_id_counter,
                    self.external_id_counter,
                )
                for (tlbr, score_keep) in zip(dets, scores_keep)
            ]
        else:
            detections = []

        """ Add newly detected tracklets to tracked_stracks"""
        unconfirmed = []
        tracked_stracks = []  # type: list[STrack]

        for track in self.tracked_tracks:
            if not track.is_activated:
                unconfirmed.append(track)
            else:
                tracked_stracks.append(track)

        """ Step 2: First association, with high score detection boxes"""
        strack_pool = joint_tracks(tracked_stracks, self.lost_tracks)
        # Predict the current location with KF
        STrack.multi_predict(strack_pool, self.shared_kalman)
        dists = matching.iou_distance(strack_pool, detections)

        dists = matching.fuse_score(dists, detections)
        matches, u_track, u_detection = matching.linear_assignment(
            dists, thresh=self.minimum_matching_threshold
        )

        for itracked, idet in matches:
            track = strack_pool[itracked]
            det = detections[idet]
            if track.state == TrackState.Tracked:
                track.update(detections[idet], self.frame_id)
                activated_starcks.append(track)
            else:
                track.re_activate(det, self.frame_id)
                refind_stracks.append(track)

        """ Step 3: Second association, with low score detection boxes"""
        # association the untrack to the low score detections
        if len(dets_second) > 0:
            """Detections"""
            detections_second = [
                STrack(
                    STrack.tlbr_to_tlwh(tlbr),
                    score_second,
                    self.minimum_consecutive_frames,
                    self.shared_kalman,
                    self.internal_id_counter,
                    self.external_id_counter,
                )
                for (tlbr, score_second) in zip(dets_second, scores_second)
            ]
        else:
            detections_second = []
        r_tracked_stracks = [
            strack_pool[i]
            for i in u_track
            if strack_pool[i].state == TrackState.Tracked
        ]
        dists = matching.iou_distance(r_tracked_stracks, detections_second)
        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]
            if track.state == TrackState.Tracked:
                track.update(det, self.frame_id)
                activated_starcks.append(track)
            else:
                track.re_activate(det, self.frame_id)
                refind_stracks.append(track)

        for it in u_track:
            track = r_tracked_stracks[it]
            if not track.state == TrackState.Lost:
                track.state = TrackState.Lost
                lost_stracks.append(track)

        """Deal with unconfirmed tracks, usually tracks with only one beginning frame"""
        detections = [detections[i] for i in u_detection]
        dists = matching.iou_distance(unconfirmed, detections)

        dists = matching.fuse_score(dists, detections)
        matches, u_unconfirmed, u_detection = matching.linear_assignment(
            dists, thresh=0.7
        )
        for itracked, idet in matches:
            unconfirmed[itracked].update(detections[idet], self.frame_id)
            activated_starcks.append(unconfirmed[itracked])
        for it in u_unconfirmed:
            track = unconfirmed[it]
            track.state = TrackState.Removed
            removed_stracks.append(track)

        """ Step 4: Init new stracks"""
        for inew in u_detection:
            track = detections[inew]
            if track.score < self.det_thresh:
                continue
            track.activate(self.kalman_filter, self.frame_id)
            activated_starcks.append(track)
        """ Step 5: Update state"""
        for track in self.lost_tracks:
            if self.frame_id - track.frame_id > self.max_time_lost:
                track.state = TrackState.Removed
                removed_stracks.append(track)

        self.tracked_tracks = [
            t for t in self.tracked_tracks if t.state == TrackState.Tracked
        ]
        self.tracked_tracks = joint_tracks(self.tracked_tracks, activated_starcks)
        self.tracked_tracks = joint_tracks(self.tracked_tracks, refind_stracks)
        self.lost_tracks = sub_tracks(self.lost_tracks, self.tracked_tracks)
        self.lost_tracks.extend(lost_stracks)
        self.lost_tracks = sub_tracks(self.lost_tracks, self.removed_tracks)
        self.removed_tracks = removed_stracks
        self.tracked_tracks, self.lost_tracks = remove_duplicate_tracks(
            self.tracked_tracks, self.lost_tracks
        )
        output_stracks = [track for track in self.tracked_tracks if track.is_activated]

        return output_stracks

reset(self)

Resets the internal state of the ByteTrack tracker.

This method clears the tracking data, including tracked, lost, and removed tracks, as well as resetting the frame counter. It's particularly useful when processing multiple videos sequentially, ensuring the tracker starts with a clean state for each new video.

Source code in supertracker/bytetrack/core.py

def reset(self) -> None:
    """
    Resets the internal state of the ByteTrack tracker.

    This method clears the tracking data, including tracked, lost,
    and removed tracks, as well as resetting the frame counter. It's
    particularly useful when processing multiple videos sequentially,
    ensuring the tracker starts with a clean state for each new video.
    """
    self.frame_id = 0
    self.internal_id_counter.reset()
    self.external_id_counter.reset()
    self.tracked_tracks = []
    self.lost_tracks = []
    self.removed_tracks = []

update_with_detections(self, detections)

Updates the tracker with the provided detections and returns the updated detection results.

Parameters:

Name	Type	Description	Default
detections	Detections	The detections to pass through the tracker.	required

Source code in supertracker/bytetrack/core.py

def update_with_detections(self, detections: Detections) -> Detections:
    """
    Updates the tracker with the provided detections and returns the updated
    detection results.

    Args:
        detections (Detections): The detections to pass through the tracker.

    """
    tensors = np.hstack(
        (
            detections.xyxy,
            detections.confidence[:, np.newaxis],
        )
    )
    tracks = self.update_with_tensors(tensors=tensors)

    if len(tracks) > 0:
        detection_bounding_boxes = np.asarray([det[:4] for det in tensors])
        track_bounding_boxes = np.asarray([track.tlbr for track in tracks])

        ious = box_iou_batch(detection_bounding_boxes, track_bounding_boxes)

        iou_costs = 1 - ious

        matches, _, _ = matching.linear_assignment(iou_costs, 0.5)
        detections.tracker_id = np.full(len(detections), -1, dtype=int)
        for i_detection, i_track in matches:
            detections.tracker_id[i_detection] = int(
                tracks[i_track].external_track_id
            )

        return detections[detections.tracker_id != -1]

    else:
        detections = Detections.empty()
        detections.tracker_id = np.array([], dtype=int)

        return detections

update_with_tensors(self, tensors)

Updates the tracker with the provided tensors and returns the updated tracks.

Parameters:

Name	Type	Description	Default
tensors	ndarray	The new tensors to update with.	required

Returns:

Type	Description
List[STrack]	Updated tracks.

Source code in supertracker/bytetrack/core.py

def update_with_tensors(self, tensors: np.ndarray) -> List[STrack]:
    """
    Updates the tracker with the provided tensors and returns the updated tracks.

    Parameters:
        tensors: The new tensors to update with.

    Returns:
        List[STrack]: Updated tracks.
    """
    self.frame_id += 1
    activated_starcks = []
    refind_stracks = []
    lost_stracks = []
    removed_stracks = []

    scores = tensors[:, 4]
    bboxes = tensors[:, :4]

    remain_inds = scores > self.track_activation_threshold
    inds_low = scores > 0.1
    inds_high = scores < self.track_activation_threshold

    inds_second = np.logical_and(inds_low, inds_high)
    dets_second = bboxes[inds_second]
    dets = bboxes[remain_inds]
    scores_keep = scores[remain_inds]
    scores_second = scores[inds_second]

    if len(dets) > 0:
        """Detections"""
        detections = [
            STrack(
                STrack.tlbr_to_tlwh(tlbr),
                score_keep,
                self.minimum_consecutive_frames,
                self.shared_kalman,
                self.internal_id_counter,
                self.external_id_counter,
            )
            for (tlbr, score_keep) in zip(dets, scores_keep)
        ]
    else:
        detections = []

    """ Add newly detected tracklets to tracked_stracks"""
    unconfirmed = []
    tracked_stracks = []  # type: list[STrack]

    for track in self.tracked_tracks:
        if not track.is_activated:
            unconfirmed.append(track)
        else:
            tracked_stracks.append(track)

    """ Step 2: First association, with high score detection boxes"""
    strack_pool = joint_tracks(tracked_stracks, self.lost_tracks)
    # Predict the current location with KF
    STrack.multi_predict(strack_pool, self.shared_kalman)
    dists = matching.iou_distance(strack_pool, detections)

    dists = matching.fuse_score(dists, detections)
    matches, u_track, u_detection = matching.linear_assignment(
        dists, thresh=self.minimum_matching_threshold
    )

    for itracked, idet in matches:
        track = strack_pool[itracked]
        det = detections[idet]
        if track.state == TrackState.Tracked:
            track.update(detections[idet], self.frame_id)
            activated_starcks.append(track)
        else:
            track.re_activate(det, self.frame_id)
            refind_stracks.append(track)

    """ Step 3: Second association, with low score detection boxes"""
    # association the untrack to the low score detections
    if len(dets_second) > 0:
        """Detections"""
        detections_second = [
            STrack(
                STrack.tlbr_to_tlwh(tlbr),
                score_second,
                self.minimum_consecutive_frames,
                self.shared_kalman,
                self.internal_id_counter,
                self.external_id_counter,
            )
            for (tlbr, score_second) in zip(dets_second, scores_second)
        ]
    else:
        detections_second = []
    r_tracked_stracks = [
        strack_pool[i]
        for i in u_track
        if strack_pool[i].state == TrackState.Tracked
    ]
    dists = matching.iou_distance(r_tracked_stracks, detections_second)
    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]
        if track.state == TrackState.Tracked:
            track.update(det, self.frame_id)
            activated_starcks.append(track)
        else:
            track.re_activate(det, self.frame_id)
            refind_stracks.append(track)

    for it in u_track:
        track = r_tracked_stracks[it]
        if not track.state == TrackState.Lost:
            track.state = TrackState.Lost
            lost_stracks.append(track)

    """Deal with unconfirmed tracks, usually tracks with only one beginning frame"""
    detections = [detections[i] for i in u_detection]
    dists = matching.iou_distance(unconfirmed, detections)

    dists = matching.fuse_score(dists, detections)
    matches, u_unconfirmed, u_detection = matching.linear_assignment(
        dists, thresh=0.7
    )
    for itracked, idet in matches:
        unconfirmed[itracked].update(detections[idet], self.frame_id)
        activated_starcks.append(unconfirmed[itracked])
    for it in u_unconfirmed:
        track = unconfirmed[it]
        track.state = TrackState.Removed
        removed_stracks.append(track)

    """ Step 4: Init new stracks"""
    for inew in u_detection:
        track = detections[inew]
        if track.score < self.det_thresh:
            continue
        track.activate(self.kalman_filter, self.frame_id)
        activated_starcks.append(track)
    """ Step 5: Update state"""
    for track in self.lost_tracks:
        if self.frame_id - track.frame_id > self.max_time_lost:
            track.state = TrackState.Removed
            removed_stracks.append(track)

    self.tracked_tracks = [
        t for t in self.tracked_tracks if t.state == TrackState.Tracked
    ]
    self.tracked_tracks = joint_tracks(self.tracked_tracks, activated_starcks)
    self.tracked_tracks = joint_tracks(self.tracked_tracks, refind_stracks)
    self.lost_tracks = sub_tracks(self.lost_tracks, self.tracked_tracks)
    self.lost_tracks.extend(lost_stracks)
    self.lost_tracks = sub_tracks(self.lost_tracks, self.removed_tracks)
    self.removed_tracks = removed_stracks
    self.tracked_tracks, self.lost_tracks = remove_duplicate_tracks(
        self.tracked_tracks, self.lost_tracks
    )
    output_stracks = [track for track in self.tracked_tracks if track.is_activated]

    return output_stracks

joint_tracks(track_list_a, track_list_b)

Joins two lists of tracks, ensuring that the resulting list does not contain tracks with duplicate internal_track_id values.

Parameters:

Name	Type	Description	Default
track_list_a	List[supertracker.bytetrack.single_object_track.STrack]	First list of tracks (with internal_track_id attribute).	required
track_list_b	List[supertracker.bytetrack.single_object_track.STrack]	Second list of tracks (with internal_track_id attribute).	required

Returns:

Type	Description
List[supertracker.bytetrack.single_object_track.STrack]	Combined list of tracks from track_list_a and track_list_b without duplicate internal_track_id values.

Source code in supertracker/bytetrack/core.py

def joint_tracks(
    track_list_a: List[STrack], track_list_b: List[STrack]
) -> List[STrack]:
    """
    Joins two lists of tracks, ensuring that the resulting list does not
    contain tracks with duplicate internal_track_id values.

    Parameters:
        track_list_a: First list of tracks (with internal_track_id attribute).
        track_list_b: Second list of tracks (with internal_track_id attribute).

    Returns:
        Combined list of tracks from track_list_a and track_list_b
            without duplicate internal_track_id values.
    """
    seen_track_ids = set()
    result = []

    for track in track_list_a + track_list_b:
        if track.internal_track_id not in seen_track_ids:
            seen_track_ids.add(track.internal_track_id)
            result.append(track)

    return result

sub_tracks(track_list_a, track_list_b)

Returns a list of tracks from track_list_a after removing any tracks that share the same internal_track_id with tracks in track_list_b.

Parameters:

Name	Type	Description	Default
track_list_a	List[supertracker.bytetrack.single_object_track.STrack]	List of tracks (with internal_track_id attribute).	required
track_list_b	List[supertracker.bytetrack.single_object_track.STrack]	List of tracks (with internal_track_id attribute) to be subtracted from track_list_a.	required

Returns:

Type	Description
List[int]	List of remaining tracks from track_list_a after subtraction.

Source code in supertracker/bytetrack/core.py

def sub_tracks(track_list_a: List[STrack], track_list_b: List[STrack]) -> List[int]:
    """
    Returns a list of tracks from track_list_a after removing any tracks
    that share the same internal_track_id with tracks in track_list_b.

    Parameters:
        track_list_a: List of tracks (with internal_track_id attribute).
        track_list_b: List of tracks (with internal_track_id attribute) to
            be subtracted from track_list_a.
    Returns:
        List of remaining tracks from track_list_a after subtraction.
    """
    tracks = {track.internal_track_id: track for track in track_list_a}
    track_ids_b = {track.internal_track_id for track in track_list_b}

    for track_id in track_ids_b:
        tracks.pop(track_id, None)

    return list(tracks.values())

kalman_filter

KalmanFilter

A simple Kalman filter for tracking bounding boxes in image space.

The 8-dimensional state space

x, y, a, h, vx, vy, va, vh

contains the bounding box center position (x, y), aspect ratio a, height h, and their respective velocities.

Object motion follows a constant velocity model. The bounding box location (x, y, a, h) is taken as direct observation of the state space (linear observation model).

Source code in supertracker/bytetrack/kalman_filter.py

class KalmanFilter:
    """
    A simple Kalman filter for tracking bounding boxes in image space.

    The 8-dimensional state space

        x, y, a, h, vx, vy, va, vh

    contains the bounding box center position (x, y), aspect ratio a, height h,
    and their respective velocities.

    Object motion follows a constant velocity model. The bounding box location
    (x, y, a, h) is taken as direct observation of the state space (linear
    observation model).
    """

    def __init__(self):
        ndim, dt = 4, 1.0

        self._motion_mat = np.eye(2 * ndim, 2 * ndim)
        for i in range(ndim):
            self._motion_mat[i, ndim + i] = dt
        self._update_mat = np.eye(ndim, 2 * ndim)
        self._std_weight_position = 1.0 / 20
        self._std_weight_velocity = 1.0 / 160

    def initiate(self, measurement: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
        """
        Create track from an unassociated measurement.

        Args:
            measurement (ndarray): Bounding box coordinates (x, y, a, h) with
                center position (x, y), aspect ratio a, and height h.

        Returns:
            Tuple[ndarray, ndarray]: Returns the mean vector (8 dimensional) and
                covariance matrix (8x8 dimensional) of the new track.
                Unobserved velocities are initialized to 0 mean.
        """
        mean_pos = measurement
        mean_vel = np.zeros_like(mean_pos)
        mean = np.r_[mean_pos, mean_vel]

        std = [
            2 * self._std_weight_position * measurement[3],
            2 * self._std_weight_position * measurement[3],
            1e-2,
            2 * self._std_weight_position * measurement[3],
            10 * self._std_weight_velocity * measurement[3],
            10 * self._std_weight_velocity * measurement[3],
            1e-5,
            10 * self._std_weight_velocity * measurement[3],
        ]
        covariance = np.diag(np.square(std))
        return mean, covariance

    def predict(
        self, mean: np.ndarray, covariance: np.ndarray
    ) -> Tuple[np.ndarray, np.ndarray]:
        """
        Run Kalman filter prediction step.

        Args:
            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:
            Tuple[ndarray, ndarray]: Returns 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]))

        mean = np.dot(mean, self._motion_mat.T)
        covariance = (
            np.linalg.multi_dot((self._motion_mat, covariance, self._motion_mat.T))
            + motion_cov
        )

        return mean, covariance

    def project(
        self, mean: np.ndarray, covariance: np.ndarray
    ) -> Tuple[np.ndarray, np.ndarray]:
        """
        Project state distribution to measurement space.

        Args:
            mean (ndarray): The state's mean vector (8 dimensional array).
            covariance (ndarray): The state's covariance matrix (8x8 dimensional).

        Returns:
            Tuple[ndarray, ndarray]: Returns the projected mean and
                covariance matrix of the given state estimate.
        """
        std = [
            self._std_weight_position * mean[3],
            self._std_weight_position * mean[3],
            1e-1,
            self._std_weight_position * mean[3],
        ]
        innovation_cov = np.diag(np.square(std))

        mean = np.dot(self._update_mat, mean)
        covariance = np.linalg.multi_dot(
            (self._update_mat, covariance, self._update_mat.T)
        )
        return mean, covariance + innovation_cov

    def multi_predict(
        self, mean: np.ndarray, covariance: np.ndarray
    ) -> Tuple[np.ndarray, np.ndarray]:
        """
        Run Kalman filter prediction step (Vectorized version).

        Args:
            mean (ndarray): The Nx8 dimensional mean matrix
                of the object states at the previous time step.
            covariance (ndarray): The Nx8x8 dimensional covariance matrices
                of the object states at the previous time step.

        Returns:
            Tuple[ndarray, ndarray]: Returns 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 * np.ones_like(mean[:, 3]),
            self._std_weight_position * mean[:, 3],
        ]
        std_vel = [
            self._std_weight_velocity * mean[:, 3],
            self._std_weight_velocity * mean[:, 3],
            1e-5 * np.ones_like(mean[:, 3]),
            self._std_weight_velocity * mean[:, 3],
        ]
        sqr = np.square(np.r_[std_pos, std_vel]).T

        motion_cov = []
        for i in range(len(mean)):
            motion_cov.append(np.diag(sqr[i]))
        motion_cov = np.asarray(motion_cov)

        mean = np.dot(mean, self._motion_mat.T)
        left = np.dot(self._motion_mat, covariance).transpose((1, 0, 2))
        covariance = np.dot(left, self._motion_mat.T) + motion_cov

        return mean, covariance

    def update(
        self, mean: np.ndarray, covariance: np.ndarray, measurement: np.ndarray
    ) -> Tuple[np.ndarray, np.ndarray]:
        """
        Run Kalman filter correction step.

        Args:
            mean (ndarray): The predicted state's mean vector (8 dimensional).
            covariance (ndarray): The state's covariance matrix (8x8 dimensional).
            measurement (ndarray): The 4-dimensional measurement vector (x, y, a, h),
                where (x, y) is the center position, a the aspect ratio,
                and h the height of the bounding box.

        Returns:
            Tuple[ndarray, ndarray]: Returns the measurement-corrected
                state distribution.
        """
        projected_mean, projected_cov = self.project(mean, covariance)

        chol_factor, lower = scipy.linalg.cho_factor(
            projected_cov, lower=True, check_finite=False
        )
        kalman_gain = scipy.linalg.cho_solve(
            (chol_factor, lower),
            np.dot(covariance, self._update_mat.T).T,
            check_finite=False,
        ).T
        innovation = measurement - projected_mean

        new_mean = mean + np.dot(innovation, kalman_gain.T)
        new_covariance = covariance - np.linalg.multi_dot(
            (kalman_gain, projected_cov, kalman_gain.T)
        )
        return new_mean, new_covariance

initiate(self, measurement)

Create track from an unassociated measurement.

Parameters:

Name	Type	Description	Default
measurement	ndarray	Bounding box coordinates (x, y, a, h) with center position (x, y), aspect ratio a, and height h.	required

Returns:

Type	Description
Tuple[ndarray, ndarray]	Returns the mean vector (8 dimensional) and covariance matrix (8x8 dimensional) of the new track. Unobserved velocities are initialized to 0 mean.

Source code in supertracker/bytetrack/kalman_filter.py

def initiate(self, measurement: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
    """
    Create track from an unassociated measurement.

    Args:
        measurement (ndarray): Bounding box coordinates (x, y, a, h) with
            center position (x, y), aspect ratio a, and height h.

    Returns:
        Tuple[ndarray, ndarray]: Returns the mean vector (8 dimensional) and
            covariance matrix (8x8 dimensional) of the new track.
            Unobserved velocities are initialized to 0 mean.
    """
    mean_pos = measurement
    mean_vel = np.zeros_like(mean_pos)
    mean = np.r_[mean_pos, mean_vel]

    std = [
        2 * self._std_weight_position * measurement[3],
        2 * self._std_weight_position * measurement[3],
        1e-2,
        2 * self._std_weight_position * measurement[3],
        10 * self._std_weight_velocity * measurement[3],
        10 * self._std_weight_velocity * measurement[3],
        1e-5,
        10 * self._std_weight_velocity * measurement[3],
    ]
    covariance = np.diag(np.square(std))
    return mean, covariance

multi_predict(self, mean, covariance)

Run Kalman filter prediction step (Vectorized version).

Parameters:

Name	Type	Description	Default
mean	ndarray	The Nx8 dimensional mean matrix of the object states at the previous time step.	required
covariance	ndarray	The Nx8x8 dimensional covariance matrices of the object states at the previous time step.	required

Returns:

Type	Description
Tuple[ndarray, ndarray]	Returns the mean vector and covariance matrix of the predicted state. Unobserved velocities are initialized to 0 mean.

Source code in supertracker/bytetrack/kalman_filter.py

def multi_predict(
    self, mean: np.ndarray, covariance: np.ndarray
) -> Tuple[np.ndarray, np.ndarray]:
    """
    Run Kalman filter prediction step (Vectorized version).

    Args:
        mean (ndarray): The Nx8 dimensional mean matrix
            of the object states at the previous time step.
        covariance (ndarray): The Nx8x8 dimensional covariance matrices
            of the object states at the previous time step.

    Returns:
        Tuple[ndarray, ndarray]: Returns 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 * np.ones_like(mean[:, 3]),
        self._std_weight_position * mean[:, 3],
    ]
    std_vel = [
        self._std_weight_velocity * mean[:, 3],
        self._std_weight_velocity * mean[:, 3],
        1e-5 * np.ones_like(mean[:, 3]),
        self._std_weight_velocity * mean[:, 3],
    ]
    sqr = np.square(np.r_[std_pos, std_vel]).T

    motion_cov = []
    for i in range(len(mean)):
        motion_cov.append(np.diag(sqr[i]))
    motion_cov = np.asarray(motion_cov)

    mean = np.dot(mean, self._motion_mat.T)
    left = np.dot(self._motion_mat, covariance).transpose((1, 0, 2))
    covariance = np.dot(left, self._motion_mat.T) + motion_cov

    return mean, covariance

predict(self, mean, covariance)

Run Kalman filter prediction step.

Parameters:

Name	Type	Description	Default
mean	ndarray	The 8 dimensional mean vector of the object state at the previous time step.	required
covariance	ndarray	The 8x8 dimensional covariance matrix of the object state at the previous time step.	required

Returns:

Type	Description
Tuple[ndarray, ndarray]	Returns the mean vector and covariance matrix of the predicted state. Unobserved velocities are initialized to 0 mean.

Source code in supertracker/bytetrack/kalman_filter.py

def predict(
    self, mean: np.ndarray, covariance: np.ndarray
) -> Tuple[np.ndarray, np.ndarray]:
    """
    Run Kalman filter prediction step.

    Args:
        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:
        Tuple[ndarray, ndarray]: Returns 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]))

    mean = np.dot(mean, self._motion_mat.T)
    covariance = (
        np.linalg.multi_dot((self._motion_mat, covariance, self._motion_mat.T))
        + motion_cov
    )

    return mean, covariance

project(self, mean, covariance)

Project state distribution to measurement space.

Parameters:

Name	Type	Description	Default
mean	ndarray	The state's mean vector (8 dimensional array).	required
covariance	ndarray	The state's covariance matrix (8x8 dimensional).	required

Returns:

Type	Description
Tuple[ndarray, ndarray]	Returns the projected mean and covariance matrix of the given state estimate.

Source code in supertracker/bytetrack/kalman_filter.py

def project(
    self, mean: np.ndarray, covariance: np.ndarray
) -> Tuple[np.ndarray, np.ndarray]:
    """
    Project state distribution to measurement space.

    Args:
        mean (ndarray): The state's mean vector (8 dimensional array).
        covariance (ndarray): The state's covariance matrix (8x8 dimensional).

    Returns:
        Tuple[ndarray, ndarray]: Returns the projected mean and
            covariance matrix of the given state estimate.
    """
    std = [
        self._std_weight_position * mean[3],
        self._std_weight_position * mean[3],
        1e-1,
        self._std_weight_position * mean[3],
    ]
    innovation_cov = np.diag(np.square(std))

    mean = np.dot(self._update_mat, mean)
    covariance = np.linalg.multi_dot(
        (self._update_mat, covariance, self._update_mat.T)
    )
    return mean, covariance + innovation_cov

update(self, mean, covariance, measurement)

Run Kalman filter correction step.

Parameters:

Name	Type	Description	Default
mean	ndarray	The predicted state's mean vector (8 dimensional).	required
covariance	ndarray	The state's covariance matrix (8x8 dimensional).	required
measurement	ndarray	The 4-dimensional measurement vector (x, y, a, h), where (x, y) is the center position, a the aspect ratio, and h the height of the bounding box.	required

Returns:

Type	Description
Tuple[ndarray, ndarray]	Returns the measurement-corrected state distribution.

Source code in supertracker/bytetrack/kalman_filter.py

def update(
    self, mean: np.ndarray, covariance: np.ndarray, measurement: np.ndarray
) -> Tuple[np.ndarray, np.ndarray]:
    """
    Run Kalman filter correction step.

    Args:
        mean (ndarray): The predicted state's mean vector (8 dimensional).
        covariance (ndarray): The state's covariance matrix (8x8 dimensional).
        measurement (ndarray): The 4-dimensional measurement vector (x, y, a, h),
            where (x, y) is the center position, a the aspect ratio,
            and h the height of the bounding box.

    Returns:
        Tuple[ndarray, ndarray]: Returns the measurement-corrected
            state distribution.
    """
    projected_mean, projected_cov = self.project(mean, covariance)

    chol_factor, lower = scipy.linalg.cho_factor(
        projected_cov, lower=True, check_finite=False
    )
    kalman_gain = scipy.linalg.cho_solve(
        (chol_factor, lower),
        np.dot(covariance, self._update_mat.T).T,
        check_finite=False,
    ).T
    innovation = measurement - projected_mean

    new_mean = mean + np.dot(innovation, kalman_gain.T)
    new_covariance = covariance - np.linalg.multi_dot(
        (kalman_gain, projected_cov, kalman_gain.T)
    )
    return new_mean, new_covariance

single_object_track

STrack

Source code in supertracker/bytetrack/single_object_track.py

class STrack:
    def __init__(
        self,
        tlwh: npt.NDArray[np.float32],
        score: npt.NDArray[np.float32],
        minimum_consecutive_frames: int,
        shared_kalman: KalmanFilter,
        internal_id_counter: IdCounter,
        external_id_counter: IdCounter,
    ):
        self.state = TrackState.New
        self.is_activated = False
        self.start_frame = 0
        self.frame_id = 0

        self._tlwh = np.asarray(tlwh, dtype=np.float32)
        self.kalman_filter = None
        self.shared_kalman = shared_kalman
        self.mean, self.covariance = None, None
        self.is_activated = False

        self.score = score
        self.tracklet_len = 0

        self.minimum_consecutive_frames = minimum_consecutive_frames

        self.internal_id_counter = internal_id_counter
        self.external_id_counter = external_id_counter
        self.internal_track_id = self.internal_id_counter.NO_ID
        self.external_track_id = self.external_id_counter.NO_ID

    def predict(self) -> None:
        mean_state = self.mean.copy()
        if self.state != TrackState.Tracked:
            mean_state[7] = 0
        self.mean, self.covariance = self.kalman_filter.predict(
            mean_state, self.covariance
        )

    @staticmethod
    def multi_predict(stracks: List[STrack], shared_kalman: KalmanFilter) -> None:
        if len(stracks) > 0:
            multi_mean = []
            multi_covariance = []
            for i, st in enumerate(stracks):
                multi_mean.append(st.mean.copy())
                multi_covariance.append(st.covariance)
                if st.state != TrackState.Tracked:
                    multi_mean[i][7] = 0

            multi_mean, multi_covariance = shared_kalman.multi_predict(
                np.asarray(multi_mean), np.asarray(multi_covariance)
            )
            for i, (mean, cov) in enumerate(zip(multi_mean, multi_covariance)):
                stracks[i].mean = mean
                stracks[i].covariance = cov

    def activate(self, kalman_filter: KalmanFilter, frame_id: int) -> None:
        """Start a new tracklet"""
        self.kalman_filter = kalman_filter
        self.internal_track_id = self.internal_id_counter.new_id()
        self.mean, self.covariance = self.kalman_filter.initiate(
            self.tlwh_to_xyah(self._tlwh)
        )

        self.tracklet_len = 0
        self.state = TrackState.Tracked
        if frame_id == 1:
            self.is_activated = True

        if self.minimum_consecutive_frames == 1:
            self.external_track_id = self.external_id_counter.new_id()

        self.frame_id = frame_id
        self.start_frame = frame_id

    def re_activate(self, new_track: STrack, frame_id: int) -> None:
        self.mean, self.covariance = self.kalman_filter.update(
            self.mean, self.covariance, self.tlwh_to_xyah(new_track.tlwh)
        )
        self.tracklet_len = 0
        self.state = TrackState.Tracked

        self.frame_id = frame_id
        self.score = new_track.score

    def update(self, new_track: STrack, frame_id: int) -> None:
        """
        Update a matched track
        :type new_track: STrack
        :type frame_id: int
        :type update_feature: bool
        :return:
        """
        self.frame_id = frame_id
        self.tracklet_len += 1

        new_tlwh = new_track.tlwh
        self.mean, self.covariance = self.kalman_filter.update(
            self.mean, self.covariance, self.tlwh_to_xyah(new_tlwh)
        )
        self.state = TrackState.Tracked
        if self.tracklet_len == self.minimum_consecutive_frames:
            self.is_activated = True
            if self.external_track_id == self.external_id_counter.NO_ID:
                self.external_track_id = self.external_id_counter.new_id()

        self.score = new_track.score

    @property
    def tlwh(self) -> npt.NDArray[np.float32]:
        """Get current position in bounding box format `(top left x, top left y,
        width, height)`.
        """
        if self.mean is None:
            return self._tlwh.copy()
        ret = self.mean[:4].copy()
        ret[2] *= ret[3]
        ret[:2] -= ret[2:] / 2
        return ret

    @property
    def tlbr(self) -> npt.NDArray[np.float32]:
        """Convert bounding box to format `(min x, min y, max x, max y)`, i.e.,
        `(top left, bottom right)`.
        """
        ret = self.tlwh.copy()
        ret[2:] += ret[:2]
        return ret

    @staticmethod
    def tlwh_to_xyah(tlwh) -> npt.NDArray[np.float32]:
        """Convert bounding box to format `(center x, center y, aspect ratio,
        height)`, where the aspect ratio is `width / height`.
        """
        ret = np.asarray(tlwh).copy()
        ret[:2] += ret[2:] / 2
        ret[2] /= ret[3]
        return ret

    def to_xyah(self) -> npt.NDArray[np.float32]:
        return self.tlwh_to_xyah(self.tlwh)

    @staticmethod
    def tlbr_to_tlwh(tlbr) -> npt.NDArray[np.float32]:
        ret = np.asarray(tlbr).copy()
        ret[2:] -= ret[:2]
        return ret

    @staticmethod
    def tlwh_to_tlbr(tlwh) -> npt.NDArray[np.float32]:
        ret = np.asarray(tlwh).copy()
        ret[2:] += ret[:2]
        return ret

    def __repr__(self) -> str:
        return "OT_{}_({}-{})".format(
            self.internal_track_id, self.start_frame, self.frame_id
        )

tlbr: npt.NDArray[np.float32] property readonly

Convert bounding box to format (min x, min y, max x, max y), i.e., (top left, bottom right).

tlwh: npt.NDArray[np.float32] property readonly

Get current position in bounding box format (top left x, top left y, width, height).

activate(self, kalman_filter, frame_id)

Start a new tracklet

Source code in supertracker/bytetrack/single_object_track.py

def activate(self, kalman_filter: KalmanFilter, frame_id: int) -> None:
    """Start a new tracklet"""
    self.kalman_filter = kalman_filter
    self.internal_track_id = self.internal_id_counter.new_id()
    self.mean, self.covariance = self.kalman_filter.initiate(
        self.tlwh_to_xyah(self._tlwh)
    )

    self.tracklet_len = 0
    self.state = TrackState.Tracked
    if frame_id == 1:
        self.is_activated = True

    if self.minimum_consecutive_frames == 1:
        self.external_track_id = self.external_id_counter.new_id()

    self.frame_id = frame_id
    self.start_frame = frame_id

tlwh_to_xyah(tlwh) staticmethod

Convert bounding box to format (center x, center y, aspect ratio, height), where the aspect ratio is width / height.

Source code in supertracker/bytetrack/single_object_track.py

@staticmethod
def tlwh_to_xyah(tlwh) -> npt.NDArray[np.float32]:
    """Convert bounding box to format `(center x, center y, aspect ratio,
    height)`, where the aspect ratio is `width / height`.
    """
    ret = np.asarray(tlwh).copy()
    ret[:2] += ret[2:] / 2
    ret[2] /= ret[3]
    return ret

update(self, new_track, frame_id)

Update a matched track :type new_track: STrack :type frame_id: int :type update_feature: bool :return:

Source code in supertracker/bytetrack/single_object_track.py

def update(self, new_track: STrack, frame_id: int) -> None:
    """
    Update a matched track
    :type new_track: STrack
    :type frame_id: int
    :type update_feature: bool
    :return:
    """
    self.frame_id = frame_id
    self.tracklet_len += 1

    new_tlwh = new_track.tlwh
    self.mean, self.covariance = self.kalman_filter.update(
        self.mean, self.covariance, self.tlwh_to_xyah(new_tlwh)
    )
    self.state = TrackState.Tracked
    if self.tracklet_len == self.minimum_consecutive_frames:
        self.is_activated = True
        if self.external_track_id == self.external_id_counter.NO_ID:
            self.external_track_id = self.external_id_counter.new_id()

    self.score = new_track.score