Coverage for custom_components/supernotify/model.py: 98%
354 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 14:29 +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 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 # TODO: only works for scenario or recipient, not action call
130 self.target: Target | None
132 if config.get(CONF_TARGET):
133 if self.data:
134 self.target = Target(config.get(CONF_TARGET), target_data=self.data, target_specific_data=target_specific)
135 else:
136 self.target = Target(config.get(CONF_TARGET))
137 else:
138 self.target = None
140 def data_value(self, key: str) -> Any: # ruff: ignore[any-type]
141 return self.data.get(key) if self.data else None
143 def as_dict(self, **_kwargs: Any) -> dict[str, Any]:
144 return {CONF_TARGET: self.target.as_dict() if self.target else None, CONF_ENABLED: self.enabled, CONF_DATA: self.data}
147class SelectionRule:
148 def __init__(self, config: str | list[str] | dict | SelectionRule | None) -> None:
149 # local import: see TransportConfig.__init__ for why this can't be module-level
150 from .options import SELECT_EXCLUDE, SELECT_INCLUDE
152 self.include: list[str] | None = None
153 self.exclude: list[str] | None = None
154 if config is None:
155 return
156 if isinstance(config, SelectionRule):
157 self.include = config.include
158 self.exclude = config.exclude
159 elif isinstance(config, str):
160 self.include = [config]
161 elif isinstance(config, list):
162 self.include = config
163 else:
164 if config.get(SELECT_INCLUDE):
165 self.include = ensure_list(config.get(SELECT_INCLUDE))
166 if config.get(SELECT_EXCLUDE):
167 self.exclude = ensure_list(config.get(SELECT_EXCLUDE))
169 def match(self, v: str | Iterable[str] | None) -> bool:
170 if self.include is None and self.exclude is None:
171 return True
172 if isinstance(v, str) or v is None:
173 if self.exclude is not None and v is not None and any(re.fullmatch(pat, v) for pat in self.exclude):
174 return False
175 if self.include is not None and (v is None or not any(re.fullmatch(pat, v) for pat in self.include)):
176 return False
177 else:
178 if self.exclude is not None:
179 for vv in v:
180 if any(re.fullmatch(pat, vv) for pat in self.exclude):
181 return False
182 if self.include is not None:
183 return any(any(re.fullmatch(pat, vv) for pat in self.include) for vv in v)
184 return True
187class DataFilter:
188 r"""Accepts a dict structure and returns a filtered copy, with arbitrary-depth key filtering.
190 Config format (same structure applies recursively at each level):
191 str | list -- shorthand: include only keys matching these patterns
192 dict:
193 include: list[str] -- include only keys matching these patterns
194 exclude: list[str] -- exclude keys matching these patterns
195 exclude: dict -- exclude tree: null value = exclude that key,
196 dict value = keep key but apply tree recursively to its value
197 <key>: sub-config -- any other key: sub-filter applied to that key's dict value
199 Patterns are matched with re.fullmatch. Sub-filter key lookup is exact (not regex).
200 include and exclude can be combined; any non-reserved key adds a sub-filter.
201 """
203 def __init__(self, config: str | list[str] | dict | None) -> None:
204 self._include: list[str] | None = None
205 self._exclude: list[str] | None = None
206 self._sub: dict[str, DataFilter] = {}
207 if config is None:
208 return
209 if isinstance(config, str):
210 self._include = [config]
211 elif isinstance(config, list):
212 self._include = config
213 else:
214 self._init_from_dict(config)
216 def _init_from_dict(self, config: dict) -> None:
217 # local import: see TransportConfig.__init__ for why this can't be module-level
218 from .options import SELECT_EXCLUDE, SELECT_INCLUDE
220 include_val = config.get(SELECT_INCLUDE)
221 exclude_val = config.get(SELECT_EXCLUDE)
222 if isinstance(include_val, dict):
223 # include as dict: keys = include patterns, non-null values = sub-filters
224 self._include = list(include_val.keys())
225 for k, v in include_val.items():
226 if v is not None:
227 self._sub[k] = DataFilter(v)
228 elif include_val is not None:
229 self._include = ensure_list(include_val)
230 if isinstance(exclude_val, dict):
231 excludes, subs = DataFilter._parse_exclude_tree(exclude_val)
232 self._exclude = excludes or None
233 self._sub.update(subs)
234 elif exclude_val is not None:
235 self._exclude = ensure_list(exclude_val)
236 for k, v in config.items():
237 if k in (SELECT_INCLUDE, SELECT_EXCLUDE) or k in self._sub:
238 continue
239 if isinstance(v, dict) and (SELECT_INCLUDE in v or SELECT_EXCLUDE in v):
240 # value is an explicit DataFilter config (has reserved keys) → sub-filter only, all keys pass
241 self._sub[k] = DataFilter(v)
242 else:
243 # null or value without reserved keys → include pattern (+ sub-filter if non-null)
244 if self._include is None:
245 self._include = []
246 self._include.append(k)
247 if v is not None:
248 self._sub[k] = DataFilter(v)
250 @staticmethod
251 def _parse_exclude_tree(tree: dict) -> tuple[list[str], dict[str, DataFilter]]:
252 excludes: list[str] = []
253 subs: dict[str, DataFilter] = {}
254 for k, v in tree.items():
255 if v is None:
256 excludes.append(k)
257 else:
258 subs[k] = DataFilter._exclude_tree_to_filter(v)
259 return excludes, subs
261 @staticmethod
262 def _exclude_tree_to_filter(tree: dict) -> DataFilter:
263 df = DataFilter(None)
264 excludes, subs = DataFilter._parse_exclude_tree(tree)
265 df._exclude = excludes or None
266 df._sub = subs
267 return df
269 def _match(self, key: str) -> bool:
270 if self._exclude is None and self._include is None:
271 return True
272 if self._exclude is not None and any(re.fullmatch(p, key) for p in self._exclude):
273 return False
274 return self._include is None or any(re.fullmatch(p, key) for p in self._include)
276 def apply(self, data: dict[str, Any], *, prune_empty: bool = False) -> dict[str, Any]:
277 result: dict[str, Any] = {}
278 for key, value in data.items():
279 if not self._match(key):
280 _LOGGER.debug("SUPERNOTIFY Pruning %s:%s", key, value)
281 continue
282 if key in self._sub and isinstance(value, dict):
283 value = self._sub[key].apply(value, prune_empty=prune_empty)
284 if prune_empty and value == {}:
285 _LOGGER.debug("SUPERNOTIFY Pruning empty %s", key)
286 continue
287 result[key] = value
288 return result
291class DeliveryConfig:
292 """Shared config for transport defaults and Delivery definitions"""
294 def __init__(self, conf: ConfigType, delivery_defaults: DeliveryConfig | None = None) -> None:
296 if delivery_defaults is not None:
297 # use transport defaults where no delivery level override
298 self.target: Target | None = Target(conf.get(CONF_TARGET)) if CONF_TARGET in conf else delivery_defaults.target
299 self.target_required: TargetRequired = conf.get(CONF_TARGET_REQUIRED, delivery_defaults.target_required)
300 self.target_usage: str = conf.get(CONF_TARGET_USAGE) or delivery_defaults.target_usage
301 self.action: str | None = conf.get(CONF_ACTION) or delivery_defaults.action
302 self.debug: bool = conf.get(CONF_DEBUG, delivery_defaults.debug)
304 self.data: ConfigType = dict(delivery_defaults.data) if isinstance(delivery_defaults.data, dict) else {}
305 self.data.update(conf.get(CONF_DATA, {}))
306 self.inclusion: list[str] = conf.get(CONF_INCLUSION, delivery_defaults.inclusion)
307 self.priority: list[str] = conf.get(CONF_PRIORITY, delivery_defaults.priority)
308 self.selection_rank: SelectionRank = conf.get(CONF_SELECTION_RANK, delivery_defaults.selection_rank)
309 self.options: ConfigType = conf.get(CONF_OPTIONS, {})
310 # only override options not set in config
311 if isinstance(delivery_defaults.options, dict):
312 for opt in delivery_defaults.options:
313 self.options.setdefault(opt, delivery_defaults.options[opt])
314 self.alias: str | None = conf.get(CONF_ALIAS, delivery_defaults.alias)
315 self.template: str | None = conf.get(CONF_TEMPLATE, delivery_defaults.template)
316 self.message: str | None = conf.get(CONF_MESSAGE, delivery_defaults.message)
317 self.title: str | None = conf.get(CONF_TITLE, delivery_defaults.title)
318 self.occupancy: str = conf.get(CONF_OCCUPANCY, delivery_defaults.occupancy)
319 self.conditions_config: list[ConfigType] | None = conf.get(CONF_CONDITIONS, delivery_defaults.conditions_config)
320 else:
321 # construct the transport defaults
322 self.target = Target(conf.get(CONF_TARGET)) if conf.get(CONF_TARGET) else None
323 self.target_required = conf.get(CONF_TARGET_REQUIRED, TargetRequired.ALWAYS)
324 self.target_usage = conf.get(CONF_TARGET_USAGE, TARGET_USE_ON_NO_ACTION_TARGETS)
325 self.action = conf.get(CONF_ACTION)
326 self.debug = conf.get(CONF_DEBUG, False)
327 self.options = conf.get(CONF_OPTIONS, {})
328 self.data = conf.get(CONF_DATA, {})
329 self.inclusion = conf.get(CONF_INCLUSION, [INCLUSION_DEFAULT])
330 self.priority = conf.get(CONF_PRIORITY, list(PRIORITY_VALUES.keys()))
331 self.selection_rank = conf.get(CONF_SELECTION_RANK, SelectionRank.ANY)
332 self.alias = conf.get(CONF_ALIAS)
333 self.template = conf.get(CONF_TEMPLATE)
334 self.message = conf.get(CONF_MESSAGE)
335 self.title = conf.get(CONF_TITLE)
336 self.occupancy = conf.get(CONF_OCCUPANCY, OCCUPANCY_ALL)
337 self.conditions_config = conf.get(CONF_CONDITIONS)
339 def as_dict(self, **_kwargs: Any) -> dict[str, Any]:
340 return {
341 CONF_TARGET: self.target.as_dict() if self.target else None,
342 CONF_ACTION: self.action,
343 CONF_OPTIONS: self.options,
344 CONF_DATA: self.data,
345 CONF_INCLUSION: self.inclusion,
346 CONF_PRIORITY: self.priority,
347 CONF_SELECTION_RANK: str(self.selection_rank),
348 CONF_TARGET_REQUIRED: str(self.target_required),
349 CONF_TARGET_USAGE: self.target_usage,
350 CONF_ALIAS: self.alias,
351 CONF_TEMPLATE: self.template,
352 CONF_MESSAGE: self.message,
353 CONF_TITLE: self.title,
354 CONF_OCCUPANCY: self.occupancy,
355 CONF_CONDITIONS: self.conditions_config,
356 }
358 def __repr__(self) -> str:
359 """Log friendly representation"""
360 return str(self.as_dict())
363@dataclass
364class ConditionVariables:
365 """Variables presented to all condition evaluations
367 Attributes
368 ----------
369 applied_scenarios (list[str]): Scenarios that have been applied
370 required_scenarios (list[str]): Scenarios that must be applied
371 constrain_scenarios (list[str]): Only scenarios in this list, or in explicit apply_scenarios, can be applied
372 notification_priority (str): Priority of the notification
373 notification_message (str): Message of the notification
374 notification_title (str): Title of the notification
375 occupancy (list[str]): List of occupancy scenarios
376 notification_data (dict[str,Any]): Additional data passed on notify action call
378 """
380 applied_scenarios: list[str] = field(default_factory=list)
381 required_scenarios: list[str] = field(default_factory=list)
382 constrain_scenarios: list[str] = field(default_factory=list)
383 notification_priority: str = PRIORITY_MEDIUM
384 notification_message: str | None = ""
385 notification_title: str | None = ""
386 occupancy: list[str] = field(default_factory=list)
388 def __init__(
389 self,
390 applied_scenarios: list[str] | None = None,
391 required_scenarios: list[str] | None = None,
392 constrain_scenarios: list[str] | None = None,
393 delivery_priority: str | None = PRIORITY_MEDIUM,
394 occupiers: dict[str, list[Any]] | None = None,
395 message: str | None = None,
396 title: str | None = None,
397 notification_data: dict[str, Any] | None = None,
398 ) -> None:
399 occupiers = occupiers or {}
400 self.occupancy = []
401 if not occupiers.get(STATE_NOT_HOME) and not occupiers.get(STATE_HOME):
402 self.occupancy.append("UNDEFINED_OCCUPANTS")
403 if not occupiers.get(STATE_NOT_HOME) and occupiers.get(STATE_HOME):
404 self.occupancy.append("ALL_HOME")
405 elif occupiers.get(STATE_NOT_HOME) and not occupiers.get(STATE_HOME):
406 self.occupancy.append("ALL_AWAY")
407 if len(occupiers.get(STATE_HOME, [])) == 1:
408 self.occupancy.extend(["LONE_HOME", "SOME_HOME"])
409 elif len(occupiers.get(STATE_HOME, [])) > 1 and occupiers.get(STATE_NOT_HOME):
410 self.occupancy.extend(["MULTI_HOME", "SOME_HOME"])
411 self.applied_scenarios = applied_scenarios or []
412 self.required_scenarios = required_scenarios or []
413 self.constrain_scenarios = constrain_scenarios or []
414 self.notification_priority = delivery_priority or PRIORITY_MEDIUM
415 self.notification_message = message
416 self.notification_title = title
417 self.notification_data: dict[str, Any] = notification_data or {}
419 def as_dict(self, **_kwargs: Any) -> TemplateVarsType:
420 return {
421 "applied_scenarios": self.applied_scenarios,
422 "required_scenarios": self.required_scenarios,
423 "constrain_scenarios": self.constrain_scenarios,
424 "notification_message": self.notification_message,
425 "notification_title": self.notification_title,
426 "notification_priority": self.notification_priority,
427 "occupancy": self.occupancy,
428 "notification_data": self.notification_data,
429 }
432class SuppressionReason(StrEnum):
433 SNOOZED = "SNOOZED"
434 DUPE = "DUPE"
435 NO_SCENARIO = "NO_SCENARIO"
436 NO_ACTION = "NO_ACTION"
437 NO_TARGET = "NO_TARGET"
438 INVALID_ACTION_DATA = "INVALID_ACTION_DATA"
439 TRANSPORT_DISABLED = "TRANSPORT_DISABLED"
440 PRIORITY = "PRIORITY"
441 DELIVERY_CONDITION = "DELIVERY_CONDITION"
442 UNKNOWN = "UNKNOWN"
443 ERROR = "ERROR"
446class TargetRequired(StrEnum):
447 ALWAYS = auto()
448 NEVER = auto()
449 OPTIONAL = auto()
451 @classmethod
452 def _missing_(cls, value: Any) -> TargetRequired | None: # ruff: ignore[any-type]
453 """Backward compatibility for binary values"""
454 if value is True or (isinstance(value, str) and value.lower() in ("true", "on")):
455 return cls.ALWAYS
456 if value is False or (isinstance(value, str) and value.lower() in ("false", "off")):
457 return cls.OPTIONAL
458 return None
461class TargetType(StrEnum):
462 pass
465class GlobalTargetType(TargetType):
466 NONCRITICAL = "NONCRITICAL"
467 EVERYTHING = "EVERYTHING"
470class RecipientType(StrEnum):
471 USER = "USER"
472 EVERYONE = "EVERYONE"
475class QualifiedTargetType(TargetType):
476 TRANSPORT = "TRANSPORT"
477 DELIVERY = "DELIVERY"
478 CAMERA = "CAMERA"
479 PRIORITY = "PRIORITY"
480 MOBILE = "MOBILE"
483class CommandType(StrEnum):
484 SNOOZE = "SNOOZE"
485 SILENCE = "SILENCE"
486 NORMAL = "NORMAL"
489class MessageOnlyPolicy(StrEnum):
490 STANDARD = "STANDARD" # independent title and message
491 USE_TITLE = "USE_TITLE" # use title in place of message, no title
492 # use combined title and message as message, no title
493 COMBINE_TITLE = "COMBINE_TITLE"
496class DebugTrace:
497 def __init__(
498 self,
499 message: str | None,
500 title: str | None,
501 data: dict[str, Any] | None,
502 target: dict[str, list[str]] | list[str] | str | None,
503 debug: bool = True,
504 ) -> None:
505 self.debug: bool = debug
506 self.message: str | None = message
507 self.title: str | None = title
508 self.data: dict[str, Any] | None = dict(data) if data else data
509 self.target: dict[str, list[str]] | list[str] | str | None = list(target) if target else target
510 self.resolved: dict[str, dict[str, Any]] = {}
511 self.delivery_selection: dict[str, list[str]] = {}
512 self.delivery_provenance: dict[str, dict[str, list[str]]] = {}
513 self.delivery_artefacts: dict[str, Any] = {}
514 self.delivery_exceptions: dict[str, dict[str, list[list[str]]]] = {}
515 self._last_stage: dict[str, str] = {}
516 self._last_target: dict[str, Any] = {}
518 def contents(self, **_kwargs: Any) -> dict[str, Any]:
519 results: dict[str, Any] = {
520 "arguments": {
521 "message": self.message,
522 "title": self.title,
523 "data": self.data,
524 "target": self.target,
525 },
526 "delivery_selection": self.delivery_selection,
527 "resolved": self.resolved,
528 }
529 if self.delivery_artefacts:
530 results["delivery_artefacts"] = self.delivery_artefacts
531 if self.delivery_exceptions:
532 results["delivery_exceptions"] = self.delivery_exceptions
533 return results
535 def record_target(self, delivery_name: str, stage: str, computed: Target | list[Target]) -> None:
536 """Debug support for recording detailed target resolution in archived notification"""
537 if not self.debug:
538 return
539 self.resolved.setdefault(delivery_name, {})
540 self.resolved[delivery_name].setdefault(stage, {})
541 self._last_target.setdefault(delivery_name, {})
542 self._last_target[delivery_name].setdefault(stage, {})
543 if isinstance(computed, Target):
544 combined = computed
545 else:
546 combined = Target()
547 for target in ensure_list(computed):
548 combined += target
549 new_target: dict[str, Any] = combined.as_dict()
550 result: str | dict[str, Any] = new_target
551 if self._last_stage.get(delivery_name):
552 last_target = self._last_target[delivery_name][self._last_stage[delivery_name]]
553 if last_target is not None and last_target == result:
554 result = "NO_CHANGE"
556 self.resolved[delivery_name][stage] = result
557 self._last_stage[delivery_name] = stage
558 self._last_target[delivery_name][stage] = new_target
560 def record_delivery_selection(self, stage: str, delivery_selection: list[str]) -> None:
561 """Debug support for recording detailed target resolution in archived notification"""
562 if not self.debug:
563 return
564 self.delivery_selection[stage] = delivery_selection
566 def record_delivery_provenance(self, delivery: str, effect: str, source: str) -> None:
567 """Record which source switched a delivery on or off, where `record_delivery_selection`
568 only has the combined list per stage.
570 `effect` is `enabled_by` or `disabled_by`, `source` is `default`, `call`,
571 `scenario:<name>` or `recipient:<name>`.
573 Unlike the rest of the trace this is recorded without `debug`, since it is small - a few
574 names per delivery - and archived with every notification, as its `delivery_provenance`.
575 """
576 sources = self.delivery_provenance.setdefault(delivery, {}).setdefault(effect, [])
577 if source not in sources:
578 sources.append(source)
580 def record_delivery_artefact(self, delivery: str, artefact_name: str, artefact: Any) -> None: # ruff: ignore[any-type]
581 if not self.debug:
582 return
583 self.delivery_artefacts.setdefault(delivery, {})
584 self.delivery_artefacts[delivery][artefact_name] = artefact
586 def record_delivery_exception(self, delivery: str, context: str, exception: Exception) -> None:
587 if not self.debug:
588 return
589 self.delivery_exceptions.setdefault(delivery, {})
590 self.delivery_exceptions[delivery].setdefault(context, [])
591 self.delivery_exceptions[delivery][context].append(format_exception(exception))
594class NotifyEntityPlatform:
595 """No simple base class in NotifyEntity to reuse, plus lack of support for Context passing"""
597 @abc.abstractmethod
598 async def async_send_message(
599 self, message: str, title: str | None = None, target: str | None = None, context: HAContext | None = None
600 ) -> None:
601 """implement message"""