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