Spaces:
Runtime error
Runtime error
| import cv2 | |
| import numpy as np | |
| import scipy | |
| import lap | |
| from scipy.spatial.distance import cdist | |
| from trackers.bytetrack import kalman_filter | |
| import time | |
| def merge_matches(m1, m2, shape): | |
| O,P,Q = shape | |
| m1 = np.asarray(m1) | |
| m2 = np.asarray(m2) | |
| M1 = scipy.sparse.coo_matrix((np.ones(len(m1)), (m1[:, 0], m1[:, 1])), shape=(O, P)) | |
| M2 = scipy.sparse.coo_matrix((np.ones(len(m2)), (m2[:, 0], m2[:, 1])), shape=(P, Q)) | |
| mask = M1*M2 | |
| match = mask.nonzero() | |
| match = list(zip(match[0], match[1])) | |
| unmatched_O = tuple(set(range(O)) - set([i for i, j in match])) | |
| unmatched_Q = tuple(set(range(Q)) - set([j for i, j in match])) | |
| return match, unmatched_O, unmatched_Q | |
| def _indices_to_matches(cost_matrix, indices, thresh): | |
| matched_cost = cost_matrix[tuple(zip(*indices))] | |
| matched_mask = (matched_cost <= thresh) | |
| matches = indices[matched_mask] | |
| unmatched_a = tuple(set(range(cost_matrix.shape[0])) - set(matches[:, 0])) | |
| unmatched_b = tuple(set(range(cost_matrix.shape[1])) - set(matches[:, 1])) | |
| return matches, unmatched_a, unmatched_b | |
| def linear_assignment(cost_matrix, thresh): | |
| 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 = [], [], [] | |
| cost, x, y = lap.lapjv(cost_matrix, extend_cost=True, cost_limit=thresh) | |
| for ix, mx in enumerate(x): | |
| if mx >= 0: | |
| matches.append([ix, mx]) | |
| unmatched_a = np.where(x < 0)[0] | |
| unmatched_b = np.where(y < 0)[0] | |
| matches = np.asarray(matches) | |
| return matches, unmatched_a, unmatched_b | |
| def ious(atlbrs, btlbrs): | |
| """ | |
| Compute cost based on IoU | |
| :type atlbrs: list[tlbr] | np.ndarray | |
| :type atlbrs: list[tlbr] | np.ndarray | |
| :rtype ious np.ndarray | |
| """ | |
| ious = np.zeros((len(atlbrs), len(btlbrs)), dtype=np.float32) | |
| if ious.size == 0: | |
| return ious | |
| ious = bbox_ious( | |
| np.ascontiguousarray(atlbrs, dtype=np.float32), | |
| np.ascontiguousarray(btlbrs, dtype=np.float32) | |
| ) | |
| return ious | |
| def iou_distance(atracks, btracks): | |
| """ | |
| Compute cost based on IoU | |
| :type atracks: list[STrack] | |
| :type btracks: list[STrack] | |
| :rtype cost_matrix np.ndarray | |
| """ | |
| if (len(atracks)>0 and isinstance(atracks[0], np.ndarray)) or (len(btracks) > 0 and isinstance(btracks[0], np.ndarray)): | |
| atlbrs = atracks | |
| btlbrs = btracks | |
| else: | |
| atlbrs = [track.tlbr for track in atracks] | |
| btlbrs = [track.tlbr for track in btracks] | |
| _ious = ious(atlbrs, btlbrs) | |
| cost_matrix = 1 - _ious | |
| return cost_matrix | |
| def v_iou_distance(atracks, btracks): | |
| """ | |
| Compute cost based on IoU | |
| :type atracks: list[STrack] | |
| :type btracks: list[STrack] | |
| :rtype cost_matrix np.ndarray | |
| """ | |
| if (len(atracks)>0 and isinstance(atracks[0], np.ndarray)) or (len(btracks) > 0 and isinstance(btracks[0], np.ndarray)): | |
| atlbrs = atracks | |
| btlbrs = btracks | |
| else: | |
| atlbrs = [track.tlwh_to_tlbr(track.pred_bbox) for track in atracks] | |
| btlbrs = [track.tlwh_to_tlbr(track.pred_bbox) for track in btracks] | |
| _ious = ious(atlbrs, btlbrs) | |
| cost_matrix = 1 - _ious | |
| return cost_matrix | |
| def embedding_distance(tracks, detections, metric='cosine'): | |
| """ | |
| :param tracks: list[STrack] | |
| :param detections: list[BaseTrack] | |
| :param metric: | |
| :return: cost_matrix np.ndarray | |
| """ | |
| cost_matrix = np.zeros((len(tracks), len(detections)), dtype=np.float32) | |
| if cost_matrix.size == 0: | |
| return cost_matrix | |
| det_features = np.asarray([track.curr_feat for track in detections], dtype=np.float32) | |
| #for i, track in enumerate(tracks): | |
| #cost_matrix[i, :] = np.maximum(0.0, cdist(track.smooth_feat.reshape(1,-1), det_features, metric)) | |
| track_features = np.asarray([track.smooth_feat for track in tracks], dtype=np.float32) | |
| cost_matrix = np.maximum(0.0, cdist(track_features, det_features, metric)) # Nomalized features | |
| return cost_matrix | |
| def gate_cost_matrix(kf, cost_matrix, tracks, detections, only_position=False): | |
| if cost_matrix.size == 0: | |
| return cost_matrix | |
| gating_dim = 2 if only_position else 4 | |
| gating_threshold = kalman_filter.chi2inv95[gating_dim] | |
| measurements = np.asarray([det.to_xyah() for det in detections]) | |
| for row, track in enumerate(tracks): | |
| gating_distance = kf.gating_distance( | |
| track.mean, track.covariance, measurements, only_position) | |
| cost_matrix[row, gating_distance > gating_threshold] = np.inf | |
| return cost_matrix | |
| def fuse_motion(kf, cost_matrix, tracks, detections, only_position=False, lambda_=0.98): | |
| if cost_matrix.size == 0: | |
| return cost_matrix | |
| gating_dim = 2 if only_position else 4 | |
| gating_threshold = kalman_filter.chi2inv95[gating_dim] | |
| measurements = np.asarray([det.to_xyah() for det in detections]) | |
| for row, track in enumerate(tracks): | |
| gating_distance = kf.gating_distance( | |
| track.mean, track.covariance, measurements, only_position, metric='maha') | |
| cost_matrix[row, gating_distance > gating_threshold] = np.inf | |
| cost_matrix[row] = lambda_ * cost_matrix[row] + (1 - lambda_) * gating_distance | |
| return cost_matrix | |
| def fuse_iou(cost_matrix, tracks, detections): | |
| if cost_matrix.size == 0: | |
| return cost_matrix | |
| reid_sim = 1 - cost_matrix | |
| iou_dist = iou_distance(tracks, detections) | |
| iou_sim = 1 - iou_dist | |
| fuse_sim = reid_sim * (1 + iou_sim) / 2 | |
| det_scores = np.array([det.score for det in detections]) | |
| det_scores = np.expand_dims(det_scores, axis=0).repeat(cost_matrix.shape[0], axis=0) | |
| #fuse_sim = fuse_sim * (1 + det_scores) / 2 | |
| fuse_cost = 1 - fuse_sim | |
| return fuse_cost | |
| def fuse_score(cost_matrix, detections): | |
| if cost_matrix.size == 0: | |
| return cost_matrix | |
| iou_sim = 1 - cost_matrix | |
| det_scores = np.array([det.score for det in detections]) | |
| det_scores = np.expand_dims(det_scores, axis=0).repeat(cost_matrix.shape[0], axis=0) | |
| fuse_sim = iou_sim * det_scores | |
| fuse_cost = 1 - fuse_sim | |
| return fuse_cost | |
| def bbox_ious(boxes, query_boxes): | |
| """ | |
| Parameters | |
| ---------- | |
| boxes: (N, 4) ndarray of float | |
| query_boxes: (K, 4) ndarray of float | |
| Returns | |
| ------- | |
| overlaps: (N, K) ndarray of overlap between boxes and query_boxes | |
| """ | |
| N = boxes.shape[0] | |
| K = query_boxes.shape[0] | |
| overlaps = np.zeros((N, K), dtype=np.float32) | |
| for k in range(K): | |
| box_area = ( | |
| (query_boxes[k, 2] - query_boxes[k, 0] + 1) * | |
| (query_boxes[k, 3] - query_boxes[k, 1] + 1) | |
| ) | |
| for n in range(N): | |
| iw = ( | |
| min(boxes[n, 2], query_boxes[k, 2]) - | |
| max(boxes[n, 0], query_boxes[k, 0]) + 1 | |
| ) | |
| if iw > 0: | |
| ih = ( | |
| min(boxes[n, 3], query_boxes[k, 3]) - | |
| max(boxes[n, 1], query_boxes[k, 1]) + 1 | |
| ) | |
| if ih > 0: | |
| ua = float( | |
| (boxes[n, 2] - boxes[n, 0] + 1) * | |
| (boxes[n, 3] - boxes[n, 1] + 1) + | |
| box_area - iw * ih | |
| ) | |
| overlaps[n, k] = iw * ih / ua | |
| return overlaps |