Use a rolling average of iou to determine if an object is no longer stationary (#9381)

* Use a rolling average of iou to determine if an object is no longer stationary

* Use different box variation to designate when an object is stationary on debug

* In progress

* Use average of boxes instead of average of iou

* Update frigate/track/norfair_tracker.py

Co-authored-by: Blake Blackshear <blake@frigate.video>

---------

Co-authored-by: Blake Blackshear <blake@frigate.video>

frigate/object_processing.py

@@ -489,6 +489,10 @@ class CameraState:
             # draw the bounding boxes on the frame
             for obj in tracked_objects.values():
                 if obj["frame_time"] == frame_time:
+                    if obj["stationary"]:
+                        color = (220, 220, 220)
+                        thickness = 1
+                    else:
                         thickness = 2
                         color = self.config.model.colormap[obj["label"]]
                 else:

frigate/track/norfair_tracker.py

@@ -17,10 +17,15 @@ from frigate.ptz.autotrack import PtzMotionEstimator
 from frigate.track import ObjectTracker
 from frigate.types import PTZMetricsTypes
 from frigate.util.image import intersection_over_union
+from frigate.util.object import average_boxes
 
 logger = logging.getLogger(__name__)
 
 
+THRESHOLD_STATIONARY_IOU_AVERAGE = 0.6
+MAX_STATIONARY_HISTORY = 10
+
+
 # Normalizes distance from estimate relative to object size
 # Other ideas:
 # - if estimates are inaccurate for first N detections, compare with last_detection (may be fine)
@@ -74,6 +79,7 @@ class NorfairTracker(ObjectTracker):
         self.untracked_object_boxes: list[list[int]] = []
         self.disappeared = {}
         self.positions = {}
+        self.stationary_box_history: dict[str, list[list[int, int, int, int]]] = {}
         self.camera_config = config
         self.detect_config = config.detect
         self.ptz_metrics = ptz_metrics
@@ -127,6 +133,7 @@ class NorfairTracker(ObjectTracker):
             "xmax": self.detect_config.width,
             "ymax": self.detect_config.height,
         }
+        self.stationary_box_history[id] = []
 
     def deregister(self, id, track_id):
         del self.tracked_objects[id]
@@ -138,22 +145,24 @@ class NorfairTracker(ObjectTracker):
 
     # tracks the current position of the object based on the last N bounding boxes
     # returns False if the object has moved outside its previous position
-    def update_position(self, id, box):
+    def update_position(self, id: str, box: list[int, int, int, int]):
         position = self.positions[id]
-        position_box = (
+        self.stationary_box_history[id].append(box)
-            position["xmin"],
+
-            position["ymin"],
+        if len(self.stationary_box_history[id]) > MAX_STATIONARY_HISTORY:
-            position["xmax"],
+            self.stationary_box_history[id] = self.stationary_box_history[id][
-            position["ymax"],
+                -MAX_STATIONARY_HISTORY:
+            ]
+
+        avg_iou = intersection_over_union(
+            box, average_boxes(self.stationary_box_history[id])
         )
 
         xmin, ymin, xmax, ymax = box
 
-        iou = intersection_over_union(position_box, box)
-
         # if the iou drops below the threshold
         # assume the object has moved to a new position and reset the computed box
-        if iou < 0.6:
+        if avg_iou < THRESHOLD_STATIONARY_IOU_AVERAGE:
             self.positions[id] = {
                 "xmins": [xmin],
                 "ymins": [ymin],
@@ -220,6 +229,7 @@ class NorfairTracker(ObjectTracker):
             ):
                 self.tracked_objects[id]["position_changes"] += 1
             self.tracked_objects[id]["motionless_count"] = 0
+            self.stationary_box_history[id] = []
 
         self.tracked_objects[id].update(obj)
 

frigate/util/object.py

@@ -323,6 +323,22 @@ def reduce_boxes(boxes, iou_threshold=0.0):
     return [tuple(c) for c in clusters]
 
 
+def average_boxes(boxes: list[list[int, int, int, int]]) -> list[int, int, int, int]:
+    """Return a box that is the average of a list of boxes."""
+    x_mins = []
+    y_mins = []
+    x_max = []
+    y_max = []
+
+    for box in boxes:
+        x_mins.append(box[0])
+        y_mins.append(box[1])
+        x_max.append(box[2])
+        y_max.append(box[3])
+
+    return [np.mean(x_mins), np.mean(y_mins), np.mean(x_max), np.mean(y_max)]
+
+
 def intersects_any(box_a, boxes):
     for box in boxes:
         if box_overlaps(box_a, box):