Coverage for custom_components/supernotify/delivery.py: 97%
224 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-01 18:25 +0000
1from __future__ import annotations
3import logging
4from typing import TYPE_CHECKING, Any
6from homeassistant.const import (
7 ATTR_DEVICE_ID,
8 ATTR_FRIENDLY_NAME,
9 ATTR_NAME,
10 CONF_ACTION,
11 CONF_ALIAS,
12 CONF_CONDITIONS,
13 CONF_DEBUG,
14 CONF_ENABLED,
15 CONF_NAME,
16 CONF_OPTIONS,
17 CONF_TARGET,
18)
20from custom_components.supernotify.model import ConditionVariables, DeliveryConfig, SelectionRule, Target
22from .const import (
23 ATTR_ENABLED,
24 ATTR_MOBILE_APP_ID,
25 CONF_DATA,
26 CONF_MESSAGE,
27 CONF_OCCUPANCY,
28 CONF_SELECTION,
29 CONF_TARGET_REQUIRED,
30 CONF_TARGET_USAGE,
31 CONF_TEMPLATE,
32 CONF_TITLE,
33 CONF_TRANSPORT,
34 OCCUPANCY_ALL,
35 OPTION_DATA_KEYS_EXCLUDE_RE,
36 OPTION_DATA_KEYS_INCLUDE_RE,
37 OPTION_DATA_KEYS_SELECT,
38 OPTION_DEVICE_AREA_SELECT,
39 OPTION_DEVICE_DISCOVERY,
40 OPTION_DEVICE_DOMAIN,
41 OPTION_DEVICE_LABEL_SELECT,
42 OPTION_DEVICE_MANUFACTURER_SELECT,
43 OPTION_DEVICE_MODEL_SELECT,
44 OPTION_DEVICE_OS_SELECT,
45 OPTION_TARGET_CATEGORIES,
46 OPTION_TARGET_INCLUDE_RE,
47 OPTION_TARGET_SELECT,
48 RESERVED_DELIVERY_NAMES,
49 SELECT_EXCLUDE,
50 SELECT_INCLUDE,
51 SELECTION_DEFAULT,
52 SELECTION_FALLBACK,
53 SELECTION_FALLBACK_ON_ERROR,
54)
56if TYPE_CHECKING:
57 from homeassistant.helpers.typing import ConfigType
59 from custom_components.supernotify.hass_api import DeviceInfo
60 from custom_components.supernotify.transport import Transport
62 from .context import Context
63 from .schema import ConditionsFunc
65_LOGGER = logging.getLogger(__name__)
68class Delivery(DeliveryConfig):
69 def __init__(self, name: str, conf: ConfigType, transport: Transport) -> None:
70 conf = conf or {}
71 self.name: str = name
72 self.alias: str | None = conf.get(CONF_ALIAS)
73 self.transport: Transport = transport
74 transport_defaults: DeliveryConfig = self.transport.delivery_defaults
75 super().__init__(conf, delivery_defaults=transport_defaults)
76 self.template: str | None = conf.get(CONF_TEMPLATE)
77 self.message: str | None = conf.get(CONF_MESSAGE)
78 self.title: str | None = conf.get(CONF_TITLE)
79 self.enabled: bool = conf.get(CONF_ENABLED, self.transport.enabled)
80 self.occupancy: str = conf.get(CONF_OCCUPANCY, OCCUPANCY_ALL)
81 self.conditions_config: list[ConfigType] | None = conf.get(CONF_CONDITIONS)
82 self.conditions: ConditionsFunc | None = None
83 self.transport_data: dict[str, Any] = {}
84 if self.options.get(OPTION_TARGET_SELECT):
85 self.target_selector: SelectionRule | None = SelectionRule(self.options.get(OPTION_TARGET_SELECT))
86 else:
87 self.target_selector = None
88 self.upgrade_deprecations(conf)
90 async def initialize(self, context: Context) -> bool:
91 errors = 0
92 if self.name in RESERVED_DELIVERY_NAMES:
93 _LOGGER.warning("SUPERNOTIFY Delivery uses reserved word %s", self.name)
94 context.hass_api.raise_issue(
95 f"delivery_{self.name}_reserved_name",
96 issue_key="delivery_reserved_name",
97 issue_map={"delivery": self.name},
98 learn_more_url="https://supernotify.rhizomatics.org.uk/deliveries",
99 )
100 errors += 1
101 if not self.transport.validate_action(self.action):
102 _LOGGER.warning("SUPERNOTIFY Invalid action definition for delivery %s (%s)", self.name, self.action)
103 context.hass_api.raise_issue(
104 f"delivery_{self.name}_invalid_action",
105 issue_key="delivery_invalid_action",
106 issue_map={"delivery": self.name, "action": self.action or ""},
107 learn_more_url="https://supernotify.rhizomatics.org.uk/deliveries",
108 )
109 errors += 1
111 if self.conditions_config:
112 try:
113 self.conditions = await context.hass_api.build_conditions(
114 self.conditions_config, validate=True, strict=True, name=self.name
115 )
116 passed = True
117 exception = ""
118 except Exception as e:
119 passed = False
120 exception = str(e)
121 if not passed:
122 _LOGGER.warning("SUPERNOTIFY Invalid delivery conditions for %s: %s", self.name, self.conditions_config)
123 context.hass_api.raise_issue(
124 f"delivery_{self.name}_invalid_condition",
125 issue_key="delivery_invalid_condition",
126 issue_map={"delivery": self.name, "condition": str(self.conditions_config), "exception": exception},
127 learn_more_url="https://supernotify.rhizomatics.org.uk/deliveries",
128 )
129 errors += 1
131 self.discover_devices(context)
132 self.transport_data = self.transport.setup_delivery_options(self.options, self.name)
133 return errors == 0
135 def upgrade_deprecations(self, conf: ConfigType) -> None:
136 # v1.9.0
137 if (
138 OPTION_DATA_KEYS_INCLUDE_RE in self.options or OPTION_DATA_KEYS_EXCLUDE_RE in self.options
139 ) and not self.options.get(OPTION_DATA_KEYS_SELECT):
140 _LOGGER.warning(
141 "SUPERNOTIFY Deprecated use of data_keys_include_re/data_keys_exclude_re options - use data_keys_select"
142 )
143 self.options[OPTION_DATA_KEYS_SELECT] = {
144 SELECT_INCLUDE: self.options.get(OPTION_DATA_KEYS_INCLUDE_RE),
145 SELECT_EXCLUDE: self.options.get(OPTION_DATA_KEYS_EXCLUDE_RE),
146 }
147 # v1.9.0
148 if OPTION_TARGET_INCLUDE_RE in self.options and not self.options.get(OPTION_TARGET_SELECT):
149 _LOGGER.warning("SUPERNOTIFY Deprecated use of target_include_re option - use target_select")
150 self.options[OPTION_TARGET_SELECT] = {SELECT_INCLUDE: self.options.get(OPTION_TARGET_INCLUDE_RE)}
152 def discover_devices(self, context: Context) -> None:
153 if self.options.get(OPTION_DEVICE_DISCOVERY, False):
154 for domain in self.options.get(OPTION_DEVICE_DOMAIN, []):
155 discovered: int = 0
156 added: int = 0
157 for d in context.hass_api.discover_devices(
158 domain,
159 device_model_select=SelectionRule(self.options.get(OPTION_DEVICE_MODEL_SELECT)),
160 device_manufacturer_select=SelectionRule(self.options.get(OPTION_DEVICE_MANUFACTURER_SELECT)),
161 device_os_select=SelectionRule(self.options.get(OPTION_DEVICE_OS_SELECT)),
162 device_area_select=SelectionRule(self.options.get(OPTION_DEVICE_AREA_SELECT)),
163 device_label_select=SelectionRule(self.options.get(OPTION_DEVICE_LABEL_SELECT)),
164 ):
165 discovered += 1
166 if self.target is None:
167 self.target = Target()
168 if domain == "mobile_app":
169 mobile_app: DeviceInfo | None = context.hass_api.mobile_app_by_device_id(d.device_id)
170 if mobile_app and mobile_app.action:
171 mobile_app_id = mobile_app.mobile_app_id if mobile_app else None
172 if mobile_app_id and mobile_app_id not in self.target.mobile_app_ids:
173 _LOGGER.debug(
174 f"SUPERNOTIFY Found mobile {d.model} device {d.device_name} for {domain}, id {d.device_id}"
175 )
176 self.target.extend(ATTR_MOBILE_APP_ID, mobile_app_id)
177 added += 1
178 else:
179 _LOGGER.debug(f"SUPERNOTIFY Skipped mobile without notify entity {d.device_name}, id {d.device_id}")
180 else:
181 if d.device_id not in self.target.device_ids:
182 _LOGGER.debug(f"SUPERNOTIFY Found {d.model} device {d.device_name} for {domain}, id {d.device_id}")
183 self.target.extend(ATTR_DEVICE_ID, d.device_id)
184 added += 1
186 _LOGGER.info(f"SUPERNOTIFY {self.name} Device discovery for {domain} found {discovered} devices, added {added}")
188 def select_targets(self, target: Target) -> Target:
189 def selected(category: str, targets: list[str]) -> list[str]:
190 if OPTION_TARGET_CATEGORIES in self.options and category not in self.options[OPTION_TARGET_CATEGORIES]:
191 return []
192 if self.target_selector:
193 return [t for t in targets if self.target_selector.match(t)]
194 return targets
196 filtered_target = Target({k: selected(k, v) for k, v in target.targets.items()}, target_data=target.target_data)
197 # TODO: in model class
198 if target.target_specific_data:
199 filtered_target.target_specific_data = {
200 (c, t): data
201 for (c, t), data in target.target_specific_data.items()
202 if c in target.targets and t in target.targets[c]
203 }
204 return filtered_target
206 def evaluate_conditions(self, condition_variables: ConditionVariables) -> bool | None:
207 if not self.enabled:
208 return False
209 if self.conditions is None:
210 return True
211 # TODO: reconsider hass_api injection
212 return self.transport.hass_api.evaluate_conditions(self.conditions, condition_variables)
214 def option(self, option_name: str, default: str | bool) -> str | bool:
215 """Get an option value from delivery config or transport default options"""
216 opt: str | bool | None = None
217 if option_name in self.options:
218 opt = self.options[option_name]
219 if opt is None:
220 _LOGGER.debug(
221 "SUPERNOTIFY No default in delivery %s for option %s, setting to default %s", self.name, option_name, default
222 )
223 opt = default
224 return opt
226 def option_bool(self, option_name: str, default: bool = False) -> bool:
227 return bool(self.option(option_name, default=default))
229 def option_str(self, option_name: str) -> str:
230 return str(self.option(option_name, default=""))
232 def as_dict(self, **_kwargs: Any) -> dict[str, Any]:
233 base = super().as_dict()
234 base.update({
235 CONF_NAME: self.name,
236 CONF_ALIAS: self.alias,
237 CONF_TRANSPORT: self.transport.name,
238 CONF_TEMPLATE: self.template,
239 CONF_MESSAGE: self.message,
240 CONF_TITLE: self.title,
241 CONF_ENABLED: self.enabled,
242 CONF_OCCUPANCY: self.occupancy,
243 CONF_CONDITIONS: self.conditions,
244 })
245 return base
247 def attributes(self) -> dict[str, Any]:
248 """For exposure as entity state"""
249 attrs: dict[str, Any] = {
250 ATTR_NAME: self.name,
251 ATTR_ENABLED: self.enabled,
252 CONF_TRANSPORT: self.transport.name,
253 CONF_ACTION: self.action,
254 CONF_OPTIONS: self.options,
255 CONF_SELECTION: self.selection,
256 CONF_TARGET: self.target,
257 CONF_TARGET_REQUIRED: self.target_required,
258 CONF_TARGET_USAGE: self.target_usage,
259 CONF_DATA: self.data,
260 CONF_DEBUG: self.debug,
261 }
262 if self.alias:
263 attrs[ATTR_FRIENDLY_NAME] = self.alias
264 return attrs
267class DeliveryRegistry:
268 def __init__(
269 self,
270 deliveries: ConfigType | None = None,
271 transport_configs: ConfigType | None = None,
272 transport_types: list[type[Transport]] | dict[type[Transport], dict[str, Any]] | None = None,
273 # for unit tests only
274 transport_instances: list[Transport] | None = None,
275 ) -> None:
276 # raw configured deliveries
277 self._config_deliveries: ConfigType = deliveries if isinstance(deliveries, dict) else {}
278 # validated deliveries
279 self._deliveries: dict[str, Delivery] = {}
280 self.transports: dict[str, Transport] = {}
281 self._transport_configs: ConfigType = transport_configs or {}
282 self._fallback_on_error: list[Delivery] = []
283 self._fallback_by_default: list[Delivery] = []
284 self._implicit_deliveries: list[Delivery] = []
285 # test harness support
286 self._transport_types: dict[type[Transport], dict[str, Any]]
287 if isinstance(transport_types, list):
288 self._transport_types = {t: {} for t in transport_types}
289 else:
290 self._transport_types = transport_types or {}
291 self._transport_instances: list[Transport] | None = transport_instances
293 async def initialize(self, context: Context) -> None:
294 await self.initialize_transports(context)
295 await self.autogenerate_deliveries(context)
296 self.initialize_deliveries()
298 def initialize_deliveries(self) -> None:
299 for delivery in self._deliveries.values():
300 if delivery.enabled:
301 if SELECTION_FALLBACK_ON_ERROR in delivery.selection:
302 self._fallback_on_error.append(delivery)
303 if SELECTION_FALLBACK in delivery.selection:
304 self._fallback_by_default.append(delivery)
305 if SELECTION_DEFAULT in delivery.selection:
306 self._implicit_deliveries.append(delivery)
307 # delivery.selection can also be SELECTION_BY_SCENARIO
308 # or SELECTION_EXPLICIT to have it only used where asked for
309 # default is SELECTION_DEFAULT so every delivery used implicitly
311 def enable(self, delivery_name: str) -> bool:
312 delivery = self._deliveries.get(delivery_name)
313 if delivery and not delivery.enabled:
314 _LOGGER.info(f"SUPERNOTIFY Enabling delivery {delivery_name}")
315 delivery.enabled = True
316 return True
317 return False
319 def disable(self, delivery_name: str) -> bool:
320 delivery = self._deliveries.get(delivery_name)
321 if delivery and delivery.enabled:
322 _LOGGER.info(f"SUPERNOTIFY Disabling delivery {delivery_name}")
323 delivery.enabled = False
324 return True
325 return False
327 @property
328 def deliveries(self) -> dict[str, Delivery]:
329 return dict(self._deliveries.items())
331 @property
332 def enabled_deliveries(self) -> dict[str, Delivery]:
333 return {d: dconf for d, dconf in self._deliveries.items() if dconf.enabled}
335 @property
336 def disabled_deliveries(self) -> dict[str, Delivery]:
337 return {d: dconf for d, dconf in self._deliveries.items() if not dconf.enabled}
339 @property
340 def fallback_by_default_deliveries(self) -> list[Delivery]:
341 return [d for d in self._fallback_by_default if d.enabled]
343 @property
344 def fallback_on_error_deliveries(self) -> list[Delivery]:
345 return [d for d in self._fallback_on_error if d.enabled]
347 @property
348 def implicit_deliveries(self) -> list[Delivery]:
349 """Deliveries switched on all the time for implicit selection"""
350 return [d for d in self._implicit_deliveries if d.enabled]
352 async def initialize_transports(self, context: Context) -> None:
353 """Use configure_for_tests() to set transports to mocks or manually created fixtures"""
354 if self._transport_instances:
355 for transport in self._transport_instances:
356 self.transports[transport.name] = transport
357 await transport.initialize()
358 await self.initialize_transport_deliveries(context, transport)
359 if self._transport_types:
360 for transport_class, kwargs in self._transport_types.items():
361 transport_config: ConfigType = self._transport_configs.get(transport_class.name, {})
362 transport = transport_class(context, transport_config, **kwargs)
363 self.transports[transport_class.name] = transport
364 await transport.initialize()
365 await self.initialize_transport_deliveries(context, transport)
366 self.transports[transport_class.name] = transport
368 unconfigured_deliveries = [dc for d, dc in self._config_deliveries.items() if d not in self._deliveries]
369 for bad_del in unconfigured_deliveries:
370 # presumably there was no transport for these
371 context.hass_api.raise_issue(
372 f"delivery_{bad_del.get(CONF_NAME)}_for_transport_{bad_del.get(CONF_TRANSPORT)}_failed_to_configure",
373 issue_key="delivery_unknown_transport",
374 issue_map={"delivery": bad_del.get(CONF_NAME), "transport": bad_del.get(CONF_TRANSPORT)},
375 learn_more_url="https://supernotify.rhizomatics.org.uk/deliveries",
376 )
377 _LOGGER.info("SUPERNOTIFY Configured deliveries %s", "; ".join(self._deliveries.keys()))
379 async def initialize_transport_deliveries(self, context: Context, transport: Transport) -> None:
380 """Validate and initialize deliveries at startup for this transport"""
381 validated_deliveries: dict[str, Delivery] = {}
382 deliveries_for_this_transport = {
383 d: dc for d, dc in self._config_deliveries.items() if dc.get(CONF_TRANSPORT) == transport.name
384 }
385 for d, dc in deliveries_for_this_transport.items():
386 # don't care about ENABLED here since disabled deliveries can be overridden later
387 delivery = Delivery(d, dc, transport)
388 if not await delivery.initialize(context):
389 _LOGGER.error(f"SUPERNOTIFY Ignoring delivery {d} with errors")
390 else:
391 validated_deliveries[d] = delivery
393 self._deliveries.update(validated_deliveries)
395 _LOGGER.debug(
396 "SUPERNOTIFY Validated transport %s, default action %s, valid deliveries: %s",
397 transport.name,
398 transport.delivery_defaults.action,
399 [d for d in self._deliveries.values() if d.enabled and d.transport == transport],
400 )
402 async def autogenerate_deliveries(self, context: Context) -> None:
403 # If the config has no deliveries, check if a default delivery should be auto-generated
404 # where there is a empty config, supernotify can at least handle NotifyEntities sensibly
406 autogenerated: dict[str, Delivery] = {}
407 for transport in [t for t in self.transports.values() if t.enabled]:
408 if any(dc.get(CONF_TRANSPORT) == transport.name for dc in self._config_deliveries.values()):
409 # don't auto-configure if there's an explicit delivery configured for this transport
410 continue
412 transport_definition: DeliveryConfig | None = transport.auto_configure(context.hass_api)
413 if transport_definition:
414 _LOGGER.debug(
415 "SUPERNOTIFY Building default delivery for %s from transport %s", transport.name, transport_definition
416 )
417 # belt and braces transport checking its own discovery
418 if transport.validate_action(transport_definition.action):
419 # auto generate a delivery that will be implicitly selected
420 default_delivery = Delivery(f"DEFAULT_{transport.name}", transport_definition.as_dict(), transport)
421 await default_delivery.initialize(context)
422 default_delivery.enabled = transport.enabled
423 autogenerated[default_delivery.name] = default_delivery
424 _LOGGER.info(
425 "SUPERNOTIFY Auto-generating a default delivery for %s from transport %s",
426 transport.name,
427 transport_definition,
428 )
429 else:
430 _LOGGER.debug("SUPERNOTIFY No default delivery or transport_definition for transport %s", transport.name)
431 if autogenerated:
432 self._deliveries.update(autogenerated)