fix double event publishes for stationary objects with attributes

This commit is contained in:
Josh Hawkins 2026-05-07 12:59:16 -05:00
parent d8905e0dfa
commit b03959e45c
2 changed files with 25 additions and 22 deletions

View File

@ -330,7 +330,12 @@ class TrackedObject:
if self.obj_data["position_changes"] != obj_data["position_changes"]:
significant_change = True
if self.obj_data["attributes"] != obj_data["attributes"]:
# disappearance of a per-frame attribute can be caused by detection
# skipping the object on a frame (stationary objects on non-interval
# frames), so only flag when a new attribute label appears
prev_labels = {a["label"] for a in self.obj_data["attributes"]}
curr_labels = {a["label"] for a in obj_data["attributes"]}
if curr_labels - prev_labels:
significant_change = True
# if the state changed between stationary and active

View File

@ -438,34 +438,32 @@ def process_frames(
else:
object_tracker.update_frame_times(frame_name, frame_time)
# group the attribute detections based on what label they apply to
attribute_detections: dict[str, list[TrackedObjectAttribute]] = {}
for label, attribute_labels in attributes_map.items():
attribute_detections[label] = [
TrackedObjectAttribute(d)
for d in consolidated_detections
if d[0] in attribute_labels
]
# build detections
detections = {}
for obj in object_tracker.tracked_objects.values():
detections[obj["id"]] = {**obj, "attributes": []}
# find the best object for each attribute to be assigned to
# assign each detected attribute to the best matching object.
# iterate consolidated_detections once so attributes that appear under
# multiple parent labels in attributes_map (e.g. license_plate is in
# both "car" and "motorcycle") are not appended more than once
all_objects: list[dict[str, Any]] = object_tracker.tracked_objects.values()
for attributes in attribute_detections.values():
for attribute in attributes:
filtered_objects = filter(
lambda o: attribute.label in attributes_map.get(o["label"], []),
all_objects,
)
selected_object_id = attribute.find_best_object(filtered_objects)
detected_attributes = [
TrackedObjectAttribute(d)
for d in consolidated_detections
if d[0] in all_attributes
]
for attribute in detected_attributes:
filtered_objects = filter(
lambda o: attribute.label in attributes_map.get(o["label"], []),
all_objects,
)
selected_object_id = attribute.find_best_object(filtered_objects)
if selected_object_id is not None:
detections[selected_object_id]["attributes"].append(
attribute.get_tracking_data()
)
if selected_object_id is not None:
detections[selected_object_id]["attributes"].append(
attribute.get_tracking_data()
)
# debug object tracking
if False: