/
E.Ionenko
/
OpenFB
Обзор
Документация
Войти
/
E.Ionenko
/
OpenFB
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
resources/function_blocks/Tracker/Tracking/tracker.py
201 строка
7 KB
Oleg Botnar
Added basic version of openFB
18 дек 2025, 13:04
18 дек 2025, 13:04
d12442f
Код
Авторство
О чём код?
from __future__ import absolute_import import numpy as np from . import kalman_filter from .track import Track from scipy.optimize import linear_sum_assignment import uuid def iou(bbox, candidates): """Computer intersection over union. Parameters ---------- bbox : ndarray A bounding box in format `(top left x, top left y, width, height)`. candidates : ndarray A matrix of candidate bounding boxes (one per row) in the same format as `bbox`. Returns ------- ndarray The intersection over union in [0, 1] between the `bbox` and each candidate. A higher score means a larger fraction of the `bbox` is occluded by the candidate. """ bbox_tl, bbox_br = bbox[:2], bbox[:2] + bbox[2:] candidates_tl = candidates[:, :2] candidates_br = candidates[:, :2] + candidates[:, 2:] tl = np.c_[np.maximum(bbox_tl[0], candidates_tl[:, 0])[:, np.newaxis], np.maximum(bbox_tl[1], candidates_tl[:, 1])[:, np.newaxis]] br = np.c_[np.minimum(bbox_br[0], candidates_br[:, 0])[:, np.newaxis], np.minimum(bbox_br[1], candidates_br[:, 1])[:, np.newaxis]] wh = np.maximum(0., br - tl) area_intersection = wh.prod(axis=1) area_bbox = bbox[2:].prod() area_candidates = candidates[:, 2:].prod(axis=1) return area_intersection / (area_bbox + area_candidates - area_intersection) def iou_cost(tracks, detections, track_indices, detection_indices): """An intersection over union distance metric. Parameters ---------- tracks : List[Track] A list of tracks. detections : List[Detection] A list of detections. track_indices : Optional[List[int]] A list of indices to tracks that should be matched. Defaults to all `tracks`. detection_indices : Optional[List[int]] A list of indices to detections that should be matched. Defaults to all `detections`. Returns ------- ndarray Returns a cost matrix of shape len(track_indices), len(detection_indices) where entry (i, j) is `1 - iou(tracks[track_indices[i]], detections[detection_indices[j]])`. """ cost_matrix = np.zeros((len(track_indices), len(detection_indices))) for row, track_idx in enumerate(track_indices): # if tracks[track_idx].time_since_update > 1: # cost_matrix[row, :] = linear_assignment.INFTY_COST # continue bbox = tracks[track_idx].to_tlwh() candidates = np.asarray([detections[i].tlwh for i in detection_indices]) cost_matrix[row, :] = 1. - iou(bbox, candidates) return cost_matrix class Tracker: """ This is the multi-target tracker. Parameters ---------- max_age : int Maximum number of missed misses before a track is deleted. n_init : int Number of consecutive detections before the track is confirmed. The track state is set to `Deleted` if a miss occurs within the first `n_init` frames. Attributes ---------- max_age : int Maximum number of missed misses before a track is deleted. n_init : int Number of frames that a track remains in initialization phase. kf : kalman_filter.KalmanFilter A Kalman filter to filter target trajectories in image space. tracks : List[Track] The list of active tracks at the current time step. """ def __init__(self, max_iou_distance=0.7, max_age=30, n_init=3): self.max_distance = max_iou_distance self.max_age = max_age self.n_init = n_init self.kf = kalman_filter.KalmanFilter() self.tracks = [] def min_cost_matching(self, detections): """Solve linear assignment problem. Parameters ---------- detections : List[detection.Detection] A list of detections at the current time step. Returns ------- (List[(int, int)], List[int], List[int]) Returns a tuple with the following three entries: * A list of matched track and detection indices. * A list of unmatched track indices. * A list of unmatched detection indices. """ track_indices = np.arange(len(self.tracks)) detection_indices = np.arange(len(detections)) if len(detection_indices) == 0 or len(track_indices) == 0: return [], track_indices, detection_indices # Nothing to match. cost_matrix = iou_cost( self.tracks, detections, track_indices, detection_indices) cost_matrix[cost_matrix > self.max_distance] = self.max_distance + 1e-5 indices = np.array(linear_sum_assignment(cost_matrix)).T matches, unmatched_tracks, unmatched_detections = [], [], [] for col, detection_idx in enumerate(detection_indices): if col not in indices[:, 1]: unmatched_detections.append(detection_idx) for row, track_idx in enumerate(track_indices): if row not in indices[:, 0]: unmatched_tracks.append(track_idx) for row, col in indices: track_idx = track_indices[row] detection_idx = detection_indices[col] if cost_matrix[row, col] > self.max_distance: unmatched_tracks.append(track_idx) unmatched_detections.append(detection_idx) else: matches.append((track_idx, detection_idx)) return matches, unmatched_tracks, unmatched_detections def update(self, detections): """Perform measurement update and track management. Parameters ---------- detections : List[detection] A list of detections at the current time step. """ for track in self.tracks: track.predict(self.kf) matches, unmatched_tracks, unmatched_detections = self.min_cost_matching(detections) # Update track set. for track_idx, detection_idx in matches: self.tracks[track_idx].update( self.kf, detections[detection_idx]) for track_idx in unmatched_tracks: self.tracks[track_idx].mark_missed() for detection_idx in unmatched_detections: self._initiate_track(detections[detection_idx]) self.tracks = [t for t in self.tracks if not t.is_deleted()] return self.tracks def _initiate_track(self, detection): mean, covariance = self.kf.initiate(detection.to_xyah()) new_track_id = uuid.uuid4().hex self.tracks.append(Track(detection, mean, covariance, new_track_id, self.n_init, self.max_age))