Coverage for custom_components/supernotify/engine.py: 98%
197 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 21:14 +0000
1"""Supernotify service, extending BaseNotificationService"""
3from __future__ import annotations
5import json
6import logging
7from dataclasses import asdict
8from traceback import format_exception
9from typing import TYPE_CHECKING, Any
11from homeassistant.const import (
12 EVENT_HOMEASSISTANT_STOP,
13)
14from homeassistant.core import (
15 Context as HAContext,
16)
17from homeassistant.core import (
18 Event,
19 HomeAssistant,
20 callback,
21)
22from homeassistant.helpers.json import ExtendedJSONEncoder
23from homeassistant.helpers.typing import ConfigType
25from .archive import NotificationArchive
26from .common import DupeChecker
27from .const import (
28 ATTR_ACTION,
29 ATTR_DATA,
30 CONF_ACTION_GROUPS,
31 CONF_ARCHIVE,
32 CONF_CAMERAS,
33 CONF_DELIVERY,
34 CONF_DELIVERY_CONTROL,
35 CONF_DUPE_CHECK,
36 CONF_HOUSEKEEPING,
37 CONF_HOUSEKEEPING_TIME,
38 CONF_LINKS,
39 CONF_MEDIA_PATH,
40 CONF_MEDIA_STORAGE_DAYS,
41 CONF_MEDIA_URL_PREFIX,
42 CONF_MOBILE_DISCOVERY,
43 CONF_RECIPIENTS,
44 CONF_RECIPIENTS_DISCOVERY,
45 CONF_SCENARIO_CONTROL,
46 CONF_SCENARIOS,
47 CONF_SNOOZE,
48 CONF_TEMPLATE_PATH,
49 CONF_TRANSPORTS,
50 OVERRIDE_KIND_DELIVERY,
51 OVERRIDE_KIND_RECIPIENT,
52 OVERRIDE_KIND_SCENARIO,
53 OVERRIDE_KIND_TRANSPORT,
54 OVERRIDE_KINDS,
55 PRIORITY_MEDIUM,
56)
57from .context import Context
58from .delivery import DeliveryRegistry
59from .exceptions import UncategorizedTargetError
60from .hass_api import HomeAssistantAPI
61from .media_grab import MediaStorage
62from .model import ConditionVariables, SuppressionReason
63from .notification import Notification
64from .people import PeopleRegistry, Recipient
65from .scenario import ScenarioRegistry
66from .sensor import SupernotifyCounterSensor
67from .snoozer import Snoozer
68from .static_config import TRANSPORTS
70if TYPE_CHECKING:
71 import datetime as dt
72 from collections.abc import Iterable
74 from .switch import Overridable, SupernotifyOverridableSwitch
76_LOGGER = logging.getLogger(__name__)
79class SupernotifyEngine:
80 """Owns the Context/registries/transports and actually delivers notifications.
82 This is the shared engine behind every entrypoint - notify.supernotify (via the
83 SuperNotificationService legacy shim), supernotify.notify, and the NotifyEntity platform
84 (RecipientNotifyEntity) - so it deliberately has no dependency on
85 BaseNotificationService or anything else specific to the legacy notify platform. If/when
86 HA core drops BaseNotificationService, only SuperNotificationService and its wiring in
87 __init__.py need to go; this class and everything else built on it are unaffected.
88 """
90 def __init__(
91 self,
92 hass: HomeAssistant,
93 deliveries: dict[str, dict[str, Any]] | None = None,
94 template_path: str | None = None,
95 media_path: str | None = None,
96 media_url_prefix: str | None = None,
97 archive: dict[str, Any] | None = None,
98 housekeeping: dict[str, Any] | None = None,
99 recipients_discovery: bool = True,
100 mobile_discovery: bool = True,
101 recipients: list[dict[str, Any]] | None = None,
102 mobile_actions: dict[str, Any] | None = None,
103 scenarios: dict[str, dict[str, Any]] | None = None,
104 links: list[str] | None = None,
105 transport_configs: dict[str, Any] | None = None,
106 cameras: list[dict[str, Any]] | None = None,
107 dupe_check: dict[str, Any] | None = None,
108 snooze: dict[str, Any] | None = None,
109 scenario_control: dict[str, Any] | None = None,
110 delivery_control: dict[str, Any] | None = None,
111 ) -> None:
112 """Initialize the service."""
113 self.last_notification: Notification | None = None
114 self.housekeeping: dict[str, Any] = housekeeping or {}
115 # The counts live only in these entities, which restore their own value across restarts;
116 # sensor.py hands them to Home Assistant once its platform loads
117 self.notifications_sensor = SupernotifyCounterSensor("notifications", "notifications")
118 self.failures_sensor = SupernotifyCounterSensor("failures", "failures")
119 # Every switch overriding a configured enabled flag, by unique_id - populated by switch.py
120 # as each is added to Home Assistant
121 self.override_switches: dict[str, SupernotifyOverridableSwitch] = {}
122 hass_api = HomeAssistantAPI(hass)
124 people_registry = PeopleRegistry(
125 recipients or [], hass_api, discover=recipients_discovery, mobile_discovery=mobile_discovery
126 )
127 self.context = Context(
128 hass_api,
129 people_registry,
130 ScenarioRegistry(scenarios or {}, scenario_control, people_registry),
131 DeliveryRegistry(deliveries or {}, transport_configs or {}, TRANSPORTS, delivery_control=delivery_control),
132 DupeChecker(dupe_check or {}),
133 NotificationArchive(archive or {}, hass_api),
134 MediaStorage(
135 media_path,
136 media_url_prefix=media_url_prefix,
137 days=self.housekeeping.get(CONF_MEDIA_STORAGE_DAYS, 7),
138 ),
139 Snoozer(snooze),
140 links or [],
141 recipients or [],
142 mobile_actions,
143 template_path,
144 cameras=cameras,
145 )
147 async def initialize(self) -> None:
148 await self.context.initialize()
149 self.context.hass_api.initialize()
150 await self.context.people_registry.initialize()
151 await self.context.delivery_registry.initialize(self.context)
152 await self.context.scenario_registry.initialize(
153 self.context.delivery_registry,
154 self.context.mobile_actions,
155 self.context.hass_api,
156 )
157 await self.context.archive.initialize()
158 await self.context.media_storage.initialize(self.context.hass_api)
159 await self.context.snoozer.initialize(self.context.hass_api)
161 # Every entity - switches, binary_sensors and counters - is a real platform entity, added
162 # once this method returns (see __init__.py), and keeps its own state current
163 self.context.hass_api.subscribe_event("mobile_app_notification_action", self.on_mobile_action)
165 housekeeping_schedule = self.housekeeping.get(CONF_HOUSEKEEPING_TIME)
166 if housekeeping_schedule:
167 _LOGGER.info("SUPERNOTIFY Setting up housekeeping schedule at: %s", housekeeping_schedule)
168 self.context.hass_api.subscribe_time(
169 housekeeping_schedule.hour, housekeeping_schedule.minute, housekeeping_schedule.second, self.async_nightly_tasks
170 )
171 else:
172 _LOGGER.info(
173 "SUPERNOTIFY Housekeeping disabled. Storage must be manually managed if using attachments or image snapshots"
174 )
176 self.context.hass_api.subscribe_event(EVENT_HOMEASSISTANT_STOP, self.async_shutdown)
178 async def async_shutdown(self, event: Event) -> None:
179 _LOGGER.info("SUPERNOTIFY Shutting down, %s (%s)", event.event_type, event.time_fired)
180 self.shutdown()
182 def shutdown(self) -> None:
183 self.context.hass_api.disconnect()
184 _LOGGER.info("SUPERNOTIFY Shut down")
186 @property
187 def counter_sensors(self) -> list[SupernotifyCounterSensor]:
188 return [self.notifications_sensor, self.failures_sensor]
190 @property
191 def sent(self) -> int:
192 return self.notifications_sensor.count
194 @property
195 def failures(self) -> int:
196 return self.failures_sensor.count
198 async def async_send_message(
199 self,
200 message: str = "",
201 title: str | None = None,
202 target: list[str] | str | dict[str, Any] | None = None,
203 context: HAContext | None = None,
204 **kwargs: Any,
205 ) -> Notification | None:
206 """Send a message via chosen transport, returning the notification, if one could be made"""
207 data = kwargs.get(ATTR_DATA, {})
208 notification = None
209 _LOGGER.debug("SUPERNOTIFY Message: %s, target: %s, data: %s", message, target, data)
211 if context is None:
212 # only reachable when async_send_message is invoked directly rather than via a
213 # registered action (e.g. SuperNotificationService._async_notify_message_service, or
214 # supernotify.notify in async_setup_supplemental_actions) - without this fallback,
215 # downstream service calls for this notification would each get their own
216 # unrelated Context, leaving them unlinked in the logbook/recorder
217 _LOGGER.debug("SUPERNOTIFY No context supplied, generating new one")
218 context = HAContext()
220 try:
221 notification = Notification(self.context, message, title, target, action_data=data, ha_context=context)
222 await notification.initialize()
223 if await notification.deliver():
224 self.notifications_sensor.increment()
225 elif notification.failed:
226 _LOGGER.error("SUPERNOTIFY Failed to deliver %s, error count %s", notification.id, notification.error_count)
227 else:
228 if notification.delivered == 0:
229 codes: list[SuppressionReason] = notification._skip_reasons
230 reason: str = ",".join(str(code) for code in codes)
231 problem: bool = codes != [SuppressionReason.DUPE]
232 else:
233 problem = True
234 reason = "No delivery envelopes generated"
235 if problem:
236 _LOGGER.warning("SUPERNOTIFY No deliveries made for %s: %s", notification.id, reason)
237 else:
238 _LOGGER.debug("SUPERNOTIFY Deliveries suppressed for %s: %s", notification.id, reason)
240 except Exception as err:
241 # fault barrier of last resort, integration failures should be caught within envelope delivery
242 _LOGGER.exception("SUPERNOTIFY Failed to send message %s", message)
243 self.failures_sensor.increment()
244 if notification is not None:
245 notification._delivery_error = format_exception(err)
247 if notification is None:
248 _LOGGER.warning("SUPERNOTIFY NULL Notification, %s", message)
249 else:
250 self.last_notification = notification
251 await self.context.archive.archive(notification)
252 _LOGGER.debug(
253 "SUPERNOTIFY %s deliveries, %s failed, %s skipped, %s suppressed",
254 notification.delivered,
255 notification.failed,
256 notification.skipped,
257 notification.suppressed,
258 )
259 if notification.uncategorized_targets:
260 # raised only now, at the very end - every target that could be delivered
261 # already has been, so one uncategorized target must never get in the way
262 # of the rest of the notification going out
263 raise UncategorizedTargetError(notification.delivered, notification.uncategorized_targets)
264 return notification
266 async def async_dry_run(
267 self,
268 message: str = "",
269 title: str | None = None,
270 target: list[str] | str | dict[str, Any] | None = None,
271 data: dict[str, Any] | None = None,
272 ) -> dict[str, Any]:
273 """Which deliveries a notification would use right now, and who it would reach, without sending it"""
274 notification = Notification(self.context, message, title, target, action_data=data)
275 await notification.initialize()
276 return notification.plan()
278 @callback
279 def refresh_entities(self) -> None:
280 """Re-publish the current state of every entity SuperNotify provides.
282 Entities are only refreshed once added to Home Assistant, so this is a safe no-op for any
283 whose platform hasn't loaded (e.g. in tests that build SupernotifyEngine directly without
284 a config entry). Must run in the event loop, as it writes entity state.
285 """
286 self.notifications_sensor.refresh()
287 self.failures_sensor.refresh()
288 self.context.scenario_registry.async_refresh_scenario_states()
289 for entity in self.context.people_registry.recipient_entities():
290 entity.async_write_ha_state()
291 for legacy_entity in self.context.delivery_registry.legacy_entities():
292 legacy_entity.async_write_ha_state()
293 for switch in self.override_switches.values():
294 switch.async_write_ha_state()
296 def _overridables(self, kind: str) -> Iterable[Overridable]:
297 overridables: dict[str, Iterable[Overridable]] = {
298 OVERRIDE_KIND_SCENARIO: self.context.scenario_registry.scenarios.values(),
299 OVERRIDE_KIND_RECIPIENT: self.context.people_registry.people.values(),
300 OVERRIDE_KIND_DELIVERY: self.context.delivery_registry.deliveries.values(),
301 OVERRIDE_KIND_TRANSPORT: self.context.delivery_registry.transports.values(),
302 }
303 return overridables[kind]
305 @callback
306 def reset_overrides(self, kinds: Iterable[str] = OVERRIDE_KINDS) -> dict[str, list[str]]:
307 """Put everything switched on or off at runtime back to its configured enabled state,
308 returning the names reset for each kind.
310 Walks the scenarios, recipients, deliveries and transports themselves rather than their
311 switches, so one whose switch is disabled in the entity registry is reset too.
312 """
313 reset: dict[str, list[str]] = {}
314 for kind in kinds:
315 names = reset[kind] = []
316 for item in self._overridables(kind):
317 if item.enabled == item.config_enabled:
318 continue
319 switch = self.override_switches.get(f"{kind}_{item.name}")
320 if switch is not None:
321 switch.async_set_enabled(item.config_enabled)
322 else:
323 item.enabled = item.config_enabled
324 self._async_refresh_related(kind, item.name)
325 names.append(item.name)
326 return reset
328 @callback
329 def _async_refresh_related(self, kind: str, name: str) -> None:
330 """Re-publish the binary_sensor following the enabled flag of something with no switch
331 to do it - the same as that switch's own _refresh_related()."""
332 if kind == OVERRIDE_KIND_SCENARIO:
333 self.context.scenario_registry.async_refresh_entity(name)
334 elif kind == OVERRIDE_KIND_RECIPIENT:
335 self.context.people_registry.async_refresh_entity(name)
336 else:
337 self.context.delivery_registry.async_refresh_entity(f"{kind}_{name}")
339 def enquire_implicit_deliveries(self) -> dict[str, Any]:
340 v: dict[str, list[str]] = {}
341 for t in self.context.delivery_registry.transports:
342 for d in self.context.delivery_registry.implicit_deliveries:
343 if d.transport.name == t:
344 v.setdefault(t, [])
345 v[t].append(d.name)
346 return v
348 def enquire_deliveries_by_scenario(self) -> dict[str, dict[str, list[str]]]:
349 return {
350 name: {
351 "enabled": scenario.enabling_deliveries(),
352 "disabled": scenario.disabling_deliveries(),
353 "applies": scenario.relevant_deliveries(),
354 }
355 for name, scenario in self.context.scenario_registry.scenarios.items()
356 if scenario.enabled
357 }
359 async def enquire_occupancy(self) -> dict[str, list[dict[str, Any]]]:
360 occupancy = self.context.people_registry.determine_occupancy()
361 return {k: [v.as_dict() for v in vs] for k, vs in occupancy.items()}
363 async def enquire_active_scenarios(self) -> list[str]:
364 occupiers: dict[str, list[Recipient]] = self.context.people_registry.determine_occupancy()
365 cvars = ConditionVariables([], [], [], PRIORITY_MEDIUM, occupiers, None, None)
366 return [s.name for s in self.context.scenario_registry.scenarios.values() if s.evaluate(cvars)]
368 async def trace_active_scenarios(self) -> tuple[list[dict[str, Any]], list[dict[str, Any]], dict[str, Any]]:
369 occupiers: dict[str, list[Recipient]] = self.context.people_registry.determine_occupancy()
370 cvars = ConditionVariables([], [], [], PRIORITY_MEDIUM, occupiers, None, None)
372 def safe_json(v: Any) -> Any: # ruff: ignore[any-type]
373 return json.loads(json.dumps(v, cls=ExtendedJSONEncoder))
375 enabled = []
376 disabled = []
377 dcvars = asdict(cvars)
378 for s in self.context.scenario_registry.scenarios.values():
379 if await s.trace(cvars):
380 enabled.append(safe_json(s.attributes(include_trace=True)))
381 else:
382 disabled.append(safe_json(s.attributes(include_trace=True)))
383 return enabled, disabled, dcvars
385 def enquire_scenarios(self) -> dict[str, dict[str, Any]]:
386 return {s.name: s.attributes(include_condition=False) for s in self.context.scenario_registry.scenarios.values()}
388 def enquire_snoozes(self) -> list[dict[str, Any]]:
389 return self.context.snoozer.export()
391 def clear_snoozes(self) -> int:
392 return self.context.snoozer.clear()
394 def enquire_recipients(self) -> list[dict[str, Any]]:
395 return [p.as_dict() for p in self.context.people_registry.people.values()]
397 @callback
398 def on_mobile_action(self, event: Event) -> None:
399 """Listen for mobile actions relevant to snooze and silence notifications
401 Example Action:
402 event_type: mobile_app_notification_action
403 data:
404 foo: a
405 origin: REMOTE
406 time_fired: "2024-04-20T13:14:09.360708+00:00"
407 context:
408 id: 01HVXT93JGWEDW0KE57Z0X6Z1K
409 parent_id: null
410 user_id: a9dbae1a5abf33dbbad52ff82201bb17
411 """
412 event_name = event.data.get(ATTR_ACTION)
413 if event_name is None or not event_name.startswith("SUPERNOTIFY_"):
414 return # event not intended for here
415 self.context.snoozer.handle_command_event(event, self.context.people_registry.enabled_recipients())
417 @callback
418 async def async_nightly_tasks(self, now: dt.datetime) -> None:
419 _LOGGER.info("SUPERNOTIFY Housekeeping starting as scheduled at %s", now)
420 await self.context.archive.cleanup()
421 self.context.snoozer.purge_snoozes()
422 await self.context.media_storage.cleanup()
423 _LOGGER.info("SUPERNOTIFY Housekeeping completed")
426def build_supernotify_engine(hass: HomeAssistant, config: ConfigType) -> SupernotifyEngine:
427 """Construct a SupernotifyEngine from a fully validated FULL_CONFIG_SCHEMA config dict.
429 Used by the config-entry setup (async_setup_entry in __init__.py), the sole owner of
430 registering notify.supernotify.
431 """
432 return SupernotifyEngine(
433 hass,
434 deliveries=config[CONF_DELIVERY],
435 template_path=config[CONF_TEMPLATE_PATH],
436 media_path=config[CONF_MEDIA_PATH],
437 media_url_prefix=config.get(CONF_MEDIA_URL_PREFIX),
438 archive=config[CONF_ARCHIVE],
439 housekeeping=config[CONF_HOUSEKEEPING],
440 mobile_discovery=config[CONF_MOBILE_DISCOVERY],
441 recipients_discovery=config[CONF_RECIPIENTS_DISCOVERY],
442 recipients=config[CONF_RECIPIENTS],
443 mobile_actions=config[CONF_ACTION_GROUPS],
444 scenarios=config[CONF_SCENARIOS],
445 links=config[CONF_LINKS],
446 transport_configs=config[CONF_TRANSPORTS],
447 cameras=config[CONF_CAMERAS],
448 dupe_check=config[CONF_DUPE_CHECK],
449 snooze=config[CONF_SNOOZE],
450 scenario_control=config.get(CONF_SCENARIO_CONTROL),
451 delivery_control=config.get(CONF_DELIVERY_CONTROL),
452 )