Coverage for custom_components / supernotify / media_grab.py: 93%
334 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-11 22:18 +0000
1from __future__ import annotations
3import asyncio
4import datetime as dt
5import io
6import logging
7import time
8from enum import StrEnum, auto
9from http import HTTPStatus
10from io import BytesIO
11from typing import TYPE_CHECKING, Any, cast
13import aiofiles
14import aiofiles.os
15import homeassistant.util.dt as dt_util
16from aiohttp import ClientResponse, ClientSession, ClientTimeout
17from anyio import Path
18from homeassistant.const import STATE_HOME, STATE_UNAVAILABLE
19from PIL import Image
21from custom_components.supernotify.const import (
22 ATTR_JPEG_OPTS,
23 ATTR_MEDIA_CAMERA_DELAY,
24 ATTR_MEDIA_CAMERA_ENTITY_ID,
25 ATTR_MEDIA_CAMERA_PTZ_PRESET,
26 ATTR_MEDIA_SNAPSHOT_PATH,
27 ATTR_MEDIA_SNAPSHOT_URL,
28 ATTR_PNG_OPTS,
29 CONF_ALT_CAMERA,
30 CONF_CAMERA,
31 CONF_DEVICE_TRACKER,
32 CONF_OPTIONS,
33 CONF_PTZ_CAMERA,
34 CONF_PTZ_DELAY,
35 CONF_PTZ_METHOD,
36 CONF_PTZ_PRESET_DEFAULT,
37 MEDIA_OPTION_REPROCESS,
38 OPTION_JPEG,
39 OPTION_PNG,
40 PTZ_METHOD_FRIGATE,
41 PTZ_METHOD_ONVIF,
42)
44if TYPE_CHECKING:
45 from homeassistant.components.image import ImageEntity
46 from homeassistant.core import State
48 from .context import Context
49 from .hass_api import HomeAssistantAPI
51_LOGGER = logging.getLogger(__name__)
54class ReprocessOption(StrEnum):
55 ALWAYS = auto()
56 NEVER = auto()
57 PRESERVE = auto()
60async def snapshot_from_url(
61 hass_api: HomeAssistantAPI,
62 snapshot_url: str,
63 notification_id: str,
64 media_path: Path,
65 hass_base_url: str | None,
66 remote_timeout: int = 15,
67) -> Path | None:
68 """Download a snapshot URL and save raw bytes. No reprocessing."""
69 hass_base_url = hass_base_url or ""
70 try:
71 raw_dir: Path = Path(media_path) / "raw"
72 await raw_dir.mkdir(parents=True, exist_ok=True)
74 image_url = snapshot_url if snapshot_url.startswith("http") else f"{hass_base_url}{snapshot_url}"
75 websession: ClientSession = hass_api.http_session()
76 r: ClientResponse = await websession.get(image_url, timeout=ClientTimeout(total=remote_timeout))
77 if r.status != HTTPStatus.OK:
78 _LOGGER.warning("SUPERNOTIFY Unable to retrieve %s: %s", image_url, r.status)
79 else:
80 bitmap: bytes | None = await r.content.read()
81 if bitmap:
82 ext = _detect_image_ext(bitmap)
83 raw_path: Path = raw_dir / f"{notification_id}.{ext}"
84 async with aiofiles.open(raw_path, "wb") as f:
85 await f.write(bitmap)
86 _LOGGER.debug("SUPERNOTIFY Fetched raw image from %s to %s", image_url, raw_path)
87 return raw_path
89 _LOGGER.warning("SUPERNOTIFY Failed to snap image from %s", snapshot_url)
90 except Exception as e:
91 _LOGGER.exception("SUPERNOTIFY Image snap fail: %s", e)
93 return None
96async def move_camera_to_ptz_preset(
97 hass_api: HomeAssistantAPI, camera_entity_id: str, preset: str | int, method: str = PTZ_METHOD_ONVIF
98) -> None:
99 try:
100 _LOGGER.info("SUPERNOTIFY Executing PTZ by %s to %s for %s", method, preset, camera_entity_id)
101 if method == PTZ_METHOD_FRIGATE:
102 await hass_api.call_service(
103 "frigate",
104 "ptz",
105 service_data={"action": "preset", "argument": preset},
106 target={"entity_id": camera_entity_id},
107 return_response=False,
108 blocking=True,
109 )
111 elif method == PTZ_METHOD_ONVIF:
112 await hass_api.call_service(
113 "onvif",
114 "ptz",
115 service_data={"move_mode": "GotoPreset", "preset": preset},
116 target={"entity_id": camera_entity_id},
117 return_response=False,
118 blocking=True,
119 )
120 else:
121 _LOGGER.warning("SUPERNOTIFY Unknown PTZ method %s", method)
122 except Exception as e:
123 _LOGGER.warning("SUPERNOTIFY Unable to move %s to ptz preset %s: %s", camera_entity_id, preset, e)
126async def snap_image_entity(
127 hass_api: HomeAssistantAPI,
128 entity_id: str,
129 media_path: Path,
130 notification_id: str,
131) -> Path | None:
132 """Read an image entity and save raw bytes. No reprocessing."""
133 raw_path: Path | None = None
134 try:
135 image_entity: ImageEntity | None = cast("ImageEntity|None", hass_api.domain_entity("image", entity_id))
136 if image_entity:
137 bitmap: bytes | None = await image_entity.async_image()
138 if bitmap:
139 raw_dir: Path = Path(media_path) / "raw"
140 await raw_dir.mkdir(parents=True, exist_ok=True)
141 ext = _detect_image_ext(bitmap)
142 raw_path = raw_dir / f"{notification_id}.{ext}"
143 async with aiofiles.open(raw_path, "wb") as f:
144 await f.write(bitmap)
145 except Exception as e:
146 _LOGGER.warning("SUPERNOTIFY Unable to snap image %s: %s", entity_id, e)
147 if raw_path is None:
148 _LOGGER.warning("SUPERNOTIFY Unable to save from image entity %s", entity_id)
149 return raw_path
152async def snap_camera(
153 hass_api: HomeAssistantAPI,
154 camera_entity_id: str,
155 notification_id: str,
156 media_path: Path,
157 max_camera_wait: int = 20,
158) -> Path | None:
159 """Snap a camera and save the raw image. No reprocessing."""
160 if not camera_entity_id:
161 _LOGGER.warning("SUPERNOTIFY Empty camera entity id for snap")
162 return None
164 raw_path: Path | None = None
165 try:
166 raw_dir: Path = Path(media_path) / "raw"
167 await raw_dir.mkdir(parents=True, exist_ok=True)
168 raw_path = raw_dir / f"{notification_id}.jpg"
169 share_root = Path(media_path)
170 share_path = share_root / raw_path.relative_to(Path(media_path))
171 await hass_api.call_service(
172 "camera",
173 "snapshot",
174 service_data={"entity_id": camera_entity_id, "filename": share_path},
175 return_response=False,
176 blocking=True,
177 )
179 cutoff_time = time.time() + max_camera_wait
180 while time.time() < cutoff_time and not await raw_path.exists():
181 _LOGGER.info("Image file not available yet at %s, pausing", raw_path)
182 await asyncio.sleep(1)
184 except Exception as e:
185 _LOGGER.warning("Failed to snap avail camera %s to %s: %s", camera_entity_id, raw_path, e)
186 raw_path = None
188 return raw_path
191def camera_available(hass_api: HomeAssistantAPI, camera_config: dict[str, Any], non_entity: bool = False) -> bool:
192 state: State | None = None
193 tracker_entity_id: str
194 camera_entity_id: str = camera_config[CONF_CAMERA]
195 try:
196 if camera_config.get(CONF_DEVICE_TRACKER):
197 tracker_entity_id = camera_config[CONF_DEVICE_TRACKER]
198 state = hass_api.get_state(camera_config[CONF_DEVICE_TRACKER])
199 if state and state.state == STATE_HOME:
200 return True
201 _LOGGER.debug("SUPERNOTIFY Skipping camera %s tracker %s state %s", camera_entity_id, tracker_entity_id, state)
202 else:
203 tracker_entity_id = camera_entity_id
204 state = hass_api.get_state(camera_entity_id)
205 if state and state.state != STATE_UNAVAILABLE:
206 return True
207 if state is None and non_entity:
208 return True
209 _LOGGER.debug("SUPERNOTIFY Skipping camera %s with state %s", camera_entity_id, state)
210 if state is None:
211 if tracker_entity_id == camera_entity_id:
212 _LOGGER.warning(
213 "SUPERNOTIFY Camera %s tracker %s has no entity state",
214 camera_entity_id,
215 tracker_entity_id,
216 )
217 else:
218 _LOGGER.warning(
219 "SUPERNOTIFY Camera %s device_tracker %s seems missing",
220 camera_entity_id,
221 camera_config[CONF_DEVICE_TRACKER],
222 )
223 return False
225 except Exception as e:
226 _LOGGER.exception("SUPERNOTIFY Unable to determine camera state: %s, %s", camera_config, e)
227 return False
230def select_avail_camera(hass_api: HomeAssistantAPI, cameras: dict[str, Any], camera_entity_id: str) -> str | None:
231 avail_camera_entity_id: str | None = None
233 preferred_cam = cameras.get(camera_entity_id)
234 # test support FIXME
235 if preferred_cam and CONF_CAMERA not in preferred_cam:
236 preferred_cam[CONF_CAMERA] = camera_entity_id
237 if preferred_cam is None:
238 # assume unconfigured camera available
239 return camera_entity_id
240 if camera_available(hass_api, preferred_cam):
241 return camera_entity_id
243 alt_cams: list[dict[str, Any]] = [cameras[c] for c in preferred_cam.get(CONF_ALT_CAMERA, []) if c in cameras]
244 alt_cams.extend(
245 {CONF_CAMERA: entity_id} for entity_id in preferred_cam.get(CONF_ALT_CAMERA, []) if entity_id not in cameras
246 )
247 for alt_cam in alt_cams:
248 if camera_available(hass_api, alt_cam):
249 _LOGGER.info("SUPERNOTIFY Selecting available camera %s rather than %s", alt_cam[CONF_CAMERA], camera_entity_id)
250 return alt_cam[CONF_CAMERA]
252 if avail_camera_entity_id is None:
253 _LOGGER.warning("%s not available, finding best alternative available", camera_entity_id)
254 if camera_available(hass_api, preferred_cam, non_entity=True):
255 _LOGGER.info("SUPERNOTIFY Selecting camera %s with no known entity", camera_entity_id)
256 return camera_entity_id
257 for alt_cam in alt_cams:
258 if camera_available(hass_api, alt_cam, non_entity=True):
259 _LOGGER.info(
260 "SUPERNOTIFY Selecting alt camera %s with no known entity for %s", alt_cam[CONF_CAMERA], camera_entity_id
261 )
262 return alt_cam[CONF_CAMERA]
264 return None
267def _detect_image_ext(bitmap: bytes) -> str:
268 """Detect image format from raw bytes, returning a file extension."""
269 try:
270 img = Image.open(io.BytesIO(bitmap))
271 fmt = (img.format or "").lower()
272 return "jpg" if fmt in ("jpg", "jpeg") else fmt or "img"
273 except Exception:
274 return "img"
277async def snap_notification_image(notification: Notification, context: Context) -> Path | None: # type: ignore # noqa: F821
278 """Delivery-neutral image acquisition: PTZ movement, camera snap, URL fetch, or image entity.
280 Caches the raw image path on notification._raw_image_path. Safe to call multiple times;
281 subsequent calls return the cached path immediately.
282 """
283 if getattr(notification, "_raw_image_path", None) is not None:
284 return notification._raw_image_path # type: ignore[attr-defined]
286 if notification.media.get(ATTR_MEDIA_SNAPSHOT_PATH) is not None:
287 return Path(notification.media[ATTR_MEDIA_SNAPSHOT_PATH])
289 snapshot_url = notification.media.get(ATTR_MEDIA_SNAPSHOT_URL)
290 camera_entity_id = notification.media.get(ATTR_MEDIA_CAMERA_ENTITY_ID)
291 media_path: Path | None = context.media_storage.media_path
293 if not media_path or (not snapshot_url and not camera_entity_id):
294 return None
295 if not context.hass_api:
296 return None
298 raw_path: Path | None = None
299 if snapshot_url:
300 raw_path = await snapshot_from_url(
301 context.hass_api, snapshot_url, notification.id, media_path, context.hass_api.internal_url
302 )
303 elif camera_entity_id.startswith("image."):
304 raw_path = await snap_image_entity(context.hass_api, camera_entity_id, media_path, notification.id)
305 else:
306 active_camera_entity_id = select_avail_camera(context.hass_api, context.cameras, camera_entity_id)
307 if active_camera_entity_id:
308 camera_config = context.cameras.get(active_camera_entity_id, {})
309 camera_ptz_entity_id: str = camera_config.get(CONF_PTZ_CAMERA, active_camera_entity_id)
310 camera_delay = notification.media.get(ATTR_MEDIA_CAMERA_DELAY, camera_config.get(CONF_PTZ_DELAY))
311 camera_ptz_preset_default = camera_config.get(CONF_PTZ_PRESET_DEFAULT)
312 camera_ptz_method = camera_config.get(CONF_PTZ_METHOD, PTZ_METHOD_ONVIF)
313 camera_ptz_preset = notification.media.get(ATTR_MEDIA_CAMERA_PTZ_PRESET)
314 _LOGGER.debug(
315 "SUPERNOTIFY snapping camera %s, ptz %s->%s (%s), delay %s secs",
316 active_camera_entity_id,
317 camera_ptz_preset,
318 camera_ptz_preset_default,
319 camera_ptz_entity_id,
320 camera_delay,
321 )
322 if camera_ptz_preset:
323 await move_camera_to_ptz_preset(
324 context.hass_api, camera_ptz_entity_id, camera_ptz_preset, method=camera_ptz_method
325 )
326 if camera_delay:
327 _LOGGER.debug("SUPERNOTIFY Waiting %s secs before snapping", camera_delay)
328 await asyncio.sleep(camera_delay)
329 raw_path = await snap_camera(
330 context.hass_api,
331 active_camera_entity_id,
332 notification.id,
333 media_path=media_path,
334 max_camera_wait=15,
335 )
336 if camera_ptz_preset and camera_ptz_preset_default:
337 await move_camera_to_ptz_preset(
338 context.hass_api, camera_ptz_entity_id, camera_ptz_preset_default, method=camera_ptz_method
339 )
341 if raw_path is None:
342 _LOGGER.warning("SUPERNOTIFY No media available to attach (%s,%s)", snapshot_url, camera_entity_id)
343 notification._raw_image_path = raw_path # type: ignore[attr-defined]
344 return raw_path
347async def grab_image(notification: Notification, delivery: Delivery, context: Context) -> Path | None: # type: ignore # noqa: F821
348 """Get a delivery-ready image, reprocessing the raw snap with delivery-specific settings.
350 The raw snap is cached on the notification; reprocessed variants are cached by filename
351 so multiple deliveries with the same settings share the processed file.
353 Filename convention:
354 raw/{nid}.{ext} — delivery-neutral camera output
355 image/{nid}.jpg — default reprocessing (ALWAYS, no extra opts)
356 image/{nid}_{hashed_opts}}.jpg — delivery-specific opts or non-ALWAYS reprocess mode
357 """
358 if notification.media.get(ATTR_MEDIA_SNAPSHOT_PATH) is not None:
359 return Path(notification.media[ATTR_MEDIA_SNAPSHOT_PATH])
361 raw_path = await snap_notification_image(notification, context)
362 if raw_path is None:
363 return None
365 media_path: Path | None = context.media_storage.media_path
366 if not media_path or not context.hass_api:
367 return None
369 delivery_config = notification.delivery_data(delivery)
370 jpeg_opts = notification.media.get(ATTR_JPEG_OPTS, delivery_config.get(CONF_OPTIONS, {}).get(OPTION_JPEG))
371 png_opts = notification.media.get(ATTR_PNG_OPTS, delivery_config.get(CONF_OPTIONS, {}).get(OPTION_PNG))
372 reprocess_option = (
373 notification.media.get(MEDIA_OPTION_REPROCESS, delivery_config.get(CONF_OPTIONS, {}).get(MEDIA_OPTION_REPROCESS))
374 or "always"
375 )
376 reprocess: ReprocessOption = ReprocessOption.ALWAYS
377 try:
378 reprocess = ReprocessOption(reprocess_option)
379 except Exception:
380 _LOGGER.warning("SUPERNOTIFY Invalid reprocess option: %s", reprocess_option)
382 if reprocess == ReprocessOption.NEVER:
383 return raw_path
385 raw_ext = raw_path.suffix.lstrip(".").lower()
386 relevant_opts: dict[str, Any] = jpeg_opts if raw_ext in ("jpg", "jpeg") else png_opts if raw_ext == "png" else {}
387 is_default = not relevant_opts and reprocess == ReprocessOption.ALWAYS
388 if is_default:
389 processed_name = f"{notification.id}.jpg"
390 else:
391 key = hex(hash((reprocess_option, *tuple(relevant_opts.values()))))[-12:]
392 processed_name = f"{notification.id}_{key}.jpg"
393 processed_path = Path(media_path) / "image" / processed_name
395 if await processed_path.exists():
396 return await processed_path.resolve()
398 async with await raw_path.open("rb") as f:
399 bitmap: bytes = await f.read()
400 return await write_image_from_bitmap(
401 context.hass_api, bitmap, processed_path, reprocess=reprocess, jpeg_opts=jpeg_opts, png_opts=png_opts
402 )
405async def write_image_from_bitmap(
406 hass_api: HomeAssistantAPI,
407 bitmap: bytes | None,
408 output_path: Path,
409 reprocess: ReprocessOption = ReprocessOption.ALWAYS,
410 output_format: str | None = None,
411 jpeg_opts: dict[str, Any] | None = None,
412 png_opts: dict[str, Any] | None = None,
413) -> Path | None:
414 """Reprocess a raw image bitmap and write to an explicit output path."""
415 if bitmap is None:
416 _LOGGER.debug("SUPERNOTIFY Empty bitmap for image")
417 return None
418 input_format: str = "img"
419 try:
420 await output_path.parent.mkdir(parents=True, exist_ok=True)
422 image = await hass_api.create_job(Image.open, io.BytesIO(bitmap))
424 input_format = image.format.lower() if image.format else "img"
425 if reprocess == ReprocessOption.ALWAYS:
426 # rewrite to remove metadata, incl custom CCTV comments that confuse python MIMEImage
427 clean_image: Image.Image = Image.new(image.mode, image.size)
428 clean_image.putdata(image.getdata()) # being removed in 2027
429 # clean_image.putdata(image.get_flattened_data()) # added in jan 2026
430 image = clean_image
432 buffer = BytesIO()
433 img_args: dict[str, Any] = {}
434 if reprocess in (ReprocessOption.ALWAYS, ReprocessOption.PRESERVE):
435 if input_format in ("jpg", "jpeg") and jpeg_opts:
436 img_args.update(jpeg_opts)
437 elif input_format == "png" and png_opts:
438 img_args.update(png_opts)
440 image.save(buffer, output_format or input_format, **img_args)
442 output_path = await output_path.resolve()
443 async with aiofiles.open(output_path, "wb") as file:
444 await file.write(buffer.getbuffer())
445 return output_path
446 except TypeError:
447 # probably a jpeg or png option
448 _LOGGER.exception("SUPERNOTIFY Image snap fail")
449 except Exception:
450 _LOGGER.exception("SUPERNOTIFY Failure saving %s bitmap", input_format)
451 return None
454class MediaStorage:
455 def __init__(
456 self,
457 media_path: str | None,
458 media_url_prefix: str | None = None,
459 days: int = 7,
460 ) -> None:
461 self.media_path: Path | None = Path(media_path) if media_path else None
462 self.last_purge: dt.datetime | None = None
463 self.media_url_prefix = media_url_prefix
464 self.purge_minute_interval = 60 * 6
465 self.days = days
467 async def initialize(self, hass_api: HomeAssistantAPI) -> None:
468 self.hass_api = hass_api # TODO: should not be set on initialize
469 if self.media_path is not None and not self.media_path.is_absolute():
470 self.media_path = await self.media_path.absolute()
471 _LOGGER.info("SUPERNOTIFY media path updated to %s", self.media_path)
472 if self.media_path and not await self.media_path.exists():
473 _LOGGER.info("SUPERNOTIFY media path not found at %s", self.media_path)
474 try:
475 await self.media_path.mkdir(parents=True, exist_ok=True)
476 except Exception as e:
477 _LOGGER.warning("SUPERNOTIFY media path %s cannot be created: %s", self.media_path, e)
478 hass_api.raise_issue(
479 "media_path",
480 "media_path",
481 {"path": str(self.media_path), "error": str(e)},
482 learn_more_url="https://supernotify.rhizomatics.org.uk/#getting-started",
483 )
484 self.media_path = None
485 if self.media_path is not None:
486 _LOGGER.info("SUPERNOTIFY abs media path: %s", await self.media_path.absolute())
488 if self.media_url_prefix is not None and self.media_path is not None:
489 if await hass_api.register_web_path(self.media_path, self.media_url_prefix):
490 _LOGGER.info("SUPERNOTIFY Media at %s available with prefixed URL %s", self.media_path, self.media_url_prefix)
491 else:
492 self.media_url_prefix = None
494 async def object_url(self, relative_path: Path) -> str | None:
495 """Convert a local image path to an externally accessible URL via the registered static path."""
496 if self.media_url_prefix is None or self.media_path is None:
497 _LOGGER.debug("SUPERNOTIFY Unable to generate object_url for %s", relative_path)
498 return None
499 try:
500 relative = relative_path.relative_to(await self.media_path.absolute())
501 return self.hass_api.abs_url(f"{self.media_url_prefix}/{relative}")
502 except ValueError as e:
503 _LOGGER.warning("SUPERNOTIFY Invalid media path for URL %s: %s", relative_path, e)
504 return None
506 async def share_path(self, artefact_path: Path) -> str | None:
507 """Return the HA media-share path for a local file (e.g. '/media/images/raw/foo.jpg').
509 Used by mobile push: the companion app resolves share-rooted paths against the HA base URL.
510 Returns None when media_path is unconfigured or path is outside the media tree.
511 """
512 if self.media_path is None or self.media_url_prefix is None:
513 return None
514 try:
515 if artefact_path.is_absolute():
516 relative = artefact_path.relative_to(await self.media_path.absolute())
517 else:
518 relative = artefact_path
519 return str(Path(self.media_url_prefix) / relative)
520 except ValueError as e:
521 _LOGGER.debug("SUPERNOTIFY Failure creating shared media path for %s:%s", artefact_path, e)
522 return None
524 async def size(self) -> int:
525 path: Path | None = self.media_path
526 if path and await path.exists():
527 return sum(1 for p in await aiofiles.os.listdir(path))
528 return 0
530 async def cleanup(self, days: int | None = None, force: bool = False) -> int:
531 if (
532 not force
533 and self.last_purge is not None
534 and self.last_purge > dt.datetime.now(dt.UTC) - dt.timedelta(minutes=self.purge_minute_interval)
535 ):
536 _LOGGER.debug(
537 "SUPERNOTIFY Media storage cleanup skipped, force: %s, last purge: %s, purge interval: %s",
538 force,
539 self.last_purge,
540 self.purge_minute_interval,
541 )
542 return 0
543 days = days or self.days
544 if days == 0 or self.media_path is None:
545 _LOGGER.debug("SUPERNOTIFY Media storage cleanup skipped, %s days, %s", days, self.media_path)
546 return 0
548 cutoff = dt.datetime.now(dt.UTC) - dt.timedelta(days=days)
549 cutoff = cutoff.astimezone(dt.UTC)
550 purged: int = 0
551 skipped: int = 0
552 if self.media_path and await self.media_path.exists():
553 try:
554 queue: list[Path] = [self.media_path]
555 while queue:
556 current = queue.pop()
557 for entry in await aiofiles.os.scandir(current):
558 if entry.is_dir():
559 queue.append(Path(entry.path))
560 elif entry.is_file() and dt_util.utc_from_timestamp(entry.stat().st_mtime) <= cutoff:
561 _LOGGER.debug("SUPERNOTIFY Purging %s", entry.path)
562 await aiofiles.os.unlink(Path(entry.path))
563 purged += 1
564 else:
565 skipped += 1
566 except Exception as e:
567 _LOGGER.warning("SUPERNOTIFY Unable to clean up media storage at %s: %s", self.media_path, e, exc_info=True)
568 _LOGGER.info("SUPERNOTIFY Purged %s media storage for cutoff %s, skipped %s", purged, cutoff, skipped)
569 self.last_purge = dt.datetime.now(dt.UTC)
570 else:
571 _LOGGER.warning("SUPERNOTIFY Skipping media storage cleanup for unknown path %s", self.media_path)
572 return purged