refactor(tracker): clean up ByteTracker implementation

- Extract _xyah_from_xyxy to module-level function (was duplicated in KalmanFilter and STrack)
- Remove unused _sync_data method
- Simplify linear_assignment unmatched index computation
- Use sets for duplicate track removal
- Remove dead code and stale comments

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

utils/tracker.py

@@ -3,6 +3,23 @@ import numpy as np
 from scipy.optimize import linear_sum_assignment
 import scipy.linalg
 
+
+def _xyah_from_xyxy(xyxy):
+    """Convert bounding box (x1, y1, x2, y2) to (center_x, center_y, aspect_ratio, height)."""
+    bbox = np.asarray(xyxy).copy()
+    cx = (bbox[0] + bbox[2]) / 2.0
+    cy = (bbox[1] + bbox[3]) / 2.0
+    w = bbox[2] - bbox[0]
+    h = bbox[3] - bbox[1]
+
+    ret = np.zeros(4, dtype=bbox.dtype)
+    ret[0] = cx
+    ret[1] = cy
+    ret[2] = w / h
+    ret[3] = h
+    return ret
+
+
 class KalmanFilter:
     """
     A simple Kalman Filter for tracking bounding boxes in image space.
@@ -27,19 +44,19 @@ class KalmanFilter:
 
     def initiate(self, measurement):
         """Create track from unassociated measurement.
-        
+
         Parameters
         ----------
-        measurement : dbo
+        measurement : ndarray
             Bounding box coordinates (x1, y1, x2, y2) with confidence score.
-            
+
         Returns
         -------
         (mean, covariance)
             Returns the mean vector (8 dimensional) and covariance matrix (8x8)
             of the new track.
         """
-        mean_pos = self._xyah_from_xyxy(measurement)
+        mean_pos = _xyah_from_xyxy(measurement)
         mean = np.r_[mean_pos, np.zeros_like(mean_pos)]
 
         std = [
@@ -79,14 +96,14 @@ class KalmanFilter:
             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(self._motion_mat, mean)
         covariance = (
@@ -117,7 +134,7 @@ class KalmanFilter:
             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))
@@ -177,34 +194,13 @@ class KalmanFilter:
         else:
             raise ValueError("invalid distance metric")
 
-    def _xyah_from_xyxy(self, xyxy):
-        """Convert bounding box to format `(center x, center y, aspect ratio,
-        height)`, where the aspect ratio is `width / height`.
-        """
-        bbox = np.asarray(xyxy).copy()
-        cx = (bbox[0] + bbox[2]) / 2.0
-        cy = (bbox[1] + bbox[3]) / 2.0
-        w = bbox[2] - bbox[0]
-        h = bbox[3] - bbox[1]
-        
-        ret = np.zeros(4, dtype=bbox.dtype)
-        ret[0] = cx
-        ret[1] = cy
-        ret[2] = w / h
-        ret[3] = h
-        return ret
-
 
 class STrack:
     """
     Single object track. Wrapper around KalmanFilter state.
     """
-    
-    def __init__(self, tlwh, score, label):
-        # wait, input is xyxy usually in our pipeline
-        # ByteTrack usually uses tlwh internally.
-        # Let's standardize to input xyxy.
 
+    def __init__(self, tlwh, score, label):
         self._tlwh = np.asarray(self._tlwh_from_xyxy(tlwh), dtype=np.float32)
         self.is_activated = False
         self.track_id = 0
@@ -216,22 +212,22 @@ class STrack:
         self.frame_id = 0
         self.time_since_update = 0
         self.hits = 0
-        
+
         # Multi-frame history
         self.history = []
-        
+
         # Kalman Filter
         self.kalman_filter = None
         self.mean = None
         self.covariance = None
-        
+
 
     def _tlwh_from_xyxy(self, xyxy):
         """Convert xyxy to tlwh."""
         w = xyxy[2] - xyxy[0]
         h = xyxy[3] - xyxy[1]
         return [xyxy[0], xyxy[1], w, h]
-        
+
     def _xyxy_from_tlwh(self, tlwh):
         """Convert tlwh to xyxy."""
         x1 = tlwh[0]
@@ -239,7 +235,7 @@ class STrack:
         x2 = x1 + tlwh[2]
         y2 = y1 + tlwh[3]
         return [x1, y1, x2, y2]
-    
+
     @property
     def tlwh(self):
         """Get current position in bounding box format `(top left x, top left y,
@@ -265,7 +261,7 @@ class STrack:
         """Start a new track (tentative until min_hits reached)."""
         self.kalman_filter = kalman_filter
         self.track_id = self.next_id()
-        self.mean, self.covariance = self.kalman_filter.initiate(self.tlbr) # Initiate needs xyxy
+        self.mean, self.covariance = self.kalman_filter.initiate(self.tlbr)
 
         self.state = 2 # Tracked
         self.frame_id = frame_id
@@ -276,13 +272,13 @@ class STrack:
     def re_activate(self, new_track, frame_id, new_id=False):
         """Reactivate a lost track with a new detection."""
         self.mean, self.covariance = self.kalman_filter.update(
-            self.mean, self.covariance, self._xyah_from_xyxy(new_track.tlbr)
+            self.mean, self.covariance, _xyah_from_xyxy(new_track.tlbr)
         )
         self.time_since_update = 0
         self.state = 2 # Tracked
         self.frame_id = frame_id
         self.score = new_track.score
-        
+
         if new_id:
             self.track_id = self.next_id()
 
@@ -294,35 +290,17 @@ class STrack:
         self.hits += 1
 
         self.mean, self.covariance = self.kalman_filter.update(
-            self.mean, self.covariance, self._xyah_from_xyxy(new_track.tlbr)
+            self.mean, self.covariance, _xyah_from_xyxy(new_track.tlbr)
         )
         self.state = 2 # Tracked
         if self.hits >= min_hits:
             self.is_activated = True  # Confirmed after N consecutive hits
-        
+
     def predict(self):
         """Propagate tracking state distribution one time step forward."""
         if self.mean is None: return
-        if self.state != 2: # Only predict if tracked? ByteTrack predicts always?
-             # Standard implementation predicts for all active/lost tracks
-             pass
         self.mean, self.covariance = self.kalman_filter.predict(self.mean, self.covariance)
 
-    def _xyah_from_xyxy(self, xyxy):
-        """Internal helper for measurement conversion."""
-        bbox = np.asarray(xyxy).copy()
-        cx = (bbox[0] + bbox[2]) / 2.0
-        cy = (bbox[1] + bbox[3]) / 2.0
-        w = bbox[2] - bbox[0]
-        h = bbox[3] - bbox[1]
-        
-        ret = np.zeros(4, dtype=bbox.dtype)
-        ret[0] = cx
-        ret[1] = cy
-        ret[2] = w / h
-        ret[3] = h
-        return ret
-
     @staticmethod
     def next_id():
         # Global counter
@@ -359,39 +337,32 @@ class ByteTracker:
     def update(self, detections_list):
         """
         Update the tracker with a list of detections.
-        
+
         Args:
             detections_list: List of dicts, each having:
                 - bbox: [x1, y1, x2, y2]
                 - score: float
                 - label: str
                 - (optional) other keys preserved
-        
+
         Returns:
             List of dicts with 'track_id' added/updated.
         """
         self.frame_id += 1
-        
-        # 0. STrack Conversion using generic interface
+
         activated_stracks = []
         refind_stracks = []
         lost_stracks = []
         removed_stracks = []
 
-        scores = [d['score'] for d in detections_list]
-        bboxes = [d['bbox'] for d in detections_list]
-        
         # Split into high and low confidence
         detections = []
         detections_second = []
-        
-        # Need to keep mapping to original dict to populate results later
-        # We wrap original dict in STrack
-        
+
         for d in detections_list:
             score = d['score']
             if score < self.track_low_thresh:
-                continue  # Background noise — discard entirely
+                continue  # Background noise -- discard entirely
 
             t = STrack(d['bbox'], score, d['label'])
             t.original_data = d  # Link back
@@ -416,7 +387,7 @@ class ByteTracker:
 
         # 2. First association (High score)
         dists = iou_distance(strack_pool, detections)
-        dists = fuse_score(dists, detections) # Optional? ByteTrack uses it
+        dists = fuse_score(dists, detections)
         matches, u_track, u_detection = linear_assignment(dists, thresh=self.match_thresh)
 
         for itracked, idet in matches:
@@ -429,9 +400,6 @@ class ByteTracker:
                 track.re_activate(det, self.frame_id, new_id=False)
                 refind_stracks.append(track)
 
-            # Persist data
-            self._sync_data(track, det)
-
         # 3. Second association (Low score)
         # Match unmatched tracks to low score detections
         r_tracked_stracks = [strack_pool[i] for i in u_track if strack_pool[i].state == 2]
@@ -448,8 +416,6 @@ class ByteTracker:
                 track.re_activate(det, self.frame_id, new_id=False)
                 refind_stracks.append(track)
 
-            self._sync_data(track, det)
-
         for it in u_track:
             track = r_tracked_stracks[it]
             if not track.state == 3: # If not already lost
@@ -467,20 +433,19 @@ class ByteTracker:
                 det = remaining_dets[idet]
                 track.update(det, self.frame_id, min_hits=self.min_hits)
                 activated_stracks.append(track)
-                self._sync_data(track, det)
 
             # Update u_detection to only contain indices not matched to unconfirmed
             matched_det_indices = set(u_detection[idet] for _, idet in matches_unc) if len(matches_unc) > 0 else set()
             u_detection = [i for i in u_detection if i not in matched_det_indices]
 
-            # Unconfirmed tracks that didn't match → remove (too noisy to keep)
+            # Unconfirmed tracks that didn't match -> remove (too noisy to keep)
             for it in u_unconfirmed:
                 track = unconfirmed[it]
                 track.state = 4  # Removed
                 removed_stracks.append(track)
 
         elif unconfirmed:
-            # No detections left to match — remove all unconfirmed
+            # No detections left to match -- remove all unconfirmed
             for track in unconfirmed:
                 track.state = 4  # Removed
                 removed_stracks.append(track)
@@ -496,7 +461,6 @@ class ByteTracker:
 
             track.activate(self.kalman_filter, self.frame_id)
             activated_stracks.append(track)
-            self._sync_data(track, track)  # Sync self
 
         if rejected_by_thresh > 0 and self.frame_id <= 5:
             logging.warning(
@@ -521,19 +485,9 @@ class ByteTracker:
                 self.removed_stracks.append(track)
         self.lost_stracks = [t for t in self.lost_stracks if self.frame_id - t.frame_id <= self.track_buffer]
 
-        # 7. Final Output Construction
-        # We need to update the original dictionaries in detections_list IN PLACE, 
-        # or return a new list. The logic in inference.py expects us to modify detections dicts
-        # or we might want to return the tracked ones.
-        # But wait, we iterate `detections_list` at start.
-        # We want to return ONLY the currently tracked/active objects? 
-        # Usually inference pipeline draws ALL detections, but standard tracking ONLY output active tracks.
-        # If we only output active tracks, we might suppress valid high-confidence detections that just started?
-        # No, activated_stracks includes new ones.
-        
-        # Let's collect all active tracks
+        # 7. Collect all active tracks for output
         output_stracks = [t for t in self.tracked_stracks if t.is_activated]
-        
+
         results = []
         for track in output_stracks:
             d_out = track.original_data.copy() if hasattr(track, 'original_data') else {}
@@ -563,10 +517,6 @@ class ByteTracker:
 
         return results
 
-    def _sync_data(self, track, det_source):
-        """Sync label from detection source to track."""
-        pass
-
 
 # --- Helper Functions ---
 
@@ -574,48 +524,35 @@ def linear_assignment(cost_matrix, thresh):
     """Linear assignment with threshold using scipy."""
     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 = [], [], []
-    
-    # Scipy linear_sum_assignment finds min cost
+
     row_ind, col_ind = linear_sum_assignment(cost_matrix)
-    
+
+    matched_rows = set(row_ind)
+    matched_cols = set(col_ind)
+
+    matches = []
+    unmatched_a = list(set(range(cost_matrix.shape[0])) - matched_rows)
+    unmatched_b = list(set(range(cost_matrix.shape[1])) - matched_cols)
+
     for r, c in zip(row_ind, col_ind):
         if cost_matrix[r, c] <= thresh:
             matches.append((r, c))
         else:
             unmatched_a.append(r)
             unmatched_b.append(c)
-            
-    # Add accumulation of indices that weren't selected
-    # (scipy returns perfect matching for square, but partial for rectangular)
-    # Actually scipy matches rows to cols. Any row not in row_ind is unmatched?
-    # No, row_ind covers all rows if N < M.
-    
-    if cost_matrix.shape[0] > cost_matrix.shape[1]: # More rows than cols
-         unmatched_a += list(set(range(cost_matrix.shape[0])) - set(row_ind))
-    elif cost_matrix.shape[0] < cost_matrix.shape[1]: # More cols than rows
-         unmatched_b += list(set(range(cost_matrix.shape[1])) - set(col_ind))
-         
-    # Also filter out threshold failures
-    for r, c in zip(row_ind, col_ind):
-        if cost_matrix[r, c] > thresh:
-             if r not in unmatched_a: unmatched_a.append(r)
-             if c not in unmatched_b: unmatched_b.append(c)
 
-    # Clean up
     matches = np.array(matches) if len(matches) > 0 else np.empty((0, 2), dtype=int)
     return matches, unmatched_a, unmatched_b
 
 
 def iou_distance(atracks, btracks):
     """Compute IOU cost matrix between tracks and detections."""
-    if (len(atracks) == 0 and len(btracks) == 0) or len(atracks) == 0 or len(btracks) == 0:
+    if len(atracks) == 0 or len(btracks) == 0:
         return np.zeros((len(atracks), len(btracks)), dtype=float)
-        
+
     atlbrs = [track.tlbr for track in atracks]
     btlbrs = [track.tlbr for track in btracks]
-    
+
     _ious = bbox_ious(np.array(atlbrs), np.array(btlbrs))
     cost_matrix = 1 - _ious
     return cost_matrix
@@ -629,12 +566,12 @@ def bbox_ious(boxes1, boxes2):
     inter_rect_y1 = np.maximum(b1_y1[:, None], b2_y1)
     inter_rect_x2 = np.minimum(b1_x2[:, None], b2_x2)
     inter_rect_y2 = np.minimum(b1_y2[:, None], b2_y2)
-    
+
     inter_area = np.maximum(inter_rect_x2 - inter_rect_x1, 0) * np.maximum(inter_rect_y2 - inter_rect_y1, 0)
-    
+
     b1_area = (b1_x2 - b1_x1) * (b1_y2 - b1_y1)
     b2_area = (b2_x2 - b2_x1) * (b2_y2 - b2_y1)
-    
+
     iou = inter_area / (b1_area[:, None] + b2_area - inter_area + 1e-6)
     return iou
 
@@ -678,26 +615,24 @@ def sub_stracks(tlist_a, tlist_b):
 def remove_duplicate_stracks(stracksa, stracksb):
     pdist = iou_distance(stracksa, stracksb)
     pairs = np.where(pdist < 0.15)
-    dupa, dupb = list(pairs[0]), list(pairs[1])
-    for a, b in zip(dupa, dupb):
+    dupa, dupb = set(), set()
+    for a, b in zip(pairs[0], pairs[1]):
         time_a = stracksa[a].frame_id - stracksa[a].start_frame
         time_b = stracksb[b].frame_id - stracksb[b].start_frame
         if time_a > time_b:
-            dupb.append(b) # Bug in orig ByteTrack? It assumes removing from list. 
-                           # We mark for removal.
+            dupb.add(b)
         else:
-            dupa.append(a)
-            
-    res_a = [t for i, t in enumerate(stracksa) if not i in dupa]
-    res_b = [t for i, t in enumerate(stracksb) if not i in dupb]
+            dupa.add(a)
+
+    res_a = [t for i, t in enumerate(stracksa) if i not in dupa]
+    res_b = [t for i, t in enumerate(stracksb) if i not in dupb]
     return res_a, res_b
 
 
-# Monkey patch for multi_predict since STrack is not in a module
-def multi_predict(stracks, kalman_filter):
+def _multi_predict(stracks, kalman_filter):
     for t in stracks:
         if t.state != 2:
             t.mean[7] = 0 # reset velocity h if lost
         t.mean, t.covariance = kalman_filter.predict(t.mean, t.covariance)
 
-STrack.multi_predict = static_method_multi_predict = multi_predict
+STrack.multi_predict = staticmethod(_multi_predict)