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