up code

.gitignore

@@ -0,0 +1,18 @@
+# Virtual envirionment
+/myenv
+/venv 
+
+# Data
+/data
+video/*
+!video/.gitkeep
+
+out_video/*
+!out_video/.gitkeep
+
+# Cache
+__pycache__/
+
+# Ignore all log
+*.log
+/flagged

README.md

@@ -1,12 +1,13 @@
 ---
-title: FloodTrafficSolution
-emoji: 🏆
-colorFrom: gray
-colorTo: gray
+title: TrafficDetection
+emoji: 🐠
+colorFrom: red
+colorTo: pink
 sdk: gradio
-sdk_version: 4.19.2
+sdk_version: 4.12.0
 app_file: app.py
 pinned: false
+license: mit
 ---
 
 Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

accountService.json

@@ -0,0 +1,13 @@
+{
+  "type": "service_account",
+  "project_id": "trafficsolution-395806",
+  "private_key_id": "d734f0c912aa65cad15281d8a3fda8bd73cfefb7",
+  "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC96Qujkh6pxk1B\nb0OMihmUZIu0h+HoSS9brkmz2bkH6K/+mnSlM9QKawvcnHSlZCZZrtozorjnNJUG\nW+kFFHvgCsr0uPMO71zDGCASnd2eK1TLYnPwd7iATYdhv/JdkzsQq6NDePUN4ErT\nw3hDW52gZ9AOvfNPjdUvGEPCG+MmeGu85hbe6RNhCQFrUbZDrFr2qBTbgeaUx5/h\nD107ofbyLEcLg1AFHaZW4mc7g58jCxnBiZY1Vp06TGPkHi0hfg9JZ+vw5FIBrizR\nvHoz0hrGwyUYLgIUhrNkQElWyidmmWqFnPVaa9EznJBDdxEOFjy5XPvQGLbsGZzu\n/YcpRLJjAgMBAAECggEAAPyXe2OUdjRaDWd7Fd1X0a4ODcOQ+Gh87wCCUIVmkTRz\nyaJazcrtpJo2xxAnCP7ytg+TAaasSEM9LPnIRab8F8dmk4vsc1zqAfd7ny0jzCEU\ngTQDPzIYCs6cT/Uz2KmdqnIB2KVlvIs4SE5hLxRoRr+TVKhG/dhSTbrWC0mizf3a\n4qp0yA10YiJzaTvR30vP737lLIfwTfN/BLVLXZwwTOcY9RtndkxHwoBevakcN/w+\nibVftQykh7vKbsuAIdvPe5ZBxWmd/7rcj1ckDqQYKZC/Lzyg6lMX9fK1886OfUwi\no7fGSr2xKgEOtsqKOG1o58QW9AJ6tvdWsKslrVF8wQKBgQDizpwbDHh/x2HPT87n\nA41asW7j1MJ4+kx/pce2v2wX9Y9mJNIT7SvMIpuDV1Hda7CMsxG6o4OMDg7R7lAx\nZKDf43MPKdKXBcLgZKGeYpQKmFBfRrTZTuc5s3ZpEGD3mrQqXfbPE+FHR3LkgXMS\n9NUKkXA+jf7Ey4tZJ/FBkROyIwKBgQDWWqwjSW2eDxBrG7abSWFdRDkQYIHmu1Xu\n1SvG4suNuEXFTix72Pv1bzLIzWAV2ZcOaGD9db5U7llDVDXKmTHiycU6QPNY/AYl\njgjkQGZvKyIjbx+TkV0ZbJ5YLfgwSDsB3OftgBGQiqItL/f0WXBPuKgoUfDNVAQv\nePq4Rn5iwQKBgQCgoI6yIVzdsgk1V2JC/aMIjgzo0e+A2lN/O5t7c4pMVZ7fSkh1\n0ExqYEAe/qJ93BzHKLLvqYoqkHIyf0LQGGJO7bi5DneUeYIQhXRoxYvvDPwu3daK\n6HSdytmbgvwJn8jlEHMRKRUs9A8MghlLXZXbzDDgPJQoL7daxwz30JVQXQKBgCpX\nyv1FegPpEeM2nEHcowEbgvI7Pi3n/7eoIHevQWqDZjTtllf4qvz2tF96QmrOispr\n0A0Bf1Gjq7pjHiUhUftsxCsXvXpsfoJYpmUAW7vGF35Maz5pTqzBCh59JjPTcOST\nULVXwSB+Yj7u5No9+LOm1IDyftDwscr26QAR3NGBAoGAaIGiv1TakFzTf5KS/Vuc\nSVOlE4Dx2FG5V37W9ee8sFFjreOsm/2Bfw62tpt+pjO94gtTfh0Gfe5Eu8tr/xHl\nnDKzXbyv+LKZLdrYMF++zI6b00U48AXz4G2LzgQVgb+ItK6MsU4GyWH2/tQY0LEV\n9noIYbdc7Q3vs8OfnOkK4RQ=\n-----END PRIVATE KEY-----\n",
+  "client_email": "firebase-adminsdk-m3k6v@trafficsolution-395806.iam.gserviceaccount.com",
+  "client_id": "114258949821756902394",
+  "auth_uri": "https://accounts.google.com/o/oauth2/auth",
+  "token_uri": "https://oauth2.googleapis.com/token",
+  "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
+  "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/firebase-adminsdk-m3k6v%40trafficsolution-395806.iam.gserviceaccount.com",
+  "universe_domain": "googleapis.com"
+}

app.py

@@ -0,0 +1,172 @@
+import os
+import gradio as gr
+import cv2
+import pandas as pd
+import random
+from datetime import datetime
+
+import firebase_admin
+from firebase_admin import credentials
+from firebase_admin import firestore
+
+from ultralytics import YOLO
+from tracker import Tracker
+from utils import ID2LABEL, MODEL_PATH, AUTHEN_ACCOUNT, compute_color_for_labels
+
+
+cred = credentials.Certificate(AUTHEN_ACCOUNT)
+firebase_admin.initialize_app(cred)
+db = firestore.client()
+
+colors = [(random.randint(0, 255), random.randint(0, 255), random.randint(0, 255)) 
+          for j in range(10)]
+
+detection_threshold = 0.1
+model = YOLO(MODEL_PATH)
+
+def addToDatabase(ss_id, obj_ids):
+    try:
+        new_doc = db.collection("TrafficData").document()
+        print(new_doc.id)
+        data = {
+            "SS_ID": ss_id,
+            "TF_COUNT_CAR": len(obj_ids['car']),
+            "TF_COUNT_MOTOBIKE": len(obj_ids['bicycle']) + len(obj_ids['motocycle']),
+            "TF_COUNT_OTHERS": len(obj_ids['bus']) + len(obj_ids['truck']) + len(obj_ids['other']),
+            "TF_ID": new_doc.id,
+            "TF_TIME": datetime.utcnow()
+
+        }
+        try:
+            db.collection("TrafficData").document(new_doc.id).set(data)
+            print("Sucessfully saved to database")
+        except:
+            print("Can't upload a new data")
+
+    except:
+        print("Can't create a new data")
+
+
+def traffic_counting(video):
+   
+    obj_ids = {"person": [], 
+                  "bicycle": [], 
+                  "car": [], 
+                  "motocycle": [], 
+                  "bus": [], 
+                  "truck": [], 
+                  "other": []}
+
+    cap = cv2.VideoCapture(video)
+    ret, frame = cap.read()
+
+    tracker = Tracker()
+    while ret:
+        results = model.predict(frame)
+
+        for result in results:
+            detections = []
+            for r in result.boxes.data.tolist():
+                x1, y1, x2, y2, score, class_id = r
+                x1 = int(x1)
+                x2 = int(x2)
+                y1 = int(y1)
+                y2 = int(y2)
+                class_id = int(class_id)
+                if score > detection_threshold:
+                    detections.append([x1, y1, x2, y2, class_id, score])
+                
+
+            tracker.update(frame, detections)
+
+            for track in tracker.tracks:
+                bbox = track.bbox
+                x1, y1, x2, y2 = bbox
+                track_id = track.track_id
+                class_id = track.class_id
+
+                cv2.rectangle(frame, (int(x1), int(y1)), (int(x2), int(y2)), (compute_color_for_labels(class_id)), 3)
+                label_name = ID2LABEL[class_id] if class_id in ID2LABEL.keys() else "other"
+                if track_id not in obj_ids[label_name]:
+                    obj_ids[label_name].append(track_id)
+
+                cv2.putText(frame,f"{label_name}-{track_id}", 
+                            (int(x1) + 5, int(y1) - 5), 
+                            cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 1, cv2.LINE_AA )
+    
+        # Count each type of traffic
+        output_data = {key: len(value) for key, value in obj_ids.items()}
+        df = pd.DataFrame(list(output_data.items()), columns=['Type', 'Number'])
+       
+        yield frame, df
+        ret, frame = cap.read()
+        
+    
+    cap.release()
+    cv2.destroyAllWindows()
+    video_path = video.replace("\\", "/")
+    # addToDatabase(video_path.split("/")[-1][:-4], obj_ids)
+
+
+# input_video = gr.Video(label="Input Video")
+# output_video = gr.outputs.Video(label="Processing Video")
+# output_data = gr.Dataframe(interactive=False, label="Traffic's Frequency")
+
+# demo = gr.Interface(traffic_counting,
+#                     inputs=input_video,
+#                     outputs=[output_video, output_data],
+#                     examples=[os.path.join('video', x) for x in os.listdir('video') if x != ".gitkeep"],
+#                     allow_flagging='never'
+#                     )
+def traffic_detection(image):
+    
+    results = model.predict(image)
+    detections = []
+    obj_ids = {"person": [], 
+                  "bicycle": [], 
+                  "car": [], 
+                  "motocycle": [], 
+                  "bus": [], 
+                  "truck": [], 
+                  "other": []}
+
+    for result in results:
+        for r in result.boxes.data.tolist():
+            x1, y1, x2, y2, score, class_id = r
+            x1 = int(x1)
+            x2 = int(x2)
+            y1 = int(y1)
+            y2 = int(y2)
+            class_id = int(class_id)
+            if score > detection_threshold:
+                detections.append([x1, y1, x2, y2, class_id, score])
+            cv2.rectangle(image, (int(x1), int(y1)), (int(x2), int(y2)), (compute_color_for_labels(class_id)), 1)
+            label_name = ID2LABEL[class_id] if class_id in ID2LABEL.keys() else "other"
+            cv2.putText(image,f"{label_name}", 
+                            (int(x1) + 5, int(y1) - 5), 
+                            cv2.FONT_HERSHEY_SIMPLEX, 0.3,compute_color_for_labels(class_id), 1, cv2.LINE_AA )         
+       
+        # Count each type of traffic
+        output_data = {key: len(value) for key, value in obj_ids.items()}
+        df = pd.DataFrame(list(output_data.items()), columns=['Type', 'Number'])
+        yield image, df
+
+        
+                
+    
+
+# Input is a image
+input_image = gr.Image(label="Input Image")
+output_image = gr.Image(type="filepath", label="Processing Image")
+output_data = gr.Dataframe(interactive=False, label="Traffic's Frequency")
+demo = gr.Interface(traffic_detection,
+                    inputs=input_image,
+                    outputs=[output_image, output_data],
+                    examples=[os.path.join('image', x) for x in os.listdir('image') if x != ".gitkeep"],
+                    allow_flagging='never'
+                    )
+
+
+if __name__ == "__main__":
+    demo.queue()
+    demo.launch(share= False)

deep_sort/detection.py

@@ -0,0 +1,55 @@
+# vim: expandtab:ts=4:sw=4
+import numpy as np
+
+
+class Detection(object):
+    """
+    This class represents a bounding box detection in a single image.
+
+    Parameters
+    ----------
+    tlwh : array_like
+        Bounding box in format `(x, y, w, h)`.
+    confidence : float
+        Detector confidence score.
+    feature : array_like
+        A feature vector that describes the object contained in this image.
+
+    Attributes
+    ----------
+    tlwh : ndarray
+        Bounding box in format `(top left x, top left y, width, height)`.
+    confidence : ndarray
+        Detector confidence score.
+    class_name : ndarray
+        Detector class.
+    feature : ndarray | NoneType
+        A feature vector that describes the object contained in this image.
+
+    """
+
+    def __init__(self, tlwh, confidence, class_name, feature):
+        self.tlwh = np.asarray(tlwh, dtype=np.float32)
+        self.confidence = float(confidence)
+        self.class_name = class_name
+        self.feature = np.asarray(feature, dtype=np.float32)
+
+    def get_class(self):
+        return self.class_name
+
+    def to_tlbr(self):
+        """Convert bounding box to format `(min x, min y, max x, max y)`, i.e.,
+        `(top left, bottom right)`.
+        """
+        ret = self.tlwh.copy()
+        ret[2:] += ret[:2]
+        return ret
+
+    def to_xyah(self):
+        """Convert bounding box to format `(center x, center y, aspect ratio,
+        height)`, where the aspect ratio is `width / height`.
+        """
+        ret = self.tlwh.copy()
+        ret[:2] += ret[2:] / 2
+        ret[2] /= ret[3]
+        return ret

deep_sort/iou_matching.py

@@ -0,0 +1,81 @@
+# vim: expandtab:ts=4:sw=4
+from __future__ import absolute_import
+import numpy as np
+from . import linear_assignment
+
+
+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=None,
+             detection_indices=None):
+    """An intersection over union distance metric.
+
+    Parameters
+    ----------
+    tracks : List[deep_sort.track.Track]
+        A list of tracks.
+    detections : List[deep_sort.detection.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]])`.
+
+    """
+    if track_indices is None:
+        track_indices = np.arange(len(tracks))
+    if detection_indices is None:
+        detection_indices = np.arange(len(detections))
+
+    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

deep_sort/kalman_filter.py

@@ -0,0 +1,229 @@
+# vim: expandtab:ts=4:sw=4
+import numpy as np
+import scipy.linalg
+
+
+"""
+Table for the 0.95 quantile of the chi-square distribution with N degrees of
+freedom (contains values for N=1, ..., 9). Taken from MATLAB/Octave's chi2inv
+function and used as Mahalanobis gating threshold.
+"""
+chi2inv95 = {
+    1: 3.8415,
+    2: 5.9915,
+    3: 7.8147,
+    4: 9.4877,
+    5: 11.070,
+    6: 12.592,
+    7: 14.067,
+    8: 15.507,
+    9: 16.919}
+
+
+class KalmanFilter(object):
+    """
+    A simple Kalman filter for tracking bounding boxes in image space.
+
+    The 8-dimensional state space
+
+        x, y, a, h, vx, vy, va, vh
+
+    contains the bounding box center position (x, y), aspect ratio a, height h,
+    and their respective velocities.
+
+    Object motion follows a constant velocity model. The bounding box location
+    (x, y, a, h) is taken as direct observation of the state space (linear
+    observation model).
+
+    """
+
+    def __init__(self):
+        ndim, dt = 4, 1.
+
+        # Create Kalman filter model matrices.
+        self._motion_mat = np.eye(2 * ndim, 2 * ndim)
+        for i in range(ndim):
+            self._motion_mat[i, ndim + i] = dt
+        self._update_mat = np.eye(ndim, 2 * ndim)
+
+        # Motion and observation uncertainty are chosen relative to the current
+        # state estimate. These weights control the amount of uncertainty in
+        # the model. This is a bit hacky.
+        self._std_weight_position = 1. / 20
+        self._std_weight_velocity = 1. / 160
+
+    def initiate(self, measurement):
+        """Create track from unassociated measurement.
+
+        Parameters
+        ----------
+        measurement : ndarray
+            Bounding box coordinates (x, y, a, h) with center position (x, y),
+            aspect ratio a, and height h.
+
+        Returns
+        -------
+        (ndarray, ndarray)
+            Returns the mean vector (8 dimensional) and covariance matrix (8x8
+            dimensional) of the new track. Unobserved velocities are initialized
+            to 0 mean.
+
+        """
+        mean_pos = measurement
+        mean_vel = np.zeros_like(mean_pos)
+        mean = np.r_[mean_pos, mean_vel]
+
+        std = [
+            2 * self._std_weight_position * measurement[3],
+            2 * self._std_weight_position * measurement[3],
+            1e-2,
+            2 * self._std_weight_position * measurement[3],
+            10 * self._std_weight_velocity * measurement[3],
+            10 * self._std_weight_velocity * measurement[3],
+            1e-5,
+            10 * self._std_weight_velocity * measurement[3]]
+        covariance = np.diag(np.square(std))
+        return mean, covariance
+
+    def predict(self, mean, covariance):
+        """Run Kalman filter prediction step.
+
+        Parameters
+        ----------
+        mean : ndarray
+            The 8 dimensional mean vector of the object state at the previous
+            time step.
+        covariance : ndarray
+            The 8x8 dimensional covariance matrix of the object state at the
+            previous time step.
+
+        Returns
+        -------
+        (ndarray, ndarray)
+            Returns the mean vector and covariance matrix of the predicted
+            state. Unobserved velocities are initialized to 0 mean.
+
+        """
+        std_pos = [
+            self._std_weight_position * mean[3],
+            self._std_weight_position * mean[3],
+            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 = np.linalg.multi_dot((
+            self._motion_mat, covariance, self._motion_mat.T)) + motion_cov
+
+        return mean, covariance
+
+    def project(self, mean, covariance):
+        """Project state distribution to measurement space.
+
+        Parameters
+        ----------
+        mean : ndarray
+            The state's mean vector (8 dimensional array).
+        covariance : ndarray
+            The state's covariance matrix (8x8 dimensional).
+
+        Returns
+        -------
+        (ndarray, ndarray)
+            Returns the projected mean and covariance matrix of the given state
+            estimate.
+
+        """
+        std = [
+            self._std_weight_position * mean[3],
+            self._std_weight_position * mean[3],
+            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))
+        return mean, covariance + innovation_cov
+
+    def update(self, mean, covariance, measurement):
+        """Run Kalman filter correction step.
+
+        Parameters
+        ----------
+        mean : ndarray
+            The predicted state's mean vector (8 dimensional).
+        covariance : ndarray
+            The state's covariance matrix (8x8 dimensional).
+        measurement : ndarray
+            The 4 dimensional measurement vector (x, y, a, h), where (x, y)
+            is the center position, a the aspect ratio, and h the height of the
+            bounding box.
+
+        Returns
+        -------
+        (ndarray, ndarray)
+            Returns the measurement-corrected state distribution.
+
+        """
+        projected_mean, projected_cov = self.project(mean, covariance)
+
+        chol_factor, lower = scipy.linalg.cho_factor(
+            projected_cov, lower=True, check_finite=False)
+        kalman_gain = scipy.linalg.cho_solve(
+            (chol_factor, lower), np.dot(covariance, self._update_mat.T).T,
+            check_finite=False).T
+        innovation = measurement - projected_mean
+
+        new_mean = mean + np.dot(innovation, kalman_gain.T)
+        new_covariance = covariance - np.linalg.multi_dot((
+            kalman_gain, projected_cov, kalman_gain.T))
+        return new_mean, new_covariance
+
+    def gating_distance(self, mean, covariance, measurements,
+                        only_position=False):
+        """Compute gating distance between state distribution and measurements.
+
+        A suitable distance threshold can be obtained from `chi2inv95`. If
+        `only_position` is False, the chi-square distribution has 4 degrees of
+        freedom, otherwise 2.
+
+        Parameters
+        ----------
+        mean : ndarray
+            Mean vector over the state distribution (8 dimensional).
+        covariance : ndarray
+            Covariance of the state distribution (8x8 dimensional).
+        measurements : ndarray
+            An Nx4 dimensional matrix of N measurements, each in
+            format (x, y, a, h) where (x, y) is the bounding box center
+            position, a the aspect ratio, and h the height.
+        only_position : Optional[bool]
+            If True, distance computation is done with respect to the bounding
+            box center position only.
+
+        Returns
+        -------
+        ndarray
+            Returns an array of length N, where the i-th element contains the
+            squared Mahalanobis distance between (mean, covariance) and
+            `measurements[i]`.
+
+        """
+        mean, covariance = self.project(mean, covariance)
+        if only_position:
+            mean, covariance = mean[:2], covariance[:2, :2]
+            measurements = measurements[:, :2]
+
+        cholesky_factor = np.linalg.cholesky(covariance)
+        d = measurements - mean
+        z = scipy.linalg.solve_triangular(
+            cholesky_factor, d.T, lower=True, check_finite=False,
+            overwrite_b=True)
+        squared_maha = np.sum(z * z, axis=0)
+        return squared_maha

deep_sort/linear_assignment.py

@@ -0,0 +1,191 @@
+# vim: expandtab:ts=4:sw=4
+from __future__ import absolute_import
+import numpy as np
+from scipy.optimize import linear_sum_assignment
+from . import kalman_filter
+
+
+INFTY_COST = 1e+5
+
+
+def min_cost_matching(
+        distance_metric, max_distance, tracks, detections, track_indices=None,
+        detection_indices=None):
+    """Solve linear assignment problem.
+
+    Parameters
+    ----------
+    distance_metric : Callable[List[Track], List[Detection], List[int], List[int]) -> ndarray
+        The distance metric is given a list of tracks and detections as well as
+        a list of N track indices and M detection indices. The metric should
+        return the NxM dimensional cost matrix, where element (i, j) is the
+        association cost between the i-th track in the given track indices and
+        the j-th detection in the given detection_indices.
+    max_distance : float
+        Gating threshold. Associations with cost larger than this value are
+        disregarded.
+    tracks : List[track.Track]
+        A list of predicted tracks at the current time step.
+    detections : List[detection.Detection]
+        A list of detections at the current time step.
+    track_indices : List[int]
+        List of track indices that maps rows in `cost_matrix` to tracks in
+        `tracks` (see description above).
+    detection_indices : List[int]
+        List of detection indices that maps columns in `cost_matrix` to
+        detections in `detections` (see description above).
+
+    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.
+
+    """
+    if track_indices is None:
+        track_indices = np.arange(len(tracks))
+    if detection_indices is None:
+        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 = distance_metric(
+        tracks, detections, track_indices, detection_indices)
+    cost_matrix[cost_matrix > max_distance] = max_distance + 1e-5
+    indices = linear_sum_assignment(cost_matrix)
+    indices = np.asarray(indices)
+    indices = np.transpose(indices)
+    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] > 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 matching_cascade(
+        distance_metric, max_distance, cascade_depth, tracks, detections,
+        track_indices=None, detection_indices=None):
+    """Run matching cascade.
+
+    Parameters
+    ----------
+    distance_metric : Callable[List[Track], List[Detection], List[int], List[int]) -> ndarray
+        The distance metric is given a list of tracks and detections as well as
+        a list of N track indices and M detection indices. The metric should
+        return the NxM dimensional cost matrix, where element (i, j) is the
+        association cost between the i-th track in the given track indices and
+        the j-th detection in the given detection indices.
+    max_distance : float
+        Gating threshold. Associations with cost larger than this value are
+        disregarded.
+    cascade_depth: int
+        The cascade depth, should be se to the maximum track age.
+    tracks : List[track.Track]
+        A list of predicted tracks at the current time step.
+    detections : List[detection.Detection]
+        A list of detections at the current time step.
+    track_indices : Optional[List[int]]
+        List of track indices that maps rows in `cost_matrix` to tracks in
+        `tracks` (see description above). Defaults to all tracks.
+    detection_indices : Optional[List[int]]
+        List of detection indices that maps columns in `cost_matrix` to
+        detections in `detections` (see description above). Defaults to all
+        detections.
+
+    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.
+
+    """
+    if track_indices is None:
+        track_indices = list(range(len(tracks)))
+    if detection_indices is None:
+        detection_indices = list(range(len(detections)))
+
+    unmatched_detections = detection_indices
+    matches = []
+    for level in range(cascade_depth):
+        if len(unmatched_detections) == 0:  # No detections left
+            break
+
+        track_indices_l = [
+            k for k in track_indices
+            if tracks[k].time_since_update == 1 + level
+        ]
+        if len(track_indices_l) == 0:  # Nothing to match at this level
+            continue
+
+        matches_l, _, unmatched_detections = \
+            min_cost_matching(
+                distance_metric, max_distance, tracks, detections,
+                track_indices_l, unmatched_detections)
+        matches += matches_l
+    unmatched_tracks = list(set(track_indices) - set(k for k, _ in matches))
+    return matches, unmatched_tracks, unmatched_detections
+
+
+def gate_cost_matrix(
+        kf, cost_matrix, tracks, detections, track_indices, detection_indices,
+        gated_cost=INFTY_COST, only_position=False):
+    """Invalidate infeasible entries in cost matrix based on the state
+    distributions obtained by Kalman filtering.
+
+    Parameters
+    ----------
+    kf : The Kalman filter.
+    cost_matrix : ndarray
+        The NxM dimensional cost matrix, where N is the number of track indices
+        and M is the number of detection indices, such that entry (i, j) is the
+        association cost between `tracks[track_indices[i]]` and
+        `detections[detection_indices[j]]`.
+    tracks : List[track.Track]
+        A list of predicted tracks at the current time step.
+    detections : List[detection.Detection]
+        A list of detections at the current time step.
+    track_indices : List[int]
+        List of track indices that maps rows in `cost_matrix` to tracks in
+        `tracks` (see description above).
+    detection_indices : List[int]
+        List of detection indices that maps columns in `cost_matrix` to
+        detections in `detections` (see description above).
+    gated_cost : Optional[float]
+        Entries in the cost matrix corresponding to infeasible associations are
+        set this value. Defaults to a very large value.
+    only_position : Optional[bool]
+        If True, only the x, y position of the state distribution is considered
+        during gating. Defaults to False.
+
+    Returns
+    -------
+    ndarray
+        Returns the modified cost matrix.
+
+    """
+    gating_dim = 2 if only_position else 4
+    gating_threshold = kalman_filter.chi2inv95[gating_dim]
+    measurements = np.asarray(
+        [detections[i].to_xyah() for i in detection_indices])
+    for row, track_idx in enumerate(track_indices):
+        track = tracks[track_idx]
+        gating_distance = kf.gating_distance(
+            track.mean, track.covariance, measurements, only_position)
+        cost_matrix[row, gating_distance > gating_threshold] = gated_cost
+    return cost_matrix

deep_sort/nn_matching.py

@@ -0,0 +1,177 @@
+# vim: expandtab:ts=4:sw=4
+import numpy as np
+
+
+def _pdist(a, b):
+    """Compute pair-wise squared distance between points in `a` and `b`.
+
+    Parameters
+    ----------
+    a : array_like
+        An NxM matrix of N samples of dimensionality M.
+    b : array_like
+        An LxM matrix of L samples of dimensionality M.
+
+    Returns
+    -------
+    ndarray
+        Returns a matrix of size len(a), len(b) such that eleement (i, j)
+        contains the squared distance between `a[i]` and `b[j]`.
+
+    """
+    a, b = np.asarray(a), np.asarray(b)
+    if len(a) == 0 or len(b) == 0:
+        return np.zeros((len(a), len(b)))
+    a2, b2 = np.square(a).sum(axis=1), np.square(b).sum(axis=1)
+    r2 = -2. * np.dot(a, b.T) + a2[:, None] + b2[None, :]
+    r2 = np.clip(r2, 0., float(np.inf))
+    return r2
+
+
+def _cosine_distance(a, b, data_is_normalized=False):
+    """Compute pair-wise cosine distance between points in `a` and `b`.
+
+    Parameters
+    ----------
+    a : array_like
+        An NxM matrix of N samples of dimensionality M.
+    b : array_like
+        An LxM matrix of L samples of dimensionality M.
+    data_is_normalized : Optional[bool]
+        If True, assumes rows in a and b are unit length vectors.
+        Otherwise, a and b are explicitly normalized to lenght 1.
+
+    Returns
+    -------
+    ndarray
+        Returns a matrix of size len(a), len(b) such that eleement (i, j)
+        contains the squared distance between `a[i]` and `b[j]`.
+
+    """
+    if not data_is_normalized:
+        a = np.asarray(a) / np.linalg.norm(a, axis=1, keepdims=True)
+        b = np.asarray(b) / np.linalg.norm(b, axis=1, keepdims=True)
+    return 1. - np.dot(a, b.T)
+
+
+def _nn_euclidean_distance(x, y):
+    """ Helper function for nearest neighbor distance metric (Euclidean).
+
+    Parameters
+    ----------
+    x : ndarray
+        A matrix of N row-vectors (sample points).
+    y : ndarray
+        A matrix of M row-vectors (query points).
+
+    Returns
+    -------
+    ndarray
+        A vector of length M that contains for each entry in `y` the
+        smallest Euclidean distance to a sample in `x`.
+
+    """
+    distances = _pdist(x, y)
+    return np.maximum(0.0, distances.min(axis=0))
+
+
+def _nn_cosine_distance(x, y):
+    """ Helper function for nearest neighbor distance metric (cosine).
+
+    Parameters
+    ----------
+    x : ndarray
+        A matrix of N row-vectors (sample points).
+    y : ndarray
+        A matrix of M row-vectors (query points).
+
+    Returns
+    -------
+    ndarray
+        A vector of length M that contains for each entry in `y` the
+        smallest cosine distance to a sample in `x`.
+
+    """
+    distances = _cosine_distance(x, y)
+    return distances.min(axis=0)
+
+
+class NearestNeighborDistanceMetric(object):
+    """
+    A nearest neighbor distance metric that, for each target, returns
+    the closest distance to any sample that has been observed so far.
+
+    Parameters
+    ----------
+    metric : str
+        Either "euclidean" or "cosine".
+    matching_threshold: float
+        The matching threshold. Samples with larger distance are considered an
+        invalid match.
+    budget : Optional[int]
+        If not None, fix samples per class to at most this number. Removes
+        the oldest samples when the budget is reached.
+
+    Attributes
+    ----------
+    samples : Dict[int -> List[ndarray]]
+        A dictionary that maps from target identities to the list of samples
+        that have been observed so far.
+
+    """
+
+    def __init__(self, metric, matching_threshold, budget=None):
+
+
+        if metric == "euclidean":
+            self._metric = _nn_euclidean_distance
+        elif metric == "cosine":
+            self._metric = _nn_cosine_distance
+        else:
+            raise ValueError(
+                "Invalid metric; must be either 'euclidean' or 'cosine'")
+        self.matching_threshold = matching_threshold
+        self.budget = budget
+        self.samples = {}
+
+    def partial_fit(self, features, targets, active_targets):
+        """Update the distance metric with new data.
+
+        Parameters
+        ----------
+        features : ndarray
+            An NxM matrix of N features of dimensionality M.
+        targets : ndarray
+            An integer array of associated target identities.
+        active_targets : List[int]
+            A list of targets that are currently present in the scene.
+
+        """
+        for feature, target in zip(features, targets):
+            self.samples.setdefault(target, []).append(feature)
+            if self.budget is not None:
+                self.samples[target] = self.samples[target][-self.budget:]
+        self.samples = {k: self.samples[k] for k in active_targets}
+
+    def distance(self, features, targets):
+        """Compute distance between features and targets.
+
+        Parameters
+        ----------
+        features : ndarray
+            An NxM matrix of N features of dimensionality M.
+        targets : List[int]
+            A list of targets to match the given `features` against.
+
+        Returns
+        -------
+        ndarray
+            Returns a cost matrix of shape len(targets), len(features), where
+            element (i, j) contains the closest squared distance between
+            `targets[i]` and `features[j]`.
+
+        """
+        cost_matrix = np.zeros((len(targets), len(features)))
+        for i, target in enumerate(targets):
+            cost_matrix[i, :] = self._metric(self.samples[target], features)
+        return cost_matrix

deep_sort/preprocessing.py

@@ -0,0 +1,74 @@
+# vim: expandtab:ts=4:sw=4
+import numpy as np
+import cv2
+
+
+def non_max_suppression(boxes, classes, max_bbox_overlap, scores=None):
+    """Suppress overlapping detections.
+
+    Original code from [1]_ has been adapted to include confidence score.
+
+    .. [1] http://www.pyimagesearch.com/2015/02/16/
+           faster-non-maximum-suppression-python/
+
+    Examples
+    --------
+
+        >>> boxes = [d.roi for d in detections]
+        >>> classes = [d.classes for d in detections]
+        >>> scores = [d.confidence for d in detections]
+        >>> indices = non_max_suppression(boxes, max_bbox_overlap, scores)
+        >>> detections = [detections[i] for i in indices]
+
+    Parameters
+    ----------
+    boxes : ndarray
+        Array of ROIs (x, y, width, height).
+    max_bbox_overlap : float
+        ROIs that overlap more than this values are suppressed.
+    scores : Optional[array_like]
+        Detector confidence score.
+
+    Returns
+    -------
+    List[int]
+        Returns indices of detections that have survived non-maxima suppression.
+
+    """
+    if len(boxes) == 0:
+        return []
+
+    boxes = boxes.astype(np.float)
+    pick = []
+
+    x1 = boxes[:, 0]
+    y1 = boxes[:, 1]
+    x2 = boxes[:, 2] + boxes[:, 0]
+    y2 = boxes[:, 3] + boxes[:, 1]
+
+    area = (x2 - x1 + 1) * (y2 - y1 + 1)
+    if scores is not None:
+        idxs = np.argsort(scores)
+    else:
+        idxs = np.argsort(y2)
+
+    while len(idxs) > 0:
+        last = len(idxs) - 1
+        i = idxs[last]
+        pick.append(i)
+
+        xx1 = np.maximum(x1[i], x1[idxs[:last]])
+        yy1 = np.maximum(y1[i], y1[idxs[:last]])
+        xx2 = np.minimum(x2[i], x2[idxs[:last]])
+        yy2 = np.minimum(y2[i], y2[idxs[:last]])
+
+        w = np.maximum(0, xx2 - xx1 + 1)
+        h = np.maximum(0, yy2 - yy1 + 1)
+
+        overlap = (w * h) / area[idxs[:last]]
+
+        idxs = np.delete(
+            idxs, np.concatenate(
+                ([last], np.where(overlap > max_bbox_overlap)[0])))
+
+    return pick

deep_sort/track.py

@@ -0,0 +1,170 @@
+# vim: expandtab:ts=4:sw=4
+
+
+class TrackState:
+    """
+    Enumeration type for the single target track state. Newly created tracks are
+    classified as `tentative` until enough evidence has been collected. Then,
+    the track state is changed to `confirmed`. Tracks that are no longer alive
+    are classified as `deleted` to mark them for removal from the set of active
+    tracks.
+
+    """
+
+    Tentative = 1
+    Confirmed = 2
+    Deleted = 3
+
+
+class Track:
+    """
+    A single target track with state space `(x, y, a, h)` and associated
+    velocities, where `(x, y)` is the center of the bounding box, `a` is the
+    aspect ratio and `h` is the height.
+
+    Parameters
+    ----------
+    mean : ndarray
+        Mean vector of the initial state distribution.
+    covariance : ndarray
+        Covariance matrix of the initial state distribution.
+    track_id : int
+        A unique track identifier.
+    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.
+    max_age : int
+        The maximum number of consecutive misses before the track state is
+        set to `Deleted`.
+    feature : Optional[ndarray]
+        Feature vector of the detection this track originates from. If not None,
+        this feature is added to the `features` cache.
+
+    Attributes
+    ----------
+    mean : ndarray
+        Mean vector of the initial state distribution.
+    covariance : ndarray
+        Covariance matrix of the initial state distribution.
+    track_id : int
+        A unique track identifier.
+    hits : int
+        Total number of measurement updates.
+    age : int
+        Total number of frames since first occurance.
+    time_since_update : int
+        Total number of frames since last measurement update.
+    state : TrackState
+        The current track state.
+    features : List[ndarray]
+        A cache of features. On each measurement update, the associated feature
+        vector is added to this list.
+
+    """
+
+    def __init__(self, mean, covariance, track_id, n_init, max_age,
+                 feature=None, class_name=None):
+        self.mean = mean
+        self.covariance = covariance
+        self.track_id = track_id
+        self.hits = 1
+        self.age = 1
+        self.time_since_update = 0
+
+        self.state = TrackState.Tentative
+        self.features = []
+        if feature is not None:
+            self.features.append(feature)
+
+        self._n_init = n_init
+        self._max_age = max_age
+        self.class_name = class_name
+
+    def to_tlwh(self):
+        """Get current position in bounding box format `(top left x, top left y,
+        width, height)`.
+
+        Returns
+        -------
+        ndarray
+            The bounding box.
+
+        """
+        ret = self.mean[:4].copy()
+        ret[2] *= ret[3]
+        ret[:2] -= ret[2:] / 2
+        return ret
+
+    def to_tlbr(self):
+        """Get current position in bounding box format `(min x, miny, max x,
+        max y)`.
+
+        Returns
+        -------
+        ndarray
+            The bounding box.
+
+        """
+        ret = self.to_tlwh()
+        ret[2:] = ret[:2] + ret[2:]
+        return ret
+    
+    def get_class(self):
+        return self.class_name
+
+    def predict(self, kf):
+        """Propagate the state distribution to the current time step using a
+        Kalman filter prediction step.
+
+        Parameters
+        ----------
+        kf : kalman_filter.KalmanFilter
+            The Kalman filter.
+
+        """
+        self.mean, self.covariance = kf.predict(self.mean, self.covariance)
+        self.age += 1
+        self.time_since_update += 1
+
+    def update(self, kf, detection):
+        """Perform Kalman filter measurement update step and update the feature
+        cache.
+
+        Parameters
+        ----------
+        kf : kalman_filter.KalmanFilter
+            The Kalman filter.
+        detection : Detection
+            The associated detection.
+
+        """
+        self.mean, self.covariance = kf.update(
+            self.mean, self.covariance, detection.to_xyah())
+        self.features.append(detection.feature)
+
+        self.hits += 1
+        self.time_since_update = 0
+        if self.state == TrackState.Tentative and self.hits >= self._n_init:
+            self.state = TrackState.Confirmed
+
+    def mark_missed(self):
+        """Mark this track as missed (no association at the current time step).
+        """
+        if self.state == TrackState.Tentative:
+            self.state = TrackState.Deleted
+        elif self.time_since_update > self._max_age:
+            self.state = TrackState.Deleted
+
+    def is_tentative(self):
+        """Returns True if this track is tentative (unconfirmed).
+        """
+        return self.state == TrackState.Tentative
+
+    def is_confirmed(self):
+        """Returns True if this track is confirmed."""
+        return self.state == TrackState.Confirmed
+
+    def is_deleted(self):
+        """Returns True if this track is dead and should be deleted."""
+        return self.state == TrackState.Deleted

deep_sort/tracker.py

@@ -0,0 +1,139 @@
+# vim: expandtab:ts=4:sw=4
+from __future__ import absolute_import
+import numpy as np
+from . import kalman_filter
+from . import linear_assignment
+from . import iou_matching
+from .track import Track
+
+
+class Tracker:
+    """
+    This is the multi-target tracker.
+
+    Parameters
+    ----------
+    metric : nn_matching.NearestNeighborDistanceMetric
+        A distance metric for measurement-to-track association.
+    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
+    ----------
+    metric : nn_matching.NearestNeighborDistanceMetric
+        The distance metric used for measurement to track association.
+    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, metric, max_iou_distance=0.7, max_age=60, n_init=3):
+        self.metric = metric
+        self.max_iou_distance = max_iou_distance
+        self.max_age = max_age
+        self.n_init = n_init
+
+        self.kf = kalman_filter.KalmanFilter()
+        self.tracks = []
+        self._next_id = 1
+
+    def predict(self):
+        """Propagate track state distributions one time step forward.
+
+        This function should be called once every time step, before `update`.
+        """
+        for track in self.tracks:
+            track.predict(self.kf)
+
+    def update(self, detections):
+        """Perform measurement update and track management.
+
+        Parameters
+        ----------
+        detections : List[deep_sort.detection.Detection]
+            A list of detections at the current time step.
+
+        """
+        # Run matching cascade.
+        matches, unmatched_tracks, unmatched_detections = \
+            self._match(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()]
+
+        # Update distance metric.
+        active_targets = [t.track_id for t in self.tracks if t.is_confirmed()]
+        features, targets = [], []
+        for track in self.tracks:
+            if not track.is_confirmed():
+                continue
+            features += track.features
+            targets += [track.track_id for _ in track.features]
+            track.features = []
+        self.metric.partial_fit(
+            np.asarray(features), np.asarray(targets), active_targets)
+
+    def _match(self, detections):
+
+        def gated_metric(tracks, dets, track_indices, detection_indices):
+            features = np.array([dets[i].feature for i in detection_indices])
+            targets = np.array([tracks[i].track_id for i in track_indices])
+            cost_matrix = self.metric.distance(features, targets)
+            cost_matrix = linear_assignment.gate_cost_matrix(
+                self.kf, cost_matrix, tracks, dets, track_indices,
+                detection_indices)
+
+            return cost_matrix
+
+        # Split track set into confirmed and unconfirmed tracks.
+        confirmed_tracks = [
+            i for i, t in enumerate(self.tracks) if t.is_confirmed()]
+        unconfirmed_tracks = [
+            i for i, t in enumerate(self.tracks) if not t.is_confirmed()]
+
+        # Associate confirmed tracks using appearance features.
+        matches_a, unmatched_tracks_a, unmatched_detections = \
+            linear_assignment.matching_cascade(
+                gated_metric, self.metric.matching_threshold, self.max_age,
+                self.tracks, detections, confirmed_tracks)
+
+        # Associate remaining tracks together with unconfirmed tracks using IOU.
+        iou_track_candidates = unconfirmed_tracks + [
+            k for k in unmatched_tracks_a if
+            self.tracks[k].time_since_update == 1]
+        unmatched_tracks_a = [
+            k for k in unmatched_tracks_a if
+            self.tracks[k].time_since_update != 1]
+        matches_b, unmatched_tracks_b, unmatched_detections = \
+            linear_assignment.min_cost_matching(
+                iou_matching.iou_cost, self.max_iou_distance, self.tracks,
+                detections, iou_track_candidates, unmatched_detections)
+
+        matches = matches_a + matches_b
+        unmatched_tracks = list(set(unmatched_tracks_a + unmatched_tracks_b))
+        return matches, unmatched_tracks, unmatched_detections
+
+    def _initiate_track(self, detection):
+        mean, covariance = self.kf.initiate(detection.to_xyah())
+        class_name = detection.get_class()
+        self.tracks.append(Track(
+            mean, covariance, self._next_id, self.n_init, self.max_age,
+            detection.feature, class_name))
+        self._next_id += 1

model_data/best.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:834cf683bc111bfe901203be83b65b1aa7267c07e8e75ee74ac359408c6ea46a
+size 22552601

model_data/yolov8m.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:6c25b0b63b1a433843f06d821a9ac1deb8d5805f74f0f38772c7308c5adc55a5
+size 52117635

object_tracker_demo.py

@@ -0,0 +1,89 @@
+import os
+import random
+
+import cv2
+from ultralytics import YOLO
+from tracker import Tracker
+
+id2label = {0: "person", 1: "bicycle", 2:"car", 3:"motocycle", 5:"bus", 7:"truck"} 
+palette = (2 ** 11 - 1, 2 ** 15 - 1, 2 ** 20 - 1)
+# object_counter = {"person": 0, "car": 0, "motobike": 0, "bus": 0, "truck": 0, "other": 0}
+object_ids = {"person": [], "bicycle": [], "car": [], "motocycle": [], "bus": [], "truck": [], "other": []}
+
+video_path = os.path.join('.', 'video', 'NgaTu_01.mp4')
+# video_out_path = os.path.join('.', 'out.mp4')
+
+cap = cv2.VideoCapture(video_path)
+ret, frame = cap.read()
+
+cap_out = cv2.VideoWriter('out.avi', cv2.VideoWriter_fourcc(*'MJPG'), cap.get(cv2.CAP_PROP_FPS), (frame.shape[1], frame.shape[0]))
+
+model = YOLO('model_data/yolov8m.pt')
+
+tracker = Tracker()
+
+colors = [(random.randint(0, 255), random.randint(0, 255), random.randint(0, 255)) for j in range(10)]
+
+detection_threshold = 0.7
+
+
+def compute_color_for_labels(label):
+
+    if label == 0: # person
+        color = (85, 45, 255)
+    elif label == 2: # Car
+        color = (222, 82, 175)
+    elif label == 3: # Motobike
+        color = (0, 204, 255)
+    elif label == 5: # Bus
+        color = (0, 149, 255)
+    else:
+        color = [int((p * (label ** 2 - label + 1)) % 255) for p in palette]
+    return tuple(color)
+
+while ret:
+
+    results = model.predict(frame)
+
+    for result in results:
+        detections = []
+        for r in result.boxes.data.tolist():
+            x1, y1, x2, y2, score, class_id = r
+            x1 = int(x1)
+            x2 = int(x2)
+            y1 = int(y1)
+            y2 = int(y2)
+            class_id = int(class_id)
+            if score > detection_threshold:
+                detections.append([x1, y1, x2, y2, class_id, score])
+            
+
+        tracker.update(frame, detections)
+
+        for track in tracker.tracks:
+            bbox = track.bbox
+            x1, y1, x2, y2 = bbox
+            track_id = track.track_id
+            class_id = track.class_id
+
+            cv2.rectangle(frame, (int(x1), int(y1)), (int(x2), int(y2)), (compute_color_for_labels(class_id)), 3)
+
+            label_name = id2label[class_id] if class_id in id2label.keys() else "other"
+            if track_id not in object_ids[label_name]:
+                object_ids[label_name].append(track_id)
+
+            cv2.putText(frame,f"{label_name}-{track_id}", 
+                        (int(x1) + 5, int(y1) - 5), 
+                        cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 1, cv2.LINE_AA )
+    
+    cap_out.write(frame)
+    cv2.imshow('frame',frame)
+    cv2.waitKey(2)
+
+    ret, frame = cap.read()
+
+cap.release()
+cv2.destroyAllWindows()
+
+
+print(object_ids)

requirements.txt

@@ -0,0 +1,8 @@
+gradio==3.41.2
+gradio_client==0.5.0
+opencv-python==4.8.0.76
+pandas==2.0.3
+ultralytics==8.0.178
+scipy==1.11.2
+tensorflow==2.11.0
+firebase_admin

tools/freeze_model.py

@@ -0,0 +1,219 @@
+# vim: expandtab:ts=4:sw=4
+import argparse
+import tensorflow as tf
+import tensorflow.contrib.slim as slim
+
+
+def _batch_norm_fn(x, scope=None):
+    if scope is None:
+        scope = tf.get_variable_scope().name + "/bn"
+    return slim.batch_norm(x, scope=scope)
+
+
+def create_link(
+        incoming, network_builder, scope, nonlinearity=tf.nn.elu,
+        weights_initializer=tf.truncated_normal_initializer(stddev=1e-3),
+        regularizer=None, is_first=False, summarize_activations=True):
+    if is_first:
+        network = incoming
+    else:
+        network = _batch_norm_fn(incoming, scope=scope + "/bn")
+        network = nonlinearity(network)
+        if summarize_activations:
+            tf.summary.histogram(scope+"/activations", network)
+
+    pre_block_network = network
+    post_block_network = network_builder(pre_block_network, scope)
+
+    incoming_dim = pre_block_network.get_shape().as_list()[-1]
+    outgoing_dim = post_block_network.get_shape().as_list()[-1]
+    if incoming_dim != outgoing_dim:
+        assert outgoing_dim == 2 * incoming_dim, \
+            "%d != %d" % (outgoing_dim, 2 * incoming)
+        projection = slim.conv2d(
+            incoming, outgoing_dim, 1, 2, padding="SAME", activation_fn=None,
+            scope=scope+"/projection", weights_initializer=weights_initializer,
+            biases_initializer=None, weights_regularizer=regularizer)
+        network = projection + post_block_network
+    else:
+        network = incoming + post_block_network
+    return network
+
+
+def create_inner_block(
+        incoming, scope, nonlinearity=tf.nn.elu,
+        weights_initializer=tf.truncated_normal_initializer(1e-3),
+        bias_initializer=tf.zeros_initializer(), regularizer=None,
+        increase_dim=False, summarize_activations=True):
+    n = incoming.get_shape().as_list()[-1]
+    stride = 1
+    if increase_dim:
+        n *= 2
+        stride = 2
+
+    incoming = slim.conv2d(
+        incoming, n, [3, 3], stride, activation_fn=nonlinearity, padding="SAME",
+        normalizer_fn=_batch_norm_fn, weights_initializer=weights_initializer,
+        biases_initializer=bias_initializer, weights_regularizer=regularizer,
+        scope=scope + "/1")
+    if summarize_activations:
+        tf.summary.histogram(incoming.name + "/activations", incoming)
+
+    incoming = slim.dropout(incoming, keep_prob=0.6)
+
+    incoming = slim.conv2d(
+        incoming, n, [3, 3], 1, activation_fn=None, padding="SAME",
+        normalizer_fn=None, weights_initializer=weights_initializer,
+        biases_initializer=bias_initializer, weights_regularizer=regularizer,
+        scope=scope + "/2")
+    return incoming
+
+
+def residual_block(incoming, scope, nonlinearity=tf.nn.elu,
+                   weights_initializer=tf.truncated_normal_initializer(1e3),
+                   bias_initializer=tf.zeros_initializer(), regularizer=None,
+                   increase_dim=False, is_first=False,
+                   summarize_activations=True):
+
+    def network_builder(x, s):
+        return create_inner_block(
+            x, s, nonlinearity, weights_initializer, bias_initializer,
+            regularizer, increase_dim, summarize_activations)
+
+    return create_link(
+        incoming, network_builder, scope, nonlinearity, weights_initializer,
+        regularizer, is_first, summarize_activations)
+
+
+def _create_network(incoming, reuse=None, weight_decay=1e-8):
+    nonlinearity = tf.nn.elu
+    conv_weight_init = tf.truncated_normal_initializer(stddev=1e-3)
+    conv_bias_init = tf.zeros_initializer()
+    conv_regularizer = slim.l2_regularizer(weight_decay)
+    fc_weight_init = tf.truncated_normal_initializer(stddev=1e-3)
+    fc_bias_init = tf.zeros_initializer()
+    fc_regularizer = slim.l2_regularizer(weight_decay)
+
+    def batch_norm_fn(x):
+        return slim.batch_norm(x, scope=tf.get_variable_scope().name + "/bn")
+
+    network = incoming
+    network = slim.conv2d(
+        network, 32, [3, 3], stride=1, activation_fn=nonlinearity,
+        padding="SAME", normalizer_fn=batch_norm_fn, scope="conv1_1",
+        weights_initializer=conv_weight_init, biases_initializer=conv_bias_init,
+        weights_regularizer=conv_regularizer)
+    network = slim.conv2d(
+        network, 32, [3, 3], stride=1, activation_fn=nonlinearity,
+        padding="SAME", normalizer_fn=batch_norm_fn, scope="conv1_2",
+        weights_initializer=conv_weight_init, biases_initializer=conv_bias_init,
+        weights_regularizer=conv_regularizer)
+
+    # NOTE(nwojke): This is missing a padding="SAME" to match the CNN
+    # architecture in Table 1 of the paper. Information on how this affects
+    # performance on MOT 16 training sequences can be found in
+    # issue 10 https://github.com/nwojke/deep_sort/issues/10
+    network = slim.max_pool2d(network, [3, 3], [2, 2], scope="pool1")
+
+    network = residual_block(
+        network, "conv2_1", nonlinearity, conv_weight_init, conv_bias_init,
+        conv_regularizer, increase_dim=False, is_first=True)
+    network = residual_block(
+        network, "conv2_3", nonlinearity, conv_weight_init, conv_bias_init,
+        conv_regularizer, increase_dim=False)
+
+    network = residual_block(
+        network, "conv3_1", nonlinearity, conv_weight_init, conv_bias_init,
+        conv_regularizer, increase_dim=True)
+    network = residual_block(
+        network, "conv3_3", nonlinearity, conv_weight_init, conv_bias_init,
+        conv_regularizer, increase_dim=False)
+
+    network = residual_block(
+        network, "conv4_1", nonlinearity, conv_weight_init, conv_bias_init,
+        conv_regularizer, increase_dim=True)
+    network = residual_block(
+        network, "conv4_3", nonlinearity, conv_weight_init, conv_bias_init,
+        conv_regularizer, increase_dim=False)
+
+    feature_dim = network.get_shape().as_list()[-1]
+    network = slim.flatten(network)
+
+    network = slim.dropout(network, keep_prob=0.6)
+    network = slim.fully_connected(
+        network, feature_dim, activation_fn=nonlinearity,
+        normalizer_fn=batch_norm_fn, weights_regularizer=fc_regularizer,
+        scope="fc1", weights_initializer=fc_weight_init,
+        biases_initializer=fc_bias_init)
+
+    features = network
+
+    # Features in rows, normalize axis 1.
+    features = slim.batch_norm(features, scope="ball", reuse=reuse)
+    feature_norm = tf.sqrt(
+        tf.constant(1e-8, tf.float32) +
+        tf.reduce_sum(tf.square(features), [1], keepdims=True))
+    features = features / feature_norm
+    return features, None
+
+
+def _network_factory(weight_decay=1e-8):
+
+    def factory_fn(image, reuse):
+            with slim.arg_scope([slim.batch_norm, slim.dropout],
+                                is_training=False):
+                with slim.arg_scope([slim.conv2d, slim.fully_connected,
+                                     slim.batch_norm, slim.layer_norm],
+                                    reuse=reuse):
+                    features, logits = _create_network(
+                        image, reuse=reuse, weight_decay=weight_decay)
+                    return features, logits
+
+    return factory_fn
+
+
+def _preprocess(image):
+    image = image[:, :, ::-1]  # BGR to RGB
+    return image
+
+
+def parse_args():
+    """Parse command line arguments.
+    """
+    parser = argparse.ArgumentParser(description="Freeze old model")
+    parser.add_argument(
+        "--checkpoint_in",
+        default="resources/networks/mars-small128.ckpt-68577",
+        help="Path to checkpoint file")
+    parser.add_argument(
+        "--graphdef_out",
+        default="resources/networks/mars-small128.pb")
+    return parser.parse_args()
+
+
+def main():
+    args = parse_args()
+
+    with tf.Session(graph=tf.Graph()) as session:
+        input_var = tf.placeholder(
+            tf.uint8, (None, 128, 64, 3), name="images")
+        image_var = tf.map_fn(
+            lambda x: _preprocess(x), tf.cast(input_var, tf.float32),
+            back_prop=False)
+
+        factory_fn = _network_factory()
+        features, _ = factory_fn(image_var, reuse=None)
+        features = tf.identity(features, name="features")
+
+        saver = tf.train.Saver(slim.get_variables_to_restore())
+        saver.restore(session, args.checkpoint_in)
+
+        output_graph_def = tf.graph_util.convert_variables_to_constants(
+            session, tf.get_default_graph().as_graph_def(),
+            [features.name.split(":")[0]])
+        with tf.gfile.GFile(args.graphdef_out, "wb") as file_handle:
+            file_handle.write(output_graph_def.SerializeToString())
+
+
+if __name__ == "__main__":
+    main()

tools/generate_detections.py

@@ -0,0 +1,218 @@
+# vim: expandtab:ts=4:sw=4
+import os
+import errno
+import argparse
+import numpy as np
+import cv2
+import tensorflow.compat.v1 as tf
+
+#tf.compat.v1.disable_eager_execution()
+
+physical_devices = tf.config.experimental.list_physical_devices('GPU')
+if len(physical_devices) > 0:
+    tf.config.experimental.set_memory_growth(physical_devices[0], True)
+
+def _run_in_batches(f, data_dict, out, batch_size):
+    data_len = len(out)
+    num_batches = int(data_len / batch_size)
+
+    s, e = 0, 0
+    for i in range(num_batches):
+        s, e = i * batch_size, (i + 1) * batch_size
+        batch_data_dict = {k: v[s:e] for k, v in data_dict.items()}
+        out[s:e] = f(batch_data_dict)
+    if e < len(out):
+        batch_data_dict = {k: v[e:] for k, v in data_dict.items()}
+        out[e:] = f(batch_data_dict)
+
+
+def extract_image_patch(image, bbox, patch_shape):
+    """Extract image patch from bounding box.
+
+    Parameters
+    ----------
+    image : ndarray
+        The full image.
+    bbox : array_like
+        The bounding box in format (x, y, width, height).
+    patch_shape : Optional[array_like]
+        This parameter can be used to enforce a desired patch shape
+        (height, width). First, the `bbox` is adapted to the aspect ratio
+        of the patch shape, then it is clipped at the image boundaries.
+        If None, the shape is computed from :arg:`bbox`.
+
+    Returns
+    -------
+    ndarray | NoneType
+        An image patch showing the :arg:`bbox`, optionally reshaped to
+        :arg:`patch_shape`.
+        Returns None if the bounding box is empty or fully outside of the image
+        boundaries.
+
+    """
+    bbox = np.array(bbox)
+    if patch_shape is not None:
+        # correct aspect ratio to patch shape
+        target_aspect = float(patch_shape[1]) / patch_shape[0]
+        new_width = target_aspect * bbox[3]
+        bbox[0] -= (new_width - bbox[2]) / 2
+        bbox[2] = new_width
+
+    # convert to top left, bottom right
+    bbox[2:] += bbox[:2]
+    bbox = bbox.astype(np.int32)
+
+    # clip at image boundaries
+    bbox[:2] = np.maximum(0, bbox[:2])
+    bbox[2:] = np.minimum(np.asarray(image.shape[:2][::-1]) - 1, bbox[2:])
+    if np.any(bbox[:2] >= bbox[2:]):
+        return None
+    sx, sy, ex, ey = bbox
+    image = image[sy:ey, sx:ex]
+    image = cv2.resize(image, tuple(patch_shape[::-1]))
+    return image
+
+
+class ImageEncoder(object):
+
+    def __init__(self, checkpoint_filename, input_name="images",
+                 output_name="features"):
+        self.session = tf.Session()
+        with tf.gfile.GFile(checkpoint_filename, "rb") as file_handle:
+            graph_def = tf.GraphDef()
+            graph_def.ParseFromString(file_handle.read())
+        tf.import_graph_def(graph_def, name="net")
+        self.input_var = tf.get_default_graph().get_tensor_by_name(
+            "%s:0" % input_name)
+        self.output_var = tf.get_default_graph().get_tensor_by_name(
+            "%s:0" % output_name)
+
+        assert len(self.output_var.get_shape()) == 2
+        assert len(self.input_var.get_shape()) == 4
+        self.feature_dim = self.output_var.get_shape().as_list()[-1]
+        self.image_shape = self.input_var.get_shape().as_list()[1:]
+
+    def __call__(self, data_x, batch_size=32):
+        out = np.zeros((len(data_x), self.feature_dim), np.float32)
+        _run_in_batches(
+            lambda x: self.session.run(self.output_var, feed_dict=x),
+            {self.input_var: data_x}, out, batch_size)
+        return out
+
+
+def create_box_encoder(model_filename, input_name="images",
+                       output_name="features", batch_size=32):
+    image_encoder = ImageEncoder(model_filename, input_name, output_name)
+    image_shape = image_encoder.image_shape
+
+    def encoder(image, boxes):
+        image_patches = []
+        for box in boxes:
+            patch = extract_image_patch(image, box, image_shape[:2])
+            if patch is None:
+                print("WARNING: Failed to extract image patch: %s." % str(box))
+                patch = np.random.uniform(
+                    0., 255., image_shape).astype(np.uint8)
+            image_patches.append(patch)
+        image_patches = np.asarray(image_patches)
+        return image_encoder(image_patches, batch_size)
+
+    return encoder
+
+
+def generate_detections(encoder, mot_dir, output_dir, detection_dir=None):
+    """Generate detections with features.
+
+    Parameters
+    ----------
+    encoder : Callable[image, ndarray] -> ndarray
+        The encoder function takes as input a BGR color image and a matrix of
+        bounding boxes in format `(x, y, w, h)` and returns a matrix of
+        corresponding feature vectors.
+    mot_dir : str
+        Path to the MOTChallenge directory (can be either train or test).
+    output_dir
+        Path to the output directory. Will be created if it does not exist.
+    detection_dir
+        Path to custom detections. The directory structure should be the default
+        MOTChallenge structure: `[sequence]/det/det.txt`. If None, uses the
+        standard MOTChallenge detections.
+
+    """
+    if detection_dir is None:
+        detection_dir = mot_dir
+    try:
+        os.makedirs(output_dir)
+    except OSError as exception:
+        if exception.errno == errno.EEXIST and os.path.isdir(output_dir):
+            pass
+        else:
+            raise ValueError(
+                "Failed to created output directory '%s'" % output_dir)
+
+    for sequence in os.listdir(mot_dir):
+        print("Processing %s" % sequence)
+        sequence_dir = os.path.join(mot_dir, sequence)
+
+        image_dir = os.path.join(sequence_dir, "img1")
+        image_filenames = {
+            int(os.path.splitext(f)[0]): os.path.join(image_dir, f)
+            for f in os.listdir(image_dir)}
+
+        detection_file = os.path.join(
+            detection_dir, sequence, "det/det.txt")
+        detections_in = np.loadtxt(detection_file, delimiter=',')
+        detections_out = []
+
+        frame_indices = detections_in[:, 0].astype(np.int32)
+        min_frame_idx = frame_indices.astype(np.int32).min()
+        max_frame_idx = frame_indices.astype(np.int32).max()
+        for frame_idx in range(min_frame_idx, max_frame_idx + 1):
+            print("Frame %05d/%05d" % (frame_idx, max_frame_idx))
+            mask = frame_indices == frame_idx
+            rows = detections_in[mask]
+
+            if frame_idx not in image_filenames:
+                print("WARNING could not find image for frame %d" % frame_idx)
+                continue
+            bgr_image = cv2.imread(
+                image_filenames[frame_idx], cv2.IMREAD_COLOR)
+            features = encoder(bgr_image, rows[:, 2:6].copy())
+            detections_out += [np.r_[(row, feature)] for row, feature
+                               in zip(rows, features)]
+
+        output_filename = os.path.join(output_dir, "%s.npy" % sequence)
+        np.save(
+            output_filename, np.asarray(detections_out), allow_pickle=False)
+
+
+def parse_args():
+    """Parse command line arguments.
+    """
+    parser = argparse.ArgumentParser(description="Re-ID feature extractor")
+    parser.add_argument(
+        "--model",
+        default="resources/networks/mars-small128.pb",
+        help="Path to freezed inference graph protobuf.")
+    parser.add_argument(
+        "--mot_dir", help="Path to MOTChallenge directory (train or test)",
+        required=True)
+    parser.add_argument(
+        "--detection_dir", help="Path to custom detections. Defaults to "
+        "standard MOT detections Directory structure should be the default "
+        "MOTChallenge structure: [sequence]/det/det.txt", default=None)
+    parser.add_argument(
+        "--output_dir", help="Output directory. Will be created if it does not"
+        " exist.", default="detections")
+    return parser.parse_args()
+
+
+def main():
+    args = parse_args()
+    encoder = create_box_encoder(args.model, batch_size=32)
+    generate_detections(encoder, args.mot_dir, args.output_dir,
+                        args.detection_dir)
+
+
+if __name__ == "__main__":
+    main()

tracker.py

@@ -0,0 +1,67 @@
+from deep_sort.tracker import Tracker as DeepSortTracker
+from tools import generate_detections as gdet
+from deep_sort import nn_matching
+from deep_sort.detection import Detection
+import numpy as np
+
+
+class Tracker:
+    tracker = None
+    encoder = None
+    tracks = None
+
+    def __init__(self):
+        max_cosine_distance = 0.4
+        nn_budget = None
+
+        encoder_model_filename = 'model_data/mars-small128.pb'
+
+        metric = nn_matching.NearestNeighborDistanceMetric("cosine", max_cosine_distance, nn_budget)
+        self.tracker = DeepSortTracker(metric)
+        self.encoder = gdet.create_box_encoder(encoder_model_filename, batch_size=1)
+
+    def update(self, frame, detections):
+
+        if len(detections) == 0:
+            self.tracker.predict()
+            self.tracker.update([])  
+            self.update_tracks()
+            return
+
+        bboxes = np.asarray([d[:-2] for d in detections])
+        bboxes[:, 2:] = bboxes[:, 2:] - bboxes[:, 0:2]
+        scores = [d[-1] for d in detections]
+        class_ids = [d[-2] for d in detections]
+        features = self.encoder(frame, bboxes)
+
+        dets = []
+        for bbox_id, bbox in enumerate(bboxes):
+            dets.append(Detection(bbox, scores[bbox_id], class_ids[bbox_id], features[bbox_id]))
+
+        self.tracker.predict()
+        self.tracker.update(dets)
+        self.update_tracks()
+
+    def update_tracks(self):
+        tracks = []
+        for track in self.tracker.tracks:
+            if not track.is_confirmed() or track.time_since_update > 1:
+                continue
+            bbox = track.to_tlbr()
+            class_id = track.get_class()
+            id = track.track_id
+
+            tracks.append(Track(id, bbox, class_id))
+
+        self.tracks = tracks
+
+
+class Track:
+    track_id = None
+    bbox = None
+    class_id = None
+
+    def __init__(self, id, bbox, class_id):
+        self.track_id = id
+        self.bbox = bbox
+        self.class_id = class_id

utils.py

@@ -0,0 +1,22 @@
+ID2LABEL = {0: "person", 1: "bicycle", 2:"car", 3:"motocycle", 5:"bus", 7:"truck"} 
+PALETTE = (2 ** 11 - 1, 2 ** 15 - 1, 2 ** 20 - 1)
+
+MODEL_PATH = "model_data/yolov8m.pt"
+
+# MODEL_PATH = "model_data/best.pt"
+
+AUTHEN_ACCOUNT = 'accountService.json'
+
+def compute_color_for_labels(label):
+
+    if label == 0: # person
+        color = (85, 45, 255)
+    elif label == 2: # Car
+        color = (222, 82, 175)
+    elif label == 3: # Motobike
+        color = (0, 204, 255)
+    elif label == 5: # Bus
+        color = (0, 149, 255)
+    else:
+        color = [int((p * (label ** 2 - label + 1)) % 255) for p in PALETTE]
+    return tuple(color)