diff --git a/frigate/review/maintainer.py b/frigate/review/maintainer.py index b770bbd8e0..f97940d4a3 100644 --- a/frigate/review/maintainer.py +++ b/frigate/review/maintainer.py @@ -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, diff --git a/frigate/test/test_review_flow.py b/frigate/test/test_review_flow.py new file mode 100644 index 0000000000..2a1b5ee097 --- /dev/null +++ b/frigate/test/test_review_flow.py @@ -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()