Merge 0db2a34426 into f065cc8642

frigate/test/test_obects.py

@@ -1,22 +1,6 @@
-import random
 import unittest
 
-import numpy as np
-
 from frigate.track.tracked_object import TrackedObjectAttribute
-from frigate.util.object import average_boxes
-
-
-class TestBoxStatistics(unittest.TestCase):
-    def test_average_boxes_matches_numpy(self) -> None:
-        rng = random.Random(0)
-        for _ in range(5000):
-            boxes = [
-                [rng.randint(0, 4000) for _ in range(4)]
-                for _ in range(rng.randint(1, 10))
-            ]
-            expected = [float(np.mean([b[i] for b in boxes])) for i in range(4)]
-            self.assertEqual(average_boxes(boxes), expected)
 
 
 class TestAttribute(unittest.TestCase):

frigate/test/test_stationary_classifier.py

@@ -1,39 +0,0 @@
-"""Tests for stationary object classification thresholds."""
-
-import unittest
-
-from frigate.track.stationary_classifier import (
-    DEFAULT_OBJECT_THRESHOLDS,
-    DYNAMIC_OBJECT_THRESHOLDS,
-    NON_STATIONARY_OBJECT_THRESHOLDS,
-    STATIONARY_OBJECT_THRESHOLDS,
-    StationaryThresholds,
-    get_stationary_threshold,
-)
-
-
-class TestStationaryThresholds(unittest.TestCase):
-    def test_known_labels_return_expected_singletons(self) -> None:
-        self.assertIs(get_stationary_threshold("package"), STATIONARY_OBJECT_THRESHOLDS)
-        self.assertIs(get_stationary_threshold("car"), DYNAMIC_OBJECT_THRESHOLDS)
-        self.assertIs(
-            get_stationary_threshold("license_plate"),
-            NON_STATIONARY_OBJECT_THRESHOLDS,
-        )
-
-    def test_unknown_label_returns_shared_default(self) -> None:
-        # an unknown label must reuse the shared default instance, not allocate
-        # a fresh one on every call (this runs per object per frame)
-        first = get_stationary_threshold("person")
-        second = get_stationary_threshold("dog")
-        self.assertIs(first, DEFAULT_OBJECT_THRESHOLDS)
-        self.assertIs(second, DEFAULT_OBJECT_THRESHOLDS)
-
-    def test_default_matches_a_fresh_instance(self) -> None:
-        # the shared default must be value-equivalent to the previous
-        # per-call StationaryThresholds()
-        self.assertEqual(DEFAULT_OBJECT_THRESHOLDS, StationaryThresholds())
-
-
-if __name__ == "__main__":
-    unittest.main()

frigate/test/test_video.py

@@ -82,24 +82,6 @@ class TestRegion(unittest.TestCase):
 
         assert len(cluster_candidates) == 2
 
-    def test_cluster_candidates_partition_boxes(self):
-        # every box index must appear in exactly one cluster (no box used twice,
-        # none dropped) - the invariant the used-box tracking enforces
-        boxes = [
-            (100, 100, 200, 200),
-            (202, 150, 252, 200),
-            (210, 160, 260, 210),
-            (900, 900, 950, 950),
-            (905, 905, 955, 955),
-        ]
-
-        cluster_candidates = get_cluster_candidates(
-            self.frame_shape, self.min_region_size, boxes
-        )
-
-        assigned = [idx for cluster in cluster_candidates for idx in cluster]
-        self.assertEqual(sorted(assigned), list(range(len(boxes))))
-
     def test_transliterate_to_latin(self):
         self.assertEqual(transliterate_to_latin("frégate"), "fregate")
         self.assertEqual(transliterate_to_latin("utilité"), "utilite")

frigate/track/norfair_tracker.py

@@ -641,11 +641,9 @@ class NorfairTracker(ObjectTracker):
             self.deregister(self.track_id_map[e_id], e_id)
 
         # update list of object boxes that don't have a tracked object yet
-        tracked_object_boxes = {
-            tuple(obj["box"]) for obj in self.tracked_objects.values()
-        }
+        tracked_object_boxes = [obj["box"] for obj in self.tracked_objects.values()]
         self.untracked_object_boxes = [
-            o[2] for o in detections if tuple(o[2]) not in tracked_object_boxes
+            o[2] for o in detections if o[2] not in tracked_object_boxes
         ]
 
     def print_objects_as_table(self, tracked_objects: Sequence) -> None:

frigate/track/stationary_classifier.py

@@ -63,9 +63,6 @@ NON_STATIONARY_OBJECT_THRESHOLDS = StationaryThresholds(
     max_stationary_history=4,
 )
 
-# Default thresholds for any other object label
-DEFAULT_OBJECT_THRESHOLDS = StationaryThresholds()
-
 
 def get_stationary_threshold(label: str) -> StationaryThresholds:
     """Get the stationary thresholds for a given object label."""
@@ -79,7 +76,7 @@ def get_stationary_threshold(label: str) -> StationaryThresholds:
     if label in NON_STATIONARY_OBJECT_THRESHOLDS.objects:
         return NON_STATIONARY_OBJECT_THRESHOLDS
 
-    return DEFAULT_OBJECT_THRESHOLDS
+    return StationaryThresholds()
 
 
 class StationaryMotionClassifier:

frigate/util/object.py

@@ -339,13 +339,18 @@ def reduce_boxes(boxes, iou_threshold=0.0):
 
 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."""
-    n = len(boxes)
-    return [
-        sum(box[0] for box in boxes) / n,
-        sum(box[1] for box in boxes) / n,
-        sum(box[2] for box in boxes) / n,
-        sum(box[3] for box in boxes) / n,
-    ]
+    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 median_of_boxes(boxes: list[list[int, int, int, int]]) -> list[int, int, int, int]:
@@ -396,13 +401,13 @@ def get_cluster_candidates(frame_shape, min_region, boxes):
     # determined by the max_region size minus half the box + 20%
     # TODO: see if we can do this with numpy
     cluster_candidates = []
-    used_boxes = set()
+    used_boxes = []
     # loop over each box
     for current_index, b in enumerate(boxes):
         if current_index in used_boxes:
             continue
         cluster = [current_index]
-        used_boxes.add(current_index)
+        used_boxes.append(current_index)
         cluster_boundary = get_cluster_boundary(b, min_region)
         # find all other boxes that fit inside the boundary
         for compare_index, compare_box in enumerate(boxes):
@@ -431,7 +436,7 @@ def get_cluster_candidates(frame_shape, min_region, boxes):
 
             if should_cluster:
                 cluster.append(compare_index)
-                used_boxes.add(compare_index)
+                used_boxes.append(compare_index)
         cluster_candidates.append(cluster)
 
     # return the unique clusters only
@@ -553,7 +558,6 @@ def reduce_detections(
                 current_detection = sorted_by_area[current_detection_idx]
                 current_label = current_detection[0]
                 current_box = current_detection[2]
-                current_area = area(current_box)
                 overlap = 0
                 for to_check_idx in range(
                     min(current_detection_idx + 1, len(sorted_by_area)),
@@ -564,14 +568,14 @@ def reduce_detections(
                     # if area of current detection / area of check < 5% they should not be compared
                     # this covers cases where a large car parked in a driveway doesn't block detections
                     # of cars in the street behind it
-                    if current_area / area(to_check) < 0.05:
+                    if area(current_box) / area(to_check) < 0.05:
                         continue
 
                     intersect_box = intersection(current_box, to_check)
                     # if % of smaller detection is inside of another detection, consolidate
-                    if intersect_box is not None and area(
-                        intersect_box
-                    ) / current_area > LABEL_CONSOLIDATION_MAP.get(
+                    if intersect_box is not None and area(intersect_box) / area(
+                        current_box
+                    ) > LABEL_CONSOLIDATION_MAP.get(
                         current_label, LABEL_CONSOLIDATION_DEFAULT
                     ):
                         overlap = 1