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