diff --git a/frigate/ptz/autotrack.py b/frigate/ptz/autotrack.py index 6b9a85a1ad..cd0edec684 100644 --- a/frigate/ptz/autotrack.py +++ b/frigate/ptz/autotrack.py @@ -585,26 +585,36 @@ class PtzAutoTracker(threading.Thread): def _predict_movement_time(self, camera, pan, tilt): combined_movement = abs(pan) + abs(tilt) - input_data = np.array([self.intercept[camera], combined_movement]) + input_data = np.array([1, combined_movement]) 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]), + 1, + ) + + # 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 @@ -744,6 +734,8 @@ class PtzAutoTracker(threading.Thread): continue else: + pan_tilt_times = None + if ( camera_config.onvif.autotracking.zooming == ZoomingModeEnum.relative @@ -755,17 +747,28 @@ class PtzAutoTracker(threading.Thread): await self._wait_until_stopped(camera, metrics) + # _zoom_absolute resets the metrics timestamps + pan_tilt_times = ( + metrics.start_time.value, + metrics.stop_time.value, + ) + if zoom > 0 and metrics.zoom_level.value != zoom: await self.onvif._zoom_absolute(camera, zoom, 1) await self._wait_until_stopped(camera, metrics) + start_time, stop_time = pan_tilt_times or ( + metrics.start_time.value, + metrics.stop_time.value, + ) + if camera_config.onvif.autotracking.movement_weights: logger.debug( f"{camera}: Predicted movement time: {self._predict_movement_time(camera, pan, tilt)}" ) logger.debug( - f"{camera}: Actual movement time: {metrics.stop_time.value - metrics.start_time.value}" + f"{camera}: Actual movement time: {stop_time - start_time}" ) # save metrics for better estimate calculations @@ -781,8 +784,8 @@ class PtzAutoTracker(threading.Thread): { "pan": pan, "tilt": tilt, - "start_timestamp": metrics.start_time.value, - "end_timestamp": metrics.stop_time.value, + "start_timestamp": start_time, + "end_timestamp": stop_time, } ) @@ -845,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 @@ -863,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( @@ -891,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), ] diff --git a/frigate/test/test_ptz_autotrack.py b/frigate/test/test_ptz_autotrack.py index f06f38c702..fa84f0a672 100644 --- a/frigate/test/test_ptz_autotrack.py +++ b/frigate/test/test_ptz_autotrack.py @@ -11,11 +11,14 @@ KeyError on the autotracker thread or silently keep the wrong state: running its motion estimator for a camera that could not autotrack. """ +import asyncio import unittest -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock + +import numpy as np from frigate.camera import PTZMetrics -from frigate.config import FrigateConfig +from frigate.config import FrigateConfig, ZoomingModeEnum from frigate.config.camera.updater import CameraConfigUpdateEnum from frigate.ptz.autotrack import PtzAutoTracker, calculate_max_target_box @@ -123,6 +126,118 @@ class TestAutotrackerDisable(unittest.TestCase): self.assertIs(payload, autotracking) +class TestAutotrackerPrediction(unittest.TestCase): + def test_movement_time_uses_the_fitted_intercept_unchanged(self) -> None: + tracker = _make_tracker() + tracker.intercept = {CAMERA: 0.5} + tracker.move_coefficients = {CAMERA: [0.3, 0.8]} + + self.assertAlmostEqual(tracker._predict_movement_time(CAMERA, 0.25, 0.25), 0.7) + + def test_shrinking_box_keeps_a_positive_target_box(self) -> None: + # 60 px shrinking 6 px/frame reaches zero after 2 s at 5 fps + target_box = self._target_box_for((0, 0, 60, 60), [3, 3, -3, -3], 2.0) + + self.assertGreater(target_box, 0) + self.assertTrue(np.isfinite(target_box)) + + 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: + return self._target_box_for((900, 400, 1000, 600), velocity, predicted_time) + + def _target_box_for( + self, + box: tuple[int, int, int, int], + velocity: list[float], + predicted_time: float, + ) -> float: + tracker = _make_tracker() + tracker.tracked_object_history = {CAMERA: [{"box": box}]} + 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 TestAutotrackerMoveQueue(unittest.IsolatedAsyncioTestCase): + async def test_absolute_zoom_does_not_overwrite_the_pan_tilt_sample(self) -> None: + tracker = _make_tracker() + autotracking = tracker.config.cameras[CAMERA].onvif.autotracking + autotracking.zooming = ZoomingModeEnum.absolute + autotracking.calibrate_on_startup = True + metrics = tracker.ptz_metrics[CAMERA] + + tracker.stop_event = MagicMock() + tracker.stop_event.is_set.side_effect = [False, True] + tracker.move_queues = {CAMERA: asyncio.Queue()} + tracker.move_queue_locks = {CAMERA: asyncio.Lock()} + tracker.intercept = {CAMERA: 0.5} + tracker.move_metrics = {CAMERA: []} + tracker._wait_until_stopped = AsyncMock() + tracker._calculate_move_coefficients = MagicMock() + + async def move_relative(*_args) -> None: + metrics.start_time.value = 100.0 + metrics.stop_time.value = 102.0 + + async def zoom_absolute(*_args) -> None: + metrics.start_time.value = 103.0 + metrics.stop_time.value = 0 + + tracker.onvif._move_relative = AsyncMock(side_effect=move_relative) + tracker.onvif._zoom_absolute = AsyncMock(side_effect=zoom_absolute) + + tracker.move_queues[CAMERA].put_nowait((50.0, 0.5, 0.5, 0.5)) + await tracker._process_move_queue(CAMERA) + + tracker.onvif._zoom_absolute.assert_awaited_once() + self.assertEqual( + tracker.move_metrics[CAMERA], + [ + { + "pan": 0.5, + "tilt": 0.5, + "start_timestamp": 100.0, + "end_timestamp": 102.0, + } + ], + ) + + class TestMaxTargetBox(unittest.TestCase): def test_follows_zoom_factor(self) -> None: self.assertAlmostEqual(calculate_max_target_box(0.5), 0.6**2)