Tracking

A vendored ByteTrack tracker for DFINE.predict_video(track=True) — per-frame Results gain a persistent boxes.id across a video. numpy + scipy only.

dfine.track.ByteTrack

ByteTrack(frame_rate: float = 30.0, track_thresh: float = 0.25, match_thresh: float = 0.8, track_buffer: int = 30)

Stateful tracker: feed it one :class:Results per frame, get ids back.

Construct one per video (state persists across :meth:update calls), then call :meth:update on each frame's detections. The returned :class:Results carries only the currently active tracks, each with boxes.id set to its track id.

Source code in dfine/track/__init__.py

def __init__(
    self,
    frame_rate: float = 30.0,
    track_thresh: float = 0.25,
    match_thresh: float = 0.8,
    track_buffer: int = 30,
):
    self._tracker = BYTETracker(
        track_thresh=track_thresh,
        match_thresh=match_thresh,
        track_buffer=track_buffer,
        frame_rate=frame_rate,
    )

update

update(result: Results) -> Results

Track result's detections into the running tracks; return tracked boxes.

Source code in dfine/track/__init__.py

def update(self, result: Results) -> Results:
    """Track ``result``'s detections into the running tracks; return tracked boxes."""
    import torch

    boxes = result.boxes
    if len(boxes):
        xyxy = boxes.xyxy.detach().cpu().numpy().astype(np.float32).reshape(-1, 4)
        scores = boxes.conf.detach().cpu().numpy().astype(np.float32).reshape(-1)
        cls = boxes.cls.detach().cpu().numpy().reshape(-1)
    else:
        xyxy = np.zeros((0, 4), dtype=np.float32)
        scores = np.zeros((0,), dtype=np.float32)
        cls = np.zeros((0,), dtype=np.float32)

    tracks = self._tracker.update(xyxy, scores, cls)
    if tracks:
        t_xyxy = np.stack([t.xyxy for t in tracks]).astype(np.float32)
        t_conf = np.array([t.score for t in tracks], dtype=np.float32)
        t_cls = np.array([int(t.cls) for t in tracks], dtype=np.int64)
        t_id = np.array([int(t.track_id) for t in tracks], dtype=np.int64)
    else:
        t_xyxy = np.zeros((0, 4), dtype=np.float32)
        t_conf = np.zeros((0,), dtype=np.float32)
        t_cls = np.zeros((0,), dtype=np.int64)
        t_id = np.zeros((0,), dtype=np.int64)

    tracked = Boxes(
        xyxy=torch.from_numpy(t_xyxy),
        conf=torch.from_numpy(t_conf),
        cls=torch.from_numpy(t_cls),
        id=torch.from_numpy(t_id),
    )
    return Results(result.orig_img, tracked, result.names)

dfine.track.byte_tracker.BYTETracker

BYTETracker(track_thresh: float = 0.25, match_thresh: float = 0.8, track_buffer: int = 30, frame_rate: float = 30.0)

ByteTrack: Kalman prediction + two-stage IoU association.

track_thresh splits detections into high/low score; match_thresh is the max IoU-distance for a valid first-stage match; track_buffer (scaled by frame_rate) is how many frames a lost track survives before removal.

Source code in dfine/track/byte_tracker.py

def __init__(
    self,
    track_thresh: float = 0.25,
    match_thresh: float = 0.8,
    track_buffer: int = 30,
    frame_rate: float = 30.0,
):
    self.tracked_stracks: list[STrack] = []
    self.lost_stracks: list[STrack] = []
    self.removed_stracks: list[STrack] = []
    self.frame_id = 0
    self.track_thresh = track_thresh
    self.match_thresh = match_thresh
    self.det_thresh = track_thresh + 0.1
    self.max_time_lost = int(frame_rate / 30.0 * track_buffer)
    self.kalman_filter = KalmanFilterXYAH()
    STrack.reset_id()  # ids start at 1 for each new tracker (deterministic)

update

update(xyxy: ndarray, scores: ndarray, cls: ndarray) -> list[STrack]

Advance one frame; return the currently active tracks.

Source code in dfine/track/byte_tracker.py

def update(self, xyxy: np.ndarray, scores: np.ndarray, cls: np.ndarray) -> list[STrack]:
    """Advance one frame; return the currently active tracks."""
    self.frame_id += 1
    xyxy = np.asarray(xyxy, dtype=np.float32).reshape(-1, 4)
    scores = np.asarray(scores, dtype=np.float32).reshape(-1)
    cls = np.asarray(cls).reshape(-1)

    activated, refind, lost, removed = [], [], [], []

    high = scores >= self.track_thresh
    low = (scores > 0.1) & (scores < self.track_thresh)
    dets = self._init_tracks(xyxy[high], scores[high], cls[high])
    dets_second = self._init_tracks(xyxy[low], scores[low], cls[low])

    unconfirmed = [t for t in self.tracked_stracks if not t.is_activated]
    tracked = [t for t in self.tracked_stracks if t.is_activated]

    # --- stage 1: high-score detections vs tracked + lost tracks ----------
    pool = _joint(tracked, self.lost_stracks)
    for t in pool:
        t.predict()
    matches, u_track, u_det = _linear_assignment(_iou_distance(pool, dets), self.match_thresh)
    for it, idet in matches:
        track, det = pool[it], dets[idet]
        if track.state == TrackState.Tracked:
            track.update(det, self.frame_id)
            activated.append(track)
        else:
            track.re_activate(det, self.frame_id)
            refind.append(track)

    # --- stage 2: leftover tracked tracks vs low-score detections ---------
    r_tracked = [pool[i] for i in u_track if pool[i].state == TrackState.Tracked]
    matches, u_track2, _ = _linear_assignment(_iou_distance(r_tracked, dets_second), 0.5)
    for it, idet in matches:
        track, det = r_tracked[it], dets_second[idet]
        if track.state == TrackState.Tracked:
            track.update(det, self.frame_id)
            activated.append(track)
        else:
            track.re_activate(det, self.frame_id)
            refind.append(track)
    for it in u_track2:
        track = r_tracked[it]
        if track.state != TrackState.Lost:
            track.mark_lost()
            lost.append(track)

    # --- unconfirmed tracks (single-detection new tracks) -----------------
    dets = [dets[i] for i in u_det]
    matches, u_unconfirmed, u_det = _linear_assignment(_iou_distance(unconfirmed, dets), 0.7)
    for it, idet in matches:
        unconfirmed[it].update(dets[idet], self.frame_id)
        activated.append(unconfirmed[it])
    for it in u_unconfirmed:
        track = unconfirmed[it]
        track.mark_removed()
        removed.append(track)

    # --- start new tracks from strong leftover detections -----------------
    for idet in u_det:
        track = dets[idet]
        if track.score < self.det_thresh:
            continue
        track.activate(self.kalman_filter, self.frame_id)
        activated.append(track)

    # --- expire long-lost tracks -----------------------------------------
    for track in self.lost_stracks:
        if self.frame_id - track.frame_id > self.max_time_lost:
            track.mark_removed()
            removed.append(track)

    # --- bookkeeping ------------------------------------------------------
    self.tracked_stracks = [t for t in self.tracked_stracks if t.state == TrackState.Tracked]
    self.tracked_stracks = _joint(self.tracked_stracks, activated)
    self.tracked_stracks = _joint(self.tracked_stracks, refind)
    self.lost_stracks = _subtract(self.lost_stracks, self.tracked_stracks)
    self.lost_stracks.extend(lost)
    self.lost_stracks = _subtract(self.lost_stracks, self.removed_stracks)
    self.removed_stracks.extend(removed)
    self.tracked_stracks, self.lost_stracks = _remove_duplicates(
        self.tracked_stracks, self.lost_stracks
    )
    return [t for t in self.tracked_stracks if t.is_activated]