(self, output_results)
| 51 | |
| 52 | |
| 53 | def step(self, output_results): |
| 54 | scores = output_results["scores"] |
| 55 | classes = output_results["labels"] |
| 56 | bboxes = output_results["boxes"] # x1y1x2y2 |
| 57 | track_bboxes = output_results["track_boxes"] if "track_boxes" in output_results else None # x1y1x2y2 |
| 58 | |
| 59 | results = list() |
| 60 | results_dict = dict() |
| 61 | |
| 62 | tracks = list() |
| 63 | |
| 64 | for idx in range(scores.shape[0]): |
| 65 | if idx in self.tracks_dict and track_bboxes is not None: |
| 66 | self.tracks_dict[idx]["bbox"] = track_bboxes[idx, :].cpu().numpy().tolist() |
| 67 | |
| 68 | if scores[idx] >= self.score_thresh: |
| 69 | obj = dict() |
| 70 | obj["score"] = float(scores[idx]) |
| 71 | obj["bbox"] = bboxes[idx, :].cpu().numpy().tolist() |
| 72 | results.append(obj) |
| 73 | results_dict[idx] = obj |
| 74 | |
| 75 | tracks = [v for v in self.tracks_dict.values()] + self.unmatched_tracks |
| 76 | N = len(results) |
| 77 | M = len(tracks) |
| 78 | |
| 79 | ret = list() |
| 80 | unmatched_tracks = [t for t in range(M)] |
| 81 | unmatched_dets = [d for d in range(N)] |
| 82 | if N > 0 and M > 0: |
| 83 | det_box = torch.stack([torch.tensor(obj['bbox']) for obj in results], dim=0) # N x 4 |
| 84 | track_box = torch.stack([torch.tensor(obj['bbox']) for obj in tracks], dim=0) # M x 4 |
| 85 | cost_bbox = 1.0 - box_ops.generalized_box_iou(det_box, track_box) # N x M |
| 86 | |
| 87 | matched_indices = linear_sum_assignment(cost_bbox) |
| 88 | unmatched_dets = [d for d in range(N) if not (d in matched_indices[0])] |
| 89 | unmatched_tracks = [d for d in range(M) if not (d in matched_indices[1])] |
| 90 | |
| 91 | matches = [[],[]] |
| 92 | for (m0, m1) in zip(matched_indices[0], matched_indices[1]): |
| 93 | if cost_bbox[m0, m1] > 1.2: |
| 94 | unmatched_dets.append(m0) |
| 95 | unmatched_tracks.append(m1) |
| 96 | else: |
| 97 | matches[0].append(m0) |
| 98 | matches[1].append(m1) |
| 99 | |
| 100 | for (m0, m1) in zip(matches[0], matches[1]): |
| 101 | track = results[m0] |
| 102 | track['tracking_id'] = tracks[m1]['tracking_id'] |
| 103 | track['age'] = 1 |
| 104 | track['active'] = 1 |
| 105 | pre_box = tracks[m1]['bbox'] |
| 106 | cur_box = track['bbox'] |
| 107 | # pre_cx, pre_cy = (pre_box[0] + pre_box[2]) / 2, (pre_box[1] + pre_box[3]) / 2 |
| 108 | # cur_cx, cur_cy = (cur_box[0] + cur_box[2]) / 2, (cur_box[1] + cur_box[3]) / 2 |
| 109 | # track['vxvy'] = [cur_cx - pre_cx, cur_cy - pre_cy] |
| 110 | ret.append(track) |
no outgoing calls
no test coverage detected