Coverage for custom_components / supernotify / notify.py: 89%
316 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-11 22:18 +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.components.notify import (
12 NotifyEntity,
13 NotifyEntityFeature,
14)
15from homeassistant.components.notify.legacy import BaseNotificationService
16from homeassistant.const import (
17 EVENT_HOMEASSISTANT_STOP,
18 STATE_OFF,
19 STATE_ON,
20 STATE_UNKNOWN,
21 EntityCategory,
22 Platform,
23)
24from homeassistant.core import (
25 Event,
26 EventStateChangedData,
27 HomeAssistant,
28 ServiceCall,
29 State,
30 SupportsResponse,
31 callback,
32)
33from homeassistant.helpers.json import ExtendedJSONEncoder
34from homeassistant.helpers.reload import async_setup_reload_service
36from . import DOMAIN, PLATFORMS
37from .archive import ARCHIVE_PURGE_MIN_INTERVAL, NotificationArchive
38from .common import DupeChecker, sanitize
39from .const import (
40 ATTR_ACTION,
41 ATTR_DATA,
42 CONF_ACTION_GROUPS,
43 CONF_ACTIONS,
44 CONF_ARCHIVE,
45 CONF_CAMERAS,
46 CONF_DELIVERY,
47 CONF_DUPE_CHECK,
48 CONF_HOUSEKEEPING,
49 CONF_HOUSEKEEPING_TIME,
50 CONF_LINKS,
51 CONF_MEDIA_PATH,
52 CONF_MEDIA_STORAGE_DAYS,
53 CONF_MEDIA_URL_PREFIX,
54 CONF_MOBILE_DISCOVERY,
55 CONF_RECIPIENTS,
56 CONF_RECIPIENTS_DISCOVERY,
57 CONF_SCENARIOS,
58 CONF_SNOOZE,
59 CONF_TEMPLATE_PATH,
60 CONF_TRANSPORTS,
61 PRIORITY_MEDIUM,
62)
63from .context import Context
64from .delivery import DeliveryRegistry
65from .hass_api import HomeAssistantAPI
66from .media_grab import MediaStorage
67from .model import ConditionVariables, SuppressionReason
68from .notification import Notification
69from .people import PeopleRegistry, Recipient
70from .scenario import ScenarioRegistry
71from .schema import SUPERNOTIFY_SCHEMA as PLATFORM_SCHEMA
72from .snoozer import Snoozer
73from .transports.alexa_devices import AlexaDevicesTransport
74from .transports.alexa_media_player import AlexaMediaPlayerTransport
75from .transports.chime import ChimeTransport
76from .transports.email import EmailTransport
77from .transports.generic import GenericTransport
78from .transports.gotify import GotifyTransport
79from .transports.lametric import LaMetricTransport
80from .transports.media_player import MediaPlayerTransport
81from .transports.mobile_push import MobilePushTransport
82from .transports.mqtt import MQTTTransport
83from .transports.notify_entity import NotifyEntityTransport
84from .transports.ntfy import NtfyTransport
85from .transports.persistent import PersistentTransport
86from .transports.pushover import PushoverTransport
87from .transports.sms import SMSTransport
88from .transports.telegram import TelegramTransport
89from .transports.tts import TTSTransport
91if TYPE_CHECKING:
92 import datetime as dt
94 from homeassistant.helpers import entity_registry as er
95 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
97 from .scenario import Scenario
98 from .transport import Transport
100PARALLEL_UPDATES = 0
102_LOGGER = logging.getLogger(__name__)
104TRANSPORTS: list[type[Transport]] = [
105 EmailTransport,
106 SMSTransport,
107 MQTTTransport,
108 AlexaDevicesTransport,
109 AlexaMediaPlayerTransport,
110 MobilePushTransport,
111 MediaPlayerTransport,
112 ChimeTransport,
113 PersistentTransport,
114 GenericTransport,
115 TTSTransport,
116 NotifyEntityTransport,
117 NtfyTransport,
118 GotifyTransport,
119 TelegramTransport,
120 LaMetricTransport,
121 PushoverTransport,
122] # No auto-discovery of transport plugins so manual class registration required here
125async def async_get_service(
126 hass: HomeAssistant,
127 config: ConfigType,
128 discovery_info: DiscoveryInfoType | None = None,
129) -> SupernotifyAction:
130 """Notify specific component setup - see async_setup_legacy in legacy BaseNotificationService"""
131 _ = PLATFORM_SCHEMA # schema must be imported even if not used for HA platform detection
132 _ = discovery_info
134 await async_setup_reload_service(hass, DOMAIN, PLATFORMS)
136 service = SupernotifyAction(
137 hass,
138 deliveries=config[CONF_DELIVERY],
139 template_path=config[CONF_TEMPLATE_PATH],
140 media_path=config[CONF_MEDIA_PATH],
141 media_url_prefix=config.get(CONF_MEDIA_URL_PREFIX),
142 archive=config[CONF_ARCHIVE],
143 housekeeping=config[CONF_HOUSEKEEPING],
144 mobile_discovery=config[CONF_MOBILE_DISCOVERY],
145 recipients_discovery=config[CONF_RECIPIENTS_DISCOVERY],
146 recipients=config[CONF_RECIPIENTS],
147 mobile_actions=config[CONF_ACTION_GROUPS],
148 scenarios=config[CONF_SCENARIOS],
149 links=config[CONF_LINKS],
150 transport_configs=config[CONF_TRANSPORTS],
151 cameras=config[CONF_CAMERAS],
152 dupe_check=config[CONF_DUPE_CHECK],
153 snooze=config[CONF_SNOOZE],
154 )
155 await service.initialize()
157 def supplemental_action_enquire_configuration(_call: ServiceCall) -> dict[str, Any]:
158 return {
159 CONF_DELIVERY: config.get(CONF_DELIVERY, {}),
160 CONF_LINKS: config.get(CONF_LINKS, ()),
161 CONF_TEMPLATE_PATH: config.get(CONF_TEMPLATE_PATH, None),
162 CONF_MEDIA_PATH: config.get(CONF_MEDIA_PATH, None),
163 CONF_ARCHIVE: config.get(CONF_ARCHIVE, {}),
164 CONF_MOBILE_DISCOVERY: config.get(CONF_MOBILE_DISCOVERY, ()),
165 CONF_RECIPIENTS_DISCOVERY: config.get(CONF_RECIPIENTS_DISCOVERY, ()),
166 CONF_RECIPIENTS: config.get(CONF_RECIPIENTS, ()),
167 CONF_ACTIONS: config.get(CONF_ACTIONS, {}),
168 CONF_HOUSEKEEPING: config.get(CONF_HOUSEKEEPING, {}),
169 CONF_ACTION_GROUPS: config.get(CONF_ACTION_GROUPS, {}),
170 CONF_SCENARIOS: list(config.get(CONF_SCENARIOS, {}).keys()),
171 CONF_TRANSPORTS: config.get(CONF_TRANSPORTS, {}),
172 CONF_CAMERAS: config.get(CONF_CAMERAS, {}),
173 CONF_DUPE_CHECK: config.get(CONF_DUPE_CHECK, {}),
174 CONF_SNOOZE: config.get(CONF_SNOOZE, {}),
175 }
177 def supplemental_action_refresh_entities(_call: ServiceCall) -> None:
178 return service.expose_entities()
180 def supplemental_action_enquire_implicit_deliveries(_call: ServiceCall) -> dict[str, Any]:
181 return service.enquire_implicit_deliveries()
183 def supplemental_action_enquire_deliveries_by_scenario(_call: ServiceCall) -> dict[str, Any]:
184 return service.enquire_deliveries_by_scenario()
186 def supplemental_action_enquire_last_notification(_call: ServiceCall) -> dict[str, Any]:
187 return service.last_notification.contents() if service.last_notification else {}
189 async def supplemental_action_enquire_active_scenarios(call: ServiceCall) -> dict[str, Any]:
190 trace = call.data.get("trace", False)
191 result: dict[str, Any] = {"scenarios": await service.enquire_active_scenarios()}
192 if trace:
193 result["trace"] = await service.trace_active_scenarios()
194 return result
196 def supplemental_action_enquire_scenarios(_call: ServiceCall) -> dict[str, Any]:
197 return {"scenarios": service.enquire_scenarios()}
199 async def supplemental_action_enquire_occupancy(_call: ServiceCall) -> dict[str, Any]:
200 return {"scenarios": await service.enquire_occupancy()}
202 def supplemental_action_enquire_snoozes(_call: ServiceCall) -> dict[str, Any]:
203 return {"snoozes": service.enquire_snoozes()}
205 def supplemental_action_clear_snoozes(_call: ServiceCall) -> dict[str, Any]:
206 return {"cleared": service.clear_snoozes()}
208 def supplemental_action_enquire_recipients(_call: ServiceCall) -> dict[str, Any]:
209 return {"recipients": service.enquire_recipients()}
211 async def supplemental_action_purge_archive(call: ServiceCall) -> dict[str, Any]:
212 days = call.data.get("days")
213 if not service.context.archive.enabled:
214 return {"error": "No archive configured"}
215 purged = await service.context.archive.cleanup(days=days, force=True)
216 arch_size = await service.context.archive.size()
217 return {
218 "purged": purged,
219 "remaining": arch_size,
220 "interval": ARCHIVE_PURGE_MIN_INTERVAL,
221 "days": service.context.archive.archive_days if days is None else days,
222 }
224 async def supplemental_action_purge_media(call: ServiceCall) -> dict[str, Any]:
225 days = call.data.get("days")
226 if not service.context.media_storage.media_path:
227 return {"error": "No media storage configured"}
228 purged = await service.context.media_storage.cleanup(days=days, force=True)
229 size = await service.context.media_storage.size()
230 return {
231 "purged": purged,
232 "remaining": size,
233 "interval": service.context.media_storage.purge_minute_interval,
234 "days": service.context.media_storage.days if days is None else days,
235 }
237 hass.services.async_register(
238 DOMAIN,
239 "enquire_configuration",
240 supplemental_action_enquire_configuration,
241 supports_response=SupportsResponse.ONLY,
242 )
243 hass.services.async_register(
244 DOMAIN,
245 "enquire_implicit_deliveries",
246 supplemental_action_enquire_implicit_deliveries,
247 supports_response=SupportsResponse.ONLY,
248 )
249 hass.services.async_register(
250 DOMAIN,
251 "enquire_deliveries_by_scenario",
252 supplemental_action_enquire_deliveries_by_scenario,
253 supports_response=SupportsResponse.ONLY,
254 )
255 hass.services.async_register(
256 DOMAIN,
257 "enquire_last_notification",
258 supplemental_action_enquire_last_notification,
259 supports_response=SupportsResponse.ONLY,
260 )
261 hass.services.async_register(
262 DOMAIN,
263 "enquire_active_scenarios",
264 supplemental_action_enquire_active_scenarios,
265 supports_response=SupportsResponse.ONLY,
266 )
267 hass.services.async_register(
268 DOMAIN,
269 "enquire_scenarios",
270 supplemental_action_enquire_scenarios,
271 supports_response=SupportsResponse.ONLY,
272 )
273 hass.services.async_register(
274 DOMAIN,
275 "enquire_occupancy",
276 supplemental_action_enquire_occupancy,
277 supports_response=SupportsResponse.ONLY,
278 )
279 hass.services.async_register(
280 DOMAIN,
281 "enquire_recipients",
282 supplemental_action_enquire_recipients,
283 supports_response=SupportsResponse.ONLY,
284 )
285 hass.services.async_register(
286 DOMAIN,
287 "enquire_snoozes",
288 supplemental_action_enquire_snoozes,
289 supports_response=SupportsResponse.ONLY,
290 )
291 hass.services.async_register(
292 DOMAIN,
293 "clear_snoozes",
294 supplemental_action_clear_snoozes,
295 supports_response=SupportsResponse.ONLY,
296 )
297 hass.services.async_register(
298 DOMAIN,
299 "purge_archive",
300 supplemental_action_purge_archive,
301 supports_response=SupportsResponse.ONLY,
302 )
303 hass.services.async_register(
304 DOMAIN,
305 "purge_media",
306 supplemental_action_purge_media,
307 supports_response=SupportsResponse.ONLY,
308 )
309 hass.services.async_register(
310 DOMAIN,
311 "refresh_entities",
312 supplemental_action_refresh_entities,
313 supports_response=SupportsResponse.NONE,
314 )
316 return service
319class SupernotifyEntity(NotifyEntity):
320 """Implement supernotify as a NotifyEntity platform."""
322 _attr_has_entity_name = True
323 _attr_name = "supernotify"
325 def __init__(
326 self,
327 unique_id: str,
328 platform: SupernotifyAction,
329 ) -> None:
330 """Initialize the SuperNotify entity."""
331 self._attr_unique_id = unique_id
332 self._attr_supported_features = NotifyEntityFeature.TITLE
333 self._platform = platform
335 async def async_send_message(
336 self, message: str, title: str | None = None, target: str | list[str] | None = None, data: dict[str, Any] | None = None
337 ) -> None:
338 """Send a message to a user."""
339 await self._platform.async_send_message(message, title=title, target=target, data=data)
342class SupernotifyAction(BaseNotificationService):
343 """Implement SuperNotify Action"""
345 def __init__(
346 self,
347 hass: HomeAssistant,
348 deliveries: dict[str, dict[str, Any]] | None = None,
349 template_path: str | None = None,
350 media_path: str | None = None,
351 media_url_prefix: str | None = None,
352 archive: dict[str, Any] | None = None,
353 housekeeping: dict[str, Any] | None = None,
354 recipients_discovery: bool = True,
355 mobile_discovery: bool = True,
356 recipients: list[dict[str, Any]] | None = None,
357 mobile_actions: dict[str, Any] | None = None,
358 scenarios: dict[str, dict[str, Any]] | None = None,
359 links: list[str] | None = None,
360 transport_configs: dict[str, Any] | None = None,
361 cameras: list[dict[str, Any]] | None = None,
362 dupe_check: dict[str, Any] | None = None,
363 snooze: dict[str, Any] | None = None,
364 ) -> None:
365 """Initialize the service."""
366 self.last_notification: Notification | None = None
367 self.failures: int = 0
368 self.housekeeping: dict[str, Any] = housekeeping or {}
369 self.sent: int = 0
370 hass_api = HomeAssistantAPI(hass)
372 self.context = Context(
373 hass_api,
374 PeopleRegistry(recipients or [], hass_api, discover=recipients_discovery, mobile_discovery=mobile_discovery),
375 ScenarioRegistry(scenarios or {}),
376 DeliveryRegistry(deliveries or {}, transport_configs or {}, TRANSPORTS),
377 DupeChecker(dupe_check or {}),
378 NotificationArchive(archive or {}, hass_api),
379 MediaStorage(
380 media_path,
381 media_url_prefix=media_url_prefix,
382 days=self.housekeeping.get(CONF_MEDIA_STORAGE_DAYS, 7),
383 ),
384 Snoozer(snooze),
385 links or [],
386 recipients or [],
387 mobile_actions,
388 template_path,
389 cameras=cameras,
390 )
392 self.exposed_entities: list[str] = []
394 async def initialize(self) -> None:
395 await self.context.initialize()
396 self.context.hass_api.initialize()
397 self.context.people_registry.initialize()
398 await self.context.delivery_registry.initialize(self.context)
399 await self.context.scenario_registry.initialize(
400 self.context.delivery_registry,
401 self.context.mobile_actions,
402 self.context.hass_api,
403 )
404 await self.context.archive.initialize()
405 await self.context.media_storage.initialize(self.context.hass_api)
407 self.expose_entities()
408 self.context.hass_api.subscribe_event("mobile_app_notification_action", self.on_mobile_action)
409 self.context.hass_api.subscribe_state(self.exposed_entities, self._entity_state_change_listener)
411 housekeeping_schedule = self.housekeeping.get(CONF_HOUSEKEEPING_TIME)
412 if housekeeping_schedule:
413 _LOGGER.info("SUPERNOTIFY setting up housekeeping schedule at: %s", housekeeping_schedule)
414 self.context.hass_api.subscribe_time(
415 housekeeping_schedule.hour, housekeeping_schedule.minute, housekeeping_schedule.second, self.async_nightly_tasks
416 )
417 else:
418 _LOGGER.info(
419 "SUPERNOTIFY Housekeeping disabled. Storage must be manually managed if using attachments or image snapshots"
420 )
422 self.context.hass_api.subscribe_event(EVENT_HOMEASSISTANT_STOP, self.async_shutdown)
424 async def async_shutdown(self, event: Event) -> None:
425 _LOGGER.info("SUPERNOTIFY shutting down, %s (%s)", event.event_type, event.time_fired)
426 self.shutdown()
428 async def async_unregister_services(self) -> None:
429 _LOGGER.info("SUPERNOTIFY unregistering")
430 self.shutdown()
431 return await super().async_unregister_services()
433 def shutdown(self) -> None:
434 self.context.hass_api.disconnect()
435 _LOGGER.info("SUPERNOTIFY shut down")
437 async def async_send_message(
438 self, message: str = "", title: str | None = None, target: list[str] | str | None = None, **kwargs: Any
439 ) -> None:
440 """Send a message via chosen transport."""
441 data = kwargs.get(ATTR_DATA, {})
442 notification = None
443 _LOGGER.debug("Message: %s, target: %s, data: %s", message, target, data)
445 try:
446 notification = Notification(self.context, message, title, target, data)
447 await notification.initialize()
448 if await notification.deliver():
449 self.sent += 1
450 self.context.hass_api.set_state(f"sensor.{DOMAIN}_notifications", self.sent)
451 elif notification.failed:
452 _LOGGER.error("SUPERNOTIFY Failed to deliver %s, error count %s", notification.id, notification.error_count)
453 else:
454 if notification.delivered == 0:
455 codes: list[SuppressionReason] = notification._skip_reasons
456 reason: str = ",".join(str(code) for code in codes)
457 problem: bool = codes != [SuppressionReason.DUPE]
458 else:
459 problem = True
460 reason = "No delivery envelopes generated"
461 if problem:
462 _LOGGER.warning("SUPERNOTIFY No deliveries made for %s: %s", notification.id, reason)
463 else:
464 _LOGGER.debug("SUPERNOTIFY Deliveries suppressed for %s: %s", notification.id, reason)
466 except Exception as err:
467 # fault barrier of last resort, integration failures should be caught within envelope delivery
468 _LOGGER.exception("SUPERNOTIFY Failed to send message %s", message)
469 self.failures += 1
470 if notification is not None:
471 notification._delivery_error = format_exception(err)
472 self.context.hass_api.set_state(f"sensor.{DOMAIN}_failures", self.failures)
474 if notification is None:
475 _LOGGER.warning("SUPERNOTIFY NULL Notification, %s", message)
476 else:
477 self.last_notification = notification
478 await self.context.archive.archive(notification)
479 _LOGGER.debug(
480 "SUPERNOTIFY %s deliveries, %s failed, %s skipped, %s suppressed",
481 notification.delivered,
482 notification.failed,
483 notification.skipped,
484 notification.suppressed,
485 )
487 async def _entity_state_change_listener(self, event: Event[EventStateChangedData]) -> None:
488 changes = 0
489 if event is not None:
490 _LOGGER.debug(f"SUPERNOTIFY {event.event_type} event for entity: {event.data}")
491 new_state: State | None = event.data["new_state"]
492 if new_state and event.data["entity_id"].startswith(f"binary_sensor.{DOMAIN}_scenario_"):
493 scenario: Scenario | None = self.context.scenario_registry.scenarios.get(
494 event.data["entity_id"].replace(f"binary_sensor.{DOMAIN}_scenario_", "")
495 )
496 if scenario is None:
497 _LOGGER.warning(f"SUPERNOTIFY Event for unknown scenario {event.data['entity_id']}")
498 else:
499 if new_state.state == "off" and scenario.enabled:
500 scenario.enabled = False
501 _LOGGER.info(f"SUPERNOTIFY Disabling scenario {scenario.name}")
502 changes += 1
503 elif new_state.state == "on" and not scenario.enabled:
504 scenario.enabled = True
505 _LOGGER.info(f"SUPERNOTIFY Enabling scenario {scenario.name}")
506 changes += 1
507 else:
508 _LOGGER.info(f"SUPERNOTIFY No change to scenario {scenario.name}, already {new_state}")
509 elif new_state and event.data["entity_id"].startswith(f"binary_sensor.{DOMAIN}_delivery_"):
510 delivery_name: str = event.data["entity_id"].replace(f"binary_sensor.{DOMAIN}_delivery_", "")
511 if new_state.state == "off":
512 if self.context.delivery_registry.disable(delivery_name):
513 changes += 1
514 elif new_state.state == "on":
515 if self.context.delivery_registry.enable(delivery_name):
516 changes += 1
517 else:
518 _LOGGER.info(f"SUPERNOTIFY No change to delivery {delivery_name} for state {new_state.state}")
519 elif new_state and event.data["entity_id"].startswith(f"binary_sensor.{DOMAIN}_transport_"):
520 transport: Transport | None = self.context.delivery_registry.transports.get(
521 event.data["entity_id"].replace(f"binary_sensor.{DOMAIN}_transport_", "")
522 )
523 if transport is None:
524 _LOGGER.warning(f"SUPERNOTIFY Event for unknown transport {event.data['entity_id']}")
525 else:
526 if new_state.state == "off" and transport.enabled:
527 transport.enabled = False
528 _LOGGER.info(f"SUPERNOTIFY Disabling transport {transport.name}")
529 changes += 1
530 elif new_state.state == "on" and not transport.enabled:
531 transport.enabled = True
532 _LOGGER.info(f"SUPERNOTIFY Enabling transport {transport.name}")
533 changes += 1
534 else:
535 _LOGGER.info(f"SUPERNOTIFY No change to transport {transport.name}, already {new_state}")
536 elif new_state and event.data["entity_id"].startswith(f"binary_sensor.{DOMAIN}_recipient_"):
537 recipient: Recipient | None = self.context.people_registry.people.get(
538 event.data["entity_id"].replace(f"binary_sensor.{DOMAIN}_recipient_", "person.")
539 )
540 if recipient is None:
541 _LOGGER.warning(f"SUPERNOTIFY Event for unknown recipient {event.data['entity_id']}")
542 else:
543 if new_state.state == "off" and recipient.enabled:
544 recipient.enabled = False
545 _LOGGER.info(f"SUPERNOTIFY Disabling recipient {recipient.entity_id}")
546 changes += 1
547 elif new_state.state == "on" and not recipient.enabled:
548 recipient.enabled = True
549 _LOGGER.info(f"SUPERNOTIFY Enabling recipient {recipient.entity_id}")
550 changes += 1
551 else:
552 _LOGGER.info(f"SUPERNOTIFY No change to recipient {recipient.entity_id}, already {new_state}")
554 else:
555 _LOGGER.warning("SUPERNOTIFY entity event with nothing to do:%s", event)
557 def expose_entity(
558 self,
559 entity_name: str,
560 state: str,
561 attributes: dict[str, Any],
562 platform: str = Platform.BINARY_SENSOR,
563 original_name: str | None = None,
564 original_icon: str | None = None,
565 entity_registry: er.EntityRegistry | None = None,
566 ) -> None:
567 """Expose a technical entity in Home Assistant representing internal state and attributes"""
568 entity_id: str
569 if entity_registry is not None:
570 try:
571 entry: er.RegistryEntry = entity_registry.async_get_or_create(
572 platform,
573 DOMAIN,
574 entity_name,
575 entity_category=EntityCategory.DIAGNOSTIC,
576 original_name=original_name,
577 original_icon=original_icon,
578 )
579 entity_id = entry.entity_id
580 except Exception as e:
581 _LOGGER.warning("SUPERNOTIFY Unable to register entity %s: %s", entity_name, e)
582 # continue anyway even if not registered as state is independent of entity
583 entity_id = f"{platform}.{DOMAIN}_{entity_name}"
584 try:
585 self.context.hass_api.set_state(entity_id, state, attributes)
586 self.exposed_entities.append(entity_id)
587 except Exception as e:
588 _LOGGER.error("SUPERNOTIFY Unable to set state for entity %s: %s", entity_id, e)
590 def expose_entities(self) -> None:
591 # Create on the fly entities for key internal config and state
592 ent_reg: er.EntityRegistry | None = self.context.hass_api.entity_registry()
593 if ent_reg is None:
594 _LOGGER.error("SUPERNOTIFY Unable to access entity registry to expose entities")
595 return
597 self.context.hass_api.set_state(f"sensor.{DOMAIN}_failures", self.failures)
598 self.context.hass_api.set_state(f"sensor.{DOMAIN}_notifications", self.sent)
600 for scenario in self.context.scenario_registry.scenarios.values():
601 self.expose_entity(
602 f"scenario_{scenario.name}",
603 state=STATE_UNKNOWN,
604 attributes=sanitize(scenario.attributes(include_condition=False)),
605 original_name=f"{scenario.name} Scenario",
606 original_icon="mdi:clipboard-text",
607 entity_registry=ent_reg,
608 )
609 for transport in self.context.delivery_registry.transports.values():
610 self.expose_entity(
611 f"transport_{transport.name}",
612 state=STATE_ON if transport.enabled else STATE_OFF,
613 attributes=sanitize(transport.attributes()),
614 original_name=f"{transport.name} Transport Adaptor",
615 original_icon="mdi:truck-fast",
616 entity_registry=ent_reg,
617 )
619 for delivery in self.context.delivery_registry.deliveries.values():
620 self.expose_entity(
621 f"delivery_{delivery.name}",
622 state=STATE_ON if delivery.enabled else STATE_OFF,
623 attributes=sanitize(delivery.attributes()),
624 original_name=f"{delivery.name} Delivery Configuration",
625 original_icon="mdi:package-variant",
626 entity_registry=ent_reg,
627 )
629 for recipient in self.context.people_registry.people.values():
630 self.expose_entity(
631 f"recipient_{recipient.name}",
632 state=STATE_ON if recipient.enabled else STATE_OFF,
633 attributes=sanitize(recipient.attributes()),
634 original_name=f"{recipient.name}",
635 original_icon="mdi:account-arrow-left",
636 entity_registry=ent_reg,
637 )
639 def enquire_implicit_deliveries(self) -> dict[str, Any]:
640 v: dict[str, list[str]] = {}
641 for t in self.context.delivery_registry.transports:
642 for d in self.context.delivery_registry.implicit_deliveries:
643 if d.transport.name == t:
644 v.setdefault(t, [])
645 v[t].append(d.name)
646 return v
648 def enquire_deliveries_by_scenario(self) -> dict[str, dict[str, list[str]]]:
649 return {
650 name: {
651 "enabled": scenario.enabling_deliveries(),
652 "disabled": scenario.disabling_deliveries(),
653 "applies": scenario.relevant_deliveries(),
654 }
655 for name, scenario in self.context.scenario_registry.scenarios.items()
656 if scenario.enabled
657 }
659 async def enquire_occupancy(self) -> dict[str, list[dict[str, Any]]]:
660 occupancy = self.context.people_registry.determine_occupancy()
661 return {k: [v.as_dict() for v in vs] for k, vs in occupancy.items()}
663 async def enquire_active_scenarios(self) -> list[str]:
664 occupiers: dict[str, list[Recipient]] = self.context.people_registry.determine_occupancy()
665 cvars = ConditionVariables([], [], [], PRIORITY_MEDIUM, occupiers, None, None)
666 return [s.name for s in self.context.scenario_registry.scenarios.values() if s.evaluate(cvars)]
668 async def trace_active_scenarios(self) -> tuple[list[dict[str, Any]], list[dict[str, Any]], dict[str, Any]]:
669 occupiers: dict[str, list[Recipient]] = self.context.people_registry.determine_occupancy()
670 cvars = ConditionVariables([], [], [], PRIORITY_MEDIUM, occupiers, None, None)
672 def safe_json(v: Any) -> Any:
673 return json.loads(json.dumps(v, cls=ExtendedJSONEncoder))
675 enabled = []
676 disabled = []
677 dcvars = asdict(cvars)
678 for s in self.context.scenario_registry.scenarios.values():
679 if await s.trace(cvars):
680 enabled.append(safe_json(s.attributes(include_trace=True)))
681 else:
682 disabled.append(safe_json(s.attributes(include_trace=True)))
683 return enabled, disabled, dcvars
685 def enquire_scenarios(self) -> dict[str, dict[str, Any]]:
686 return {s.name: s.attributes(include_condition=False) for s in self.context.scenario_registry.scenarios.values()}
688 def enquire_snoozes(self) -> list[dict[str, Any]]:
689 return self.context.snoozer.export()
691 def clear_snoozes(self) -> int:
692 return self.context.snoozer.clear()
694 def enquire_recipients(self) -> list[dict[str, Any]]:
695 return [p.as_dict() for p in self.context.people_registry.people.values()]
697 @callback
698 def on_mobile_action(self, event: Event) -> None:
699 """Listen for mobile actions relevant to snooze and silence notifications
701 Example Action:
702 event_type: mobile_app_notification_action
703 data:
704 foo: a
705 origin: REMOTE
706 time_fired: "2024-04-20T13:14:09.360708+00:00"
707 context:
708 id: 01HVXT93JGWEDW0KE57Z0X6Z1K
709 parent_id: null
710 user_id: a9dbae1a5abf33dbbad52ff82201bb17
711 """
712 event_name = event.data.get(ATTR_ACTION)
713 if event_name is None or not event_name.startswith("SUPERNOTIFY_"):
714 return # event not intended for here
715 self.context.snoozer.handle_command_event(event, self.context.people_registry.enabled_recipients())
717 @callback
718 async def async_nightly_tasks(self, now: dt.datetime) -> None:
719 _LOGGER.info("SUPERNOTIFY Housekeeping starting as scheduled at %s", now)
720 await self.context.archive.cleanup()
721 self.context.snoozer.purge_snoozes()
722 await self.context.media_storage.cleanup()
723 _LOGGER.info("SUPERNOTIFY Housekeeping completed")