Coverage for custom_components/supernotify/delivery.py: 99%
292 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-08 17:42 +0000
1from __future__ import annotations
3import logging
4from enum import StrEnum, auto
5from typing import TYPE_CHECKING, Any
7from homeassistant.const import (
8 ATTR_DEVICE_ID,
9 ATTR_FRIENDLY_NAME,
10 ATTR_NAME,
11 CONF_ACTION,
12 CONF_ALIAS,
13 CONF_CONDITIONS,
14 CONF_DEBUG,
15 CONF_ENABLED,
16 CONF_NAME,
17 CONF_OPTIONS,
18 CONF_TARGET,
19)
20from homeassistant.core import callback
22from custom_components.supernotify.target import Target, TargetEntityCategory
24from .common import ensure_list
25from .const import (
26 ATTR_ENABLED,
27 ATTR_LAST_OUTCOME,
28 ATTR_LAST_SENT_AT,
29 ATTR_LAST_SKIP_REASON,
30 ATTR_MOBILE_APP_ID,
31 ATTR_TRANSPORT_ENABLED,
32 CONF_DATA,
33 CONF_DEFAULT_INCLUSION,
34 CONF_DELIVERY_DEFAULTS,
35 CONF_FALLBACK,
36 CONF_INCLUSION,
37 CONF_LOAD,
38 CONF_MESSAGE,
39 CONF_NOTIFY_ENTITY,
40 CONF_OCCUPANCY,
41 CONF_TARGET_REQUIRED,
42 CONF_TARGET_USAGE,
43 CONF_TEMPLATE,
44 CONF_TITLE,
45 CONF_TRANSPORT,
46 CONF_VOICE_OCCUPANCY,
47 INCLUSION_DEFAULT,
48 INCLUSION_EXPLICIT,
49 INCLUSION_FALLBACK,
50 INCLUSION_FALLBACK_ON_ERROR,
51 RESERVED_DELIVERY_NAMES,
52)
53from .model import ConditionVariables, DeliveryConfig, SelectionRule, TransportFeature
54from .options import (
55 OPTION_DATA_KEYS_EXCLUDE_RE,
56 OPTION_DATA_KEYS_INCLUDE_RE,
57 OPTION_DATA_KEYS_SELECT,
58 OPTION_DEVICE_AREA_SELECT,
59 OPTION_DEVICE_DISCOVERY,
60 OPTION_DEVICE_DOMAIN,
61 OPTION_DEVICE_LABEL_SELECT,
62 OPTION_DEVICE_MANUFACTURER_SELECT,
63 OPTION_DEVICE_MODEL_SELECT,
64 OPTION_DEVICE_OS_SELECT,
65 OPTION_TARGET_CATEGORIES,
66 OPTION_TARGET_INCLUDE_RE,
67 OPTION_TARGET_SELECT,
68 SELECT_EXCLUDE,
69 SELECT_INCLUDE,
70)
71from .static_config import TRANSPORT_NAMES
73if TYPE_CHECKING:
74 import datetime as dt
76 from homeassistant.helpers.typing import ConfigType
78 from custom_components.supernotify.hass_api import TrackedDeviceDetails
79 from custom_components.supernotify.transport import Transport
81 from .binary_sensor import SupernotifyLegacyBinarySensor
82 from .context import Context
83 from .schema import ConditionsFunc
85_LOGGER = logging.getLogger(__name__)
88class DeliveryProvenance(StrEnum):
89 DEFAULT_STANDARD = auto()
90 EXTRA_STANDARD = auto()
91 CONFIG = auto()
94class Delivery(DeliveryConfig):
95 def __init__(
96 self, name: str, conf: ConfigType, transport: Transport, provenance: DeliveryProvenance = DeliveryProvenance.CONFIG
97 ) -> None:
98 conf = conf or {}
99 self.name: str = name
100 self.provenance: DeliveryProvenance = provenance
101 self.transport: Transport = transport
102 self._raw_conf: ConfigType = conf
103 # whether this delivery's own config explicitly declared `inclusion:` (or its
104 # deprecated `selection:` alias, already normalised to `inclusion` by schema.py's
105 # cv.deprecated() before this runs) - as opposed to merely inheriting the transport's
106 # own bare default. A deliberate `inclusion: default` on a delivery someone configured
107 # themselves is trusted at face value for implicit selection (see
108 # DeliveryRegistry.implicit_deliveries), unlike a standard delivery's inherited default.
109 self.explicit_inclusion: bool = CONF_INCLUSION in conf
110 transport_defaults: DeliveryConfig = self.transport.delivery_defaults
111 super().__init__(conf, delivery_defaults=transport_defaults)
112 # deprecated option aliases must be rewritten before anything below reads self.options
113 self.migrate_config()
114 if isinstance(self.target, Target):
115 # a value set directly on this delivery's own `target:` is exclusively scoped to
116 # it - unlike a blended notification-level target list - so it's safe to claim an
117 # unqualified value (no shape a validator recognises) for this transport. (The
118 # isinstance check, not just a None check, is deliberate: a test double `Mock()`
119 # transport can leave `self.target` as an auto-mocked attribute rather than None.)
120 self.target = self.reclassify_unqualified_target(self.target)
121 # as configured, which enabled can be overridden from at runtime by the delivery switch
122 self.config_enabled: bool = conf.get(CONF_ENABLED, self.transport.config_enabled)
123 self.enabled: bool = self.config_enabled
124 # deliveries to try, in order, if this one fails - see Notification._fall_back()
125 self.fallback: list[str] = list(conf.get(CONF_FALLBACK) or [])
126 self.notify_entity: str | None = conf.get(CONF_NOTIFY_ENTITY)
127 # set once its notify entity is made, see exposed_notify_entities() in notify.py
128 self.notify_entity_id: str | None = None
129 # set by the registry when another delivery lists this one in its `fallback:`
130 self.is_fallback: bool = False
131 self.conditions: ConditionsFunc | None = None
132 self.transport_data: dict[str, Any] = {}
133 # how the latest live notification went for this delivery - see record_outcome()
134 self.last_sent_at: dt.datetime | None = None
135 self.last_outcome: str | None = None
136 self.last_skip_reason: str | None = None
137 if self.options.get(OPTION_TARGET_SELECT):
138 self.target_selector: SelectionRule | None = SelectionRule(self.options.get(OPTION_TARGET_SELECT))
139 else:
140 self.target_selector = None
142 async def initialize(self, context: Context) -> bool:
143 self.check_lost_implicit_inclusion(context)
144 errors = await self.validate_config(context)
145 self.discover_devices(context)
146 self.transport_data = self.transport.setup_delivery_options(self.options, self.name)
147 return errors == 0
149 def migrate_config(self) -> None:
150 """Rewrite deprecated option aliases in place. Called from `__init__`, before
152 anything else reads `self.options` - unlike the checks in `validate_config()` and
153 `check_lost_implicit_inclusion()`, this never needs `Context`, so there's no reason
154 to defer it to `initialize()` and risk a constructed-but-not-initialized `Delivery`
155 (as in many unit tests) seeing the pre-migration option name.
156 """
157 # v1.9.0
158 if (
159 OPTION_DATA_KEYS_INCLUDE_RE in self.options or OPTION_DATA_KEYS_EXCLUDE_RE in self.options
160 ) and not self.options.get(OPTION_DATA_KEYS_SELECT):
161 _LOGGER.warning(
162 "SUPERNOTIFY Deprecated use of data_keys_include_re/data_keys_exclude_re options - use data_keys_select"
163 )
164 self.options[OPTION_DATA_KEYS_SELECT] = {
165 SELECT_INCLUDE: self.options.get(OPTION_DATA_KEYS_INCLUDE_RE),
166 SELECT_EXCLUDE: self.options.get(OPTION_DATA_KEYS_EXCLUDE_RE),
167 }
168 # v1.9.0
169 if OPTION_TARGET_INCLUDE_RE in self.options and not self.options.get(OPTION_TARGET_SELECT):
170 _LOGGER.warning("SUPERNOTIFY Deprecated use of target_include_re option - use target_select")
171 self.options[OPTION_TARGET_SELECT] = {SELECT_INCLUDE: self.options.get(OPTION_TARGET_INCLUDE_RE)}
173 def check_lost_implicit_inclusion(self, context: Context) -> None:
174 """v2.11.0: flag a behaviour change that can't be silently auto-upgraded, unlike
176 `migrate_config()`'s option aliases - we can't know what inclusion the user actually
177 wants, so this only raises a repair issue for them to act on. Needs `Context`
178 (`delivery_registry.default_inclusion`, `hass_api.raise_issue`), so - like
179 `validate_config()` - it's deferred to `initialize()` rather than run from `__init__`,
180 to avoid raising issues for a `Delivery` that's constructed but never initialized (as
181 in many unit tests).
182 """
183 if (
184 self.provenance == DeliveryProvenance.CONFIG
185 and CONF_INCLUSION not in self._raw_conf
186 and INCLUSION_DEFAULT not in self.inclusion
187 and context.delivery_registry.default_inclusion is None
188 ):
189 _LOGGER.info(
190 "SUPERNOTIFY Delivery %s has no explicit inclusion, but transport %s no longer defaults to "
191 "'default' - it will not fire implicitly",
192 self.name,
193 self.transport.name,
194 )
195 context.hass_api.raise_issue(
196 f"delivery_{self.name}_lost_implicit_inclusion",
197 issue_key="delivery_lost_implicit_inclusion",
198 issue_map={"delivery": self.name, "transport": self.transport.name},
199 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/deliveries/",
200 )
202 async def validate_config(self, context: Context) -> int:
203 """Genuine per-delivery validation - as opposed to the version-history migration
205 concerns in `migrate_config()` and `check_lost_implicit_inclusion()`. Each failure is
206 raised as a repair issue and counted, so `DeliveryRegistry` can drop this delivery
207 rather than run it in a state nobody asked for.
208 """
209 errors = 0
210 if self.name in TRANSPORT_NAMES and self.transport.name != self.name:
211 _LOGGER.warning(
212 "SUPERNOTIFY Delivery %s is a reserved name for the standard delivery of %s transport", self.name, self.name
213 )
214 context.hass_api.raise_issue(
215 f"delivery_{self.name}_reserved_name",
216 issue_key="delivery_reserved_name",
217 issue_map={"delivery": self.name},
218 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/deliveries/",
219 )
220 if self.name in RESERVED_DELIVERY_NAMES:
221 _LOGGER.warning("SUPERNOTIFY Delivery uses reserved word %s", self.name)
222 context.hass_api.raise_issue(
223 f"delivery_{self.name}_reserved_name",
224 issue_key="delivery_reserved_name",
225 issue_map={"delivery": self.name},
226 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/deliveries/",
227 )
228 errors += 1
229 if not self.transport.validate_action(self.action):
230 _LOGGER.warning("SUPERNOTIFY Invalid action definition for delivery %s (%s)", self.name, self.action)
231 context.hass_api.raise_issue(
232 f"delivery_{self.name}_invalid_action",
233 issue_key="delivery_invalid_action",
234 issue_map={"delivery": self.name, "action": self.action or ""},
235 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/deliveries/",
236 )
237 errors += 1
239 if self.conditions_config:
240 try:
241 self.conditions = await context.hass_api.build_conditions(
242 self.conditions_config, validate=True, strict=True, name=self.name
243 )
244 passed = True
245 exception = ""
246 except Exception as e:
247 passed = False
248 exception = str(e)
249 if not passed:
250 _LOGGER.warning("SUPERNOTIFY Invalid delivery conditions for %s: %s", self.name, self.conditions_config)
251 context.hass_api.raise_issue(
252 f"delivery_{self.name}_invalid_condition",
253 issue_key="delivery_invalid_condition",
254 issue_map={"delivery": self.name, "condition": str(self.conditions_config), "exception": exception},
255 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/deliveries/",
256 )
257 errors += 1
258 return errors
260 def discover_devices(self, context: Context) -> None:
261 if self.options.get(OPTION_DEVICE_DISCOVERY, False):
262 for domain in self.options.get(OPTION_DEVICE_DOMAIN, []):
263 discovered: int = 0
264 added: int = 0
265 for d in context.hass_api.discover_devices(
266 domain,
267 device_model_select=SelectionRule(self.options.get(OPTION_DEVICE_MODEL_SELECT)),
268 device_manufacturer_select=SelectionRule(self.options.get(OPTION_DEVICE_MANUFACTURER_SELECT)),
269 device_os_select=SelectionRule(self.options.get(OPTION_DEVICE_OS_SELECT)),
270 device_area_select=SelectionRule(self.options.get(OPTION_DEVICE_AREA_SELECT)),
271 device_label_select=SelectionRule(self.options.get(OPTION_DEVICE_LABEL_SELECT)),
272 ):
273 discovered += 1
274 if self.target is None:
275 self.target = Target()
276 if domain == "mobile_app":
277 mobile_app: TrackedDeviceDetails | None = context.hass_api.mobile_app_by_device_id(d.device_id)
278 if mobile_app and mobile_app.action:
279 mobile_app_id = mobile_app.mobile_app_id if mobile_app else None
280 if mobile_app_id and mobile_app_id not in self.target.mobile_app_ids:
281 _LOGGER.debug(
282 f"SUPERNOTIFY Found mobile {d.model} device {d.device_name} for {domain}, id {d.device_id}"
283 )
284 self.target.extend(ATTR_MOBILE_APP_ID, mobile_app_id)
285 added += 1
286 else:
287 _LOGGER.debug(f"SUPERNOTIFY Skipped mobile without notify entity {d.device_name}, id {d.device_id}")
288 else:
289 if d.device_id not in self.target.device_ids:
290 _LOGGER.debug(
291 "SUPERNOTIFY Found %s device %s for %s, id %s", d.model, d.device_name, domain, d.device_id
292 )
293 self.target.extend(ATTR_DEVICE_ID, d.device_id)
294 added += 1
296 _LOGGER.info(f"SUPERNOTIFY {self.name} Device discovery for {domain} found {discovered} devices, added {added}")
298 @property
299 def target_categories(self) -> list[str | TargetEntityCategory]:
300 """The target categories this delivery accepts - the query point for "what does this
302 delivery support", so callers never need to look at `Transport` and `OPTION_TARGET_
303 CATEGORIES` separately. For every transport other than `generic`, this is direct
304 delegation to `self.transport.target_categories` (a delivery rarely needs to widen
305 what its transport understands). `generic` is the bring-your-own-categories
306 transport - it declares nothing itself, so a delivery's own `OPTION_TARGET_CATEGORIES`
307 (e.g. a made-up "slack_channel" category) is what actually defines its categories.
309 The configured list is listed first, so it takes precedence as the reclassification
310 fallback in `reclassify_unqualified_target()` when both are present.
311 """
312 configured = ensure_list(self.options.get(OPTION_TARGET_CATEGORIES))
313 return [*configured, *self.transport.target_categories]
315 def reclassify_unqualified_target(self, target: Target) -> Target:
316 """Reclassify this delivery's uncategorised target values into its primary target
318 category, since a value with no distinguishing shape (e.g. an MQTT topic, a Discord
319 channel ID, or a made-up category for `generic`, like a Slack channel) would
320 otherwise never survive `select_targets()` on its own. Left alone if this delivery's
321 `target_categories` explicitly lists the uncategorised bucket itself - that's a
322 deliberate choice to accept unqualified values exactly as they are - or if it has no
323 plain-string category to fall back to at all.
325 Only safe to call on a `Target` that is exclusively scoped to this one delivery - its
326 own configured `target:`, or a per-delivery override - never on a blended,
327 notification-level target list. There, an unqualified value must stay unclaimed
328 rather than being guessed at: it could belong to a different delivery entirely, and
329 claiming it here would leak it away from wherever it actually belongs.
330 """
331 unqualified = target.targets.get(Target.UNKNOWN_CUSTOM_CATEGORY)
332 if not unqualified:
333 return target
334 declared = self.target_categories
335 if Target.UNKNOWN_CUSTOM_CATEGORY in declared:
336 return target
337 primary = next((c for c in declared if isinstance(c, str)), None)
338 if primary is None:
339 # this target is exclusively scoped to this delivery (the safety precondition
340 # above), so if there's genuinely nowhere for it to go, it's not a value meant
341 # for a different delivery - it's just unmappable, and would otherwise be
342 # dropped with no visible explanation
343 _LOGGER.warning(
344 "SUPERNOTIFY Delivery %s (%s) has no target category to accept unqualified target(s) %s - "
345 "dropping. Known categories for this delivery: %s",
346 self.name,
347 self.transport.name,
348 unqualified,
349 [c if isinstance(c, str) else "entity_id" for c in declared] or "none",
350 )
351 return target
352 result = target.safe_copy()
353 result.targets.pop(Target.UNKNOWN_CUSTOM_CATEGORY, None)
354 result.extend(primary, unqualified)
355 return result
357 def select_targets(self, target: Target) -> Target:
358 return target.select(
359 self.target_categories, (self.name, self.transport.name), self.transport.hass_api, self.target_selector
360 )
362 def evaluate_conditions(self, condition_variables: ConditionVariables) -> bool | None:
363 if not self.enabled:
364 return False
365 if self.conditions is None:
366 return True
367 # TODO: reconsider hass_api injection
368 return self.transport.hass_api.evaluate_conditions(self.conditions, condition_variables)
370 def option(self, option_name: str, default: str | bool) -> str | bool:
371 """Get an option value from delivery config or transport default options"""
372 opt: str | bool | None = None
373 if option_name in self.options:
374 opt = self.options[option_name]
375 if opt is None:
376 _LOGGER.debug(
377 "SUPERNOTIFY No default in delivery %s for option %s, setting to default %s", self.name, option_name, default
378 )
379 opt = self.options[option_name] = default
380 return opt
382 def option_bool(self, option_name: str, default: bool = False) -> bool:
383 return bool(self.option(option_name, default=default))
385 def option_str(self, option_name: str) -> str:
386 return str(self.option(option_name, default=""))
388 def as_dict(self, **_kwargs: Any) -> dict[str, Any]:
389 base = super().as_dict()
390 base.update({
391 CONF_NAME: self.name,
392 CONF_ALIAS: self.alias,
393 CONF_TRANSPORT: self.transport.name,
394 CONF_TEMPLATE: self.template,
395 CONF_MESSAGE: self.message,
396 CONF_TITLE: self.title,
397 CONF_ENABLED: self.enabled,
398 CONF_OCCUPANCY: self.occupancy,
399 CONF_CONDITIONS: self.conditions,
400 })
401 return base
403 def record_outcome(self, outcome: str, at: dt.datetime, sent: bool = False, skip_reason: str | None = None) -> None:
404 """Keep how the latest live notification went for this delivery: `outcome` is success,
405 error or skipped, `sent` whether anything actually went out (so `last_sent_at` moves on),
406 and `skip_reason` why it was skipped or suppressed. Shown on the delivery switch."""
407 self.last_outcome = outcome
408 self.last_skip_reason = skip_reason
409 if sent:
410 self.last_sent_at = at
412 def attributes(self) -> dict[str, Any]:
413 """For exposure as entity state"""
414 attrs: dict[str, Any] = {
415 ATTR_NAME: self.name,
416 ATTR_ENABLED: self.enabled,
417 CONF_TRANSPORT: self.transport.name,
418 # a delivery is only used while its transport is enabled too
419 ATTR_TRANSPORT_ENABLED: self.transport.enabled,
420 CONF_ACTION: self.action,
421 CONF_OPTIONS: self.options,
422 CONF_INCLUSION: self.inclusion,
423 CONF_TARGET: self.target,
424 CONF_TARGET_REQUIRED: self.target_required,
425 CONF_TARGET_USAGE: self.target_usage,
426 CONF_DATA: self.data,
427 CONF_DEBUG: self.debug,
428 }
429 if self.fallback:
430 # so a dashboard can show which deliveries stand in when this one fails
431 attrs[CONF_FALLBACK] = self.fallback
432 if self.alias:
433 attrs[ATTR_FRIENDLY_NAME] = self.alias
434 if self.last_sent_at is not None:
435 attrs[ATTR_LAST_SENT_AT] = self.last_sent_at
436 if self.last_outcome is not None:
437 attrs[ATTR_LAST_OUTCOME] = self.last_outcome
438 if self.last_skip_reason is not None:
439 attrs[ATTR_LAST_SKIP_REASON] = self.last_skip_reason
440 return attrs
443class DeliveryRegistry:
444 def __init__(
445 self,
446 deliveries: ConfigType | None = None,
447 transport_configs: ConfigType | None = None,
448 transport_types: list[type[Transport]] | dict[type[Transport], dict[str, Any]] | None = None,
449 # for unit tests only
450 transport_instances: list[Transport] | None = None,
451 delivery_control: ConfigType | None = None,
452 ) -> None:
453 # raw configured deliveries
454 self._config_deliveries: ConfigType = deliveries if isinstance(deliveries, dict) else {}
455 # validated deliveries
456 self._deliveries: dict[str, Delivery] = {}
457 self.transports: dict[str, Transport] = {}
458 self._transport_configs: ConfigType = transport_configs or {}
459 # The deprecated delivery and transport binary_sensors, by unique_id - populated by
460 # binary_sensor.py as each is added, only for an existing install that still has them
461 self._entities: dict[str, SupernotifyLegacyBinarySensor] = {}
463 self._transport_types: dict[type[Transport], dict[str, Any]]
464 if isinstance(transport_types, list):
465 self._transport_types = {t: {} for t in transport_types}
466 else:
467 self._transport_types = transport_types or {}
468 # test harness support
469 self._transport_instances: list[Transport] | None = transport_instances
470 # the Delivery Control options, each None when not set, leaving transports' own defaults
471 delivery_control = delivery_control or {}
472 self.default_inclusion: str | None = delivery_control.get(CONF_DEFAULT_INCLUSION)
473 self.voice_occupancy: str | None = delivery_control.get(CONF_VOICE_OCCUPANCY)
475 async def initialize(self, context: Context) -> None:
476 await self.initialize_transports(context)
478 def unload_unused_transports(self) -> None:
479 """Drop any transport that ended up with no delivery at all - explicit or auto-generated.
481 Deliberately deferred until both initialize_transport_deliveries() (explicit) and
482 build_standard_deliveries() (implicit) have run, rather than decided per-transport up
483 front: whether a transport is worth having can only be known once the full, resolved
484 set of deliveries exists - for a transport like `generic` (bring-your-own-action,
485 entirely delivery-driven), there's no transport-level state to check in advance at all.
486 """
487 used_transport_names = {d.transport.name for d in self._deliveries.values()}
488 for name in list(self.transports):
489 if name not in used_transport_names:
490 # routine for a transport never mentioned in config, only worth an info if it was
491 level = logging.INFO if name in self._transport_configs else logging.DEBUG
492 _LOGGER.log(level, "SUPERNOTIFY %s transport has no deliveries, unloading", name)
493 del self.transports[name]
495 def register_entity(self, unique_id: str, entity: SupernotifyLegacyBinarySensor) -> None:
496 """Called by a delivery or transport binary_sensor's async_added_to_hass()."""
497 self._entities[unique_id] = entity
499 def unregister_entity(self, unique_id: str) -> None:
500 """Called by a delivery or transport binary_sensor's async_will_remove_from_hass()."""
501 self._entities.pop(unique_id, None)
503 def legacy_entities(self) -> list[SupernotifyLegacyBinarySensor]:
504 """Every registered delivery and transport binary_sensor - used by supernotify.refresh_entities."""
505 return list(self._entities.values())
507 @callback
508 def async_refresh_entity(self, unique_id: str) -> None:
509 """Re-publish one delivery or transport binary_sensor now, by its unique_id"""
510 entity = self._entities.get(unique_id)
511 if entity is not None:
512 entity.async_write_ha_state()
514 @property
515 def deliveries(self) -> dict[str, Delivery]:
516 return dict(self._deliveries.items())
518 def resolve_name(self, name: str) -> str:
519 """Backward compatibility for the original 'DEFAULT_x' auto-configured naming,
520 long since replaced by plain transport names: a reference to the old 'DEFAULT_x'
521 form resolves to the current 'x' delivery, if that's what actually exists now."""
522 if name not in self._deliveries and name.startswith("DEFAULT_"):
523 plain_name = name.removeprefix("DEFAULT_")
524 if plain_name in self._deliveries:
525 return plain_name
526 return name
528 @property
529 def enabled_deliveries(self) -> dict[str, Delivery]:
530 return {d: dconf for d, dconf in self._deliveries.items() if dconf.enabled}
532 @property
533 def disabled_deliveries(self) -> dict[str, Delivery]:
534 return {d: dconf for d, dconf in self._deliveries.items() if not dconf.enabled}
536 # Computed on each call, not cached at startup, so that a delivery enabled at runtime (by its
537 # switch) is included just like one enabled in config. delivery.inclusion can also be
538 # INCLUSION_BY_SCENARIO or INCLUSION_EXPLICIT to have it only used where asked for.
540 @property
541 def fallback_by_default_deliveries(self) -> list[Delivery]:
542 return [d for d in self._deliveries.values() if d.enabled and INCLUSION_FALLBACK in d.inclusion]
544 @property
545 def fallback_on_error_deliveries(self) -> list[Delivery]:
546 return [d for d in self._deliveries.values() if d.enabled and INCLUSION_FALLBACK_ON_ERROR in d.inclusion]
548 @property
549 def choosable_deliveries(self) -> dict[str, Delivery]:
550 """Deliveries a notification can ask for - not ones only a scenario turns on, or only a fallback.
551 What the action editor's Delivery list and the LLM tools offer."""
552 return {
553 name: d for name, d in self._deliveries.items() if {INCLUSION_DEFAULT, INCLUSION_EXPLICIT}.intersection(d.inclusion)
554 }
556 @property
557 def implicit_deliveries(self) -> list[Delivery]:
558 """Deliveries actually capable of implicit inclusion - for an auto-generated standard
559 delivery, `default` inclusion alone isn't enough any more, its transport also has to
560 declare a `unique_target_categories` or `fallback_target_categories` for
561 `Notification.select_deliveries()` to ever actually match against; without one, a
562 nominally `default` standard delivery (most transports that don't say otherwise)
563 structurally can never be auto-selected, the same as an explicit-only one. A delivery
564 whose own config explicitly declared `inclusion: default` (`Delivery.explicit_
565 inclusion`) is exempt from that check - that's already a deliberate, explicit choice,
566 not something merely inherited from the transport's now-universal bare `default`
567 (`TransportConfig()`'s own constructor, `model.py` - `Transport.inclusion_mode` used to
568 override this down to `explicit` for most transports; now that it's gone, every
569 transport that doesn't declare categories or set its own `delivery_defaults.inclusion`
570 nominally inherits `default` too, without it being a deliberate choice by anyone), so
571 it's trusted at face value exactly as before this mechanism existed (e.g. a fixed-action
572 `generic` delivery with no target concept at all, meant to fire on every notification).
573 A delivery that leaves `inclusion:` unset entirely still goes through the category
574 check, even though it inherits the same nominal `default` value - inheriting isn't
575 choosing. This is the single source of truth for "implicit-capable" - both the actual
576 selection logic and this reporting property (the action editor's implied default list,
577 `enquire_implicit_deliveries`, diagnostics) read it."""
578 return [
579 d
580 for d in self._deliveries.values()
581 if d.enabled
582 and INCLUSION_DEFAULT in d.inclusion
583 and (d.explicit_inclusion or d.transport.unique_target_categories or d.transport.fallback_target_categories)
584 ]
586 def apply_delivery_control(self, transport: Transport, transport_config: ConfigType) -> None:
587 """Give a transport's deliveries the Delivery Control defaults, where the transport's own
588 YAML `delivery_defaults` doesn't set them. A delivery's own YAML still wins over both."""
589 own: ConfigType = transport_config.get(CONF_DELIVERY_DEFAULTS) or {}
590 defaults: DeliveryConfig = transport.delivery_defaults
591 if self.default_inclusion and CONF_INCLUSION not in own:
592 defaults.inclusion = [self.default_inclusion]
593 if self.voice_occupancy and CONF_OCCUPANCY not in own and transport.supported_features & TransportFeature.SPOKEN:
594 defaults.occupancy = self.voice_occupancy
596 async def initialize_transports(self, context: Context) -> None:
597 if self._transport_instances:
598 """Used by configure_for_tests() and TestingContext to set transports to mocks or manually created fixtures"""
599 for transport in self._transport_instances:
600 self.transports[transport.name] = transport
601 await transport.initialize()
602 await self.initialize_transport_deliveries(context, transport)
604 if self._transport_types:
605 # production usage
606 for transport_class, kwargs in self._transport_types.items():
607 transport_config: ConfigType = self._transport_configs.get(transport_class.name, {})
608 if not transport_config.get(CONF_LOAD, True):
609 # not just disabled: excluded entirely, so no deliveries or entities either
610 _LOGGER.debug("SUPERNOTIFY %s transport configured not to load", transport_class.name)
611 continue
612 transport = transport_class(context, transport_config, **kwargs)
613 self.apply_delivery_control(transport, transport_config)
614 if not transport.is_viable(context.hass_api):
615 # quietly skip transports for integrations that aren't installed at all
616 level = (
617 logging.DEBUG
618 if transport.integration_domain
619 and not context.hass_api.integration_loaded(transport.integration_domain)
620 else logging.INFO
621 )
622 _LOGGER.log(level, "SUPERNOTIFY %s transport has no viable configuration, not loaded", transport_class.name)
623 continue
624 self.transports[transport_class.name] = transport
625 await transport.initialize()
626 await self.initialize_transport_deliveries(context, transport)
627 self.transports[transport_class.name] = transport
629 unconfigured_deliveries = [dc for d, dc in self._config_deliveries.items() if d not in self._deliveries]
630 for bad_del in unconfigured_deliveries:
631 # presumably there was no transport for these
632 context.hass_api.raise_issue(
633 f"delivery_{bad_del.get(CONF_NAME)}_for_transport_{bad_del.get(CONF_TRANSPORT)}_failed_to_configure",
634 issue_key="delivery_unknown_transport",
635 issue_map={
636 "delivery": bad_del.get(CONF_NAME),
637 "transport": bad_del.get(CONF_TRANSPORT),
638 "transports": ", ".join(sorted(self.transports)),
639 },
640 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/deliveries/",
641 )
643 self.unload_unused_transports()
644 for delivery in self._deliveries.values():
645 for name in delivery.fallback:
646 if name not in self._deliveries or name == delivery.name:
647 _LOGGER.warning(
648 "SUPERNOTIFY Delivery %s has fallback %s, which is not another delivery", delivery.name, name
649 )
650 else:
651 self._deliveries[name].is_fallback = True
652 _LOGGER.info("SUPERNOTIFY Configured deliveries %s", "; ".join(self._deliveries.keys()))
654 async def initialize_transport_deliveries(self, context: Context, transport: Transport) -> None:
655 """Validate and initialize deliveries at startup for this transport"""
656 validated_deliveries: dict[str, Delivery] = {}
657 configured_deliveries: dict[str, ConfigType] = {
658 d: dc for d, dc in self._config_deliveries.items() if dc.get(CONF_TRANSPORT) == transport.name
659 }
660 # hackily put here, since build_standard_deliveries can have side-effect of updating default deliveries
661 standard_deliveries: dict[str, ConfigType] = transport.build_standard_deliveries(context.hass_api)
663 for d, dc in configured_deliveries.items():
664 # don't care about ENABLED here since disabled deliveries can be overridden later
665 delivery = Delivery(d, dc, transport, DeliveryProvenance.CONFIG)
666 if not await delivery.initialize(context):
667 _LOGGER.error(f"SUPERNOTIFY Ignoring configured delivery {d} with errors")
668 else:
669 validated_deliveries[d] = delivery
671 # merge in remaining standard deliveries but allow local override
672 for d, dc in standard_deliveries.items():
673 if d in configured_deliveries:
674 _LOGGER.info("SUPERNOTIFY Default standard delivery %s overridden by config", d)
675 else:
676 provenance = DeliveryProvenance.DEFAULT_STANDARD if d == transport.name else DeliveryProvenance.EXTRA_STANDARD
677 delivery = Delivery(d, dc, transport, provenance=provenance)
679 if not await delivery.initialize(context):
680 _LOGGER.error(f"SUPERNOTIFY Ignoring standard delivery {d} with errors")
681 else:
682 validated_deliveries[d] = delivery
684 self._deliveries.update(validated_deliveries)
686 _LOGGER.debug(
687 "SUPERNOTIFY Validated transport %s, default action %s, valid deliveries: %s",
688 transport.name,
689 transport.delivery_defaults.action,
690 [d for d in self._deliveries.values() if d.enabled and d.transport == transport],
691 )