mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-01 20:36:51 +03:00
Fix detection not correctly being started after an alert ends (#24534)
* Fix detection not correctly being started after an alert ends * Improve coverage * Fix incorrect behavior * Handle unsent thumbs * Add more tests
This commit is contained in:
Nicolas Mowen
+135
-55
@@ -66,6 +66,9 @@ class PendingReviewSegment:
|
||||
self.zones = zones
|
||||
self.audio = audio
|
||||
self.classification_state_changes: list[dict[str, Any]] = []
|
||||
# detection-level activity after the last alert activity, split by the
|
||||
# detection cutoff, these are published when the alert is cut off
|
||||
self.pending_detections: list[PendingReviewSegment] = []
|
||||
self.thumb_time: float | None = None
|
||||
self.last_alert_time: float | None = None
|
||||
self.last_detection_time: float = frame_time
|
||||
@@ -83,6 +86,20 @@ class PendingReviewSegment:
|
||||
CLIPS_DIR, f"review/thumb-{self.camera}-{self.id}.webp"
|
||||
)
|
||||
|
||||
def add_object(self, obj: dict[str, Any], attributes: list[str]) -> None:
|
||||
"""Add a tracked object's label, sub label, and zones to the segment."""
|
||||
if not obj["sub_label"]:
|
||||
self.detections[obj["id"]] = obj["label"]
|
||||
elif obj["sub_label"][0] in attributes:
|
||||
self.detections[obj["id"]] = obj["sub_label"][0]
|
||||
else:
|
||||
self.detections[obj["id"]] = f"{obj['label']}-verified"
|
||||
self.sub_labels[obj["id"]] = obj["sub_label"][0]
|
||||
|
||||
for zone in obj["current_zones"]:
|
||||
if zone not in self.zones:
|
||||
self.zones.append(zone)
|
||||
|
||||
def update_frame(
|
||||
self,
|
||||
camera_config: CameraConfig,
|
||||
@@ -435,9 +452,29 @@ class ReviewSegmentMaintainer(threading.Thread):
|
||||
segment.last_detection_time = now
|
||||
|
||||
prev_data = segment.get_data(False)
|
||||
return self._publish_segment_end(segment, prev_data)
|
||||
end_time = self._publish_segment_end(segment, prev_data)
|
||||
self._publish_pending_detections(segment, None)
|
||||
return end_time
|
||||
return None
|
||||
|
||||
def _publish_pending_detections(
|
||||
self, segment: PendingReviewSegment, ongoing_since: float | None
|
||||
) -> None:
|
||||
"""Publish the detections held while an ended alert was active.
|
||||
|
||||
A detection with activity after ongoing_since stays open, only the
|
||||
latest can. With None every detection is ended, this does not read the
|
||||
camera config since a removed camera is no longer in it.
|
||||
"""
|
||||
for pending in segment.pending_detections:
|
||||
self._activate_segment(pending)
|
||||
self._publish_segment_start(pending)
|
||||
|
||||
if ongoing_since is None or pending.last_detection_time < ongoing_since:
|
||||
self._publish_segment_end(pending, pending.get_data(False))
|
||||
|
||||
segment.pending_detections = []
|
||||
|
||||
def get_manual_event_severity(self, camera: str, label: str) -> SeverityEnum | None:
|
||||
"""Determine the review severity for a manual event label.
|
||||
|
||||
@@ -470,6 +507,55 @@ class ReviewSegmentMaintainer(threading.Thread):
|
||||
self.indefinite_events.pop(camera, None)
|
||||
self.recent_classification_state_changes.pop(camera, None)
|
||||
|
||||
def _track_pending_detection(
|
||||
self,
|
||||
segment: PendingReviewSegment,
|
||||
camera_config: CameraConfig,
|
||||
frame_name: str,
|
||||
frame_time: float,
|
||||
objects: list[dict[str, Any]],
|
||||
) -> None:
|
||||
"""Hold detection-level activity seen after an alert's last alert
|
||||
activity, starting a separate detection when the gap since the previous
|
||||
activity exceeds the detection cutoff."""
|
||||
pending = segment.pending_detections[-1] if segment.pending_detections else None
|
||||
|
||||
if pending is None or frame_time > (
|
||||
pending.last_detection_time + camera_config.review.detections.cutoff_time
|
||||
):
|
||||
pending = PendingReviewSegment(
|
||||
segment.camera,
|
||||
frame_time,
|
||||
SeverityEnum.detection,
|
||||
{},
|
||||
sub_labels={},
|
||||
audio=set(),
|
||||
zones=[],
|
||||
)
|
||||
segment.pending_detections.append(pending)
|
||||
|
||||
pending.last_detection_time = frame_time
|
||||
|
||||
for obj in objects:
|
||||
pending.add_object(obj, self.config.all_attributes)
|
||||
|
||||
if len(objects) <= pending.frame_active_count:
|
||||
return
|
||||
|
||||
try:
|
||||
yuv_frame = self.frame_manager.get(
|
||||
frame_name, camera_config.frame_shape_yuv
|
||||
)
|
||||
except FileNotFoundError:
|
||||
return
|
||||
|
||||
if yuv_frame is None:
|
||||
logger.debug(f"Failed to get frame {frame_name} from SHM")
|
||||
return
|
||||
|
||||
pending.update_frame(camera_config, yuv_frame, objects)
|
||||
self.frame_manager.close(frame_name)
|
||||
|
||||
def update_existing_segment(
|
||||
self,
|
||||
segment: PendingReviewSegment,
|
||||
@@ -505,6 +591,21 @@ class ReviewSegmentMaintainer(threading.Thread):
|
||||
should_update_state = True
|
||||
should_update_image = True
|
||||
|
||||
# alert activity resumed, so the pending detection activity
|
||||
# falls within this alert
|
||||
for pending in segment.pending_detections:
|
||||
segment.detections.update(pending.detections)
|
||||
segment.sub_labels.update(pending.sub_labels)
|
||||
|
||||
for zone in pending.zones:
|
||||
if zone not in segment.zones:
|
||||
segment.zones.append(zone)
|
||||
|
||||
Path(pending.frame_path).unlink(missing_ok=True)
|
||||
should_update_state = True
|
||||
|
||||
segment.pending_detections = []
|
||||
|
||||
if activity.has_activity_category(SeverityEnum.detection):
|
||||
if (
|
||||
segment.last_detection_time is None
|
||||
@@ -512,6 +613,8 @@ class ReviewSegmentMaintainer(threading.Thread):
|
||||
):
|
||||
segment.last_detection_time = frame_time
|
||||
|
||||
pending_objects: list[dict[str, Any]] = []
|
||||
|
||||
for object in activity.get_all_objects():
|
||||
# Alert-level objects should always be added (they extend/upgrade the segment)
|
||||
# Detection-level objects should only be added if:
|
||||
@@ -522,23 +625,22 @@ class ReviewSegmentMaintainer(threading.Thread):
|
||||
|
||||
if not is_alert_object and segment.severity == SeverityEnum.alert:
|
||||
# This is a detection-level object
|
||||
if (
|
||||
segment.last_alert_time is not None
|
||||
and frame_time > segment.last_alert_time
|
||||
):
|
||||
pending_objects.append(object)
|
||||
|
||||
# Only add if it started during the alert's active period
|
||||
if object["start_time"] > segment.last_alert_time:
|
||||
continue
|
||||
|
||||
if not object["sub_label"]:
|
||||
segment.detections[object["id"]] = object["label"]
|
||||
elif object["sub_label"][0] in self.config.all_attributes:
|
||||
segment.detections[object["id"]] = object["sub_label"][0]
|
||||
else:
|
||||
segment.detections[object["id"]] = f"{object['label']}-verified"
|
||||
segment.sub_labels[object["id"]] = object["sub_label"][0]
|
||||
segment.add_object(object, self.config.all_attributes)
|
||||
|
||||
# keep zones up to date
|
||||
if len(object["current_zones"]) > 0:
|
||||
for zone in object["current_zones"]:
|
||||
if zone not in segment.zones:
|
||||
segment.zones.append(zone)
|
||||
if pending_objects:
|
||||
self._track_pending_detection(
|
||||
segment, camera_config, frame_name, frame_time, pending_objects
|
||||
)
|
||||
|
||||
if len(activity.get_all_objects()) > segment.frame_active_count:
|
||||
should_update_state = True
|
||||
@@ -590,50 +692,28 @@ class ReviewSegmentMaintainer(threading.Thread):
|
||||
except FileNotFoundError:
|
||||
return
|
||||
|
||||
if (
|
||||
segment.severity == SeverityEnum.alert
|
||||
and segment.last_alert_time is not None
|
||||
and frame_time
|
||||
> (segment.last_alert_time + camera_config.review.alerts.cutoff_time)
|
||||
):
|
||||
needs_new_detection = (
|
||||
segment.last_detection_time > segment.last_alert_time
|
||||
and (
|
||||
segment.last_detection_time
|
||||
+ camera_config.review.detections.cutoff_time
|
||||
)
|
||||
> frame_time
|
||||
)
|
||||
last_detection_time = segment.last_detection_time
|
||||
|
||||
end_time = self._publish_segment_end(segment, prev_data)
|
||||
|
||||
if needs_new_detection:
|
||||
new_detections: dict[str, str] = {}
|
||||
new_zones = set()
|
||||
|
||||
for o in activity.categorized_objects["detections"]:
|
||||
new_detections[o["id"]] = o["label"]
|
||||
new_zones.update(o["current_zones"])
|
||||
|
||||
if new_detections:
|
||||
new_segment = PendingReviewSegment(
|
||||
segment.camera,
|
||||
end_time,
|
||||
SeverityEnum.detection,
|
||||
new_detections,
|
||||
sub_labels={},
|
||||
audio=set(),
|
||||
zones=list(new_zones),
|
||||
)
|
||||
self._activate_segment(new_segment)
|
||||
self._publish_segment_start(new_segment)
|
||||
new_segment.last_detection_time = last_detection_time
|
||||
elif segment.severity == SeverityEnum.detection and frame_time > (
|
||||
# detection-level activity must not keep an alert open, it continues
|
||||
# in a new detection segment once the alert is cut off
|
||||
if (
|
||||
segment.severity == SeverityEnum.alert
|
||||
and segment.last_alert_time is not None
|
||||
and frame_time
|
||||
> (segment.last_alert_time + camera_config.review.alerts.cutoff_time)
|
||||
):
|
||||
self._publish_segment_end(segment, prev_data)
|
||||
self._publish_pending_detections(
|
||||
segment, frame_time - camera_config.review.detections.cutoff_time
|
||||
)
|
||||
elif (
|
||||
not has_activity
|
||||
and segment.severity == SeverityEnum.detection
|
||||
and frame_time
|
||||
> (
|
||||
segment.last_detection_time
|
||||
+ camera_config.review.detections.cutoff_time
|
||||
):
|
||||
self._publish_segment_end(segment, prev_data)
|
||||
)
|
||||
):
|
||||
self._publish_segment_end(segment, prev_data)
|
||||
|
||||
def check_if_new_segment(
|
||||
self,
|
||||
|
||||
@@ -0,0 +1,857 @@
|
||||
"""Tests for how tracked objects flow into review segments.
|
||||
|
||||
Frames are fed through the maintainer's run loop with mocked subscribers so
|
||||
the dispatch between starting and updating segments is exercised, and the
|
||||
published review updates are checked for the expected alert, detection, or
|
||||
lack of a review item.
|
||||
"""
|
||||
|
||||
import json
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import numpy as np
|
||||
|
||||
from frigate.comms.detections_updater import DetectionTypeEnum
|
||||
from frigate.config import FrigateConfig
|
||||
from frigate.review.maintainer import ReviewSegmentMaintainer
|
||||
|
||||
CAMERA = "front_door"
|
||||
|
||||
BASE_CONFIG = """
|
||||
mqtt:
|
||||
enabled: False
|
||||
record:
|
||||
enabled: True
|
||||
cameras:
|
||||
front_door:
|
||||
ffmpeg:
|
||||
inputs:
|
||||
- path: rtsp://10.0.0.1:554/video
|
||||
roles:
|
||||
- detect
|
||||
detect:
|
||||
width: 640
|
||||
height: 360
|
||||
fps: 5
|
||||
zones:
|
||||
driveway:
|
||||
coordinates: 0,0,320,0,320,360,0,360
|
||||
yard:
|
||||
coordinates: 320,0,640,0,640,360,320,360
|
||||
%s
|
||||
"""
|
||||
|
||||
|
||||
class ReviewFlowTestCase(unittest.TestCase):
|
||||
review_config = ""
|
||||
|
||||
def setUp(self) -> None:
|
||||
self.clips_dir = clips_dir = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(clips_dir.cleanup)
|
||||
clips_patch = patch("frigate.review.maintainer.CLIPS_DIR", clips_dir.name)
|
||||
clips_patch.start()
|
||||
self.addCleanup(clips_patch.stop)
|
||||
|
||||
self.maintainer = self._make_maintainer(self.review_config)
|
||||
|
||||
def _make_maintainer(self, review_config: str) -> ReviewSegmentMaintainer:
|
||||
"""Build a maintainer without invoking __init__ (avoids needing ZMQ
|
||||
sockets and shared memory)."""
|
||||
maintainer = ReviewSegmentMaintainer.__new__(ReviewSegmentMaintainer)
|
||||
threading.Thread.__init__(maintainer)
|
||||
maintainer.config = FrigateConfig.parse_yaml(BASE_CONFIG % review_config)
|
||||
maintainer.active_review_segments = {}
|
||||
maintainer.indefinite_events = {}
|
||||
maintainer.recent_classification_state_changes = {}
|
||||
maintainer.requestor = MagicMock()
|
||||
maintainer.review_publisher = MagicMock()
|
||||
maintainer.config_subscriber = MagicMock()
|
||||
maintainer.config_subscriber.check_for_updates.return_value = {}
|
||||
maintainer.detection_subscriber = MagicMock()
|
||||
maintainer.frame_manager = MagicMock()
|
||||
maintainer.frame_manager.get.side_effect = lambda _name, shape: np.zeros(
|
||||
shape, np.uint8
|
||||
)
|
||||
return maintainer
|
||||
|
||||
@property
|
||||
def stationary_threshold(self) -> int:
|
||||
return self.maintainer.config.cameras[CAMERA].detect.stationary.threshold
|
||||
|
||||
@property
|
||||
def alert_cutoff(self) -> int:
|
||||
return self.maintainer.config.cameras[CAMERA].review.alerts.cutoff_time
|
||||
|
||||
@property
|
||||
def detection_cutoff(self) -> int:
|
||||
return self.maintainer.config.cameras[CAMERA].review.detections.cutoff_time
|
||||
|
||||
def tracked(
|
||||
self,
|
||||
obj_id: str,
|
||||
label: str,
|
||||
frame_time: float,
|
||||
*,
|
||||
start_time: float = 0,
|
||||
zones: list[str] | None = None,
|
||||
stationary: bool = False,
|
||||
loitering: bool = False,
|
||||
moved: bool = True,
|
||||
false_positive: bool = False,
|
||||
sub_label: tuple[str, float] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Build a tracked object as published by the object processor."""
|
||||
return {
|
||||
"id": obj_id,
|
||||
"label": label,
|
||||
"sub_label": sub_label,
|
||||
"frame_time": frame_time,
|
||||
"start_time": start_time,
|
||||
"motionless_count": self.stationary_threshold if stationary else 0,
|
||||
"pending_loitering": loitering,
|
||||
"position_changes": 1 if moved else 0,
|
||||
"false_positive": false_positive,
|
||||
"current_zones": zones or [],
|
||||
"box": (100, 100, 200, 200),
|
||||
}
|
||||
|
||||
def feed(self, *frames: tuple[float, list[dict[str, Any]]]) -> None:
|
||||
"""Run the maintainer loop over the given (frame_time, objects) frames."""
|
||||
queue = [
|
||||
(
|
||||
DetectionTypeEnum.video.value,
|
||||
(CAMERA, f"{CAMERA}_{frame_time}", frame_time, objects, [], []),
|
||||
)
|
||||
for frame_time, objects in frames
|
||||
]
|
||||
self.maintainer.stop_event = threading.Event()
|
||||
|
||||
def next_update(timeout: float) -> Any:
|
||||
if not queue:
|
||||
self.maintainer.stop_event.set()
|
||||
return None
|
||||
|
||||
return queue.pop(0)
|
||||
|
||||
self.maintainer.detection_subscriber.check_for_update.side_effect = next_update
|
||||
self.maintainer.run()
|
||||
|
||||
def reviews(self) -> list[dict[str, Any]]:
|
||||
"""All review updates published on the reviews topic."""
|
||||
return [
|
||||
json.loads(c.args[1])
|
||||
for c in self.maintainer.requestor.send_data.call_args_list
|
||||
if c.args[0] == "reviews"
|
||||
]
|
||||
|
||||
def review_summary(self) -> list[tuple[str, str]]:
|
||||
return [(r["type"], r["after"]["severity"]) for r in self.reviews()]
|
||||
|
||||
def non_update_summary(self) -> list[tuple[str, str]]:
|
||||
return [(t, s) for t, s in self.review_summary() if t != "update"]
|
||||
|
||||
def thumbnails(self) -> set[str]:
|
||||
"""Names of the review thumbnails on disk."""
|
||||
return {t.name for t in Path(self.clips_dir.name, "review").iterdir()}
|
||||
|
||||
def assert_no_review(self) -> None:
|
||||
self.assertEqual(self.reviews(), [])
|
||||
self.assertIsNone(self.maintainer.active_review_segments.get(CAMERA))
|
||||
|
||||
|
||||
class TestReviewSeverity(ReviewFlowTestCase):
|
||||
def test_alert_label_creates_alert(self) -> None:
|
||||
self.feed((1, [self.tracked("p1", "person", 1)]))
|
||||
|
||||
self.assertEqual(self.review_summary(), [("new", "alert")])
|
||||
self.assertEqual(self.reviews()[0]["after"]["data"]["objects"], ["person"])
|
||||
|
||||
def test_non_alert_label_creates_detection(self) -> None:
|
||||
self.feed((1, [self.tracked("d1", "dog", 1)]))
|
||||
|
||||
self.assertEqual(self.review_summary(), [("new", "detection")])
|
||||
self.assertEqual(self.reviews()[0]["after"]["data"]["objects"], ["dog"])
|
||||
|
||||
def test_alert_and_detection_objects_create_single_alert(self) -> None:
|
||||
self.feed((1, [self.tracked("p1", "person", 1), self.tracked("d1", "dog", 1)]))
|
||||
|
||||
self.assertEqual(self.review_summary(), [("new", "alert")])
|
||||
self.assertCountEqual(
|
||||
self.reviews()[0]["after"]["data"]["objects"], ["person", "dog"]
|
||||
)
|
||||
|
||||
def test_no_objects_creates_nothing(self) -> None:
|
||||
self.feed((1, []), (2, []))
|
||||
|
||||
self.assert_no_review()
|
||||
|
||||
|
||||
class TestIgnoredObjects(ReviewFlowTestCase):
|
||||
def test_stationary_object_creates_nothing(self) -> None:
|
||||
self.feed((1, [self.tracked("p1", "person", 1, stationary=True)]))
|
||||
|
||||
self.assert_no_review()
|
||||
|
||||
def test_stationary_loitering_object_creates_alert(self) -> None:
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1, stationary=True, loitering=True)])
|
||||
)
|
||||
|
||||
self.assertEqual(self.review_summary(), [("new", "alert")])
|
||||
|
||||
def test_object_that_never_moved_creates_nothing(self) -> None:
|
||||
self.feed((1, [self.tracked("p1", "person", 1, moved=False)]))
|
||||
|
||||
self.assert_no_review()
|
||||
|
||||
def test_object_not_detected_in_current_frame_creates_nothing(self) -> None:
|
||||
self.feed((2, [self.tracked("p1", "person", 1)]))
|
||||
|
||||
self.assert_no_review()
|
||||
|
||||
def test_false_positive_creates_nothing(self) -> None:
|
||||
self.feed((1, [self.tracked("p1", "person", 1, false_positive=True)]))
|
||||
|
||||
self.assert_no_review()
|
||||
|
||||
|
||||
class TestAlertRequiredZones(ReviewFlowTestCase):
|
||||
review_config = """
|
||||
review:
|
||||
alerts:
|
||||
required_zones: driveway
|
||||
"""
|
||||
|
||||
def test_alert_label_in_required_zone_creates_alert(self) -> None:
|
||||
self.feed((1, [self.tracked("p1", "person", 1, zones=["driveway"])]))
|
||||
|
||||
self.assertEqual(self.review_summary(), [("new", "alert")])
|
||||
self.assertEqual(self.reviews()[0]["after"]["data"]["zones"], ["driveway"])
|
||||
|
||||
def test_alert_label_outside_required_zone_creates_detection(self) -> None:
|
||||
self.feed((1, [self.tracked("p1", "person", 1, zones=["yard"])]))
|
||||
|
||||
self.assertEqual(self.review_summary(), [("new", "detection")])
|
||||
|
||||
def test_alert_label_in_no_zone_creates_detection(self) -> None:
|
||||
self.feed((1, [self.tracked("p1", "person", 1)]))
|
||||
|
||||
self.assertEqual(self.review_summary(), [("new", "detection")])
|
||||
|
||||
def test_detection_upgrades_to_alert_when_object_enters_required_zone(
|
||||
self,
|
||||
) -> None:
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1, zones=["yard"])]),
|
||||
(2, [self.tracked("p1", "person", 2, zones=["driveway"])]),
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
self.review_summary(), [("new", "detection"), ("update", "alert")]
|
||||
)
|
||||
self.assertEqual(
|
||||
self.reviews()[0]["after"]["id"], self.reviews()[1]["after"]["id"]
|
||||
)
|
||||
|
||||
|
||||
class TestDetectionRequiredZones(ReviewFlowTestCase):
|
||||
review_config = """
|
||||
review:
|
||||
detections:
|
||||
required_zones: yard
|
||||
"""
|
||||
|
||||
def test_detection_label_in_required_zone_creates_detection(self) -> None:
|
||||
self.feed((1, [self.tracked("d1", "dog", 1, zones=["yard"])]))
|
||||
|
||||
self.assertEqual(self.review_summary(), [("new", "detection")])
|
||||
|
||||
def test_detection_label_outside_required_zone_creates_nothing(self) -> None:
|
||||
self.feed((1, [self.tracked("d1", "dog", 1, zones=["driveway"])]))
|
||||
|
||||
self.assert_no_review()
|
||||
|
||||
def test_alert_label_ignores_detection_required_zones(self) -> None:
|
||||
self.feed((1, [self.tracked("p1", "person", 1, zones=["driveway"])]))
|
||||
|
||||
self.assertEqual(self.review_summary(), [("new", "alert")])
|
||||
|
||||
def test_activity_outside_required_zone_after_alert_is_not_held(self) -> None:
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1, start_time=1)]),
|
||||
(5, [self.tracked("d1", "dog", 5, start_time=5, zones=["driveway"])]),
|
||||
)
|
||||
self.assertEqual(
|
||||
self.maintainer.active_review_segments[CAMERA].pending_detections, []
|
||||
)
|
||||
|
||||
self.feed((2 + self.alert_cutoff, []))
|
||||
|
||||
self.assertEqual(
|
||||
self.non_update_summary(), [("new", "alert"), ("end", "alert")]
|
||||
)
|
||||
self.assertIsNone(self.maintainer.active_review_segments.get(CAMERA))
|
||||
self.assertEqual(len(self.thumbnails()), 1)
|
||||
|
||||
def test_activity_inside_required_zone_after_alert_is_split_out(self) -> None:
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1, start_time=1)]),
|
||||
(5, [self.tracked("d1", "dog", 5, start_time=5, zones=["yard"])]),
|
||||
(2 + self.alert_cutoff, []),
|
||||
)
|
||||
|
||||
# the dog is past the detection cutoff, so it is ended right away
|
||||
self.assertEqual(
|
||||
self.non_update_summary(),
|
||||
[
|
||||
("new", "alert"),
|
||||
("end", "alert"),
|
||||
("new", "detection"),
|
||||
("end", "detection"),
|
||||
],
|
||||
)
|
||||
detection = self.reviews()[-1]["after"]
|
||||
self.assertEqual(detection["data"]["objects"], ["dog"])
|
||||
self.assertEqual(detection["data"]["zones"], ["yard"])
|
||||
|
||||
|
||||
class TestDetectionLabels(ReviewFlowTestCase):
|
||||
review_config = """
|
||||
review:
|
||||
detections:
|
||||
labels:
|
||||
- dog
|
||||
"""
|
||||
|
||||
def test_listed_label_creates_detection(self) -> None:
|
||||
self.feed((1, [self.tracked("d1", "dog", 1)]))
|
||||
|
||||
self.assertEqual(self.review_summary(), [("new", "detection")])
|
||||
|
||||
def test_unlisted_label_creates_nothing(self) -> None:
|
||||
self.feed((1, [self.tracked("c1", "cat", 1)]))
|
||||
|
||||
self.assert_no_review()
|
||||
|
||||
def test_unlisted_object_is_left_out_of_detection(self) -> None:
|
||||
self.feed((1, [self.tracked("d1", "dog", 1), self.tracked("c1", "cat", 1)]))
|
||||
|
||||
self.assertEqual(self.review_summary(), [("new", "detection")])
|
||||
self.assertEqual(self.reviews()[0]["after"]["data"]["objects"], ["dog"])
|
||||
|
||||
|
||||
class TestAlertsDisabled(ReviewFlowTestCase):
|
||||
review_config = """
|
||||
review:
|
||||
alerts:
|
||||
enabled: False
|
||||
"""
|
||||
|
||||
def test_alert_label_creates_detection(self) -> None:
|
||||
self.feed((1, [self.tracked("p1", "person", 1)]))
|
||||
|
||||
self.assertEqual(self.review_summary(), [("new", "detection")])
|
||||
|
||||
|
||||
class TestDetectionsDisabled(ReviewFlowTestCase):
|
||||
review_config = """
|
||||
review:
|
||||
detections:
|
||||
enabled: False
|
||||
"""
|
||||
|
||||
def test_alert_label_creates_alert(self) -> None:
|
||||
self.feed((1, [self.tracked("p1", "person", 1)]))
|
||||
|
||||
self.assertEqual(self.review_summary(), [("new", "alert")])
|
||||
|
||||
def test_non_alert_label_creates_nothing(self) -> None:
|
||||
self.feed((1, [self.tracked("d1", "dog", 1)]))
|
||||
|
||||
self.assert_no_review()
|
||||
|
||||
def test_activity_after_alert_is_not_held(self) -> None:
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1, start_time=1)]),
|
||||
(5, [self.tracked("d1", "dog", 5, start_time=5)]),
|
||||
)
|
||||
self.assertEqual(
|
||||
self.maintainer.active_review_segments[CAMERA].pending_detections, []
|
||||
)
|
||||
|
||||
self.feed((2 + self.alert_cutoff, []))
|
||||
|
||||
self.assertEqual(
|
||||
self.non_update_summary(), [("new", "alert"), ("end", "alert")]
|
||||
)
|
||||
self.assertIsNone(self.maintainer.active_review_segments.get(CAMERA))
|
||||
self.assertEqual(len(self.thumbnails()), 1)
|
||||
|
||||
|
||||
class TestAlertsAndDetectionsDisabled(ReviewFlowTestCase):
|
||||
review_config = """
|
||||
review:
|
||||
alerts:
|
||||
enabled: False
|
||||
detections:
|
||||
enabled: False
|
||||
"""
|
||||
|
||||
def test_nothing_is_created(self) -> None:
|
||||
self.feed((1, [self.tracked("p1", "person", 1), self.tracked("d1", "dog", 1)]))
|
||||
|
||||
self.assert_no_review()
|
||||
|
||||
|
||||
class TestReviewLifecycle(ReviewFlowTestCase):
|
||||
def test_detection_upgrades_to_alert_when_alert_object_appears(self) -> None:
|
||||
self.feed(
|
||||
(1, [self.tracked("d1", "dog", 1)]),
|
||||
(2, [self.tracked("d1", "dog", 2), self.tracked("p1", "person", 2)]),
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
self.review_summary(), [("new", "detection"), ("update", "alert")]
|
||||
)
|
||||
new, update = self.reviews()
|
||||
self.assertEqual(new["after"]["id"], update["after"]["id"])
|
||||
self.assertCountEqual(update["after"]["data"]["objects"], ["dog", "person"])
|
||||
|
||||
def test_alert_does_not_downgrade_when_only_detection_objects_remain(
|
||||
self,
|
||||
) -> None:
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1), self.tracked("d1", "dog", 1)]),
|
||||
(2, [self.tracked("d1", "dog", 2)]),
|
||||
(3, [self.tracked("d1", "dog", 3)]),
|
||||
)
|
||||
|
||||
self.assertEqual({severity for _, severity in self.review_summary()}, {"alert"})
|
||||
self.assertEqual(
|
||||
self.maintainer.active_review_segments[CAMERA].severity.value, "alert"
|
||||
)
|
||||
|
||||
def test_alert_stays_open_until_cutoff(self) -> None:
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1)]),
|
||||
(1 + self.alert_cutoff, []),
|
||||
)
|
||||
|
||||
self.assertNotIn("end", [t for t, _ in self.review_summary()])
|
||||
self.assertIsNotNone(self.maintainer.active_review_segments.get(CAMERA))
|
||||
|
||||
def test_alert_ends_after_cutoff_at_last_alert_activity(self) -> None:
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1)]),
|
||||
(5, [self.tracked("p1", "person", 5)]),
|
||||
(6, []),
|
||||
(5 + self.alert_cutoff + 1, []),
|
||||
)
|
||||
|
||||
end = self.reviews()[-1]
|
||||
self.assertEqual(end["type"], "end")
|
||||
self.assertEqual(end["after"]["severity"], "alert")
|
||||
self.assertEqual(end["after"]["end_time"], 5)
|
||||
self.assertIsNone(self.maintainer.active_review_segments.get(CAMERA))
|
||||
|
||||
def test_ongoing_alert_activity_extends_alert(self) -> None:
|
||||
last_activity = 1 + self.alert_cutoff * 2
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1)]),
|
||||
(
|
||||
1 + self.alert_cutoff,
|
||||
[self.tracked("p1", "person", 1 + self.alert_cutoff)],
|
||||
),
|
||||
(last_activity, [self.tracked("p1", "person", last_activity)]),
|
||||
(last_activity + 1, []),
|
||||
)
|
||||
|
||||
self.assertNotIn("end", [t for t, _ in self.review_summary()])
|
||||
self.assertEqual(
|
||||
self.maintainer.active_review_segments[CAMERA].last_alert_time,
|
||||
last_activity,
|
||||
)
|
||||
|
||||
def test_detection_ends_after_cutoff_at_last_detection_activity(self) -> None:
|
||||
self.feed(
|
||||
(1, [self.tracked("d1", "dog", 1)]),
|
||||
(5, [self.tracked("d1", "dog", 5)]),
|
||||
(5 + self.detection_cutoff, []),
|
||||
)
|
||||
self.assertNotIn("end", [t for t, _ in self.review_summary()])
|
||||
|
||||
self.feed((5 + self.detection_cutoff + 1, []))
|
||||
|
||||
end = self.reviews()[-1]
|
||||
self.assertEqual(end["type"], "end")
|
||||
self.assertEqual(end["after"]["severity"], "detection")
|
||||
self.assertEqual(end["after"]["end_time"], 5)
|
||||
self.assertIsNone(self.maintainer.active_review_segments.get(CAMERA))
|
||||
|
||||
def test_stationary_object_does_not_extend_alert(self) -> None:
|
||||
self.feed(
|
||||
(1, [self.tracked("c1", "car", 1)]),
|
||||
(2, [self.tracked("c1", "car", 2, stationary=True)]),
|
||||
(
|
||||
2 + self.alert_cutoff,
|
||||
[self.tracked("c1", "car", 2 + self.alert_cutoff, stationary=True)],
|
||||
),
|
||||
)
|
||||
|
||||
end = self.reviews()[-1]
|
||||
self.assertEqual(end["type"], "end")
|
||||
self.assertEqual(end["after"]["end_time"], 1)
|
||||
|
||||
def test_new_activity_after_end_creates_new_review(self) -> None:
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1)]),
|
||||
(2 + self.alert_cutoff, []),
|
||||
(3 + self.alert_cutoff, [self.tracked("d1", "dog", 3 + self.alert_cutoff)]),
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
self.non_update_summary(),
|
||||
[("new", "alert"), ("end", "alert"), ("new", "detection")],
|
||||
)
|
||||
ids = {r["after"]["id"] for r in self.reviews() if r["type"] != "update"}
|
||||
self.assertEqual(len(ids), 2)
|
||||
|
||||
def test_alert_splits_into_detection_when_detection_activity_continues(
|
||||
self,
|
||||
) -> None:
|
||||
# the person leaves after the first frame while the dog keeps moving
|
||||
dog_frames = [
|
||||
(t, [self.tracked("d1", "dog", t, start_time=1)])
|
||||
for t in range(11, 2 + self.alert_cutoff + 10, 10)
|
||||
]
|
||||
self.feed(
|
||||
(
|
||||
1,
|
||||
[
|
||||
self.tracked("p1", "person", 1, start_time=1),
|
||||
self.tracked("d1", "dog", 1, start_time=1),
|
||||
],
|
||||
),
|
||||
*dog_frames,
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
self.non_update_summary(),
|
||||
[("new", "alert"), ("end", "alert"), ("new", "detection")],
|
||||
)
|
||||
alert_end = next(r for r in self.reviews() if r["type"] == "end")
|
||||
self.assertEqual(alert_end["after"]["end_time"], 1)
|
||||
|
||||
detection = self.maintainer.active_review_segments[CAMERA]
|
||||
self.assertEqual(detection.severity.value, "detection")
|
||||
self.assertEqual(detection.start_time, 11)
|
||||
self.assertEqual(list(detection.detections.values()), ["dog"])
|
||||
|
||||
# the detection ends once the dog stops moving
|
||||
last_dog_time = dog_frames[-1][0]
|
||||
self.feed((last_dog_time + self.detection_cutoff + 1, []))
|
||||
|
||||
end = self.reviews()[-1]
|
||||
self.assertEqual(end["type"], "end")
|
||||
self.assertEqual(end["after"]["severity"], "detection")
|
||||
self.assertEqual(end["after"]["id"], detection.id)
|
||||
self.assertEqual(end["after"]["end_time"], last_dog_time)
|
||||
|
||||
def test_detection_starting_after_alert_activity_is_split_out(self) -> None:
|
||||
# the dog only shows up after the person has left
|
||||
dog_frames = [
|
||||
(t, [self.tracked("d1", "dog", t, start_time=11)])
|
||||
for t in range(11, 2 + self.alert_cutoff + 10, 10)
|
||||
]
|
||||
self.feed((1, [self.tracked("p1", "person", 1, start_time=1)]), *dog_frames)
|
||||
|
||||
self.assertEqual(
|
||||
self.non_update_summary(),
|
||||
[("new", "alert"), ("end", "alert"), ("new", "detection")],
|
||||
)
|
||||
alert_end = next(r for r in self.reviews() if r["type"] == "end")
|
||||
self.assertEqual(alert_end["after"]["end_time"], 1)
|
||||
self.assertEqual(alert_end["after"]["data"]["objects"], ["person"])
|
||||
|
||||
detection = self.maintainer.active_review_segments[CAMERA]
|
||||
self.assertEqual(detection.severity.value, "detection")
|
||||
self.assertEqual(detection.start_time, 11)
|
||||
self.assertEqual(list(detection.detections.values()), ["dog"])
|
||||
|
||||
def test_detection_leaving_before_alert_cutoff_gets_detection(self) -> None:
|
||||
# the dog comes and goes after the person left, all before the alert
|
||||
# cutoff, so nothing is active when the alert ends
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1, start_time=1)]),
|
||||
(11, [self.tracked("d1", "dog", 11, start_time=11)]),
|
||||
(21, [self.tracked("d1", "dog", 21, start_time=11)]),
|
||||
(31, []),
|
||||
(2 + self.alert_cutoff, []),
|
||||
(22 + self.detection_cutoff, []),
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
self.non_update_summary(),
|
||||
[
|
||||
("new", "alert"),
|
||||
("end", "alert"),
|
||||
("new", "detection"),
|
||||
("end", "detection"),
|
||||
],
|
||||
)
|
||||
ends = [r["after"] for r in self.reviews() if r["type"] == "end"]
|
||||
self.assertEqual(ends[0]["end_time"], 1)
|
||||
self.assertEqual(ends[0]["data"]["objects"], ["person"])
|
||||
self.assertEqual(ends[1]["data"]["objects"], ["dog"])
|
||||
self.assertEqual(ends[1]["start_time"], 11)
|
||||
self.assertEqual(ends[1]["end_time"], 21)
|
||||
|
||||
def test_detection_older_than_detection_cutoff_gets_detection(self) -> None:
|
||||
# the dog is gone longer than the detection cutoff by the time the
|
||||
# alert ends, its activity still needs a detection
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1, start_time=1)]),
|
||||
(3, [self.tracked("d1", "dog", 3, start_time=3)]),
|
||||
(5, [self.tracked("d1", "dog", 5, start_time=3)]),
|
||||
(6, []),
|
||||
(2 + self.alert_cutoff, []),
|
||||
(3 + self.alert_cutoff, []),
|
||||
)
|
||||
self.assertGreater(2 + self.alert_cutoff, 5 + self.detection_cutoff)
|
||||
|
||||
self.assertEqual(
|
||||
self.non_update_summary(),
|
||||
[
|
||||
("new", "alert"),
|
||||
("end", "alert"),
|
||||
("new", "detection"),
|
||||
("end", "detection"),
|
||||
],
|
||||
)
|
||||
ends = [r["after"] for r in self.reviews() if r["type"] == "end"]
|
||||
self.assertEqual(ends[0]["end_time"], 1)
|
||||
self.assertEqual(ends[0]["data"]["objects"], ["person"])
|
||||
self.assertEqual(ends[1]["data"]["objects"], ["dog"])
|
||||
self.assertEqual(ends[1]["start_time"], 3)
|
||||
self.assertEqual(ends[1]["end_time"], 5)
|
||||
self.assertIsNone(self.maintainer.active_review_segments.get(CAMERA))
|
||||
|
||||
def test_separate_detection_activity_after_alert_is_not_combined(self) -> None:
|
||||
# the dog and cat are seen further apart than the detection cutoff
|
||||
# while the alert is waiting to be cut off
|
||||
cat_time = 3 + self.detection_cutoff + 6
|
||||
self.assertLess(cat_time, 1 + self.alert_cutoff)
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1, start_time=1)]),
|
||||
(3, [self.tracked("d1", "dog", 3, start_time=3)]),
|
||||
(4, []),
|
||||
(cat_time, [self.tracked("c1", "cat", cat_time, start_time=cat_time)]),
|
||||
(2 + self.alert_cutoff, []),
|
||||
(cat_time + self.detection_cutoff + 1, []),
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
self.non_update_summary(),
|
||||
[
|
||||
("new", "alert"),
|
||||
("end", "alert"),
|
||||
("new", "detection"),
|
||||
("end", "detection"),
|
||||
("new", "detection"),
|
||||
("end", "detection"),
|
||||
],
|
||||
)
|
||||
ends = [r["after"] for r in self.reviews() if r["type"] == "end"]
|
||||
self.assertEqual(ends[0]["data"]["objects"], ["person"])
|
||||
self.assertEqual(ends[1]["data"]["objects"], ["dog"])
|
||||
self.assertEqual((ends[1]["start_time"], ends[1]["end_time"]), (3, 3))
|
||||
self.assertEqual(ends[2]["data"]["objects"], ["cat"])
|
||||
self.assertEqual(
|
||||
(ends[2]["start_time"], ends[2]["end_time"]), (cat_time, cat_time)
|
||||
)
|
||||
for detection in ends[1:]:
|
||||
self.assertTrue(Path(detection["thumb_path"]).is_file())
|
||||
self.assertIsNotNone(detection["data"]["thumb_time"])
|
||||
|
||||
def test_resumed_alert_publishes_pending_detection_objects(self) -> None:
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1, start_time=1)]),
|
||||
(5, [self.tracked("d1", "dog", 5, start_time=5)]),
|
||||
(10, [self.tracked("p1", "person", 10, start_time=1)]),
|
||||
)
|
||||
|
||||
latest = self.reviews()[-1]
|
||||
self.assertEqual(latest["type"], "update")
|
||||
self.assertEqual(latest["after"]["severity"], "alert")
|
||||
self.assertCountEqual(latest["after"]["data"]["objects"], ["person", "dog"])
|
||||
|
||||
def test_split_detection_has_thumbnail_of_its_activity(self) -> None:
|
||||
dog_frames = [
|
||||
(t, [self.tracked("d1", "dog", t, start_time=11)])
|
||||
for t in range(11, 2 + self.alert_cutoff + 10, 10)
|
||||
]
|
||||
self.feed((1, [self.tracked("p1", "person", 1, start_time=1)]), *dog_frames)
|
||||
|
||||
new_detection = next(
|
||||
r["after"]
|
||||
for r in self.reviews()
|
||||
if r["type"] == "new" and r["after"]["severity"] == "detection"
|
||||
)
|
||||
self.assertTrue(Path(new_detection["thumb_path"]).is_file())
|
||||
# captured from the dog's first frame, not a later fallback frame
|
||||
self.assertIsNotNone(new_detection["data"]["thumb_time"])
|
||||
self.assertIn(
|
||||
f"{CAMERA}_11",
|
||||
[c.args[0] for c in self.maintainer.frame_manager.get.call_args_list],
|
||||
)
|
||||
|
||||
def assert_force_end_publishes_pending_detection(self, topic: str) -> None:
|
||||
# the dog is held as a pending detection when the alert is ended
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1, start_time=1)]),
|
||||
(5, [self.tracked("d1", "dog", 5, start_time=5)]),
|
||||
)
|
||||
|
||||
if topic == "remove":
|
||||
# the config updater has already dropped the camera
|
||||
self.maintainer.config.cameras.pop(CAMERA)
|
||||
|
||||
self.maintainer.config_subscriber.check_for_updates.side_effect = [
|
||||
{topic: [CAMERA]}
|
||||
]
|
||||
self.feed()
|
||||
|
||||
self.assertEqual(
|
||||
self.non_update_summary(),
|
||||
[
|
||||
("new", "alert"),
|
||||
("end", "alert"),
|
||||
("new", "detection"),
|
||||
("end", "detection"),
|
||||
],
|
||||
)
|
||||
end = self.reviews()[-1]["after"]
|
||||
self.assertEqual(end["data"]["objects"], ["dog"])
|
||||
self.assertEqual((end["start_time"], end["end_time"]), (5, 5))
|
||||
self.assertIsNone(self.maintainer.active_review_segments.get(CAMERA))
|
||||
|
||||
# every thumbnail on disk belongs to a published review
|
||||
published = {Path(r["after"]["thumb_path"]).name for r in self.reviews()}
|
||||
self.assertEqual(self.thumbnails(), published)
|
||||
|
||||
def test_disabled_camera_publishes_pending_detections(self) -> None:
|
||||
self.assert_force_end_publishes_pending_detection("enabled")
|
||||
|
||||
def test_removed_camera_publishes_pending_detections(self) -> None:
|
||||
self.assert_force_end_publishes_pending_detection("remove")
|
||||
|
||||
def test_pending_thumbnail_removed_when_alert_resumes(self) -> None:
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1, start_time=1)]),
|
||||
(5, [self.tracked("d1", "dog", 5, start_time=5)]),
|
||||
)
|
||||
alert = self.maintainer.active_review_segments[CAMERA]
|
||||
pending_thumb = Path(alert.pending_detections[0].frame_path)
|
||||
self.assertTrue(pending_thumb.is_file())
|
||||
|
||||
self.feed((10, [self.tracked("p1", "person", 10, start_time=1)]))
|
||||
|
||||
self.assertEqual(alert.pending_detections, [])
|
||||
self.assertFalse(pending_thumb.exists())
|
||||
self.assertEqual(self.thumbnails(), {Path(alert.frame_path).name})
|
||||
|
||||
def test_multiple_pending_objects_share_one_detection(self) -> None:
|
||||
# objects within the detection cutoff of each other, whether seen
|
||||
# together or later, make up one detection
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1, start_time=1)]),
|
||||
(
|
||||
5,
|
||||
[
|
||||
self.tracked("d1", "dog", 5, start_time=5),
|
||||
self.tracked("c1", "cat", 5, start_time=5),
|
||||
],
|
||||
),
|
||||
(20, [self.tracked("b1", "bird", 20, start_time=20)]),
|
||||
(21, []),
|
||||
(2 + self.alert_cutoff, []),
|
||||
(21 + self.detection_cutoff, []),
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
self.non_update_summary(),
|
||||
[
|
||||
("new", "alert"),
|
||||
("end", "alert"),
|
||||
("new", "detection"),
|
||||
("end", "detection"),
|
||||
],
|
||||
)
|
||||
end = self.reviews()[-1]["after"]
|
||||
self.assertCountEqual(end["data"]["objects"], ["dog", "cat", "bird"])
|
||||
self.assertCountEqual(end["data"]["detections"], ["d1", "c1", "b1"])
|
||||
self.assertEqual((end["start_time"], end["end_time"]), (5, 20))
|
||||
# the thumbnail is framed on both objects seen together, the bird
|
||||
# alone is fewer objects so it does not replace it
|
||||
frames = [c.args[0] for c in self.maintainer.frame_manager.get.call_args_list]
|
||||
self.assertIn(f"{CAMERA}_5", frames)
|
||||
self.assertNotIn(f"{CAMERA}_20", frames)
|
||||
|
||||
|
||||
class TestSplitDetectionLabels(ReviewFlowTestCase):
|
||||
review_config = """
|
||||
review:
|
||||
alerts:
|
||||
labels:
|
||||
- person
|
||||
"""
|
||||
|
||||
def test_split_detection_keeps_sub_labels_and_zones(self) -> None:
|
||||
self.feed(
|
||||
(1, [self.tracked("p1", "person", 1, start_time=1)]),
|
||||
(
|
||||
5,
|
||||
[
|
||||
self.tracked(
|
||||
"d1",
|
||||
"dog",
|
||||
5,
|
||||
start_time=5,
|
||||
zones=["yard"],
|
||||
sub_label=("Rex", 0.95),
|
||||
),
|
||||
self.tracked(
|
||||
"c1",
|
||||
"car",
|
||||
5,
|
||||
start_time=5,
|
||||
zones=["driveway"],
|
||||
sub_label=("fedex", 0.9),
|
||||
),
|
||||
],
|
||||
),
|
||||
(2 + self.alert_cutoff, []),
|
||||
)
|
||||
|
||||
ends = [r["after"] for r in self.reviews() if r["type"] == "end"]
|
||||
self.assertEqual(len(ends), 2)
|
||||
alert_end, detection = ends
|
||||
self.assertEqual(alert_end["data"]["objects"], ["person"])
|
||||
self.assertEqual(alert_end["data"]["sub_labels"], [])
|
||||
self.assertEqual(alert_end["data"]["zones"], [])
|
||||
|
||||
self.assertEqual(detection["severity"], "detection")
|
||||
# attributes replace the label, other sub labels verify it
|
||||
self.assertCountEqual(detection["data"]["objects"], ["dog-verified", "fedex"])
|
||||
self.assertEqual(detection["data"]["verified_objects"], ["dog-verified"])
|
||||
self.assertEqual(detection["data"]["sub_labels"], ["Rex"])
|
||||
self.assertCountEqual(detection["data"]["zones"], ["yard", "driveway"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user