core/homeassistant/components/template/config.py

"""Template config validator."""

from collections.abc import Callable
from contextlib import suppress
import itertools
import logging
from typing import Any

import voluptuous as vol

from homeassistant.components.alarm_control_panel import (
    DOMAIN as DOMAIN_ALARM_CONTROL_PANEL,
)
from homeassistant.components.binary_sensor import DOMAIN as DOMAIN_BINARY_SENSOR
from homeassistant.components.blueprint import (
    is_blueprint_instance_config,
    schemas as blueprint_schemas,
)
from homeassistant.components.button import DOMAIN as DOMAIN_BUTTON
from homeassistant.components.cover import DOMAIN as DOMAIN_COVER
from homeassistant.components.event import DOMAIN as DOMAIN_EVENT
from homeassistant.components.fan import DOMAIN as DOMAIN_FAN
from homeassistant.components.image import DOMAIN as DOMAIN_IMAGE
from homeassistant.components.light import DOMAIN as DOMAIN_LIGHT
from homeassistant.components.lock import DOMAIN as DOMAIN_LOCK
from homeassistant.components.number import DOMAIN as DOMAIN_NUMBER
from homeassistant.components.select import DOMAIN as DOMAIN_SELECT
from homeassistant.components.sensor import DOMAIN as DOMAIN_SENSOR
from homeassistant.components.switch import DOMAIN as DOMAIN_SWITCH
from homeassistant.components.update import DOMAIN as DOMAIN_UPDATE
from homeassistant.components.vacuum import DOMAIN as DOMAIN_VACUUM
from homeassistant.components.weather import DOMAIN as DOMAIN_WEATHER
from homeassistant.config import async_log_schema_error, config_without_domain
from homeassistant.const import (
    CONF_ACTION,
    CONF_ACTIONS,
    CONF_BINARY_SENSORS,
    CONF_CONDITION,
    CONF_CONDITIONS,
    CONF_NAME,
    CONF_SENSORS,
    CONF_TRIGGER,
    CONF_TRIGGERS,
    CONF_UNIQUE_ID,
    CONF_VARIABLES,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv, issue_registry as ir
from homeassistant.helpers.condition import async_validate_conditions_config
from homeassistant.helpers.issue_registry import IssueSeverity
from homeassistant.helpers.template import Template
from homeassistant.helpers.trigger import async_validate_trigger_config
from homeassistant.helpers.typing import ConfigType
from homeassistant.setup import async_notify_setup_error
from homeassistant.util import yaml as yaml_util

from . import (
    alarm_control_panel as alarm_control_panel_platform,
    binary_sensor as binary_sensor_platform,
    button as button_platform,
    cover as cover_platform,
    event as event_platform,
    fan as fan_platform,
    image as image_platform,
    light as light_platform,
    lock as lock_platform,
    number as number_platform,
    select as select_platform,
    sensor as sensor_platform,
    switch as switch_platform,
    update as update_platform,
    vacuum as vacuum_platform,
    weather as weather_platform,
)
from .const import CONF_DEFAULT_ENTITY_ID, DOMAIN, PLATFORMS, TemplateConfig
from .helpers import (
    async_get_blueprints,
    create_legacy_template_issue,
    rewrite_legacy_to_modern_configs,
)

_LOGGER = logging.getLogger(__name__)

PACKAGE_MERGE_HINT = "list"


def validate_binary_sensor_auto_off_has_trigger(obj: dict) -> dict:
    """Validate that binary sensors with auto_off have triggers."""
    if CONF_TRIGGERS not in obj and DOMAIN_BINARY_SENSOR in obj:
        binary_sensors: list[ConfigType] = obj[DOMAIN_BINARY_SENSOR]
        for binary_sensor in binary_sensors:
            if binary_sensor_platform.CONF_AUTO_OFF not in binary_sensor:
                continue

            identifier = f"{CONF_NAME}: {binary_sensor_platform.DEFAULT_NAME}"
            if (
                (name := binary_sensor.get(CONF_NAME))
                and isinstance(name, Template)
                and name.template != binary_sensor_platform.DEFAULT_NAME
            ):
                identifier = f"{CONF_NAME}: {name.template}"
            elif default_entity_id := binary_sensor.get(CONF_DEFAULT_ENTITY_ID):
                identifier = f"{CONF_DEFAULT_ENTITY_ID}: {default_entity_id}"
            elif unique_id := binary_sensor.get(CONF_UNIQUE_ID):
                identifier = f"{CONF_UNIQUE_ID}: {unique_id}"

            raise vol.Invalid(
                f"The auto_off option for template binary sensor: {identifier} "
                "requires a trigger, remove the auto_off option or rewrite "
                "configuration to use a trigger"
            )

    return obj


def ensure_domains_do_not_have_trigger_or_action(*keys: str) -> Callable[[dict], dict]:
    """Validate that config does not contain trigger and action."""
    domains = set(keys)

    def validate(obj: dict):
        options = set(obj.keys())
        if found_domains := domains.intersection(options):
            invalid = {CONF_TRIGGERS, CONF_ACTIONS}
            if found_invalid := invalid.intersection(set(obj.keys())):
                raise vol.Invalid(
                    f"Unsupported option(s) found for domain {found_domains.pop()}, please remove ({', '.join(found_invalid)}) from your configuration",
                )

        return obj

    return validate


def create_trigger_format_issue(
    hass: HomeAssistant, config: ConfigType, option: str
) -> None:
    """Create a warning when a rogue trigger or action is found."""
    issue_id = hex(hash(frozenset(config)))
    yaml_config = yaml_util.dump(config)
    ir.async_create_issue(
        hass,
        DOMAIN,
        issue_id,
        is_fixable=False,
        severity=IssueSeverity.WARNING,
        translation_key=f"config_format_{option}",
        translation_placeholders={"config": yaml_config},
    )


def validate_trigger_format(
    hass: HomeAssistant, config_section: ConfigType, raw_config: ConfigType
) -> None:
    """Validate the config section."""
    options = set(config_section.keys())

    if CONF_TRIGGERS in options and not options.intersection(
        [CONF_SENSORS, CONF_BINARY_SENSORS, *PLATFORMS]
    ):
        _LOGGER.warning(
            "Invalid template configuration found, trigger option is missing matching domain"
        )
        create_trigger_format_issue(hass, raw_config, CONF_TRIGGERS)

    elif CONF_ACTIONS in options and CONF_TRIGGERS not in options:
        _LOGGER.warning(
            "Invalid template configuration found, action option requires a trigger"
        )
        create_trigger_format_issue(hass, raw_config, CONF_ACTIONS)


def _backward_compat_schema(value: Any | None) -> Any:
    """Backward compatibility for automations."""

    value = cv.renamed(CONF_TRIGGER, CONF_TRIGGERS)(value)
    value = cv.renamed(CONF_ACTION, CONF_ACTIONS)(value)
    return cv.renamed(CONF_CONDITION, CONF_CONDITIONS)(value)


CONFIG_SECTION_SCHEMA = vol.All(
    _backward_compat_schema,
    vol.Schema(
        {
            vol.Optional(CONF_ACTIONS): cv.SCRIPT_SCHEMA,
            vol.Optional(CONF_BINARY_SENSORS): cv.schema_with_slug_keys(
                binary_sensor_platform.BINARY_SENSOR_LEGACY_YAML_SCHEMA
            ),
            vol.Optional(CONF_CONDITIONS): cv.CONDITIONS_SCHEMA,
            vol.Optional(CONF_SENSORS): cv.schema_with_slug_keys(
                sensor_platform.SENSOR_LEGACY_YAML_SCHEMA
            ),
            vol.Optional(CONF_TRIGGERS): cv.TRIGGER_SCHEMA,
            vol.Optional(CONF_UNIQUE_ID): cv.string,
            vol.Optional(CONF_VARIABLES): cv.SCRIPT_VARIABLES_SCHEMA,
            vol.Optional(DOMAIN_ALARM_CONTROL_PANEL): vol.All(
                cv.ensure_list,
                [alarm_control_panel_platform.ALARM_CONTROL_PANEL_YAML_SCHEMA],
            ),
            vol.Optional(DOMAIN_BINARY_SENSOR): vol.All(
                cv.ensure_list, [binary_sensor_platform.BINARY_SENSOR_YAML_SCHEMA]
            ),
            vol.Optional(DOMAIN_BUTTON): vol.All(
                cv.ensure_list, [button_platform.BUTTON_YAML_SCHEMA]
            ),
            vol.Optional(DOMAIN_COVER): vol.All(
                cv.ensure_list, [cover_platform.COVER_YAML_SCHEMA]
            ),
            vol.Optional(DOMAIN_EVENT): vol.All(
                cv.ensure_list, [event_platform.EVENT_YAML_SCHEMA]
            ),
            vol.Optional(DOMAIN_FAN): vol.All(
                cv.ensure_list, [fan_platform.FAN_YAML_SCHEMA]
            ),
            vol.Optional(DOMAIN_IMAGE): vol.All(
                cv.ensure_list, [image_platform.IMAGE_YAML_SCHEMA]
            ),
            vol.Optional(DOMAIN_LIGHT): vol.All(
                cv.ensure_list, [light_platform.LIGHT_YAML_SCHEMA]
            ),
            vol.Optional(DOMAIN_LOCK): vol.All(
                cv.ensure_list, [lock_platform.LOCK_YAML_SCHEMA]
            ),
            vol.Optional(DOMAIN_NUMBER): vol.All(
                cv.ensure_list, [number_platform.NUMBER_YAML_SCHEMA]
            ),
            vol.Optional(DOMAIN_SELECT): vol.All(
                cv.ensure_list, [select_platform.SELECT_YAML_SCHEMA]
            ),
            vol.Optional(DOMAIN_SENSOR): vol.All(
                cv.ensure_list, [sensor_platform.SENSOR_YAML_SCHEMA]
            ),
            vol.Optional(DOMAIN_SWITCH): vol.All(
                cv.ensure_list, [switch_platform.SWITCH_YAML_SCHEMA]
            ),
            vol.Optional(DOMAIN_UPDATE): vol.All(
                cv.ensure_list, [update_platform.UPDATE_YAML_SCHEMA]
            ),
            vol.Optional(DOMAIN_VACUUM): vol.All(
                cv.ensure_list, [vacuum_platform.VACUUM_YAML_SCHEMA]
            ),
            vol.Optional(DOMAIN_WEATHER): vol.All(
                cv.ensure_list, [weather_platform.WEATHER_YAML_SCHEMA]
            ),
        },
    ),
    ensure_domains_do_not_have_trigger_or_action(
        DOMAIN_BUTTON,
    ),
    validate_binary_sensor_auto_off_has_trigger,
)

TEMPLATE_BLUEPRINT_SCHEMA = vol.All(
    _backward_compat_schema, blueprint_schemas.BLUEPRINT_SCHEMA
)


def _merge_section_variables(config: ConfigType, section_variables: ConfigType) -> None:
    """Merges a template entity configuration's variables with the section variables."""
    if (variables := config.pop(CONF_VARIABLES, None)) and isinstance(variables, dict):
        config[CONF_VARIABLES] = {**section_variables, **variables}
    else:
        config[CONF_VARIABLES] = section_variables


async def _async_resolve_template_config(
    hass: HomeAssistant,
    config: ConfigType,
) -> TemplateConfig:
    """If a config item requires a blueprint, resolve that item to an actual config."""
    raw_config = None
    raw_blueprint_inputs = None

    with suppress(ValueError):  # Invalid config
        raw_config = dict(config)

    original_config = config
    config = _backward_compat_schema(config)
    if is_blueprint_instance_config(config):
        blueprints = async_get_blueprints(hass)

        blueprint_inputs = await blueprints.async_inputs_from_config(config)
        raw_blueprint_inputs = blueprint_inputs.config_with_inputs

        config = blueprint_inputs.async_substitute()

        platforms = [platform for platform in PLATFORMS if platform in config]
        if len(platforms) > 1:
            raise vol.Invalid("more than one platform defined per blueprint")
        if len(platforms) == 1:
            platform = platforms.pop()
            for prop in (CONF_NAME, CONF_UNIQUE_ID):
                if prop in config:
                    config[platform][prop] = config.pop(prop)
            # State based template entities remove CONF_VARIABLES because they pass
            # blueprint inputs to the template entities. Trigger based template entities
            # retain CONF_VARIABLES because the variables are always executed between
            # the trigger and action.
            if CONF_TRIGGERS not in config and CONF_VARIABLES in config:
                _merge_section_variables(config[platform], config.pop(CONF_VARIABLES))

        raw_config = dict(config)

    # Trigger based template entities retain CONF_VARIABLES because the variables are
    # always executed between the trigger and action.
    elif CONF_TRIGGERS not in config and CONF_VARIABLES in config:
        # State based template entities have 2 layers of variables.  Variables at the section level
        # and variables at the entity level should be merged together at the entity level.
        section_variables = config.pop(CONF_VARIABLES)
        platform_config: list[ConfigType] | ConfigType
        platforms = [platform for platform in PLATFORMS if platform in config]
        for platform in platforms:
            platform_config = config[platform]
            if platform in PLATFORMS:
                if isinstance(platform_config, dict):
                    platform_config = [platform_config]

                for entity_config in platform_config:
                    _merge_section_variables(entity_config, section_variables)

    validate_trigger_format(hass, config, original_config)
    template_config = TemplateConfig(CONFIG_SECTION_SCHEMA(config))
    template_config.raw_blueprint_inputs = raw_blueprint_inputs
    template_config.raw_config = raw_config

    return template_config


async def async_validate_config_section(
    hass: HomeAssistant, config: ConfigType
) -> TemplateConfig:
    """Validate an entire config section for the template integration."""

    validated_config = await _async_resolve_template_config(hass, config)

    if CONF_TRIGGERS in validated_config:
        validated_config[CONF_TRIGGERS] = await async_validate_trigger_config(
            hass, validated_config[CONF_TRIGGERS]
        )

    if CONF_CONDITIONS in validated_config:
        validated_config[CONF_CONDITIONS] = await async_validate_conditions_config(
            hass, validated_config[CONF_CONDITIONS]
        )

    return validated_config


async def async_validate_config(hass: HomeAssistant, config: ConfigType) -> ConfigType:
    """Validate config."""

    configs = []
    for key in config:
        if DOMAIN not in key:
            continue

        if key == DOMAIN or (key.startswith(DOMAIN) and len(key.split()) > 1):
            configs.append(cv.ensure_list(config[key]))

    if not configs:
        return config

    config_sections = []

    for cfg in itertools.chain(*configs):
        try:
            template_config: TemplateConfig = await async_validate_config_section(
                hass, cfg
            )
        except vol.Invalid as err:
            async_log_schema_error(err, DOMAIN, cfg, hass)
            async_notify_setup_error(hass, DOMAIN)
            continue

        legacy_warn_printed = False

        for old_key, new_key, legacy_fields in (
            (
                CONF_SENSORS,
                DOMAIN_SENSOR,
                sensor_platform.LEGACY_FIELDS,
            ),
            (
                CONF_BINARY_SENSORS,
                DOMAIN_BINARY_SENSOR,
                binary_sensor_platform.LEGACY_FIELDS,
            ),
        ):
            if old_key not in template_config:
                continue

            if not legacy_warn_printed:
                legacy_warn_printed = True
                _LOGGER.warning(
                    "The entity definition format under template: differs from the"
                    " platform "
                    "configuration format. See "
                    "https://www.home-assistant.io/integrations/template#configuration-for-trigger-based-template-sensors"
                )

            definitions = (
                list(template_config[new_key]) if new_key in template_config else []
            )
            for definition in rewrite_legacy_to_modern_configs(
                hass, new_key, template_config[old_key], legacy_fields
            ):
                create_legacy_template_issue(hass, definition, new_key)
                definitions.append(definition)
            template_config = TemplateConfig({**template_config, new_key: definitions})

        config_sections.append(template_config)

    # Create a copy of the configuration with all config for current
    # component removed and add validated config back in.
    config = config_without_domain(config, DOMAIN)
    config[DOMAIN] = config_sections

    return config