Adding input shift

[BartvanMarrewijk]

Dear Roboflow,

First of all many thanks for making this github open source.
I am using the trackers in a conveyor setting to track objects over time. However, since the conveyor belt speed is not constant both and it is full of objects >200 Bytetrack and SortTracker fail. It incorrectly matches the objects.

However, by adding a simple input U vector, the shift in x or y position both trackers work fine. The shift of the conveyor can be easily determined by for example opencv matched (cv2.matchTemplate)

    def predict(self, u: np.ndarray| None = None) -> None:
        """
        Predict the next state of the bounding box (applies the state transition).
        """
        # Predict state
        if u is not None:
            # Ensure u has the correct shape (n, 1)
            if u.ndim == 1:
                u = u.reshape(-1, 1)
            self.state = (self.F @ self.state).astype(np.float32) + (self.B @ u).astype(np.float32)
        else:
            self.state = (self.F @ self.state).astype(np.float32)
        # Predict error covariance
        self.P = (self.F @ self.P @ self.F.T + self.Q).astype(np.float32)

        # Increase time since update
        self.time_since_update += 1

I think this would be a nice feature, let me know what you think. Maybe a simple wrapper can already help.

[SkalskiP]

Hi @BartvanMarrewijk 👋🏻 Thanks a lot for your interest in Trackers.

Would you be able to share an example video that shows the issue?

The fix you used works, but it feels more like a workaround. We want Trackers to handle cases like this out of the box. If you can share a sample, we can look into a more general and reusable solution.

[BartvanMarrewijk]

Hereby a link to the video with default (Sort) and the custom implementation with input from the template matcher.
https://filesender.surf.nl/?s=download&token=381e1e47-9ab2-4afd-8f5e-737fec7bd37b
(uploading directly to github did not work).

I understand that preferably it handles cases like this out of the box, but with many objects, and a large shift default trackers in my opinion do not work, or I am doing something wrong. The input can be relatively easy determined using cv2. for a more concrete example, please have a look at my fork:
https://github.com/VISNet-framework/trackers/blob/wur_devel/main_example_gar.py

[BartvanMarrewijk]

sorry, another remark. Another example, for 3D data might be interesting as well. In these situations, objects at the back move slowly with respect to objects in the foreground, so tracking them on xyz location + optional box information might better than boxes only.

[SkalskiP]

Hi @BartvanMarrewijk 👋🏻

The reason we lose tracks is that both SORT and ByteTrack use IoU to associate objects between consecutive frames. In your case, the FPS of the video is so low that the tracked objects jump from one position to another between frames. As a result, the overlap is too small and the tracker cannot match them correctly.

Why is the video running at such a low FPS? Is there a way to increase the input video FPS?

[BartvanMarrewijk]

@SkalskiP
Thank you for your time.

I do fully agree that the quality of the video is poor, low image quality and low fps.
However, I don't think you can always or want to improve the FPS. For example, if you are running a setup with multiple camera's, sometimes it is not possible to run with a high fps. A high FPS also means that your deep learning network has to process all those frames, which is not always desirable, GPU load model size etc.

In industrial setup, with a conveyor belt i think tracking by using the input "u" in a kalman filter makes sense. Yes, it is not out of the box, however, it has to possibility to run with a low FPS, enabling a higher throughput and or better deep learning model.

[roel-klein]

I think this would be nice to add.

By default, ByteTrack and SORT use independent Kalman filters for each object, but in many use-cases it is necessary to add global motion compensation.

This global motion can already be known beforehand, or be estimated in a variety of ways.
For example, ORB/SIFT feature matching, optical flow, or cross-correlation based, most already available within opencv.
Examples of this are also included in BOT-SORT (https://github.com/NirAharon/BoT-SORT/blob/main/tracker/gmc.py) and Ultralytics (https://docs.ultralytics.com/reference/trackers/utils/gmc/).

So I think to keep this modular, it would be nice to let the users give an optional global state update in the tracker.update methods.

Some functions or examples could be included for the global motion estimation, but in general this problem does not have a one-size-fits-all solution.

[SkalskiP]

We’re already working on BoT-SORT here: #232.

That’s why I don’t want to merge solutions like this into trackers right away. I’d rather wait until the BoT-SORT work is further along. Then we might be able to connect these ideas and end up with a cleaner approach.

[github-actions]

This issue has not had activity in 28 days. If this is still relevant, please add context or a minimal reproducible example. Inactive issues are deprioritized over time.

[AlexBodner]

@BartvanMarrewijk your trick with the input vector looks like an interesting take, but Im afraid that it could be very overfitted to each scenario. Can you share how does your full pipeline work (to adjust u vector)? Do you think this new feature #403 can replace this? Or maybe even setting another prior for the KF

[BartvanMarrewijk]

Dear @AlexBodner, sorry for the late reply.
Not sure how it replace the #403 feature, because if the pixel shift is large I suspect all iou based methods will fail if there is no compensation for camera shift. The trick especially works in conveyor belt settings, so top down view, you can use cv2.templateplate matching for this or ORB. You can have a look at the code below how it works.

Note that the trick is even better for 3D detections and instead of using the bounding box uses a 3D location XYZ (XYZ) , and use the Y shift of the conveyor for example. I hope that in the near future we can add Detection3D to the supervision.

from pathlib import Path

import cv2
import natsort
import numpy as np
import pandas as pd
import supervision as sv
from supervision.detection.core import Detections

from trackers import CustomSORTTracker
from trackers.utils.match_template import estimate_shift_match_template

def estimate_shift_match_template(
        bottom_incoming_image : npt.NDArray,
        top_stitched_image: npt.NDArray,
        starting_roi_xyxy: list[int],
        max_shift : int,
        method = cv2.TM_CCOEFF,
        movement: str = "y-axis", # or x-axis
        direction: str = "decrease" # decrease
    ) -> tuple[int, npt.NDArray]:
    """
    Estimates the shift in pixels from top_stitched_image to bottom_incoming_image.
    Arguments:
        - bottom_incoming_image : shifted image to align, grayscale np.uint8
        - top_stitched_image : starting image to align with, grayscale np.uint8
        - starting_roi_xyxy: bounding box of the bottom_stitched_image to use for align.
        - max_shift : maximum vertical shift in pixels
    Returns:
        pixelshift : Estimated number pixels of that should be added from incoming image
        errors : For each pixelshift, the sum of the MAE.
    """
    roi_x1 = starting_roi_xyxy[0]
    roi_y1 = starting_roi_xyxy[1]
    roi_x2 = starting_roi_xyxy[2]
    roi_y2 = starting_roi_xyxy[3]

    if movement=="y-axis":
        height = bottom_incoming_image.shape[0]
        max_shift_x = 0
        max_shift_y = min(max_shift, height)

        if roi_y1 < max_shift_y:
            warnings.warn(
                f"The region of interest y-position {roi_y1} is smaller than "
                f"maximum shift {max_shift}. Adjusting y-position to "
                f"{max_shift}, to make it possible to check previous pixel "
                f"lines. Adjusting either max_shift or still_box_xyxy is "
                f"recommended to prevent possible runtimeerrors.",
                stacklevel=2,
            )
            # further shift starting roi y-position if necessary
            # to make sure we can look at the max_shift pixels before it
            roi_y2 += max_shift_y - roi_y1
            roi_y1 += max_shift_y - roi_y1
            if roi_y2 > height:
                raise RuntimeError("Bounding box exceeded image border, reduce max "
                                   "shift or roi box size")

        # convert roi indices to be negative, relative to image end
        # this way we can use the same roi indices for a longer stitched image
        # and a shorter incoming image
        roi_y1 -= bottom_incoming_image.shape[0]
        roi_y2 -= bottom_incoming_image.shape[0]

    elif movement=="x-axis":
        width = bottom_incoming_image.shape[1]
        max_shift_y = 0
        max_shift_x = min(max_shift, width)

        if roi_x1 < max_shift_x:
            warnings.warn(
                f"The region of interest x-position {roi_x1} is smaller than "
                f"maximum shift {max_shift}. Adjusting x-position to "
                f"{max_shift}, to make it possible to check previous pixel "
                f"lines. Adjusting either max_shift or still_box_xyxy is "
                f"recommended to prevent possible runtimeerrors.",
                stacklevel=2,
            )
            # further shift starting roi y-position if necessary
            # to make sure we can look at the max_shift pixels before it
            roi_x2 += max_shift_x - roi_x1
            roi_x1 += max_shift_x - roi_x1
            if roi_x2 > width:
                raise RuntimeError("Bounding box exceeded image border, reduce max "
                                   "shift or roi box size")

        # convert roi indices to be negative, relative to image end
        # this way we can use the same roi indices for a longer stitched image
        # and a shorter incoming image
        roi_x1 -= bottom_incoming_image.shape[1]
        roi_x2 -= bottom_incoming_image.shape[1]

    else:
        KeyError()

    if direction=="decrease":
        # Apply template Matching
        search = bottom_incoming_image[roi_y1-max_shift_y:roi_y2,
                                       roi_x1-max_shift_x:roi_x2]
        index_order = -1
    elif direction=="increase":
        search = bottom_incoming_image[roi_y1:roi_y2+max_shift_y,
                                       roi_x1:roi_x2+max_shift_x]
        index_order = 1


    template = top_stitched_image[roi_y1:roi_y2, roi_x1:roi_x2]
    res = cv2.matchTemplate(search, template, method)
    if movement=="x-axis":
        res = res.T

    # If the method is TM_SQDIFF or TM_SQDIFF_NORMED, lower is better (error metric)
    # for other, higher is better(similarity metric)
    if method in [cv2.TM_SQDIFF, cv2.TM_SQDIFF_NORMED]:
        pixelshift = int(np.argmin(res[::index_order, 0])) * 1*index_order
    else:
        pixelshift = int(np.argmax(res[::index_order, 0])) * 1*index_order

    # import matplotlib.pyplot as plt
    # cv2.imwrite("search.png", search)
    # cv2.imwrite("searchT.png", template)
    # plt.plot(res[::index_order, 0])
    # plt.show()
    return pixelshift, res


tracker = CustomSORTTracker(
    lost_track_buffer=30,
    frame_rate=30,
    track_activation_threshold=0.25,
    minimum_consecutive_frames=3,
    minimum_iou_threshold=0.2 # for demo low
)

main_folder = Path("test/data/belt_tulip/")

input_files = natsort.natsorted(main_folder.glob("*.json"))

box_annotator = sv.BoxAnnotator()
label_annotator = sv.LabelAnnotator()
mask_annotator = sv.MaskAnnotator()

# https://supervision.roboflow.com/0.27.0/notebooks/count-objects-crossing-the-line/#process-video
START = sv.Point(550, 432)
END = sv.Point(550, 0)

line_zone = sv.LineZone(start=START, end=END
                        , minimum_crossing_threshold=2,
                        triggering_anchors=[sv.Position.CENTER])

line_zone_annotator = sv.LineZoneAnnotator(
    thickness=1,
    text_thickness=1,
    text_scale=1,
    text_orient_to_line=True)

prev_img = None
for i, annot_name in enumerate(input_files):
## replace lines with other annotation format
    detections = Detections.from_darwin(
                    json_name=annot_name,
                    with_masks=True,
                    classes=["Potato"],
                    skip_unknown_classes=True,
                    with_track_ids=False,
                    # with_ellipse_as=False,
                )

    frame_bgr = cv2.imread(str(annot_name.parent / (annot_name.stem + ".png")))
    next_img = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2GRAY)

    if i==0:
        tmp_shift = 0
        shift = tmp_shift
    else:
        # if df.iloc[i-1]["pixelshift"]*-1!=0:
        #     print("debugging")
        shift, res = estimate_shift_match_template(bottom_incoming_image=next_img,
                                            top_stitched_image=prev_img,
                                            starting_roi_xyxy=[300, 40, 800, 375],
                                            max_shift=110,
                                            movement="x-axis", # or x-axis
                                            direction= "decrease" # decrease
        )
        # shift = 0

        print("calculated shift", shift, annot_name.name)

    detections = tracker.update(detections, u=np.array([shift, 0, shift, 0]))

    annotated_frame = box_annotator.annotate(frame_bgr, detections)

    # disable dbecause was time consuming
    # annotated_frame = mask_annotator.annotate(annotated_frame, detections[0 ])
    # Convert tracker_id to string and handle empty tracker_id
    if detections.tracker_id is None or len(detections.tracker_id) == 0:
        # Handle empty tracker or None
        detections.tracker_id = ["Unknown"] * len(detections)
    else:
        detections.tracker_id = [
            str(t_id) if t_id is not None else "Unknown"
            for t_id in detections.tracker_id
        ]
    annotated_frame = label_annotator.annotate(annotated_frame,
                                               detections,
                                               labels=detections.tracker_id)

    if shift!=0:
        line_zone.trigger(detections)
    line_zone_annotator.annotate(annotated_frame, line_counter=line_zone)

    cv2.imshow("CustomFilter", annotated_frame)
    if cv2.waitKey(1) & 0xFF == ord("q"):
        break
    prev_img = next_img.copy()

print("finished")

[github-actions]

This issue has not had activity in 28 days. If this is still relevant, please add context or a minimal reproducible example. Inactive issues are deprioritized over time.