Coverage for custom_components/supernotify/delivery.py: 98%
259 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 21:14 +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_MOBILE_APP_ID,
28 ATTR_TRANSPORT_ENABLED,
29 CONF_DATA,
30 CONF_DEFAULT_INCLUSION,
31 CONF_DELIVERY_DEFAULTS,
32 CONF_INCLUSION,
33 CONF_LOAD,
34 CONF_MESSAGE,
35 CONF_OCCUPANCY,
36 CONF_TARGET_REQUIRED,
37 CONF_TARGET_USAGE,
38 CONF_TEMPLATE,
39 CONF_TITLE,
40 CONF_TRANSPORT,
41 CONF_VOICE_OCCUPANCY,
42 INCLUSION_DEFAULT,
43 INCLUSION_EXPLICIT,
44 INCLUSION_FALLBACK,
45 INCLUSION_FALLBACK_ON_ERROR,
46 RESERVED_DELIVERY_NAMES,
47)
48from .model import ConditionVariables, DeliveryConfig, SelectionRule, TransportFeature
49from .options import (
50 OPTION_DATA_KEYS_EXCLUDE_RE,
51 OPTION_DATA_KEYS_INCLUDE_RE,
52 OPTION_DATA_KEYS_SELECT,
53 OPTION_DEVICE_AREA_SELECT,
54 OPTION_DEVICE_DISCOVERY,
55 OPTION_DEVICE_DOMAIN,
56 OPTION_DEVICE_LABEL_SELECT,
57 OPTION_DEVICE_MANUFACTURER_SELECT,
58 OPTION_DEVICE_MODEL_SELECT,
59 OPTION_DEVICE_OS_SELECT,
60 OPTION_TARGET_CATEGORIES,
61 OPTION_TARGET_INCLUDE_RE,
62 OPTION_TARGET_SELECT,
63 SELECT_EXCLUDE,
64 SELECT_INCLUDE,
65)
66from .static_config import TRANSPORT_NAMES
68if TYPE_CHECKING:
69 from homeassistant.helpers.typing import ConfigType
71 from custom_components.supernotify.hass_api import TrackedDeviceDetails
72 from custom_components.supernotify.transport import Transport
74 from .binary_sensor import SupernotifyLegacyBinarySensor
75 from .context import Context
76 from .schema import ConditionsFunc
78_LOGGER = logging.getLogger(__name__)
81class DeliveryProvenance(StrEnum):
82 DEFAULT_STANDARD = auto()
83 EXTRA_STANDARD = auto()
84 CONFIG = auto()
87class Delivery(DeliveryConfig):
88 def __init__(
89 self, name: str, conf: ConfigType, transport: Transport, provenance: DeliveryProvenance = DeliveryProvenance.CONFIG
90 ) -> None:
91 conf = conf or {}
92 self.name: str = name
93 self.provenance: DeliveryProvenance = provenance
94 self.transport: Transport = transport
95 self._raw_conf: ConfigType = conf
96 transport_defaults: DeliveryConfig = self.transport.delivery_defaults
97 super().__init__(conf, delivery_defaults=transport_defaults)
98 if isinstance(self.target, Target):
99 # a value set directly on this delivery's own `target:` is exclusively scoped to
100 # it - unlike a blended notification-level target list - so it's safe to claim an
101 # unqualified value (no shape a validator recognises) for this transport. (The
102 # isinstance check, not just a None check, is deliberate: a test double `Mock()`
103 # transport can leave `self.target` as an auto-mocked attribute rather than None.)
104 self.target = self.reclassify_unqualified_target(self.target)
105 # as configured, which enabled can be overridden from at runtime by the delivery switch
106 self.config_enabled: bool = conf.get(CONF_ENABLED, self.transport.config_enabled)
107 self.enabled: bool = self.config_enabled
108 self.conditions: ConditionsFunc | None = None
109 self.transport_data: dict[str, Any] = {}
110 if self.options.get(OPTION_TARGET_SELECT):
111 self.target_selector: SelectionRule | None = SelectionRule(self.options.get(OPTION_TARGET_SELECT))
112 else:
113 self.target_selector = None
114 self.upgrade_deprecations(conf)
116 async def initialize(self, context: Context) -> bool:
117 errors = 0
118 if self.name in TRANSPORT_NAMES and self.transport.name != self.name:
119 _LOGGER.warning(
120 "SUPERNOTIFY Delivery %s is a reserved name for the standard delivery of %s transport", self.name, self.name
121 )
122 context.hass_api.raise_issue(
123 f"delivery_{self.name}_reserved_name",
124 issue_key="delivery_reserved_name",
125 issue_map={"delivery": self.name},
126 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/deliveries/",
127 )
128 if (
129 CONF_INCLUSION not in self._raw_conf
130 and INCLUSION_DEFAULT not in self.inclusion
131 and context.delivery_registry.default_inclusion is None
132 ):
133 _LOGGER.info(
134 "SUPERNOTIFY Delivery %s has no explicit inclusion, but transport %s no longer defaults to "
135 "'default' - it will not fire implicitly",
136 self.name,
137 self.transport.name,
138 )
139 context.hass_api.raise_issue(
140 f"delivery_{self.name}_lost_implicit_inclusion",
141 issue_key="delivery_lost_implicit_inclusion",
142 issue_map={"delivery": self.name, "transport": self.transport.name},
143 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/deliveries/",
144 )
145 if self.name in RESERVED_DELIVERY_NAMES:
146 _LOGGER.warning("SUPERNOTIFY Delivery uses reserved word %s", self.name)
147 context.hass_api.raise_issue(
148 f"delivery_{self.name}_reserved_name",
149 issue_key="delivery_reserved_name",
150 issue_map={"delivery": self.name},
151 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/deliveries/",
152 )
153 errors += 1
154 if not self.transport.validate_action(self.action):
155 _LOGGER.warning("SUPERNOTIFY Invalid action definition for delivery %s (%s)", self.name, self.action)
156 context.hass_api.raise_issue(
157 f"delivery_{self.name}_invalid_action",
158 issue_key="delivery_invalid_action",
159 issue_map={"delivery": self.name, "action": self.action or ""},
160 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/deliveries/",
161 )
162 errors += 1
164 if self.conditions_config:
165 try:
166 self.conditions = await context.hass_api.build_conditions(
167 self.conditions_config, validate=True, strict=True, name=self.name
168 )
169 passed = True
170 exception = ""
171 except Exception as e:
172 passed = False
173 exception = str(e)
174 if not passed:
175 _LOGGER.warning("SUPERNOTIFY Invalid delivery conditions for %s: %s", self.name, self.conditions_config)
176 context.hass_api.raise_issue(
177 f"delivery_{self.name}_invalid_condition",
178 issue_key="delivery_invalid_condition",
179 issue_map={"delivery": self.name, "condition": str(self.conditions_config), "exception": exception},
180 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/deliveries/",
181 )
182 errors += 1
184 self.discover_devices(context)
185 self.transport_data = self.transport.setup_delivery_options(self.options, self.name)
186 return errors == 0
188 def upgrade_deprecations(self, conf: ConfigType) -> None:
189 # v1.9.0
190 if (
191 OPTION_DATA_KEYS_INCLUDE_RE in self.options or OPTION_DATA_KEYS_EXCLUDE_RE in self.options
192 ) and not self.options.get(OPTION_DATA_KEYS_SELECT):
193 _LOGGER.warning(
194 "SUPERNOTIFY Deprecated use of data_keys_include_re/data_keys_exclude_re options - use data_keys_select"
195 )
196 self.options[OPTION_DATA_KEYS_SELECT] = {
197 SELECT_INCLUDE: self.options.get(OPTION_DATA_KEYS_INCLUDE_RE),
198 SELECT_EXCLUDE: self.options.get(OPTION_DATA_KEYS_EXCLUDE_RE),
199 }
200 # v1.9.0
201 if OPTION_TARGET_INCLUDE_RE in self.options and not self.options.get(OPTION_TARGET_SELECT):
202 _LOGGER.warning("SUPERNOTIFY Deprecated use of target_include_re option - use target_select")
203 self.options[OPTION_TARGET_SELECT] = {SELECT_INCLUDE: self.options.get(OPTION_TARGET_INCLUDE_RE)}
205 def discover_devices(self, context: Context) -> None:
206 if self.options.get(OPTION_DEVICE_DISCOVERY, False):
207 for domain in self.options.get(OPTION_DEVICE_DOMAIN, []):
208 discovered: int = 0
209 added: int = 0
210 for d in context.hass_api.discover_devices(
211 domain,
212 device_model_select=SelectionRule(self.options.get(OPTION_DEVICE_MODEL_SELECT)),
213 device_manufacturer_select=SelectionRule(self.options.get(OPTION_DEVICE_MANUFACTURER_SELECT)),
214 device_os_select=SelectionRule(self.options.get(OPTION_DEVICE_OS_SELECT)),
215 device_area_select=SelectionRule(self.options.get(OPTION_DEVICE_AREA_SELECT)),
216 device_label_select=SelectionRule(self.options.get(OPTION_DEVICE_LABEL_SELECT)),
217 ):
218 discovered += 1
219 if self.target is None:
220 self.target = Target()
221 if domain == "mobile_app":
222 mobile_app: TrackedDeviceDetails | None = context.hass_api.mobile_app_by_device_id(d.device_id)
223 if mobile_app and mobile_app.action:
224 mobile_app_id = mobile_app.mobile_app_id if mobile_app else None
225 if mobile_app_id and mobile_app_id not in self.target.mobile_app_ids:
226 _LOGGER.debug(
227 f"SUPERNOTIFY Found mobile {d.model} device {d.device_name} for {domain}, id {d.device_id}"
228 )
229 self.target.extend(ATTR_MOBILE_APP_ID, mobile_app_id)
230 added += 1
231 else:
232 _LOGGER.debug(f"SUPERNOTIFY Skipped mobile without notify entity {d.device_name}, id {d.device_id}")
233 else:
234 if d.device_id not in self.target.device_ids:
235 _LOGGER.debug(f"SUPERNOTIFY Found {d.model} device {d.device_name} for {domain}, id {d.device_id}")
236 self.target.extend(ATTR_DEVICE_ID, d.device_id)
237 added += 1
239 _LOGGER.info(f"SUPERNOTIFY {self.name} Device discovery for {domain} found {discovered} devices, added {added}")
241 @property
242 def target_categories(self) -> list[str | TargetEntityCategory]:
243 """The target categories this delivery accepts - the query point for "what does this
245 delivery support", so callers never need to look at `Transport` and `OPTION_TARGET_
246 CATEGORIES` separately. For every transport other than `generic`, this is direct
247 delegation to `self.transport.target_categories` (a delivery rarely needs to widen
248 what its transport understands). `generic` is the bring-your-own-categories
249 transport - it declares nothing itself, so a delivery's own `OPTION_TARGET_CATEGORIES`
250 (e.g. a made-up "slack_channel" category) is what actually defines its categories.
252 The configured list is listed first, so it takes precedence as the reclassification
253 fallback in `reclassify_unqualified_target()` when both are present.
254 """
255 configured = ensure_list(self.options.get(OPTION_TARGET_CATEGORIES))
256 return [*configured, *self.transport.target_categories]
258 def reclassify_unqualified_target(self, target: Target) -> Target:
259 """Reclassify this delivery's uncategorised target values into its primary target
261 category, since a value with no distinguishing shape (e.g. an MQTT topic, a Discord
262 channel ID, or a made-up category for `generic`, like a Slack channel) would
263 otherwise never survive `select_targets()` on its own. Left alone if this delivery's
264 `target_categories` explicitly lists the uncategorised bucket itself - that's a
265 deliberate choice to accept unqualified values exactly as they are - or if it has no
266 plain-string category to fall back to at all.
268 Only safe to call on a `Target` that is exclusively scoped to this one delivery - its
269 own configured `target:`, or a per-delivery override - never on a blended,
270 notification-level target list. There, an unqualified value must stay unclaimed
271 rather than being guessed at: it could belong to a different delivery entirely, and
272 claiming it here would leak it away from wherever it actually belongs.
273 """
274 unqualified = target.targets.get(Target.UNKNOWN_CUSTOM_CATEGORY)
275 if not unqualified:
276 return target
277 declared = self.target_categories
278 if Target.UNKNOWN_CUSTOM_CATEGORY in declared:
279 return target
280 primary = next((c for c in declared if isinstance(c, str)), None)
281 if primary is None:
282 # this target is exclusively scoped to this delivery (the safety precondition
283 # above), so if there's genuinely nowhere for it to go, it's not a value meant
284 # for a different delivery - it's just unmappable, and would otherwise be
285 # dropped with no visible explanation
286 _LOGGER.warning(
287 "SUPERNOTIFY Delivery %s (%s) has no target category to accept unqualified target(s) %s - "
288 "dropping. Known categories for this delivery: %s",
289 self.name,
290 self.transport.name,
291 unqualified,
292 [c if isinstance(c, str) else "entity_id" for c in declared] or "none",
293 )
294 return target
295 result = target.safe_copy()
296 result.targets.pop(Target.UNKNOWN_CUSTOM_CATEGORY, None)
297 result.extend(primary, unqualified)
298 return result
300 def select_targets(self, target: Target) -> Target:
301 return target.select(
302 self.target_categories, (self.name, self.transport.name), self.transport.hass_api, self.target_selector
303 )
305 def evaluate_conditions(self, condition_variables: ConditionVariables) -> bool | None:
306 if not self.enabled:
307 return False
308 if self.conditions is None:
309 return True
310 # TODO: reconsider hass_api injection
311 return self.transport.hass_api.evaluate_conditions(self.conditions, condition_variables)
313 def option(self, option_name: str, default: str | bool) -> str | bool:
314 """Get an option value from delivery config or transport default options"""
315 opt: str | bool | None = None
316 if option_name in self.options:
317 opt = self.options[option_name]
318 if opt is None:
319 _LOGGER.debug(
320 "SUPERNOTIFY No default in delivery %s for option %s, setting to default %s", self.name, option_name, default
321 )
322 opt = self.options[option_name] = default
323 return opt
325 def option_bool(self, option_name: str, default: bool = False) -> bool:
326 return bool(self.option(option_name, default=default))
328 def option_str(self, option_name: str) -> str:
329 return str(self.option(option_name, default=""))
331 def as_dict(self, **_kwargs: Any) -> dict[str, Any]:
332 base = super().as_dict()
333 base.update({
334 CONF_NAME: self.name,
335 CONF_ALIAS: self.alias,
336 CONF_TRANSPORT: self.transport.name,
337 CONF_TEMPLATE: self.template,
338 CONF_MESSAGE: self.message,
339 CONF_TITLE: self.title,
340 CONF_ENABLED: self.enabled,
341 CONF_OCCUPANCY: self.occupancy,
342 CONF_CONDITIONS: self.conditions,
343 })
344 return base
346 def attributes(self) -> dict[str, Any]:
347 """For exposure as entity state"""
348 attrs: dict[str, Any] = {
349 ATTR_NAME: self.name,
350 ATTR_ENABLED: self.enabled,
351 CONF_TRANSPORT: self.transport.name,
352 # a delivery is only used while its transport is enabled too
353 ATTR_TRANSPORT_ENABLED: self.transport.enabled,
354 CONF_ACTION: self.action,
355 CONF_OPTIONS: self.options,
356 CONF_INCLUSION: self.inclusion,
357 CONF_TARGET: self.target,
358 CONF_TARGET_REQUIRED: self.target_required,
359 CONF_TARGET_USAGE: self.target_usage,
360 CONF_DATA: self.data,
361 CONF_DEBUG: self.debug,
362 }
363 if self.alias:
364 attrs[ATTR_FRIENDLY_NAME] = self.alias
365 return attrs
368class DeliveryRegistry:
369 def __init__(
370 self,
371 deliveries: ConfigType | None = None,
372 transport_configs: ConfigType | None = None,
373 transport_types: list[type[Transport]] | dict[type[Transport], dict[str, Any]] | None = None,
374 # for unit tests only
375 transport_instances: list[Transport] | None = None,
376 delivery_control: ConfigType | None = None,
377 ) -> None:
378 # raw configured deliveries
379 self._config_deliveries: ConfigType = deliveries if isinstance(deliveries, dict) else {}
380 # validated deliveries
381 self._deliveries: dict[str, Delivery] = {}
382 self.transports: dict[str, Transport] = {}
383 self._transport_configs: ConfigType = transport_configs or {}
384 # The deprecated delivery and transport binary_sensors, by unique_id - populated by
385 # binary_sensor.py as each is added, only for an existing install that still has them
386 self._entities: dict[str, SupernotifyLegacyBinarySensor] = {}
388 self._transport_types: dict[type[Transport], dict[str, Any]]
389 if isinstance(transport_types, list):
390 self._transport_types = {t: {} for t in transport_types}
391 else:
392 self._transport_types = transport_types or {}
393 # test harness support
394 self._transport_instances: list[Transport] | None = transport_instances
395 # the Delivery Control options, each None when not set, leaving transports' own defaults
396 delivery_control = delivery_control or {}
397 self.default_inclusion: str | None = delivery_control.get(CONF_DEFAULT_INCLUSION)
398 self.voice_occupancy: str | None = delivery_control.get(CONF_VOICE_OCCUPANCY)
400 async def initialize(self, context: Context) -> None:
401 await self.initialize_transports(context)
403 def unload_unused_transports(self) -> None:
404 """Drop any transport that ended up with no delivery at all - explicit or auto-generated.
406 Deliberately deferred until both initialize_transport_deliveries() (explicit) and
407 build_standard_deliveries() (implicit) have run, rather than decided per-transport up
408 front: whether a transport is worth having can only be known once the full, resolved
409 set of deliveries exists - for a transport like `generic` (bring-your-own-action,
410 entirely delivery-driven), there's no transport-level state to check in advance at all.
411 """
412 used_transport_names = {d.transport.name for d in self._deliveries.values()}
413 for name in list(self.transports):
414 if name not in used_transport_names:
415 _LOGGER.info("SUPERNOTIFY %s transport has no deliveries, unloading", name)
416 del self.transports[name]
418 def register_entity(self, unique_id: str, entity: SupernotifyLegacyBinarySensor) -> None:
419 """Called by a delivery or transport binary_sensor's async_added_to_hass()."""
420 self._entities[unique_id] = entity
422 def unregister_entity(self, unique_id: str) -> None:
423 """Called by a delivery or transport binary_sensor's async_will_remove_from_hass()."""
424 self._entities.pop(unique_id, None)
426 def legacy_entities(self) -> list[SupernotifyLegacyBinarySensor]:
427 """Every registered delivery and transport binary_sensor - used by supernotify.refresh_entities."""
428 return list(self._entities.values())
430 @callback
431 def async_refresh_entity(self, unique_id: str) -> None:
432 """Re-publish one delivery or transport binary_sensor now, by its unique_id"""
433 entity = self._entities.get(unique_id)
434 if entity is not None:
435 entity.async_write_ha_state()
437 @property
438 def deliveries(self) -> dict[str, Delivery]:
439 return dict(self._deliveries.items())
441 def resolve_name(self, name: str) -> str:
442 """Backward compatibility for the original 'DEFAULT_x' auto-configured naming,
443 long since replaced by plain transport names: a reference to the old 'DEFAULT_x'
444 form resolves to the current 'x' delivery, if that's what actually exists now."""
445 if name not in self._deliveries and name.startswith("DEFAULT_"):
446 plain_name = name.removeprefix("DEFAULT_")
447 if plain_name in self._deliveries:
448 return plain_name
449 return name
451 @property
452 def enabled_deliveries(self) -> dict[str, Delivery]:
453 return {d: dconf for d, dconf in self._deliveries.items() if dconf.enabled}
455 @property
456 def disabled_deliveries(self) -> dict[str, Delivery]:
457 return {d: dconf for d, dconf in self._deliveries.items() if not dconf.enabled}
459 # Computed on each call, not cached at startup, so that a delivery enabled at runtime (by its
460 # switch) is included just like one enabled in config. delivery.inclusion can also be
461 # INCLUSION_BY_SCENARIO or INCLUSION_EXPLICIT to have it only used where asked for.
463 @property
464 def fallback_by_default_deliveries(self) -> list[Delivery]:
465 return [d for d in self._deliveries.values() if d.enabled and INCLUSION_FALLBACK in d.inclusion]
467 @property
468 def fallback_on_error_deliveries(self) -> list[Delivery]:
469 return [d for d in self._deliveries.values() if d.enabled and INCLUSION_FALLBACK_ON_ERROR in d.inclusion]
471 @property
472 def choosable_deliveries(self) -> dict[str, Delivery]:
473 """Deliveries a notification can ask for - not ones only a scenario turns on, or only a fallback.
474 What the action editor's Delivery list and the LLM tools offer."""
475 return {
476 name: d for name, d in self._deliveries.items() if {INCLUSION_DEFAULT, INCLUSION_EXPLICIT}.intersection(d.inclusion)
477 }
479 @property
480 def implicit_deliveries(self) -> list[Delivery]:
481 """Deliveries switched on all the time via implicit inclusion"""
482 return [d for d in self._deliveries.values() if d.enabled and INCLUSION_DEFAULT in d.inclusion]
484 def apply_delivery_control(self, transport: Transport, transport_config: ConfigType) -> None:
485 """Give a transport's deliveries the Delivery Control defaults, where the transport's own
486 YAML `delivery_defaults` doesn't set them. A delivery's own YAML still wins over both."""
487 own: ConfigType = transport_config.get(CONF_DELIVERY_DEFAULTS) or {}
488 defaults: DeliveryConfig = transport.delivery_defaults
489 if self.default_inclusion and CONF_INCLUSION not in own:
490 defaults.inclusion = [self.default_inclusion]
491 if self.voice_occupancy and CONF_OCCUPANCY not in own and transport.supported_features & TransportFeature.SPOKEN:
492 defaults.occupancy = self.voice_occupancy
494 async def initialize_transports(self, context: Context) -> None:
495 if self._transport_instances:
496 """Used by configure_for_tests() and TestingContext to set transports to mocks or manually created fixtures"""
497 for transport in self._transport_instances:
498 self.transports[transport.name] = transport
499 await transport.initialize()
500 await self.initialize_transport_deliveries(context, transport)
502 if self._transport_types:
503 # production usage
504 for transport_class, kwargs in self._transport_types.items():
505 transport_config: ConfigType = self._transport_configs.get(transport_class.name, {})
506 if not transport_config.get(CONF_LOAD, True):
507 # not just disabled: excluded entirely, so no deliveries or entities either
508 _LOGGER.debug("SUPERNOTIFY %s transport configured not to load", transport_class.name)
509 continue
510 transport = transport_class(context, transport_config, **kwargs)
511 self.apply_delivery_control(transport, transport_config)
512 if not transport.is_viable(context.hass_api):
513 _LOGGER.info("SUPERNOTIFY %s transport has no viable configuration, not loaded", transport_class.name)
514 continue
515 self.transports[transport_class.name] = transport
516 await transport.initialize()
517 await self.initialize_transport_deliveries(context, transport)
518 self.transports[transport_class.name] = transport
520 unconfigured_deliveries = [dc for d, dc in self._config_deliveries.items() if d not in self._deliveries]
521 for bad_del in unconfigured_deliveries:
522 # presumably there was no transport for these
523 context.hass_api.raise_issue(
524 f"delivery_{bad_del.get(CONF_NAME)}_for_transport_{bad_del.get(CONF_TRANSPORT)}_failed_to_configure",
525 issue_key="delivery_unknown_transport",
526 issue_map={
527 "delivery": bad_del.get(CONF_NAME),
528 "transport": bad_del.get(CONF_TRANSPORT),
529 "transports": ", ".join(sorted(self.transports)),
530 },
531 learn_more_url="https://supernotify.rhizomatics.org.uk/configuration/deliveries/",
532 )
534 self.unload_unused_transports()
535 _LOGGER.info("SUPERNOTIFY Configured deliveries %s", "; ".join(self._deliveries.keys()))
537 async def initialize_transport_deliveries(self, context: Context, transport: Transport) -> None:
538 """Validate and initialize deliveries at startup for this transport"""
539 validated_deliveries: dict[str, Delivery] = {}
540 configured_deliveries: dict[str, ConfigType] = {
541 d: dc for d, dc in self._config_deliveries.items() if dc.get(CONF_TRANSPORT) == transport.name
542 }
543 # hackily put here, since build_standard_deliveries can have side-effect of updating default deliveries
544 standard_deliveries: dict[str, ConfigType] = transport.build_standard_deliveries(context.hass_api)
546 for d, dc in configured_deliveries.items():
547 # don't care about ENABLED here since disabled deliveries can be overridden later
548 delivery = Delivery(d, dc, transport, DeliveryProvenance.CONFIG)
549 if not await delivery.initialize(context):
550 _LOGGER.error(f"SUPERNOTIFY Ignoring configured delivery {d} with errors")
551 else:
552 validated_deliveries[d] = delivery
554 # merge in remaining standard deliveries but allow local override
555 for d, dc in standard_deliveries.items():
556 if d in configured_deliveries:
557 _LOGGER.info("SUPERNOTIFY Default standard delivery %s overridden by config", d)
558 else:
559 provenance = DeliveryProvenance.DEFAULT_STANDARD if d == transport.name else DeliveryProvenance.EXTRA_STANDARD
560 delivery = Delivery(d, dc, transport, provenance=provenance)
562 if not await delivery.initialize(context):
563 _LOGGER.error(f"SUPERNOTIFY Ignoring standard delivery {d} with errors")
564 else:
565 validated_deliveries[d] = delivery
567 self._deliveries.update(validated_deliveries)
569 _LOGGER.debug(
570 "SUPERNOTIFY Validated transport %s, default action %s, valid deliveries: %s",
571 transport.name,
572 transport.delivery_defaults.action,
573 [d for d in self._deliveries.values() if d.enabled and d.transport == transport],
574 )