diff --git a/frigate/test/test_get_histogram.py b/frigate/test/test_get_histogram.py new file mode 100644 index 0000000000..f7fb33737d --- /dev/null +++ b/frigate/test/test_get_histogram.py @@ -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) diff --git a/frigate/test/test_norfair_tracker_modes.py b/frigate/test/test_norfair_tracker_modes.py new file mode 100644 index 0000000000..effc264eed --- /dev/null +++ b/frigate/test/test_norfair_tracker_modes.py @@ -0,0 +1,146 @@ +"""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_track_list_edit_while_disabled_keeps_objects(self): + tracker = self.make_tracker(enabled=False) + self.run_frames(tracker, 10) + ids = set(tracker.tracked_objects) + + self.apply_onvif_update(tracker, _config(enabled=False, track=["car"])) + 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), + ) diff --git a/frigate/track/norfair_tracker.py b/frigate/track/norfair_tracker.py index e69e1305a3..ec807ee93f 100644 --- a/frigate/track/norfair_tracker.py +++ b/frigate/track/norfair_tracker.py @@ -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,91 @@ 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 + + if not autotracking.enabled_in_config: + return (False, ()) + + ptz_labels = tuple( + label + for label in self.ptz_object_type_configs + if label in autotracking.track + ) + return (True, 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: + """Pick the default tracker from saved config, not the runtime toggle.""" + 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 +568,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 +621,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) diff --git a/frigate/util/image.py b/frigate/util/image.py index b403f9750e..37c2e0c895 100644 --- a/frigate/util/image.py +++ b/frigate/util/image.py @@ -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] diff --git a/frigate/video/detect.py b/frigate/video/detect.py index 676b2bb4c5..d1048e4270 100644 --- a/frigate/video/detect.py +++ b/frigate/video/detect.py @@ -302,6 +302,7 @@ def process_frames( motion_detector.autotracking_enabled = ( camera_config.onvif.autotracking.enabled ) + object_tracker.sync_trackers() if ( not camera_enabled