predict velocity and zoom size for objects approaching the autotracked camera

The velocity check required both box corners to move in the same direction, but the corners of an approaching object diverge, so its velocity was rejected and zeroed and the camera aimed where the object had been. That check is removed, and the existing magnitude, delta, and variance checks still reject noise. The zoom size prediction fit area against epoch time with no intercept, which returned roughly the mean area instead of a forecast. It now scales the target box by the growth the corner velocities predict over the movement time.

frigate/ptz/autotrack.py

@@ -589,22 +589,32 @@ class PtzAutoTracker(threading.Thread):
 
         return np.dot(self.move_coefficients[camera], input_data)
 
-    def _predict_area_after_time(self, camera, time):
-        return np.dot(
-            self.tracked_object_metrics[camera]["area_coefficients"],
-            [self.tracked_object_history[camera][-1]["frame_time"] + time],
-        )
-
     def _predict_target_box(self, camera, predicted_time):
-        target_box = self.tracked_object_metrics[camera]["target_box"]
+        tom = self.tracked_object_metrics[camera]
+        target_box = tom["target_box"]
 
         if not predicted_time:
             return target_box
 
-        frame_shape = self.config.cameras[camera].frame_shape
+        camera_config = self.config.cameras[camera]
+        left, top, right, bottom = self.tracked_object_history[camera][-1]["box"]
+        size = max(right - left, bottom - top)
 
-        return target_box + self._predict_area_after_time(camera, predicted_time) / (
-            frame_shape[0] * frame_shape[1]
+        if size <= 0:
+            return target_box
+
+        # the corner velocities diverge as the object approaches the camera
+        frames = camera_config.detect.fps * predicted_time
+        velocity = tom["velocity"]
+        predicted_size = max(
+            right - left + frames * (velocity[2] - velocity[0]),
+            bottom - top + frames * (velocity[3] - velocity[1]),
+            0,
+        )
+
+        # target_box is the squared size raised to the zoom factor
+        return target_box * (predicted_size / size) ** (
+            2 * camera_config.onvif.autotracking.zoom_factor
         )
 
     def _calculate_tracked_object_metrics(self, camera, obj):
@@ -660,26 +670,6 @@ class PtzAutoTracker(threading.Thread):
 
         filtered_areas = remove_outliers(areas) if len(areas) > 3 else areas
 
-        # Filter entries that are not touching the frame edge
-        filtered_areas_not_touching_edge = [
-            entry
-            for entry in filtered_areas
-            if self._touching_frame_edges(camera, entry["box"]) == 0
-        ]
-
-        # Calculate regression for area change predictions
-        if len(filtered_areas_not_touching_edge):
-            X = np.array(
-                [item["frame_time"] for item in filtered_areas_not_touching_edge]
-            )
-            y = np.array([item["area"] for item in filtered_areas_not_touching_edge])
-
-            tom["area_coefficients"] = np.linalg.lstsq(X.reshape(-1, 1), y, rcond=None)[
-                0
-            ]
-        else:
-            tom["area_coefficients"] = np.array([0])
-
         weights = np.arange(1, len(filtered_areas) + 1)
         weighted_area = np.average(
             [item["area"] for item in filtered_areas], weights=weights
@@ -858,7 +848,6 @@ class PtzAutoTracker(threading.Thread):
         # Thresholds
         x_mags_thresh = camera_width / camera_fps / 2
         y_mags_thresh = camera_height / camera_fps / 2
-        dir_thresh = 0.93
         delta_thresh = 20
         var_thresh = 10
 
@@ -876,24 +865,12 @@ class PtzAutoTracker(threading.Thread):
         stdev_list = np.std(velocities, axis=0)
         high_variances = np.any(stdev_list > var_thresh)
 
-        # Check direction difference
+        # the corners of an object moving toward or away from the camera move
+        # in different directions, so the delta bounds how far they may differ
         velocities = np.round(velocities)
-        invalid_dirs = False
-        if np.all(np.linalg.norm(velocities, axis=1)):
-            cosine_sim = np.dot(velocities[0], velocities[1]) / (
-                np.linalg.norm(velocities[0]) * np.linalg.norm(velocities[1])
-            )
-            dir_thresh = 0.6 if np.all(delta < delta_thresh / 2) else dir_thresh
-            invalid_dirs = cosine_sim < dir_thresh
 
         # Combine
-        invalid = (
-            invalid_x_mags
-            or invalid_y_mags
-            or invalid_dirs
-            or invalid_delta
-            or high_variances
-        )
+        invalid = invalid_x_mags or invalid_y_mags or invalid_delta or high_variances
 
         if invalid:
             logger.debug(
@@ -904,7 +881,6 @@ class PtzAutoTracker(threading.Thread):
                         for var_name, is_invalid in [
                             ("invalid_x_mags", invalid_x_mags),
                             ("invalid_y_mags", invalid_y_mags),
-                            ("invalid_dirs", invalid_dirs),
                             ("invalid_delta", invalid_delta),
                             ("high_variances", high_variances),
                         ]

frigate/test/test_ptz_autotrack.py

@@ -14,6 +14,8 @@ KeyError on the autotracker thread or silently keep the wrong state:
 import unittest
 from unittest.mock import MagicMock
 
+import numpy as np
+
 from frigate.camera import PTZMetrics
 from frigate.config import FrigateConfig
 from frigate.config.camera.updater import CameraConfigUpdateEnum
@@ -123,6 +125,51 @@ class TestAutotrackerDisable(unittest.TestCase):
         self.assertIs(payload, autotracking)
 
 
+class TestAutotrackerPrediction(unittest.TestCase):
+    def test_approaching_object_keeps_its_velocity(self) -> None:
+        # the box grows, so the corners move in opposite directions
+        tracker = _make_tracker()
+        obj = MagicMock()
+        obj.obj_data = {"estimate_velocity": np.array([[-3.0, -4.0], [3.0, 1.0]])}
+
+        valid, velocity = tracker._get_valid_velocity(CAMERA, obj)
+
+        self.assertTrue(valid)
+        self.assertEqual(list(velocity), [-3, -4, 3, 1])
+
+    def test_corners_that_disagree_too_much_are_rejected(self) -> None:
+        tracker = _make_tracker()
+        obj = MagicMock()
+        obj.obj_data = {"estimate_velocity": np.array([[-15.0, 0.0], [15.0, 0.0]])}
+
+        valid, velocity = tracker._get_valid_velocity(CAMERA, obj)
+
+        self.assertFalse(valid)
+        self.assertFalse(np.any(velocity))
+
+    def _target_box(self, velocity: list[float], predicted_time: float) -> float:
+        tracker = _make_tracker()
+        tracker.tracked_object_history = {CAMERA: [{"box": (900, 400, 1000, 600)}]}
+        tracker.tracked_object_metrics = {
+            CAMERA: {"target_box": 0.2, "velocity": np.array(velocity)}
+        }
+
+        return tracker._predict_target_box(CAMERA, predicted_time)
+
+    def test_target_box_grows_with_an_approaching_object(self) -> None:
+        # the 200px height gains 4px per frame at 5 fps for 1 second
+        zoom_factor = 0.3
+
+        self.assertAlmostEqual(
+            self._target_box([-1, -2, 1, 2], 1.0),
+            0.2 * (220 / 200) ** (2 * zoom_factor),
+        )
+
+    def test_target_box_is_unchanged_without_growth_or_time(self) -> None:
+        self.assertAlmostEqual(self._target_box([5, 0, 5, 0], 1.0), 0.2)
+        self.assertEqual(self._target_box([-1, -2, 1, 2], 0), 0.2)
+
+
 class TestMaxTargetBox(unittest.TestCase):
     def test_follows_zoom_factor(self) -> None:
         self.assertAlmostEqual(calculate_max_target_box(0.5), 0.6**2)