mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-11 09:12:48 +03:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9ef7c22777 |
@@ -1 +1 @@
|
|||||||
scikit-build == 0.18.*
|
scikit-build == 0.19.*
|
||||||
|
|||||||
@@ -1,41 +0,0 @@
|
|||||||
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)
|
|
||||||
@@ -1,136 +0,0 @@
|
|||||||
"""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,9 +180,48 @@ class NorfairTracker(ObjectTracker):
|
|||||||
}
|
}
|
||||||
|
|
||||||
self.trackers: dict[str, dict[str, Tracker]] = {}
|
self.trackers: dict[str, dict[str, Tracker]] = {}
|
||||||
self.default_tracker: dict[str, Tracker] = {}
|
# Handle static trackers
|
||||||
self.tracker_selection: tuple[bool | None, tuple[str, ...]] | None = None
|
for obj_type, tracker_config in self.object_type_configs.items():
|
||||||
self.sync_trackers()
|
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]
|
||||||
|
),
|
||||||
|
}
|
||||||
|
|
||||||
if self.camera_config.onvif.autotracking.enabled:
|
if self.camera_config.onvif.autotracking.enabled:
|
||||||
self.ptz_motion_estimator = PtzMotionEstimator(
|
self.ptz_motion_estimator = PtzMotionEstimator(
|
||||||
@@ -218,86 +257,18 @@ class NorfairTracker(ObjectTracker):
|
|||||||
|
|
||||||
return Tracker(**tracker_params)
|
return Tracker(**tracker_params)
|
||||||
|
|
||||||
def _tracker_selection(self) -> tuple[bool | None, tuple[str, ...]]:
|
def get_tracker(self, object_type: str) -> Tracker:
|
||||||
autotracking = self.camera_config.onvif.autotracking
|
"""Get the appropriate tracker based on object type and camera mode."""
|
||||||
ptz_labels = tuple(
|
mode = (
|
||||||
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"
|
"ptz"
|
||||||
if self.camera_config.onvif.autotracking.enabled_in_config
|
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"
|
else "static"
|
||||||
)
|
)
|
||||||
|
if object_type in self.trackers:
|
||||||
def get_tracker(self, object_type: str) -> Tracker:
|
return self.trackers[object_type][mode]
|
||||||
"""Get the tracker that match_and_update feeds this label's detections to."""
|
return self.default_tracker[mode]
|
||||||
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:
|
def register(self, track_id: str, obj: dict[str, Any]) -> None:
|
||||||
rand_id = "".join(random.choices(string.ascii_lowercase + string.digits, k=6))
|
rand_id = "".join(random.choices(string.ascii_lowercase + string.digits, k=6))
|
||||||
@@ -563,7 +534,7 @@ class NorfairTracker(ObjectTracker):
|
|||||||
points = np.array([[obj[2][0], obj[2][1]], [obj[2][2], obj[2][3]]])
|
points = np.array([[obj[2][0], obj[2][1]], [obj[2][2], obj[2][3]]])
|
||||||
|
|
||||||
embedding = None
|
embedding = None
|
||||||
if self.camera_config.onvif.autotracking.enabled and yuv_frame is not None:
|
if self.camera_config.onvif.autotracking.enabled:
|
||||||
embedding = get_histogram(
|
embedding = get_histogram(
|
||||||
yuv_frame, obj[2][0], obj[2][1], obj[2][2], obj[2][3]
|
yuv_frame, obj[2][0], obj[2][1], obj[2][2], obj[2][3]
|
||||||
)
|
)
|
||||||
@@ -616,7 +587,12 @@ class NorfairTracker(ObjectTracker):
|
|||||||
default_detections.extend(dets)
|
default_detections.extend(dets)
|
||||||
|
|
||||||
# Update default tracker with untracked detections
|
# Update default tracker with untracked detections
|
||||||
tracked_objects = self.default_tracker[self.default_mode()].update(
|
mode = (
|
||||||
|
"ptz"
|
||||||
|
if self.camera_config.onvif.autotracking.enabled_in_config
|
||||||
|
else "static"
|
||||||
|
)
|
||||||
|
tracked_objects = self.default_tracker[mode].update(
|
||||||
detections=default_detections, coord_transformations=coord_transformations
|
detections=default_detections, coord_transformations=coord_transformations
|
||||||
)
|
)
|
||||||
all_tracked_objects.extend(tracked_objects)
|
all_tracked_objects.extend(tracked_objects)
|
||||||
|
|||||||
+3
-36
@@ -1241,42 +1241,9 @@ def get_image_from_recording(
|
|||||||
return image_data
|
return image_data
|
||||||
|
|
||||||
|
|
||||||
def get_histogram(
|
def get_histogram(image, x_min, y_min, x_max, y_max):
|
||||||
image: np.ndarray, x_min: int, y_min: int, x_max: int, y_max: int
|
image_bgr = cv2.cvtColor(image, cv2.COLOR_YUV2BGR_I420)
|
||||||
) -> np.ndarray:
|
image_bgr = image_bgr[y_min:y_max, x_min:x_max]
|
||||||
"""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(
|
hist = cv2.calcHist(
|
||||||
[image_bgr], [0, 1, 2], None, [8, 8, 8], [0, 256, 0, 256, 0, 256]
|
[image_bgr], [0, 1, 2], None, [8, 8, 8], [0, 256, 0, 256, 0, 256]
|
||||||
|
|||||||
@@ -302,7 +302,6 @@ def process_frames(
|
|||||||
motion_detector.autotracking_enabled = (
|
motion_detector.autotracking_enabled = (
|
||||||
camera_config.onvif.autotracking.enabled
|
camera_config.onvif.autotracking.enabled
|
||||||
)
|
)
|
||||||
object_tracker.sync_trackers()
|
|
||||||
|
|
||||||
if (
|
if (
|
||||||
not camera_enabled
|
not camera_enabled
|
||||||
|
|||||||
Reference in New Issue
Block a user