Coverage for custom_components/supernotify/notification.py: 94%
705 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 14:29 +0000
1from __future__ import annotations
3import asyncio
4import datetime as dt
5import logging
6import uuid
7from dataclasses import dataclass
8from traceback import format_exception
9from typing import TYPE_CHECKING, Any, cast
11import homeassistant.util.dt as dt_util
12import voluptuous as vol
13from homeassistant.components.notify.const import ATTR_DATA
14from homeassistant.const import ATTR_DEVICE_ID, ATTR_ENTITY_ID
15from voluptuous import humanize
17from custom_components.supernotify.delivery import DeliveryProvenance
18from custom_components.supernotify.schema import SelectionRank
20from .archive import ArchivableObject
21from .common import ensure_list, nullable_ensure_list, sanitize
22from .const import (
23 ATTR_ACTION_GROUPS,
24 ATTR_ACTIONS,
25 ATTR_DEBUG,
26 ATTR_DELIVERY,
27 ATTR_DELIVERY_SELECTION,
28 ATTR_FORCE_RESEND,
29 ATTR_IMAGE,
30 ATTR_MEDIA,
31 ATTR_MEDIA_CAMERA_ENTITY_ID,
32 ATTR_MEDIA_CLIP_URL,
33 ATTR_MEDIA_SNAPSHOT_URL,
34 ATTR_MESSAGE_HTML,
35 ATTR_PERSON_ID,
36 ATTR_PRIORITY,
37 ATTR_RECIPIENTS,
38 ATTR_SCENARIOS_APPLY,
39 ATTR_SCENARIOS_CONSTRAIN,
40 ATTR_SCENARIOS_REQUIRE,
41 ATTR_SPOKEN_MESSAGE,
42 ATTR_VIDEO,
43 DELIVERY_SELECTION_EXPLICIT,
44 DELIVERY_SELECTION_FIXED,
45 DELIVERY_SELECTION_IMPLICIT,
46 INCLUSION_DEFAULT,
47 PRIORITY_MEDIUM,
48 PRIORITY_VALUES,
49 TARGET_USE_FIXED,
50 TARGET_USE_MERGE_ALWAYS,
51 TARGET_USE_MERGE_ON_DELIVERY_TARGETS,
52 TARGET_USE_ON_NO_ACTION_TARGETS,
53 TARGET_USE_ON_NO_DELIVERY_TARGETS,
54)
55from .envelope import Envelope
56from .media_grab import snap_notification_image as _snap_notification_image
57from .model import (
58 ConditionVariables,
59 DebugTrace,
60 DeliveryCustomization,
61 SuppressionReason,
62 Target,
63 TargetRequired,
64 TransportFeature,
65)
66from .options import OPTION_UNIQUE_TARGETS
67from .schema import ACTION_DATA_SCHEMA, STRICT_ACTION_DATA_SCHEMA, DeliveryOutcome, EnvelopeOutcome, OutcomeSelection
69if TYPE_CHECKING:
70 from homeassistant.core import Context as HAContext
72 from .context import Context
73 from .delivery import Delivery, DeliveryRegistry
74 from .people import PeopleRegistry, Recipient
75 from .scenario import Scenario
76 from .transport import (
77 Transport,
78 )
80_LOGGER = logging.getLogger(__name__)
82_VERSION: str = "unknown"
85def set_version(version: str) -> None:
86 """Called once from async_setup_entry with the integration's manifest version.
88 Avoids a blocking manifest.json read on the event loop at import time.
89 """
90 global _VERSION
91 _VERSION = version
94# supernotify specific data items not to be passed to transports in data
95# the existence of this is probably an indication of something not quite right
97type DeliveryName = str
100@dataclass
101class DeliveryTargetOverride:
102 fixed: list[str]
103 include: list[str]
104 exclude: list[str]
106 def as_dict(self, **_kwargs: Any) -> dict[str, Any]:
107 return {"fixed": self.fixed, "include": self.include, "exclude": self.exclude}
110class Notification(ArchivableObject):
111 def __init__(
112 self,
113 context: Context,
114 message: str | None = None,
115 title: str | None = None,
116 target: list[str] | str | dict[str, Any] | None = None,
117 action_data: dict[str, Any] | None = None,
118 ha_context: HAContext | None = None,
119 ) -> None:
120 self.created: dt.datetime = dt.datetime.now(tz=dt_util.get_default_time_zone())
121 self.debug_trace: DebugTrace = DebugTrace(
122 message=message,
123 title=title,
124 data=action_data,
125 target=target,
126 debug=bool((action_data or {}).get(ATTR_DEBUG, False)),
127 )
128 self.message: str | None = message
129 self.context: Context = context
130 self.ha_context: HAContext | None = ha_context
131 self.people_registry: PeopleRegistry = context.people_registry
132 self.delivery_registry: DeliveryRegistry = context.delivery_registry
133 action_data = action_data or {}
135 if action_data.get(ATTR_RECIPIENTS) and not isinstance(target, dict):
136 _LOGGER.warning("SUPERNOTIFY recipients key in notify action data deprecated, list recipients in target instead")
137 target = ensure_list(target) + ensure_list(action_data.get(ATTR_RECIPIENTS))
139 self._target: Target | None = Target(self.convert_notify_entities(target)) if target else None
140 # targeting this integration's own device is never a real target - treat it as a request to
141 # also apply each default delivery's own target alongside whatever else was targeted (see
142 # HomeAssistantAPI.is_own_device and the merge in generate_targets)
143 self._apply_delivery_defaults: bool = False
144 if self._target and self._target.device_ids:
145 own_device_ids = [d for d in self._target.device_ids if context.hass_api.is_own_device(d)]
146 if own_device_ids:
147 self._target.remove(ATTR_DEVICE_ID, own_device_ids)
148 self._apply_delivery_defaults = True
149 self._already_selected: Target = Target()
150 self._recorded_person_ids: set[str] = set() # recipients whose notify entity this notification has updated
151 self._title: str | None = title
152 self.id = str(uuid.uuid1())
153 self.delivered: int = 0
154 self.error_count: int = 0
155 self.skipped: int = 0
156 self.failed: int = 0
157 self.suppressed: int = 0
158 self.fallback: int = 0
159 self.dupe: bool = False
160 self.deliveries: dict[DeliveryName, dict[EnvelopeOutcome, list[str] | list[Envelope] | dict[str, Any]]] = {}
161 self.delivery_exceptions: dict[DeliveryName, list[str]] = {}
162 self.uncategorized_targets: dict[DeliveryName, list[str]] = {}
163 self._skip_reasons: list[SuppressionReason] = []
165 self.validate_action_data(action_data)
166 # for compatibility with other notify calls, pass thru surplus data to underlying delivery transports
167 self.extra_data: dict[str, Any] = {
168 k: v for k, v in action_data.items() if k not in STRICT_ACTION_DATA_SCHEMA(action_data)
169 }
171 action_data = {k: v for k, v in action_data.items() if k not in self.extra_data}
172 self.extra_data.update(action_data.get(ATTR_DATA, {})) # nested `data` could be supernotify or target service
174 self.priority: str = action_data.get(ATTR_PRIORITY, PRIORITY_MEDIUM)
175 self.message_html: str | None = action_data.get(ATTR_MESSAGE_HTML)
176 self.spoken_message: str | None = action_data.get(ATTR_SPOKEN_MESSAGE)
177 self.force_resend: bool = action_data.get(ATTR_FORCE_RESEND, False)
178 self.required_scenario_names: list[str] = ensure_list(action_data.get(ATTR_SCENARIOS_REQUIRE))
179 self.applied_scenario_names: list[str] = ensure_list(action_data.get(ATTR_SCENARIOS_APPLY))
180 self.constrain_scenario_names: list[str] = ensure_list(action_data.get(ATTR_SCENARIOS_CONSTRAIN))
181 self.delivery_selection: str | None = action_data.get(ATTR_DELIVERY_SELECTION)
182 self.delivery_overrides: dict[str, DeliveryCustomization] = {}
184 delivery_data: Any | None = action_data.get(ATTR_DELIVERY)
185 if self.delivery_selection is None:
186 if isinstance(delivery_data, list) or (isinstance(delivery_data, str) and delivery_data):
187 self.delivery_selection = DELIVERY_SELECTION_EXPLICIT
188 else:
189 self.delivery_selection = DELIVERY_SELECTION_IMPLICIT
190 if isinstance(delivery_data, list):
191 # a bare list of deliveries implies intent to restrict
192 _LOGGER.debug("SUPERNOTIFY Defaulting delivery selection as explicit for list %s", delivery_data)
193 self.delivery_overrides = {
194 k: DeliveryCustomization(config=None, default_enabled=True) for k in action_data.get(ATTR_DELIVERY, [])
195 }
196 elif isinstance(delivery_data, str) and delivery_data:
197 # a bare list of deliveries implies intent to restrict
198 _LOGGER.debug("SUPERNOTIFY Defaulting delivery selection as explicit for single %s", delivery_data)
199 self.delivery_overrides = {delivery_data: DeliveryCustomization(config=None, default_enabled=True)}
200 elif isinstance(delivery_data, dict):
201 # whereas a dict may be used to tune or restrict
202 _LOGGER.debug("SUPERNOTIFY Defaulting delivery selection as implicit for mapping %s", delivery_data)
203 self.delivery_overrides = {
204 k: DeliveryCustomization(config={**(v or {})}, default_enabled=True)
205 for k, v in action_data.get(ATTR_DELIVERY, {}).items()
206 }
207 elif delivery_data:
208 _LOGGER.warning("SUPERNOTIFY Unable to interpret delivery data %s", delivery_data)
210 self.action_groups: list[str] | None = nullable_ensure_list(action_data.get(ATTR_ACTION_GROUPS))
211 self.media: dict[str, Any] = action_data.get(ATTR_MEDIA) or {}
212 self.debug: bool = action_data.get(ATTR_DEBUG, False)
213 self.actions: list[dict[str, Any]] = ensure_list(action_data.get(ATTR_ACTIONS))
215 self.selected_deliveries: dict[str, DeliveryTargetOverride | None] = {}
216 self.enabled_scenarios: dict[str, Scenario] = {}
217 self.selected_scenario_names: list[str] = []
218 self._suppression_reason: SuppressionReason | None = None
219 self._raw_image_path: Any = None
220 self._delivery_error: list[str] | None = None
221 self.condition_variables: ConditionVariables
222 self.unknown_names: dict[str, list[str]] = self._unknown_names()
224 def _unknown_names(self) -> dict[str, list[str]]:
225 """Delivery and scenario names in the call that aren't configured, which are otherwise ignored"""
226 if not (
227 self.delivery_overrides
228 or self.required_scenario_names
229 or self.applied_scenario_names
230 or self.constrain_scenario_names
231 ):
232 return {}
233 registry: DeliveryRegistry = self.delivery_registry
234 known_deliveries: set[str] = {*registry.deliveries, *registry.transports}
235 scenarios: list[str] = list(self.context.scenario_registry.scenarios)
236 requested: dict[str, tuple[list[str], list[str]]] = {
237 ATTR_DELIVERY: (
238 [n for n in self.delivery_overrides if registry.resolve_name(n) not in known_deliveries],
239 list(registry.deliveries),
240 ),
241 ATTR_SCENARIOS_REQUIRE: ([n for n in self.required_scenario_names if n not in scenarios], scenarios),
242 ATTR_SCENARIOS_APPLY: ([n for n in self.applied_scenario_names if n not in scenarios], scenarios),
243 ATTR_SCENARIOS_CONSTRAIN: ([n for n in self.constrain_scenario_names if n not in scenarios], scenarios),
244 }
245 unknown: dict[str, list[str]] = {}
246 for field, (names, valid) in requested.items():
247 if names:
248 _LOGGER.warning(
249 "SUPERNOTIFY Ignoring unknown %s %s, configured names are: %s",
250 field,
251 ", ".join(names),
252 ", ".join(valid) or "none",
253 )
254 unknown[field] = names
255 return unknown
257 async def initialize(self) -> None:
258 """Async post-construction initialization"""
259 self.occupancy: dict[str, list[Recipient]] = self.people_registry.determine_occupancy()
260 self.condition_variables = ConditionVariables(
261 self.applied_scenario_names,
262 self.required_scenario_names,
263 self.constrain_scenario_names,
264 self.priority,
265 self.occupancy,
266 self.message,
267 self._title,
268 self.extra_data,
269 ) # requires occupancy first
271 enabled_scenario_names: list[str] = list(self.applied_scenario_names) or []
272 self.selected_scenario_names = await self.select_scenarios()
273 enabled_scenario_names.extend(self.selected_scenario_names)
274 if self.constrain_scenario_names:
275 enabled_scenario_names = [
276 s for s in enabled_scenario_names if (s in self.constrain_scenario_names or s in self.applied_scenario_names)
277 ]
278 if self.required_scenario_names and not any(s in enabled_scenario_names for s in self.required_scenario_names):
279 _LOGGER.info("SUPERNOTIFY Suppressing notification, no required scenarios enabled")
280 self.selected_deliveries = {}
281 self.suppress(SuppressionReason.NO_SCENARIO)
282 else:
283 for s in enabled_scenario_names:
284 scenario_obj: Scenario | None = self.context.scenario_registry.scenarios.get(s)
285 if scenario_obj is not None:
286 self.enabled_scenarios[s] = scenario_obj
288 self.selected_deliveries = self.select_deliveries()
289 if self.context.snoozer.is_global_snooze(self.priority):
290 self.suppress(SuppressionReason.SNOOZED)
291 self.apply_enabled_scenarios()
293 if not self.media:
294 self.media = self.media_requirements(self.extra_data)
296 def convert_notify_entities(
297 self, target: list[str] | str | dict[str, Any] | None = None
298 ) -> list[str] | str | dict[str, Any] | None:
299 """Short circuit supernotify notify entities so they're handled directly so not
300 going round in circles via calls to notify.send_message. A genuine other-integration
301 notify entity is left alone for NotifyEntityTransport to handle normally.
303 Defined here rather than in models/Target since requires access to registries.
305 supernotify.notify's `target:` field also accepts the dict shape Home Assistant's own
306 target selector produces, e.g. {"entity_id": ["person.jey", "notify.recipient_alice"]} -
307 unlike this integration's other dict-shaped targets (recipient/delivery config), that
308 `entity_id` list is Home Assistant's raw picker output, not pre-sorted by category: a
309 person entity picked that way still needs to end up as a `person_id`, same as if it had
310 been typed directly into notify.supernotify's flat target list, or it would otherwise be
311 silently dropped by Target() (whose entity_id category explicitly excludes the person
312 domain - see Target.is_entity_id/is_person_id in model.py).
313 """
314 if not target:
315 return target
316 known_entities: dict[str, Recipient] = self.people_registry.notify_entities()
318 if isinstance(target, dict):
319 entity_ids = ensure_list(target.get(ATTR_ENTITY_ID))
320 if not entity_ids:
321 return target
322 remaining_entity_ids: list[str] = []
323 person_ids: list[str] = list(ensure_list(target.get(ATTR_PERSON_ID)))
324 for e in entity_ids:
325 if e in known_entities:
326 # e.g. switch to a person id if a notify entity for recipient
327 person_ids.append(known_entities[e].entity_id)
328 elif Target.is_person_id(e):
329 person_ids.append(e)
330 else:
331 remaining_entity_ids.append(e)
332 updated_target: dict[str, list[str] | Any] = {**target, ATTR_ENTITY_ID: remaining_entity_ids}
333 if person_ids:
334 updated_target[ATTR_PERSON_ID] = person_ids
335 return updated_target
337 updated: list[str] = []
338 changed: bool = False
339 for t in ensure_list(target):
340 if t in known_entities:
341 # e.g. switch to a person id if a notify entity for recipient
342 updated.append(known_entities[t].entity_id)
343 changed = True
344 else:
345 updated.append(t)
347 if changed:
348 return updated
349 return target
351 def outcome(self) -> DeliveryOutcome:
352 if self.error_count > 0:
353 return DeliveryOutcome.ERROR
354 if self.dupe:
355 return DeliveryOutcome.DUPE
356 if not self.delivered:
357 return DeliveryOutcome.NO_DELIVERY
358 if self.fallback:
359 return DeliveryOutcome.FALLBACK_DELIVERY
360 if self.skipped:
361 return DeliveryOutcome.PARTIAL_DELIVERY
362 return DeliveryOutcome.SUCCESS
364 def diagnostics_selected(self, outcome_policy: OutcomeSelection) -> bool:
365 """A notification sent with `debug: true` asked for its trace, so it is archived
366 with the full diagnostic content whatever the configured `diagnostics` outcomes"""
367 return bool(self.debug) or super().diagnostics_selected(outcome_policy)
369 def media_requirements(self, data: dict[str, Any]) -> dict[str, Any]:
370 """If no media defined, look for iOS / Android actions that have media defined
372 Example is the Frigate blueprint, which generates `image`, `video` etc
373 in the `data` section, that can also be used for email attachments
374 """
375 media_dict: dict[str, Any | None] = {}
376 if not data:
377 return {}
378 if data.get(ATTR_IMAGE):
379 media_dict[ATTR_MEDIA_SNAPSHOT_URL] = data.get(ATTR_IMAGE)
380 if data.get(ATTR_VIDEO):
381 media_dict[ATTR_MEDIA_CLIP_URL] = data.get(ATTR_VIDEO)
382 if data.get("attachment", {}).get("url"):
383 url = data["attachment"]["url"]
384 if url and url.endswith(".mp4") and not media_dict.get(ATTR_MEDIA_CLIP_URL):
385 media_dict[ATTR_MEDIA_CLIP_URL] = url
386 elif url and (url.endswith((".jpg", ".jpeg", ".png"))) and not media_dict.get(ATTR_MEDIA_SNAPSHOT_URL):
387 media_dict[ATTR_MEDIA_SNAPSHOT_URL] = url
388 return media_dict
390 def validate_action_data(self, action_data: dict[str, Any]) -> None:
391 if action_data.get(ATTR_PRIORITY):
392 if isinstance(action_data.get(ATTR_PRIORITY), (str, int, float)):
393 if action_data.get(ATTR_PRIORITY) not in PRIORITY_VALUES:
394 _LOGGER.info("SUPERNOTIFY Custom priority %s", action_data.get(ATTR_PRIORITY))
395 else:
396 _LOGGER.info("SUPERNOTIFY Invalid priority %s", action_data.get(ATTR_PRIORITY))
397 self.suppress(SuppressionReason.INVALID_ACTION_DATA)
398 raise vol.Invalid("Priority value must be a simple value")
399 try:
400 humanize.validate_with_humanized_errors(action_data, cast("vol.Schema", ACTION_DATA_SCHEMA))
401 except vol.Invalid as e:
402 _LOGGER.warning("SUPERNOTIFY Invalid action data %s: %s", action_data, e)
403 self.suppress(SuppressionReason.INVALID_ACTION_DATA)
404 raise
405 except vol.error.Error as e2:
406 _LOGGER.warning("SUPERNOTIFY Failed to validate action data %s: %s", action_data, e2)
407 self.suppress(SuppressionReason.INVALID_ACTION_DATA)
408 raise vol.Invalid(f"Unable to validate action data - {e2}") from e2
410 def apply_enabled_scenarios(self) -> None:
411 """Set media and action_groups from scenario if defined, first come first applied"""
412 action_groups: list[str] = []
413 for scenario in self.enabled_scenarios.values():
414 if scenario.media:
415 if self.media:
416 self.media.update(scenario.media)
417 else:
418 self.media = scenario.media
419 if scenario.action_groups:
420 action_groups.extend(ag for ag in scenario.action_groups if ag not in action_groups)
421 # self.action_groups only accessed from inside Envelope
422 if self.action_groups:
423 self.action_groups.extend(action_groups)
424 else:
425 self.action_groups = action_groups
427 def select_deliveries(self) -> dict[str, DeliveryTargetOverride | None]:
428 # 'DEFAULT_x' backward compatibility: a delivery referenced by its bare
429 # transport name resolves to the auto-configured 'DEFAULT_x', if that's what
430 # actually exists in the registry.
431 resolve_name = self.context.delivery_registry.resolve_name
433 scenario_enable_deliveries: list[str] = []
434 scenario_disable_deliveries: list[str] = []
435 default_enable_deliveries: list[str] = []
436 recipients_enable_deliveries: list[str] = []
437 recipients_disable_deliveries: list[str] = []
438 all_recipients: list[Recipient] = self.all_recipients()
440 trace: DebugTrace = self.debug_trace
441 if self.delivery_selection != DELIVERY_SELECTION_FIXED:
442 for scenario in self.enabled_scenarios.values():
443 for d in scenario.enabling_deliveries():
444 name = resolve_name(d)
445 scenario_enable_deliveries.append(name)
446 trace.record_delivery_provenance(name, "enabled_by", f"scenario:{scenario.name}")
447 for scenario in self.enabled_scenarios.values():
448 for d in scenario.disabling_deliveries():
449 name = resolve_name(d)
450 scenario_disable_deliveries.append(name)
451 trace.record_delivery_provenance(name, "disabled_by", f"scenario:{scenario.name}")
453 scenario_enable_deliveries = list(dict.fromkeys(scenario_enable_deliveries))
454 scenario_disable_deliveries = list(dict.fromkeys(scenario_disable_deliveries))
456 for recipient in all_recipients:
457 for d in recipient.enabling_delivery_names():
458 name = resolve_name(d)
459 recipients_enable_deliveries.append(name)
460 trace.record_delivery_provenance(name, "enabled_by", f"recipient:{recipient.name}")
461 # See also Recipient.target() for the disabled cases
462 for d in recipient.disabling_delivery_names():
463 name = resolve_name(d)
464 recipients_disable_deliveries.append(name)
465 trace.record_delivery_provenance(name, "disabled_by", f"recipient:{recipient.name}")
466 if self.delivery_selection == DELIVERY_SELECTION_IMPLICIT:
467 # all deliveries with INCLUSION_DEFAULT in CONF_INCLUSION
468 default_enable_deliveries = [d.name for d in self.context.delivery_registry.implicit_deliveries]
469 for d in default_enable_deliveries:
470 trace.record_delivery_provenance(d, "enabled_by", "default")
472 trace.record_delivery_selection("scenario_enable_deliveries", scenario_enable_deliveries)
473 trace.record_delivery_selection("scenario_disable_deliveries", scenario_disable_deliveries)
474 trace.record_delivery_selection("default_enable_deliveries", default_enable_deliveries)
475 trace.record_delivery_selection("recipient_enable_deliveries", recipients_enable_deliveries)
477 override_enable_deliveries: list[str] = []
478 override_disable_deliveries: list[str] = []
480 # apply the deliveries defined in the notification action call
481 for delivery_name, delivery_override in self.delivery_overrides.items():
482 delivery = resolve_name(delivery_name)
483 if (
484 (delivery_override is None or delivery_override.enabled is not False)
485 and delivery in self.context.delivery_registry.enabled_deliveries
486 ) or (
487 (delivery_override is not None and delivery_override.enabled is not False)
488 and delivery in self.context.delivery_registry.disabled_deliveries
489 ):
490 override_enable_deliveries.append(delivery)
491 trace.record_delivery_provenance(delivery, "enabled_by", "call")
492 elif delivery_override is not None and delivery_override.enabled is False:
493 override_disable_deliveries.append(delivery)
494 trace.record_delivery_provenance(delivery, "disabled_by", "call")
496 all_global_enabled: list[str] = list(
497 dict.fromkeys(scenario_enable_deliveries + default_enable_deliveries + override_enable_deliveries)
498 )
499 all_enabled: list[str] = all_global_enabled + recipients_enable_deliveries
500 # override_enable_deliveries takes precedence: if the action call explicitly
501 # re-enables a delivery that a scenario disabled, remove it from all_disabled.
502 all_disabled: list[str] = [
503 d for d in scenario_disable_deliveries + override_disable_deliveries if d not in override_enable_deliveries
504 ]
505 override_enabled: list[str] = list(dict.fromkeys(scenario_enable_deliveries + override_enable_deliveries))
506 trace.record_delivery_selection("override_disable_deliveries", override_disable_deliveries)
507 trace.record_delivery_selection("override_enable_deliveries", override_enable_deliveries)
509 unsorted_maybe_objs: list[Delivery | None] = [
510 self.delivery_registry.deliveries.get(d) for d in all_enabled if d not in all_disabled
511 ]
512 unsorted_objs: list[Delivery] = [
513 d for d in unsorted_maybe_objs if d is not None and (d.enabled or d.name in override_enabled)
514 ]
515 first: list[str] = [d.name for d in unsorted_objs if d.selection_rank == SelectionRank.FIRST]
516 anywhere: list[str] = [d.name for d in unsorted_objs if d.selection_rank == SelectionRank.ANY]
517 config_last: list[str] = [
518 d.name
519 for d in unsorted_objs
520 if d.selection_rank == SelectionRank.LAST and d.provenance == DeliveryProvenance.CONFIG
521 ]
522 auto_last: list[str] = [
523 d.name
524 for d in unsorted_objs
525 if d.selection_rank == SelectionRank.LAST and d.provenance != DeliveryProvenance.CONFIG
526 ]
527 selected: list[str] = first + anywhere + config_last + auto_last
528 trace.record_delivery_selection("ranked", selected)
530 selected_deliveries: dict[str, DeliveryTargetOverride | None] = dict.fromkeys(selected)
531 personal_deliveries = [d for d in selected if d in recipients_enable_deliveries]
532 for personal_delivery in personal_deliveries:
533 fixed_targets: list[str] = []
534 include_targets: list[str] = []
535 exclude_targets: list[str] = []
536 for recipient in all_recipients:
537 if personal_delivery in (resolve_name(d) for d in recipient.enabling_delivery_names()):
538 if personal_delivery in all_global_enabled:
539 # delivery already selected, make sure this recipient also included
540 include_targets.append(recipient.entity_id)
541 else:
542 # this delivery has been explicitly enabled for individual recipients
543 fixed_targets.append(recipient.entity_id)
544 elif personal_delivery in (resolve_name(d) for d in recipient.disabling_delivery_names()):
545 exclude_targets.append(recipient.entity_id)
546 selected_deliveries[personal_delivery] = DeliveryTargetOverride(
547 fixed=fixed_targets, exclude=exclude_targets, include=include_targets
548 )
549 return selected_deliveries
551 def suppress(self, reason: SuppressionReason) -> None:
552 self._suppression_reason = reason
553 if reason not in self._skip_reasons:
554 self._skip_reasons.append(reason)
555 _LOGGER.info("SUPERNOTIFY Suppressing notification, reason:%s, id:%s", reason, self.id)
557 async def deliver(self) -> bool:
558 _LOGGER.debug(
559 "Message: %s, notification: %s, deliveries: %s",
560 self.message,
561 self.id,
562 self.selected_deliveries,
563 )
565 for delivery_name in self.selected_deliveries:
566 self.deliveries[delivery_name] = {}
568 if self._suppression_reason is not None:
569 _LOGGER.info("SUPERNOTIFY Suppressing globally silenced/snoozed notification (%s)", self.id)
570 for delivery_name in self.selected_deliveries:
571 delivery = self.context.delivery_registry.deliveries.get(delivery_name)
572 self.record_result(delivery, suppression_reason=self._suppression_reason)
573 else:
574 # Deliveries for transports that call grab_image() are deferred so that
575 # PTZ movement runs concurrently with non-image deliveries (chime, TTS, etc.)
576 camera_configured = bool(self.media.get(ATTR_MEDIA_CAMERA_ENTITY_ID) or self.media.get(ATTR_MEDIA_SNAPSHOT_URL))
577 immediate_deliveries: dict[str, DeliveryTargetOverride | None] = {}
578 deferred_deliveries: dict[str, DeliveryTargetOverride | None] = {}
579 for delivery_name, details in self.selected_deliveries.items():
580 d = self.context.delivery_registry.deliveries.get(delivery_name)
581 if d and camera_configured and d.transport.supported_features & TransportFeature.SNAPSHOT_IMAGE:
582 deferred_deliveries[delivery_name] = details
583 else:
584 immediate_deliveries[delivery_name] = details
586 # Start image grab immediately so PTZ runs while immediate deliveries execute
587 image_task: asyncio.Task | None = None
588 if deferred_deliveries:
589 image_task = asyncio.create_task(_snap_notification_image(self, self.context, ha_context=self.ha_context))
591 _LOGGER.debug("SUPERNOTIFY Scheduling %s immediate deliveries", len(deferred_deliveries))
592 await self._schedule_deliveries(immediate_deliveries)
594 # Ensure image is ready before running image-requiring deliveries
595 if image_task is not None:
596 wait_timeout: int = 30
597 try:
598 _LOGGER.debug("SUPERNOTIFY Waiting up to %s for image grab to complete", wait_timeout)
599 async with asyncio.timeout(wait_timeout): # TODO: configurable time-out
600 await image_task
601 except Exception:
602 _LOGGER.exception("SUPERNOTIFY Failed to pre-grab image")
603 if not image_task.done():
604 image_task.cancel()
606 _LOGGER.debug("SUPERNOTIFY Scheduling %s deferred deliveries", len(deferred_deliveries))
607 await self._schedule_deliveries(deferred_deliveries)
609 if self.delivered == 0 and not self._suppression_reason:
610 if self.failed == 0 and not self.dupe:
611 for delivery in self.context.delivery_registry.fallback_by_default_deliveries:
612 _LOGGER.info(
613 "SUPERNOTIFY No delivery succeeded, activating fallback_by_default: %s",
614 delivery.name,
615 )
616 if delivery.name not in self.selected_deliveries:
617 await self.call_transport(delivery)
618 self.fallback += 1
620 if self.failed > 0:
621 for delivery in self.context.delivery_registry.fallback_on_error_deliveries:
622 _LOGGER.warning(
623 "SUPERNOTIFY Delivery failed, activating fallback_on_error: %s",
624 delivery.name,
625 )
626 if delivery.name not in self.selected_deliveries:
627 await self.call_transport(delivery)
628 self.fallback += 1
630 return self.delivered > 0
632 async def _schedule_deliveries(self, deliveries: dict[str, DeliveryTargetOverride | None]) -> None:
633 delivery_coros = []
634 for delivery_name, target_overrides in deliveries.items():
635 delivery = self.context.delivery_registry.deliveries.get(delivery_name)
636 if delivery:
637 delivery_coros.append(self.call_transport(delivery, target_overrides))
638 else:
639 _LOGGER.error("SUPERNOTIFY Unexpected missing delivery %s", delivery_name)
640 if delivery_coros:
641 results = await asyncio.gather(*delivery_coros, return_exceptions=True)
642 for result in results:
643 if isinstance(result, BaseException):
644 _LOGGER.error("SUPERNOTIFY Unexpected error in parallel delivery: %s", result)
646 def delivery_skip_reason(self, delivery: Delivery) -> SuppressionReason | None:
647 """Why a selected delivery won't be attempted at all, checked before any targets are worked out"""
648 if not delivery.transport.enabled:
649 return SuppressionReason.TRANSPORT_DISABLED
650 if self.context.snoozer.is_delivery_snoozed(self.priority, delivery, self.media.get(ATTR_MEDIA_CAMERA_ENTITY_ID)):
651 return SuppressionReason.SNOOZED
652 if self.delivery_selection != DELIVERY_SELECTION_FIXED:
653 delivery_priorities: list[str] = delivery.priority
654 if self.priority and delivery_priorities and self.priority not in delivery_priorities:
655 return SuppressionReason.PRIORITY
656 if not delivery.evaluate_conditions(self.condition_variables):
657 return SuppressionReason.DELIVERY_CONDITION
658 return None
660 def plan(self) -> dict[str, Any]:
661 """Work out which deliveries would send, and to whom, without sending anything.
663 Makes the same checks as deliver(), except the dupe check, since the dupe checker
664 remembers what it's asked about.
665 """
666 deliveries: dict[str, dict[str, Any]] = {}
667 if self._suppression_reason is None:
668 for delivery_name, target_override in self.selected_deliveries.items():
669 delivery: Delivery | None = self.context.delivery_registry.deliveries.get(delivery_name)
670 if delivery is None:
671 continue
672 reason: SuppressionReason | None = self.delivery_skip_reason(delivery)
673 targets: list[Target] = [] if reason else self.generate_targets(delivery, target_override)
674 # the same test generate_envelopes() makes, without building the envelopes
675 if not reason and not any(
676 t.has_resolved_target() or delivery.target_required != TargetRequired.ALWAYS for t in targets
677 ):
678 reason = SuppressionReason.NO_TARGET
679 if reason:
680 deliveries[delivery_name] = {"skipped": str(reason)}
681 else:
682 deliveries[delivery_name] = {
683 "recipients": sorted({p for t in targets for p in t.person_ids}),
684 "targets": [t.direct().as_dict(redact=True) for t in targets if t.has_resolved_target()],
685 }
686 result: dict[str, Any] = {
687 "priority": self.priority,
688 "scenarios": list(self.enabled_scenarios),
689 "occupancy": {state: [r.entity_id for r in people] for state, people in self.occupancy.items()},
690 "deliveries": deliveries,
691 "delivery_provenance": self.debug_trace.delivery_provenance,
692 }
693 if self._suppression_reason is not None:
694 result["suppressed"] = str(self._suppression_reason)
695 elif not any("skipped" not in d for d in deliveries.values()):
696 result["fallback"] = [d.name for d in self.context.delivery_registry.fallback_by_default_deliveries]
697 return result
699 async def call_transport(self, delivery: Delivery, target_override: DeliveryTargetOverride | None = None) -> None:
700 try:
701 transport: Transport = delivery.transport
702 skip_reason: SuppressionReason | None = self.delivery_skip_reason(delivery)
703 if skip_reason is not None:
704 self.record_result(delivery, suppression_reason=skip_reason)
705 _LOGGER.debug("SUPERNOTIFY Skipping delivery %s, %s", delivery, skip_reason)
706 return
708 targets: list[Target] = self.generate_targets(delivery, target_override)
709 envelopes: list[Envelope] = self.generate_envelopes(delivery, targets)
710 if not envelopes:
711 if delivery.target_required == TargetRequired.ALWAYS and (
712 not targets or not any(t.has_resolved_target() for t in targets)
713 ):
714 reason: SuppressionReason = SuppressionReason.NO_TARGET
715 else:
716 reason = SuppressionReason.UNKNOWN
717 self.record_result(delivery, targets=targets, suppression_reason=reason)
718 return
720 for envelope in envelopes:
721 if not envelope.force_resend and self.context.dupe_checker.check(envelope):
722 _LOGGER.debug("SUPERNOTIFY Suppressing dupe envelope, %s", self.message)
723 self.record_result(delivery, envelope, suppression_reason=SuppressionReason.DUPE)
724 continue
725 try:
726 if await transport.deliver(envelope, debug_trace=self.debug_trace):
727 self.record_result(delivery, envelope)
728 else:
729 _LOGGER.info("SUPERNOTIFY No delivery for %s (id: %s)", delivery.name, envelope.notification_id)
730 self.record_result(delivery, envelope, suppression_reason=SuppressionReason.UNKNOWN)
731 except Exception as e2:
732 envelope.error_count = envelope.error_count + 1
733 transport.record_error(str(e2), method="deliver")
734 transport.log_delivery_failure(e2, "SUPERNOTIFY Failed to deliver %s", delivery.name)
735 envelope.delivery_error = format_exception(e2)
736 self.record_result(delivery, envelope)
738 except Exception as e:
739 _LOGGER.exception(
740 "SUPERNOTIFY Failed to notify using delivery %s via %s",
741 delivery.name,
742 type(delivery.transport).__name__,
743 )
744 self.delivery_exceptions.setdefault(delivery.name, [])
745 self.delivery_exceptions[delivery.name].append("\n".join(format_exception(e)))
746 self.record_result(delivery, None, suppression_reason=SuppressionReason.ERROR)
748 def record_result(
749 self,
750 delivery: Delivery | None,
751 envelope: Envelope | None = None,
752 targets: list[Target] | None = None,
753 suppression_reason: SuppressionReason | None = None,
754 ) -> None:
755 """Debugging (and unit test) support for notifications that failed or were skipped"""
756 if delivery and envelope:
757 self.delivered += envelope.delivered
758 self.error_count += envelope.error_count
759 self.deliveries.setdefault(delivery.name, {})
760 if envelope.delivered:
761 self.deliveries[delivery.name].setdefault(EnvelopeOutcome.SUCCESS, [])
762 self.deliveries[delivery.name][EnvelopeOutcome.SUCCESS].append(envelope) # type: ignore
763 envelope.record_recipient_notifications(self._recorded_person_ids)
764 else:
765 if suppression_reason:
766 envelope.skip_reason = suppression_reason
767 if suppression_reason not in self._skip_reasons:
768 self._skip_reasons.append(suppression_reason)
769 if suppression_reason == SuppressionReason.DUPE:
770 self.dupe = True
771 if envelope.error_count:
772 self.deliveries[delivery.name].setdefault(EnvelopeOutcome.ERROR, [])
773 self.deliveries[delivery.name][EnvelopeOutcome.ERROR].append(envelope) # type: ignore
774 self.failed += 1
775 else:
776 self.deliveries[delivery.name].setdefault(EnvelopeOutcome.SUPPRESSED, [])
777 self.deliveries[delivery.name][EnvelopeOutcome.SUPPRESSED].append(envelope) # type: ignore
778 self.suppressed += 1
780 if not envelope:
781 delivery_name: str = delivery.name if delivery else "!UNKNOWN!"
782 skip_summary: dict[str, Any] = {
783 "target_required": delivery.target_required if delivery else "!UNKNOWN!",
784 "suppression_reason": str(suppression_reason),
785 }
786 self.deliveries.setdefault(delivery_name, {})
787 if targets:
788 skip_summary["targets"] = targets
789 self.deliveries[delivery_name][EnvelopeOutcome.SKIPPED] = skip_summary
790 self.skipped += 1
791 if suppression_reason == SuppressionReason.ERROR:
792 self.error_count += 1
793 self.failed += 1
795 def contents(self, diagnostics: bool = False, **_kwargs: Any) -> dict[str, Any]:
796 """ArchiveableObject implementation"""
797 minimal = not diagnostics
798 object_refs = ["context", "ha_context", "people_registry", "delivery_registry"]
799 keys_only = ["enabled_scenarios"]
800 debug_only = ["debug_trace"]
801 exposed_if_populated = [
802 "_delivery_error",
803 "message_html",
804 "spoken_message",
805 "extra_data",
806 "actions",
807 "_suppression_reason",
808 "delivery_provenance",
809 "unknown_names",
810 ]
811 # fine tune dict order to ease the eye-burden when reviewing archived notifications
812 preferred_order = [
813 "id",
814 "outcome",
815 "created",
816 "message",
817 "spoken_message",
818 "message_html",
819 "priority",
820 "stats",
821 "delivered",
822 "failed",
823 "suppressed",
824 "skipped",
825 "fallback",
826 "error_count",
827 "dupe",
828 "force_resend",
829 "delivery_selection",
830 "delivery_overrides",
831 "delivery_selection",
832 "selected_deliveries",
833 "delivery_provenance",
834 "applied_scenario_names",
835 "constrain_scenario_names",
836 "required_scenario_names",
837 "enabled_scenarios",
838 "selected_scenario_names",
839 "delivery_exceptions",
840 "uncategorized_targets",
841 "unassigned_targets",
842 "unknown_names",
843 "extra_data",
844 "deliveries",
845 "actions",
846 "delivery_exceptions",
847 "original_context",
848 "version",
849 ]
850 # preferred fields
851 raw: dict[str, Any] = dict(self.__dict__)
852 raw["unassigned_targets"] = self._unassigned_targets()
853 raw["stats"] = self._delivery_stats()
854 raw["version"] = _VERSION
855 raw["outcome"] = self.outcome()
856 raw["delivery_provenance"] = self.debug_trace.delivery_provenance
858 result: dict[str, Any] = {}
859 if self.ha_context is not None:
860 # Context.as_dict() takes no kwargs, so it can't go through sanitize() like everything else
861 original_context = dict(self.ha_context.as_dict())
862 user_id = original_context.get("user_id")
863 if user_id:
864 user_name = self.people_registry.name_for_user_id(user_id)
865 if user_name:
866 original_context["user"] = user_name
867 raw["original_context"] = original_context
868 result.update({
869 k: sanitize(raw[k], minimal=minimal, occupancy_only=True, top_level_keys_only=(minimal and k in keys_only))
870 for k in preferred_order
871 if k in raw and (k not in exposed_if_populated or raw[k])
872 })
874 # all the rest not explicitly excluded
875 result.update({
876 k: sanitize(v, minimal=minimal, occupancy_only=True)
877 for k, v in raw.items()
878 if k not in result
879 and k not in exposed_if_populated
880 and k not in object_refs
881 and not k.startswith("_")
882 and (not minimal or k not in keys_only)
883 and (not minimal or k not in debug_only)
884 })
885 # the exposed only if populated fields not already placed by preferred_order
886 result.update({
887 k: sanitize(raw[k], minimal=minimal, occupancy_only=True)
888 for k in exposed_if_populated
889 if k not in result and raw.get(k)
890 })
891 return result
893 def _delivery_stats(self) -> dict[str, Any]:
894 # delivery_stats: aggregate delivery metrics
895 result: dict[str, Any] = {}
896 try:
897 all_durations: dict[str, float] = {}
898 total_ok = 0
899 total_all = 0
900 for d_name, outcomes in self.deliveries.items():
901 for envelope in outcomes.get(EnvelopeOutcome.SUCCESS, []):
902 dur = sum(c.contents().get("elapsed", 0) for c in getattr(envelope, "calls", [])) * 1000
903 all_durations[d_name] = dur
904 total_ok += 1
905 total_all += 1
906 for _envelope in outcomes.get(EnvelopeOutcome.ERROR, []):
907 all_durations.setdefault(d_name, 0)
908 total_all += 1
909 if outcomes.get(EnvelopeOutcome.SKIPPED):
910 total_all += 1
911 if all_durations:
912 result = {
913 "total_duration_ms": round(sum(all_durations.values()), 1),
914 "slowest_delivery": max(all_durations, key=lambda k: all_durations[k]),
915 "fastest_delivery": min(all_durations, key=lambda k: all_durations[k]),
916 "delivery_success_rate": round(total_ok / total_all, 2) if total_all else 1.0,
917 }
918 except Exception as e:
919 _LOGGER.warning("SUPERNOTIFY delivery_stats computation failed: %s", e)
920 return result
922 def base_filename(self) -> str:
923 """ArchiveableObject implementation"""
924 return f"{self.created.isoformat()[:16].replace(':', '-')}_{self.id}"
926 def delivery_data(self, delivery: Delivery) -> dict[str, Any]:
927 if delivery is None:
928 return {}
929 delivery_override: DeliveryCustomization | None = self.delivery_overrides.get(delivery.name)
930 if delivery_override is None:
931 delivery_override = self.delivery_overrides.get(delivery.transport.name)
932 # a copy: callers (Envelope.__init__) pop message/title out of this, and the same
933 # override can now be shared by more than one delivery on the same transport
934 return dict(delivery_override.data) if delivery_override and delivery_override.data else {}
936 @property
937 def delivered_envelopes(self) -> list[Envelope]:
938 result: list[Envelope] = []
939 for delivery_result in self.deliveries.values():
940 result.extend(cast("list[Envelope]", delivery_result.get(EnvelopeOutcome.SUCCESS, [])))
941 return result
943 @property
944 def undelivered_envelopes(self) -> list[Envelope]:
945 result: list[Envelope] = []
946 for delivery_result in self.deliveries.values():
947 result.extend(cast("list[Envelope]", delivery_result.get(EnvelopeOutcome.SUPPRESSED, [])))
948 result.extend(cast("list[Envelope]", delivery_result.get(EnvelopeOutcome.ERROR, [])))
949 return result
951 async def select_scenarios(self) -> list[str]:
952 return [s.name for s in self.context.scenario_registry.scenarios.values() if s.evaluate(self.condition_variables)]
954 def _filter_resolve_select_targets(
955 self, computed_target: Target, delivery: Delivery, stages: tuple[str, str, str, str]
956 ) -> Target:
957 """Snooze-filter, resolve indirect/scenario targets, then apply delivery target selection."""
958 computed_target = self.context.snoozer.filter_recipients(
959 computed_target, self.priority, delivery, self.media.get(ATTR_MEDIA_CAMERA_ENTITY_ID)
960 )
961 self.debug_trace.record_target(delivery.name, stages[0], computed_target)
962 for indirect_target in self.resolve_indirect_targets(computed_target, delivery):
963 computed_target += indirect_target
964 self.debug_trace.record_target(delivery.name, stages[1], computed_target)
965 computed_target += self.resolve_scenario_targets(delivery)
966 self.debug_trace.record_target(delivery.name, stages[2], computed_target)
967 computed_target = delivery.select_targets(computed_target)
968 self.debug_trace.record_target(delivery.name, stages[3], computed_target)
969 return computed_target
971 def generate_targets(self, delivery: Delivery, target_override: DeliveryTargetOverride | None = None) -> list[Target]:
973 if delivery.target_required == TargetRequired.NEVER:
974 # don't waste time computing targets for deliveries that don't need them
975 return [Target(None, target_data=delivery.data)]
977 computed_target: Target
979 if delivery.target_usage == TARGET_USE_FIXED:
980 if delivery.target:
981 computed_target = delivery.target.safe_copy()
982 self.debug_trace.record_target(delivery.name, "100_delivery_default_fixed", computed_target)
983 else:
984 computed_target = Target(None, target_data=delivery.data)
985 self.debug_trace.record_target(delivery.name, "101_delivery_default_fixed_empty", computed_target)
986 elif target_override is not None and target_override.fixed:
987 # personal deliveries
988 computed_target = Target(target_override.fixed)
989 self.debug_trace.record_target(delivery.name, "102_delivery_default_fixed", computed_target)
991 elif not self._target:
992 # Unless there are explicit targets, include everyone on the people registry
993 computed_target = self.default_person_ids(delivery)
994 self.debug_trace.record_target(delivery.name, "201_no_action_target", computed_target)
995 else:
996 computed_target = self._target.safe_copy()
997 if target_override is not None:
998 if target_override.include:
999 computed_target += Target(target_override.include)
1000 if target_override.exclude:
1001 computed_target -= Target(target_override.exclude)
1002 self.debug_trace.record_target(delivery.name, "202_action_target", computed_target)
1003 if self._apply_delivery_defaults and INCLUSION_DEFAULT in delivery.inclusion:
1004 computed_target += self.default_person_ids(delivery)
1005 self.debug_trace.record_target(delivery.name, "203_action_target_plus_defaults", computed_target)
1007 # 1st round of filtering for snooze and resolving people->direct targets
1008 computed_target = self._filter_resolve_select_targets(
1009 computed_target,
1010 delivery,
1011 ("300_post_snooze", "310_resolve_indirect", "320_resolved_scenario_targets", "330_delivery_selection"),
1012 )
1013 primary_count = len(computed_target)
1015 if delivery.target_usage == TARGET_USE_ON_NO_DELIVERY_TARGETS:
1016 # person_ids survive selection, but aren't targets to deliver to in their own right
1017 if not computed_target.has_resolved_target() and delivery.target:
1018 computed_target += delivery.target
1019 self.debug_trace.record_target(delivery.name, "400_delivery_default_no_delivery_targets", computed_target)
1020 elif delivery.target_usage == TARGET_USE_ON_NO_ACTION_TARGETS:
1021 if not self._target and delivery.target:
1022 computed_target += delivery.target
1023 self.debug_trace.record_target(delivery.name, "401_delivery_default_no_action_targets", computed_target)
1024 elif delivery.target_usage == TARGET_USE_MERGE_ON_DELIVERY_TARGETS:
1025 # merge in the delivery defaults if there's a target defined in action call
1026 if computed_target.has_resolved_target() and delivery.target:
1027 computed_target += delivery.target
1028 self.debug_trace.record_target(delivery.name, "402_delivery_merge_on_delivery_targets", computed_target)
1029 elif delivery.target_usage == TARGET_USE_MERGE_ALWAYS:
1030 # merge in the delivery defaults even if there's not a target defined in action call
1031 if delivery.target:
1032 computed_target += delivery.target
1033 self.debug_trace.record_target(delivery.name, "403_delivery_merge_always_targets", computed_target)
1034 elif delivery.target_usage == TARGET_USE_FIXED:
1035 _LOGGER.debug("SUPERNOTIFY Fixed target on delivery %s", delivery.name)
1036 self.debug_trace.record_target(delivery.name, "404_fixed_target", computed_target)
1037 else:
1038 self.debug_trace.record_target(delivery.name, "405_no_target_usage_match", computed_target)
1039 _LOGGER.debug("SUPERNOTIFY No useful target definition for delivery %s", delivery.name)
1041 if len(computed_target) > primary_count:
1042 _LOGGER.debug(
1043 "SUPERNOTIFY Delivery config added %s targets for %s", len(computed_target) - primary_count, delivery.name
1044 )
1046 # 2nd round of filtering for snooze and resolving people->direct targets after delivery target applied
1047 computed_target = self._filter_resolve_select_targets(
1048 computed_target,
1049 delivery,
1050 ("501_post_snooze", "502_resolve_indirect", "503_resolved_scenario_targets", "504_delivery_selection"),
1051 )
1053 # If the action call explicitly specified a target for this delivery, it takes
1054 # precedence over all resolved/merged targets above.
1055 delivery_override: DeliveryCustomization | None = self.delivery_overrides.get(delivery.name)
1056 if delivery_override is None:
1057 # if override doesn't use a valid delivery name, try a transport name instead
1058 delivery_override = self.delivery_overrides.get(delivery.transport.name)
1059 if delivery_override and delivery_override.target and delivery_override.target.has_targets():
1060 # exclusively scoped to this one delivery, so safe to claim an unqualified value
1061 override_target = delivery.reclassify_unqualified_target(delivery_override.target)
1062 self._record_uncategorized(delivery, override_target)
1063 # handle and resolve indirect targets, like person->mobile device or email
1064 for indirect_target in self.resolve_indirect_targets(override_target, delivery):
1065 override_target += indirect_target
1066 computed_target = delivery.select_targets(override_target)
1067 self.debug_trace.record_target(delivery.name, "600_delivery_override_target", computed_target)
1069 split_targets: list[Target] = computed_target.split_by_target_data()
1070 self.debug_trace.record_target(delivery.name, "610_delivery_split_targets", split_targets)
1072 direct_targets: list[Target] = [t.direct() for t in split_targets]
1073 self.debug_trace.record_target(delivery.name, "620_narrow_to_direct", direct_targets)
1075 if delivery.options.get(OPTION_UNIQUE_TARGETS, False):
1076 direct_targets = [t - self._already_selected for t in direct_targets]
1077 self.debug_trace.record_target(delivery.name, "630_make_unique_across_deliveries", direct_targets)
1078 for direct_target in direct_targets:
1079 self._already_selected += direct_target
1080 # after the uniqueness bookkeeping above, so that only ever deals in what's delivered to
1081 self._attach_person_ids(delivery, computed_target.person_ids, direct_targets)
1082 self.debug_trace.record_target(delivery.name, "999_final_cut", direct_targets)
1083 return direct_targets
1085 def _attach_person_ids(self, delivery: Delivery, person_ids: list[str], envelope_targets: list[Target]) -> None:
1086 """Record on each envelope target which of the notification's recipients it reaches
1088 A recipient is only added to the envelope targets holding one of their own targets
1089 for this delivery, so one with nothing this delivery can send to, or whose addresses
1090 were all already sent in an earlier envelope, isn't counted as notified by this one.
1091 """
1092 for person_id in person_ids:
1093 recipient: Recipient | None = self.people_registry.people.get(person_id)
1094 if recipient is None:
1095 continue
1096 recipient_targets: set[str] = set(recipient.target(delivery.name).resolved_targets())
1097 for envelope_target in envelope_targets:
1098 if recipient_targets.intersection(envelope_target.resolved_targets()):
1099 envelope_target.extend(ATTR_PERSON_ID, person_id)
1101 def resolve_scenario_targets(self, delivery: Delivery) -> Target:
1102 resolved: Target = Target()
1103 for scenario in self.enabled_scenarios.values():
1104 customization: DeliveryCustomization | None = scenario.delivery_customization(delivery.name)
1105 if customization and customization.target and customization.target.has_targets():
1106 # exclusively scoped to this one delivery, so safe to claim an unqualified value
1107 reclassified = delivery.reclassify_unqualified_target(customization.target)
1108 self._record_uncategorized(delivery, reclassified)
1109 resolved += reclassified
1110 return resolved
1112 def _record_uncategorized(self, delivery: Delivery, target: Target) -> None:
1113 """Track a delivery-scoped target's leftover uncategorised values, once
1115 `Delivery.reclassify_unqualified_target()` has had its one chance to place them (it
1116 already logs a warning) - collected here so the caller (`SupernotifyEngine.
1117 async_send_message()`) can raise a single `UncategorizedTargetError` for the whole
1118 notification, once delivery has fully finished and everything deliverable is sent.
1119 """
1120 if target.has_unknown_targets():
1121 self.uncategorized_targets.setdefault(delivery.name, [])
1122 self.uncategorized_targets[delivery.name].extend(target.custom_ids(Target.UNKNOWN_CUSTOM_CATEGORY))
1124 def _unassigned_targets(self) -> dict[str, list[str]]:
1125 """Archive-only diagnostic: requested target values with a recognisable category
1127 (entity_id, email, phone, mobile_app_id, device_id, or an explicit custom category)
1128 that never ended up in any envelope across the whole notification - as opposed to
1129 `uncategorized_targets`, which tracks values with no recognisable shape at all.
1130 Grouped by category, like `uncategorized_targets`, so a gap is easy to place when
1131 debugging. Computed here rather than tracked as an instance attribute, since it's
1132 only ever needed for the archive.
1133 """
1134 if not self._target:
1135 return {}
1136 result: dict[str, list[str]] = {}
1137 for category, targets in self._target.direct().targets.items():
1138 if category == Target.UNKNOWN_CUSTOM_CATEGORY:
1139 continue
1140 unassigned = [t for t in targets if t not in self._already_selected.targets.get(category, [])]
1141 if unassigned:
1142 result[category] = unassigned
1143 return result
1145 def all_recipients(self) -> list[Recipient]:
1146 recipients: list[Recipient] = []
1147 if self._target:
1148 # explicit targets given
1149 recipients.extend(
1150 self.people_registry.people[pers_ent_id]
1151 for pers_ent_id in self._target.person_ids
1152 if pers_ent_id in self.people_registry.people and self.people_registry.people[pers_ent_id].enabled
1153 )
1154 else:
1155 # default to all known recipients
1156 recipients = self.people_registry.enabled_recipients()
1157 return recipients
1159 def default_person_ids(self, delivery: Delivery) -> Target:
1160 # If target not specified on service call or delivery, then default to std list of recipients
1161 people: list[Recipient] = self.people_registry.filter_recipients_by_occupancy(delivery.occupancy)
1162 return Target({
1163 ATTR_PERSON_ID: [p.entity_id for p in people if p.entity_id and delivery.name not in p.disabling_delivery_names()]
1164 })
1166 def resolve_indirect_targets(self, target: Target, delivery: Delivery) -> list[Target]:
1167 # enrich data selected in configuration for this delivery, from direct target definition or attrs like email or phone
1168 resolved: Target = Target()
1169 additional: list[Target] = []
1171 for person_id in target.person_ids:
1172 recipient: Recipient | None = self.people_registry.people.get(person_id)
1173 if recipient and recipient.enabled:
1174 recipient_target = recipient.target(delivery.name)
1175 if recipient_target.target_specific_data:
1176 additional.append(recipient_target)
1177 else:
1178 resolved += recipient_target
1179 else:
1180 _LOGGER.debug("SUPERNOTIFY Skipping recipient %s with enabled switched off", person_id)
1182 return [resolved, *additional]
1184 def generate_envelopes(self, delivery: Delivery, targets: list[Target]) -> list[Envelope]:
1185 # now the list of recipients determined, resolve this to target addresses or entities
1187 envelopes: list[Envelope] = []
1188 for target in targets:
1189 # a target is always generated, even if there are no recipients
1190 if target.has_resolved_target() or delivery.target_required != TargetRequired.ALWAYS:
1191 envelope_data = {}
1193 # least priority - delivery derived data
1194 envelope_data.update(delivery.data)
1195 # next least priority - target derived data
1196 if target.target_data:
1197 envelope_data.update(target.target_data)
1199 # scenario applied at cross-delivery level in apply_enabled_scenarios
1200 for scenario in self.enabled_scenarios.values():
1201 customization: DeliveryCustomization | None = scenario.delivery_customization(delivery.name)
1202 if customization and customization.data:
1203 envelope_data.update(customization.data)
1205 # apply data from action call last to prioritize it
1206 envelope_data.update(self.extra_data) # action call data
1208 envelopes.append(
1209 Envelope(delivery, self, target, envelope_data, context=self.context, ha_context=self.ha_context)
1210 )
1212 return envelopes