"""Trigger entity.""" from __future__ import annotations from collections.abc import Callable from typing import Any from homeassistant.const import CONF_STATE, CONF_VARIABLES from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.script_variables import ScriptVariables from homeassistant.helpers.template import _SENTINEL from homeassistant.helpers.trigger_template_entity import TriggerBaseEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity from . import TriggerUpdateCoordinator from .entity import AbstractTemplateEntity class TriggerEntity( # pylint: disable=hass-enforce-class-module TriggerBaseEntity, CoordinatorEntity[TriggerUpdateCoordinator], AbstractTemplateEntity, ): """Template entity based on trigger data.""" def __init__( self, hass: HomeAssistant, coordinator: TriggerUpdateCoordinator, config: dict, ) -> None: """Initialize the entity.""" CoordinatorEntity.__init__(self, coordinator) TriggerBaseEntity.__init__(self, hass, config) AbstractTemplateEntity.__init__(self, hass, config) self._entity_variables: ScriptVariables | None = config.get(CONF_VARIABLES) self._rendered_entity_variables: dict | None = None self._state_render_error = False async def async_added_to_hass(self) -> None: """Handle being added to Home Assistant.""" await super().async_added_to_hass() if self.coordinator.data is not None: self._process_data() def _set_unique_id(self, unique_id: str | None) -> None: """Set unique id.""" if unique_id and self.coordinator.unique_id: self._unique_id = f"{self.coordinator.unique_id}-{unique_id}" else: self._unique_id = unique_id def setup_state_template( self, option: str, attribute: str, validator: Callable[[Any], Any] | None = None, on_update: Callable[[Any], None] | None = None, ) -> None: """Set up a template that manages the main state of the entity.""" if self.add_template(option, attribute, validator, on_update): self._to_render_simple.append(option) self._parse_result.add(option) def setup_template( self, option: str, attribute: str, validator: Callable[[Any], Any] | None = None, on_update: Callable[[Any], None] | None = None, ) -> None: """Set up a template that manages any property or attribute of the entity. Parameters ---------- option The configuration key provided by ConfigFlow or the yaml option attribute The name of the attribute to link to. This attribute must exist unless a custom on_update method is supplied. validator: Optional function that validates the rendered result. on_update: Called to store the template result rather than storing it the supplied attribute. Passed the result of the validator. """ self.setup_state_template(option, attribute, validator, on_update) @property def referenced_blueprint(self) -> str | None: """Return referenced blueprint or None.""" return self.coordinator.referenced_blueprint @property def available(self) -> bool: """Return availability of the entity.""" if self._state_render_error: return False return super().available @callback def _render_script_variables(self) -> dict: """Render configured variables.""" return self._rendered_entity_variables or {} def _render_templates(self, variables: dict[str, Any]) -> None: """Render templates.""" self._state_render_error = False rendered = dict(self._static_rendered) # If state fails to render, the entity should go unavailable. Render the # state as a simple template because the result should always be a string or None. if CONF_STATE in self._to_render_simple: if ( result := self._render_single_template(CONF_STATE, variables) ) is _SENTINEL: self._rendered = self._static_rendered self._state_render_error = True return rendered[CONF_STATE] = result self._render_single_templates(rendered, variables, [CONF_STATE]) self._render_attributes(rendered, variables) self._rendered = rendered def _handle_rendered_results(self) -> bool: """Get a rendered result and return the value.""" # Handle any templates. for option, entity_template in self._templates.items(): value = _SENTINEL if (rendered := self._rendered.get(option)) is not None: value = rendered if entity_template.validator: value = entity_template.validator(rendered) # Capture templates that did not render a result due to an exception and # ensure the state object updates. _SENTINEL is used to differentiate # templates that render None. if value is _SENTINEL: return True if entity_template.on_update: entity_template.on_update(value) else: setattr(self, entity_template.attribute, value) return True if len(self._rendered) > 0: # In some cases, the entity may be state optimistic or # attribute optimistic, in these scenarios the state needs # to update. return True return False @callback def _process_data(self) -> None: """Process new data.""" coordinator_variables = self.coordinator.data["run_variables"] if self._entity_variables: entity_variables = self._entity_variables.async_simple_render( coordinator_variables ) self._rendered_entity_variables = { **coordinator_variables, **entity_variables, } else: self._rendered_entity_variables = coordinator_variables variables = self._template_variables(self._rendered_entity_variables) self.async_set_context(self.coordinator.data["context"]) if self._render_availability_template(variables): self._render_templates(variables) write_state = False # While transitioning platforms to the new framework, this # if-statement is necessary for backward compatibility with existing # trigger based platforms. if self._templates: # Handle any results that were rendered. write_state = self._handle_rendered_results() # Check availability after rendering the results because the state # template could render the entity unavailable if not self.available: write_state = True if write_state: self.async_write_ha_state() else: self.async_write_ha_state() @callback def _handle_coordinator_update(self) -> None: """Handle updated data from the coordinator. While transitioning platforms to the new framework, this function is necessary for backward compatibility with existing trigger based platforms. """ self._process_data()