feat(Q5): package delivery state machine with sunset-aware reminders

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

vigilar/detection/package.py

@@ -0,0 +1,99 @@
+"""Package delivery detection state machine."""
+
+import datetime
+import logging
+import time
+from enum import StrEnum
+
+from vigilar.detection.solar import get_sunset
+
+log = logging.getLogger(__name__)
+
+PACKAGE_CLASS_IDS = {24, 26, 28}  # backpack, handbag, suitcase
+
+
+class PackageState(StrEnum):
+    IDLE = "IDLE"
+    PRESENT = "PRESENT"
+    REMINDED = "REMINDED"
+    COLLECTED = "COLLECTED"
+
+
+class PackageTracker:
+    def __init__(self, camera_id: str, latitude: float, longitude: float):
+        self.camera_id = camera_id
+        self._latitude = latitude
+        self._longitude = longitude
+        self._state = PackageState.IDLE
+        self._person_last_seen: float = 0
+        self._person_departure_threshold: float = 30.0
+        self._package_bbox: list[float] | None = None
+        self._detected_at: float = 0
+        self._reminder_at: float = 0
+        self._person_returned: bool = False
+
+    @property
+    def state(self) -> PackageState:
+        return self._state
+
+    def on_person_detected(self, now: float) -> None:
+        self._person_last_seen = now
+        if self._state in (PackageState.PRESENT, PackageState.REMINDED):
+            self._person_returned = True
+
+    def check_for_package(self, detections: list[dict], now: float) -> bool:
+        if self._state != PackageState.IDLE:
+            return False
+        if self._person_last_seen == 0:
+            return False
+        if now - self._person_last_seen < self._person_departure_threshold:
+            return False
+        for det in detections:
+            if det.get("class_id") in PACKAGE_CLASS_IDS:
+                self._state = PackageState.PRESENT
+                self._package_bbox = det.get("bbox")
+                self._detected_at = now
+                self._reminder_at = self._calculate_reminder_time(now)
+                self._person_returned = False
+                log.info("Package detected on %s", self.camera_id)
+                return True
+        return False
+
+    def check_collected(self, detections: list[dict], now: float) -> bool:
+        if self._state not in (PackageState.PRESENT, PackageState.REMINDED):
+            return False
+        if not self._person_returned:
+            return False
+        for det in detections:
+            if det.get("class_id") in PACKAGE_CLASS_IDS:
+                return False
+        self._state = PackageState.COLLECTED
+        return True
+
+    def check_reminder(self, now: float) -> bool:
+        if self._state != PackageState.PRESENT:
+            return False
+        if now >= self._reminder_at:
+            self._state = PackageState.REMINDED
+            return True
+        return False
+
+    def reset(self) -> None:
+        self._state = PackageState.IDLE
+        self._package_bbox = None
+        self._detected_at = 0
+        self._reminder_at = 0
+        self._person_returned = False
+
+    def _calculate_reminder_time(self, now: float) -> float:
+        three_hours = now + 3 * 3600
+        try:
+            today = datetime.date.today()
+            sunset_time = get_sunset(self._latitude, self._longitude, today)
+            sunset_dt = datetime.datetime.combine(today, sunset_time)
+            sunset_ts = sunset_dt.timestamp()
+            if sunset_ts < now:
+                sunset_ts += 86400
+            return min(sunset_ts, three_hours)
+        except Exception:
+            return three_hours