Coverage for custom_components/supernotify/model.py: 98%
363 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-08 17:42 +0000
1from __future__ import annotations
3import abc
4import logging
5import re
6from dataclasses import dataclass, field
7from enum import IntFlag, StrEnum, auto
8from traceback import format_exception
9from typing import TYPE_CHECKING, Any
11from homeassistant.const import (
12 CONF_ACTION,
13 CONF_ALIAS,
14 CONF_CONDITIONS,
15 CONF_DEBUG,
16 CONF_ENABLED,
17 CONF_OPTIONS,
18 CONF_TARGET,
19 STATE_HOME,
20 STATE_NOT_HOME,
21)
22from homeassistant.core import (
23 Context as HAContext,
24)
26from .common import canonical_priority, ensure_list
27from .const import (
28 CONF_DATA,
29 CONF_DELIVERY_DEFAULTS,
30 CONF_DEVICE_DISCOVERY,
31 CONF_DEVICE_DOMAIN,
32 CONF_DEVICE_MODEL_EXCLUDE,
33 CONF_DEVICE_MODEL_INCLUDE,
34 CONF_INCLUSION,
35 CONF_MESSAGE,
36 CONF_OCCUPANCY,
37 CONF_PRIORITY,
38 CONF_SELECTION_RANK,
39 CONF_TARGET_REQUIRED,
40 CONF_TARGET_USAGE,
41 CONF_TEMPLATE,
42 CONF_TITLE,
43 INCLUSION_DEFAULT,
44 OCCUPANCY_ALL,
45 PRIORITY_MEDIUM,
46 PRIORITY_VALUES,
47 TARGET_USE_ON_NO_ACTION_TARGETS,
48)
49from .schema import SelectionRank
50from .target import Target
52if TYPE_CHECKING:
53 from collections.abc import Iterable
55 from homeassistant.helpers.typing import ConfigType, TemplateVarsType
57_LOGGER = logging.getLogger(__name__)
60class TransportFeature(IntFlag):
61 MESSAGE = 1
62 TITLE = 2
63 IMAGES = 4
64 VIDEO = 8
65 ACTIONS = 16
66 TEMPLATE_FILE = 32
67 SNAPSHOT_IMAGE = 64 # transports will be deferred if a camera PTZ is defined
68 SPOKEN = 128
69 SOUND = 256 # sirens, chimes, buzzers, all non-spoken audio
72class TransportConfig:
73 def __init__(self, conf: ConfigType | None = None, class_config: TransportConfig | None = None) -> None:
74 # local import: options.py imports SelectionRule from this module, so importing
75 # its constants back at module level here would be circular
76 from .options import (
77 OPTION_DEVICE_DISCOVERY,
78 OPTION_DEVICE_DOMAIN,
79 OPTION_DEVICE_MODEL_SELECT,
80 SELECT_EXCLUDE,
81 SELECT_INCLUDE,
82 )
84 conf = conf or {}
85 if class_config is not None:
86 self.enabled: bool = conf.get(CONF_ENABLED, class_config.enabled)
87 self.alias = conf.get(CONF_ALIAS)
88 self.delivery_defaults: DeliveryConfig = DeliveryConfig(
89 conf.get(CONF_DELIVERY_DEFAULTS, {}), class_config.delivery_defaults or None
90 )
91 else:
92 self.enabled = conf.get(CONF_ENABLED, True)
93 self.alias = conf.get(CONF_ALIAS)
94 self.delivery_defaults = DeliveryConfig(conf.get(CONF_DELIVERY_DEFAULTS) or {})
96 # deprecation support
97 device_domain = conf.get(CONF_DEVICE_DOMAIN)
98 if device_domain is not None:
99 _LOGGER.warning("SUPERNOTIFY device_domain on transport deprecated, use options instead")
100 self.delivery_defaults.options[OPTION_DEVICE_DOMAIN] = device_domain
101 device_model_include = conf.get(CONF_DEVICE_MODEL_INCLUDE)
102 device_model_exclude = conf.get(CONF_DEVICE_MODEL_EXCLUDE)
103 if device_model_include is not None or device_model_exclude is not None:
104 _LOGGER.warning("SUPERNOTIFY device_model_include/exclude on transport deprecated, use options instead")
105 self.delivery_defaults.options[OPTION_DEVICE_MODEL_SELECT] = {
106 SELECT_INCLUDE: device_model_include,
107 SELECT_EXCLUDE: device_model_exclude,
108 }
109 device_discovery = conf.get(CONF_DEVICE_DISCOVERY)
110 if device_discovery is not None and self.delivery_defaults.options.get(OPTION_DEVICE_DISCOVERY) is None:
111 _LOGGER.warning("SUPERNOTIFY device_discovery on transport deprecated, use options instead")
112 self.delivery_defaults.options[OPTION_DEVICE_DISCOVERY] = device_discovery
115class DeliveryCustomization:
116 def __init__(
117 self, config: ConfigType | None = None, target_specific: bool = False, default_enabled: bool | None = None
118 ) -> None:
119 config = config or {}
120 # defining a customization doesn't imply that the delivery is always enabled -
121 # default_enabled only fills in when the `enabled` key is omitted entirely (dict.get
122 # default only applies when the key is absent), never overrides an explicit
123 # `enabled: None`/`false`/`true` - e.g. Scenario.enabling_deliveries() relies on this to
124 # treat a directly-named delivery with no `enabled` key as enabling it (matching the
125 # list/string delivery config forms), but not one explicitly set to `enabled: None`
126 # just to carry other data (e.g. a priority override).
127 self.enabled: bool | None = config.get(CONF_ENABLED, default_enabled)
128 self.data: dict[str, Any] | None = config.get(CONF_DATA)
129 self.target: Target | None
131 if config.get(CONF_TARGET):
132 if self.data:
133 self.target = Target(config.get(CONF_TARGET), target_data=self.data, target_specific_data=target_specific)
134 else:
135 self.target = Target(config.get(CONF_TARGET))
136 else:
137 self.target = None
139 def data_value(self, key: str) -> Any: # ruff: ignore[any-type]
140 return self.data.get(key) if self.data else None
142 def as_dict(self, **_kwargs: Any) -> dict[str, Any]:
143 return {CONF_TARGET: self.target.as_dict() if self.target else None, CONF_ENABLED: self.enabled, CONF_DATA: self.data}
146class SelectionRule:
147 def __init__(self, config: str | list[str] | dict | SelectionRule | None) -> None:
148 # local import: see TransportConfig.__init__ for why this can't be module-level
149 from .options import SELECT_EXCLUDE, SELECT_INCLUDE
151 self.include: list[str] | None = None
152 self.exclude: list[str] | None = None
153 if config is None:
154 return
155 if isinstance(config, SelectionRule):
156 self.include = config.include
157 self.exclude = config.exclude
158 elif isinstance(config, str):
159 self.include = [config]
160 elif isinstance(config, list):
161 self.include = config
162 else:
163 if config.get(SELECT_INCLUDE):
164 self.include = ensure_list(config.get(SELECT_INCLUDE))
165 if config.get(SELECT_EXCLUDE):
166 self.exclude = ensure_list(config.get(SELECT_EXCLUDE))
168 def match(self, v: str | Iterable[str] | None) -> bool:
169 if self.include is None and self.exclude is None:
170 return True
171 if isinstance(v, str) or v is None:
172 if self.exclude is not None and v is not None and any(re.fullmatch(pat, v) for pat in self.exclude):
173 return False
174 if self.include is not None and (v is None or not any(re.fullmatch(pat, v) for pat in self.include)):
175 return False
176 else:
177 if self.exclude is not None:
178 for vv in v:
179 if any(re.fullmatch(pat, vv) for pat in self.exclude):
180 return False
181 if self.include is not None:
182 return any(any(re.fullmatch(pat, vv) for pat in self.include) for vv in v)
183 return True
186class DataFilter:
187 r"""Accepts a dict structure and returns a filtered copy, with arbitrary-depth key filtering.
189 Config format (same structure applies recursively at each level):
190 str | list -- shorthand: include only keys matching these patterns
191 dict:
192 include: list[str] -- include only keys matching these patterns
193 exclude: list[str] -- exclude keys matching these patterns
194 exclude: dict -- exclude tree: null value = exclude that key,
195 dict value = keep key but apply tree recursively to its value
196 <key>: sub-config -- any other key: sub-filter applied to that key's dict value
198 Patterns are matched with re.fullmatch. Sub-filter key lookup is exact (not regex).
199 include and exclude can be combined; any non-reserved key adds a sub-filter.
200 """
202 def __init__(self, config: str | list[str] | dict | None) -> None:
203 self._include: list[str] | None = None
204 self._exclude: list[str] | None = None
205 self._sub: dict[str, DataFilter] = {}
206 if config is None:
207 return
208 if isinstance(config, str):
209 self._include = [config]
210 elif isinstance(config, list):
211 self._include = config
212 else:
213 self._init_from_dict(config)
215 def _init_from_dict(self, config: dict) -> None:
216 # local import: see TransportConfig.__init__ for why this can't be module-level
217 from .options import SELECT_EXCLUDE, SELECT_INCLUDE
219 include_val = config.get(SELECT_INCLUDE)
220 exclude_val = config.get(SELECT_EXCLUDE)
221 if isinstance(include_val, dict):
222 # include as dict: keys = include patterns, non-null values = sub-filters
223 self._include = list(include_val.keys())
224 for k, v in include_val.items():
225 if v is not None:
226 self._sub[k] = DataFilter(v)
227 elif include_val is not None:
228 self._include = ensure_list(include_val)
229 if isinstance(exclude_val, dict):
230 excludes, subs = DataFilter._parse_exclude_tree(exclude_val)
231 self._exclude = excludes or None
232 self._sub.update(subs)
233 elif exclude_val is not None:
234 self._exclude = ensure_list(exclude_val)
235 for k, v in config.items():
236 if k in (SELECT_INCLUDE, SELECT_EXCLUDE) or k in self._sub:
237 continue
238 if isinstance(v, dict) and (SELECT_INCLUDE in v or SELECT_EXCLUDE in v):
239 # value is an explicit DataFilter config (has reserved keys) → sub-filter only, all keys pass
240 self._sub[k] = DataFilter(v)
241 else:
242 # null or value without reserved keys → include pattern (+ sub-filter if non-null)
243 if self._include is None:
244 self._include = []
245 self._include.append(k)
246 if v is not None:
247 self._sub[k] = DataFilter(v)
249 @staticmethod
250 def _parse_exclude_tree(tree: dict) -> tuple[list[str], dict[str, DataFilter]]:
251 excludes: list[str] = []
252 subs: dict[str, DataFilter] = {}
253 for k, v in tree.items():
254 if v is None:
255 excludes.append(k)
256 else:
257 subs[k] = DataFilter._exclude_tree_to_filter(v)
258 return excludes, subs
260 @staticmethod
261 def _exclude_tree_to_filter(tree: dict) -> DataFilter:
262 df = DataFilter(None)
263 excludes, subs = DataFilter._parse_exclude_tree(tree)
264 df._exclude = excludes or None
265 df._sub = subs
266 return df
268 def _match(self, key: str) -> bool:
269 if self._exclude is None and self._include is None:
270 return True
271 if self._exclude is not None and any(re.fullmatch(p, key) for p in self._exclude):
272 return False
273 return self._include is None or any(re.fullmatch(p, key) for p in self._include)
275 def apply(self, data: dict[str, Any], *, prune_empty: bool = False) -> dict[str, Any]:
276 result: dict[str, Any] = {}
277 for key, value in data.items():
278 if not self._match(key):
279 _LOGGER.debug("SUPERNOTIFY Pruning %s:%s", key, value)
280 continue
281 if key in self._sub and isinstance(value, dict):
282 value = self._sub[key].apply(value, prune_empty=prune_empty)
283 if prune_empty and value == {}:
284 _LOGGER.debug("SUPERNOTIFY Pruning empty %s", key)
285 continue
286 result[key] = value
287 return result
290class DeliveryConfig:
291 """Shared config for transport defaults and Delivery definitions"""
293 def __init__(self, conf: ConfigType, delivery_defaults: DeliveryConfig | None = None) -> None:
295 if delivery_defaults is not None:
296 # use transport defaults where no delivery level override
297 self.target: Target | None = Target(conf.get(CONF_TARGET)) if CONF_TARGET in conf else delivery_defaults.target
298 self.target_required: TargetRequired = conf.get(CONF_TARGET_REQUIRED, delivery_defaults.target_required)
299 self.target_usage: str = conf.get(CONF_TARGET_USAGE) or delivery_defaults.target_usage
300 self.action: str | None = conf.get(CONF_ACTION) or delivery_defaults.action
301 self.debug: bool = conf.get(CONF_DEBUG, delivery_defaults.debug)
303 self.data: ConfigType = dict(delivery_defaults.data) if isinstance(delivery_defaults.data, dict) else {}
304 self.data.update(conf.get(CONF_DATA, {}))
305 self.inclusion: list[str] = conf.get(CONF_INCLUSION, delivery_defaults.inclusion)
306 self.priority: list[str] = conf.get(CONF_PRIORITY, delivery_defaults.priority)
307 self.selection_rank: SelectionRank = conf.get(CONF_SELECTION_RANK, delivery_defaults.selection_rank)
308 self.options: ConfigType = conf.get(CONF_OPTIONS, {})
309 # only override options not set in config
310 if isinstance(delivery_defaults.options, dict):
311 for opt in delivery_defaults.options:
312 self.options.setdefault(opt, delivery_defaults.options[opt])
313 self.alias: str | None = conf.get(CONF_ALIAS, delivery_defaults.alias)
314 self.template: str | None = conf.get(CONF_TEMPLATE, delivery_defaults.template)
315 self.message: str | None = conf.get(CONF_MESSAGE, delivery_defaults.message)
316 self.title: str | None = conf.get(CONF_TITLE, delivery_defaults.title)
317 self.occupancy: str = conf.get(CONF_OCCUPANCY, delivery_defaults.occupancy)
318 self.conditions_config: list[ConfigType] | None = conf.get(CONF_CONDITIONS, delivery_defaults.conditions_config)
319 else:
320 # construct the transport defaults
321 self.target = Target(conf.get(CONF_TARGET)) if conf.get(CONF_TARGET) else None
322 self.target_required = conf.get(CONF_TARGET_REQUIRED, TargetRequired.ALWAYS)
323 self.target_usage = conf.get(CONF_TARGET_USAGE, TARGET_USE_ON_NO_ACTION_TARGETS)
324 self.action = conf.get(CONF_ACTION)
325 self.debug = conf.get(CONF_DEBUG, False)
326 self.options = conf.get(CONF_OPTIONS, {})
327 self.data = conf.get(CONF_DATA, {})
328 self.inclusion = conf.get(CONF_INCLUSION, [INCLUSION_DEFAULT])
329 self.priority = [canonical_priority(p) for p in conf.get(CONF_PRIORITY, list(PRIORITY_VALUES.keys()))]
330 self.selection_rank = conf.get(CONF_SELECTION_RANK, SelectionRank.ANY)
331 self.alias = conf.get(CONF_ALIAS)
332 self.template = conf.get(CONF_TEMPLATE)
333 self.message = conf.get(CONF_MESSAGE)
334 self.title = conf.get(CONF_TITLE)
335 self.occupancy = conf.get(CONF_OCCUPANCY, OCCUPANCY_ALL)
336 self.conditions_config = conf.get(CONF_CONDITIONS)
338 def as_dict(self, **_kwargs: Any) -> dict[str, Any]:
339 return {
340 CONF_TARGET: self.target.as_dict() if self.target else None,
341 CONF_ACTION: self.action,
342 CONF_OPTIONS: self.options,
343 CONF_DATA: self.data,
344 CONF_INCLUSION: self.inclusion,
345 CONF_PRIORITY: self.priority,
346 CONF_SELECTION_RANK: str(self.selection_rank),
347 CONF_TARGET_REQUIRED: str(self.target_required),
348 CONF_TARGET_USAGE: self.target_usage,
349 CONF_ALIAS: self.alias,
350 CONF_TEMPLATE: self.template,
351 CONF_MESSAGE: self.message,
352 CONF_TITLE: self.title,
353 CONF_OCCUPANCY: self.occupancy,
354 CONF_CONDITIONS: self.conditions_config,
355 }
357 def __repr__(self) -> str:
358 """Log friendly representation"""
359 return str(self.as_dict())
362@dataclass
363class ConditionVariables:
364 """Variables presented to all condition evaluations
366 Attributes
367 ----------
368 applied_scenarios (list[str]): Scenarios that have been applied
369 required_scenarios (list[str]): Scenarios that must be applied
370 constrain_scenarios (list[str]): Only scenarios in this list, or in explicit apply_scenarios, can be applied
371 notification_priority (str): Priority of the notification
372 notification_message (str): Message of the notification
373 notification_title (str): Title of the notification
374 occupancy (list[str]): List of occupancy scenarios
375 notification_data (dict[str,Any]): Additional data passed on notify action call
376 camera (dict[str,Any]): Camera of the notification, its `entity_id` and anything configured for it in `cameras`
377 notification_actions (list[dict[str,Any]]): Actions passed on notify action call
378 notification_action_groups (list[str]): Action groups passed on notify action call
380 """
382 applied_scenarios: list[str] = field(default_factory=list)
383 required_scenarios: list[str] = field(default_factory=list)
384 constrain_scenarios: list[str] = field(default_factory=list)
385 notification_priority: str = PRIORITY_MEDIUM
386 notification_message: str | None = ""
387 notification_title: str | None = ""
388 occupancy: list[str] = field(default_factory=list)
390 def __init__(
391 self,
392 applied_scenarios: list[str] | None = None,
393 required_scenarios: list[str] | None = None,
394 constrain_scenarios: list[str] | None = None,
395 delivery_priority: str | None = PRIORITY_MEDIUM,
396 occupiers: dict[str, list[Any]] | None = None,
397 message: str | None = None,
398 title: str | None = None,
399 notification_data: dict[str, Any] | None = None,
400 camera: dict[str, Any] | None = None,
401 actions: list[dict[str, Any]] | None = None,
402 action_groups: list[str] | None = None,
403 ) -> None:
404 occupiers = occupiers or {}
405 self.occupancy = []
406 if not occupiers.get(STATE_NOT_HOME) and not occupiers.get(STATE_HOME):
407 self.occupancy.append("UNDEFINED_OCCUPANTS")
408 if not occupiers.get(STATE_NOT_HOME) and occupiers.get(STATE_HOME):
409 self.occupancy.append("ALL_HOME")
410 elif occupiers.get(STATE_NOT_HOME) and not occupiers.get(STATE_HOME):
411 self.occupancy.append("ALL_AWAY")
412 if len(occupiers.get(STATE_HOME, [])) == 1:
413 self.occupancy.extend(["LONE_HOME", "SOME_HOME"])
414 elif len(occupiers.get(STATE_HOME, [])) > 1 and occupiers.get(STATE_NOT_HOME):
415 self.occupancy.extend(["MULTI_HOME", "SOME_HOME"])
416 self.applied_scenarios = applied_scenarios or []
417 self.required_scenarios = required_scenarios or []
418 self.constrain_scenarios = constrain_scenarios or []
419 self.notification_priority = delivery_priority or PRIORITY_MEDIUM
420 self.notification_message = message
421 self.notification_title = title
422 self.notification_data: dict[str, Any] = notification_data or {}
423 self.camera: dict[str, Any] = camera or {}
424 self.notification_actions: list[dict[str, Any]] = list(actions or [])
425 self.notification_action_groups: list[str] = list(action_groups or [])
427 def as_dict(self, **_kwargs: Any) -> TemplateVarsType:
428 return {
429 "applied_scenarios": self.applied_scenarios,
430 "required_scenarios": self.required_scenarios,
431 "constrain_scenarios": self.constrain_scenarios,
432 "notification_message": self.notification_message,
433 "notification_title": self.notification_title,
434 "notification_priority": self.notification_priority,
435 "occupancy": self.occupancy,
436 "notification_data": self.notification_data,
437 "camera": self.camera,
438 "notification_actions": self.notification_actions,
439 "notification_action_groups": self.notification_action_groups,
440 }
443class SuppressionReason(StrEnum):
444 SNOOZED = "SNOOZED"
445 DUPE = "DUPE"
446 NO_SCENARIO = "NO_SCENARIO"
447 NO_ACTION = "NO_ACTION"
448 NO_TARGET = "NO_TARGET"
449 INVALID_ACTION_DATA = "INVALID_ACTION_DATA"
450 TRANSPORT_DISABLED = "TRANSPORT_DISABLED"
451 PRIORITY = "PRIORITY"
452 DELIVERY_CONDITION = "DELIVERY_CONDITION"
453 OCCUPANCY = "OCCUPANCY"
454 UNKNOWN = "UNKNOWN"
455 ERROR = "ERROR"
457 @property
458 def is_rule(self) -> bool:
459 """Whether this reason reflects a configured rule (occupancy, priority, condition,
460 snooze, a disabled transport, an inactive required scenario) rather than an unexpected
461 failure. A rule-based skip is exactly what was configured to happen, so it shouldn't
462 count toward a notification's outcome the way a genuine miss does."""
463 return self in {
464 SuppressionReason.SNOOZED,
465 SuppressionReason.NO_SCENARIO,
466 SuppressionReason.TRANSPORT_DISABLED,
467 SuppressionReason.PRIORITY,
468 SuppressionReason.DELIVERY_CONDITION,
469 SuppressionReason.OCCUPANCY,
470 }
473class TargetRequired(StrEnum):
474 ALWAYS = auto()
475 NEVER = auto()
476 OPTIONAL = auto()
478 @classmethod
479 def _missing_(cls, value: Any) -> TargetRequired | None: # ruff: ignore[any-type]
480 """Backward compatibility for binary values"""
481 if value is True or (isinstance(value, str) and value.lower() in ("true", "on")):
482 return cls.ALWAYS
483 if value is False or (isinstance(value, str) and value.lower() in ("false", "off")):
484 return cls.OPTIONAL
485 return None
488class TargetType(StrEnum):
489 pass
492class GlobalTargetType(TargetType):
493 NONCRITICAL = "NONCRITICAL"
494 EVERYTHING = "EVERYTHING"
497class RecipientType(StrEnum):
498 USER = "USER"
499 EVERYONE = "EVERYONE"
502class QualifiedTargetType(TargetType):
503 TRANSPORT = "TRANSPORT"
504 DELIVERY = "DELIVERY"
505 CAMERA = "CAMERA"
506 PRIORITY = "PRIORITY"
507 MOBILE = "MOBILE"
508 TAG = "TAG"
511class CommandType(StrEnum):
512 SNOOZE = "SNOOZE"
513 SILENCE = "SILENCE"
514 NORMAL = "NORMAL"
517class MessageOnlyPolicy(StrEnum):
518 STANDARD = "STANDARD" # independent title and message, title used as message if no message on a message-only transport
519 MESSAGE_ONLY = "MESSAGE_ONLY" # independent title and message, never falls back to title
520 USE_TITLE = "USE_TITLE" # use title in place of message, no title
521 # use combined title and message as message, no title
522 COMBINE_TITLE = "COMBINE_TITLE"
525class DebugTrace:
526 def __init__(
527 self,
528 message: str | None,
529 title: str | None,
530 data: dict[str, Any] | None,
531 target: dict[str, list[str]] | list[str] | str | None,
532 debug: bool = True,
533 ) -> None:
534 self.debug: bool = debug
535 self.message: str | None = message
536 self.title: str | None = title
537 self.data: dict[str, Any] | None = dict(data) if data else data
538 self.target: dict[str, list[str]] | list[str] | str | None = list(target) if target else target
539 self.resolved: dict[str, dict[str, Any]] = {}
540 self.delivery_selection: dict[str, list[str]] = {}
541 self.delivery_provenance: dict[str, dict[str, list[str]]] = {}
542 self.delivery_artefacts: dict[str, Any] = {}
543 self.delivery_exceptions: dict[str, dict[str, list[list[str]]]] = {}
544 self._last_stage: dict[str, str] = {}
545 self._last_target: dict[str, Any] = {}
547 def contents(self, **_kwargs: Any) -> dict[str, Any]:
548 results: dict[str, Any] = {
549 "arguments": {
550 "message": self.message,
551 "title": self.title,
552 "data": self.data,
553 "target": self.target,
554 },
555 "delivery_selection": self.delivery_selection,
556 "resolved": self.resolved,
557 }
558 if self.delivery_artefacts:
559 results["delivery_artefacts"] = self.delivery_artefacts
560 if self.delivery_exceptions:
561 results["delivery_exceptions"] = self.delivery_exceptions
562 return results
564 def record_target(self, delivery_name: str, stage: str, computed: Target | list[Target]) -> None:
565 """Debug support for recording detailed target resolution in archived notification"""
566 if not self.debug:
567 return
568 self.resolved.setdefault(delivery_name, {})
569 self.resolved[delivery_name].setdefault(stage, {})
570 self._last_target.setdefault(delivery_name, {})
571 self._last_target[delivery_name].setdefault(stage, {})
572 if isinstance(computed, Target):
573 combined = computed
574 else:
575 combined = Target()
576 for target in ensure_list(computed):
577 combined += target
578 new_target: dict[str, Any] = combined.as_dict()
579 result: str | dict[str, Any] = new_target
580 if self._last_stage.get(delivery_name):
581 last_target = self._last_target[delivery_name][self._last_stage[delivery_name]]
582 if last_target is not None and last_target == result:
583 result = "NO_CHANGE"
585 self.resolved[delivery_name][stage] = result
586 self._last_stage[delivery_name] = stage
587 self._last_target[delivery_name][stage] = new_target
589 def record_delivery_selection(self, stage: str, delivery_selection: list[str]) -> None:
590 """Debug support for recording detailed target resolution in archived notification"""
591 if not self.debug:
592 return
593 self.delivery_selection[stage] = delivery_selection
595 def record_delivery_provenance(self, delivery: str, effect: str, source: str) -> None:
596 """Record which source switched a delivery on or off, where `record_delivery_selection`
597 only has the combined list per stage.
599 `effect` is `enabled_by` or `disabled_by`, `source` is `default`, `call`,
600 `scenario:<name>` or `recipient:<name>`.
602 Unlike the rest of the trace this is recorded without `debug`, since it is small - a few
603 names per delivery - and archived with every notification, as its `delivery_provenance`.
604 """
605 sources = self.delivery_provenance.setdefault(delivery, {}).setdefault(effect, [])
606 if source not in sources:
607 sources.append(source)
609 def record_delivery_artefact(self, delivery: str, artefact_name: str, artefact: Any) -> None: # ruff: ignore[any-type]
610 if not self.debug:
611 return
612 self.delivery_artefacts.setdefault(delivery, {})
613 self.delivery_artefacts[delivery][artefact_name] = artefact
615 def record_delivery_exception(self, delivery: str, context: str, exception: Exception) -> None:
616 if not self.debug:
617 return
618 self.delivery_exceptions.setdefault(delivery, {})
619 self.delivery_exceptions[delivery].setdefault(context, [])
620 self.delivery_exceptions[delivery][context].append(format_exception(exception))
623class NotifyEntityPlatform:
624 """No simple base class in NotifyEntity to reuse, plus lack of support for Context passing"""
626 @abc.abstractmethod
627 async def async_send_message(
628 self,
629 message: str,
630 title: str | None = None,
631 target: str | None = None,
632 context: HAContext | None = None,
633 via_notify: str | None = None,
634 ) -> None:
635 """implement message"""