Coverage for custom_components / supernotify / snoozer.py: 95%
174 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-11 22:18 +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 .const import ATTR_ACTION, ATTR_MOBILE_APP_ID, ATTR_PERSON_ID, CONF_SNOOZE_TIME, PRIORITY_CRITICAL, PRIORITY_MEDIUM
11from .model import CommandType, GlobalTargetType, QualifiedTargetType, RecipientType, Target, TargetType
13if TYPE_CHECKING:
14 from homeassistant.core import Event
16 from .delivery import Delivery
17 from .people import PeopleRegistry, Recipient
19_LOGGER = logging.getLogger(__name__)
22class Snooze:
23 target: str | list[str] | None
24 target_type: TargetType
25 snoozed_at: dt.datetime
26 snooze_until: dt.datetime | None = None
27 recipient_type: RecipientType
28 recipient: str | None
29 reason: str | None = None
31 def __init__(
32 self,
33 target_type: TargetType,
34 recipient_type: RecipientType,
35 target: str | list[str] | None = None,
36 recipient: str | None = None,
37 snooze_for: timedelta | None = None,
38 reason: str | None = None,
39 ) -> None:
40 self.snoozed_at = dt_util.now()
41 self.target = target
42 self.target_type = target_type
43 self.recipient_type: RecipientType = recipient_type
44 self.recipient = recipient
45 self.reason = reason
46 self.snooze_until = None
47 if snooze_for:
48 self.snooze_until = self.snoozed_at + snooze_for
50 def std_recipient(self) -> str | None:
51 return self.recipient if self.recipient_type == RecipientType.USER else RecipientType.EVERYONE
53 def short_key(self) -> str:
54 # only one GLOBAL can be active at a time
55 target = "GLOBAL" if self.target_type in GlobalTargetType else f"{self.target_type}_{self.target}"
56 return f"{target}_{self.std_recipient()}"
58 def __eq__(self, other: object) -> bool:
59 """Check if two snoozes for the same thing"""
60 if not isinstance(other, Snooze):
61 return False
62 return self.short_key() == other.short_key()
64 def __repr__(self) -> str:
65 """Return a string representation of the object."""
66 target = "GLOBAL" if self.target_type in GlobalTargetType else f"{self.target_type}_{self.target}"
67 return f"Snooze({target}, {self.std_recipient()})"
69 def active(self) -> bool:
70 return self.snooze_until is None or self.snooze_until > dt_util.now()
72 def export(self) -> dict[str, Any]:
73 return {
74 "target_type": self.target_type,
75 "target": self.target,
76 "recipient_type": self.recipient_type,
77 "recipient": self.recipient,
78 "reason": self.reason,
79 "snoozed_at": dt_util.as_local(self.snoozed_at).strftime("%H:%M:%S") if self.snoozed_at else None,
80 "snooze_until": dt_util.as_local(self.snooze_until).strftime("%H:%M:%S") if self.snooze_until else None,
81 }
84class Snoozer:
85 """Manage snoozing"""
87 def __init__(self, config: dict[str, Any] | None = None, people_registry: PeopleRegistry | None = None) -> None:
88 self.snoozes: dict[str, Snooze] = {}
89 self.people_registry: PeopleRegistry | None = people_registry
90 self.config = config or {}
91 self.snooze_period = timedelta(seconds=self.config.get(CONF_SNOOZE_TIME, 60 * 60))
93 def handle_command_event(self, event: Event, people: list[Recipient] | None = None) -> None:
94 people = people or []
95 try:
96 cmd: CommandType
97 target_type: TargetType | None = None
98 target: str | None = None
99 snooze_for: timedelta = self.snooze_period
100 recipient_type: RecipientType | None = None
101 event_name = event.data.get(ATTR_ACTION)
103 if not event_name:
104 _LOGGER.warning(
105 "SUPERNOTIFY Invalid Mobile Action: %s, %s, %s, %s",
106 event.origin,
107 event.time_fired,
108 event.data,
109 event.context,
110 )
111 return
113 _LOGGER.debug(
114 "SUPERNOTIFY Mobile Action: %s, %s, %s, %s", event.origin, event.time_fired, event.data, event.context
115 )
116 event_parts: list[str] = event_name.split("_")
117 if len(event_parts) < 4:
118 _LOGGER.warning("SUPERNOTIFY Malformed mobile event action %s", event_name)
119 return
120 cmd = CommandType[event_parts[1]]
121 recipient_type = RecipientType[event_parts[2]]
122 if event_parts[3] in QualifiedTargetType and len(event_parts) > 4:
123 target_type = QualifiedTargetType[event_parts[3]]
124 target = event_parts[4]
125 snooze_for = timedelta(minutes=int(event_parts[-1])) if len(event_parts) == 6 else self.snooze_period
126 elif event_parts[3] in GlobalTargetType and len(event_parts) >= 4:
127 target_type = GlobalTargetType[event_parts[3]]
128 snooze_for = timedelta(minutes=int(event_parts[-1])) if len(event_parts) == 5 else self.snooze_period
130 if cmd is None or target_type is None or recipient_type is None:
131 _LOGGER.warning("SUPERNOTIFY Invalid mobile event name %s", event_name)
132 return
134 except KeyError as ke:
135 _LOGGER.warning("SUPERNOTIFY Unknown enum in event %s: %s", event, ke)
136 return
137 except Exception as e:
138 _LOGGER.warning("SUPERNOTIFY Unable to analyze event %s: %s", event, e)
139 return
141 try:
142 recipient: str | None = None
143 if recipient_type == RecipientType.USER:
144 target_people = [
145 p.entity_id
146 for p in people
147 if p.user_id == event.context.user_id and event.context.user_id is not None and p.entity_id
148 ]
149 if target_people:
150 recipient = target_people[0]
151 _LOGGER.debug("SUPERNOTIFY mobile action from %s mapped to %s", event.context.user_id, recipient)
152 else:
153 _LOGGER.warning("SUPERNOTIFY Unable to find person for action from %s", event.context.user_id)
154 return
156 self.register_snooze(cmd, target_type, target, recipient_type, recipient, snooze_for)
158 except Exception as e:
159 _LOGGER.warning("SUPERNOTIFY Unable to handle event %s: %s", event, e)
161 def register_snooze(
162 self,
163 cmd: CommandType,
164 target_type: TargetType,
165 target: str | None,
166 recipient_type: RecipientType,
167 recipient: str | None,
168 snooze_for: timedelta | None,
169 reason: str = "User command",
170 ) -> None:
171 if cmd == CommandType.SNOOZE:
172 snooze = Snooze(target_type, recipient_type, target, recipient, snooze_for, reason=reason)
173 self.snoozes[snooze.short_key()] = snooze
174 elif cmd == CommandType.SILENCE:
175 snooze = Snooze(target_type, recipient_type, target, recipient, reason=reason)
176 self.snoozes[snooze.short_key()] = snooze
177 elif cmd == CommandType.NORMAL:
178 anti_snooze = Snooze(target_type, recipient_type, target, recipient)
179 to_del = [k for k, v in self.snoozes.items() if v.short_key() == anti_snooze.short_key()]
180 for k in to_del:
181 del self.snoozes[k]
182 else:
183 _LOGGER.warning( # type: ignore
184 "SUPERNOTIFY Invalid mobile cmd %s (target_type: %s, target: %s, recipient_type: %s)",
185 cmd,
186 target_type,
187 target,
188 recipient_type,
189 )
191 def purge_snoozes(self) -> None:
192 to_del = [k for k, v in self.snoozes.items() if not v.active()]
193 for k in to_del:
194 del self.snoozes[k]
196 def clear(self) -> int:
197 cleared = len(self.snoozes)
198 self.snoozes.clear()
199 return cleared
201 def export(self) -> list[dict[str, Any]]:
202 return [s.export() for s in self.snoozes.values()]
204 def current_snoozes(self, priority: str, delivery: Delivery) -> list[Snooze]:
205 inscope_snoozes: list[Snooze] = []
207 for snooze in self.snoozes.values():
208 if snooze.active():
209 match snooze.target_type:
210 case GlobalTargetType.EVERYTHING:
211 inscope_snoozes.append(snooze)
212 case GlobalTargetType.NONCRITICAL:
213 if priority != PRIORITY_CRITICAL:
214 inscope_snoozes.append(snooze)
215 case QualifiedTargetType.DELIVERY:
216 if snooze.target == delivery.name:
217 inscope_snoozes.append(snooze)
218 case QualifiedTargetType.PRIORITY:
219 if snooze.target == priority:
220 inscope_snoozes.append(snooze)
221 case QualifiedTargetType.MOBILE:
222 inscope_snoozes.append(snooze)
223 case QualifiedTargetType.TRANSPORT:
224 if snooze.target == delivery.transport.name:
225 inscope_snoozes.append(snooze)
226 case QualifiedTargetType.CAMERA:
227 inscope_snoozes.append(snooze)
228 case _:
229 _LOGGER.warning("SUPERNOTIFY Unhandled target type %s", snooze.target_type)
231 return inscope_snoozes
233 def is_global_snooze(self, priority: str = PRIORITY_MEDIUM) -> bool:
234 for snooze in self.snoozes.values():
235 if snooze.active():
236 match snooze.target_type:
237 case GlobalTargetType.EVERYTHING:
238 return True
239 case GlobalTargetType.NONCRITICAL:
240 if priority != PRIORITY_CRITICAL:
241 return True
243 return False
245 def filter_recipients(self, recipients: Target, priority: str, delivery: Delivery) -> Target:
246 inscope_snoozes = self.current_snoozes(priority, delivery)
247 for snooze in inscope_snoozes:
248 if snooze.recipient_type == RecipientType.USER:
249 # assume the everyone checks are made before notification gets this far
250 if (
251 (snooze.target_type == QualifiedTargetType.DELIVERY and snooze.target == delivery.name)
252 or (snooze.target_type == QualifiedTargetType.TRANSPORT and snooze.target == delivery.transport.name)
253 or (
254 snooze.target_type == QualifiedTargetType.PRIORITY
255 and (snooze.target == priority or (isinstance(snooze.target, list) and priority in snooze.target))
256 )
257 or snooze.target_type == GlobalTargetType.EVERYTHING
258 or (snooze.target_type == GlobalTargetType.NONCRITICAL and priority != PRIORITY_CRITICAL)
259 ):
260 recipients_to_remove = []
261 for recipient in recipients.person_ids:
262 if recipient == snooze.recipient:
263 recipients_to_remove.append(recipient)
264 _LOGGER.info("SUPERNOTIFY Snoozing %s", snooze.recipient)
266 recipients.remove(ATTR_PERSON_ID, recipients_to_remove)
268 if snooze.target_type == QualifiedTargetType.MOBILE:
269 to_remove: list[str] = []
270 for recipient in recipients.mobile_app_ids:
271 if recipient == snooze.target:
272 _LOGGER.debug("SUPERNOTIFY Snoozing %s for %s", snooze.recipient, snooze.target)
273 to_remove.append(recipient)
274 if to_remove:
275 recipients.remove(ATTR_MOBILE_APP_ID, to_remove)
276 return recipients