Coverage for custom_components/supernotify/notification.py: 89%
416 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-10-26 08:54 +0000
1import asyncio
2import datetime as dt
3import logging
4import uuid
5from pathlib import Path
6from traceback import format_exception
7from typing import Any
9import voluptuous as vol
10from homeassistant.components.notify.const import ATTR_DATA, ATTR_TARGET
11from homeassistant.const import CONF_ENABLED, CONF_NAME, CONF_TARGET, STATE_HOME, STATE_NOT_HOME
12from homeassistant.helpers.template import Template
13from jinja2 import TemplateError
14from voluptuous import humanize
16from custom_components.supernotify import (
17 ACTION_DATA_SCHEMA,
18 ATTR_ACTION_GROUPS,
19 ATTR_ACTIONS,
20 ATTR_DEBUG,
21 ATTR_DELIVERY,
22 ATTR_DELIVERY_SELECTION,
23 ATTR_JPEG_OPTS,
24 ATTR_MEDIA,
25 ATTR_MEDIA_CAMERA_DELAY,
26 ATTR_MEDIA_CAMERA_ENTITY_ID,
27 ATTR_MEDIA_CAMERA_PTZ_PRESET,
28 ATTR_MEDIA_CLIP_URL,
29 ATTR_MEDIA_SNAPSHOT_URL,
30 ATTR_MESSAGE_HTML,
31 ATTR_PRIORITY,
32 ATTR_RECIPIENTS,
33 ATTR_SCENARIOS_APPLY,
34 ATTR_SCENARIOS_CONSTRAIN,
35 ATTR_SCENARIOS_REQUIRE,
36 CONF_DATA,
37 CONF_DELIVERY,
38 CONF_MESSAGE,
39 CONF_OCCUPANCY,
40 CONF_OPTIONS,
41 CONF_PERSON,
42 CONF_PRIORITY,
43 CONF_PTZ_DELAY,
44 CONF_PTZ_METHOD,
45 CONF_PTZ_PRESET_DEFAULT,
46 CONF_RECIPIENTS,
47 CONF_SELECTION,
48 CONF_TITLE,
49 DELIVERY_SELECTION_EXPLICIT,
50 DELIVERY_SELECTION_FIXED,
51 DELIVERY_SELECTION_IMPLICIT,
52 OCCUPANCY_ALL,
53 OCCUPANCY_ALL_IN,
54 OCCUPANCY_ALL_OUT,
55 OCCUPANCY_ANY_IN,
56 OCCUPANCY_ANY_OUT,
57 OCCUPANCY_NONE,
58 OCCUPANCY_ONLY_IN,
59 OCCUPANCY_ONLY_OUT,
60 PRIORITY_MEDIUM,
61 PRIORITY_VALUES,
62 SCENARIO_DEFAULT,
63 SCENARIO_NULL,
64 SELECTION_BY_SCENARIO,
65 STRICT_ACTION_DATA_SCHEMA,
66 ConditionVariables,
67 MessageOnlyPolicy,
68)
69from custom_components.supernotify.archive import ArchivableObject
70from custom_components.supernotify.common import DebugTrace, safe_extend
71from custom_components.supernotify.delivery_method import (
72 OPTION_MESSAGE_USAGE,
73 OPTION_SIMPLIFY_TEXT,
74 OPTION_STRIP_URLS,
75 DeliveryMethod,
76)
77from custom_components.supernotify.envelope import Envelope
78from custom_components.supernotify.scenario import Scenario
80from .common import ensure_dict, ensure_list
81from .configuration import Context
82from .media_grab import move_camera_to_ptz_preset, select_avail_camera, snap_camera, snap_image, snapshot_from_url
84_LOGGER = logging.getLogger(__name__)
87class Notification(ArchivableObject):
88 def __init__(
89 self,
90 context: Context,
91 message: str | None = None,
92 title: str | None = None,
93 target: list[str] | str | None = None,
94 action_data: dict[str, Any] | None = None,
95 ) -> None:
96 self.created: dt.datetime = dt.datetime.now(tz=dt.UTC)
97 self.debug_trace: DebugTrace = DebugTrace(message=message, title=title, data=action_data, target=target)
98 self._message: str | None = message
99 self.context: Context = context
100 action_data = action_data or {}
101 self.target: list[str] = ensure_list(target)
102 self._title: str | None = title
103 self.id = str(uuid.uuid1())
104 self.snapshot_image_path: Path | None = None
105 self.delivered: int = 0
106 self.errored: int = 0
107 self.skipped: int = 0
108 self.delivered_envelopes: list[Envelope] = []
109 self.undelivered_envelopes: list[Envelope] = []
110 self.delivery_error: list[str] | None = None
112 self.validate_action_data(action_data)
113 # for compatibility with other notify calls, pass thru surplus data to underlying delivery methods
114 self.data: dict[str, Any] = {k: v for k, v in action_data.items() if k not in STRICT_ACTION_DATA_SCHEMA(action_data)}
115 action_data = {k: v for k, v in action_data.items() if k not in self.data}
117 self.priority: str = action_data.get(ATTR_PRIORITY, PRIORITY_MEDIUM)
118 self.message_html: str | None = action_data.get(ATTR_MESSAGE_HTML)
119 self.required_scenario_names: list[str] = ensure_list(action_data.get(ATTR_SCENARIOS_REQUIRE))
120 self.applied_scenario_names: list[str] = ensure_list(action_data.get(ATTR_SCENARIOS_APPLY))
121 self.constrain_scenario_names: list[str] = ensure_list(action_data.get(ATTR_SCENARIOS_CONSTRAIN))
122 self.delivery_selection: str | None = action_data.get(ATTR_DELIVERY_SELECTION)
123 self.delivery_overrides_type: str = action_data.get(ATTR_DELIVERY).__class__.__name__
124 self.delivery_overrides: dict[str, Any] = ensure_dict(action_data.get(ATTR_DELIVERY))
125 self.action_groups: list[str] | None = action_data.get(ATTR_ACTION_GROUPS)
126 self.recipients_override: list[str] | None = action_data.get(ATTR_RECIPIENTS)
127 self.data.update(action_data.get(ATTR_DATA, {}))
128 self.media: dict[str, Any] = action_data.get(ATTR_MEDIA) or {}
129 self.debug: bool = action_data.get(ATTR_DEBUG, False)
130 self.actions: dict[str, Any] = action_data.get(ATTR_ACTIONS) or {}
131 self.delivery_results: dict[str, Any] = {}
132 self.delivery_errors: dict[str, Any] = {}
134 self.selected_delivery_names: list[str] = []
135 self.enabled_scenarios: dict[str, Scenario] = {}
136 self.selected_scenario_names: list[str] = []
137 self.people_by_occupancy: list[dict[str, Any]] = []
138 self.globally_disabled: bool = False
139 self.occupancy: dict[str, list[dict[str, Any]]] = {}
140 self.condition_variables: ConditionVariables | None = None
142 async def initialize(self) -> None:
143 """Async post-construction initialization"""
144 if self.delivery_selection is None:
145 if self.delivery_overrides_type in ("list", "str"):
146 # a bare list of deliveries implies intent to restrict
147 _LOGGER.debug("SUPERNOTIFY defaulting delivery selection as explicit for type %s", self.delivery_overrides_type)
148 self.delivery_selection = DELIVERY_SELECTION_EXPLICIT
149 else:
150 # whereas a dict may be used to tune or restrict
151 self.delivery_selection = DELIVERY_SELECTION_IMPLICIT
152 _LOGGER.debug("SUPERNOTIFY defaulting delivery selection as implicit for type %s", self.delivery_overrides_type)
154 self.occupancy = self.context.determine_occupancy()
155 self.condition_variables = ConditionVariables(
156 self.applied_scenario_names,
157 self.required_scenario_names,
158 self.constrain_scenario_names,
159 self.priority,
160 self.occupancy,
161 self._message,
162 self._title,
163 ) # requires occupancy first
165 enabled_scenario_names: list[str] = list(self.applied_scenario_names) or []
166 self.selected_scenario_names = await self.select_scenarios()
167 enabled_scenario_names.extend(self.selected_scenario_names)
168 if self.constrain_scenario_names:
169 enabled_scenario_names = [
170 s
171 for s in enabled_scenario_names
172 if (s in self.constrain_scenario_names or s in self.applied_scenario_names) and s != SCENARIO_NULL
173 ]
174 if self.required_scenario_names and not any(s in enabled_scenario_names for s in self.required_scenario_names):
175 _LOGGER.info("SUPERNOTIFY suppressing notification, no required scenarios enabled")
176 self.selected_delivery_names = []
177 self.globally_disabled = True
178 else:
179 for s in enabled_scenario_names:
180 scenario_obj = self.context.scenarios.get(s)
181 if scenario_obj is not None:
182 self.enabled_scenarios[s] = scenario_obj
184 self.selected_delivery_names = self.select_deliveries()
185 self.globally_disabled = self.context.snoozer.is_global_snooze(self.priority)
186 self.default_media_from_actions()
187 self.apply_enabled_scenarios()
189 def validate_action_data(self, action_data: dict[str, Any]) -> None:
190 if action_data.get(ATTR_PRIORITY) and action_data.get(ATTR_PRIORITY) not in PRIORITY_VALUES:
191 _LOGGER.warning("SUPERNOTIFY invalid priority %s - overriding to medium", action_data.get(ATTR_PRIORITY))
192 action_data[ATTR_PRIORITY] = PRIORITY_MEDIUM
193 try:
194 humanize.validate_with_humanized_errors(action_data, ACTION_DATA_SCHEMA)
195 except vol.Invalid as e:
196 _LOGGER.warning("SUPERNOTIFY invalid service data %s: %s", action_data, e)
197 raise
199 def apply_enabled_scenarios(self) -> None:
200 """Set media and action_groups from scenario if defined, first come first applied"""
201 action_groups: list[str] = []
202 for scen_obj in self.enabled_scenarios.values():
203 if scen_obj.media and not self.media:
204 self.media.update(scen_obj.media)
205 if scen_obj.action_groups:
206 action_groups.extend(ag for ag in scen_obj.action_groups if ag not in action_groups)
207 if action_groups:
208 self.action_groups = action_groups
210 def select_deliveries(self) -> list[str]:
211 scenario_enable_deliveries: list[str] = []
212 default_enable_deliveries: list[str] = []
213 scenario_disable_deliveries: list[str] = []
215 if self.delivery_selection != DELIVERY_SELECTION_FIXED:
216 for scenario_name in self.enabled_scenarios:
217 scenario_enable_deliveries.extend(self.context.delivery_by_scenario.get(scenario_name, ()))
218 if self.delivery_selection == DELIVERY_SELECTION_IMPLICIT:
219 default_enable_deliveries = self.context.delivery_by_scenario.get(SCENARIO_DEFAULT, [])
221 override_enable_deliveries = []
222 override_disable_deliveries = []
224 for delivery, delivery_override in self.delivery_overrides.items():
225 if (delivery_override is None or delivery_override.get(CONF_ENABLED, True)) and delivery in self.context.deliveries:
226 override_enable_deliveries.append(delivery)
227 elif delivery_override is not None and not delivery_override.get(CONF_ENABLED, True):
228 override_disable_deliveries.append(delivery)
230 if self.delivery_selection != DELIVERY_SELECTION_FIXED:
231 scenario_disable_deliveries = [
232 d
233 for d, dc in self.context.deliveries.items()
234 if dc.get(CONF_SELECTION) == SELECTION_BY_SCENARIO and d not in scenario_enable_deliveries
235 ]
236 all_enabled = list(set(scenario_enable_deliveries + default_enable_deliveries + override_enable_deliveries))
237 all_disabled = scenario_disable_deliveries + override_disable_deliveries
238 if self.debug_trace:
239 self.debug_trace.delivery_selection["override_disable_deliveries"] = override_disable_deliveries
240 self.debug_trace.delivery_selection["override_enable_deliveries"] = override_enable_deliveries
241 self.debug_trace.delivery_selection["scenario_enable_deliveries"] = scenario_enable_deliveries
242 self.debug_trace.delivery_selection["default_enable_deliveries"] = default_enable_deliveries
243 self.debug_trace.delivery_selection["scenario_disable_deliveries"] = scenario_disable_deliveries
245 return [d for d in all_enabled if d not in all_disabled]
247 def default_media_from_actions(self) -> None:
248 """If no media defined, look for iOS / Android actions that have media defined"""
249 if self.media:
250 return
251 if self.data.get("image"):
252 self.media[ATTR_MEDIA_SNAPSHOT_URL] = self.data.get("image")
253 if self.data.get("video"):
254 self.media[ATTR_MEDIA_CLIP_URL] = self.data.get("video")
255 if self.data.get("attachment", {}).get("url"):
256 url = self.data["attachment"]["url"]
257 if url and url.endswith(".mp4") and not self.media.get(ATTR_MEDIA_CLIP_URL):
258 self.media[ATTR_MEDIA_CLIP_URL] = url
259 elif (
260 url
261 and (url.endswith(".jpg") or url.endswith(".jpeg") or url.endswith(".png"))
262 and not self.media.get(ATTR_MEDIA_SNAPSHOT_URL)
263 ):
264 self.media[ATTR_MEDIA_SNAPSHOT_URL] = url
266 def _render_scenario_templates(
267 self, original: str | None, template_field: str, matching_ctx: str, delivery_name: str
268 ) -> str | None:
269 template_scenario_names = self.context.content_scenario_templates.get(template_field, {}).get(delivery_name, [])
270 if not template_scenario_names:
271 return original
272 context_vars = self.condition_variables.as_dict() if self.condition_variables else {}
273 rendered = original if original is not None else ""
274 for scen_obj in [obj for name, obj in self.enabled_scenarios.items() if name in template_scenario_names]:
275 context_vars[matching_ctx] = rendered
276 try:
277 template_format = scen_obj.delivery.get(delivery_name, {}).get(CONF_DATA, {}).get(template_field)
278 if template_format is not None:
279 template = Template(template_format, self.context.hass)
280 rendered = template.async_render(variables=context_vars)
281 except TemplateError as e:
282 _LOGGER.warning("SUPERNOTIFIER Rendering template %s for %s failed: %s", template_field, delivery_name, e)
283 return rendered
285 def message(self, delivery_name: str) -> str | None:
286 # message and title reverse the usual defaulting, delivery config overrides runtime call
287 delivery_config: dict[str, Any] = self.context.deliveries.get(delivery_name, {})
288 msg: str | None = delivery_config.get(CONF_MESSAGE, self._message)
289 delivery_method: DeliveryMethod = self.context.delivery_method(delivery_name)
290 message_usage: str = str(delivery_method.option_str(OPTION_MESSAGE_USAGE, delivery_config))
291 if message_usage.upper() == MessageOnlyPolicy.USE_TITLE:
292 title = self.title(delivery_name, ignore_usage=True)
293 if title:
294 msg = title
295 elif message_usage.upper() == MessageOnlyPolicy.COMBINE_TITLE:
296 title = self.title(delivery_name, ignore_usage=True)
297 if title:
298 msg = f"{title} {msg}"
299 if (
300 delivery_method.option_bool(OPTION_SIMPLIFY_TEXT, delivery_config) is True
301 or delivery_method.option_bool(OPTION_STRIP_URLS, delivery_config) is True
302 ):
303 msg = delivery_method.simplify(msg, strip_urls=delivery_method.option_bool(OPTION_STRIP_URLS, delivery_config))
305 msg = self._render_scenario_templates(msg, "message_template", "notification_message", delivery_name)
306 if msg is None: # keep mypy happy
307 return None
308 return str(msg)
310 def title(self, delivery_name: str, ignore_usage: bool = False) -> str | None:
311 # message and title reverse the usual defaulting, delivery config overrides runtime call
312 delivery_config = self.context.deliveries.get(delivery_name, {})
313 delivery_method: DeliveryMethod = self.context.delivery_method(delivery_name)
314 message_usage = delivery_method.option_str(OPTION_MESSAGE_USAGE, delivery_config)
315 if not ignore_usage and message_usage.upper() in (MessageOnlyPolicy.USE_TITLE, MessageOnlyPolicy.COMBINE_TITLE):
316 title = None
317 else:
318 title = delivery_config.get(CONF_TITLE, self._title)
319 if (
320 delivery_method.option_bool(OPTION_SIMPLIFY_TEXT, delivery_config) is True
321 or delivery_method.option_bool(OPTION_STRIP_URLS, delivery_config) is True
322 ):
323 title = delivery_method.simplify(
324 title, strip_urls=delivery_method.option_bool(OPTION_STRIP_URLS, delivery_config)
325 )
326 title = self._render_scenario_templates(title, "title_template", "notification_title", delivery_name)
327 if title is None:
328 return None
329 return str(title)
331 def suppress(self) -> None:
332 self.globally_disabled = True
333 _LOGGER.info("SUPERNOTIFY Suppressing notification (%s)", self.id)
335 async def deliver(self) -> bool:
336 if self.globally_disabled:
337 _LOGGER.info("SUPERNOTIFY Suppressing globally silenced/snoozed notification (%s)", self.id)
338 self.skipped += 1
339 return False
341 _LOGGER.debug(
342 "Message: %s, notification: %s, deliveries: %s",
343 self._message,
344 self.id,
345 self.selected_delivery_names,
346 )
348 for delivery in self.selected_delivery_names:
349 await self.call_delivery_method(delivery)
351 if self.delivered == 0 and self.errored == 0:
352 for delivery in self.context.fallback_by_default:
353 if delivery not in self.selected_delivery_names:
354 await self.call_delivery_method(delivery)
356 if self.delivered == 0 and self.errored > 0:
357 for delivery in self.context.fallback_on_error:
358 if delivery not in self.selected_delivery_names:
359 await self.call_delivery_method(delivery)
361 return self.delivered > 0
363 async def call_delivery_method(self, delivery: str) -> None:
364 try:
365 delivery_method: DeliveryMethod = self.context.delivery_method(delivery)
366 delivery_config = delivery_method.delivery_config(delivery)
368 delivery_priorities = delivery_config.get(CONF_PRIORITY) or ()
369 if self.priority and delivery_priorities and self.priority not in delivery_priorities:
370 _LOGGER.debug("SUPERNOTIFY Skipping delivery %s based on priority (%s)", delivery, self.priority)
371 self.skipped += 1
372 return
373 if not await delivery_method.evaluate_delivery_conditions(delivery_config, self.condition_variables):
374 _LOGGER.debug("SUPERNOTIFY Skipping delivery %s based on conditions", delivery)
375 self.skipped += 1
376 return
378 recipients = self.generate_recipients(delivery, delivery_method)
379 envelopes = self.generate_envelopes(delivery, delivery_method, recipients)
380 for envelope in envelopes:
381 try:
382 await delivery_method.deliver(envelope)
383 self.delivered += envelope.delivered
384 self.errored += envelope.errored
385 if envelope.delivered:
386 self.delivered_envelopes.append(envelope)
387 else:
388 self.undelivered_envelopes.append(envelope)
389 except Exception as e2:
390 _LOGGER.warning("SUPERNOTIFY Failed to deliver %s: %s", envelope.delivery_name, e2)
391 _LOGGER.debug("SUPERNOTIFY %s", e2, exc_info=True)
392 self.errored += 1
393 envelope.delivery_error = format_exception(e2)
394 self.undelivered_envelopes.append(envelope)
396 except Exception as e:
397 _LOGGER.warning("SUPERNOTIFY Failed to notify using %s: %s", delivery, e)
398 _LOGGER.debug("SUPERNOTIFY %s delivery failure", delivery, exc_info=True)
399 self.delivery_errors[delivery] = format_exception(e)
401 def hash(self) -> int:
402 return hash((self._message, self._title))
404 def contents(self, minimal: bool = False) -> dict[str, Any]:
405 """ArchiveableObject implementation"""
406 sanitized = {k: v for k, v in self.__dict__.items() if k not in ("context")}
407 sanitized["delivered_envelopes"] = [e.contents(minimal=minimal) for e in self.delivered_envelopes]
408 sanitized["undelivered_envelopes"] = [e.contents(minimal=minimal) for e in self.undelivered_envelopes]
409 sanitized["enabled_scenarios"] = {k: v.contents(minimal=minimal) for k, v in self.enabled_scenarios.items()}
410 if self.debug_trace:
411 sanitized["debug_trace"] = self.debug_trace.contents()
412 else:
413 del sanitized["debug_trace"]
414 return sanitized
416 def base_filename(self) -> str:
417 """ArchiveableObject implementation"""
418 return f"{self.created.isoformat()[:16]}_{self.id}"
420 def delivery_data(self, delivery_name: str) -> dict[str, Any]:
421 delivery_override = self.delivery_overrides.get(delivery_name)
422 return delivery_override.get(CONF_DATA) if delivery_override else {}
424 def delivery_scenarios(self, delivery_name: str) -> dict[str, Scenario]:
425 return {
426 s: obj for s, obj in self.enabled_scenarios.items() if delivery_name in self.context.delivery_by_scenario.get(s, [])
427 }
429 async def select_scenarios(self) -> list[str]:
430 return [s.name for s in self.context.scenarios.values() if await s.evaluate(self.condition_variables)]
432 def merge(self, attribute: str, delivery_name: str) -> dict[str, Any]:
433 delivery: dict[str, Any] = self.delivery_overrides.get(delivery_name, {})
434 base: dict[str, Any] = delivery.get(attribute, {})
435 for scenario in self.enabled_scenarios.values():
436 if scenario and hasattr(scenario, attribute):
437 base.update(getattr(scenario, attribute))
438 if hasattr(self, attribute):
439 base.update(getattr(self, attribute))
440 return base
442 def record_resolve(self, delivery_name: str, category: str, resolved: str | list[Any] | None) -> None:
443 """Debug support for recording detailed target resolution in archived notification"""
444 self.debug_trace.resolved.setdefault(delivery_name, {})
445 self.debug_trace.resolved[delivery_name].setdefault(category, [])
446 if isinstance(resolved, list):
447 self.debug_trace.resolved[delivery_name][category].extend(resolved)
448 else:
449 self.debug_trace.resolved[delivery_name][category].append(resolved)
451 def filter_people_by_occupancy(self, occupancy: str) -> list[dict[str, Any]]:
452 people = list(self.context.people.values())
453 if occupancy == OCCUPANCY_ALL:
454 return people
455 if occupancy == OCCUPANCY_NONE:
456 return []
458 away = self.occupancy[STATE_NOT_HOME]
459 at_home = self.occupancy[STATE_HOME]
460 if occupancy == OCCUPANCY_ALL_IN:
461 return people if len(away) == 0 else []
462 if occupancy == OCCUPANCY_ALL_OUT:
463 return people if len(at_home) == 0 else []
464 if occupancy == OCCUPANCY_ANY_IN:
465 return people if len(at_home) > 0 else []
466 if occupancy == OCCUPANCY_ANY_OUT:
467 return people if len(away) > 0 else []
468 if occupancy == OCCUPANCY_ONLY_IN:
469 return at_home
470 if occupancy == OCCUPANCY_ONLY_OUT:
471 return away
473 _LOGGER.warning("SUPERNOTIFY Unknown occupancy tested: %s", occupancy)
474 return []
476 def generate_recipients(self, delivery_name: str, delivery_method: DeliveryMethod) -> list[dict[str, Any]]:
477 delivery_config: dict[str, Any] = delivery_method.delivery_config(delivery_name)
479 recipients: list[dict[str, Any]] = []
480 if self.target:
481 # first priority is explicit target set on notify call, which overrides everything else
482 for t in self.target:
483 if t in self.context.people:
484 recipients.append(self.context.people[t])
485 self.record_resolve(
486 delivery_name,
487 "1a_person_target",
488 t,
489 )
490 else:
491 recipients.append({ATTR_TARGET: t})
492 self.record_resolve(delivery_name, "1b_non_person_target", t)
493 _LOGGER.debug("SUPERNOTIFY %s Overriding with explicit targets: %s", __name__, recipients)
494 else:
495 # second priority is explicit target on delivery
496 if delivery_config and CONF_TARGET in delivery_config and delivery_config[CONF_TARGET]:
497 recipients.extend({ATTR_TARGET: e} for e in delivery_config.get(CONF_TARGET, []))
498 self.record_resolve(delivery_name, "2b_delivery_config_target", delivery_config.get(CONF_TARGET))
499 _LOGGER.debug("SUPERNOTIFY %s Using delivery config targets: %s", __name__, recipients)
501 # next priority is explicit recipients on delivery
502 if delivery_config and CONF_RECIPIENTS in delivery_config and delivery_config[CONF_RECIPIENTS]:
503 recipients.extend(delivery_config[CONF_RECIPIENTS])
504 self.record_resolve(delivery_name, "2c_delivery_config_recipient", delivery_config.get(CONF_RECIPIENTS))
505 _LOGGER.debug("SUPERNOTIFY %s Using overridden recipients: %s", delivery_name, recipients)
507 # If target not specified on service call or delivery, then default to std list of recipients
508 elif not delivery_config or CONF_TARGET not in delivery_config:
509 recipients = self.filter_people_by_occupancy(delivery_config.get(CONF_OCCUPANCY, OCCUPANCY_ALL))
510 self.record_resolve(delivery_name, "2d_recipients_by_occupancy", recipients)
511 recipients = [
512 r for r in recipients if self.recipients_override is None or r.get(CONF_PERSON) in self.recipients_override
513 ]
514 self.record_resolve(
515 delivery_name, "2d_recipient_names_by_occupancy_filtered", [r.get(CONF_PERSON) for r in recipients]
516 )
517 _LOGGER.debug("SUPERNOTIFY %s Using recipients: %s", delivery_name, recipients)
519 return self.context.snoozer.filter_recipients(
520 recipients, self.priority, delivery_name, delivery_method, self.selected_delivery_names, self.context.deliveries
521 )
523 def generate_envelopes(
524 self, delivery_name: str, method: DeliveryMethod, recipients: list[dict[str, Any]]
525 ) -> list[Envelope]:
526 # now the list of recipients determined, resolve this to target addresses or entities
528 delivery_config: dict[str, Any] = method.delivery_config(delivery_name)
529 default_data: dict[str, Any] = delivery_config.get(CONF_DATA, {})
530 default_targets: list[str] = []
531 custom_envelopes: list[Envelope] = []
533 for recipient in recipients:
534 recipient_targets: list[str] = []
535 enabled: bool = True
536 custom_data: dict[str, Any] = {}
537 # reuse standard recipient attributes like email or phone
538 safe_extend(recipient_targets, method.recipient_target(recipient))
539 # use entities or targets set at a method level for recipient
540 if CONF_DELIVERY in recipient and delivery_config[CONF_NAME] in recipient.get(CONF_DELIVERY, {}):
541 recp_meth_cust = recipient.get(CONF_DELIVERY, {}).get(delivery_config[CONF_NAME], {})
542 safe_extend(recipient_targets, recp_meth_cust.get(CONF_TARGET, []))
543 custom_data = recp_meth_cust.get(CONF_DATA)
544 enabled = recp_meth_cust.get(CONF_ENABLED, True)
545 elif ATTR_TARGET in recipient:
546 # non person recipient
547 safe_extend(default_targets, recipient.get(ATTR_TARGET))
548 if enabled:
549 if custom_data:
550 envelope_data = {}
551 envelope_data.update(default_data)
552 envelope_data.update(self.data)
553 envelope_data.update(custom_data)
554 custom_envelopes.append(Envelope(delivery_name, self, recipient_targets, envelope_data))
555 else:
556 default_targets.extend(recipient_targets)
558 envelope_data = {}
559 envelope_data.update(default_data)
560 envelope_data.update(self.data)
562 bundled_envelopes = [*custom_envelopes, Envelope(delivery_name, self, default_targets, envelope_data)]
563 filtered_envelopes = []
564 for envelope in bundled_envelopes:
565 pre_filter_count = len(envelope.targets)
566 _LOGGER.debug("SUPERNOTIFY Prefiltered targets: %s", envelope.targets)
567 targets = [t for t in envelope.targets if method.select_target(t)]
568 if len(targets) < pre_filter_count:
569 _LOGGER.warning(
570 "SUPERNOTIFY %s target list filtered out %s",
571 method.method,
572 [t for t in envelope.targets if not method.select_target(t)],
573 )
574 if not targets:
575 _LOGGER.debug("SUPERNOTIFY %s No targets resolved out of %s", method.method, pre_filter_count)
576 else:
577 envelope.targets = targets
578 filtered_envelopes.append(envelope)
580 if not filtered_envelopes:
581 # not all delivery methods require explicit targets, or can default them internally
582 filtered_envelopes = [Envelope(delivery_name, self, data=envelope_data)]
583 return filtered_envelopes
585 async def grab_image(self, delivery_name: str) -> Path | None:
586 snapshot_url = self.media.get(ATTR_MEDIA_SNAPSHOT_URL)
587 camera_entity_id = self.media.get(ATTR_MEDIA_CAMERA_ENTITY_ID)
588 delivery_config = self.delivery_data(delivery_name)
589 jpeg_opts = self.media.get(ATTR_JPEG_OPTS, delivery_config.get(CONF_OPTIONS, {}).get(ATTR_JPEG_OPTS))
591 if not snapshot_url and not camera_entity_id:
592 return None
594 image_path: Path | None = None
595 if self.snapshot_image_path is not None:
596 return self.snapshot_image_path
597 if snapshot_url and self.context.media_path and self.context.hass:
598 image_path = await snapshot_from_url(
599 self.context.hass, snapshot_url, self.id, self.context.media_path, self.context.hass_internal_url, jpeg_opts
600 )
601 elif camera_entity_id and camera_entity_id.startswith("image.") and self.context.hass and self.context.media_path:
602 image_path = await snap_image(self.context, camera_entity_id, self.context.media_path, self.id, jpeg_opts)
603 elif camera_entity_id:
604 if not self.context.hass or not self.context.media_path:
605 _LOGGER.warning("SUPERNOTIFY No homeassistant ref or media path for camera %s", camera_entity_id)
606 return None
607 active_camera_entity_id = select_avail_camera(self.context.hass, self.context.cameras, camera_entity_id)
608 if active_camera_entity_id:
609 camera_config = self.context.cameras.get(active_camera_entity_id, {})
610 camera_delay = self.media.get(ATTR_MEDIA_CAMERA_DELAY, camera_config.get(CONF_PTZ_DELAY))
611 camera_ptz_preset_default = camera_config.get(CONF_PTZ_PRESET_DEFAULT)
612 camera_ptz_method = camera_config.get(CONF_PTZ_METHOD)
613 camera_ptz_preset = self.media.get(ATTR_MEDIA_CAMERA_PTZ_PRESET)
614 _LOGGER.debug(
615 "SUPERNOTIFY snapping camera %s, ptz %s->%s, delay %s secs",
616 active_camera_entity_id,
617 camera_ptz_preset,
618 camera_ptz_preset_default,
619 camera_delay,
620 )
621 if camera_ptz_preset:
622 await move_camera_to_ptz_preset(
623 self.context.hass, active_camera_entity_id, camera_ptz_preset, method=camera_ptz_method
624 )
625 if camera_delay:
626 _LOGGER.debug("SUPERNOTIFY Waiting %s secs before snapping", camera_delay)
627 await asyncio.sleep(camera_delay)
628 image_path = await snap_camera(
629 self.context.hass,
630 active_camera_entity_id,
631 media_path=self.context.media_path,
632 max_camera_wait=15,
633 jpeg_opts=jpeg_opts,
634 )
635 if camera_ptz_preset and camera_ptz_preset_default:
636 await move_camera_to_ptz_preset(
637 self.context.hass, active_camera_entity_id, camera_ptz_preset_default, method=camera_ptz_method
638 )
640 if image_path is None:
641 _LOGGER.warning("SUPERNOTIFY No media available to attach (%s,%s)", snapshot_url, camera_entity_id)
642 return None
643 self.snapshot_image_path = image_path
644 return image_path