Coverage for custom_components/camera_autonotify/tracker.py: 100%
219 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:32 +0000
1"""What to notify for the reviews of one camera or zone, as Frigate reports them over MQTT
3Pure logic, with no Home Assistant in it: the time is passed in, and anything to happen later is handed back as
4a `Wait` for the caller to time, and then call `on_timer` for.
5"""
7from __future__ import annotations
9import json
10from collections.abc import Mapping
11from dataclasses import dataclass, field, fields
12from enum import StrEnum
13from typing import Any
15from homeassistant.util import slugify
17SEVERITY_ALERT = "alert"
18SEVERITY_DETECTION = "detection"
19TYPE_NEW = "new"
20TYPE_UPDATE = "update"
21TYPE_END = "end"
22# a review that never ends, from a lost MQTT message or a restart of Frigate, is forgotten after this long
23STALE_AFTER = 3600
26def _slugs(zones: tuple[str, ...] | set[str]) -> set[str]:
27 return {slugify(zone) for zone in zones}
30class Reason(StrEnum):
31 """Why a notification is sent"""
33 NEW = "new"
34 LABEL = "label"
35 ZONE = "zone"
36 SUB_LABEL = "sub_label"
37 SEVERITY = "severity"
38 FINAL = "final"
41@dataclass(frozen=True)
42class Settings:
43 """The filters and timing of the Frigate blueprint, with its defaults where it has them"""
45 severities: frozenset[str] = frozenset({SEVERITY_ALERT})
46 # any, where empty
47 labels: frozenset[str] = frozenset()
48 zones: tuple[str, ...] = ()
49 # every zone chosen, rather than any of them
50 zone_multi: bool = False
51 # and entered in the order chosen
52 zone_order: bool = False
53 # seconds after a review is notified before another review is
54 cooldown: float = 30
55 # seconds held back before the first notification of a review, for a better snapshot
56 initial_delay: float = 0
57 # a last notification once the review has ended, after this many seconds
58 final_update: bool = False
59 final_delay: float = 0
60 # seconds from the first notification of a review after which nothing more is sent for it
61 timeout: float = 600
62 # only the first notification of a review makes a sound
63 alert_once: bool = True
64 # told again when a sub label, such as a face or plate, is recognised
65 update_sub_label: bool = True
67 @classmethod
68 def from_options(cls, options: Mapping[str, Any]) -> Settings:
69 """From a camera's options, which are named as these, and only kept where they differ from the default"""
70 given = {f.name: options[f.name] for f in fields(cls) if f.name in options}
71 for name in ("severities", "labels"):
72 if name in given:
73 given[name] = frozenset(given[name])
74 if "zones" in given:
75 given["zones"] = tuple(given["zones"])
76 return cls(**given)
79@dataclass(frozen=True)
80class Review:
81 """A review as Frigate reports it, in the `after` of its MQTT message"""
83 id: str
84 camera: str
85 severity: str
86 objects: tuple[str, ...]
87 sub_labels: tuple[str, ...]
88 # as written in the Frigate config, in the order entered
89 zones: tuple[str, ...]
90 detections: tuple[str, ...]
91 ended: bool
93 @classmethod
94 def from_payload(cls, after: dict[str, Any]) -> Review:
95 data = after.get("data") or {}
96 # audio, such as a scream or a bark, is as much what was seen as an object is
97 objects = [*data.get("objects", []), *data.get("audio", [])]
98 sub_labels = [label[0] if isinstance(label, list) else label for label in data.get("sub_labels") or []]
99 return cls(
100 id=after["id"],
101 camera=after["camera"],
102 severity=after.get("severity") or SEVERITY_DETECTION,
103 objects=tuple(dict.fromkeys(objects)),
104 sub_labels=tuple(dict.fromkeys(str(label) for label in sub_labels if label)),
105 zones=tuple(dict.fromkeys(data.get("zones") or [])),
106 detections=tuple(data.get("detections") or []),
107 ended=after.get("end_time") is not None,
108 )
110 @property
111 def label(self) -> str:
112 """What was seen, as said in a notification, with whoever or whatever was recognised"""
113 text = " and ".join(obj.replace("_", " ").capitalize() for obj in self.objects) or "Something"
114 return f"{text} ({', '.join(self.sub_labels)})" if self.sub_labels else text
117@dataclass(frozen=True)
118class Decision:
119 """A notification to send, with the review as it then is"""
121 review: Review
122 reason: Reason
123 # the notifications sent for the review before this one
124 previous: int
125 silent: bool
127 @property
128 def is_update(self) -> bool:
129 return self.previous > 0
131 @property
132 def is_final(self) -> bool:
133 return self.reason is Reason.FINAL
135 @property
136 def detection_id(self) -> str | None:
137 """The detection whose snapshot shows the review, the first, which is what set it off"""
138 return self.review.detections[0] if self.review.detections else None
141@dataclass(frozen=True)
142class Wait:
143 """A notification held back, to be looked at again with `on_timer` at `at`"""
145 review_id: str
146 at: float
147 reason: Reason
150@dataclass
151class _Tracked:
152 review: Review
153 first_seen: float
154 # when the first notification went, or is due
155 notified_at: float | None = None
156 sent: int = 0
157 # as last notified, so an update is only for what's new since
158 objects: set[str] = field(default_factory=set)
159 zones: set[str] = field(default_factory=set)
160 sub_labels: set[str] = field(default_factory=set)
161 severity: str = ""
162 # held back, the first notification by the initial delay, or the last by the final delay
163 waiting: Reason | None = None
164 # passed over for the cooldown, so none of its updates are sent either
165 skipped: bool = False
168class ReviewTracker:
169 """The reviews of a camera, or of a zone on any camera, and the notifications each should give"""
171 def __init__(self, settings: Settings, camera: str | None = None, zone: str | None = None) -> None:
172 self.settings = settings
173 self.camera = camera
174 # a zone is matched by its slug, as its device has it, since reviews give it as written in the Frigate config
175 self.zone = slugify(zone) if zone else None
176 self._reviews: dict[str, _Tracked] = {}
177 self._last_new: float | None = None
179 def on_message(self, payload: str | bytes | bytearray | dict[str, Any], now: float) -> Decision | Wait | None:
180 """A message from `{topic_prefix}/reviews`"""
181 message = json.loads(payload) if isinstance(payload, (str, bytes, bytearray)) else payload
182 after = message.get("after")
183 if not isinstance(after, dict) or not after.get("id") or not after.get("camera"):
184 return None
185 review = Review.from_payload(after)
186 if not self._ours(review):
187 return None
188 self._forget_stale(now)
189 tracked = self._reviews.get(review.id)
190 if tracked is None:
191 if message.get("type") == TYPE_END and not self._wanted(review):
192 return None
193 tracked = self._reviews[review.id] = _Tracked(review, first_seen=now)
194 tracked.review = review
195 if tracked.skipped:
196 self._drop_if_ended(tracked)
197 return None
198 if tracked.notified_at is None:
199 return self._first(tracked, now)
200 if tracked.waiting is not None:
201 # held back, and sent as it is once it's due, with whatever has changed
202 return None
203 if now - tracked.notified_at > self.settings.timeout:
204 del self._reviews[review.id]
205 return None
206 if review.ended:
207 return self._ending(tracked, now)
208 return self._update(tracked)
210 def on_timer(self, review_id: str, now: float) -> Decision | Wait | None:
211 """A notification held back is due"""
212 tracked = self._reviews.get(review_id)
213 if tracked is None or tracked.waiting is None:
214 return None
215 reason, tracked.waiting = tracked.waiting, None
216 if reason is Reason.FINAL:
217 del self._reviews[review_id]
218 return self._decide(tracked, Reason.FINAL)
219 return self._first_decision(tracked)
221 @property
222 def tracked(self) -> list[str]:
223 return list(self._reviews)
225 def _ours(self, review: Review) -> bool:
226 if self.zone is not None:
227 return self.zone in {slugify(zone) for zone in review.zones}
228 return review.camera == self.camera
230 def _wanted(self, review: Review) -> bool:
231 settings = self.settings
232 if review.severity not in settings.severities:
233 return False
234 if settings.labels and not settings.labels & set(review.objects):
235 return False
236 if settings.zones:
237 chosen = [slugify(zone) for zone in settings.zones]
238 entered = [slugify(zone) for zone in review.zones]
239 matched = [zone for zone in chosen if zone in entered]
240 if not matched or (settings.zone_multi and len(matched) < len(chosen)):
241 return False
242 # Frigate lists the zones of a review in the order they were entered
243 positions = [entered.index(zone) for zone in matched]
244 if settings.zone_order and positions != sorted(positions):
245 return False
246 return True
248 def _first(self, tracked: _Tracked, now: float) -> Decision | Wait | None:
249 if not self._wanted(tracked.review):
250 # may yet be, as the review goes on
251 self._drop_if_ended(tracked)
252 return None
253 if self._last_new is not None and now - self._last_new < self.settings.cooldown:
254 tracked.skipped = True
255 self._drop_if_ended(tracked)
256 return None
257 self._last_new = tracked.notified_at = now
258 if self.settings.initial_delay > 0:
259 tracked.waiting = Reason.NEW
260 return Wait(tracked.review.id, now + self.settings.initial_delay, Reason.NEW)
261 return self._first_decision(tracked)
263 def _first_decision(self, tracked: _Tracked) -> Decision:
264 decision = self._decide(tracked, Reason.NEW)
265 if tracked.review.ended:
266 # over before it was told at all, so told once, with nothing to follow
267 del self._reviews[tracked.review.id]
268 return decision
270 def _update(self, tracked: _Tracked) -> Decision | None:
271 review = tracked.review
272 settings = self.settings
273 reason: Reason | None = None
274 if review.severity == SEVERITY_ALERT and tracked.severity != SEVERITY_ALERT:
275 reason = Reason.SEVERITY
276 elif (new := set(review.objects) - tracked.objects) and (not settings.labels or new & settings.labels):
277 reason = Reason.LABEL
278 elif settings.update_sub_label and set(review.sub_labels) - tracked.sub_labels:
279 reason = Reason.SUB_LABEL
280 elif settings.zones and (_slugs(review.zones) - _slugs(tracked.zones)) & _slugs(settings.zones):
281 reason = Reason.ZONE
282 return self._decide(tracked, reason) if reason is not None else None
284 def _ending(self, tracked: _Tracked, now: float) -> Decision | Wait | None:
285 if not self.settings.final_update:
286 del self._reviews[tracked.review.id]
287 return None
288 if self.settings.final_delay > 0:
289 tracked.waiting = Reason.FINAL
290 return Wait(tracked.review.id, now + self.settings.final_delay, Reason.FINAL)
291 del self._reviews[tracked.review.id]
292 return self._decide(tracked, Reason.FINAL)
294 def _decide(self, tracked: _Tracked, reason: Reason) -> Decision:
295 review = tracked.review
296 decision = Decision(review, reason, tracked.sent, silent=tracked.sent > 0 and self.settings.alert_once)
297 tracked.sent += 1
298 tracked.objects = set(review.objects)
299 tracked.zones = set(review.zones)
300 tracked.sub_labels = set(review.sub_labels)
301 tracked.severity = review.severity
302 return decision
304 def _drop_if_ended(self, tracked: _Tracked) -> None:
305 if tracked.review.ended:
306 del self._reviews[tracked.review.id]
308 def _forget_stale(self, now: float) -> None:
309 for review_id in [r for r, tracked in self._reviews.items() if now - tracked.first_seen > STALE_AFTER]:
310 del self._reviews[review_id]