Coverage for custom_components/supernotify/snoozer.py: 96%
234 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 14:29 +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 .const import (
12 ATTR_ACTION,
13 ATTR_MOBILE_APP_ID,
14 ATTR_PERSON_ID,
15 ATTR_REPLY_TEXT,
16 CONF_SNOOZE_TIME,
17 PRIORITY_CRITICAL,
18 PRIORITY_MEDIUM,
19)
20from .model import CommandType, GlobalTargetType, QualifiedTargetType, RecipientType, Target, TargetType
22if TYPE_CHECKING:
23 from homeassistant.core import Event
25 from .delivery import Delivery
26 from .hass_api import HomeAssistantAPI
27 from .people import PeopleRegistry, Recipient
29_LOGGER = logging.getLogger(__name__)
31STORAGE_VERSION = 1
32STORAGE_KEY = f"{DOMAIN}.snoozes"
35class Snooze:
36 target: str | list[str] | None
37 target_type: TargetType
38 snoozed_at: dt.datetime
39 snooze_until: dt.datetime | None = None
40 recipient_type: RecipientType
41 recipient: str | None
42 reason: str | None = None
44 def __init__(
45 self,
46 target_type: TargetType,
47 recipient_type: RecipientType,
48 target: str | list[str] | None = None,
49 recipient: str | None = None,
50 snooze_for: timedelta | None = None,
51 reason: str | None = None,
52 ) -> None:
53 self.snoozed_at = dt_util.now()
54 self.target = target
55 self.target_type = target_type
56 self.recipient_type: RecipientType = recipient_type
57 self.recipient = recipient
58 self.reason = reason
59 self.snooze_until = None
60 if snooze_for:
61 self.snooze_until = self.snoozed_at + snooze_for
63 def std_recipient(self) -> str | None:
64 return self.recipient if self.recipient_type == RecipientType.USER else RecipientType.EVERYONE
66 def short_key(self) -> str:
67 # only one GLOBAL can be active at a time
68 target = "GLOBAL" if self.target_type in GlobalTargetType else f"{self.target_type}_{self.target}"
69 return f"{target}_{self.std_recipient()}"
71 def __eq__(self, other: object) -> bool:
72 """Check if two snoozes for the same thing"""
73 if not isinstance(other, Snooze):
74 return False
75 return self.short_key() == other.short_key()
77 def __repr__(self) -> str:
78 """Return a string representation of the object."""
79 target = "GLOBAL" if self.target_type in GlobalTargetType else f"{self.target_type}_{self.target}"
80 return f"Snooze({target}, {self.std_recipient()})"
82 def active(self) -> bool:
83 return self.snooze_until is None or self.snooze_until > dt_util.now()
85 def export(self) -> dict[str, Any]:
86 return {
87 "target_type": self.target_type,
88 "target": self.target,
89 "recipient_type": self.recipient_type,
90 "recipient": self.recipient,
91 "reason": self.reason,
92 "snoozed_at": dt_util.as_local(self.snoozed_at).strftime("%H:%M:%S") if self.snoozed_at else None,
93 "snooze_until": dt_util.as_local(self.snooze_until).strftime("%H:%M:%S") if self.snooze_until else None,
94 }
96 def to_storage_dict(self) -> dict[str, Any]:
97 """Full-fidelity serialization for persistence (unlike export(), which is a display
98 summary that loses the date part of timestamps)."""
99 return {
100 "target_type_class": "GlobalTargetType"
101 if isinstance(self.target_type, GlobalTargetType)
102 else "QualifiedTargetType",
103 "target_type": str(self.target_type),
104 "target": self.target,
105 "recipient_type": str(self.recipient_type),
106 "recipient": self.recipient,
107 "reason": self.reason,
108 "snoozed_at": self.snoozed_at.isoformat() if self.snoozed_at else None,
109 "snooze_until": self.snooze_until.isoformat() if self.snooze_until else None,
110 }
112 @classmethod
113 def from_storage_dict(cls, data: dict[str, Any]) -> Snooze | None:
114 try:
115 target_type_cls = GlobalTargetType if data["target_type_class"] == "GlobalTargetType" else QualifiedTargetType
116 snooze = cls(
117 target_type_cls(data["target_type"]),
118 RecipientType(data["recipient_type"]),
119 data.get("target"),
120 data.get("recipient"),
121 reason=data.get("reason"),
122 )
123 if data.get("snoozed_at"):
124 snoozed_at = dt_util.parse_datetime(data["snoozed_at"])
125 if snoozed_at:
126 snooze.snoozed_at = snoozed_at
127 snooze.snooze_until = dt_util.parse_datetime(data["snooze_until"]) if data.get("snooze_until") else None
128 except Exception as e:
129 _LOGGER.warning("SUPERNOTIFY Discarding invalid persisted snooze %s: %s", data, e)
130 return None
131 else:
132 return snooze
135class Snoozer:
136 """Manage snoozing"""
138 def __init__(self, config: dict[str, Any] | None = None, people_registry: PeopleRegistry | None = None) -> None:
139 self.snoozes: dict[str, Snooze] = {}
140 self.people_registry: PeopleRegistry | None = people_registry
141 self.config: dict[str, Any] = config or {}
142 self.snooze_period = timedelta(seconds=self.config.get(CONF_SNOOZE_TIME, 60 * 60))
143 self.hass_api: HomeAssistantAPI | None = None
145 async def initialize(self, hass_api: HomeAssistantAPI) -> None:
146 """Restore any snoozes persisted from a previous run - HA restarts and reloads no
147 longer silently lose active snoozes/silences. Expired ones are dropped on restore."""
148 self.hass_api = hass_api
149 stored: list[dict[str, Any]] | None = await hass_api.load_storage(STORAGE_KEY, STORAGE_VERSION)
150 if not stored:
151 return
152 restored = 0
153 for entry in stored:
154 snooze: Snooze | None = Snooze.from_storage_dict(entry)
155 if snooze and snooze.active():
156 self.snoozes[snooze.short_key()] = snooze
157 restored += 1
158 if restored:
159 _LOGGER.info("SUPERNOTIFY Restored %s snooze(s) from storage", restored)
161 def _persist(self) -> None:
162 if self.hass_api is not None:
163 self.hass_api.save_storage(STORAGE_KEY, [s.to_storage_dict() for s in self.snoozes.values()], STORAGE_VERSION)
165 def handle_command_event(self, event: Event, people: list[Recipient] | None = None) -> None:
166 people = people or []
167 try:
168 cmd: CommandType
169 target_type: TargetType | None = None
170 target: str | None = None
171 snooze_for: timedelta = self.snooze_period
172 recipient_type: RecipientType | None = None
173 event_name: str | None = event.data.get(ATTR_ACTION)
175 if not event_name:
176 _LOGGER.warning(
177 "SUPERNOTIFY Invalid Mobile Action: %s, %s, %s, %s",
178 event.origin,
179 event.time_fired,
180 event.data,
181 event.context,
182 )
183 return
185 _LOGGER.debug(
186 "SUPERNOTIFY Mobile Action: %s, %s, %s, %s", event.origin, event.time_fired, event.data, event.context
187 )
188 event_parts: list[str] = event_name.split("_")
189 if len(event_parts) < 4:
190 _LOGGER.warning("SUPERNOTIFY Malformed mobile event action %s", event_name)
191 return
192 cmd = CommandType[event_parts[1]]
193 recipient_type = RecipientType[event_parts[2]]
194 # a text input reply carries the minutes, so the whole remainder is the target;
195 # otherwise a trailing number is minutes, which is ambiguous for names ending in _<digits>
196 reply_text: str | None = event.data.get(ATTR_REPLY_TEXT)
197 reply_minutes: int | None = int(reply_text) if reply_text and reply_text.strip().isdigit() else None
198 if event_parts[3] in QualifiedTargetType and len(event_parts) > 4:
199 target_type = QualifiedTargetType[event_parts[3]]
200 target_parts: list[str] = event_parts[4:]
201 if reply_text is None and len(target_parts) > 1 and target_parts[-1].isdigit():
202 reply_minutes = int(target_parts[-1])
203 target_parts = target_parts[:-1]
204 target = "_".join(target_parts)
205 elif event_parts[3] in GlobalTargetType and len(event_parts) >= 4:
206 target_type = GlobalTargetType[event_parts[3]]
207 if reply_text is None and len(event_parts) == 5:
208 reply_minutes = int(event_parts[-1])
209 if reply_minutes:
210 snooze_for = timedelta(minutes=reply_minutes)
212 if cmd is None or target_type is None or recipient_type is None:
213 _LOGGER.warning("SUPERNOTIFY Invalid mobile event name %s", event_name)
214 return
216 except KeyError as ke:
217 _LOGGER.warning("SUPERNOTIFY Unknown enum in event %s: %s", event, ke)
218 return
219 except Exception as e:
220 _LOGGER.warning("SUPERNOTIFY Unable to analyze event %s: %s", event, e)
221 return
223 try:
224 recipient: str | None = None
225 if recipient_type == RecipientType.USER:
226 target_people: list[str] = [
227 p.entity_id
228 for p in people
229 if p.user_id == event.context.user_id and event.context.user_id is not None and p.entity_id
230 ]
231 if target_people:
232 recipient = target_people[0]
233 _LOGGER.debug("SUPERNOTIFY Mobile action from %s mapped to %s", event.context.user_id, recipient)
234 else:
235 _LOGGER.warning("SUPERNOTIFY Unable to find person for action from %s", event.context.user_id)
236 return
238 self.register_snooze(cmd, target_type, target, recipient_type, recipient, snooze_for)
240 except Exception as e:
241 _LOGGER.warning("SUPERNOTIFY Unable to handle event %s: %s", event, e)
243 def register_snooze(
244 self,
245 cmd: CommandType,
246 target_type: TargetType,
247 target: str | None,
248 recipient_type: RecipientType,
249 recipient: str | None,
250 snooze_for: timedelta | None,
251 reason: str = "User command",
252 ) -> None:
253 if cmd == CommandType.SNOOZE:
254 snooze = Snooze(target_type, recipient_type, target, recipient, snooze_for, reason=reason)
255 self.snoozes[snooze.short_key()] = snooze
256 self._persist()
257 elif cmd == CommandType.SILENCE:
258 snooze = Snooze(target_type, recipient_type, target, recipient, reason=reason)
259 self.snoozes[snooze.short_key()] = snooze
260 self._persist()
261 elif cmd == CommandType.NORMAL:
262 anti_snooze = Snooze(target_type, recipient_type, target, recipient)
263 to_del: list[str] = [k for k, v in self.snoozes.items() if v.short_key() == anti_snooze.short_key()]
264 for k in to_del:
265 del self.snoozes[k]
266 if to_del:
267 self._persist()
268 else:
269 _LOGGER.warning( # type: ignore
270 "SUPERNOTIFY Invalid mobile cmd %s (target_type: %s, target: %s, recipient_type: %s)",
271 cmd,
272 target_type,
273 target,
274 recipient_type,
275 )
277 def purge_snoozes(self) -> None:
278 to_del: list[str] = [k for k, v in self.snoozes.items() if not v.active()]
279 for k in to_del:
280 del self.snoozes[k]
281 if to_del:
282 self._persist()
284 def clear(self) -> int:
285 cleared: int = len(self.snoozes)
286 self.snoozes.clear()
287 if cleared:
288 self._persist()
289 return cleared
291 def export(self) -> list[dict[str, Any]]:
292 return [s.export() for s in self.snoozes.values()]
294 def current_snoozes(self, priority: str, delivery: Delivery) -> list[Snooze]:
295 inscope_snoozes: list[Snooze] = []
297 for snooze in self.snoozes.values():
298 if snooze.active():
299 match snooze.target_type:
300 case GlobalTargetType.EVERYTHING:
301 inscope_snoozes.append(snooze)
302 case GlobalTargetType.NONCRITICAL:
303 if priority != PRIORITY_CRITICAL:
304 inscope_snoozes.append(snooze)
305 case QualifiedTargetType.DELIVERY:
306 if snooze.target == delivery.name:
307 inscope_snoozes.append(snooze)
308 case QualifiedTargetType.PRIORITY:
309 if snooze.target == priority:
310 inscope_snoozes.append(snooze)
311 case QualifiedTargetType.MOBILE:
312 inscope_snoozes.append(snooze)
313 case QualifiedTargetType.TRANSPORT:
314 if snooze.target == delivery.transport.name:
315 inscope_snoozes.append(snooze)
316 case QualifiedTargetType.CAMERA:
317 inscope_snoozes.append(snooze)
318 case _:
319 _LOGGER.warning("SUPERNOTIFY Unhandled target type %s", snooze.target_type)
321 return inscope_snoozes
323 def is_global_snooze(self, priority: str = PRIORITY_MEDIUM) -> bool:
324 for snooze in self.snoozes.values():
325 if snooze.active() and snooze.recipient_type == RecipientType.EVERYONE:
326 match snooze.target_type:
327 case GlobalTargetType.EVERYTHING:
328 return True
329 case GlobalTargetType.NONCRITICAL:
330 if priority != PRIORITY_CRITICAL:
331 return True
333 return False
335 def is_delivery_snoozed(self, priority: str, delivery: Delivery, camera_entity_id: str | None = None) -> bool:
336 """Everyone-scoped delivery, transport, priority or camera snoozes stop the whole delivery"""
337 for snooze in self.current_snoozes(priority, delivery):
338 if snooze.recipient_type != RecipientType.EVERYONE:
339 continue
340 if snooze.target_type in (
341 QualifiedTargetType.DELIVERY,
342 QualifiedTargetType.TRANSPORT,
343 QualifiedTargetType.PRIORITY,
344 ):
345 return True
346 if snooze.target_type == QualifiedTargetType.CAMERA and camera_entity_id and snooze.target == camera_entity_id:
347 return True
348 return False
350 def filter_recipients(
351 self, recipients: Target, priority: str, delivery: Delivery, camera_entity_id: str | None = None
352 ) -> Target:
353 inscope_snoozes: list[Snooze] = self.current_snoozes(priority, delivery)
354 for snooze in inscope_snoozes:
355 # everyone-scoped snoozes are checked by is_global_snooze and is_delivery_snoozed
356 if snooze.recipient_type == RecipientType.USER and (
357 (snooze.target_type == QualifiedTargetType.DELIVERY and snooze.target == delivery.name)
358 or (snooze.target_type == QualifiedTargetType.TRANSPORT and snooze.target == delivery.transport.name)
359 or (
360 snooze.target_type == QualifiedTargetType.PRIORITY
361 and (snooze.target == priority or (isinstance(snooze.target, list) and priority in snooze.target))
362 )
363 or (
364 snooze.target_type == QualifiedTargetType.CAMERA
365 and camera_entity_id is not None
366 and snooze.target == camera_entity_id
367 )
368 or snooze.target_type == GlobalTargetType.EVERYTHING
369 or (snooze.target_type == GlobalTargetType.NONCRITICAL and priority != PRIORITY_CRITICAL)
370 ):
371 recipients_to_remove: list[str] = []
372 for recipient in recipients.person_ids:
373 if recipient == snooze.recipient:
374 recipients_to_remove.append(recipient)
375 _LOGGER.info("SUPERNOTIFY Snoozing %s", snooze.recipient)
377 recipients.remove(ATTR_PERSON_ID, recipients_to_remove)
379 if snooze.target_type == QualifiedTargetType.MOBILE:
380 to_remove: list[str] = []
381 for recipient in recipients.mobile_app_ids:
382 if recipient == snooze.target:
383 _LOGGER.debug("SUPERNOTIFY Snoozing %s for %s", snooze.std_recipient(), snooze.target)
384 to_remove.append(recipient)
385 if to_remove:
386 recipients.remove(ATTR_MOBILE_APP_ID, to_remove)
387 return recipients