Coverage for custom_components/supernotify/hass_api.py: 25%
423 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-02-06 15:56 +0000
1from __future__ import annotations
3import logging
4from dataclasses import dataclass
5from functools import partial
6from typing import TYPE_CHECKING, Any
8from homeassistant.components.person import ATTR_USER_ID
9from homeassistant.const import CONF_ACTION, CONF_DEVICE_ID
10from homeassistant.helpers.aiohttp_client import async_get_clientsession
11from homeassistant.helpers.event import async_track_state_change_event, async_track_time_change
12from homeassistant.util import slugify
14if TYPE_CHECKING:
15 import asyncio
16 from collections.abc import Callable, Iterable, Iterator
18 import aiohttp
19 from homeassistant.core import CALLBACK_TYPE, HomeAssistant, Service, ServiceResponse, State
20 from homeassistant.helpers.entity import Entity
21 from homeassistant.helpers.entity_registry import EntityRegistry
22 from homeassistant.helpers.typing import ConfigType
23 from homeassistant.util.event_type import EventType
25 from .schema import ConditionsFunc
27import socket
28import threading
29from contextlib import contextmanager
30from typing import TYPE_CHECKING, cast
32import homeassistant.components.trace
33from homeassistant.components import mqtt
34from homeassistant.components.group import expand_entity_ids
35from homeassistant.components.trace.const import DATA_TRACE
36from homeassistant.components.trace.models import ActionTrace
37from homeassistant.components.trace.util import async_store_trace
38from homeassistant.core import Context as HomeAssistantContext
39from homeassistant.core import HomeAssistant, SupportsResponse
40from homeassistant.exceptions import ConditionError, ConditionErrorContainer, IntegrationError
41from homeassistant.helpers import condition as condition
42from homeassistant.helpers import device_registry as dr
43from homeassistant.helpers import entity_registry as er
44from homeassistant.helpers import issue_registry as ir
45from homeassistant.helpers.json import json_dumps
46from homeassistant.helpers.network import get_url
47from homeassistant.helpers.template import Template
48from homeassistant.helpers.trace import trace_get, trace_path
49from homeassistant.helpers.typing import ConfigType
51from . import DOMAIN
52from .const import CONF_DEVICE_LABELS, CONF_DEVICE_TRACKER, CONF_MOBILE_APP_ID
53from .model import ConditionVariables, SelectionRule
55if TYPE_CHECKING:
56 from homeassistant.core import HomeAssistant
57 from homeassistant.helpers.device_registry import DeviceEntry, DeviceRegistry
58 from homeassistant.helpers.typing import ConfigType
60# avoid importing from homeassistant.components.mobile_app.const and triggering dependency chain
62CONF_USER_ID = "user_id"
63ATTR_OS_NAME = "os_name"
64ATTR_OS_VERSION = "os_version"
65ATTR_APP_VERSION = "app_version"
66ATTR_DEVICE_NAME = "device_name"
67ATTR_MANUFACTURER = "manufacturer"
68ATTR_MODEL = "model"
70_LOGGER = logging.getLogger(__name__)
73@dataclass
74class DeviceInfo:
75 device_id: str
76 device_labels: list[str]
77 mobile_app_id: str | None = None
78 device_name: str | None = None
79 device_tracker: str | None = None
80 action: str | None = None
81 user_id: str | None = None
82 area_id: str | None = None
83 manufacturer: str | None = None
84 model: str | None = None
85 os_name: str | None = None
86 os_version: str | None = None
87 app_version: str | None = None
88 identifiers: set[tuple[str, str]] | None = None
90 def as_dict(self) -> dict[str, str | list[str] | None]:
91 return {
92 CONF_MOBILE_APP_ID: self.mobile_app_id,
93 ATTR_DEVICE_NAME: self.device_name,
94 CONF_DEVICE_ID: self.device_id,
95 CONF_USER_ID: self.user_id,
96 CONF_DEVICE_TRACKER: self.device_tracker,
97 CONF_ACTION: self.action,
98 ATTR_OS_NAME: self.os_name,
99 ATTR_OS_VERSION: self.os_version,
100 ATTR_APP_VERSION: self.app_version,
101 ATTR_MANUFACTURER: self.manufacturer,
102 ATTR_MODEL: self.model,
103 CONF_DEVICE_LABELS: self.device_labels,
104 }
106 def __eq__(self, other: Any) -> bool:
107 """Test support"""
108 return other is not None and other.as_dict() == self.as_dict()
111class HomeAssistantAPI:
112 def __init__(self, hass: HomeAssistant) -> None:
113 self._hass: HomeAssistant = hass
114 self.internal_url: str = ""
115 self.external_url: str = ""
116 self.language: str = ""
117 self.hass_name: str = "!UNDEFINED!"
118 self._entity_registry: er.EntityRegistry | None = None
119 self._device_registry: dr.DeviceRegistry | None = None
120 self._service_info: dict[tuple[str, str], Any] = {}
121 self.unsubscribes: list[CALLBACK_TYPE] = []
122 self.mobile_apps_by_tracker: dict[str, DeviceInfo] = {}
123 self.mobile_apps_by_app_id: dict[str, DeviceInfo] = {}
124 self.mobile_apps_by_device_id: dict[str, DeviceInfo] = {}
125 self.mobile_apps_by_user_id: dict[str, list[DeviceInfo]] = {}
127 def initialize(self) -> None:
128 self.hass_name = self._hass.config.location_name
129 self.language = self._hass.config.language
130 try:
131 self.internal_url = get_url(self._hass, prefer_external=False)
132 except Exception as e:
133 self.internal_url = f"http://{socket.gethostname()}"
134 _LOGGER.warning("SUPERNOTIFY could not get internal hass url, defaulting to %s: %s", self.internal_url, e)
135 try:
136 self.external_url = get_url(self._hass, prefer_external=True)
137 except Exception as e:
138 _LOGGER.warning("SUPERNOTIFY could not get external hass url, defaulting to internal url: %s", e)
139 self.external_url = self.internal_url
141 self.build_mobile_app_cache()
143 _LOGGER.debug(
144 "SUPERNOTIFY Configured for HomeAssistant instance %s at %s , %s",
145 self.hass_name,
146 self.internal_url,
147 self.external_url,
148 )
150 if not self.internal_url or not self.internal_url.startswith("http"):
151 _LOGGER.warning("SUPERNOTIFY invalid internal hass url %s", self.internal_url)
153 def disconnect(self) -> None:
154 while self.unsubscribes:
155 unsub = self.unsubscribes.pop()
156 try:
157 _LOGGER.debug("SUPERNOTIFY unsubscribing: %s", unsub)
158 unsub()
159 except Exception as e:
160 _LOGGER.error("SUPERNOTIFY failed to unsubscribe: %s", e)
161 _LOGGER.debug("SUPERNOTIFY disconnection complete")
163 def subscribe_event(self, event: EventType | str, callback: Callable) -> None:
164 self.unsubscribes.append(self._hass.bus.async_listen(event, callback))
166 def subscribe_state(self, entity_ids: str | Iterable[str], callback: Callable) -> None:
167 self.unsubscribes.append(async_track_state_change_event(self._hass, entity_ids, callback))
169 def subscribe_time(self, hour: int, minute: int, second: int, callback: Callable) -> None:
170 self.unsubscribes.append(async_track_time_change(self._hass, callback, hour=hour, minute=minute, second=second))
172 def in_hass_loop(self) -> bool:
173 return self._hass is not None and self._hass.loop_thread_id == threading.get_ident()
175 def get_state(self, entity_id: str) -> State | None:
176 return self._hass.states.get(entity_id)
178 def is_state(self, entity_id: str, state: str) -> bool:
179 return self._hass.states.is_state(entity_id, state)
181 def set_state(self, entity_id: str, state: str | int | bool, attributes: dict[str, Any] | None = None) -> None:
182 if self.in_hass_loop():
183 self._hass.states.async_set(entity_id, str(state), attributes=attributes)
184 else:
185 self._hass.states.set(entity_id, str(state), attributes=attributes)
187 def has_service(self, domain: str, service: str) -> bool:
188 return self._hass.services.has_service(domain, service)
190 def entity_ids_for_domain(self, domain: str) -> list[str]:
191 return self._hass.states.async_entity_ids(domain)
193 def domain_entity(self, domain: str, entity_id: str) -> Entity | None:
194 # TODO: must be a better hass method than this
195 return self._hass.data.get(domain, {}).get_entity(entity_id)
197 def create_job(self, func: Callable, *args: Any) -> asyncio.Future[Any]:
198 """Wrap a blocking function call in a HomeAssistant awaitable job"""
199 return self._hass.async_add_executor_job(func, *args)
201 async def call_service(
202 self,
203 domain: str,
204 service: str,
205 service_data: dict[str, Any] | None = None,
206 target: dict[str, Any] | None = None,
207 return_response: bool | None = None,
208 blocking: bool | None = None,
209 debug: bool = False,
210 ) -> ServiceResponse | None:
212 if return_response is None or blocking is None:
213 # unknown service, for example defined in generic action, check if it supports response
214 supports_response: SupportsResponse = self.service_info(domain, service)
215 if supports_response == SupportsResponse.NONE:
216 return_response = False
217 elif supports_response == SupportsResponse.ONLY:
218 return_response = True
219 else:
220 return_response = debug
221 blocking = return_response or debug
223 response: ServiceResponse | None = await self._hass.services.async_call(
224 domain,
225 service,
226 service_data=service_data,
227 blocking=blocking,
228 context=None,
229 target=target,
230 return_response=return_response,
231 )
232 if response is not None and debug:
233 _LOGGER.info("SUPERNOTIFY Service %s.%s response: %s", domain, service, response)
234 return response
236 def service_info(self, domain: str, service: str) -> SupportsResponse:
238 try:
239 if (domain, service) not in self._service_info:
240 service_objs: dict[str, dict[str, Service]] = self._hass.services.async_services()
241 service_obj: Service | None = service_objs.get(domain, {}).get(service)
242 if service_obj:
243 self._service_info[domain, service] = {
244 "supports_response": service_obj.supports_response,
245 "schema": service_obj.schema,
246 }
247 service_info: dict[str, Any] = self._service_info.get((domain, service), {})
248 supports_response: SupportsResponse | None = service_info.get("supports_response")
249 if supports_response is None:
250 _LOGGER.debug("SUPERNOTIFY Unable to find service info for %s.%s", domain, service)
252 except Exception as e:
253 _LOGGER.warning("SUPERNOTIFY Unable to get service info for %s.%s: %s", domain, service, e)
254 return supports_response or SupportsResponse.NONE # default to no response
256 def find_service(self, domain: str, module: str) -> str | None:
257 try:
258 service_objs: dict[str, dict[str, Service]] = self._hass.services.async_services()
259 if service_objs:
260 for service, domain_obj in service_objs.get(domain, {}).items():
261 if domain_obj.job and domain_obj.job.target:
262 target_module: str | None = (
263 domain_obj.job.target.__self__.__module__
264 if hasattr(domain_obj.job.target, "__self__")
265 else domain_obj.job.target.__module__
266 )
267 if target_module == module:
268 _LOGGER.debug("SUPERNOTIFY Found service %s for domain %s", domain, service)
269 return f"{domain}.{service}"
271 _LOGGER.debug("SUPERNOTIFY Unable to find service for %s", domain)
272 except Exception as e:
273 _LOGGER.warning("SUPERNOTIFY Unable to find service for %s: %s", domain, e)
274 return None
276 def http_session(self) -> aiohttp.ClientSession:
277 """Client aiohttp session for async web requests"""
278 return async_get_clientsession(self._hass)
280 def expand_group(self, entity_ids: str | list[str]) -> list[str]:
281 return expand_entity_ids(self._hass, entity_ids)
283 def template(self, template_format: str) -> Template:
284 return Template(template_format, self._hass)
286 async def trace_conditions(
287 self,
288 conditions: ConditionsFunc,
289 condition_variables: ConditionVariables,
290 trace_name: str | None = None,
291 ) -> tuple[bool | None, ActionTrace | None]:
293 result: bool | None = None
294 this_trace: ActionTrace | None = None
295 if DATA_TRACE not in self._hass.data:
296 _LOGGER.warning("SUPERNOTIFY tracing not configured, attempting to set up")
298 await homeassistant.components.trace.async_setup(self._hass, {}) # type: ignore
299 with trace_action(self._hass, trace_name or "anon_condition") as cond_trace:
300 cond_trace.set_trace(trace_get())
301 this_trace = cond_trace
302 with trace_path(["condition", "conditions"]) as _tp:
303 result = self.evaluate_conditions(conditions, condition_variables)
304 _LOGGER.debug(cond_trace.as_dict())
305 return result, this_trace
307 async def build_conditions(
308 self, condition_config: list[ConfigType], strict: bool = False, validate: bool = False, name: str = DOMAIN
309 ) -> ConditionsFunc | None:
310 capturing_logger: ConditionErrorLoggingAdaptor = ConditionErrorLoggingAdaptor(_LOGGER)
311 condition_variables: ConditionVariables = ConditionVariables()
312 cond_list: list[ConfigType]
313 try:
314 if validate:
315 cond_list = cast(
316 "list[ConfigType]", await condition.async_validate_conditions_config(self._hass, condition_config)
317 )
318 else:
319 cond_list = condition_config
320 except Exception as e:
321 _LOGGER.exception("SUPERNOTIFY Conditions validation failed: %s", e)
322 raise
323 try:
324 if strict:
325 force_strict_template_mode(cond_list, undo=False)
327 test: ConditionsFunc = await condition.async_conditions_from_config(
328 self._hass, cond_list, cast("logging.Logger", capturing_logger), name
329 )
330 if test is None:
331 raise IntegrationError(f"Invalid condition {condition_config}")
332 test(condition_variables.as_dict())
333 return test
334 except Exception as e:
335 _LOGGER.exception("SUPERNOTIFY Conditions eval failed: %s", e)
336 raise
337 finally:
338 if strict:
339 force_strict_template_mode(condition_config, undo=True)
340 if strict and capturing_logger.condition_errors and len(capturing_logger.condition_errors) > 0:
341 for exception in capturing_logger.condition_errors:
342 _LOGGER.warning("SUPERNOTIFY Invalid condition %s:%s", condition_config, exception)
343 raise capturing_logger.condition_errors[0]
345 def evaluate_conditions(
346 self,
347 conditions: ConditionsFunc,
348 condition_variables: ConditionVariables,
349 ) -> bool | None:
350 try:
351 if not condition_variables:
352 _LOGGER.warning("SUPERNOTIFY No cond vars provided for condition")
353 return conditions(condition_variables.as_dict() if condition_variables is not None else None)
354 except Exception as e:
355 _LOGGER.error("SUPERNOTIFY Condition eval failed: %s", e)
356 raise
358 def abs_url(self, fragment: str | None, prefer_external: bool = True) -> str | None:
359 base_url = self.external_url if prefer_external else self.internal_url
360 if fragment:
361 if fragment.startswith("http"):
362 return fragment
363 if fragment.startswith("/"):
364 return base_url + fragment
365 return base_url + "/" + fragment
366 return None
368 def raise_issue(
369 self,
370 issue_id: str,
371 issue_key: str,
372 issue_map: dict[str, str],
373 severity: ir.IssueSeverity = ir.IssueSeverity.WARNING,
374 learn_more_url: str = "https://supernotify.rhizomatics.org.uk",
375 is_fixable: bool = False,
376 ) -> None:
377 ir.async_create_issue(
378 self._hass,
379 DOMAIN,
380 issue_id,
381 translation_key=issue_key,
382 translation_placeholders=issue_map,
383 severity=severity,
384 learn_more_url=learn_more_url,
385 is_fixable=is_fixable,
386 )
388 def mobile_app_by_tracker(self, device_tracker: str) -> DeviceInfo | None:
389 return self.mobile_apps_by_tracker.get(device_tracker)
391 def mobile_app_by_id(self, mobile_app_id: str) -> DeviceInfo | None:
392 return self.mobile_apps_by_app_id.get(mobile_app_id)
394 def mobile_app_by_device_id(self, device_id: str) -> DeviceInfo | None:
395 return self.mobile_apps_by_device_id.get(device_id)
397 def mobile_app_by_user_id(self, user_id: str) -> list[DeviceInfo] | None:
398 return self.mobile_apps_by_user_id.get(user_id)
400 def build_mobile_app_cache(self) -> None:
401 """All enabled mobile apps"""
402 ent_reg: EntityRegistry | None = self.entity_registry()
403 if not ent_reg:
404 _LOGGER.warning("SUPERNOTIFY Unable to discover devices for - no entity registry found")
405 return
407 found: int = 0
408 complete: int = 0
409 for mobile_app_info in self.discover_devices("mobile_app"):
410 try:
411 mobile_app_id: str = f"mobile_app_{slugify(mobile_app_info.device_name)}"
412 device_tracker: str | None = None
413 notify_action: str | None = None
414 if self.has_service("notify", mobile_app_id):
415 notify_action = f"notify.{mobile_app_id}"
416 else:
417 _LOGGER.warning("SUPERNOTIFY Unable to find notify action <%s>", mobile_app_id)
419 registry_entries = ent_reg.entities.get_entries_for_device_id(mobile_app_info.device_id)
420 for reg_entry in registry_entries:
421 if reg_entry.platform == "mobile_app" and reg_entry.domain == "device_tracker":
422 device_tracker = reg_entry.entity_id
424 if device_tracker and notify_action:
425 complete += 1
427 mobile_app_info.mobile_app_id = mobile_app_id
428 mobile_app_info.device_tracker = device_tracker
429 mobile_app_info.action = notify_action
431 found += 1
432 self.mobile_apps_by_app_id[mobile_app_id] = mobile_app_info
433 self.mobile_apps_by_device_id[mobile_app_info.device_id] = mobile_app_info
434 if device_tracker:
435 self.mobile_apps_by_tracker[device_tracker] = mobile_app_info
436 if mobile_app_info.user_id is not None:
437 self.mobile_apps_by_user_id.setdefault(mobile_app_info.user_id, [])
438 self.mobile_apps_by_user_id[mobile_app_info.user_id].append(mobile_app_info)
440 except Exception as e:
441 _LOGGER.error("SUPERNOTIFY Failure examining device %s: %s", mobile_app_info, e)
443 _LOGGER.info(f"SUPERNOTIFY Found {found} enabled mobile app devices, {complete} complete config")
445 def device_config_info(self, device: DeviceEntry) -> dict[str, str | None]:
446 results: dict[str, str | None] = {ATTR_OS_NAME: None, ATTR_OS_VERSION: None, CONF_USER_ID: None, ATTR_APP_VERSION: None}
447 for config_entry_id in device.config_entries:
448 config_entry = self._hass.config_entries.async_get_entry(config_entry_id)
449 if config_entry and config_entry.data:
450 for attr in results:
451 results[attr] = config_entry.data.get(attr) or results[attr]
452 return results
454 def discover_devices(
455 self,
456 discover_domain: str,
457 device_model_select: SelectionRule | None = None,
458 device_manufacturer_select: SelectionRule | None = None,
459 device_os_select: SelectionRule | None = None,
460 device_area_select: SelectionRule | None = None,
461 device_label_select: SelectionRule | None = None,
462 ) -> list[DeviceInfo]:
463 devices: list[DeviceInfo] = []
464 dev_reg: DeviceRegistry | None = self.device_registry()
465 if dev_reg is None or not hasattr(dev_reg, "devices"):
466 _LOGGER.warning(f"SUPERNOTIFY Unable to discover devices for {discover_domain} - no device registry found")
467 return []
469 all_devs = enabled_devs = found_devs = skipped_devs = 0
470 for dev in dev_reg.devices.values():
471 all_devs += 1
473 if dev.disabled:
474 _LOGGER.debug("SUPERNOTIFY excluded disabled device %s", dev.name)
475 else:
476 enabled_devs += 1
477 for identifier in dev.identifiers:
478 if identifier and len(identifier) > 1 and identifier[0] == discover_domain:
479 _LOGGER.debug("SUPERNOTIFY discovered %s device %s for id %s", dev.model, dev.name, identifier)
480 found_devs += 1
481 if device_model_select is not None and not device_model_select.match(dev.model):
482 _LOGGER.debug("SUPERNOTIFY Skipped dev %s, no model %s match", dev.name, dev.model)
483 skipped_devs += 1
484 continue
485 if device_manufacturer_select is not None and not device_manufacturer_select.match(dev.manufacturer):
486 _LOGGER.debug("SUPERNOTIFY Skipped dev %s, no manufacturer %s match", dev.name, dev.manufacturer)
487 skipped_devs += 1
488 continue
489 device_config_info = self.device_config_info(dev)
490 if device_os_select is not None and not device_os_select.match(device_config_info[ATTR_OS_NAME]):
491 _LOGGER.debug(
492 "SUPERNOTIFY Skipped dev %s, no OS %s match", dev.name, device_config_info[ATTR_OS_NAME]
493 )
494 skipped_devs += 1
495 continue
496 if device_area_select is not None and not device_area_select.match(dev.area_id):
497 _LOGGER.debug("SUPERNOTIFY Skipped dev %s, no area %s match", dev.name, dev.area_id)
498 skipped_devs += 1
499 continue
500 if device_label_select is not None and not device_label_select.match(dev.labels):
501 _LOGGER.debug("SUPERNOTIFY Skipped dev %s, no label %s match", dev.name, dev.labels)
502 skipped_devs += 1
503 continue
504 devices.append(
505 DeviceInfo(
506 device_id=dev.id,
507 device_name=dev.name,
508 manufacturer=dev.manufacturer,
509 model=dev.model,
510 area_id=dev.area_id,
511 user_id=device_config_info[ATTR_USER_ID],
512 os_name=device_config_info[ATTR_OS_NAME],
513 os_version=device_config_info[ATTR_OS_VERSION],
514 app_version=device_config_info[ATTR_APP_VERSION],
515 device_labels=list(dev.labels) if dev.labels else [],
516 identifiers=dev.identifiers,
517 )
518 )
520 elif identifier:
521 # HomeKit has triples for identifiers, other domains may behave similarly
522 _LOGGER.debug("SUPERNOTIFY Ignoring device %s id: %s", dev.name, identifier)
523 else:
524 _LOGGER.debug( # type: ignore
525 "SUPERNOTIFY Unexpected %s device %s without id", dev.model, dev.name
526 )
528 _LOGGER.debug(f"SUPERNOTIFY {discover_domain} device discovery, all={all_devs},enabled={enabled_devs} ")
529 _LOGGER.debug(f"SUPERNOTIFY {discover_domain} skipped={skipped_devs}, found={found_devs}")
531 return devices
533 def domain_for_device(self, device_id: str, domains: list[str]) -> str | None:
534 # discover domain from device registry
535 verified_domain: str | None = None
536 device_registry = self.device_registry()
537 if device_registry:
538 device: DeviceEntry | None = device_registry.async_get(device_id)
539 if device:
540 matching_domains = [d for d, _id in device.identifiers if d in domains]
541 if matching_domains:
542 # TODO: limited to first domain found, unlikely to be more
543 return matching_domains[0]
544 _LOGGER.warning(
545 "SUPERNOTIFY A target that looks like a device_id can't be matched to supported integration: %s",
546 device_id,
547 )
548 return verified_domain
550 def entity_registry(self) -> er.EntityRegistry | None:
551 """Hass entity registry is weird, every component ends up creating its own, with a store, subscribing
552 to all entities, so do it once here
553 """ # noqa: D205
554 if self._entity_registry is not None:
555 return self._entity_registry
556 try:
557 self._entity_registry = er.async_get(self._hass)
558 except Exception as e:
559 _LOGGER.warning("SUPERNOTIFY Unable to get entity registry: %s", e)
560 return self._entity_registry
562 def device_registry(self) -> dr.DeviceRegistry | None:
563 """Hass device registry is weird, every component ends up creating its own, with a store, subscribing
564 to all devices, so do it once here
565 """ # noqa: D205
566 if self._device_registry is not None:
567 return self._device_registry
568 try:
569 self._device_registry = dr.async_get(self._hass)
570 except Exception as e:
571 _LOGGER.warning("SUPERNOTIFY Unable to get device registry: %s", e)
572 return self._device_registry
574 async def mqtt_available(self, raise_on_error: bool = True) -> bool:
575 try:
576 return await mqtt.async_wait_for_mqtt_client(self._hass) is True
577 except Exception:
578 _LOGGER.exception("SUPERNOTIFY MQTT integration failed on available check")
579 if raise_on_error:
580 raise
581 return False
583 async def mqtt_publish(
584 self, topic: str, payload: Any = None, qos: int = 0, retain: bool = False, raise_on_error: bool = True
585 ) -> None:
586 try:
587 await mqtt.async_publish(
588 self._hass,
589 topic=topic,
590 payload=json_dumps(payload),
591 qos=qos,
592 retain=retain,
593 )
594 except Exception:
595 _LOGGER.exception(f"SUPERNOTIFY MQTT publish failed to {topic}")
596 if raise_on_error:
597 raise
600class ConditionErrorLoggingAdaptor(logging.LoggerAdapter):
601 def __init__(self, *args: Any, **kwargs: Any) -> None:
602 super().__init__(*args, **kwargs)
603 self.condition_errors: list[ConditionError] = []
605 def capture(self, args: Any) -> None:
606 if args and isinstance(args, list | tuple):
607 for arg in args:
608 if isinstance(arg, ConditionErrorContainer):
609 self.condition_errors.extend(arg.errors)
610 elif isinstance(arg, ConditionError):
611 self.condition_errors.append(arg)
613 def error(self, msg: Any, *args: object, **kwargs: Any) -> None:
614 self.capture(args)
615 self.logger.error(msg, args, kwargs)
617 def warning(self, msg: Any, *args: Any, **kwargs: Any) -> None:
618 self.capture(args)
619 self.logger.warning(msg, args, kwargs)
622def force_strict_template_mode(conditions: list[ConfigType], undo: bool = False) -> None:
623 class TemplateWrapper:
624 def __init__(self, obj: Template) -> None:
625 self._obj = obj
627 def __getattr__(self, name: str) -> Any:
628 if name == "async_render_to_info":
629 return partial(self._obj.async_render_to_info, strict=True)
630 return getattr(self._obj, name)
632 def __setattr__(self, name: str, value: Any) -> None:
633 super().__setattr__(name, value)
635 def __repr__(self) -> str:
636 return self._obj.__repr__() if self._obj else "NULL TEMPLATE"
638 def wrap_template(cond: ConfigType, undo: bool) -> ConfigType:
639 for key, val in cond.items():
640 if not undo and isinstance(val, Template) and hasattr(val, "_env"):
641 cond[key] = TemplateWrapper(val)
642 elif undo and isinstance(val, TemplateWrapper):
643 cond[key] = val._obj
644 elif isinstance(val, dict):
645 wrap_template(val, undo)
646 return cond
648 if conditions is not None:
649 conditions = [wrap_template(condition, undo) for condition in conditions]
652@contextmanager
653def trace_action(
654 hass: HomeAssistant,
655 item_id: str,
656 config: dict[str, Any] | None = None,
657 context: HomeAssistantContext | None = None,
658 stored_traces: int = 5,
659) -> Iterator[ActionTrace]:
660 """Trace execution of a condition"""
661 trace = ActionTrace(item_id, config, None, context or HomeAssistantContext())
662 async_store_trace(hass, trace, stored_traces)
664 try:
665 yield trace
666 except Exception as ex:
667 if item_id:
668 trace.set_error(ex)
669 raise
670 finally:
671 if item_id:
672 trace.finished()