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