Coverage for custom_components/supernotify/snoozer.py: 97%
266 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-08 17:42 +0000
1from __future__ import annotations
3import datetime as dt
4import logging
5from datetime import timedelta
6from typing import TYPE_CHECKING, Any
8from homeassistant.util import dt as dt_util
10from . import DOMAIN
11from .common import canonical_priority, spoken_name
12from .const import (
13 ATTR_ACTION,
14 ATTR_MOBILE_APP_ID,
15 ATTR_PERSON_ID,
16 ATTR_REPLY_TEXT,
17 CONF_SNOOZE_TIME,
18 PRIORITY_CRITICAL,
19 PRIORITY_MEDIUM,
20 PRIORITY_VALUES,
21)
22from .model import CommandType, GlobalTargetType, QualifiedTargetType, RecipientType, Target, TargetType
24if TYPE_CHECKING:
25 from homeassistant.core import Event
27 from .delivery import Delivery
28 from .engine import SupernotifyEngine
29 from .hass_api import HomeAssistantAPI
30 from .people import PeopleRegistry, Recipient
32_LOGGER = logging.getLogger(__name__)
34# what a snooze covers, by the name the AI tool and supernotify.snooze give it
35SNOOZE_SCOPES: dict[str, TargetType] = {
36 "everything": GlobalTargetType.EVERYTHING,
37 "noncritical": GlobalTargetType.NONCRITICAL,
38 "delivery": QualifiedTargetType.DELIVERY,
39 "transport": QualifiedTargetType.TRANSPORT,
40 "priority": QualifiedTargetType.PRIORITY,
41 "camera": QualifiedTargetType.CAMERA,
42 "tag": QualifiedTargetType.TAG,
43}
46def snooze_name_error(
47 engine: SupernotifyEngine, scope: QualifiedTargetType, name: str | None, cmd: CommandType
48) -> tuple[str, dict[str, str]] | None:
49 """Why a snooze of this scope can't take this name, as an `exceptions` translation key and its
50 placeholders, or None - shared by the AI tool and supernotify.snooze"""
51 registry = engine.context.delivery_registry
52 valid: list[str] | None = {
53 QualifiedTargetType.DELIVERY: list(registry.deliveries),
54 QualifiedTargetType.TRANSPORT: list(registry.transports),
55 QualifiedTargetType.PRIORITY: list(PRIORITY_VALUES),
56 }.get(scope)
57 if not name:
58 return "snooze_name_needed", {"scope": scope.lower()}
59 if scope == QualifiedTargetType.PRIORITY:
60 name = canonical_priority(name)
61 if valid is not None and name not in valid:
62 return "snooze_name_unknown", {"scope": scope.lower(), "name": name, "choices": ", ".join(valid)}
63 if scope == QualifiedTargetType.CAMERA and not name.startswith("camera."):
64 return "snooze_camera_not_entity", {}
65 # a tag snooze can still be undone when what it named has gone
66 if scope == QualifiedTargetType.TAG and cmd != CommandType.NORMAL and not engine.tag_matches(name):
67 return "snooze_tag_unknown", {"name": name}
68 return None
71STORAGE_VERSION = 1
72STORAGE_KEY = f"{DOMAIN}.snoozes"
75class Snooze:
76 target: str | list[str] | None
77 target_type: TargetType
78 snoozed_at: dt.datetime
79 snooze_until: dt.datetime | None = None
80 recipient_type: RecipientType
81 recipient: str | None
82 reason: str | None = None
84 def __init__(
85 self,
86 target_type: TargetType,
87 recipient_type: RecipientType,
88 target: str | list[str] | None = None,
89 recipient: str | None = None,
90 snooze_for: timedelta | None = None,
91 reason: str | None = None,
92 ) -> None:
93 self.snoozed_at = dt_util.now()
94 self.target = target
95 self.target_type = target_type
96 self.recipient_type: RecipientType = recipient_type
97 self.recipient = recipient
98 self.reason = reason
99 self.snooze_until = None
100 if snooze_for:
101 self.snooze_until = self.snoozed_at + snooze_for
103 def std_recipient(self) -> str | None:
104 return self.recipient if self.recipient_type == RecipientType.USER else RecipientType.EVERYONE
106 def short_key(self) -> str:
107 # only one GLOBAL can be active at a time
108 target = "GLOBAL" if self.target_type in GlobalTargetType else f"{self.target_type}_{self.target}"
109 return f"{target}_{self.std_recipient()}"
111 def __eq__(self, other: object) -> bool:
112 """Check if two snoozes for the same thing"""
113 if not isinstance(other, Snooze):
114 return False
115 return self.short_key() == other.short_key()
117 def __repr__(self) -> str:
118 """Return a string representation of the object."""
119 target = "GLOBAL" if self.target_type in GlobalTargetType else f"{self.target_type}_{self.target}"
120 return f"Snooze({target}, {self.std_recipient()})"
122 def active(self) -> bool:
123 return self.snooze_until is None or self.snooze_until > dt_util.now()
125 def export(self) -> dict[str, Any]:
126 return {
127 "target_type": self.target_type,
128 "target": self.target,
129 "recipient_type": self.recipient_type,
130 "recipient": self.recipient,
131 "reason": self.reason,
132 "snoozed_at": dt_util.as_local(self.snoozed_at).strftime("%H:%M:%S") if self.snoozed_at else None,
133 "snooze_until": dt_util.as_local(self.snooze_until).strftime("%H:%M:%S") if self.snooze_until else None,
134 }
136 def to_storage_dict(self) -> dict[str, Any]:
137 """Full-fidelity serialization for persistence (unlike export(), which is a display
138 summary that loses the date part of timestamps)."""
139 return {
140 "target_type_class": "GlobalTargetType"
141 if isinstance(self.target_type, GlobalTargetType)
142 else "QualifiedTargetType",
143 "target_type": str(self.target_type),
144 "target": self.target,
145 "recipient_type": str(self.recipient_type),
146 "recipient": self.recipient,
147 "reason": self.reason,
148 "snoozed_at": self.snoozed_at.isoformat() if self.snoozed_at else None,
149 "snooze_until": self.snooze_until.isoformat() if self.snooze_until else None,
150 }
152 @classmethod
153 def from_storage_dict(cls, data: dict[str, Any]) -> Snooze | None:
154 try:
155 target_type_cls = GlobalTargetType if data["target_type_class"] == "GlobalTargetType" else QualifiedTargetType
156 snooze = cls(
157 target_type_cls(data["target_type"]),
158 RecipientType(data["recipient_type"]),
159 data.get("target"),
160 data.get("recipient"),
161 reason=data.get("reason"),
162 )
163 if data.get("snoozed_at"):
164 snoozed_at = dt_util.parse_datetime(data["snoozed_at"])
165 if snoozed_at:
166 snooze.snoozed_at = snoozed_at
167 snooze.snooze_until = dt_util.parse_datetime(data["snooze_until"]) if data.get("snooze_until") else None
168 except Exception as e:
169 _LOGGER.warning("SUPERNOTIFY Discarding invalid persisted snooze %s: %s", data, e)
170 return None
171 else:
172 return snooze
175class Snoozer:
176 """Manage snoozing"""
178 def __init__(self, config: dict[str, Any] | None = None, people_registry: PeopleRegistry | None = None) -> None:
179 self.snoozes: dict[str, Snooze] = {}
180 self.people_registry: PeopleRegistry | None = people_registry
181 self.config: dict[str, Any] = config or {}
182 self.snooze_period = timedelta(seconds=self.config.get(CONF_SNOOZE_TIME, 60 * 60))
183 self.hass_api: HomeAssistantAPI | None = None
185 async def initialize(self, hass_api: HomeAssistantAPI) -> None:
186 """Restore any snoozes persisted from a previous run - HA restarts and reloads no
187 longer silently lose active snoozes/silences. Expired ones are dropped on restore."""
188 self.hass_api = hass_api
189 stored: list[dict[str, Any]] | None = await hass_api.load_storage(STORAGE_KEY, STORAGE_VERSION)
190 if not stored:
191 return
192 restored = 0
193 for entry in stored:
194 snooze: Snooze | None = Snooze.from_storage_dict(entry)
195 if snooze and snooze.active():
196 self.snoozes[snooze.short_key()] = snooze
197 restored += 1
198 if restored:
199 _LOGGER.info("SUPERNOTIFY Restored %s snooze(s) from storage", restored)
201 def _persist(self) -> None:
202 if self.hass_api is not None:
203 self.hass_api.save_storage(STORAGE_KEY, [s.to_storage_dict() for s in self.snoozes.values()], STORAGE_VERSION)
205 def handle_command_event(self, event: Event, people: list[Recipient] | None = None) -> None:
206 people = people or []
207 try:
208 cmd: CommandType
209 target_type: TargetType | None = None
210 target: str | None = None
211 snooze_for: timedelta = self.snooze_period
212 recipient_type: RecipientType | None = None
213 event_name: str | None = event.data.get(ATTR_ACTION)
215 if not event_name:
216 _LOGGER.warning(
217 "SUPERNOTIFY Invalid Mobile Action: %s, %s, %s, %s",
218 event.origin,
219 event.time_fired,
220 event.data,
221 event.context,
222 )
223 return
225 _LOGGER.debug(
226 "SUPERNOTIFY Mobile Action: %s, %s, %s, %s", event.origin, event.time_fired, event.data, event.context
227 )
228 event_parts: list[str] = event_name.split("_")
229 if len(event_parts) < 4:
230 _LOGGER.warning("SUPERNOTIFY Malformed mobile event action %s", event_name)
231 return
232 cmd = CommandType[event_parts[1]]
233 recipient_type = RecipientType[event_parts[2]]
234 # a text input reply carries the minutes, so the whole remainder is the target;
235 # otherwise a trailing number is minutes, which is ambiguous for names ending in _<digits>
236 reply_text: str | None = event.data.get(ATTR_REPLY_TEXT)
237 reply_minutes: int | None = int(reply_text) if reply_text and reply_text.strip().isdigit() else None
238 if event_parts[3] in QualifiedTargetType and len(event_parts) > 4:
239 target_type = QualifiedTargetType[event_parts[3]]
240 target_parts: list[str] = event_parts[4:]
241 if reply_text is None and len(target_parts) > 1 and target_parts[-1].isdigit():
242 reply_minutes = int(target_parts[-1])
243 target_parts = target_parts[:-1]
244 target = "_".join(target_parts)
245 elif event_parts[3] in GlobalTargetType and len(event_parts) >= 4:
246 target_type = GlobalTargetType[event_parts[3]]
247 if reply_text is None and len(event_parts) == 5:
248 reply_minutes = int(event_parts[-1])
249 if reply_minutes:
250 snooze_for = timedelta(minutes=reply_minutes)
252 if cmd is None or target_type is None or recipient_type is None:
253 _LOGGER.warning("SUPERNOTIFY Invalid mobile event name %s", event_name)
254 return
256 except KeyError as ke:
257 _LOGGER.warning("SUPERNOTIFY Unknown enum in event %s: %s", event, ke)
258 return
259 except Exception as e:
260 _LOGGER.warning("SUPERNOTIFY Unable to analyze event %s: %s", event, e)
261 return
263 try:
264 recipient: str | None = None
265 if recipient_type == RecipientType.USER:
266 target_people: list[str] = [
267 p.entity_id
268 for p in people
269 if p.user_id == event.context.user_id and event.context.user_id is not None and p.entity_id
270 ]
271 if target_people:
272 recipient = target_people[0]
273 _LOGGER.debug("SUPERNOTIFY Mobile action from %s mapped to %s", event.context.user_id, recipient)
274 else:
275 _LOGGER.warning("SUPERNOTIFY Unable to find person for action from %s", event.context.user_id)
276 return
278 self.register_snooze(cmd, target_type, target, recipient_type, recipient, snooze_for)
280 except Exception as e:
281 _LOGGER.warning("SUPERNOTIFY Unable to handle event %s: %s", event, e)
283 def register_snooze(
284 self,
285 cmd: CommandType,
286 target_type: TargetType,
287 target: str | None,
288 recipient_type: RecipientType,
289 recipient: str | None,
290 snooze_for: timedelta | None,
291 reason: str = "User command",
292 ) -> None:
293 if target_type == QualifiedTargetType.TAG and target:
294 # so the same tag however it was said or typed is one snooze, and can be resumed
295 target = spoken_name(target)
296 if target_type == QualifiedTargetType.PRIORITY:
297 target = canonical_priority(target)
298 if cmd == CommandType.SNOOZE:
299 snooze = Snooze(target_type, recipient_type, target, recipient, snooze_for, reason=reason)
300 self.snoozes[snooze.short_key()] = snooze
301 self._persist()
302 elif cmd == CommandType.SILENCE:
303 snooze = Snooze(target_type, recipient_type, target, recipient, reason=reason)
304 self.snoozes[snooze.short_key()] = snooze
305 self._persist()
306 elif cmd == CommandType.NORMAL:
307 anti_snooze = Snooze(target_type, recipient_type, target, recipient)
308 to_del: list[str] = [k for k, v in self.snoozes.items() if v.short_key() == anti_snooze.short_key()]
309 for k in to_del:
310 del self.snoozes[k]
311 if to_del:
312 self._persist()
313 else:
314 _LOGGER.warning( # type: ignore
315 "SUPERNOTIFY Invalid mobile cmd %s (target_type: %s, target: %s, recipient_type: %s)",
316 cmd,
317 target_type,
318 target,
319 recipient_type,
320 )
322 def purge_snoozes(self) -> None:
323 to_del: list[str] = [k for k, v in self.snoozes.items() if not v.active() or self._mobile_app_gone(v)]
324 for k in to_del:
325 del self.snoozes[k]
326 if to_del:
327 self._persist()
329 def _mobile_app_gone(self, snooze: Snooze) -> bool:
330 """A snooze on a mobile device whose notify action no longer exists, e.g. re-paired under a new name"""
331 return (
332 snooze.target_type == QualifiedTargetType.MOBILE
333 and isinstance(snooze.target, str)
334 and self.hass_api is not None
335 and not self.hass_api.mobile_app_action_exists(snooze.target)
336 )
338 def clear(self) -> int:
339 cleared: int = len(self.snoozes)
340 self.snoozes.clear()
341 if cleared:
342 self._persist()
343 return cleared
345 def export(self) -> list[dict[str, Any]]:
346 """Active snoozes only: expired ones stay in memory until the nightly purge, and their
347 time-only snooze_until would otherwise read as a snooze still to come."""
348 return [s.export() for s in self.snoozes.values() if s.active()]
350 def current_snoozes(self, priority: str, delivery: Delivery) -> list[Snooze]:
351 inscope_snoozes: list[Snooze] = []
353 for snooze in self.snoozes.values():
354 if snooze.active():
355 match snooze.target_type:
356 case GlobalTargetType.EVERYTHING:
357 inscope_snoozes.append(snooze)
358 case GlobalTargetType.NONCRITICAL:
359 if priority != PRIORITY_CRITICAL:
360 inscope_snoozes.append(snooze)
361 case QualifiedTargetType.DELIVERY:
362 if snooze.target == delivery.name:
363 inscope_snoozes.append(snooze)
364 case QualifiedTargetType.PRIORITY:
365 if snooze.target == priority:
366 inscope_snoozes.append(snooze)
367 case QualifiedTargetType.MOBILE:
368 inscope_snoozes.append(snooze)
369 case QualifiedTargetType.TRANSPORT:
370 if snooze.target == delivery.transport.name:
371 inscope_snoozes.append(snooze)
372 case QualifiedTargetType.CAMERA | QualifiedTargetType.TAG:
373 inscope_snoozes.append(snooze)
374 case _:
375 _LOGGER.warning("SUPERNOTIFY Unhandled target type %s", snooze.target_type)
377 return inscope_snoozes
379 def has_tag_snooze(self) -> bool:
380 return any(s.active() and s.target_type == QualifiedTargetType.TAG for s in self.snoozes.values())
382 def is_global_snooze(self, priority: str = PRIORITY_MEDIUM) -> bool:
383 for snooze in self.snoozes.values():
384 if snooze.active() and snooze.recipient_type == RecipientType.EVERYONE:
385 match snooze.target_type:
386 case GlobalTargetType.EVERYTHING:
387 return True
388 case GlobalTargetType.NONCRITICAL:
389 if priority != PRIORITY_CRITICAL:
390 return True
392 return False
394 def is_delivery_snoozed(
395 self,
396 priority: str,
397 delivery: Delivery,
398 camera_entity_id: str | None = None,
399 tags: set[str] | None = None,
400 ) -> bool:
401 """Everyone-scoped delivery, transport, priority, camera or tag snoozes stop the whole delivery"""
402 for snooze in self.current_snoozes(priority, delivery):
403 if snooze.recipient_type != RecipientType.EVERYONE:
404 continue
405 if snooze.target_type in (
406 QualifiedTargetType.DELIVERY,
407 QualifiedTargetType.TRANSPORT,
408 QualifiedTargetType.PRIORITY,
409 ):
410 return True
411 if _snoozes_subject(snooze, camera_entity_id, tags):
412 return True
413 return False
415 def filter_recipients(
416 self,
417 recipients: Target,
418 priority: str,
419 delivery: Delivery,
420 camera_entity_id: str | None = None,
421 tags: set[str] | None = None,
422 ) -> Target:
423 inscope_snoozes: list[Snooze] = self.current_snoozes(priority, delivery)
424 for snooze in inscope_snoozes:
425 # everyone-scoped snoozes are checked by is_global_snooze and is_delivery_snoozed
426 if snooze.recipient_type == RecipientType.USER and (
427 (snooze.target_type == QualifiedTargetType.DELIVERY and snooze.target == delivery.name)
428 or (snooze.target_type == QualifiedTargetType.TRANSPORT and snooze.target == delivery.transport.name)
429 or (
430 snooze.target_type == QualifiedTargetType.PRIORITY
431 and (snooze.target == priority or (isinstance(snooze.target, list) and priority in snooze.target))
432 )
433 or _snoozes_subject(snooze, camera_entity_id, tags)
434 or snooze.target_type == GlobalTargetType.EVERYTHING
435 or (snooze.target_type == GlobalTargetType.NONCRITICAL and priority != PRIORITY_CRITICAL)
436 ):
437 recipients_to_remove: list[str] = []
438 for recipient in recipients.person_ids:
439 if recipient == snooze.recipient:
440 recipients_to_remove.append(recipient)
441 _LOGGER.info("SUPERNOTIFY Snoozing %s", snooze.recipient)
443 recipients.remove(ATTR_PERSON_ID, recipients_to_remove)
445 return self.filter_mobile_app_ids(recipients, priority, delivery)
447 def filter_mobile_app_ids(self, target: Target, priority: str, delivery: Delivery) -> Target:
448 """Drop snoozed mobile devices, leaving everything else - person_ids included - in place
450 Applied to the action's own targets, and again to each recipient's addresses as the
451 person is resolved, since that's the first point a person's mobile devices are known
452 """
453 snoozed: set[str] = {
454 str(s.target)
455 for s in self.current_snoozes(priority, delivery)
456 if s.target_type == QualifiedTargetType.MOBILE and s.target
457 }
458 if not snoozed:
459 return target
460 to_remove: list[str] = [m for m in target.mobile_app_ids if m.removeprefix("notify.") in snoozed]
461 if not to_remove:
462 return target
463 _LOGGER.debug("SUPERNOTIFY Snoozing mobile devices %s", to_remove)
464 return target - Target({ATTR_MOBILE_APP_ID: to_remove})
467def _snoozes_subject(snooze: Snooze, camera_entity_id: str | None, tags: set[str] | None) -> bool:
468 """Whether a camera or tag snooze is about this notification - tags are as spoken_name() has them"""
469 match snooze.target_type:
470 case QualifiedTargetType.CAMERA:
471 return camera_entity_id is not None and snooze.target == camera_entity_id
472 case QualifiedTargetType.TAG:
473 return isinstance(snooze.target, str) and spoken_name(snooze.target) in (tags or set())
474 return False