Coverage for custom_components/supernotify/notify.py: 89%
297 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-21 23:31 +0000
1"""Supernotify service, extending BaseNotificationService"""
3import copy
4import datetime as dt
5import json
6import logging
7from dataclasses import asdict
8from traceback import format_exception
9from typing import TYPE_CHECKING, Any
11from cachetools import TTLCache
12from homeassistant.components.notify import (
13 NotifyEntity,
14 NotifyEntityFeature,
15)
16from homeassistant.components.notify.legacy import BaseNotificationService
17from homeassistant.const import (
18 EVENT_HOMEASSISTANT_STOP,
19 STATE_OFF,
20 STATE_ON,
21 STATE_UNKNOWN,
22)
23from homeassistant.core import (
24 CALLBACK_TYPE,
25 Event,
26 EventStateChangedData,
27 HomeAssistant,
28 ServiceCall,
29 State,
30 SupportsResponse,
31 callback,
32)
33from homeassistant.helpers.event import async_track_state_change_event, async_track_time_change
34from homeassistant.helpers.json import ExtendedJSONEncoder
35from homeassistant.helpers.reload import async_setup_reload_service
36from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
38from custom_components.supernotify.archive import ARCHIVE_PURGE_MIN_INTERVAL
39from custom_components.supernotify.transport import Transport
41from . import (
42 ATTR_ACTION,
43 ATTR_DATA,
44 ATTR_DUPE_POLICY_MTSLP,
45 ATTR_DUPE_POLICY_NONE,
46 CONF_ACTION_GROUPS,
47 CONF_ACTIONS,
48 CONF_ARCHIVE,
49 CONF_CAMERAS,
50 CONF_DELIVERY,
51 CONF_DUPE_CHECK,
52 CONF_DUPE_POLICY,
53 CONF_HOUSEKEEPING,
54 CONF_HOUSEKEEPING_TIME,
55 CONF_LINKS,
56 CONF_MEDIA_PATH,
57 CONF_RECIPIENTS,
58 CONF_SCENARIOS,
59 CONF_SIZE,
60 CONF_TARGET,
61 CONF_TEMPLATE_PATH,
62 CONF_TRANSPORTS,
63 CONF_TTL,
64 DOMAIN,
65 PLATFORMS,
66 PRIORITY_MEDIUM,
67 PRIORITY_VALUES,
68)
69from . import SUPERNOTIFY_SCHEMA as PLATFORM_SCHEMA
70from .archive import NotificationArchive
71from .context import Context
72from .delivery import DeliveryRegistry
73from .hass_api import HomeAssistantAPI
74from .model import ConditionVariables, SuppressionReason
75from .notification import Notification
76from .people import PeopleRegistry
77from .scenario import ScenarioRegistry
78from .snoozer import Snoozer
79from .transports.alexa_devices import AlexaDevicesTransport
80from .transports.alexa_media_player import AlexaMediaPlayerTransport
81from .transports.chime import ChimeTransport
82from .transports.email import EmailTransport
83from .transports.generic import GenericTransport
84from .transports.media_player import MediaPlayerTransport
85from .transports.mobile_push import MobilePushTransport
86from .transports.mqtt import MQTTTransport
87from .transports.notify_entity import NotifyEntityTransport
88from .transports.persistent import PersistentTransport
89from .transports.sms import SMSTransport
91if TYPE_CHECKING:
92 from custom_components.supernotify.delivery import Delivery
94 from .scenario import Scenario
97_LOGGER = logging.getLogger(__name__)
99SNOOZE_TIME = 60 * 60 # TODO: move to configuration
101TRANSPORTS: list[type[Transport]] = [
102 EmailTransport,
103 SMSTransport,
104 MQTTTransport,
105 AlexaDevicesTransport,
106 AlexaMediaPlayerTransport,
107 MobilePushTransport,
108 MediaPlayerTransport,
109 ChimeTransport,
110 PersistentTransport,
111 GenericTransport,
112 NotifyEntityTransport,
113] # No auto-discovery of transport plugins so manual class registration required here
116async def async_get_service(
117 hass: HomeAssistant,
118 config: ConfigType,
119 discovery_info: DiscoveryInfoType | None = None,
120) -> "SupernotifyAction":
121 """Notify specific component setup - see async_setup_legacy in legacy BaseNotificationService"""
122 _ = PLATFORM_SCHEMA # schema must be imported even if not used for HA platform detection
123 _ = discovery_info
124 # for delivery in config.get(CONF_DELIVERY, {}).values():
125 # if delivery and CONF_CONDITION in delivery:
126 # try:
127 # await async_validate_condition_config(hass, delivery[CONF_CONDITION])
128 # except Exception as e:
129 # _LOGGER.error("SUPERNOTIFY delivery %s fails condition: %s", delivery[CONF_CONDITION], e)
130 # raise
132 hass.states.async_set(
133 f"{DOMAIN}.configured",
134 "True",
135 {
136 CONF_DELIVERY: config.get(CONF_DELIVERY, {}),
137 CONF_LINKS: config.get(CONF_LINKS, ()),
138 CONF_TEMPLATE_PATH: config.get(CONF_TEMPLATE_PATH, None),
139 CONF_MEDIA_PATH: config.get(CONF_MEDIA_PATH, None),
140 CONF_ARCHIVE: config.get(CONF_ARCHIVE, {}),
141 CONF_RECIPIENTS: config.get(CONF_RECIPIENTS, ()),
142 CONF_ACTIONS: config.get(CONF_ACTIONS, {}),
143 CONF_HOUSEKEEPING: config.get(CONF_HOUSEKEEPING, {}),
144 CONF_ACTION_GROUPS: config.get(CONF_ACTION_GROUPS, {}),
145 CONF_SCENARIOS: list(config.get(CONF_SCENARIOS, {}).keys()),
146 CONF_TRANSPORTS: config.get(CONF_TRANSPORTS, {}),
147 CONF_CAMERAS: config.get(CONF_CAMERAS, {}),
148 CONF_DUPE_CHECK: config.get(CONF_DUPE_CHECK, {}),
149 },
150 )
151 hass.states.async_set(f"{DOMAIN}.failures", "0")
152 hass.states.async_set(f"{DOMAIN}.sent", "0")
154 await async_setup_reload_service(hass, DOMAIN, PLATFORMS)
155 service = SupernotifyAction(
156 hass,
157 deliveries=config[CONF_DELIVERY],
158 template_path=config[CONF_TEMPLATE_PATH],
159 media_path=config[CONF_MEDIA_PATH],
160 archive=config[CONF_ARCHIVE],
161 housekeeping=config[CONF_HOUSEKEEPING],
162 recipients=config[CONF_RECIPIENTS],
163 mobile_actions=config[CONF_ACTION_GROUPS],
164 scenarios=config[CONF_SCENARIOS],
165 links=config[CONF_LINKS],
166 transport_configs=config[CONF_TRANSPORTS],
167 cameras=config[CONF_CAMERAS],
168 dupe_check=config[CONF_DUPE_CHECK],
169 )
170 await service.initialize()
172 def supplemental_action_refresh_entities(_call: ServiceCall) -> None:
173 return service.expose_entities()
175 def supplemental_action_enquire_implicit_deliveries(_call: ServiceCall) -> dict[str, Any]:
176 return service.enquire_implicit_deliveries()
178 def supplemental_action_enquire_deliveries_by_scenario(_call: ServiceCall) -> dict[str, Any]:
179 return service.enquire_deliveries_by_scenario()
181 def supplemental_action_enquire_last_notification(_call: ServiceCall) -> dict[str, Any]:
182 return service.last_notification.contents() if service.last_notification else {}
184 async def supplemental_action_enquire_active_scenarios(call: ServiceCall) -> dict[str, Any]:
185 trace = call.data.get("trace", False)
186 result: dict[str, Any] = {"scenarios": await service.enquire_active_scenarios()}
187 if trace:
188 result["trace"] = await service.trace_active_scenarios()
189 return result
191 def supplemental_action_enquire_scenarios(_call: ServiceCall) -> dict[str, Any]:
192 return {"scenarios": service.enquire_scenarios()}
194 async def supplemental_action_enquire_occupancy(_call: ServiceCall) -> dict[str, Any]:
195 return {"scenarios": await service.enquire_occupancy()}
197 def supplemental_action_enquire_snoozes(_call: ServiceCall) -> dict[str, Any]:
198 return {"snoozes": service.enquire_snoozes()}
200 def supplemental_action_clear_snoozes(_call: ServiceCall) -> dict[str, Any]:
201 return {"cleared": service.clear_snoozes()}
203 def supplemental_action_enquire_people(_call: ServiceCall) -> dict[str, Any]:
204 return {"people": service.enquire_people()}
206 async def supplemental_action_purge_archive(call: ServiceCall) -> dict[str, Any]:
207 days = call.data.get("days")
208 if not service.context.archive.enabled:
209 return {"error": "No archive configured"}
210 purged = await service.context.archive.cleanup(days=days, force=True)
211 arch_size = await service.context.archive.size()
212 return {
213 "purged": purged,
214 "remaining": arch_size,
215 "interval": ARCHIVE_PURGE_MIN_INTERVAL,
216 "days": service.context.archive.archive_days if days is None else days,
217 }
219 hass.services.async_register(
220 DOMAIN,
221 "enquire_implicit_deliveries",
222 supplemental_action_enquire_implicit_deliveries,
223 supports_response=SupportsResponse.ONLY,
224 )
225 hass.services.async_register(
226 DOMAIN,
227 "enquire_deliveries_by_scenario",
228 supplemental_action_enquire_deliveries_by_scenario,
229 supports_response=SupportsResponse.ONLY,
230 )
231 hass.services.async_register(
232 DOMAIN,
233 "enquire_last_notification",
234 supplemental_action_enquire_last_notification,
235 supports_response=SupportsResponse.ONLY,
236 )
237 hass.services.async_register(
238 DOMAIN,
239 "enquire_active_scenarios",
240 supplemental_action_enquire_active_scenarios,
241 supports_response=SupportsResponse.ONLY,
242 )
243 hass.services.async_register(
244 DOMAIN,
245 "enquire_scenarios",
246 supplemental_action_enquire_scenarios,
247 supports_response=SupportsResponse.ONLY,
248 )
249 hass.services.async_register(
250 DOMAIN,
251 "enquire_occupancy",
252 supplemental_action_enquire_occupancy,
253 supports_response=SupportsResponse.ONLY,
254 )
255 hass.services.async_register(
256 DOMAIN,
257 "enquire_people",
258 supplemental_action_enquire_people,
259 supports_response=SupportsResponse.ONLY,
260 )
261 hass.services.async_register(
262 DOMAIN,
263 "enquire_snoozes",
264 supplemental_action_enquire_snoozes,
265 supports_response=SupportsResponse.ONLY,
266 )
267 hass.services.async_register(
268 DOMAIN,
269 "clear_snoozes",
270 supplemental_action_clear_snoozes,
271 supports_response=SupportsResponse.ONLY,
272 )
273 hass.services.async_register(
274 DOMAIN,
275 "purge_archive",
276 supplemental_action_purge_archive,
277 supports_response=SupportsResponse.ONLY,
278 )
279 hass.services.async_register(
280 DOMAIN,
281 "refresh_entities",
282 supplemental_action_refresh_entities,
283 supports_response=SupportsResponse.NONE,
284 )
286 return service
289class SupernotifyEntity(NotifyEntity):
290 """Implement supernotify as a NotifyEntity platform."""
292 _attr_has_entity_name = True
293 _attr_name = "supernotify"
295 def __init__(
296 self,
297 unique_id: str,
298 platform: "SupernotifyAction",
299 ) -> None:
300 """Initialize the SuperNotify entity."""
301 self._attr_unique_id = unique_id
302 self._attr_supported_features = NotifyEntityFeature.TITLE
303 self._platform = platform
305 async def async_send_message(
306 self, message: str, title: str | None = None, target: str | list[str] | None = None, data: dict[str, Any] | None = None
307 ) -> None:
308 """Send a message to a user."""
309 await self._platform.async_send_message(message, title=title, target=target, data=data)
312class SupernotifyAction(BaseNotificationService):
313 """Implement SuperNotify Action"""
315 def __init__(
316 self,
317 hass: HomeAssistant,
318 deliveries: dict[str, dict[str, Any]] | None = None,
319 template_path: str | None = None,
320 media_path: str | None = None,
321 archive: dict[str, Any] | None = None,
322 housekeeping: dict[str, Any] | None = None,
323 recipients: list[dict[str, Any]] | None = None,
324 mobile_actions: dict[str, Any] | None = None,
325 scenarios: dict[str, dict[str, Any]] | None = None,
326 links: list[str] | None = None,
327 transport_configs: dict[str, Any] | None = None,
328 cameras: list[dict[str, Any]] | None = None,
329 dupe_check: dict[str, Any] | None = None,
330 ) -> None:
331 """Initialize the service."""
332 self.hass: HomeAssistant = hass
333 self.last_notification: Notification | None = None
334 self.failures: int = 0
335 self.housekeeping: dict[str, Any] = housekeeping or {}
336 self.sent: int = 0
337 hass_api = HomeAssistantAPI(hass)
338 self.context = Context(
339 hass_api,
340 PeopleRegistry(recipients or [], hass_api),
341 ScenarioRegistry(scenarios or {}),
342 DeliveryRegistry(deliveries or {}, transport_configs or {}, TRANSPORTS),
343 NotificationArchive(archive or {}, hass_api),
344 Snoozer(),
345 links or [],
346 recipients or [],
347 mobile_actions,
348 template_path,
349 media_path,
350 cameras=cameras,
351 )
353 self.unsubscribes: list[CALLBACK_TYPE] = []
354 self.exposed_entities: list[str] = []
355 self.dupe_check_config: dict[str, Any] = dupe_check or {}
356 # dupe check cache, key is (priority, message hash)
357 self.notification_cache: TTLCache[tuple[int, str], str] = TTLCache(
358 maxsize=self.dupe_check_config.get(CONF_SIZE, 100), ttl=self.dupe_check_config.get(CONF_TTL, 120)
359 )
361 async def initialize(self) -> None:
362 await self.context.initialize()
363 self.context.hass_api.initialize()
364 self.context.people_registry.initialize()
365 await self.context.delivery_registry.initialize(self.context)
366 await self.context.scenario_registry.initialize(
367 self.context.delivery_registry.deliveries,
368 self.context.delivery_registry.implicit_deliveries,
369 self.context.mobile_actions,
370 self.context.hass_api,
371 )
372 await self.context.archive.initialize()
374 self.expose_entities()
375 self.unsubscribes.append(self.hass.bus.async_listen("mobile_app_notification_action", self.on_mobile_action))
376 self.unsubscribes.append(
377 async_track_state_change_event(self.hass, self.exposed_entities, self._entity_state_change_listener)
378 )
380 housekeeping_schedule = self.housekeeping.get(CONF_HOUSEKEEPING_TIME)
381 if housekeeping_schedule:
382 _LOGGER.info("SUPERNOTIFY setting up housekeeping schedule at: %s", housekeeping_schedule)
383 self.unsubscribes.append(
384 async_track_time_change(
385 self.hass,
386 self.async_nightly_tasks,
387 hour=housekeeping_schedule.hour,
388 minute=housekeeping_schedule.minute,
389 second=housekeeping_schedule.second,
390 )
391 )
393 self.unsubscribes.append(self.hass.bus.async_listen(EVENT_HOMEASSISTANT_STOP, self.async_shutdown))
395 async def async_shutdown(self, event: Event) -> None:
396 _LOGGER.info("SUPERNOTIFY shutting down, %s", event)
397 self.shutdown()
399 async def async_unregister_services(self) -> None:
400 _LOGGER.info("SUPERNOTIFY unregistering")
401 self.shutdown()
402 return await super().async_unregister_services()
404 def shutdown(self) -> None:
405 for unsub in self.unsubscribes:
406 try:
407 _LOGGER.debug("SUPERNOTIFY unsubscribing: %s", unsub)
408 unsub()
409 except Exception as e:
410 _LOGGER.error("SUPERNOTIFY failed to unsubscribe: %s", e)
411 _LOGGER.info("SUPERNOTIFY shut down")
413 async def async_send_message(
414 self, message: str = "", title: str | None = None, target: list[str] | str | None = None, **kwargs: Any
415 ) -> None:
416 """Send a message via chosen transport."""
417 data = kwargs.get(ATTR_DATA, {})
418 notification = None
419 _LOGGER.debug("Message: %s, target: %s, data: %s", message, target, data)
421 try:
422 notification = Notification(self.context, message, title, target, data)
423 await notification.initialize()
424 if self.dupe_check(notification):
425 notification.suppress(SuppressionReason.DUPE)
426 else:
427 if await notification.deliver():
428 self.sent += 1
429 self.hass.states.async_set(f"{DOMAIN}.sent", str(self.sent))
430 elif notification.errored:
431 _LOGGER.error("SUPERNOTIFY Failed to deliver %s, error count %s", notification.id, notification.errored)
432 else:
433 _LOGGER.warning("SUPERNOTIFY No deliveries made for %s", notification.id)
435 except Exception as err:
436 # fault barrier of last resort, integration failures should be caught within envelope delivery
437 _LOGGER.exception("SUPERNOTIFY Failed to send message %s", message)
438 self.failures += 1
439 if notification is not None:
440 notification.delivery_error = format_exception(err)
441 self.hass.states.async_set(f"{DOMAIN}.failures", str(self.failures))
443 if notification is not None:
444 self.last_notification = notification
445 await self.context.archive.archive(notification)
446 _LOGGER.debug(
447 "SUPERNOTIFY %s deliveries, %s errors, %s skipped",
448 notification.delivered,
449 notification.errored,
450 notification.skipped,
451 )
453 async def _entity_state_change_listener(self, event: Event[EventStateChangedData]) -> None:
454 changes = 0
455 if event is not None:
456 _LOGGER.info(f"SUPERNOTIFY {event.event_type} event for entity: {event.data}")
457 new_state: State | None = event.data["new_state"]
458 if new_state and event.data["entity_id"].startswith(f"{DOMAIN}.scenario_"):
459 scenario: Scenario | None = self.context.scenario_registry.scenarios.get(
460 event.data["entity_id"].replace(f"{DOMAIN}.scenario_", "")
461 )
462 if scenario is None:
463 _LOGGER.warning(f"SUPERNOTIFY Event for unknown scenario {event.data['entity_id']}")
464 else:
465 if new_state.state == "off" and scenario.enabled:
466 scenario.enabled = False
467 _LOGGER.info(f"SUPERNOTIFY Disabling scenario {scenario.name}")
468 changes += 1
469 elif new_state.state == "on" and not scenario.enabled:
470 scenario.enabled = True
471 _LOGGER.info(f"SUPERNOTIFY Enabling scenario {scenario.name}")
472 changes += 1
473 else:
474 _LOGGER.info(f"SUPERNOTIFY No change to scenario {scenario.name}, already {new_state}")
475 elif new_state and event.data["entity_id"].startswith(f"{DOMAIN}.delivery_"):
476 delivery_config: Delivery | None = self.context.delivery_registry.deliveries.get(
477 event.data["entity_id"].replace(f"{DOMAIN}.delivery_", "")
478 )
479 if delivery_config is None:
480 _LOGGER.warning(f"SUPERNOTIFY Event for unknown delivery {event.data['entity_id']}")
481 else:
482 if new_state.state == "off" and delivery_config.enabled:
483 delivery_config.enabled = False
484 _LOGGER.info(f"SUPERNOTIFY Disabling delivery {delivery_config.name}")
485 changes += 1
486 elif new_state.state == "on" and not delivery_config.enabled:
487 delivery_config.enabled = True
488 _LOGGER.info(f"SUPERNOTIFY Enabling delivery {delivery_config.name}")
489 changes += 1
490 else:
491 _LOGGER.info(f"SUPERNOTIFY No change to delivery {delivery_config.name}, already {new_state}")
492 elif new_state and event.data["entity_id"].startswith(f"{DOMAIN}.transport_"):
493 transport: Transport | None = self.context.delivery_registry.transports.get(
494 event.data["entity_id"].replace(f"{DOMAIN}.transport_", "")
495 )
496 if transport is None:
497 _LOGGER.warning(f"SUPERNOTIFY Event for unknown methtransportod {event.data['entity_id']}")
498 else:
499 if new_state.state == "off" and transport.override_enabled:
500 transport.override_enabled = False
501 _LOGGER.info(f"SUPERNOTIFY Disabling delivery {transport.name}")
502 changes += 1
503 elif new_state.state == "on" and not transport.override_enabled:
504 transport.override_enabled = True
505 _LOGGER.info(f"SUPERNOTIFY Enabling delivery {transport.name}")
506 changes += 1
507 else:
508 _LOGGER.info(f"SUPERNOTIFY No change to transport {transport.name}, already {new_state}")
510 else:
511 _LOGGER.warning("SUPERNOTIFY entity event with nothing to do:%s", event)
512 if changes:
513 self.context.scenario_registry.refresh(
514 self.context.delivery_registry.deliveries, self.context.delivery_registry.implicit_deliveries
515 )
516 _LOGGER.debug(f"SUPERNOTIFY event had {changes} changes triggering updates to states")
518 def expose_entities(self) -> None:
519 # Create on the fly entities for key internal config and state
521 for scenario in self.context.scenario_registry.scenarios.values():
522 self.hass.states.async_set(
523 f"{DOMAIN}.scenario_{scenario.name}", STATE_UNKNOWN, scenario.attributes(include_condition=False)
524 )
525 self.exposed_entities.append(f"{DOMAIN}.scenario_{scenario.name}")
526 for transport in self.context.delivery_registry.transports.values():
527 self.hass.states.async_set(
528 f"{DOMAIN}.transport_{transport.name}",
529 STATE_ON if transport.override_enabled else STATE_OFF,
530 transport.attributes(),
531 )
532 self.exposed_entities.append(f"{DOMAIN}.transport_{transport.name}")
533 for delivery_name, delivery in self.context.delivery_registry.deliveries.items():
534 self.hass.states.async_set(
535 f"{DOMAIN}.delivery_{delivery.name}", STATE_ON if delivery.enabled else STATE_OFF, delivery.attributes()
536 )
537 self.exposed_entities.append(f"{DOMAIN}.delivery_{delivery_name}")
539 def dupe_check(self, notification: Notification) -> bool:
540 policy = self.dupe_check_config.get(CONF_DUPE_POLICY, ATTR_DUPE_POLICY_MTSLP)
541 if policy == ATTR_DUPE_POLICY_NONE:
542 return False
543 notification_hash = notification.hash()
544 if notification.priority in PRIORITY_VALUES:
545 same_or_higher_priority = PRIORITY_VALUES[PRIORITY_VALUES.index(notification.priority) :]
546 else:
547 same_or_higher_priority = [notification.priority]
548 dupe = False
549 if any((notification_hash, p) in self.notification_cache for p in same_or_higher_priority):
550 _LOGGER.debug("SUPERNOTIFY Detected dupe notification")
551 dupe = True
552 self.notification_cache[notification_hash, notification.priority] = notification.id
553 return dupe
555 def enquire_implicit_deliveries(self) -> dict[str, Any]:
556 v: dict[str, list[str]] = {}
557 for t in self.context.delivery_registry.transports:
558 for d in self.context.delivery_registry.implicit_deliveries:
559 if d.transport.name == t:
560 v.setdefault(t, [])
561 v[t].append(d.name)
562 return v
564 def enquire_deliveries_by_scenario(self) -> dict[str, list[str]]:
565 return self.context.scenario_registry.delivery_by_scenario
567 async def enquire_occupancy(self) -> dict[str, list[dict[str, Any]]]:
568 return self.context.people_registry.determine_occupancy()
570 async def enquire_active_scenarios(self) -> list[str]:
571 occupiers: dict[str, list[dict[str, Any]]] = self.context.people_registry.determine_occupancy()
572 cvars = ConditionVariables([], [], [], PRIORITY_MEDIUM, occupiers, None, None)
573 return [s.name for s in self.context.scenario_registry.scenarios.values() if await s.evaluate(cvars)]
575 async def trace_active_scenarios(self) -> tuple[list[dict[str, Any]], list[dict[str, Any]], dict[str, Any]]:
576 occupiers: dict[str, list[dict[str, Any]]] = self.context.people_registry.determine_occupancy()
577 cvars = ConditionVariables([], [], [], PRIORITY_MEDIUM, occupiers, None, None)
579 def safe_json(v: Any) -> Any:
580 return json.loads(json.dumps(v, cls=ExtendedJSONEncoder))
582 enabled = []
583 disabled = []
584 dcvars = asdict(cvars)
585 for s in self.context.scenario_registry.scenarios.values():
586 if await s.trace(cvars):
587 enabled.append(safe_json(s.attributes(include_trace=True)))
588 else:
589 disabled.append(safe_json(s.attributes(include_trace=True)))
590 return enabled, disabled, dcvars
592 def enquire_scenarios(self) -> dict[str, dict[str, Any]]:
593 return {s.name: s.attributes(include_condition=False) for s in self.context.scenario_registry.scenarios.values()}
595 def enquire_snoozes(self) -> list[dict[str, Any]]:
596 return self.context.snoozer.export()
598 def clear_snoozes(self) -> int:
599 return self.context.snoozer.clear()
601 def enquire_people(self) -> list[dict[str, Any]]:
602 response = copy.deepcopy(self.context.people_registry.people)
603 for p in response.values():
604 if CONF_TARGET in p:
605 p[CONF_TARGET] = p[CONF_TARGET].as_dict()
606 return list(response.values())
608 @callback
609 def on_mobile_action(self, event: Event) -> None:
610 """Listen for mobile actions relevant to snooze and silence notifications
612 Example Action:
613 event_type: mobile_app_notification_action
614 data:
615 foo: a
616 origin: REMOTE
617 time_fired: "2024-04-20T13:14:09.360708+00:00"
618 context:
619 id: 01HVXT93JGWEDW0KE57Z0X6Z1K
620 parent_id: null
621 user_id: e9dbae1a5abf44dbbad52ff85501bb17
622 """
623 event_name = event.data.get(ATTR_ACTION)
624 if event_name is None or not event_name.startswith("SUPERNOTIFY_"):
625 return # event not intended for here
626 self.context.snoozer.handle_command_event(event, self.context.people_registry.people)
628 @callback
629 async def async_nightly_tasks(self, now: dt.datetime) -> None:
630 _LOGGER.info("SUPERNOTIFY Housekeeping starting as scheduled at %s", now)
631 await self.context.archive.cleanup()
632 self.context.snoozer.purge_snoozes()
633 _LOGGER.info("SUPERNOTIFY Housekeeping completed")