Coverage for custom_components/supernotify/model.py: 97%
473 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-04-01 15:06 +0000
« prev ^ index » next coverage.py v7.10.6, created at 2026-04-01 15:06 +0000
1import logging
2import re
3from dataclasses import dataclass, field
4from enum import IntFlag, StrEnum, auto
5from traceback import format_exception
6from typing import TYPE_CHECKING, Any, ClassVar
8import voluptuous as vol
9from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
11# This import brings in a bunch of other dependency noises, make it manual until py3.14/lazy import/HA updated
12# from homeassistant.components.mobile_app import DOMAIN as MOBILE_APP_DOMAIN
13from homeassistant.const import (
14 ATTR_AREA_ID,
15 ATTR_DEVICE_ID,
16 ATTR_ENTITY_ID,
17 ATTR_FLOOR_ID,
18 ATTR_LABEL_ID,
19 CONF_ACTION,
20 CONF_ALIAS,
21 CONF_DEBUG,
22 CONF_ENABLED,
23 CONF_OPTIONS,
24 CONF_TARGET,
25 STATE_HOME,
26 STATE_NOT_HOME,
27)
28from homeassistant.core import valid_entity_id
30from .common import ensure_list
31from .const import (
32 ATTR_EMAIL,
33 ATTR_MOBILE_APP_ID,
34 ATTR_PERSON_ID,
35 ATTR_PHONE,
36 CONF_DATA,
37 CONF_DELIVERY_DEFAULTS,
38 CONF_DEVICE_DISCOVERY,
39 CONF_DEVICE_DOMAIN,
40 CONF_DEVICE_MODEL_EXCLUDE,
41 CONF_DEVICE_MODEL_INCLUDE,
42 CONF_PRIORITY,
43 CONF_SELECTION,
44 CONF_SELECTION_RANK,
45 CONF_TARGET_REQUIRED,
46 CONF_TARGET_USAGE,
47 OPTION_DEVICE_DISCOVERY,
48 OPTION_DEVICE_DOMAIN,
49 OPTION_DEVICE_MODEL_SELECT,
50 PRIORITY_MEDIUM,
51 PRIORITY_VALUES,
52 RE_DEVICE_ID,
53 SELECT_EXCLUDE,
54 SELECT_INCLUDE,
55 SELECTION_DEFAULT,
56 TARGET_USE_ON_NO_ACTION_TARGETS,
57)
58from .schema import SelectionRank, phone
60if TYPE_CHECKING:
61 from collections.abc import Iterable, Sequence
63 from homeassistant.helpers.typing import ConfigType, TemplateVarsType
65_LOGGER = logging.getLogger(__name__)
67# See note on import of homeassistant.components.mobile_app
68MOBILE_APP_DOMAIN = "mobile_app"
71class TransportFeature(IntFlag):
72 MESSAGE = 1
73 TITLE = 2
74 IMAGES = 4
75 VIDEO = 8
76 ACTIONS = 16
77 TEMPLATE_FILE = 32
78 SNAPSHOT_IMAGE = 64
79 SPOKEN = 128
82class Target:
83 # actual targets, that can positively identified with a validator
84 DIRECT_CATEGORIES: ClassVar[list[str]] = [ATTR_ENTITY_ID, ATTR_DEVICE_ID, ATTR_EMAIL, ATTR_PHONE, ATTR_MOBILE_APP_ID]
85 # references that lead to targets, that can positively identified with a validator
86 AUTO_INDIRECT_CATEGORIES: ClassVar[list[str]] = [ATTR_PERSON_ID]
87 # references that lead to targets, that can't be positively identified with a validator
88 EXPLICIT_INDIRECT_CATEGORIES: ClassVar[list[str]] = [ATTR_AREA_ID, ATTR_FLOOR_ID, ATTR_LABEL_ID]
89 INDIRECT_CATEGORIES = EXPLICIT_INDIRECT_CATEGORIES + AUTO_INDIRECT_CATEGORIES
90 AUTO_CATEGORIES = DIRECT_CATEGORIES + AUTO_INDIRECT_CATEGORIES
92 CATEGORIES = DIRECT_CATEGORIES + INDIRECT_CATEGORIES
94 UNKNOWN_CUSTOM_CATEGORY = "_UNKNOWN_"
96 def __init__(
97 self,
98 target: str
99 | list[str]
100 | dict[str, str]
101 | dict[str, Sequence[str]]
102 | dict[str, list[str]]
103 | dict[str, str | list[str]]
104 | None = None,
105 target_data: dict[str, Any] | None = None,
106 target_specific_data: bool = False,
107 ) -> None:
108 self.target_data: dict[str, Any] | None = None
109 self.target_specific_data: dict[tuple[str, str], dict[str, Any]] | None = None
110 self.targets: dict[str, list[str]] = {}
112 matched: list[str]
114 if isinstance(target, str):
115 target = [target]
117 if target is None:
118 pass # empty constructor is valid case for target building
119 elif isinstance(target, list):
120 # simplified and legacy way of assuming list of entities that can be discriminated by validator
121 targets_left = list(target)
122 for category in self.AUTO_CATEGORIES:
123 validator = getattr(self, f"is_{category}", None)
124 if validator is not None:
125 matched = []
126 for t in targets_left:
127 if t not in matched and validator(t):
128 self.targets.setdefault(category, [])
129 self.targets[category].append(t)
130 matched.append(t)
131 targets_left = [t for t in targets_left if t not in matched]
132 else:
133 _LOGGER.debug("SUPERNOTIFY Missing validator for selective target category %s", category)
134 if not targets_left:
135 break
136 if targets_left:
137 self.targets[self.UNKNOWN_CUSTOM_CATEGORY] = targets_left
139 elif isinstance(target, dict):
140 for category in target:
141 targets = ensure_list(target[category])
142 if not targets:
143 continue
144 if category in self.AUTO_CATEGORIES:
145 validator = getattr(self, f"is_{category}", None)
146 if validator is not None:
147 for t in targets:
148 if validator(t):
149 self.targets.setdefault(category, [])
150 if t not in self.targets[category]:
151 self.targets[category].append(t)
152 else:
153 _LOGGER.warning("SUPERNOTIFY Target skipped invalid %s target: %s", category, t)
154 else:
155 _LOGGER.debug("SUPERNOTIFY Missing validator for selective target category %s", category)
157 elif category in self.CATEGORIES:
158 # categories that can't be automatically detected, like label_id
159 self.targets[category] = targets
160 else:
161 # custom categories
162 self.targets[category] = targets
163 else:
164 _LOGGER.warning("SUPERNOTIFY Target created with no valid targets: %s", target)
166 if target_data and target_specific_data:
167 self.target_specific_data = {}
168 for category, targets in self.targets.items():
169 for t in targets:
170 self.target_specific_data[category, t] = target_data
171 if target_data and not target_specific_data:
172 self.target_data = target_data
174 # Targets by category
176 @property
177 def email(self) -> list[str]:
178 return self.targets.get(ATTR_EMAIL, [])
180 @property
181 def entity_ids(self) -> list[str]:
182 return self.targets.get(ATTR_ENTITY_ID, [])
184 @property
185 def person_ids(self) -> list[str]:
186 return self.targets.get(ATTR_PERSON_ID, [])
188 @property
189 def device_ids(self) -> list[str]:
190 return self.targets.get(ATTR_DEVICE_ID, [])
192 @property
193 def phone(self) -> list[str]:
194 return self.targets.get(ATTR_PHONE, [])
196 @property
197 def mobile_app_ids(self) -> list[str]:
198 return self.targets.get(ATTR_MOBILE_APP_ID, [])
200 def domain_entity_ids(self, domain: str | None) -> list[str]:
201 return [t for t in self.targets.get(ATTR_ENTITY_ID, []) if domain is not None and t and t.startswith(f"{domain}.")]
203 def custom_ids(self, category: str) -> list[str]:
204 return self.targets.get(category, []) if category not in self.CATEGORIES else []
206 @property
207 def area_ids(self) -> list[str]:
208 return self.targets.get(ATTR_AREA_ID, [])
210 @property
211 def floor_ids(self) -> list[str]:
212 return self.targets.get(ATTR_FLOOR_ID, [])
214 @property
215 def label_ids(self) -> list[str]:
216 return self.targets.get(ATTR_LABEL_ID, [])
218 # Selectors / validators
220 @classmethod
221 def is_device_id(cls, target: str) -> bool:
222 return re.fullmatch(RE_DEVICE_ID, target) is not None
224 @classmethod
225 def is_entity_id(cls, target: str) -> bool:
226 return valid_entity_id(target) and not target.startswith("person.")
228 @classmethod
229 def is_person_id(cls, target: str) -> bool:
230 return target.startswith("person.") and valid_entity_id(target)
232 @classmethod
233 def is_phone(cls, target: str) -> bool:
234 try:
235 return phone(target) is not None
236 except vol.Invalid:
237 return False
239 @classmethod
240 def is_mobile_app_id(cls, target: str) -> bool:
241 return not valid_entity_id(target) and target.startswith(f"{MOBILE_APP_DOMAIN}_")
243 @classmethod
244 def is_notify_entity(cls, target: str) -> bool:
245 return valid_entity_id(target) and target.startswith(f"{NOTIFY_DOMAIN}.")
247 @classmethod
248 def is_email(cls, target: str) -> bool:
249 try:
250 return vol.Email()(target) is not None # type: ignore[call-arg]
251 except vol.Invalid:
252 return False
254 def has_targets(self) -> bool:
255 return any(targets for targets in self.targets.values())
257 def has_resolved_target(self) -> bool:
258 return any(targets for category, targets in self.targets.items() if category not in self.INDIRECT_CATEGORIES)
260 def has_unknown_targets(self) -> bool:
261 return len(self.targets.get(self.UNKNOWN_CUSTOM_CATEGORY, [])) > 0
263 def for_category(self, category: str) -> list[str]:
264 return self.targets.get(category, [])
266 def resolved_targets(self) -> list[str]:
267 result: list[str] = []
268 for category, targets in self.targets.items():
269 if category not in self.INDIRECT_CATEGORIES:
270 result.extend(targets)
271 return result
273 def hash_resolved(self) -> int:
274 targets = []
275 for category in self.targets:
276 if category not in self.INDIRECT_CATEGORIES:
277 targets.extend(self.targets[category])
278 return hash(tuple(targets))
280 @property
281 def direct_categories(self) -> list[str]:
282 return self.DIRECT_CATEGORIES + [cat for cat in self.targets if cat not in self.CATEGORIES]
284 def direct(self) -> Target:
285 t = Target(
286 {cat: targets for cat, targets in self.targets.items() if cat in self.direct_categories},
287 target_data=self.target_data,
288 )
289 if self.target_specific_data:
290 t.target_specific_data = {k: v for k, v in self.target_specific_data.items() if k[0] in self.direct_categories}
291 return t
293 def extend(self, category: str, targets: list[str] | str) -> None:
294 targets = ensure_list(targets)
295 self.targets.setdefault(category, [])
296 self.targets[category].extend(t for t in targets if t not in self.targets[category])
298 def remove(self, category: str, targets: list[str] | str) -> None:
299 targets = ensure_list(targets)
300 if category in self.targets:
301 self.targets[category] = [t for t in self.targets[category] if t not in targets]
303 def safe_copy(self) -> Target:
304 t = Target(dict(self.targets), target_data=dict(self.target_data) if self.target_data else None)
305 t.target_specific_data = dict(self.target_specific_data) if self.target_specific_data else None
306 return t
308 def split_by_target_data(self) -> list[Target]:
309 if not self.target_specific_data:
310 result = self.safe_copy()
311 result.target_specific_data = None
312 return [result]
313 results: list[Target] = []
314 default: Target = self.safe_copy()
315 default.target_specific_data = None
316 last_found: dict[str, Any] | None = None
317 collected: dict[str, list[str]] = {}
318 for (category, target), data in self.target_specific_data.items():
319 if last_found is None:
320 last_found = data
321 collected = {category: [target]}
322 elif data != last_found and last_found is not None:
323 new_target: Target = Target(collected, target_data=last_found)
324 results.append(new_target)
325 default -= new_target
326 last_found = data
327 collected = {category: [target]}
328 else:
329 collected.setdefault(category, [])
330 collected[category].append(target)
331 new_target = Target(collected, target_data=last_found)
332 results.append(new_target)
333 default -= new_target
334 if default.has_targets():
335 results.append(default)
336 return results
338 def __len__(self) -> int:
339 """How many targets, whether direct or indirect"""
340 return sum(len(targets) for targets in self.targets.values())
342 def __add__(self, other: Target) -> Target:
343 """Create a new target by adding another to this one"""
344 new = Target()
345 categories = set(list(self.targets.keys()) + list(other.targets.keys()))
346 for category in categories:
347 new.targets[category] = list(self.targets.get(category, []))
348 new.targets[category].extend(t for t in other.targets.get(category, []) if t not in new.targets[category])
350 new.target_data = dict(self.target_data) if self.target_data else None
351 if other.target_data:
352 if new.target_data is None:
353 new.target_data = dict(other.target_data)
354 else:
355 new.target_data.update(other.target_data)
356 new.target_specific_data = dict(self.target_specific_data) if self.target_specific_data else None
357 if other.target_specific_data:
358 if new.target_specific_data is None:
359 new.target_specific_data = dict(other.target_specific_data)
360 else:
361 new.target_specific_data.update(other.target_specific_data)
362 return new
364 def __sub__(self, other: Target) -> Target:
365 """Create a new target by removing another from this one, ignoring target_data"""
366 new = Target()
367 new.target_data = self.target_data
368 if self.target_specific_data:
369 new.target_specific_data = {
370 k: v for k, v in self.target_specific_data.items() if k[1] not in other.targets.get(k[0], ())
371 }
372 categories = set(list(self.targets.keys()) + list(other.targets.keys()))
373 for category in categories:
374 new.targets[category] = []
375 new.targets[category].extend(t for t in self.targets.get(category, []) if t not in other.targets.get(category, []))
377 return new
379 def __eq__(self, other: object) -> bool:
380 """Compare two targets"""
381 if other is self:
382 return True
383 if other is None:
384 return False
385 if not isinstance(other, Target):
386 return NotImplemented
387 if self.target_data != other.target_data:
388 return False
389 if self.target_specific_data != other.target_specific_data:
390 return False
391 return all(self.targets.get(category, []) == other.targets.get(category, []) for category in self.CATEGORIES)
393 def as_dict(self, **_kwargs: Any) -> dict[str, list[str]]:
394 return {k: v for k, v in self.targets.items() if v}
397class TransportConfig:
398 def __init__(self, conf: ConfigType | None = None, class_config: TransportConfig | None = None) -> None:
399 conf = conf or {}
400 if class_config is not None:
401 self.enabled: bool = conf.get(CONF_ENABLED, class_config.enabled)
402 self.alias = conf.get(CONF_ALIAS)
403 self.delivery_defaults: DeliveryConfig = DeliveryConfig(
404 conf.get(CONF_DELIVERY_DEFAULTS, {}), class_config.delivery_defaults or None
405 )
406 else:
407 self.enabled = conf.get(CONF_ENABLED, True)
408 self.alias = conf.get(CONF_ALIAS)
409 self.delivery_defaults = DeliveryConfig(conf.get(CONF_DELIVERY_DEFAULTS) or {})
411 # deprecation support
412 device_domain = conf.get(CONF_DEVICE_DOMAIN)
413 if device_domain is not None:
414 _LOGGER.warning("SUPERNOTIFY device_domain on transport deprecated, use options instead")
415 self.delivery_defaults.options[OPTION_DEVICE_DOMAIN] = device_domain
416 device_model_include = conf.get(CONF_DEVICE_MODEL_INCLUDE)
417 device_model_exclude = conf.get(CONF_DEVICE_MODEL_EXCLUDE)
418 if device_model_include is not None or device_model_exclude is not None:
419 _LOGGER.warning("SUPERNOTIFY device_model_include/exclude on transport deprecated, use options instead")
420 self.delivery_defaults.options[OPTION_DEVICE_MODEL_SELECT] = {
421 SELECT_INCLUDE: device_model_include,
422 SELECT_EXCLUDE: device_model_exclude,
423 }
424 device_discovery = conf.get(CONF_DEVICE_DISCOVERY)
425 if device_discovery is not None and self.delivery_defaults.options.get(OPTION_DEVICE_DISCOVERY) is None:
426 _LOGGER.warning("SUPERNOTIFY device_discovery on transport deprecated, use options instead")
427 self.delivery_defaults.options[OPTION_DEVICE_DISCOVERY] = device_discovery
430class DeliveryCustomization:
431 def __init__(self, config: ConfigType | None, target_specific: bool = False) -> None:
432 config = config or {}
433 # perhaps should be false for wildcards
434 self.enabled: bool | None = config.get(CONF_ENABLED, True)
435 self.data: dict[str, Any] | None = config.get(CONF_DATA)
436 # TODO: only works for scenario or recipient, not action call
437 self.target: Target | None
439 if config.get(CONF_TARGET):
440 if self.data:
441 self.target = Target(config.get(CONF_TARGET), target_data=self.data, target_specific_data=target_specific)
442 else:
443 self.target = Target(config.get(CONF_TARGET))
444 else:
445 self.target = None
447 def data_value(self, key: str) -> Any:
448 return self.data.get(key) if self.data else None
450 def as_dict(self, **_kwargs: Any) -> dict[str, Any]:
451 return {CONF_TARGET: self.target.as_dict() if self.target else None, CONF_ENABLED: self.enabled, CONF_DATA: self.data}
454class SelectionRule:
455 def __init__(self, config: str | list[str] | dict | SelectionRule | None) -> None:
456 self.include: list[str] | None = None
457 self.exclude: list[str] | None = None
458 if config is None:
459 return
460 if isinstance(config, SelectionRule):
461 self.include = config.include
462 self.exclude = config.exclude
463 elif isinstance(config, str):
464 self.include = [config]
465 elif isinstance(config, list):
466 self.include = config
467 else:
468 if config.get(SELECT_INCLUDE):
469 self.include = ensure_list(config.get(SELECT_INCLUDE))
470 if config.get(SELECT_EXCLUDE):
471 self.exclude = ensure_list(config.get(SELECT_EXCLUDE))
473 def match(self, v: str | Iterable[str] | None) -> bool:
474 if self.include is None and self.exclude is None:
475 return True
476 if isinstance(v, str) or v is None:
477 if self.exclude is not None and v is not None and any(re.fullmatch(pat, v) for pat in self.exclude):
478 return False
479 if self.include is not None and (v is None or not any(re.fullmatch(pat, v) for pat in self.include)):
480 return False
481 else:
482 if self.exclude is not None:
483 for vv in v:
484 if any(re.fullmatch(pat, vv) for pat in self.exclude):
485 return False
486 if self.include is not None:
487 return any(any(re.fullmatch(pat, vv) for pat in self.include) for vv in v)
488 return True
491class DeliveryConfig:
492 """Shared config for transport defaults and Delivery definitions"""
494 def __init__(self, conf: ConfigType, delivery_defaults: DeliveryConfig | None = None) -> None:
496 if delivery_defaults is not None:
497 # use transport defaults where no delivery level override
498 self.target: Target | None = Target(conf.get(CONF_TARGET)) if CONF_TARGET in conf else delivery_defaults.target
499 self.target_required: TargetRequired = conf.get(CONF_TARGET_REQUIRED, delivery_defaults.target_required)
500 self.target_usage: str = conf.get(CONF_TARGET_USAGE) or delivery_defaults.target_usage
501 self.action: str | None = conf.get(CONF_ACTION) or delivery_defaults.action
502 self.debug: bool = conf.get(CONF_DEBUG, delivery_defaults.debug)
504 self.data: ConfigType = dict(delivery_defaults.data) if isinstance(delivery_defaults.data, dict) else {}
505 self.data.update(conf.get(CONF_DATA, {}))
506 self.selection: list[str] = conf.get(CONF_SELECTION, delivery_defaults.selection)
507 self.priority: list[str] = conf.get(CONF_PRIORITY, delivery_defaults.priority)
508 self.selection_rank: SelectionRank = conf.get(CONF_SELECTION_RANK, delivery_defaults.selection_rank)
509 self.options: ConfigType = conf.get(CONF_OPTIONS, {})
510 # only override options not set in config
511 if isinstance(delivery_defaults.options, dict):
512 for opt in delivery_defaults.options:
513 self.options.setdefault(opt, delivery_defaults.options[opt])
514 else:
515 # construct the transport defaults
516 self.target = Target(conf.get(CONF_TARGET)) if conf.get(CONF_TARGET) else None
517 self.target_required = conf.get(CONF_TARGET_REQUIRED, TargetRequired.ALWAYS)
518 self.target_usage = conf.get(CONF_TARGET_USAGE, TARGET_USE_ON_NO_ACTION_TARGETS)
519 self.action = conf.get(CONF_ACTION)
520 self.debug = conf.get(CONF_DEBUG, False)
521 self.options = conf.get(CONF_OPTIONS, {})
522 self.data = conf.get(CONF_DATA, {})
523 self.selection = conf.get(CONF_SELECTION, [SELECTION_DEFAULT])
524 self.priority = conf.get(CONF_PRIORITY, list(PRIORITY_VALUES.keys()))
525 self.selection_rank = conf.get(CONF_SELECTION_RANK, SelectionRank.ANY)
527 def as_dict(self, **_kwargs: Any) -> dict[str, Any]:
528 return {
529 CONF_TARGET: self.target.as_dict() if self.target else None,
530 CONF_ACTION: self.action,
531 CONF_OPTIONS: self.options,
532 CONF_DATA: self.data,
533 CONF_SELECTION: self.selection,
534 CONF_PRIORITY: self.priority,
535 CONF_SELECTION_RANK: str(self.selection_rank),
536 CONF_TARGET_REQUIRED: str(self.target_required),
537 CONF_TARGET_USAGE: self.target_usage,
538 }
540 def __repr__(self) -> str:
541 """Log friendly representation"""
542 return str(self.as_dict())
545@dataclass
546class ConditionVariables:
547 """Variables presented to all condition evaluations
549 Attributes
550 ----------
551 applied_scenarios (list[str]): Scenarios that have been applied
552 required_scenarios (list[str]): Scenarios that must be applied
553 constrain_scenarios (list[str]): Only scenarios in this list, or in explicit apply_scenarios, can be applied
554 notification_priority (str): Priority of the notification
555 notification_message (str): Message of the notification
556 notification_title (str): Title of the notification
557 occupancy (list[str]): List of occupancy scenarios
558 notification_data (dict[str,Any]): Additional data passed on notify action call
560 """
562 applied_scenarios: list[str] = field(default_factory=list)
563 required_scenarios: list[str] = field(default_factory=list)
564 constrain_scenarios: list[str] = field(default_factory=list)
565 notification_priority: str = PRIORITY_MEDIUM
566 notification_message: str | None = ""
567 notification_title: str | None = ""
568 occupancy: list[str] = field(default_factory=list)
570 def __init__(
571 self,
572 applied_scenarios: list[str] | None = None,
573 required_scenarios: list[str] | None = None,
574 constrain_scenarios: list[str] | None = None,
575 delivery_priority: str | None = PRIORITY_MEDIUM,
576 occupiers: dict[str, list[Any]] | None = None,
577 message: str | None = None,
578 title: str | None = None,
579 notification_data: dict[str, Any] | None = None,
580 ) -> None:
581 occupiers = occupiers or {}
582 self.occupancy = []
583 if not occupiers.get(STATE_NOT_HOME) and occupiers.get(STATE_HOME):
584 self.occupancy.append("ALL_HOME")
585 elif occupiers.get(STATE_NOT_HOME) and not occupiers.get(STATE_HOME):
586 self.occupancy.append("ALL_AWAY")
587 if len(occupiers.get(STATE_HOME, [])) == 1:
588 self.occupancy.extend(["LONE_HOME", "SOME_HOME"])
589 elif len(occupiers.get(STATE_HOME, [])) > 1 and occupiers.get(STATE_NOT_HOME):
590 self.occupancy.extend(["MULTI_HOME", "SOME_HOME"])
591 self.applied_scenarios = applied_scenarios or []
592 self.required_scenarios = required_scenarios or []
593 self.constrain_scenarios = constrain_scenarios or []
594 self.notification_priority = delivery_priority or PRIORITY_MEDIUM
595 self.notification_message = message
596 self.notification_title = title
597 self.notification_data: dict[str, Any] = notification_data or {}
599 def as_dict(self, **_kwargs: Any) -> TemplateVarsType:
600 return {
601 "applied_scenarios": self.applied_scenarios,
602 "required_scenarios": self.required_scenarios,
603 "constrain_scenarios": self.constrain_scenarios,
604 "notification_message": self.notification_message,
605 "notification_title": self.notification_title,
606 "notification_priority": self.notification_priority,
607 "occupancy": self.occupancy,
608 "notification_data": self.notification_data,
609 }
612class SuppressionReason(StrEnum):
613 SNOOZED = "SNOOZED"
614 DUPE = "DUPE"
615 NO_SCENARIO = "NO_SCENARIO"
616 NO_ACTION = "NO_ACTION"
617 NO_TARGET = "NO_TARGET"
618 INVALID_ACTION_DATA = "INVALID_ACTION_DATA"
619 TRANSPORT_DISABLED = "TRANSPORT_DISABLED"
620 PRIORITY = "PRIORITY"
621 DELIVERY_CONDITION = "DELIVERY_CONDITION"
622 UNKNOWN = "UNKNOWN"
625class TargetRequired(StrEnum):
626 ALWAYS = auto()
627 NEVER = auto()
628 OPTIONAL = auto()
630 @classmethod
631 def _missing_(cls, value: Any) -> TargetRequired | None:
632 """Backward compatibility for binary values"""
633 if value is True or (isinstance(value, str) and value.lower() in ("true", "on")):
634 return cls.ALWAYS
635 if value is False or (isinstance(value, str) and value.lower() in ("false", "off")):
636 return cls.OPTIONAL
637 return None
640class TargetType(StrEnum):
641 pass
644class GlobalTargetType(TargetType):
645 NONCRITICAL = "NONCRITICAL"
646 EVERYTHING = "EVERYTHING"
649class RecipientType(StrEnum):
650 USER = "USER"
651 EVERYONE = "EVERYONE"
654class QualifiedTargetType(TargetType):
655 TRANSPORT = "TRANSPORT"
656 DELIVERY = "DELIVERY"
657 CAMERA = "CAMERA"
658 PRIORITY = "PRIORITY"
659 MOBILE = "MOBILE"
662class CommandType(StrEnum):
663 SNOOZE = "SNOOZE"
664 SILENCE = "SILENCE"
665 NORMAL = "NORMAL"
668class MessageOnlyPolicy(StrEnum):
669 STANDARD = "STANDARD" # independent title and message
670 USE_TITLE = "USE_TITLE" # use title in place of message, no title
671 # use combined title and message as message, no title
672 COMBINE_TITLE = "COMBINE_TITLE"
675class DebugTrace:
676 def __init__(
677 self,
678 message: str | None,
679 title: str | None,
680 data: dict[str, Any] | None,
681 target: dict[str, list[str]] | list[str] | str | None,
682 ) -> None:
683 self.message: str | None = message
684 self.title: str | None = title
685 self.data: dict[str, Any] | None = dict(data) if data else data
686 self.target: dict[str, list[str]] | list[str] | str | None = list(target) if target else target
687 self.resolved: dict[str, dict[str, Any]] = {}
688 self.delivery_selection: dict[str, list[str]] = {}
689 self.delivery_artefacts: dict[str, Any] = {}
690 self.delivery_exceptions: dict[str, Any] = {}
691 self._last_stage: dict[str, str] = {}
692 self._last_target: dict[str, Any] = {}
694 def contents(self, **_kwargs: Any) -> dict[str, Any]:
695 results: dict[str, Any] = {
696 "arguments": {
697 "message": self.message,
698 "title": self.title,
699 "data": self.data,
700 "target": self.target,
701 },
702 "delivery_selection": self.delivery_selection,
703 "resolved": self.resolved,
704 }
705 if self.delivery_artefacts:
706 results["delivery_artefacts"] = self.delivery_artefacts
707 if self.delivery_artefacts:
708 results["delivery_exceptions"] = self.delivery_exceptions
709 return results
711 def record_target(self, delivery_name: str, stage: str, computed: Target | list[Target]) -> None:
712 """Debug support for recording detailed target resolution in archived notification"""
713 self.resolved.setdefault(delivery_name, {})
714 self.resolved[delivery_name].setdefault(stage, {})
715 self._last_target.setdefault(delivery_name, {})
716 self._last_target[delivery_name].setdefault(stage, {})
717 if isinstance(computed, Target):
718 combined = computed
719 else:
720 combined = Target()
721 for target in ensure_list(computed):
722 combined += target
723 new_target: dict[str, Any] = combined.as_dict()
724 result: str | dict[str, Any] = new_target
725 if self._last_stage.get(delivery_name):
726 last_target = self._last_target[delivery_name][self._last_stage[delivery_name]]
727 if last_target is not None and last_target == result:
728 result = "NO_CHANGE"
730 self.resolved[delivery_name][stage] = result
731 self._last_stage[delivery_name] = stage
732 self._last_target[delivery_name][stage] = new_target
734 def record_delivery_selection(self, stage: str, delivery_selection: list[str]) -> None:
735 """Debug support for recording detailed target resolution in archived notification"""
736 self.delivery_selection[stage] = delivery_selection
738 def record_delivery_artefact(self, delivery: str, artefact_name: str, artefact: Any) -> None:
739 self.delivery_artefacts.setdefault(delivery, {})
740 self.delivery_artefacts[delivery][artefact_name] = artefact
742 def record_delivery_exception(self, delivery: str, context: str, exception: Exception) -> None:
743 self.delivery_exceptions.setdefault(delivery, {})
744 self.delivery_exceptions[delivery].setdefault(context, [])
745 self.delivery_exceptions[delivery][context].append(format_exception(exception))