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