mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-11-12 00:23:01 +03:00
Autotracking tweaks (#24935)
* fix autotracking movement time prediction and recorded move times Movement time prediction multiplied the fitted intercept by the saved zero-step calibration time instead of 1, so predictions didn't match the regression. In absolute zoom mode, a pan/tilt followed by a zoom recorded the zoom's timestamps against the pan/tilt distance, because _zoom_absolute resets the PTZ start and stop times. The pan/tilt times are now captured before the zoom runs. * 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. * keep predicted target box positive and add regression tests A shrinking object's predicted size could reach zero, so the relative zoom ratio divided by zero and sent a NaN zoom to the camera. The predicted size is now floored at 1 px. Tests also cover the movement time intercept and the pan/tilt timestamps that absolute zoom overwrites.
This commit is contained in:
+40
-51
@@ -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),
|
||||
]
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user