verify the frame for embeddings object updates

Tracked object events only carried the object's own frame_time, which stays at its last detection when the object isn't detected in a frame, so embeddings couldn't check the slot it read and could crop face or plate data from a frame that had already been overwritten. Events now carry the time of the frame they name, and embeddings passes it to `get()`.
This commit is contained in:
Josh Hawkins
2026-10-11 23:00:46 -05:00
parent 91fbde81a0
commit d807bafd49
6 changed files with 124 additions and 16 deletions
+4 -4
View File
@@ -404,7 +404,7 @@ class CameraState:
self.send_mqtt_snapshot(new_obj, object_type)
for c in self.callbacks["start"]:
c(self.name, new_obj, frame_name)
c(self.name, new_obj, frame_name, frame_time)
for id in updated_ids:
updated_obj = tracked_objects[id]
@@ -416,7 +416,7 @@ class CameraState:
if autotracker_update or significant_update:
for c in self.callbacks["autotrack"]:
c(self.name, updated_obj, frame_name)
c(self.name, updated_obj, frame_name, frame_time)
if thumb_update and current_frame is not None:
# ensure this frame is stored in the cache
@@ -466,7 +466,7 @@ class CameraState:
):
# call event handlers
for c in self.callbacks["update"]:
c(self.name, updated_obj, frame_name)
c(self.name, updated_obj, frame_name, frame_time)
updated_obj.last_published = frame_time
# send MQTT snapshot when object first enters a required zone,
@@ -488,7 +488,7 @@ class CameraState:
removed_obj.obj_data["end_time"] = frame_time
logger.debug(f"{self.name}: end callback for object {id}")
for c in self.callbacks["end"]:
c(self.name, removed_obj, frame_name)
c(self.name, removed_obj, frame_name, frame_time)
# TODO: can i switch to looking this up and only changing when an event ends?
# maintain best objects
+18 -2
View File
@@ -8,7 +8,16 @@ from .zmq_proxy import Publisher, Subscriber
class EventUpdatePublisher(
Publisher[tuple[EventTypeEnum, EventStateEnum, str | None, str, dict[str, Any]]]
Publisher[
tuple[
EventTypeEnum,
EventStateEnum,
str | None,
str,
dict[str, Any],
float | None,
]
]
):
"""Publishes events (objects, audio, manual)."""
@@ -19,7 +28,14 @@ class EventUpdatePublisher(
def publish(
self,
payload: tuple[EventTypeEnum, EventStateEnum, str | None, str, dict[str, Any]],
payload: tuple[
EventTypeEnum,
EventStateEnum,
str | None,
str,
dict[str, Any],
float | None,
],
sub_topic: str = "",
) -> None:
super().publish(payload, sub_topic)
+2 -4
View File
@@ -517,7 +517,7 @@ class EmbeddingMaintainer(threading.Thread):
if update is None:
return
source_type, event_type, camera, frame_name, data = update
source_type, event_type, camera, frame_name, data, frame_time = update
logger.debug(
f"Received update - source_type: {source_type}, camera: {camera}, data label: {data.get('label') if data else 'None'}"
@@ -546,7 +546,7 @@ class EmbeddingMaintainer(threading.Thread):
# Create our own thumbnail based on the bounding box and the frame time
try:
yuv_frame = self.frame_manager.get(
frame_name, camera_config.frame_shape_yuv
frame_name, camera_config.frame_shape_yuv, frame_time
)
except FileNotFoundError:
logger.debug(f"Frame {frame_name} not found for camera {camera}")
@@ -583,8 +583,6 @@ class EmbeddingMaintainer(threading.Thread):
PostProcessDataEnum.tracked_object,
)
self.frame_manager.close(frame_name)
def _process_finalized(self) -> None:
"""Process the end of an event."""
while True:
+1 -1
View File
@@ -94,7 +94,7 @@ class EventProcessor(threading.Thread):
if update == None:
continue
source_type, event_type, camera, _, event_data = update
source_type, event_type, camera, _, event_data, _ = update
logger.debug(
f"Event received: {source_type} {event_type} {camera} {event_data['id']}"
@@ -7,6 +7,7 @@ the old, smaller segment.
"""
import queue
import threading
import unittest
from multiprocessing import Event, Value
from types import SimpleNamespace
@@ -14,6 +15,9 @@ from unittest.mock import MagicMock, patch
import numpy as np
from frigate.embeddings.maintainer import EmbeddingMaintainer
from frigate.events.types import EventStateEnum, EventTypeEnum
from frigate.track.object_processing import TrackedObjectProcessor
from frigate.util.image import SharedMemoryFrameManager
from frigate.video.ffmpeg import capture_frames
@@ -351,5 +355,80 @@ class TestCaptureFrames(unittest.TestCase):
self.assertEqual(self.times_during_read[1][0], 0)
class TestTrackedObjectEventsCarryFrameTime(unittest.TestCase):
"""Tracked object events carry the time of the frame they name, which can
differ from the object's own frame_time when it wasn't detected in that
frame."""
def _callbacks(self) -> tuple[dict, MagicMock]:
processor = TrackedObjectProcessor.__new__(TrackedObjectProcessor)
processor.config = SimpleNamespace(cameras={"cam": SimpleNamespace(name="cam")})
processor.event_sender = MagicMock()
processor.dispatcher = MagicMock()
processor.ptz_autotracker_thread = MagicMock()
processor.camera_states = {}
processor.camera_states_lock = threading.Lock()
processor.frame_manager = MagicMock()
processor.should_save_snapshot = MagicMock(return_value=False)
processor.should_retain_recording = MagicMock(return_value=False)
callbacks: dict = {}
with patch("frigate.track.object_processing.CameraState") as state_cls:
state_cls.return_value.on.side_effect = lambda name, fn: callbacks.update(
{name: fn}
)
processor.create_camera_state("cam")
return callbacks, processor.event_sender
def test_start_update_end_publish_frame_time(self) -> None:
callbacks, sender = self._callbacks()
obj = MagicMock()
obj.to_dict.return_value = {"id": "1", "frame_time": 10.0}
obj.obj_data = {"frame_time": 10.0}
obj.previous = {"false_positive": True}
obj.false_positive = True
callbacks["start"]("cam", obj, "cam_frame3", 12.5)
callbacks["update"]("cam", obj, "cam_frame4", 13.0)
callbacks["end"]("cam", obj, "cam_frame5", 13.5)
payloads = [c.args[0] for c in sender.publish.call_args_list]
self.assertEqual(
[(p[1], p[3], p[5]) for p in payloads],
[
(EventStateEnum.start, "cam_frame3", 12.5),
(EventStateEnum.update, "cam_frame4", 13.0),
(EventStateEnum.end, "cam_frame5", 13.5),
],
)
class TestEmbeddingsObjectUpdateVerifiesFrame(unittest.TestCase):
def test_object_update_reads_frame_at_event_frame_time(self) -> None:
maintainer = EmbeddingMaintainer.__new__(EmbeddingMaintainer)
maintainer.config = SimpleNamespace(
cameras={"cam": SimpleNamespace(frame_shape_yuv=(6, 4))},
semantic_search=SimpleNamespace(enabled=False),
)
maintainer.event_subscriber = MagicMock()
maintainer.event_subscriber.check_for_update.return_value = (
EventTypeEnum.tracked_object,
EventStateEnum.update,
"cam",
"cam_frame4",
{"id": "1", "label": "person", "frame_time": 10.0},
13.0,
)
maintainer.realtime_processors = [MagicMock()]
maintainer.post_processors = []
maintainer.frame_manager = MagicMock()
maintainer._process_updates()
maintainer.frame_manager.get.assert_called_once_with("cam_frame4", (6, 4), 13.0)
maintainer.frame_manager.close.assert_not_called()
if __name__ == "__main__":
unittest.main()
+20 -5
View File
@@ -118,7 +118,9 @@ class TrackedObjectProcessor(threading.Thread):
def create_camera_state(self, camera: str) -> None:
"""Creates a new camera state."""
def start(camera: str, obj: TrackedObject, frame_name: str) -> None:
def start(
camera: str, obj: TrackedObject, frame_name: str, frame_time: float
) -> None:
self.event_sender.publish(
(
EventTypeEnum.tracked_object,
@@ -126,10 +128,13 @@ class TrackedObjectProcessor(threading.Thread):
camera,
frame_name,
obj.to_dict(),
frame_time,
)
)
def update(camera: str, obj: TrackedObject, frame_name: str) -> None:
def update(
camera: str, obj: TrackedObject, frame_name: str, frame_time: float
) -> None:
obj.has_snapshot = self.should_save_snapshot(
camera_state.camera_config, obj
)
@@ -149,13 +154,18 @@ class TrackedObjectProcessor(threading.Thread):
camera,
frame_name,
obj.to_dict(),
frame_time,
)
)
def autotrack(camera: str, obj: TrackedObject, frame_name: str) -> None:
def autotrack(
camera: str, obj: TrackedObject, frame_name: str, frame_time: float
) -> None:
self.ptz_autotracker_thread.autotrack_object(camera, obj)
def end(camera: str, obj: TrackedObject, frame_name: str) -> None:
def end(
camera: str, obj: TrackedObject, frame_name: str, frame_time: float
) -> None:
# populate has_snapshot
obj.has_snapshot = self.should_save_snapshot(
camera_state.camera_config, obj
@@ -186,6 +196,7 @@ class TrackedObjectProcessor(threading.Thread):
camera,
frame_name,
obj.to_dict(),
frame_time,
)
)
@@ -599,6 +610,7 @@ class TrackedObjectProcessor(threading.Thread):
"type": source_type,
"draw": draw,
},
None,
)
)
@@ -664,6 +676,7 @@ class TrackedObjectProcessor(threading.Thread):
"recognized_license_plate": plate,
"recognized_license_plate_score": score,
},
None,
)
)
@@ -692,6 +705,7 @@ class TrackedObjectProcessor(threading.Thread):
None,
"",
{"id": event_id, "end_time": end_time},
None,
)
)
@@ -713,13 +727,14 @@ class TrackedObjectProcessor(threading.Thread):
def force_end_all_events(self, camera: str, camera_state: CameraState) -> None:
"""Ends all active events on camera when disabling."""
last_frame_name = camera_state.previous_frame_id
last_frame_time = camera_state.current_frame_time
for obj_id, obj in list(camera_state.tracked_objects.items()):
if "end_time" not in obj.obj_data:
logger.debug(f"Camera {camera} disabled, ending active event {obj_id}")
obj.obj_data["end_time"] = datetime.datetime.now().timestamp()
# end callbacks
for callback in camera_state.callbacks["end"]:
callback(camera, obj, last_frame_name)
callback(camera, obj, last_frame_name, last_frame_time)
# camera activity callbacks
for callback in camera_state.callbacks["camera_activity"]: