Coverage for custom_components/supernotify/notification.py: 88%
373 statements
« prev ^ index » next coverage.py v7.6.8, created at 2025-05-31 23:25 +0000
1import asyncio
2import datetime as dt
3import logging
4import uuid
5from pathlib import Path
6from traceback import format_exception
7from typing import Any, cast
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 voluptuous import humanize
14from custom_components.supernotify import (
15 ACTION_DATA_SCHEMA,
16 ATTR_ACTION_GROUPS,
17 ATTR_ACTIONS,
18 ATTR_DEBUG,
19 ATTR_DELIVERY,
20 ATTR_DELIVERY_SELECTION,
21 ATTR_JPEG_FLAGS,
22 ATTR_MEDIA,
23 ATTR_MEDIA_CAMERA_DELAY,
24 ATTR_MEDIA_CAMERA_ENTITY_ID,
25 ATTR_MEDIA_CAMERA_PTZ_PRESET,
26 ATTR_MEDIA_CLIP_URL,
27 ATTR_MEDIA_SNAPSHOT_URL,
28 ATTR_MESSAGE_HTML,
29 ATTR_PRIORITY,
30 ATTR_RECIPIENTS,
31 ATTR_SCENARIOS_APPLY,
32 ATTR_SCENARIOS_CONSTRAIN,
33 ATTR_SCENARIOS_REQUIRE,
34 CONF_DATA,
35 CONF_DELIVERY,
36 CONF_MESSAGE,
37 CONF_OCCUPANCY,
38 CONF_OPTIONS,
39 CONF_PERSON,
40 CONF_PRIORITY,
41 CONF_PTZ_DELAY,
42 CONF_PTZ_METHOD,
43 CONF_PTZ_PRESET_DEFAULT,
44 CONF_RECIPIENTS,
45 CONF_SELECTION,
46 CONF_TITLE,
47 DELIVERY_SELECTION_EXPLICIT,
48 DELIVERY_SELECTION_FIXED,
49 DELIVERY_SELECTION_IMPLICIT,
50 OCCUPANCY_ALL,
51 OCCUPANCY_ALL_IN,
52 OCCUPANCY_ALL_OUT,
53 OCCUPANCY_ANY_IN,
54 OCCUPANCY_ANY_OUT,
55 OCCUPANCY_NONE,
56 OCCUPANCY_ONLY_IN,
57 OCCUPANCY_ONLY_OUT,
58 PRIORITY_MEDIUM,
59 PRIORITY_VALUES,
60 SCENARIO_DEFAULT,
61 SCENARIO_NULL,
62 SELECTION_BY_SCENARIO,
63 STRICT_ACTION_DATA_SCHEMA,
64 ConditionVariables,
65)
66from custom_components.supernotify.archive import ArchivableObject
67from custom_components.supernotify.common import DebugTrace, safe_extend
68from custom_components.supernotify.delivery_method import DeliveryMethod
69from custom_components.supernotify.envelope import Envelope
70from custom_components.supernotify.scenario import Scenario
72from .common import ensure_dict, ensure_list
73from .configuration import SupernotificationConfiguration
74from .media_grab import move_camera_to_ptz_preset, select_avail_camera, snap_camera, snap_image, snapshot_from_url
76_LOGGER = logging.getLogger(__name__)
79class Notification(ArchivableObject):
80 def __init__(
81 self,
82 context: SupernotificationConfiguration,
83 message: str | None = None,
84 title: str | None = None,
85 target: list | str | None = None,
86 action_data: dict | None = None,
87 ) -> None:
88 self.created: dt.datetime = dt.datetime.now(tz=dt.UTC)
89 self.debug_trace: DebugTrace = DebugTrace(message=message, title=title, data=action_data, target=target)
90 self._message: str | None = message
91 self.context: SupernotificationConfiguration = context
92 action_data = action_data or {}
93 self.target: list[str] = ensure_list(target)
94 self._title: str | None = title
95 self.id = str(uuid.uuid1())
96 self.snapshot_image_path: Path | None = None
97 self.delivered: int = 0
98 self.errored: int = 0
99 self.skipped: int = 0
100 self.delivered_envelopes: list[Envelope] = []
101 self.undelivered_envelopes: list[Envelope] = []
102 self.delivery_error: list[str] | None = None
104 self.validate_action_data(action_data)
105 # for compatibility with other notify calls, pass thru surplus data to underlying delivery methods
106 self.data: dict[str, Any] = {k: v for k, v in action_data.items() if k not in STRICT_ACTION_DATA_SCHEMA(action_data)}
107 action_data = {k: v for k, v in action_data.items() if k not in self.data}
109 self.priority: str = action_data.get(ATTR_PRIORITY, PRIORITY_MEDIUM)
110 self.message_html: str | None = action_data.get(ATTR_MESSAGE_HTML)
111 self.required_scenarios: list = ensure_list(action_data.get(ATTR_SCENARIOS_REQUIRE))
112 self.applied_scenarios: list = ensure_list(action_data.get(ATTR_SCENARIOS_APPLY))
113 self.constrain_scenarios: list = ensure_list(action_data.get(ATTR_SCENARIOS_CONSTRAIN))
114 self.delivery_selection: str | None = action_data.get(ATTR_DELIVERY_SELECTION)
115 self.delivery_overrides_type: str = action_data.get(ATTR_DELIVERY).__class__.__name__
116 self.delivery_overrides: dict = ensure_dict(action_data.get(ATTR_DELIVERY))
117 self.action_groups: list[str] | None = action_data.get(ATTR_ACTION_GROUPS)
118 self.recipients_override: list[str] | None = action_data.get(ATTR_RECIPIENTS)
119 self.data.update(action_data.get(ATTR_DATA, {}))
120 self.media: dict = action_data.get(ATTR_MEDIA) or {}
121 self.debug: bool = action_data.get(ATTR_DEBUG, False)
122 self.actions: dict = action_data.get(ATTR_ACTIONS) or {}
123 self.delivery_results: dict = {}
124 self.delivery_errors: dict = {}
126 self.selected_delivery_names: list[str] = []
127 self.enabled_scenarios: list[str] = []
128 self.selected_scenarios: list[str] = []
129 self.people_by_occupancy: list = []
130 self.globally_disabled: bool = False
131 self.occupancy: dict[str, list] = {}
132 self.condition_variables: ConditionVariables | None = None
134 async def initialize(self) -> None:
135 """Async post-construction initialization"""
136 if self.delivery_selection is None:
137 if self.delivery_overrides_type in ("list", "str"):
138 # a bare list of deliveries implies intent to restrict
139 _LOGGER.debug("SUPERNOTIFY defaulting delivery selection as explicit for type %s", self.delivery_overrides_type)
140 self.delivery_selection = DELIVERY_SELECTION_EXPLICIT
141 else:
142 # whereas a dict may be used to tune or restrict
143 self.delivery_selection = DELIVERY_SELECTION_IMPLICIT
144 _LOGGER.debug("SUPERNOTIFY defaulting delivery selection as implicit for type %s", self.delivery_overrides_type)
146 self.occupancy = self.context.determine_occupancy()
147 self.condition_variables = ConditionVariables(
148 self.applied_scenarios,
149 self.required_scenarios,
150 self.constrain_scenarios,
151 self.priority,
152 self.occupancy,
153 self._message,
154 self._title,
155 ) # requires occupancy first
157 self.enabled_scenarios = list(self.applied_scenarios) or []
158 self.selected_scenarios = await self.select_scenarios()
159 self.enabled_scenarios.extend(self.selected_scenarios)
160 if self.constrain_scenarios:
161 self.enabled_scenarios = [
162 s
163 for s in self.enabled_scenarios
164 if (s in self.constrain_scenarios or s in self.applied_scenarios) and s != SCENARIO_NULL
165 ]
166 if self.required_scenarios and not any(s in self.enabled_scenarios for s in self.required_scenarios):
167 _LOGGER.info("SUPERNOTIFY suppressing notification, no required scenarios enabled")
168 self.selected_delivery_names = []
169 self.globally_disabled = True
170 else:
171 self.selected_delivery_names = self.select_deliveries()
172 self.globally_disabled = self.context.snoozer.is_global_snooze(self.priority)
173 self.default_media_from_actions()
174 self.apply_enabled_scenarios()
176 def validate_action_data(self, action_data: dict) -> None:
177 if action_data.get(ATTR_PRIORITY) and action_data.get(ATTR_PRIORITY) not in PRIORITY_VALUES:
178 _LOGGER.warning("SUPERNOTIFY invalid priority %s - overriding to medium", action_data.get(ATTR_PRIORITY))
179 action_data[ATTR_PRIORITY] = PRIORITY_MEDIUM
180 try:
181 humanize.validate_with_humanized_errors(action_data, ACTION_DATA_SCHEMA)
182 except vol.Invalid as e:
183 _LOGGER.warning("SUPERNOTIFY invalid service data %s: %s", action_data, e)
184 raise
186 def apply_enabled_scenarios(self) -> None:
187 """Set media and action_groups from scenario if defined, first come first applied"""
188 action_groups: list[str] = []
189 for scenario_name in self.enabled_scenarios:
190 scen_obj = self.context.scenarios.get(scenario_name)
191 if scen_obj:
192 if scen_obj.media and not self.media:
193 self.media.update(scen_obj.media)
194 if scen_obj.action_groups:
195 action_groups.extend(ag for ag in scen_obj.action_groups if ag not in action_groups)
196 if action_groups:
197 self.action_groups = action_groups
199 def select_deliveries(self) -> list[str]:
200 scenario_enable_deliveries: list[str] = []
201 default_enable_deliveries: list[str] = []
202 scenario_disable_deliveries: list[str] = []
204 if self.delivery_selection != DELIVERY_SELECTION_FIXED:
205 for scenario in self.enabled_scenarios:
206 scenario_enable_deliveries.extend(self.context.delivery_by_scenario.get(scenario, ()))
207 if self.delivery_selection == DELIVERY_SELECTION_IMPLICIT:
208 default_enable_deliveries = self.context.delivery_by_scenario.get(SCENARIO_DEFAULT, [])
210 override_enable_deliveries = []
211 override_disable_deliveries = []
213 for delivery, delivery_override in self.delivery_overrides.items():
214 if (delivery_override is None or delivery_override.get(CONF_ENABLED, True)) and delivery in self.context.deliveries:
215 override_enable_deliveries.append(delivery)
216 elif delivery_override is not None and not delivery_override.get(CONF_ENABLED, True):
217 override_disable_deliveries.append(delivery)
219 if self.delivery_selection != DELIVERY_SELECTION_FIXED:
220 scenario_disable_deliveries = [
221 d
222 for d, dc in self.context.deliveries.items()
223 if dc.get(CONF_SELECTION) == SELECTION_BY_SCENARIO and d not in scenario_enable_deliveries
224 ]
225 all_enabled = list(set(scenario_enable_deliveries + default_enable_deliveries + override_enable_deliveries))
226 all_disabled = scenario_disable_deliveries + override_disable_deliveries
227 if self.debug_trace:
228 self.debug_trace.delivery_selection["override_disable_deliveries"] = override_disable_deliveries
229 self.debug_trace.delivery_selection["override_enable_deliveries"] = override_enable_deliveries
230 self.debug_trace.delivery_selection["scenario_enable_deliveries"] = scenario_enable_deliveries
231 self.debug_trace.delivery_selection["default_enable_deliveries"] = default_enable_deliveries
232 self.debug_trace.delivery_selection["scenario_disable_deliveries"] = scenario_disable_deliveries
234 return [d for d in all_enabled if d not in all_disabled]
236 def default_media_from_actions(self) -> None:
237 """If no media defined, look for iOS / Android actions that have media defined"""
238 if self.media:
239 return
240 if self.data.get("image"):
241 self.media[ATTR_MEDIA_SNAPSHOT_URL] = self.data.get("image")
242 if self.data.get("video"):
243 self.media[ATTR_MEDIA_CLIP_URL] = self.data.get("video")
244 if self.data.get("attachment", {}).get("url"):
245 url = self.data["attachment"]["url"]
246 if url and url.endswith(".mp4") and not self.media.get(ATTR_MEDIA_CLIP_URL):
247 self.media[ATTR_MEDIA_CLIP_URL] = url
248 elif (
249 url
250 and (url.endswith(".jpg") or url.endswith(".jpeg") or url.endswith(".png"))
251 and not self.media.get(ATTR_MEDIA_SNAPSHOT_URL)
252 ):
253 self.media[ATTR_MEDIA_SNAPSHOT_URL] = url
255 def message(self, delivery_name: str) -> str | None:
256 # message and title reverse the usual defaulting, delivery config overrides runtime call
257 return self.context.deliveries.get(delivery_name, {}).get(CONF_MESSAGE, self._message)
259 def title(self, delivery_name: str) -> str | None:
260 # message and title reverse the usual defaulting, delivery config overrides runtime call
261 return self.context.deliveries.get(delivery_name, {}).get(CONF_TITLE, self._title)
263 def suppress(self) -> None:
264 self.globally_disabled = True
265 _LOGGER.info("SUPERNOTIFY Suppressing notification (%s)", self.id)
267 async def deliver(self) -> bool:
268 if self.globally_disabled:
269 _LOGGER.info("SUPERNOTIFY Suppressing globally silenced/snoozed notification (%s)", self.id)
270 self.skipped += 1
271 return False
273 _LOGGER.debug(
274 "Message: %s, notification: %s, deliveries: %s",
275 self._message,
276 self.id,
277 self.selected_delivery_names,
278 )
280 for delivery in self.selected_delivery_names:
281 await self.call_delivery_method(delivery)
283 if self.delivered == 0 and self.errored == 0:
284 for delivery in self.context.fallback_by_default:
285 if delivery not in self.selected_delivery_names:
286 await self.call_delivery_method(delivery)
288 if self.delivered == 0 and self.errored > 0:
289 for delivery in self.context.fallback_on_error:
290 if delivery not in self.selected_delivery_names:
291 await self.call_delivery_method(delivery)
293 return self.delivered > 0
295 async def call_delivery_method(self, delivery: str) -> None:
296 try:
297 delivery_method: DeliveryMethod = self.context.delivery_method(delivery)
298 delivery_config = delivery_method.delivery_config(delivery)
300 delivery_priorities = delivery_config.get(CONF_PRIORITY) or ()
301 if self.priority and delivery_priorities and self.priority not in delivery_priorities:
302 _LOGGER.debug("SUPERNOTIFY Skipping delivery %s based on priority (%s)", delivery, self.priority)
303 self.skipped += 1
304 return
305 if not await delivery_method.evaluate_delivery_conditions(delivery_config, self.condition_variables):
306 _LOGGER.debug("SUPERNOTIFY Skipping delivery %s based on conditions", delivery)
307 self.skipped += 1
308 return
310 recipients = self.generate_recipients(delivery, delivery_method)
311 envelopes = self.generate_envelopes(delivery, delivery_method, recipients)
312 for envelope in envelopes:
313 try:
314 await delivery_method.deliver(envelope)
315 self.delivered += envelope.delivered
316 self.errored += envelope.errored
317 if envelope.delivered:
318 self.delivered_envelopes.append(envelope)
319 else:
320 self.undelivered_envelopes.append(envelope)
321 except Exception as e2:
322 _LOGGER.warning("SUPERNOTIFY Failed to deliver %s: %s", envelope.delivery_name, e2)
323 _LOGGER.debug("SUPERNOTIFY %s", e2, exc_info=True)
324 self.errored += 1
325 envelope.delivery_error = format_exception(e2)
326 self.undelivered_envelopes.append(envelope)
328 except Exception as e:
329 _LOGGER.warning("SUPERNOTIFY Failed to notify using %s: %s", delivery, e)
330 _LOGGER.debug("SUPERNOTIFY %s delivery failure", delivery, exc_info=True)
331 self.delivery_errors[delivery] = format_exception(e)
333 def hash(self) -> int:
334 return hash((self._message, self._title))
336 def contents(self, minimal: bool = False) -> dict[str, Any]:
337 """ArchiveableObject implementation"""
338 sanitized = {k: v for k, v in self.__dict__.items() if k not in ("context")}
339 sanitized["delivered_envelopes"] = [e.contents(minimal=minimal) for e in self.delivered_envelopes]
340 sanitized["undelivered_envelopes"] = [e.contents(minimal=minimal) for e in self.undelivered_envelopes]
341 if self.debug_trace:
342 sanitized["debug_trace"] = self.debug_trace.contents()
343 else:
344 del sanitized["debug_trace"]
345 return sanitized
347 def base_filename(self) -> str:
348 """ArchiveableObject implementation"""
349 return f"{self.created.isoformat()[:16]}_{self.id}"
351 def delivery_data(self, delivery_name: str) -> dict:
352 delivery_override = self.delivery_overrides.get(delivery_name)
353 return delivery_override.get(CONF_DATA) if delivery_override else {}
355 def delivery_scenarios(self, delivery_name: str) -> dict[str, Scenario]:
356 return {
357 k: cast("Scenario", self.context.scenarios.get(k))
358 for k in self.enabled_scenarios
359 if delivery_name in self.context.delivery_by_scenario.get(k, [])
360 }
362 async def select_scenarios(self) -> list[str]:
363 return [s.name for s in self.context.scenarios.values() if await s.evaluate(self.condition_variables)]
365 def merge(self, attribute: str, delivery_name: str) -> dict:
366 delivery: dict = self.delivery_overrides.get(delivery_name, {})
367 base: dict = delivery.get(attribute, {})
368 for scenario_name in self.enabled_scenarios:
369 scenario = self.context.scenarios.get(scenario_name)
370 if scenario and hasattr(scenario, attribute):
371 base.update(getattr(scenario, attribute))
372 if hasattr(self, attribute):
373 base.update(getattr(self, attribute))
374 return base
376 def record_resolve(self, delivery_name: str, category: str, resolved: str | list | None) -> None:
377 """Debug support for recording detailed target resolution in archived notification"""
378 self.debug_trace.resolved.setdefault(delivery_name, {})
379 self.debug_trace.resolved[delivery_name].setdefault(category, [])
380 if isinstance(resolved, list):
381 self.debug_trace.resolved[delivery_name][category].extend(resolved)
382 else:
383 self.debug_trace.resolved[delivery_name][category].append(resolved)
385 def filter_people_by_occupancy(self, occupancy: str) -> list[dict]:
386 people = list(self.context.people.values())
387 if occupancy == OCCUPANCY_ALL:
388 return people
389 if occupancy == OCCUPANCY_NONE:
390 return []
392 away = self.occupancy[STATE_NOT_HOME]
393 at_home = self.occupancy[STATE_HOME]
394 if occupancy == OCCUPANCY_ALL_IN:
395 return people if len(away) == 0 else []
396 if occupancy == OCCUPANCY_ALL_OUT:
397 return people if len(at_home) == 0 else []
398 if occupancy == OCCUPANCY_ANY_IN:
399 return people if len(at_home) > 0 else []
400 if occupancy == OCCUPANCY_ANY_OUT:
401 return people if len(away) > 0 else []
402 if occupancy == OCCUPANCY_ONLY_IN:
403 return at_home
404 if occupancy == OCCUPANCY_ONLY_OUT:
405 return away
407 _LOGGER.warning("SUPERNOTIFY Unknown occupancy tested: %s", occupancy)
408 return []
410 def generate_recipients(self, delivery_name: str, delivery_method: DeliveryMethod) -> list[dict]:
411 delivery_config: dict[str, Any] = delivery_method.delivery_config(delivery_name)
413 recipients: list[dict] = []
414 if self.target:
415 # first priority is explicit target set on notify call, which overrides everything else
416 for t in self.target:
417 if t in self.context.people:
418 recipients.append(self.context.people[t])
419 self.record_resolve(
420 # type: ignore
421 delivery_name,
422 "1a_person_target",
423 t,
424 )
425 else:
426 recipients.append({ATTR_TARGET: t})
427 self.record_resolve(delivery_name, "1b_non_person_target", t)
428 _LOGGER.debug("SUPERNOTIFY %s Overriding with explicit targets: %s", __name__, recipients)
429 else:
430 # second priority is explicit entities on delivery
431 if delivery_config and CONF_ENTITIES in delivery_config and delivery_config[CONF_ENTITIES]:
432 recipients.extend({ATTR_TARGET: e} for e in delivery_config.get(CONF_ENTITIES, []))
433 self.record_resolve(delivery_name, "2a_delivery_config_entity", delivery_config.get(CONF_ENTITIES))
434 _LOGGER.debug("SUPERNOTIFY %s Using delivery config entities: %s", __name__, recipients)
435 # third priority is explicit target on delivery
436 if delivery_config and CONF_TARGET in delivery_config and delivery_config[CONF_TARGET]:
437 recipients.extend({ATTR_TARGET: e} for e in delivery_config.get(CONF_TARGET, []))
438 self.record_resolve(delivery_name, "2b_delivery_config_target", delivery_config.get(CONF_TARGET))
439 _LOGGER.debug("SUPERNOTIFY %s Using delivery config targets: %s", __name__, recipients)
441 # next priority is explicit recipients on delivery
442 if delivery_config and CONF_RECIPIENTS in delivery_config and delivery_config[CONF_RECIPIENTS]:
443 recipients.extend(delivery_config[CONF_RECIPIENTS])
444 self.record_resolve(delivery_name, "2c_delivery_config_recipient", delivery_config.get(CONF_RECIPIENTS))
445 _LOGGER.debug("SUPERNOTIFY %s Using overridden recipients: %s", delivery_name, recipients)
447 # If target not specified on service call or delivery, then default to std list of recipients
448 elif not delivery_config or (CONF_TARGET not in delivery_config and CONF_ENTITIES not in delivery_config):
449 recipients = self.filter_people_by_occupancy(delivery_config.get(CONF_OCCUPANCY, OCCUPANCY_ALL))
450 self.record_resolve(delivery_name, "2d_recipients_by_occupancy", recipients)
451 recipients = [
452 r for r in recipients if self.recipients_override is None or r.get(CONF_PERSON) in self.recipients_override
453 ]
454 self.record_resolve(
455 delivery_name, "2d_recipient_names_by_occupancy_filtered", [r.get(CONF_PERSON) for r in recipients]
456 )
457 _LOGGER.debug("SUPERNOTIFY %s Using recipients: %s", delivery_name, recipients)
459 return self.context.snoozer.filter_recipients(
460 recipients, self.priority, delivery_name, delivery_method, self.selected_delivery_names, self.context.deliveries
461 )
463 def generate_envelopes(self, delivery_name: str, method: DeliveryMethod, recipients: list[dict]) -> list[Envelope]:
464 # now the list of recipients determined, resolve this to target addresses or entities
466 delivery_config: dict = method.delivery_config(delivery_name)
467 default_data: dict = delivery_config.get(CONF_DATA, {})
468 default_targets: list = []
469 custom_envelopes: list = []
471 for recipient in recipients:
472 recipient_targets: list = []
473 enabled: bool = True
474 custom_data: dict = {}
475 # reuse standard recipient attributes like email or phone
476 safe_extend(recipient_targets, method.recipient_target(recipient))
477 # use entities or targets set at a method level for recipient
478 if CONF_DELIVERY in recipient and delivery_config[CONF_NAME] in recipient.get(CONF_DELIVERY, {}):
479 recp_meth_cust = recipient.get(CONF_DELIVERY, {}).get(delivery_config[CONF_NAME], {})
480 safe_extend(recipient_targets, recp_meth_cust.get(CONF_ENTITIES, []))
481 safe_extend(recipient_targets, recp_meth_cust.get(CONF_TARGET, []))
482 custom_data = recp_meth_cust.get(CONF_DATA)
483 enabled = recp_meth_cust.get(CONF_ENABLED, True)
484 elif ATTR_TARGET in recipient:
485 # non person recipient
486 safe_extend(default_targets, recipient.get(ATTR_TARGET))
487 if enabled:
488 if custom_data:
489 envelope_data = {}
490 envelope_data.update(default_data)
491 envelope_data.update(self.data)
492 envelope_data.update(custom_data)
493 custom_envelopes.append(Envelope(delivery_name, self, recipient_targets, envelope_data))
494 else:
495 default_targets.extend(recipient_targets)
497 envelope_data = {}
498 envelope_data.update(default_data)
499 envelope_data.update(self.data)
501 bundled_envelopes = [*custom_envelopes, Envelope(delivery_name, self, default_targets, envelope_data)]
502 filtered_envelopes = []
503 for envelope in bundled_envelopes:
504 pre_filter_count = len(envelope.targets)
505 _LOGGER.debug("SUPERNOTIFY Prefiltered targets: %s", envelope.targets)
506 targets = [t for t in envelope.targets if method.select_target(t)]
507 if len(targets) < pre_filter_count:
508 _LOGGER.debug(
509 "SUPERNOTIFY %s target list filtered by %s to %s", method.method, pre_filter_count - len(targets), targets
510 )
511 if not targets:
512 _LOGGER.debug("SUPERNOTIFY %s No targets resolved out of %s", method.method, pre_filter_count)
513 else:
514 envelope.targets = targets
515 filtered_envelopes.append(envelope)
517 if not filtered_envelopes:
518 # not all delivery methods require explicit targets, or can default them internally
519 filtered_envelopes = [Envelope(delivery_name, self, data=envelope_data)]
520 return filtered_envelopes
522 async def grab_image(self, delivery_name: str) -> Path | None:
523 snapshot_url = self.media.get(ATTR_MEDIA_SNAPSHOT_URL)
524 camera_entity_id = self.media.get(ATTR_MEDIA_CAMERA_ENTITY_ID)
525 delivery_config = self.delivery_data(delivery_name)
526 jpeg_args = self.media.get(ATTR_JPEG_FLAGS, delivery_config.get(CONF_OPTIONS, {}).get(ATTR_JPEG_FLAGS))
528 if not snapshot_url and not camera_entity_id:
529 return None
531 image_path: Path | None = None
532 if self.snapshot_image_path is not None:
533 return self.snapshot_image_path
534 if snapshot_url and self.context.media_path and self.context.hass:
535 image_path = await snapshot_from_url(
536 self.context.hass, snapshot_url, self.id, self.context.media_path, self.context.hass_internal_url, jpeg_args
537 )
538 elif camera_entity_id and camera_entity_id.startswith("image.") and self.context.hass and self.context.media_path:
539 image_path = await snap_image(self.context, camera_entity_id, self.context.media_path, self.id, jpeg_args)
540 elif camera_entity_id:
541 if not self.context.hass or not self.context.media_path:
542 _LOGGER.warning("SUPERNOTIFY No homeassistant ref or media path for camera %s", camera_entity_id)
543 return None
544 active_camera_entity_id = await select_avail_camera(self.context.hass, self.context.cameras, camera_entity_id)
545 if active_camera_entity_id:
546 camera_config = self.context.cameras.get(active_camera_entity_id, {})
547 camera_delay = self.media.get(ATTR_MEDIA_CAMERA_DELAY, camera_config.get(CONF_PTZ_DELAY))
548 camera_ptz_preset_default = camera_config.get(CONF_PTZ_PRESET_DEFAULT)
549 camera_ptz_method = camera_config.get(CONF_PTZ_METHOD)
550 camera_ptz_preset = self.media.get(ATTR_MEDIA_CAMERA_PTZ_PRESET)
551 _LOGGER.debug(
552 "SUPERNOTIFY snapping camera %s, ptz %s->%s, delay %s secs",
553 active_camera_entity_id,
554 camera_ptz_preset,
555 camera_ptz_preset_default,
556 camera_delay,
557 )
558 if camera_ptz_preset:
559 await move_camera_to_ptz_preset(
560 self.context.hass, active_camera_entity_id, camera_ptz_preset, method=camera_ptz_method
561 )
562 if camera_delay:
563 _LOGGER.debug("SUPERNOTIFY Waiting %s secs before snapping", camera_delay)
564 await asyncio.sleep(camera_delay)
565 image_path = await snap_camera(
566 self.context.hass,
567 active_camera_entity_id,
568 media_path=self.context.media_path,
569 max_camera_wait=15,
570 jpeg_args=jpeg_args,
571 )
572 if camera_ptz_preset and camera_ptz_preset_default:
573 await move_camera_to_ptz_preset(
574 self.context.hass, active_camera_entity_id, camera_ptz_preset_default, method=camera_ptz_method
575 )
577 if image_path is None:
578 _LOGGER.warning("SUPERNOTIFY No media available to attach (%s,%s)", snapshot_url, camera_entity_id)
579 return None
580 self.snapshot_image_path = image_path
581 return image_path