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