Coverage for custom_components/supernotify/hass_api.py: 97%
740 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-08 17:42 +0000
1from __future__ import annotations
3import logging
4from collections.abc import Generator, Mapping
5from dataclasses import dataclass, field
6from functools import partial
7from typing import TYPE_CHECKING, Any
9from homeassistant.components.person import ATTR_USER_ID
10from homeassistant.const import (
11 ATTR_AREA_ID,
12 ATTR_ENTITY_ID,
13 ATTR_FLOOR_ID,
14 ATTR_LABEL_ID,
15 CONF_ACTION,
16 CONF_CONDITIONS,
17 CONF_DEVICE_ID,
18)
19from homeassistant.helpers.aiohttp_client import async_get_clientsession
20from homeassistant.helpers.debounce import Debouncer
21from homeassistant.helpers.entity_registry import RegistryEntry
22from homeassistant.helpers.event import async_track_state_change_event, async_track_time_change, async_track_time_interval
23from homeassistant.helpers.storage import Store
24from homeassistant.helpers.target import TargetSelection, async_extract_referenced_entity_ids
25from homeassistant.util import slugify
27from .compat import vol
29if TYPE_CHECKING:
30 import asyncio
31 from collections.abc import Awaitable, Callable, Iterable
33 import aiohttp
34 from anyio import Path
35 from homeassistant.core import CALLBACK_TYPE, HomeAssistant, Service, ServiceResponse, State
36 from homeassistant.helpers.entity_registry import EntityRegistry
37 from homeassistant.helpers.typing import ConfigType
38 from homeassistant.util.event_type import EventType
40 from .schema import ConditionsFunc
42import socket
43import threading
44from contextlib import contextmanager
45from datetime import timedelta
46from typing import TYPE_CHECKING, cast
48import homeassistant.components.camera as ha_camera
49import homeassistant.components.image as ha_image
50import homeassistant.components.trace
51from homeassistant.components.group import DOMAIN as GROUP_DOMAIN
52from homeassistant.components.group import expand_entity_ids
53from homeassistant.components.trace.const import DATA_TRACE
54from homeassistant.components.trace.models import ActionTrace
55from homeassistant.components.trace.util import async_store_trace
56from homeassistant.const import ATTR_DOMAIN, ATTR_SERVICE, EVENT_SERVICE_REGISTERED, EVENT_SERVICE_REMOVED
57from homeassistant.core import Context as HomeAssistantContext
58from homeassistant.core import Event, HomeAssistant, SupportsResponse, callback
59from homeassistant.exceptions import ConditionError, ConditionErrorContainer, HomeAssistantError, IntegrationError
60from homeassistant.helpers import condition as condition_helper
61from homeassistant.helpers import device_registry as dr
62from homeassistant.helpers import entity_registry as er
63from homeassistant.helpers import issue_registry as ir
64from homeassistant.helpers.json import json_dumps
65from homeassistant.helpers.network import get_url
66from homeassistant.helpers.template import Template
67from homeassistant.helpers.trace import trace_get, trace_path
68from homeassistant.helpers.translation import async_get_translations
69from homeassistant.helpers.typing import ConfigType
71from . import DOMAIN
72from .common import spoken_name
73from .const import CONF_DEVICE_LABELS, CONF_DEVICE_TRACKER, CONF_MOBILE_APP_ID
74from .model import ConditionVariables, SelectionRule
76if TYPE_CHECKING:
77 from homeassistant.helpers.device_registry import DeviceEntry, DeviceRegistry
79# avoid importing from homeassistant.components.mobile_app.const and triggering dependency chain
81CONF_USER_ID = "user_id"
82ATTR_OS_NAME = "os_name"
83ATTR_OS_VERSION = "os_version"
84ATTR_APP_VERSION = "app_version"
85ATTR_DEVICE_NAME = "device_name"
86ATTR_MANUFACTURER = "manufacturer"
87ATTR_MODEL = "model"
89MOBILE_APP_DOMAIN = "mobile_app"
91_LOGGER = logging.getLogger(__name__)
94@dataclass
95class TrackedDeviceDetails:
96 device_id: str
97 device_labels: list[str] | None = None
98 mobile_app_id: str | None = None
99 device_name: str | None = None
100 device_tracker: str | None = None
101 action: str | None = None
102 user_id: str | None = None
103 area_id: str | None = None
104 manufacturer: str | None = None
105 model: str | None = None
106 os_name: str | None = None
107 os_version: str | None = None
108 app_version: str | None = None
109 identifiers: set[tuple[str, str]] | None = None
111 def as_dict(self) -> dict[str, str | list[str] | None]:
112 return {
113 CONF_MOBILE_APP_ID: self.mobile_app_id,
114 ATTR_DEVICE_NAME: self.device_name,
115 CONF_DEVICE_ID: self.device_id,
116 CONF_USER_ID: self.user_id,
117 CONF_DEVICE_TRACKER: self.device_tracker,
118 CONF_ACTION: self.action,
119 ATTR_OS_NAME: self.os_name,
120 ATTR_OS_VERSION: self.os_version,
121 ATTR_APP_VERSION: self.app_version,
122 ATTR_MANUFACTURER: self.manufacturer,
123 ATTR_MODEL: self.model,
124 CONF_DEVICE_LABELS: self.device_labels,
125 }
127 def __eq__(self, other: object) -> bool:
128 """Test support"""
129 as_dict = getattr(other, "as_dict", None)
130 return other is not None and callable(as_dict) and as_dict() == self.as_dict()
133@dataclass
134class TargetSelectorResolution:
135 """Outcome of resolving area/floor/label selectors to the entities they reference"""
137 entity_ids: list[str] = field(default_factory=list)
138 missing_areas: list[str] = field(default_factory=list)
139 missing_floors: list[str] = field(default_factory=list)
140 missing_labels: list[str] = field(default_factory=list)
142 def has_missing(self) -> bool:
143 return bool(self.missing_areas or self.missing_floors or self.missing_labels)
146def ha_device_info(entry_id: str) -> dr.DeviceInfo:
147 """Home Assistant device-registry DeviceInfo for the single 'SuperNotify' device.
149 Groups the platform entities in binary_sensor.py/sensor.py/switch.py (scenario/recipient
150 state and control, notification/failure counters) under one device in the HA device
151 registry. Unrelated to `TrackedDeviceDetails` above, which describes a discovered mobile_app
152 device for targeting.
153 """
154 return dr.DeviceInfo(
155 identifiers={(DOMAIN, entry_id)},
156 name="Supernotify",
157 manufacturer="Supernotify",
158 entry_type=dr.DeviceEntryType.SERVICE,
159 )
162def ha_transport_device_info(entry_id: str, name: str, title: str) -> dr.DeviceInfo:
163 """Home Assistant device-registry DeviceInfo for one transport, as a service, holding its
164 switch and its deliveries', in its subentry - as the SMTP integration has a service device
165 for each mail server"""
166 return dr.DeviceInfo(
167 identifiers={(DOMAIN, f"{entry_id}_transport_{name}")},
168 name=title,
169 manufacturer="Supernotify",
170 model="Transport",
171 entry_type=dr.DeviceEntryType.SERVICE,
172 )
175def _describe_unsubscribe(unsub: Callable[[], Any]) -> str:
176 """A short, cheap label for an unsubscribe callable, for logging.
178 Never str() the callable itself: the one returned by async_track_state_change_event is a
179 partial over Home Assistant's shared dict of every state-change listener in the instance,
180 so rendering it formats all of them on the event loop (megabytes on a large instance),
181 and `%.100s` only truncates after the whole string has been built.
182 """
183 target = getattr(unsub, "func", unsub) # functools.partial -> the wrapped function
184 return str(getattr(target, "__qualname__", type(target).__name__))
187class HomeAssistantAPI:
188 def __init__(self, hass: HomeAssistant) -> None:
189 self._hass: HomeAssistant = hass
190 self.internal_url: str = ""
191 self.external_url: str = ""
192 self.language: str = ""
193 self.hass_name: str = "!UNDEFINED!"
194 self.__entity_registry: er.EntityRegistry | None = None
195 self.__device_registry: dr.DeviceRegistry | None = None
196 self._service_info: dict[tuple[str, str], Any] = {}
197 self.unsubscribes: list[CALLBACK_TYPE] = []
198 self._mobile_apps_by_tracker: dict[str, TrackedDeviceDetails] = {}
199 self._mobile_apps_by_app_id: dict[str, TrackedDeviceDetails] = {}
200 self._mobile_apps_by_device_id: dict[str, TrackedDeviceDetails] = {}
201 self._mobile_apps_by_user_id: dict[str, list[TrackedDeviceDetails]] = {}
203 def initialize(self) -> None:
204 self.hass_name = self._hass.config.location_name
205 self.language = self._hass.config.language
206 try:
207 self.internal_url = get_url(self._hass, prefer_external=False)
208 except Exception as e:
209 self.internal_url = f"http://{socket.gethostname()}"
210 _LOGGER.warning("SUPERNOTIFY Internal hass url not available, defaulting to %s: %s", self.internal_url, e)
211 try:
212 self.external_url = get_url(self._hass, prefer_external=True)
213 except Exception as e:
214 _LOGGER.warning("SUPERNOTIFY External hass url not available, defaulting to internal url: %s", e)
215 self.external_url = self.internal_url
217 self.build_mobile_app_cache()
219 _LOGGER.debug(
220 "SUPERNOTIFY Configured for HomeAssistant instance %s at %s , %s",
221 self.hass_name,
222 self.internal_url,
223 self.external_url,
224 )
226 if not self.internal_url or not self.internal_url.startswith("http"):
227 _LOGGER.warning("SUPERNOTIFY Invalid internal hass url %s", self.internal_url)
229 def disconnect(self) -> None:
230 while self.unsubscribes:
231 unsub = self.unsubscribes.pop()
232 try:
233 _LOGGER.debug("SUPERNOTIFY Unsubscribing: %s", _describe_unsubscribe(unsub))
234 unsub()
235 except Exception as e:
236 _LOGGER.error("SUPERNOTIFY Failed to unsubscribe: %s", e)
237 _LOGGER.debug("SUPERNOTIFY Disconnection complete")
239 def subscribe_event(self, event: EventType[Any] | str, callback: Callable) -> None:
240 self.unsubscribes.append(self._hass.bus.async_listen(event, callback))
242 def subscribe_mobile_app_changes(self, on_change: Callable[[], Awaitable[None]], cooldown: float = 10) -> None:
243 """Call `on_change` once things settle after a mobile_app device, or its notify action, comes or goes
245 Re-pairing or renaming a phone changes its notify action, and HA registers that action a
246 little after the device itself, so both kinds of change are debounced into one call
247 """
248 debouncer: Debouncer = Debouncer(self._hass, _LOGGER, cooldown=cooldown, immediate=False, function=on_change)
250 @callback
251 def on_device(event: Event) -> None:
252 device_id: str | None = event.data.get("device_id")
253 if event.data.get("action") == "remove":
254 relevant: bool = device_id in self._mobile_apps_by_device_id
255 else:
256 dev_reg: DeviceRegistry | None = self._device_registry()
257 device = dev_reg.async_get(device_id) if dev_reg and device_id else None
258 relevant = device is not None and any(i and i[0] == MOBILE_APP_DOMAIN for i in device.identifiers)
259 if relevant:
260 debouncer.async_schedule_call()
262 @callback
263 def on_service(event: Event) -> None:
264 if event.data.get(ATTR_DOMAIN) == "notify" and str(event.data.get(ATTR_SERVICE, "")).startswith(
265 f"{MOBILE_APP_DOMAIN}_"
266 ):
267 debouncer.async_schedule_call()
269 self.subscribe_event(dr.EVENT_DEVICE_REGISTRY_UPDATED, on_device)
270 self.subscribe_event(EVENT_SERVICE_REGISTERED, on_service)
271 self.subscribe_event(EVENT_SERVICE_REMOVED, on_service)
272 self.unsubscribes.append(debouncer.async_cancel)
274 def subscribe_state(self, entity_ids: str | Iterable[str], callback: Callable) -> None:
275 self.unsubscribes.append(async_track_state_change_event(self._hass, entity_ids, callback))
277 def subscribe_time(self, hour: int, minute: int, second: int, callback: Callable) -> None:
278 self.unsubscribes.append(async_track_time_change(self._hass, callback, hour=hour, minute=minute, second=second))
280 def subscribe_interval(self, seconds: int, callback: Callable) -> None:
281 self.unsubscribes.append(async_track_time_interval(self._hass, callback, timedelta(seconds=seconds)))
283 def in_hass_loop(self) -> bool:
284 return self.hass_avail("loop_thread_id") and self._hass.loop_thread_id == threading.get_ident()
286 def get_state(self, entity_id: str) -> State | None:
287 return self._hass.states.get(entity_id)
289 def is_state(self, entity_id: str, state: str) -> bool:
290 return self._hass.states.is_state(entity_id, state)
292 def has_service(self, domain: str, service: str) -> bool:
293 return self._hass.services.has_service(domain, service)
295 def integration_loaded(self, domain: str) -> bool:
296 # guarded like hass_avail(), as tests and docgen may run with a mock hass
297 components = getattr(self._hass.config, "components", None) if self.hass_avail("config") else None
298 return isinstance(components, set) and domain in components
300 def mobile_app_action_exists(self, mobile_app_id: str) -> bool:
301 """Whether a mobile app can be notified, by its notify action or its notify entity"""
302 action: str = mobile_app_id.removeprefix("notify.")
303 return self.has_service("notify", action) or self.get_state(f"notify.{action}") is not None
305 def entity_ids_for_domain(self, domain: str) -> list[str]:
306 return self._hass.states.async_entity_ids(domain)
308 async def async_real_user_ids(self) -> dict[str, str]:
309 """Active, non-system HA user ids mapped to their login username - for discovering
310 recipients that have no Person record (see CONF_USER_ID in const.py), excluding
311 internal accounts like Supervisor/Home Assistant Content that aren't real people."""
312 users = await self._hass.auth.async_get_users()
313 result: dict[str, str] = {}
314 for user in users:
315 if not user.is_active or user.system_generated:
316 continue
317 # same lookup HA's own user-management API uses (homeassistant/components/config/auth.py)
318 username = next(
319 (cred.data.get("username") for cred in user.credentials if cred.auth_provider_type == "homeassistant"),
320 None,
321 )
322 result[user.id] = username or user.name or user.id
323 return result
325 def platform_for_entity(self, entity_id: str) -> str | None:
326 """The integration that registered this entity (RegistryEntry.platform), if any."""
327 entity_registry: EntityRegistry | None = self._entity_registry()
328 if entity_registry:
329 reg_entry: RegistryEntry | None = entity_registry.async_get(entity_id) if entity_registry else None
330 if reg_entry:
331 return reg_entry.platform
332 return None
334 def config_path(self, path: str) -> str:
335 """A path relative to the Home Assistant config directory, or the path itself if absolute."""
336 return self._hass.config.path(path)
338 def config_entry_title_for_entity(self, entity_id: str) -> str | None:
339 """The title of the config entry that registered this entity, if any."""
340 entity_registry: EntityRegistry | None = self._entity_registry()
341 reg_entry: RegistryEntry | None = entity_registry.async_get(entity_id) if entity_registry else None
342 if reg_entry is None or reg_entry.config_entry_id is None:
343 return None
344 config_entry = self._hass.config_entries.async_get_entry(reg_entry.config_entry_id)
345 return config_entry.title if config_entry else None
347 def device_for_entity(self, entity_id: str) -> str | None:
348 """The device this entity belongs to, if any."""
349 entity_registry: EntityRegistry | None = self._entity_registry()
350 reg_entry: RegistryEntry | None = entity_registry.async_get(entity_id) if entity_registry else None
351 return reg_entry.device_id if reg_entry else None
353 def entity_spoken_names(self, entity_id: str) -> set[str]:
354 """What an entity can be called, by entity_id, object_id, friendly name or alias - and by
355 object_id then domain, so 'driveway camera' is camera.driveway - as spoken_name() has them"""
356 domain, _, object_id = entity_id.partition(".")
357 names: list[str] = [entity_id, object_id, f"{object_id} {domain}"]
358 state: State | None = self._hass.states.get(entity_id)
359 if state:
360 names.append(state.name)
361 entity_registry: EntityRegistry | None = self._entity_registry()
362 reg_entry: RegistryEntry | None = entity_registry.async_get(entity_id) if entity_registry else None
363 if reg_entry:
364 names.extend(a for a in reg_entry.aliases if isinstance(a, str))
365 # from Home Assistant 2026.11 an entity's name is composed with its device's, but the
366 # friendly name stays as it was until the name is changed, so the composed one is added
367 full_entity_name: Callable[[HomeAssistant, RegistryEntry], str] | None = getattr(
368 er, "async_get_full_entity_name", None
369 )
370 if full_entity_name and (full_name := full_entity_name(self._hass, reg_entry)):
371 names.append(full_name)
372 return {spoken_name(n) for n in names}
374 def sender_entity_ids(self, ha_context: HomeAssistantContext | None) -> list[str]:
375 """The automations and scripts whose current run carries this context, i.e. what sent a notification.
377 An automation or script writes its state with the context of the run it starts, and the actions
378 of that run carry the same context (a script started by an automation gets it as parent), so
379 the sender is the automation or script whose state has the context's id or its parent's
381 WARNING: best effort only, and not to be built on further. Home Assistant has no native way
382 to get the original cause of a context, so this relies on the state still carrying the
383 context of that run - a later or parallel run replaces it, and a sender further up the
384 chain than the parent isn't found. It also scans every automation and script state on each
385 call, so callers should only use it when the answer could matter."""
386 if ha_context is None:
387 return []
388 ids: set[str] = {i for i in (ha_context.id, ha_context.parent_id) if i}
389 return [s.entity_id for s in self._hass.states.async_all(("automation", "script")) if s.context.id in ids]
391 def entity_ids_named(self, name: str) -> list[str]:
392 """Entities that can be called this, see entity_spoken_names()"""
393 wanted: str = spoken_name(name)
394 return [s.entity_id for s in self._hass.states.async_all() if wanted in self.entity_spoken_names(s.entity_id)]
396 def entity_ids_for_platform(
397 self, domain: str, platform: str, device_model_select: str | list[str] | dict | SelectionRule | None = None
398 ) -> list[str]:
399 """entity_ids in `domain` (e.g. "notify") registered by a specific integration.
401 Reads the entity registry directly (not the state machine), so a freshly
402 registered entity counts even before it has reported a first state.
404 `device_model_select` optionally filters by the backing device's model (e.g.
405 `{"exclude": ["Speaker Group"]}`), same include/exclude rule shape used elsewhere.
406 """
407 entity_registry = self._entity_registry()
408 if not entity_registry:
409 return []
410 entries: list[RegistryEntry] = [
411 e for e in entity_registry.entities.values() if e.domain == domain and e.platform == platform
412 ]
413 if device_model_select is not None:
414 model_filter = SelectionRule(device_model_select)
415 entries = [e for e in entries if model_filter.match(self._device_model(e.device_id))]
416 return [e.entity_id for e in entries]
418 def _device_model(self, device_id: str | None) -> str | None:
419 if device_id is None:
420 return None
421 dev_entry = self.find_device(device_id)
422 return dev_entry.model if dev_entry else None
424 async def async_get_camera_image(self, entity_id: str, timeout: int = 10) -> ha_camera.Image | None:
425 """Fetch a still image directly from a camera entity, via HA's own camera component API,
426 rather than triggering the camera.snapshot service and polling the filesystem for the
427 resulting file to appear."""
428 try:
429 return await ha_camera.async_get_image(self._hass, entity_id, timeout=timeout)
430 except HomeAssistantError as e:
431 _LOGGER.warning("SUPERNOTIFY Unable to get camera image for %s: %s", entity_id, e)
432 return None
434 async def async_get_image_entity_image(self, entity_id: str, timeout: int = 10) -> ha_image.Image | None:
435 """Fetch a still image directly from an image entity, via HA's own image component API."""
436 try:
437 return await ha_image.async_get_image(self._hass, entity_id, timeout=timeout)
438 except HomeAssistantError as e:
439 _LOGGER.warning("SUPERNOTIFY Unable to get image from entity %s: %s", entity_id, e)
440 return None
442 async def load_storage(self, key: str, version: int = 1) -> Any: # ruff: ignore[any-type]
443 """Load integration state previously persisted to Home Assistant's .storage/ area,
444 via HA's own Store helper, or None if nothing has been persisted yet for this key."""
445 try:
446 return await Store[Any](self._hass, version, key).async_load()
447 except Exception as e:
448 _LOGGER.warning("SUPERNOTIFY Unable to load storage %s: %s", key, e)
449 return None
451 def save_storage(self, key: str, data: Any, version: int = 1) -> None: # ruff: ignore[any-type]
452 """Persist integration state to Home Assistant's .storage/ area, via HA's own Store
453 helper. Fire-and-forget: the write happens in a tracked background task rather than
454 blocking the caller, since this is called from both sync and async contexts.
455 Sync callers include actions without @callback (e.g. clear_snoozes), which Home Assistant
456 runs in its executor, so off the event loop the thread-safe hass.create_task is used."""
457 store: Store[Any] = Store(self._hass, version, key)
458 if self.in_hass_loop():
459 self._hass.async_create_task(store.async_save(data), f"supernotify_save_{key}")
460 else:
461 self._hass.create_task(store.async_save(data), f"supernotify_save_{key}")
463 def create_job(self, func: Callable, *args: Any) -> asyncio.Future[Any]:
464 """Wrap a blocking function call in a HomeAssistant awaitable job"""
465 return self._hass.async_add_executor_job(func, *args)
467 def fire_event(
468 self, event_name: str, event_data: dict[str, Any] | None = None, context: HomeAssistantContext | None = None
469 ) -> None:
470 self._hass.bus.async_fire(event_name, event_data, context=context)
472 async def call_service(
473 self,
474 domain: str,
475 service: str,
476 service_data: dict[str, Any] | None = None,
477 target: dict[str, Any] | None = None,
478 return_response: bool | None = None,
479 blocking: bool | None = None,
480 debug: bool = False,
481 context: HomeAssistantContext | None = None,
482 ) -> ServiceResponse | None:
484 if return_response is None:
485 # unknown service, for example defined in generic action, check if it supports response
486 supports_response: SupportsResponse = self.service_info(domain, service)
487 if supports_response == SupportsResponse.NONE:
488 return_response = False
489 elif supports_response == SupportsResponse.ONLY:
490 return_response = True
491 else:
492 return_response = debug
493 if blocking is None:
494 blocking = return_response or debug
496 response: ServiceResponse | None = await self._hass.services.async_call(
497 domain,
498 service,
499 service_data=service_data,
500 blocking=blocking,
501 context=context,
502 target=target,
503 return_response=return_response,
504 )
505 if response is not None and debug:
506 _LOGGER.info("SUPERNOTIFY Service %s.%s response: %s", domain, service, response)
507 return response
509 def coerce_schema(self, domain: str, service: str, data: ConfigType) -> ConfigType:
510 if not data:
511 return data
512 try:
513 if (domain, service) not in self._service_info:
514 self.service_info(domain, service)
515 service_info = self._service_info.get((domain, service))
516 if not service_info:
517 _LOGGER.info("SUPERNOTIFY No service found to pre-validate action data for %s.%s", domain, service)
518 return data
519 if not service_info.get("schema"):
520 _LOGGER.info("SUPERNOTIFY No vol schema found to pre-validate action data for %s.%s", domain, service)
521 return data
522 service_schema = service_info["schema"]
524 while service_schema is not None and not (
525 isinstance(service_schema, vol.Schema) and isinstance(service_schema.schema, dict)
526 ):
527 if isinstance(service_schema, vol.Schema):
528 # e.g. entity services get schema wrapped as vol.Schema(vol.All(...))
529 service_schema = service_schema.schema
530 elif hasattr(service_schema, "validators") and hasattr(service_schema.validators, "__iter__"):
531 # e.g. vol.All — strip extras using first dict Schema sub-validator only
532 # (don't run the full chain; other validators may require target fields not in data)
533 service_schema = next(
534 (v for v in service_schema.validators if isinstance(v, vol.Schema) or hasattr(v, "validators")), None
535 )
536 else:
537 service_schema = None
538 if not (isinstance(service_schema, vol.Schema) and isinstance(service_schema.schema, dict)):
539 service_schema = None
540 _LOGGER.info("SUPERNOTIFY Unable to find schema for %s.%s", domain, service)
542 if service_schema:
543 coercing_schema = service_schema.extend(
544 {},
545 extra=vol.REMOVE_EXTRA if service_schema.extra == vol.PREVENT_EXTRA else service_schema.extra,
546 required=service_schema.required,
547 )
548 cleaned = coercing_schema(data)
549 else:
550 return data
551 if cleaned != data:
552 _LOGGER.debug("SUPERNOTIFY Coerced data for %s.%s from %s->%s", domain, service, data, cleaned)
553 return cleaned
554 except Exception:
555 _LOGGER.exception("SUPERNOTIFY Unable to coerce %s.%s schema for %s", domain, service, data)
556 return data
558 def service_info(self, domain: str, service: str) -> SupportsResponse:
559 supports_response: SupportsResponse | None = None
560 try:
561 if (domain, service) not in self._service_info:
562 service_objs: dict[str, Service] = self._hass.services.async_services_for_domain(domain)
563 service_obj: Service | None = service_objs.get(service)
564 if service_obj:
565 self._service_info[domain, service] = {
566 "supports_response": service_obj.supports_response,
567 "schema": service_obj.schema,
568 }
569 service_info: dict[str, Any] = self._service_info.get((domain, service), {})
570 supports_response = service_info.get("supports_response")
571 if supports_response is None:
572 _LOGGER.debug("SUPERNOTIFY Unable to find service info for %s.%s", domain, service)
574 except Exception as e:
575 _LOGGER.warning("SUPERNOTIFY Unable to get service info for %s.%s: %s", domain, service, e)
576 return supports_response or SupportsResponse.NONE # default to no response
578 def resolve_target_selectors(
579 self,
580 area_ids: list[str] | None = None,
581 floor_ids: list[str] | None = None,
582 label_ids: list[str] | None = None,
583 ) -> TargetSelectorResolution:
584 """Resolve HA area/floor/label selectors to the entities they reference, through the same
585 core helper as HA entity actions, so groups are expanded and entities inherit the area of
586 their device.
587 """
588 resolution = TargetSelectorResolution()
589 if not (area_ids or floor_ids or label_ids):
590 return resolution
591 selection: dict[str, list[str]] = {}
592 if area_ids:
593 selection[ATTR_AREA_ID] = list(area_ids)
594 if floor_ids:
595 selection[ATTR_FLOOR_ID] = list(floor_ids)
596 if label_ids:
597 selection[ATTR_LABEL_ID] = list(label_ids)
598 try:
599 selected = async_extract_referenced_entity_ids(self._hass, TargetSelection(selection), expand_group=True)
600 resolution.entity_ids = sorted(selected.referenced | selected.indirectly_referenced)
601 resolution.missing_areas = sorted(selected.missing_areas)
602 resolution.missing_floors = sorted(selected.missing_floors)
603 resolution.missing_labels = sorted(selected.missing_labels)
604 except Exception as e:
605 _LOGGER.warning("SUPERNOTIFY Unable to resolve target selectors %s: %s", selection, e)
606 return resolution
608 def resolve_target_exclusion(self, exclude: dict[str, list[str]]) -> tuple[set[str], set[str]]:
609 """Entity ids and device ids an `exclude` block of a target leaves out, as Home Assistant
610 2026.11 does for entity actions - config and diagnostic entities included, and with the
611 members of any group, so one can't come back when its group is expanded.
612 """
613 try:
614 selected = async_extract_referenced_entity_ids(
615 self._hass, TargetSelection(exclude), expand_group=True, primary_entities_only=False
616 )
617 except Exception as e:
618 _LOGGER.warning("SUPERNOTIFY Unable to resolve target exclusion %s: %s", exclude, e)
619 return set(), set()
620 entity_ids: set[str] = selected.referenced | selected.indirectly_referenced
621 for entity_id in list(entity_ids):
622 entity_ids.update(self.group_members(entity_id) or [])
623 return entity_ids, set(selected.referenced_devices)
625 def find_service(self, domain: str, module: str) -> str | None:
626 try:
627 service_objs: dict[str, Service] = self._hass.services.async_services_for_domain(domain)
628 if service_objs:
629 for service, domain_obj in service_objs.items():
630 if domain_obj.job is not None:
631 target = domain_obj.job.target
632 bound_self = getattr(target, "__self__", None)
633 target_module: str | None = bound_self.__module__ if bound_self is not None else target.__module__
634 if target_module == module:
635 # Legacy notify platforms with a targets property (e.g. alexa_media_player)
636 # register extra per-target services (notify.<platform>_<device>) that share
637 # this same bound method but hard-code their own target, ignoring any target:
638 # passed by the caller - skip those and hold out for the base platform service.
639 registered_targets = getattr(bound_self, "registered_targets", None)
640 if registered_targets is not None and service in registered_targets:
641 continue
642 _LOGGER.debug("SUPERNOTIFY Found service %s for domain %s in %s", service, domain, module)
643 return f"{domain}.{service}"
645 _LOGGER.debug("SUPERNOTIFY Unable to find service for %s, module %s", domain, module)
646 except Exception as e:
647 _LOGGER.warning("SUPERNOTIFY Unable to find service for %s, module %s : %s", domain, module, e)
648 return None
650 def smtp_recipients(self) -> dict[str, str]:
651 """E-mail address of each recipient set up in Home Assistant's SMTP integration, to its notify entity"""
652 recipients: dict[str, str] = {}
653 registry: er.EntityRegistry | None = self._entity_registry()
654 if registry is None or not self.hass_avail("config_entries"):
655 return recipients
656 try:
657 for entry in self._hass.config_entries.async_entries("smtp", include_ignore=False, include_disabled=False):
658 for subentry in entry.subentries.values():
659 # the smtp integration gives each recipient's notify entity the unique id `<entry id>_<address>`
660 if subentry.unique_id and (
661 entity_id := registry.async_get_entity_id("notify", "smtp", f"{entry.entry_id}_{subentry.unique_id}")
662 ):
663 recipients[subentry.unique_id] = entity_id
664 except Exception as e:
665 _LOGGER.warning("SUPERNOTIFY Unable to find SMTP recipients: %s", e)
666 return recipients
668 @property
669 def media_dirs(self) -> dict[str, str]:
670 """Folders Home Assistant's media source serves, by their name in `media-source://media_source/<name>/`"""
671 return dict(self._hass.config.media_dirs)
673 def find_config_entry_data(self, domain: str) -> Mapping[str, Any] | None:
674 """Return the data of the first enabled, non-ignored config entry for domain, if any."""
675 if not self.hass_avail("config_entries"):
676 return None
677 try:
678 entries = self._hass.config_entries.async_entries(domain, include_ignore=False, include_disabled=False)
679 if entries:
680 return entries[0].data
681 except Exception as e:
682 _LOGGER.warning("SUPERNOTIFY Unable to find config entry for %s: %s", domain, e)
683 return None
685 def http_session(self) -> aiohttp.ClientSession:
686 """Client aiohttp session for async web requests"""
687 return async_get_clientsession(self._hass)
689 async def translations(self, category: str, language: str | None = None) -> dict[str, str]:
690 """Supernotify's own strings for a translation category, in the given language or else the
691 configured one, keyed `component.supernotify.<category>.<key>`, English filling in any not translated"""
692 return await async_get_translations(self._hass, language or self._hass.config.language, category, [DOMAIN])
694 def expand_group(self, entity_ids: str | list[str]) -> list[str]:
695 return expand_entity_ids(self._hass, entity_ids)
697 def group_members(self, entity_id: str, _seen: set[str] | None = None) -> list[str] | None:
698 """Fully expanded members of a `group.*` helper or a platform group (media_player, light... groups
699 created by the group integration expose members in an `entity_id` state attribute). None if not a group.
701 Other entities, e.g. `scene.*` or min/max `sensor.*`, also expose an `entity_id` attribute, so anything
702 outside the `group` domain is only treated as a group if the entity registry says its platform is `group`.
703 """
704 if not self.hass_avail("states"):
705 return None
706 state = self._hass.states.get(entity_id)
707 members = state.attributes.get(ATTR_ENTITY_ID) if state else None
708 if not isinstance(members, (list, tuple)):
709 return None
710 if entity_id.partition(".")[0] != GROUP_DOMAIN and self.platform_for_entity(entity_id) != GROUP_DOMAIN:
711 return None
712 seen: set[str] = _seen if _seen is not None else set()
713 seen.add(entity_id)
714 expanded: list[str] = []
715 for member in members:
716 if member in seen:
717 continue
718 nested = self.group_members(member, seen)
719 for e in nested if nested is not None else [member]:
720 if e not in expanded:
721 expanded.append(e)
722 return expanded
724 def template(self, template_format: str) -> Template:
725 return Template(template_format, self._hass)
727 def hass_avail(self, property: str) -> bool:
728 """Guard for HA functionality, largely for tests or docgen"""
729 return self._hass is not None and getattr(self._hass, property, None) is not None
731 async def register_web_path(self, media_web_path: Path | None, url_prefix: str) -> bool:
732 if media_web_path is None or not self.hass_avail("http"):
733 return False
734 try:
735 from homeassistant.components.http import StaticPathConfig
737 await self._hass.http.async_register_static_paths([
738 StaticPathConfig(url_prefix, str(media_web_path), cache_headers=False)
739 ])
740 return True
741 except Exception as e:
742 _LOGGER.error("SUPERNOTIFY Unable to register media web exposed path for %s: %s", media_web_path, e)
743 return False
745 async def trace_conditions(
746 self,
747 conditions: ConditionsFunc,
748 condition_variables: ConditionVariables,
749 trace_name: str | None = None,
750 ) -> tuple[bool | None, ActionTrace | None]:
752 result: bool | None = None
753 this_trace: ActionTrace | None = None
754 if DATA_TRACE not in self._hass.data:
755 _LOGGER.warning("SUPERNOTIFY Tracing not configured, attempting to set up")
757 await homeassistant.components.trace.async_setup(self._hass, {}) # type: ignore
758 with trace_action(self._hass, trace_name or "anon_condition") as cond_trace:
759 cond_trace.set_trace(trace_get())
760 this_trace = cond_trace
761 with trace_path(["condition", "conditions"]) as _tp:
762 result = self.evaluate_conditions(conditions, condition_variables)
763 _LOGGER.debug(cond_trace.as_dict())
764 return result, this_trace
766 async def build_conditions(
767 self, condition_config: list[ConfigType], strict: bool = False, validate: bool = False, name: str = DOMAIN
768 ) -> ConditionsFunc | None:
769 capturing_logger: ConditionErrorLoggingAdaptor = ConditionErrorLoggingAdaptor(_LOGGER)
770 condition_variables: ConditionVariables = ConditionVariables()
771 cond_list: list[ConfigType]
772 try:
773 if validate:
774 cond_list = cast(
775 "list[ConfigType]", await condition_helper.async_validate_conditions_config(self._hass, condition_config)
776 )
777 else:
778 cond_list = condition_config
779 except Exception:
780 _LOGGER.exception("SUPERNOTIFY Conditions validation failed")
781 raise
782 try:
783 if strict:
784 # can't be an error, the entity may belong to an integration that hasn't loaded yet
785 for cond in cond_list:
786 for entity_id in sorted(condition_helper.async_extract_entities(cond)):
787 if self._hass.states.get(entity_id) is None:
788 _LOGGER.warning("SUPERNOTIFY Condition for %s refers to unknown entity %s", name, entity_id)
789 force_strict_template_mode(cond_list, undo=False)
791 test: ConditionsFunc = await condition_helper.async_conditions_from_config(
792 self._hass, cond_list, cast("logging.Logger", capturing_logger), name
793 )
794 if test is None:
795 raise IntegrationError(f"Invalid condition {condition_config}")
796 test(condition_variables.as_dict())
797 if strict:
798 # compound conditions short-circuit, so a nested template may never have been
799 # reached by the evaluation above - check each one on its own, unless it
800 # can't be evaluated yet for lack of its entity
801 for template_cond in nested_template_conditions(cond_list):
802 if any(self._hass.states.get(e) is None for e in condition_helper.async_extract_entities(template_cond)):
803 continue
804 force_strict_template_mode([template_cond], undo=False)
805 try:
806 nested_test: ConditionsFunc = await condition_helper.async_conditions_from_config(
807 self._hass, [template_cond], cast("logging.Logger", capturing_logger), name
808 )
809 nested_test(condition_variables.as_dict())
810 finally:
811 force_strict_template_mode([template_cond], undo=True)
812 return test
813 except Exception:
814 _LOGGER.exception("SUPERNOTIFY Conditions eval failed")
815 raise
816 finally:
817 if strict:
818 force_strict_template_mode(condition_config, undo=True)
819 if strict and capturing_logger.condition_errors and len(capturing_logger.condition_errors) > 0:
820 for exception in capturing_logger.condition_errors:
821 _LOGGER.warning("SUPERNOTIFY Invalid condition %s:%s", condition_config, exception)
822 raise capturing_logger.condition_errors[0]
824 def evaluate_conditions(
825 self,
826 conditions: ConditionsFunc,
827 condition_variables: ConditionVariables,
828 ) -> bool | None:
829 try:
830 if not condition_variables:
831 _LOGGER.warning("SUPERNOTIFY No cond vars provided for condition")
832 return conditions(condition_variables.as_dict() if condition_variables is not None else None)
833 except Exception as e:
834 _LOGGER.error("SUPERNOTIFY Condition eval failed: %s", e)
835 raise
837 def abs_url(self, fragment: str | None, prefer_external: bool = True) -> str | None:
838 base_url = self.external_url if prefer_external else self.internal_url
839 if fragment:
840 if fragment.startswith("http"):
841 return fragment
842 if fragment.startswith("/"):
843 return base_url + fragment
844 return base_url + "/" + fragment
845 return None
847 def raise_issue(
848 self,
849 issue_id: str,
850 issue_key: str,
851 issue_map: dict[str, str],
852 severity: ir.IssueSeverity = ir.IssueSeverity.WARNING,
853 learn_more_url: str = "https://supernotify.rhizomatics.org.uk",
854 is_fixable: bool = False,
855 ) -> None:
856 ir.async_create_issue(
857 self._hass,
858 DOMAIN,
859 issue_id,
860 translation_key=issue_key,
861 translation_placeholders=issue_map,
862 severity=severity,
863 learn_more_url=learn_more_url,
864 is_fixable=is_fixable,
865 )
867 def delete_issue(self, issue_id: str) -> None:
868 ir.async_delete_issue(self._hass, DOMAIN, issue_id)
870 def mobile_app_by_tracker(self, device_tracker: str) -> TrackedDeviceDetails | None:
871 return self._mobile_apps_by_tracker.get(device_tracker)
873 def mobile_app_by_id(self, mobile_app_id: str) -> TrackedDeviceDetails | None:
874 mobile_app_id = mobile_app_id.replace("notify.", "", 1) if mobile_app_id.startswith("notify.") else mobile_app_id
875 return self._mobile_apps_by_app_id.get(mobile_app_id)
877 def mobile_app_by_device_id(self, device_id: str) -> TrackedDeviceDetails | None:
878 return self._mobile_apps_by_device_id.get(device_id)
880 def mobile_app_by_user_id(self, user_id: str) -> list[TrackedDeviceDetails] | None:
881 return self._mobile_apps_by_user_id.get(user_id)
883 def build_mobile_app_cache(self) -> None:
884 """All enabled mobile apps"""
885 entity_registry: EntityRegistry | None = self._entity_registry()
886 if not entity_registry:
887 _LOGGER.warning("SUPERNOTIFY Unable to discover devices for - no entity registry found")
888 return
890 # rebuilt from scratch, so a device that has gone, or been re-paired under a new name, goes too
891 self._mobile_apps_by_tracker.clear()
892 self._mobile_apps_by_app_id.clear()
893 self._mobile_apps_by_device_id.clear()
894 self._mobile_apps_by_user_id.clear()
895 found: int = 0
896 complete: int = 0
897 for mobile_app_info in self.discover_devices(MOBILE_APP_DOMAIN):
898 try:
899 mobile_app_id: str = f"mobile_app_{slugify(mobile_app_info.device_name)}"
900 device_tracker: str | None = None
901 notify_action: str | None = None
902 if self.has_service("notify", mobile_app_id):
903 notify_action = f"notify.{mobile_app_id}"
904 else:
905 _LOGGER.debug("SUPERNOTIFY No notify action <%s>, so not a notifiable mobile app", mobile_app_id)
907 registry_entries = entity_registry.entities.get_entries_for_device_id(mobile_app_info.device_id)
908 for reg_entry in registry_entries:
909 if reg_entry.platform == "mobile_app" and reg_entry.domain == "device_tracker":
910 device_tracker = reg_entry.entity_id
912 if device_tracker and notify_action:
913 complete += 1
915 mobile_app_info.mobile_app_id = mobile_app_id
916 mobile_app_info.device_tracker = device_tracker
917 mobile_app_info.action = notify_action
919 found += 1
920 self._mobile_apps_by_app_id[mobile_app_id] = mobile_app_info
921 self._mobile_apps_by_device_id[mobile_app_info.device_id] = mobile_app_info
922 if device_tracker:
923 self._mobile_apps_by_tracker[device_tracker] = mobile_app_info
924 if mobile_app_info.user_id is not None:
925 self._mobile_apps_by_user_id.setdefault(mobile_app_info.user_id, [])
926 self._mobile_apps_by_user_id[mobile_app_info.user_id].append(mobile_app_info)
928 except Exception as e:
929 _LOGGER.error("SUPERNOTIFY Failure examining device %s: %s", mobile_app_info, e)
931 _LOGGER.info(f"SUPERNOTIFY Found {found} enabled mobile app devices, {complete} complete config")
933 def device_config_info(self, device: DeviceEntry) -> dict[str, str | None]:
934 results: dict[str, str | None] = {ATTR_OS_NAME: None, ATTR_OS_VERSION: None, CONF_USER_ID: None, ATTR_APP_VERSION: None}
935 try:
936 # HA 2026.8+ restricts devices to a single config entry
937 config_entry_ids: Iterable[str] = (device.config_entry_id,) # type: ignore[attr-defined]
938 except AttributeError:
939 # pre-2026.8: config_entry_id doesn't exist yet, fall back to deprecated plural set
940 config_entry_ids = device.config_entries
941 for config_entry_id in config_entry_ids:
942 config_entry = self._hass.config_entries.async_get_entry(config_entry_id)
943 if config_entry and config_entry.data:
944 for attr, value in results.items():
945 results[attr] = config_entry.data.get(attr) or value
946 return results
948 def discover_devices(
949 self,
950 discover_domain: str,
951 device_model_select: SelectionRule | None = None,
952 device_manufacturer_select: SelectionRule | None = None,
953 device_os_select: SelectionRule | None = None,
954 device_area_select: SelectionRule | None = None,
955 device_label_select: SelectionRule | None = None,
956 ) -> list[TrackedDeviceDetails]:
957 devices: list[TrackedDeviceDetails] = []
958 dev_reg: DeviceRegistry | None = self._device_registry()
959 if dev_reg is None or not hasattr(dev_reg, "devices"):
960 _LOGGER.warning(f"SUPERNOTIFY Unable to discover devices for {discover_domain} - no device registry found")
961 return []
963 all_devs = enabled_devs = found_devs = skipped_devs = 0
964 all_ha_devices: Iterable[DeviceEntry]
965 if isinstance(dev_reg.devices, Mapping):
966 # 2026.8 HA and prior
967 all_ha_devices = dev_reg.devices.values()
968 else:
969 all_ha_devices = dev_reg.devices # type: ignore[assignment]
970 for dev in all_ha_devices:
971 all_devs += 1
973 if dev.disabled:
974 _LOGGER.debug("SUPERNOTIFY Excluded disabled device %s", dev.name)
975 else:
976 enabled_devs += 1
977 for identifier in dev.identifiers:
978 if identifier and len(identifier) > 1 and identifier[0] == discover_domain:
979 _LOGGER.debug("SUPERNOTIFY Discovered %s device %s for id %s", dev.model, dev.name, identifier)
980 found_devs += 1
981 if device_model_select is not None and not device_model_select.match(dev.model):
982 _LOGGER.debug("SUPERNOTIFY Skipped dev %s, no model %s match", dev.name, dev.model)
983 skipped_devs += 1
984 continue
985 if device_manufacturer_select is not None and not device_manufacturer_select.match(dev.manufacturer):
986 _LOGGER.debug("SUPERNOTIFY Skipped dev %s, no manufacturer %s match", dev.name, dev.manufacturer)
987 skipped_devs += 1
988 continue
989 device_config_info = self.device_config_info(dev)
990 if device_os_select is not None and not device_os_select.match(device_config_info[ATTR_OS_NAME]):
991 _LOGGER.debug(
992 "SUPERNOTIFY Skipped dev %s, no OS %s match", dev.name, device_config_info[ATTR_OS_NAME]
993 )
994 skipped_devs += 1
995 continue
996 if device_area_select is not None and not device_area_select.match(dev.area_id):
997 _LOGGER.debug("SUPERNOTIFY Skipped dev %s, no area %s match", dev.name, dev.area_id)
998 skipped_devs += 1
999 continue
1000 if device_label_select is not None and not device_label_select.match(dev.labels):
1001 _LOGGER.debug("SUPERNOTIFY Skipped dev %s, no label %s match", dev.name, dev.labels)
1002 skipped_devs += 1
1003 continue
1004 devices.append(
1005 TrackedDeviceDetails(
1006 device_id=dev.id,
1007 device_name=dev.name,
1008 manufacturer=dev.manufacturer,
1009 model=dev.model,
1010 area_id=dev.area_id,
1011 user_id=device_config_info[ATTR_USER_ID],
1012 os_name=device_config_info[ATTR_OS_NAME],
1013 os_version=device_config_info[ATTR_OS_VERSION],
1014 app_version=device_config_info[ATTR_APP_VERSION],
1015 device_labels=list(dev.labels) if dev.labels else [],
1016 identifiers=dev.identifiers,
1017 )
1018 )
1020 elif identifier:
1021 # HomeKit has triples for identifiers, other domains may behave similarly
1022 _LOGGER.debug("SUPERNOTIFY Ignoring device %s id: %s", dev.name, identifier)
1023 else:
1024 _LOGGER.debug( # type: ignore
1025 "SUPERNOTIFY Unexpected %s device %s without id", dev.model, dev.name
1026 )
1028 _LOGGER.debug("SUPERNOTIFY {discover_domain} device discovery, all=%s,enabled=%s ", all_devs, enabled_devs)
1029 _LOGGER.debug("SUPERNOTIFY {discover_domain} skipped=%s, found=%s", skipped_devs, found_devs)
1031 return devices
1033 def domain_for_device(self, device_id: str, domains: list[str]) -> str | None:
1034 # discover domain from device registry
1035 verified_domain: str | None = None
1036 device_registry: DeviceRegistry | None = self._device_registry()
1037 if device_registry:
1038 device: DeviceEntry | None = self.find_device(device_id)
1039 if device:
1040 matching_domains = [d for d, _id in device.identifiers if d in domains]
1041 if matching_domains:
1042 # TODO: limited to first domain found, unlikely to be more
1043 return matching_domains[0]
1044 _LOGGER.warning(
1045 "SUPERNOTIFY A target that looks like a device_id can't be matched to supported integration: %s",
1046 device_id,
1047 )
1048 return verified_domain
1050 def _entity_registry(self) -> er.EntityRegistry | None:
1051 """Hass entity registry is weird, every component ends up creating its own, with a store, subscribing
1052 to all entities, so do it once here
1053 """
1054 if self.__entity_registry is not None:
1055 return self.__entity_registry
1056 try:
1057 self.__entity_registry = er.async_get(self._hass)
1058 except Exception as e:
1059 _LOGGER.warning("SUPERNOTIFY Unable to get entity registry: %s", e)
1060 return self.__entity_registry
1062 def _device_registry(self) -> dr.DeviceRegistry | None:
1063 """Hass device registry is weird, every component ends up creating its own, with a store, subscribing
1064 to all devices, so do it once here
1065 """
1066 if self.__device_registry is not None:
1067 return self.__device_registry
1068 try:
1069 self.__device_registry = dr.async_get(self._hass)
1070 except Exception as e:
1071 _LOGGER.warning("SUPERNOTIFY Unable to get device registry: %s", e)
1072 return self.__device_registry
1074 def find_device(self, device_id: str) -> DeviceEntry | None:
1075 reg: DeviceRegistry | None = self._device_registry()
1076 if reg is None:
1077 return None
1078 try:
1079 return reg.async_get(device_id, include_child_devices=False) # type: ignore[call-arg]
1080 except TypeError:
1081 # older HA
1082 return cast("DeviceEntry|None", reg.async_get(device_id))
1084 def is_own_device(self, device_id: str) -> bool:
1085 """True if `device_id` is this integration's own 'SuperNotify' device (see `ha_device_info`)"""
1086 device = self.find_device(device_id)
1087 return device is not None and any(domain == DOMAIN for domain, _identifier in device.identifiers)
1089 async def mqtt_available(self, raise_on_error: bool = True) -> bool:
1090 from homeassistant.components import mqtt
1092 try:
1093 return await mqtt.async_wait_for_mqtt_client(self._hass) is True
1094 except Exception:
1095 _LOGGER.exception("SUPERNOTIFY MQTT integration failed on available check")
1096 if raise_on_error:
1097 raise
1098 return False
1100 async def mqtt_publish(
1101 self,
1102 topic: str,
1103 payload: Any = None, # ruff: ignore[any-type]
1104 qos: int = 0,
1105 retain: bool = False,
1106 raise_on_error: bool = True,
1107 ) -> None:
1108 from homeassistant.components import mqtt
1110 try:
1111 await mqtt.async_publish(
1112 self._hass,
1113 topic=topic,
1114 payload=json_dumps(payload),
1115 qos=qos,
1116 retain=retain,
1117 )
1118 except Exception:
1119 _LOGGER.exception(f"SUPERNOTIFY MQTT publish failed to {topic}")
1120 if raise_on_error:
1121 raise
1124class ConditionErrorLoggingAdaptor(logging.LoggerAdapter):
1125 def __init__(self, *args: Any, **kwargs: Any) -> None:
1126 super().__init__(*args, **kwargs)
1127 self.condition_errors: list[ConditionError] = []
1129 def capture(self, args: list | tuple | None) -> None:
1130 if args and isinstance(args, list | tuple):
1131 for arg in args:
1132 if isinstance(arg, ConditionErrorContainer):
1133 self.condition_errors.extend(arg.errors)
1134 elif isinstance(arg, ConditionError):
1135 self.condition_errors.append(arg)
1137 def error(self, msg: object, *args: object, **kwargs: Any) -> None:
1138 self.capture(args)
1139 self.logger.error(msg, *args, **kwargs)
1141 def warning(self, msg: object, *args: Any, **kwargs: Any) -> None:
1142 self.capture(args)
1143 self.logger.warning(msg, *args, **kwargs)
1146class TemplateWrapper:
1147 def __init__(self, obj: Template) -> None:
1148 self._obj = obj
1150 def __getattr__(self, name: str) -> Any: # ruff: ignore[any-type]
1151 if name == "async_render_to_info":
1152 return partial(self._obj.async_render_to_info, strict=True)
1153 return getattr(self._obj, name)
1155 def __setattr__(self, name: str, value: Any) -> None: # ruff: ignore[any-type]
1156 super().__setattr__(name, value)
1158 def __repr__(self) -> str:
1159 return self._obj.__repr__() if self._obj else "NULL TEMPLATE"
1162def force_strict_template_mode(conditions: list[ConfigType], undo: bool = False) -> None:
1163 def wrap_template(cond: ConfigType, undo: bool) -> ConfigType:
1164 for key, val in cond.items():
1165 if not undo and isinstance(val, Template) and hasattr(val, "_env"):
1166 cond[key] = TemplateWrapper(val)
1167 elif undo and isinstance(val, TemplateWrapper):
1168 cond[key] = val._obj
1169 elif isinstance(val, dict):
1170 wrap_template(val, undo)
1171 return cond
1173 if conditions is not None:
1174 conditions = [wrap_template(condition, undo) for condition in conditions]
1177def nested_template_conditions(conditions: list[ConfigType], nested: bool = False) -> Generator[ConfigType]:
1178 """Conditions using a template, inside compound (and/or/not) conditions"""
1179 for cond in conditions:
1180 if isinstance(cond.get(CONF_CONDITIONS), list):
1181 yield from nested_template_conditions(cond[CONF_CONDITIONS], nested=True)
1182 elif nested and any(isinstance(val, Template) for val in cond.values()):
1183 yield cond
1186@contextmanager
1187def trace_action(
1188 hass: HomeAssistant,
1189 item_id: str,
1190 config: dict[str, Any] | None = None,
1191 context: HomeAssistantContext | None = None,
1192 stored_traces: int = 5,
1193) -> Generator[ActionTrace]:
1194 """Trace execution of a condition"""
1195 trace = ActionTrace(item_id, config, None, context or HomeAssistantContext())
1196 async_store_trace(hass, trace, stored_traces)
1198 try:
1199 yield trace
1200 except Exception as ex:
1201 if item_id:
1202 trace.set_error(ex)
1203 raise
1204 finally:
1205 if item_id:
1206 trace.finished()