mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-11 09:12:48 +03:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3f7900b257 |
@@ -0,0 +1,41 @@
|
||||
import unittest
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
from frigate.util.image import get_histogram
|
||||
|
||||
|
||||
def reference_histogram(image, x_min, y_min, x_max, y_max):
|
||||
bgr = cv2.cvtColor(image, cv2.COLOR_YUV2BGR_I420)[y_min:y_max, x_min:x_max]
|
||||
hist = cv2.calcHist([bgr], [0, 1, 2], None, [8, 8, 8], [0, 256, 0, 256, 0, 256])
|
||||
return cv2.normalize(hist, hist).flatten()
|
||||
|
||||
|
||||
class TestGetHistogram(unittest.TestCase):
|
||||
def setUp(self):
|
||||
rng = np.random.default_rng(0)
|
||||
self.frame = rng.integers(0, 255, (720 * 3 // 2, 1280), np.uint8)
|
||||
|
||||
def test_matches_full_frame_conversion_on_even_box(self):
|
||||
box = (100, 200, 400, 600)
|
||||
np.testing.assert_array_equal(
|
||||
get_histogram(self.frame, *box), reference_histogram(self.frame, *box)
|
||||
)
|
||||
|
||||
def test_odd_box_widens_to_even_edges(self):
|
||||
np.testing.assert_array_equal(
|
||||
get_histogram(self.frame, 101, 201, 399, 599),
|
||||
reference_histogram(self.frame, 100, 200, 400, 600),
|
||||
)
|
||||
|
||||
def test_box_is_clamped_to_frame(self):
|
||||
np.testing.assert_array_equal(
|
||||
get_histogram(self.frame, -10, -10, 5000, 5000),
|
||||
reference_histogram(self.frame, 0, 0, 1280, 720),
|
||||
)
|
||||
|
||||
def test_empty_box_returns_zeros(self):
|
||||
hist = get_histogram(self.frame, 50, 50, 50, 50)
|
||||
self.assertEqual(hist.shape, (512,))
|
||||
self.assertEqual(hist.sum(), 0)
|
||||
@@ -0,0 +1,136 @@
|
||||
"""Tracker selection when autotracking config changes at runtime."""
|
||||
|
||||
import unittest
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import numpy as np
|
||||
|
||||
from frigate.camera import PTZMetrics
|
||||
from frigate.config import FrigateConfig
|
||||
from frigate.track.norfair_tracker import NorfairTracker
|
||||
|
||||
CAMERA = "ptz_cam"
|
||||
BOX = (400, 200, 500, 500)
|
||||
|
||||
|
||||
def _config(enabled: bool, track: list[str] | None = None) -> FrigateConfig:
|
||||
autotracking: dict = {"enabled": enabled, "required_zones": ["zone"]}
|
||||
|
||||
if track is not None:
|
||||
autotracking["track"] = track
|
||||
|
||||
return FrigateConfig(
|
||||
**{
|
||||
"mqtt": {"enabled": False},
|
||||
"cameras": {
|
||||
CAMERA: {
|
||||
"ffmpeg": {
|
||||
"inputs": [
|
||||
{"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}
|
||||
]
|
||||
},
|
||||
"detect": {"width": 1280, "height": 720},
|
||||
"zones": {"zone": {"coordinates": "0,0,1,0,1,1,0,1"}},
|
||||
"onvif": {"host": "10.0.0.1", "autotracking": autotracking},
|
||||
}
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class TestTrackerSelection(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.frame_time = 1000.0
|
||||
|
||||
def make_tracker(self, enabled: bool) -> NorfairTracker:
|
||||
camera_config = _config(enabled).cameras[CAMERA]
|
||||
tracker = NorfairTracker(camera_config, PTZMetrics())
|
||||
tracker.frame_manager = MagicMock()
|
||||
tracker.frame_manager.get.return_value = np.zeros(
|
||||
camera_config.frame_shape_yuv, dtype=np.uint8
|
||||
)
|
||||
tracker.ptz_motion_estimator = MagicMock()
|
||||
tracker.ptz_motion_estimator.motion_estimator.return_value = None
|
||||
return tracker
|
||||
|
||||
def apply_onvif_update(self, tracker: NorfairTracker, config: FrigateConfig):
|
||||
"""Apply an onvif config update the way the camera process does."""
|
||||
tracker.camera_config.onvif = config.cameras[CAMERA].onvif
|
||||
tracker.sync_trackers()
|
||||
|
||||
def run_frames(self, tracker: NorfairTracker, count: int, label="person"):
|
||||
for _ in range(count):
|
||||
self.frame_time += 0.2
|
||||
tracker.match_and_update(
|
||||
"frame",
|
||||
self.frame_time,
|
||||
[(label, 0.9, BOX, 30000, 0.33, (0, 0, 640, 640))],
|
||||
)
|
||||
|
||||
def test_disabling_autotracking_falls_back_to_static_tracker(self):
|
||||
tracker = self.make_tracker(enabled=True)
|
||||
self.run_frames(tracker, 10)
|
||||
self.assertIs(tracker.get_tracker("person"), tracker.trackers["person"]["ptz"])
|
||||
|
||||
self.apply_onvif_update(tracker, _config(enabled=False))
|
||||
self.run_frames(tracker, 10)
|
||||
|
||||
self.assertIs(tracker.get_tracker("person"), tracker.default_tracker["static"])
|
||||
self.assertEqual(len(tracker.tracked_objects), 1)
|
||||
|
||||
def test_enabling_autotracking_uses_ptz_tracker(self):
|
||||
tracker = self.make_tracker(enabled=False)
|
||||
self.run_frames(tracker, 10)
|
||||
self.assertIs(tracker.get_tracker("person"), tracker.default_tracker["static"])
|
||||
|
||||
self.apply_onvif_update(tracker, _config(enabled=True))
|
||||
self.run_frames(tracker, 10)
|
||||
|
||||
self.assertIs(tracker.get_tracker("person"), tracker.trackers["person"]["ptz"])
|
||||
self.assertEqual(len(tracker.tracked_objects), 1)
|
||||
|
||||
def test_removing_label_from_autotracking_keeps_tracking(self):
|
||||
tracker = self.make_tracker(enabled=True)
|
||||
self.run_frames(tracker, 10)
|
||||
|
||||
self.apply_onvif_update(tracker, _config(enabled=True, track=["car"]))
|
||||
self.run_frames(tracker, 10)
|
||||
|
||||
self.assertNotIn("person", tracker.trackers)
|
||||
self.assertIs(tracker.get_tracker("person"), tracker.default_tracker["ptz"])
|
||||
self.assertEqual(len(tracker.tracked_objects), 1)
|
||||
|
||||
def test_unrelated_onvif_update_keeps_objects(self):
|
||||
tracker = self.make_tracker(enabled=True)
|
||||
self.run_frames(tracker, 10)
|
||||
ids = set(tracker.tracked_objects)
|
||||
|
||||
config = _config(enabled=True)
|
||||
config.cameras[CAMERA].onvif.password = "changed"
|
||||
self.apply_onvif_update(tracker, config)
|
||||
self.run_frames(tracker, 5)
|
||||
|
||||
self.assertEqual(set(tracker.tracked_objects), ids)
|
||||
|
||||
def test_runtime_toggle_keeps_objects(self):
|
||||
# MQTT and the autotracker only flip enabled, never enabled_in_config
|
||||
tracker = self.make_tracker(enabled=True)
|
||||
self.run_frames(tracker, 10)
|
||||
ids = set(tracker.tracked_objects)
|
||||
|
||||
tracker.camera_config.onvif.autotracking.enabled = False
|
||||
tracker.sync_trackers()
|
||||
self.run_frames(tracker, 5)
|
||||
|
||||
self.assertEqual(set(tracker.tracked_objects), ids)
|
||||
|
||||
def test_unlisted_label_uses_the_default_tracker_that_holds_it(self):
|
||||
tracker = self.make_tracker(enabled=True)
|
||||
self.run_frames(tracker, 10, label="dog")
|
||||
|
||||
default = tracker.get_tracker("dog")
|
||||
self.assertIs(default, tracker.default_tracker["ptz"])
|
||||
self.assertEqual(
|
||||
{str(o.global_id) for o in default.tracked_objects},
|
||||
set(tracker.track_id_map),
|
||||
)
|
||||
@@ -180,48 +180,9 @@ class NorfairTracker(ObjectTracker):
|
||||
}
|
||||
|
||||
self.trackers: dict[str, dict[str, Tracker]] = {}
|
||||
# Handle static trackers
|
||||
for obj_type, tracker_config in self.object_type_configs.items():
|
||||
if obj_type in self.camera_config.objects.track:
|
||||
if obj_type not in self.trackers:
|
||||
self.trackers[obj_type] = {}
|
||||
self.trackers[obj_type]["static"] = self._create_tracker(
|
||||
obj_type, tracker_config
|
||||
)
|
||||
|
||||
# Handle PTZ trackers
|
||||
for obj_type, tracker_config in self.ptz_object_type_configs.items():
|
||||
if (
|
||||
obj_type in self.camera_config.onvif.autotracking.track
|
||||
and self.camera_config.onvif.autotracking.enabled_in_config
|
||||
):
|
||||
if obj_type not in self.trackers:
|
||||
self.trackers[obj_type] = {}
|
||||
self.trackers[obj_type]["ptz"] = self._create_tracker(
|
||||
obj_type, tracker_config
|
||||
)
|
||||
|
||||
# Initialize default trackers
|
||||
self.default_tracker = {
|
||||
"static": Tracker(
|
||||
distance_function=frigate_distance,
|
||||
distance_threshold=self.default_tracker_config[ # type: ignore[arg-type]
|
||||
"distance_threshold"
|
||||
],
|
||||
initialization_delay=self.detect_config.min_initialized,
|
||||
hit_counter_max=self.detect_config.max_disappeared, # type: ignore[arg-type]
|
||||
filter_factory=self.default_tracker_config["filter_factory"], # type: ignore[arg-type]
|
||||
),
|
||||
"ptz": Tracker(
|
||||
distance_function=frigate_distance,
|
||||
distance_threshold=self.default_ptz_tracker_config[
|
||||
"distance_threshold"
|
||||
], # type: ignore[arg-type]
|
||||
initialization_delay=self.detect_config.min_initialized,
|
||||
hit_counter_max=self.detect_config.max_disappeared, # type: ignore[arg-type]
|
||||
filter_factory=self.default_ptz_tracker_config["filter_factory"], # type: ignore[arg-type]
|
||||
),
|
||||
}
|
||||
self.default_tracker: dict[str, Tracker] = {}
|
||||
self.tracker_selection: tuple[bool | None, tuple[str, ...]] | None = None
|
||||
self.sync_trackers()
|
||||
|
||||
if self.camera_config.onvif.autotracking.enabled:
|
||||
self.ptz_motion_estimator = PtzMotionEstimator(
|
||||
@@ -257,18 +218,86 @@ class NorfairTracker(ObjectTracker):
|
||||
|
||||
return Tracker(**tracker_params)
|
||||
|
||||
def get_tracker(self, object_type: str) -> Tracker:
|
||||
"""Get the appropriate tracker based on object type and camera mode."""
|
||||
mode = (
|
||||
def _tracker_selection(self) -> tuple[bool | None, tuple[str, ...]]:
|
||||
autotracking = self.camera_config.onvif.autotracking
|
||||
ptz_labels = tuple(
|
||||
label
|
||||
for label in self.ptz_object_type_configs
|
||||
if label in autotracking.track
|
||||
)
|
||||
return (autotracking.enabled_in_config, ptz_labels)
|
||||
|
||||
def sync_trackers(self) -> None:
|
||||
"""Rebuild the trackers when the config that selects them has changed.
|
||||
|
||||
The camera process receives onvif config updates at runtime, so which
|
||||
labels use a PTZ tracker can change after startup. Rebuilding drops
|
||||
norfair state, so an update that leaves the selection alone is a no-op.
|
||||
"""
|
||||
selection = self._tracker_selection()
|
||||
|
||||
if selection == self.tracker_selection:
|
||||
return
|
||||
|
||||
if self.tracker_selection is not None:
|
||||
logger.debug(
|
||||
"%s: autotracking changed, rebuilding trackers", self.camera_name
|
||||
)
|
||||
|
||||
self.tracker_selection = selection
|
||||
ptz_enabled, ptz_labels = selection
|
||||
self.trackers = {}
|
||||
|
||||
for obj_type, tracker_config in self.object_type_configs.items():
|
||||
if obj_type in self.camera_config.objects.track:
|
||||
self.trackers.setdefault(obj_type, {})["static"] = self._create_tracker(
|
||||
obj_type, tracker_config
|
||||
)
|
||||
|
||||
if ptz_enabled:
|
||||
for obj_type, tracker_config in self.ptz_object_type_configs.items():
|
||||
if obj_type in ptz_labels:
|
||||
self.trackers.setdefault(obj_type, {})["ptz"] = (
|
||||
self._create_tracker(obj_type, tracker_config)
|
||||
)
|
||||
|
||||
self.default_tracker = {
|
||||
"static": Tracker(
|
||||
distance_function=frigate_distance,
|
||||
distance_threshold=self.default_tracker_config[ # type: ignore[arg-type]
|
||||
"distance_threshold"
|
||||
],
|
||||
initialization_delay=self.detect_config.min_initialized,
|
||||
hit_counter_max=self.detect_config.max_disappeared, # type: ignore[arg-type]
|
||||
filter_factory=self.default_tracker_config["filter_factory"], # type: ignore[arg-type]
|
||||
),
|
||||
"ptz": Tracker(
|
||||
distance_function=frigate_distance,
|
||||
distance_threshold=self.default_ptz_tracker_config[
|
||||
"distance_threshold"
|
||||
], # type: ignore[arg-type]
|
||||
initialization_delay=self.detect_config.min_initialized,
|
||||
hit_counter_max=self.detect_config.max_disappeared, # type: ignore[arg-type]
|
||||
filter_factory=self.default_ptz_tracker_config["filter_factory"], # type: ignore[arg-type]
|
||||
),
|
||||
}
|
||||
|
||||
def default_mode(self) -> str:
|
||||
return (
|
||||
"ptz"
|
||||
if self.camera_config.onvif.autotracking.enabled_in_config
|
||||
and object_type in self.camera_config.onvif.autotracking.track
|
||||
and object_type in self.ptz_object_type_configs.keys()
|
||||
else "static"
|
||||
)
|
||||
if object_type in self.trackers:
|
||||
return self.trackers[object_type][mode]
|
||||
return self.default_tracker[mode]
|
||||
|
||||
def get_tracker(self, object_type: str) -> Tracker:
|
||||
"""Get the tracker that match_and_update feeds this label's detections to."""
|
||||
trackers = self.trackers.get(object_type)
|
||||
|
||||
if trackers is None:
|
||||
return self.default_tracker[self.default_mode()]
|
||||
|
||||
# sync_trackers only creates a ptz tracker for labels the config selects
|
||||
return trackers["ptz"] if "ptz" in trackers else trackers["static"]
|
||||
|
||||
def register(self, track_id: str, obj: dict[str, Any]) -> None:
|
||||
rand_id = "".join(random.choices(string.ascii_lowercase + string.digits, k=6))
|
||||
@@ -534,7 +563,7 @@ class NorfairTracker(ObjectTracker):
|
||||
points = np.array([[obj[2][0], obj[2][1]], [obj[2][2], obj[2][3]]])
|
||||
|
||||
embedding = None
|
||||
if self.camera_config.onvif.autotracking.enabled:
|
||||
if self.camera_config.onvif.autotracking.enabled and yuv_frame is not None:
|
||||
embedding = get_histogram(
|
||||
yuv_frame, obj[2][0], obj[2][1], obj[2][2], obj[2][3]
|
||||
)
|
||||
@@ -587,12 +616,7 @@ class NorfairTracker(ObjectTracker):
|
||||
default_detections.extend(dets)
|
||||
|
||||
# Update default tracker with untracked detections
|
||||
mode = (
|
||||
"ptz"
|
||||
if self.camera_config.onvif.autotracking.enabled_in_config
|
||||
else "static"
|
||||
)
|
||||
tracked_objects = self.default_tracker[mode].update(
|
||||
tracked_objects = self.default_tracker[self.default_mode()].update(
|
||||
detections=default_detections, coord_transformations=coord_transformations
|
||||
)
|
||||
all_tracked_objects.extend(tracked_objects)
|
||||
|
||||
+36
-3
@@ -1241,9 +1241,42 @@ def get_image_from_recording(
|
||||
return image_data
|
||||
|
||||
|
||||
def get_histogram(image, x_min, y_min, x_max, y_max):
|
||||
image_bgr = cv2.cvtColor(image, cv2.COLOR_YUV2BGR_I420)
|
||||
image_bgr = image_bgr[y_min:y_max, x_min:x_max]
|
||||
def get_histogram(
|
||||
image: np.ndarray, x_min: int, y_min: int, x_max: int, y_max: int
|
||||
) -> np.ndarray:
|
||||
"""Return a normalized 8x8x8 BGR histogram of a box in an I420 frame.
|
||||
|
||||
The box is cropped from each YUV plane before color conversion, so the
|
||||
cost depends on the box size rather than the frame size. Box edges are
|
||||
widened to even coordinates to keep chroma alignment.
|
||||
"""
|
||||
height = image.shape[0] * 2 // 3
|
||||
width = image.shape[1]
|
||||
x_min = max(0, x_min // 2 * 2)
|
||||
y_min = max(0, y_min // 2 * 2)
|
||||
x_max = min(width, (x_max + 1) // 2 * 2)
|
||||
y_max = min(height, (y_max + 1) // 2 * 2)
|
||||
|
||||
if x_max - x_min < 2 or y_max - y_min < 2:
|
||||
return np.zeros(512, np.float32)
|
||||
|
||||
flat = image.reshape(-1)
|
||||
y_size = height * width
|
||||
uv_size = y_size // 4
|
||||
y_plane = flat[:y_size].reshape(height, width)
|
||||
u_plane = flat[y_size : y_size + uv_size].reshape(height // 2, width // 2)
|
||||
v_plane = flat[y_size + uv_size : y_size + 2 * uv_size].reshape(
|
||||
height // 2, width // 2
|
||||
)
|
||||
|
||||
crop = np.concatenate(
|
||||
(
|
||||
y_plane[y_min:y_max, x_min:x_max].ravel(),
|
||||
u_plane[y_min // 2 : y_max // 2, x_min // 2 : x_max // 2].ravel(),
|
||||
v_plane[y_min // 2 : y_max // 2, x_min // 2 : x_max // 2].ravel(),
|
||||
)
|
||||
).reshape((y_max - y_min) * 3 // 2, x_max - x_min)
|
||||
image_bgr = cv2.cvtColor(crop, cv2.COLOR_YUV2BGR_I420)
|
||||
|
||||
hist = cv2.calcHist(
|
||||
[image_bgr], [0, 1, 2], None, [8, 8, 8], [0, 256, 0, 256, 0, 256]
|
||||
|
||||
@@ -302,6 +302,7 @@ def process_frames(
|
||||
motion_detector.autotracking_enabled = (
|
||||
camera_config.onvif.autotracking.enabled
|
||||
)
|
||||
object_tracker.sync_trackers()
|
||||
|
||||
if (
|
||||
not camera_enabled
|
||||
|
||||
Reference in New Issue
Block a user