Coverage for custom_components/supernotify/media_grab.py: 95%
358 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 14:29 +0000
1from __future__ import annotations
3import asyncio
4import datetime as dt
5import hashlib
6import io
7import json
8import logging
9from enum import StrEnum, auto
10from http import HTTPStatus
11from io import BytesIO
12from typing import TYPE_CHECKING, Any, cast
14import aiofiles
15import aiofiles.os
16import homeassistant.components.camera as ha_camera
17import homeassistant.components.image as ha_image
18import homeassistant.util.dt as dt_util
19from aiohttp import ClientResponse, ClientSession, ClientTimeout
20from anyio import Path
21from homeassistant.const import STATE_HOME, STATE_UNAVAILABLE
22from PIL import Image
24from custom_components.supernotify.const import (
25 ATTR_JPEG_OPTS,
26 ATTR_MEDIA_CAMERA_DELAY,
27 ATTR_MEDIA_CAMERA_ENTITY_ID,
28 ATTR_MEDIA_CAMERA_PTZ_PRESET,
29 ATTR_MEDIA_SNAPSHOT_PATH,
30 ATTR_MEDIA_SNAPSHOT_URL,
31 ATTR_PNG_OPTS,
32 CONF_ALT_CAMERA,
33 CONF_CAMERA,
34 CONF_DEVICE_TRACKER,
35 CONF_OPTIONS,
36 CONF_PTZ_CAMERA,
37 CONF_PTZ_DELAY,
38 CONF_PTZ_METHOD,
39 CONF_PTZ_PRESET_DEFAULT,
40 CONF_SNAP_WAIT,
41 PLATFORM_FRIGATE,
42 PTZ_DELAY_DEFAULT,
43 PTZ_METHOD_FRIGATE,
44 PTZ_METHOD_ONVIF,
45 SNAP_WAIT_DEFAULT,
46)
47from custom_components.supernotify.options import MEDIA_OPTION_REPROCESS, OPTION_JPEG, OPTION_PNG
49from .common import int_or_none
51if TYPE_CHECKING:
52 from homeassistant.core import Context as HAContext
53 from homeassistant.core import State
55 from .context import Context
56 from .delivery import Delivery
57 from .hass_api import HomeAssistantAPI
58 from .notification import Notification
60_LOGGER = logging.getLogger(__name__)
63class ReprocessOption(StrEnum):
64 ALWAYS = auto()
65 NEVER = auto()
66 PRESERVE = auto()
69async def snapshot_from_url(
70 hass_api: HomeAssistantAPI,
71 snapshot_url: str,
72 notification_id: str,
73 media_path: Path,
74 hass_base_url: str | None,
75 remote_timeout: int = 15,
76) -> Path | None:
77 """Download a snapshot URL and save raw bytes. No reprocessing."""
78 hass_base_url = hass_base_url or ""
79 try:
80 raw_dir: Path = Path(media_path) / "raw"
81 await raw_dir.mkdir(parents=True, exist_ok=True)
83 image_url = snapshot_url if snapshot_url.startswith("http") else f"{hass_base_url}{snapshot_url}"
84 websession: ClientSession = hass_api.http_session()
85 r: ClientResponse = await websession.get(image_url, timeout=ClientTimeout(total=remote_timeout))
86 if r.status != HTTPStatus.OK:
87 _LOGGER.warning("SUPERNOTIFY Unable to retrieve %s: %s", image_url, r.status)
88 else:
89 bitmap: bytes | None = await r.content.read()
90 if bitmap:
91 ext = await _detect_image_ext(hass_api, bitmap)
92 raw_path: Path = raw_dir / f"{notification_id}.{ext}"
93 async with aiofiles.open(raw_path, "wb") as f:
94 await f.write(bitmap)
95 _LOGGER.debug("SUPERNOTIFY Fetched raw image from %s to %s", image_url, raw_path)
96 return raw_path
98 _LOGGER.warning("SUPERNOTIFY Failed to snap image from %s", snapshot_url)
99 except Exception:
100 _LOGGER.exception("SUPERNOTIFY Image snap fail")
102 return None
105def infer_ptz_method(hass_api: HomeAssistantAPI, camera_entity_id: str) -> str:
106 """Guess a PTZ method from the integration that owns the camera entity.
108 Used only as a fallback for cameras with no entry in the cameras: config, where
109 there's no explicit ptz_method to consult.
110 """
111 if hass_api.platform_for_entity(camera_entity_id) == PLATFORM_FRIGATE:
112 return PTZ_METHOD_FRIGATE
113 return PTZ_METHOD_ONVIF
116async def move_camera_to_ptz_preset(
117 hass_api: HomeAssistantAPI,
118 camera_entity_id: str,
119 preset: str | int,
120 method: str = PTZ_METHOD_ONVIF,
121 ha_context: HAContext | None = None,
122) -> None:
123 try:
124 _LOGGER.info("SUPERNOTIFY Executing PTZ by %s to %s for %s", method, preset, camera_entity_id)
125 if method == PTZ_METHOD_FRIGATE:
126 await hass_api.call_service(
127 "frigate",
128 "ptz",
129 service_data={"action": "preset", "argument": preset},
130 target={"entity_id": camera_entity_id},
131 return_response=False,
132 blocking=True,
133 context=ha_context,
134 )
136 elif method == PTZ_METHOD_ONVIF:
137 await hass_api.call_service(
138 "onvif",
139 "ptz",
140 service_data={"move_mode": "GotoPreset", "preset": preset},
141 target={"entity_id": camera_entity_id},
142 return_response=False,
143 blocking=True,
144 context=ha_context,
145 )
146 else:
147 _LOGGER.warning("SUPERNOTIFY Unknown PTZ method %s", method)
148 except Exception as e:
149 _LOGGER.warning("SUPERNOTIFY Unable to move %s to ptz preset %s: %s", camera_entity_id, preset, e)
152async def snap_image_entity(
153 hass_api: HomeAssistantAPI, entity_id: str, media_path: Path, notification_id: str, max_image_wait: int = 30
154) -> Path | None:
155 """Read an image entity and save raw bytes. No reprocessing."""
156 raw_path: Path | None = None
157 try:
158 image: ha_image.Image | None = await hass_api.async_get_image_entity_image(entity_id, timeout=max_image_wait)
159 if image and image.content:
160 raw_dir: Path = Path(media_path) / "raw"
161 await raw_dir.mkdir(parents=True, exist_ok=True)
162 ext: str = await _detect_image_ext(hass_api, image.content)
163 raw_path = raw_dir / f"{notification_id}.{ext}"
164 async with aiofiles.open(raw_path, "wb") as f:
165 await f.write(image.content)
166 except Exception as e:
167 _LOGGER.warning("SUPERNOTIFY Unable to snap image %s: %s", entity_id, e)
168 if raw_path is None:
169 _LOGGER.warning("SUPERNOTIFY Unable to save from image entity %s", entity_id)
170 return raw_path
173async def snap_camera(
174 hass_api: HomeAssistantAPI,
175 camera_entity_id: str,
176 notification_id: str,
177 media_path: Path,
178 max_camera_wait: int = 20,
179) -> Path | None:
180 """Snap a camera and save the raw image. No reprocessing.
182 Fetches the still directly from the camera entity via HA's own camera component API,
183 rather than triggering the camera.snapshot service and polling the filesystem for the
184 resulting file to appear.
185 """
186 if not camera_entity_id:
187 _LOGGER.warning("SUPERNOTIFY Empty camera entity id for snap")
188 return None
190 raw_path: Path | None = None
191 try:
192 image: ha_camera.Image | None = await hass_api.async_get_camera_image(camera_entity_id, timeout=max_camera_wait)
193 if image and image.content:
194 raw_dir: Path = Path(media_path) / "raw"
195 await raw_dir.mkdir(parents=True, exist_ok=True)
196 ext: str = await _detect_image_ext(hass_api, image.content)
197 raw_path = raw_dir / f"{notification_id}.{ext}"
198 async with aiofiles.open(raw_path, "wb") as f:
199 await f.write(image.content)
200 except Exception as e:
201 _LOGGER.warning("Failed to snap avail camera %s: %s", camera_entity_id, e)
202 raw_path = None
204 return raw_path
207def camera_available(hass_api: HomeAssistantAPI, camera_config: dict[str, Any], non_entity: bool = False) -> bool:
208 state: State | None = None
209 tracker_entity_id: str
210 camera_entity_id: str = camera_config[CONF_CAMERA]
211 try:
212 if camera_config.get(CONF_DEVICE_TRACKER):
213 tracker_entity_id = camera_config[CONF_DEVICE_TRACKER]
214 state = hass_api.get_state(camera_config[CONF_DEVICE_TRACKER])
215 if state and state.state == STATE_HOME:
216 return True
217 _LOGGER.debug("SUPERNOTIFY Skipping camera %s tracker %s state %s", camera_entity_id, tracker_entity_id, state)
218 else:
219 tracker_entity_id = camera_entity_id
220 state = hass_api.get_state(camera_entity_id)
221 if state and state.state != STATE_UNAVAILABLE:
222 return True
223 if state is None and non_entity:
224 return True
225 _LOGGER.debug("SUPERNOTIFY Skipping camera %s with state %s", camera_entity_id, state)
226 if state is None:
227 if tracker_entity_id == camera_entity_id:
228 _LOGGER.warning(
229 "SUPERNOTIFY Camera %s tracker %s has no entity state",
230 camera_entity_id,
231 tracker_entity_id,
232 )
233 else:
234 _LOGGER.warning(
235 "SUPERNOTIFY Camera %s device_tracker %s seems missing",
236 camera_entity_id,
237 camera_config[CONF_DEVICE_TRACKER],
238 )
239 return False
241 except Exception:
242 _LOGGER.exception("SUPERNOTIFY Unable to determine camera state: %s", camera_config)
243 return False
246def select_avail_camera(
247 hass_api: HomeAssistantAPI, cameras: dict[str, Any], camera_entity_id: str, exclude_primary: bool = False
248) -> str | None:
249 """exclude_primary skips re-offering camera_entity_id itself, for a *configured* camera
250 (one with a `cameras:` entry - an unconfigured entity has no alternative to weigh it
251 against, so it's unaffected). A configured camera's own availability check is state-based
252 (see camera_available()) and can't detect one that's live-disabled at the device without
253 that showing up as entity state - so a caller that already knows, from an actual failed
254 fetch rather than just this heuristic, that camera_entity_id isn't currently deliverable
255 (e.g. mobile_push falling back after grab_image() found nothing) should set this, or it'll
256 just be handed the same unusable camera again.
257 """
258 avail_camera_entity_id: str | None = None
260 preferred_cam = cameras.get(camera_entity_id)
261 # test support FIXME
262 if preferred_cam and CONF_CAMERA not in preferred_cam:
263 preferred_cam[CONF_CAMERA] = camera_entity_id
264 if preferred_cam is None:
265 # no alternatives for an unconfigured entity, whether camera or image, so it's just
266 # a matter of whether it's known to be unavailable - assume it's fine if it has no state
267 if camera_available(hass_api, {CONF_CAMERA: camera_entity_id}, non_entity=True):
268 return camera_entity_id
269 return None
270 if not exclude_primary and camera_available(hass_api, preferred_cam):
271 return camera_entity_id
273 alt_cams: list[dict[str, Any]] = [cameras[c] for c in preferred_cam.get(CONF_ALT_CAMERA, []) if c in cameras]
274 alt_cams.extend(
275 {CONF_CAMERA: entity_id} for entity_id in preferred_cam.get(CONF_ALT_CAMERA, []) if entity_id not in cameras
276 )
277 for alt_cam in alt_cams:
278 if camera_available(hass_api, alt_cam):
279 _LOGGER.info("SUPERNOTIFY Selecting available camera %s rather than %s", alt_cam[CONF_CAMERA], camera_entity_id)
280 return alt_cam[CONF_CAMERA]
282 if avail_camera_entity_id is None:
283 _LOGGER.warning("SUPERNOTIFY %s not available, finding best alternative available", camera_entity_id)
284 if not exclude_primary and camera_available(hass_api, preferred_cam, non_entity=True):
285 _LOGGER.info("SUPERNOTIFY Selecting camera %s with no known entity", camera_entity_id)
286 return camera_entity_id
287 for alt_cam in alt_cams:
288 if camera_available(hass_api, alt_cam, non_entity=True):
289 _LOGGER.info(
290 "SUPERNOTIFY Selecting alt camera %s with no known entity for %s", alt_cam[CONF_CAMERA], camera_entity_id
291 )
292 return alt_cam[CONF_CAMERA]
294 return None
297async def _detect_image_ext(hass_api: HomeAssistantAPI, bitmap: bytes) -> str:
298 """Detect image format from raw bytes, returning a file extension."""
299 try:
300 img = await hass_api.create_job(Image.open, io.BytesIO(bitmap))
301 fmt = (img.format or "").lower()
302 return "jpg" if fmt in ("jpg", "jpeg") else fmt or "img"
303 except Exception as e:
304 _LOGGER.warning("SUPERNOTIFY unable to detect image, defaulting to 'img': %s", e)
305 return "img"
308async def snap_notification_image(
309 notification: Notification, context: Context, ha_context: HAContext | None = None
310) -> Path | None:
311 """Delivery-neutral image acquisition: PTZ movement, camera snap, URL fetch, or image entity.
313 Caches the raw image path on notification._raw_image_path. Safe to call multiple times;
314 subsequent calls return the cached path immediately.
315 """
316 if getattr(notification, "_raw_image_path", None) is not None:
317 return notification._raw_image_path # type: ignore[attr-defined]
319 if notification.media.get(ATTR_MEDIA_SNAPSHOT_PATH) is not None:
320 return Path(notification.media[ATTR_MEDIA_SNAPSHOT_PATH])
322 snapshot_url = notification.media.get(ATTR_MEDIA_SNAPSHOT_URL)
323 camera_entity_id = cast("str", notification.media.get(ATTR_MEDIA_CAMERA_ENTITY_ID))
324 media_path: Path | None = context.media_storage.media_path
326 if not media_path or (not snapshot_url and not camera_entity_id):
327 return None
328 if not context.hass_api:
329 return None
331 raw_path: Path | None = None
332 if snapshot_url:
333 raw_path = await snapshot_from_url(
334 context.hass_api, snapshot_url, notification.id, media_path, context.hass_api.internal_url
335 )
336 elif camera_entity_id.startswith("image."):
337 raw_path = await snap_image_entity(
338 context.hass_api, camera_entity_id, media_path, notification.id, max_image_wait=SNAP_WAIT_DEFAULT
339 )
340 else:
341 active_camera_entity_id = select_avail_camera(context.hass_api, context.cameras, camera_entity_id)
342 if active_camera_entity_id:
343 camera_config = context.cameras.get(active_camera_entity_id, {})
344 camera_ptz_entity_id: str = camera_config.get(CONF_PTZ_CAMERA, active_camera_entity_id)
345 camera_delay: int | None = int_or_none(notification.media.get(ATTR_MEDIA_CAMERA_DELAY))
346 camera_ptz_preset_default = camera_config.get(CONF_PTZ_PRESET_DEFAULT)
347 camera_ptz_preset = notification.media.get(ATTR_MEDIA_CAMERA_PTZ_PRESET)
348 camera_ptz_method = camera_config.get(CONF_PTZ_METHOD)
349 if camera_ptz_method is None:
350 camera_ptz_method = infer_ptz_method(context.hass_api, camera_ptz_entity_id)
352 if camera_ptz_preset:
353 camera_delay = (
354 camera_delay if camera_delay is not None else camera_config.get(CONF_PTZ_DELAY, PTZ_DELAY_DEFAULT)
355 )
356 _LOGGER.debug(
357 "SUPERNOTIFY Moving camera %s, ptz %s->%s (%s), delay %s secs",
358 active_camera_entity_id,
359 camera_ptz_preset,
360 camera_ptz_preset_default,
361 camera_ptz_entity_id,
362 camera_delay,
363 )
364 await move_camera_to_ptz_preset(
365 context.hass_api, camera_ptz_entity_id, camera_ptz_preset, method=camera_ptz_method, ha_context=ha_context
366 )
367 if camera_delay:
368 # pause if there's a PTZ movement, or notification explicitly asked for `camera_delay`
369 _LOGGER.debug("SUPERNOTIFY Waiting %s secs before snapping", camera_delay)
370 await asyncio.sleep(camera_delay)
372 max_camera_wait: int = camera_config.get(CONF_SNAP_WAIT, SNAP_WAIT_DEFAULT)
373 _LOGGER.debug(
374 "SUPERNOTIFY Snapping camera %s, max_wait: %s, to: %s", active_camera_entity_id, max_camera_wait, media_path
375 )
376 raw_path = await snap_camera(
377 context.hass_api,
378 active_camera_entity_id,
379 notification.id,
380 media_path=media_path,
381 max_camera_wait=max_camera_wait,
382 )
383 if camera_ptz_preset and camera_ptz_preset_default:
384 await move_camera_to_ptz_preset(
385 context.hass_api,
386 camera_ptz_entity_id,
387 camera_ptz_preset_default,
388 method=camera_ptz_method,
389 ha_context=ha_context,
390 )
392 if raw_path is None:
393 _LOGGER.warning("SUPERNOTIFY No media available to attach (%s,%s)", snapshot_url, camera_entity_id)
394 notification._raw_image_path = raw_path # type: ignore[attr-defined]
395 return raw_path
398async def grab_image(
399 notification: Notification, delivery: Delivery, context: Context, ha_context: HAContext | None = None
400) -> Path | None:
401 """Get a delivery-ready image, reprocessing the raw snap with delivery-specific settings.
403 The raw snap is cached on the notification; reprocessed variants are cached by filename
404 so multiple deliveries with the same settings share the processed file.
406 Filename convention:
407 raw/{nid}.{ext} — delivery-neutral camera output
408 image/{nid}.jpg — default reprocessing (ALWAYS, no extra opts)
409 image/{nid}_{hashed_opts}}.jpg — delivery-specific opts or non-ALWAYS reprocess mode
410 """
411 if notification.media.get(ATTR_MEDIA_SNAPSHOT_PATH) is not None:
412 return Path(notification.media[ATTR_MEDIA_SNAPSHOT_PATH])
414 raw_path = await snap_notification_image(notification, context, ha_context=ha_context)
415 if raw_path is None:
416 return None
418 media_path: Path | None = context.media_storage.media_path
419 if not media_path or not context.hass_api:
420 return None
422 delivery_config = notification.delivery_data(delivery)
423 jpeg_opts = notification.media.get(ATTR_JPEG_OPTS, delivery_config.get(CONF_OPTIONS, {}).get(OPTION_JPEG))
424 png_opts = notification.media.get(ATTR_PNG_OPTS, delivery_config.get(CONF_OPTIONS, {}).get(OPTION_PNG))
425 reprocess_option = (
426 notification.media.get(MEDIA_OPTION_REPROCESS, delivery_config.get(CONF_OPTIONS, {}).get(MEDIA_OPTION_REPROCESS))
427 or "always"
428 )
429 reprocess: ReprocessOption = ReprocessOption.ALWAYS
430 try:
431 reprocess = ReprocessOption(reprocess_option)
432 except Exception:
433 _LOGGER.warning("SUPERNOTIFY Invalid reprocess option: %s", reprocess_option)
435 if reprocess == ReprocessOption.NEVER:
436 return raw_path
438 raw_ext = raw_path.suffix.lstrip(".").lower()
439 # `jpeg_opts`/`png_opts` are None when not configured, and the reprocessed image keeps the
440 # format of the original, since that is what `write_image_from_bitmap` saves
441 relevant_opts: dict[str, Any] = (
442 (jpeg_opts or {}) if raw_ext in ("jpg", "jpeg") else (png_opts or {}) if raw_ext == "png" else {}
443 )
444 processed_ext = raw_ext or "jpg"
445 is_default = not relevant_opts and reprocess == ReprocessOption.ALWAYS
446 if is_default:
447 processed_name = f"{notification.id}.{processed_ext}"
448 else:
449 # a stable digest, unlike hash(), which is salted per process and so never matched an
450 # image reprocessed before the last restart
451 key = hashlib.sha1( # not security, just a cache key
452 json.dumps([reprocess_option, relevant_opts], sort_keys=True, default=str).encode(), usedforsecurity=False
453 ).hexdigest()[:12]
454 processed_name = f"{notification.id}_{key}.{processed_ext}"
455 processed_path = Path(media_path) / "image" / processed_name
457 if await processed_path.exists():
458 return await processed_path.resolve()
460 try:
461 async with await raw_path.open("rb") as f:
462 bitmap: bytes = await f.read()
463 except OSError as e:
464 _LOGGER.warning("SUPERNOTIFY Unable to read raw image %s: %s", raw_path, e)
465 return None
466 return await write_image_from_bitmap(
467 context.hass_api, bitmap, processed_path, reprocess=reprocess, jpeg_opts=jpeg_opts, png_opts=png_opts
468 )
471def _reprocess_bitmap(
472 bitmap: bytes,
473 reprocess: ReprocessOption,
474 output_format: str | None = None,
475 jpeg_opts: dict[str, Any] | None = None,
476 png_opts: dict[str, Any] | None = None,
477) -> tuple[str, bytes]:
478 """Decode, optionally rewrite and re-encode a bitmap, returning its format and the bytes
480 Blocking from end to end, so it is run in an executor job rather than on the event loop.
481 """
482 image: Image.Image = Image.open(io.BytesIO(bitmap))
483 input_format: str = image.format.lower() if image.format else "img"
484 if reprocess == ReprocessOption.ALWAYS:
485 # rewrite to remove metadata, incl custom CCTV comments that confuse python MIMEImage
486 clean_image: Image.Image = Image.new(image.mode, image.size)
487 # Pillow API changed in 12.1.0 and the original call will be removed in 2027
488 # https://pillow.readthedocs.io/en/stable/releasenotes/12.1.0.html#image-getdata
489 if hasattr(image, "get_flattened_data"):
490 clean_image.putdata(image.get_flattened_data()) # added in jan 2026 # ty:ignore[call-non-callable]
491 else:
492 clean_image.putdata(image.getdata()) # being removed in 2027
494 image = clean_image
496 img_args: dict[str, Any] = {}
497 if reprocess in (ReprocessOption.ALWAYS, ReprocessOption.PRESERVE):
498 if input_format in ("jpg", "jpeg") and jpeg_opts:
499 img_args.update(jpeg_opts)
500 elif input_format == "png" and png_opts:
501 img_args.update(png_opts)
503 buffer = BytesIO()
504 image.save(buffer, output_format or input_format, **img_args)
505 return input_format, buffer.getvalue()
508async def write_image_from_bitmap(
509 hass_api: HomeAssistantAPI,
510 bitmap: bytes | None,
511 output_path: Path,
512 reprocess: ReprocessOption = ReprocessOption.ALWAYS,
513 output_format: str | None = None,
514 jpeg_opts: dict[str, Any] | None = None,
515 png_opts: dict[str, Any] | None = None,
516) -> Path | None:
517 """Reprocess a raw image bitmap and write to an explicit output path."""
518 if bitmap is None:
519 _LOGGER.debug("SUPERNOTIFY Empty bitmap for image")
520 return None
521 input_format: str = "img"
522 try:
523 await output_path.parent.mkdir(parents=True, exist_ok=True)
525 # every Pillow call in one job: decoding, copying the pixels and encoding are all
526 # blocking, and only `Image.open` and `Image.new` used to be kept off the event loop
527 input_format, encoded = await hass_api.create_job(
528 _reprocess_bitmap, bitmap, reprocess, output_format, jpeg_opts, png_opts
529 )
530 buffer = BytesIO(encoded)
532 output_path = await output_path.resolve()
533 async with aiofiles.open(output_path, "wb") as file:
534 await file.write(buffer.getbuffer())
535 return output_path
536 except TypeError:
537 # probably a jpeg or png option
538 _LOGGER.exception("SUPERNOTIFY Image snap fail")
539 except Exception:
540 _LOGGER.exception("SUPERNOTIFY Failure saving %s bitmap", input_format)
541 return None
544class MediaStorage:
545 def __init__(
546 self,
547 media_path: str | None,
548 media_url_prefix: str | None = None,
549 days: int = 7,
550 ) -> None:
551 self.media_path: Path | None = Path(media_path) if media_path else None
552 self.last_purge: dt.datetime | None = None
553 self.media_url_prefix = media_url_prefix
554 self.purge_minute_interval = 60 * 6
555 self.days = days
557 async def initialize(self, hass_api: HomeAssistantAPI) -> None:
558 self.hass_api = hass_api # TODO: should not be set on initialize
559 if self.media_path is not None and not self.media_path.is_absolute():
560 self.media_path = await self.media_path.absolute()
561 _LOGGER.debug("SUPERNOTIFY Media path updated to %s", self.media_path)
562 if self.media_path and not await self.media_path.exists():
563 _LOGGER.info("SUPERNOTIFY Media path not found at %s, attempting to create.", self.media_path)
564 try:
565 await self.media_path.mkdir(parents=True, exist_ok=True)
566 except Exception as e:
567 _LOGGER.warning("SUPERNOTIFY Media path %s cannot be created: %s", self.media_path, e)
568 hass_api.raise_issue(
569 "media_path",
570 "media_path",
571 {"path": str(self.media_path), "error": str(e)},
572 learn_more_url="https://supernotify.rhizomatics.org.uk/getting_started/",
573 )
574 self.media_path = None
575 if self.media_path is not None:
576 _LOGGER.debug("SUPERNOTIFY Abs media path: %s", await self.media_path.absolute())
578 if self.media_url_prefix is not None and self.media_path is not None:
579 if await hass_api.register_web_path(self.media_path, self.media_url_prefix):
580 _LOGGER.info("SUPERNOTIFY Media at %s available with prefixed URL %s", self.media_path, self.media_url_prefix)
581 else:
582 self.media_url_prefix = None
584 async def object_url(self, relative_path: Path) -> str | None:
585 """Convert a local image path to an externally accessible URL via the registered static path."""
586 if self.media_url_prefix is None or self.media_path is None:
587 _LOGGER.debug("SUPERNOTIFY Unable to generate object_url for %s", relative_path)
588 return None
589 try:
590 relative = relative_path.relative_to(await self.media_path.absolute())
591 return self.hass_api.abs_url(f"{self.media_url_prefix}/{relative}")
592 except ValueError as e:
593 _LOGGER.warning("SUPERNOTIFY Invalid media path for URL %s: %s", relative_path, e)
594 return None
596 async def share_path(self, artefact_path: Path) -> str | None:
597 """Return the HA media-share path for a local file (e.g. '/media/images/raw/foo.jpg').
599 Used by mobile push: the companion app resolves share-rooted paths against the HA base URL.
600 Returns None when media_path is unconfigured or path is outside the media tree.
601 """
602 if self.media_path is None or self.media_url_prefix is None:
603 return None
604 try:
605 if artefact_path.is_absolute():
606 relative = artefact_path.relative_to(await self.media_path.absolute())
607 else:
608 relative = artefact_path
609 return str(Path(self.media_url_prefix) / relative)
610 except ValueError as e:
611 _LOGGER.debug("SUPERNOTIFY Failure creating shared media path for %s:%s", artefact_path, e)
612 return None
614 async def size(self) -> int:
615 path: Path | None = self.media_path
616 if path and await path.exists():
617 return sum(1 for p in await aiofiles.os.listdir(path))
618 return 0
620 async def cleanup(self, days: int | None = None, force: bool = False) -> int:
621 if (
622 not force
623 and self.last_purge is not None
624 and self.last_purge > dt.datetime.now(dt.UTC) - dt.timedelta(minutes=self.purge_minute_interval)
625 ):
626 _LOGGER.debug(
627 "SUPERNOTIFY Media storage cleanup skipped, force: %s, last purge: %s, purge interval: %s",
628 force,
629 self.last_purge,
630 self.purge_minute_interval,
631 )
632 return 0
633 days = days or self.days
634 if days == 0 or self.media_path is None:
635 _LOGGER.debug("SUPERNOTIFY Media storage cleanup skipped, %s days, %s", days, self.media_path)
636 return 0
638 cutoff = dt.datetime.now(dt.UTC) - dt.timedelta(days=days)
639 cutoff = cutoff.astimezone(dt.UTC)
640 purged: int = 0
641 skipped: int = 0
642 if self.media_path and await self.media_path.exists():
643 try:
644 queue: list[Path] = [self.media_path]
645 while queue:
646 current = queue.pop()
647 for entry in await aiofiles.os.scandir(current):
648 if entry.is_dir():
649 queue.append(Path(entry.path))
650 elif entry.is_file() and dt_util.utc_from_timestamp(entry.stat().st_mtime) <= cutoff:
651 _LOGGER.debug("SUPERNOTIFY Purging %s", entry.path)
652 await aiofiles.os.unlink(Path(entry.path))
653 purged += 1
654 else:
655 skipped += 1
656 except Exception as e:
657 _LOGGER.warning("SUPERNOTIFY Unable to clean up media storage at %s: %s", self.media_path, e, exc_info=True)
658 _LOGGER.info("SUPERNOTIFY Purged %s media storage for cutoff %s, skipped %s", purged, cutoff, skipped)
659 self.last_purge = dt.datetime.now(dt.UTC)
660 else:
661 _LOGGER.warning("SUPERNOTIFY Skipping media storage cleanup for unknown path %s", self.media_path)
662 return purged