Coverage for custom_components/supernotify/delivery.py: 18%
228 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-02-06 15:56 +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 .const 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)
56from .context import Context
58if TYPE_CHECKING:
59 from custom_components.supernotify.hass_api import DeviceInfo
61 from .schema import ConditionsFunc
63_LOGGER = logging.getLogger(__name__)
66class Delivery(DeliveryConfig):
67 def __init__(self, name: str, conf: ConfigType, transport: "Transport") -> None:
68 conf = conf or {}
69 self.name: str = name
70 self.alias: str | None = conf.get(CONF_ALIAS)
71 self.transport: Transport = transport
72 transport_defaults: DeliveryConfig = self.transport.delivery_defaults
73 super().__init__(conf, delivery_defaults=transport_defaults)
74 self.template: str | None = conf.get(CONF_TEMPLATE)
75 self.message: str | None = conf.get(CONF_MESSAGE)
76 self.title: str | None = conf.get(CONF_TITLE)
77 self.enabled: bool = conf.get(CONF_ENABLED, self.transport.enabled)
78 self.occupancy: str = conf.get(CONF_OCCUPANCY, OCCUPANCY_ALL)
79 self.conditions_config: list[ConfigType] | None = conf.get(CONF_CONDITIONS)
80 if not conf.get(CONF_CONDITIONS) and conf.get(CONF_CONDITION):
81 self.conditions_config = conf.get(CONF_CONDITION)
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()
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) -> 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)
308 def enable(self, delivery_name: str) -> bool:
309 delivery = self._deliveries.get(delivery_name)
310 if delivery and not delivery.enabled:
311 _LOGGER.info(f"SUPERNOTIFY Enabling delivery {delivery_name}")
312 delivery.enabled = True
313 return True
314 return False
316 def disable(self, delivery_name: str) -> bool:
317 delivery = self._deliveries.get(delivery_name)
318 if delivery and delivery.enabled:
319 _LOGGER.info(f"SUPERNOTIFY Disabling delivery {delivery_name}")
320 delivery.enabled = False
321 return True
322 return False
324 @property
325 def deliveries(self) -> dict[str, Delivery]:
326 return dict(self._deliveries.items())
328 @property
329 def enabled_deliveries(self) -> dict[str, Delivery]:
330 return {d: dconf for d, dconf in self._deliveries.items() if dconf.enabled}
332 @property
333 def disabled_deliveries(self) -> dict[str, Delivery]:
334 return {d: dconf for d, dconf in self._deliveries.items() if not dconf.enabled}
336 @property
337 def fallback_by_default_deliveries(self) -> list[Delivery]:
338 return [d for d in self._fallback_by_default if d.enabled]
340 @property
341 def fallback_on_error_deliveries(self) -> list[Delivery]:
342 return [d for d in self._fallback_on_error if d.enabled]
344 @property
345 def implicit_deliveries(self) -> list[Delivery]:
346 """Deliveries switched on all the time for implicit selection"""
347 return [d for d in self._implicit_deliveries if d.enabled]
349 async def initialize_transports(self, context: "Context") -> None:
350 """Use configure_for_tests() to set transports to mocks or manually created fixtures"""
351 if self._transport_instances:
352 for transport in self._transport_instances:
353 self.transports[transport.name] = transport
354 await transport.initialize()
355 await self.initialize_transport_deliveries(context, transport)
356 if self._transport_types:
357 for transport_class, kwargs in self._transport_types.items():
358 transport_config: ConfigType = self._transport_configs.get(transport_class.name, {})
359 transport = transport_class(context, transport_config, **kwargs)
360 self.transports[transport_class.name] = transport
361 await transport.initialize()
362 await self.initialize_transport_deliveries(context, transport)
363 self.transports[transport_class.name] = transport
365 unconfigured_deliveries = [dc for d, dc in self._config_deliveries.items() if d not in self._deliveries]
366 for bad_del in unconfigured_deliveries:
367 # presumably there was no transport for these
368 context.hass_api.raise_issue(
369 f"delivery_{bad_del.get(CONF_NAME)}_for_transport_{bad_del.get(CONF_TRANSPORT)}_failed_to_configure",
370 issue_key="delivery_unknown_transport",
371 issue_map={"delivery": bad_del.get(CONF_NAME), "transport": bad_del.get(CONF_TRANSPORT)},
372 learn_more_url="https://supernotify.rhizomatics.org.uk/deliveries",
373 )
374 _LOGGER.info("SUPERNOTIFY configured deliveries %s", "; ".join(self._deliveries.keys()))
376 async def initialize_transport_deliveries(self, context: Context, transport: Transport) -> None:
377 """Validate and initialize deliveries at startup for this transport"""
378 validated_deliveries: dict[str, Delivery] = {}
379 deliveries_for_this_transport = {
380 d: dc for d, dc in self._config_deliveries.items() if dc.get(CONF_TRANSPORT) == transport.name
381 }
382 for d, dc in deliveries_for_this_transport.items():
383 # don't care about ENABLED here since disabled deliveries can be overridden later
384 delivery = Delivery(d, dc, transport)
385 if not await delivery.initialize(context):
386 _LOGGER.error(f"SUPERNOTIFY Ignoring delivery {d} with errors")
387 else:
388 validated_deliveries[d] = delivery
390 self._deliveries.update(validated_deliveries)
392 _LOGGER.debug(
393 "SUPERNOTIFY Validated transport %s, default action %s, valid deliveries: %s",
394 transport.name,
395 transport.delivery_defaults.action,
396 [d for d in self._deliveries.values() if d.enabled and d.transport == transport],
397 )
399 async def autogenerate_deliveries(self, context: "Context") -> None:
400 # If the config has no deliveries, check if a default delivery should be auto-generated
401 # where there is a empty config, supernotify can at least handle NotifyEntities sensibly
403 autogenerated: dict[str, Delivery] = {}
404 for transport in [t for t in self.transports.values() if t.enabled]:
405 if any(dc for dc in self._config_deliveries.values() if dc.get(CONF_TRANSPORT) == transport.name):
406 # don't auto-configure if there's an explicit delivery configured for this transport
407 continue
409 transport_definition: DeliveryConfig | None = transport.auto_configure(context.hass_api)
410 if transport_definition:
411 _LOGGER.debug(
412 "SUPERNOTIFY Building default delivery for %s from transport %s", transport.name, transport_definition
413 )
414 # belt and braces transport checking its own discovery
415 if transport.validate_action(transport_definition.action):
416 # auto generate a delivery that will be implicitly selected
417 default_delivery = Delivery(f"DEFAULT_{transport.name}", transport_definition.as_dict(), transport)
418 await default_delivery.initialize(context)
419 default_delivery.enabled = transport.enabled
420 autogenerated[default_delivery.name] = default_delivery
421 _LOGGER.info(
422 "SUPERNOTIFY Auto-generating a default delivery for %s from transport %s",
423 transport.name,
424 transport_definition,
425 )
426 else:
427 _LOGGER.debug("SUPERNOTIFY No default delivery or transport_definition for transport %s", transport.name)
428 if autogenerated:
429 self._deliveries.update(autogenerated)