esphome/esphome/automation.py

import esphome.codegen as cg
import esphome.config_validation as cv
from esphome.const import (
    CONF_AUTOMATION_ID,
    CONF_CONDITION,
    CONF_ELSE,
    CONF_ID,
    CONF_THEN,
    CONF_TRIGGER_ID,
    CONF_TYPE_ID,
    CONF_TIME,
)
from esphome.jsonschema import jschema_extractor
from esphome.util import Registry


def maybe_simple_id(*validators):
    return maybe_conf(CONF_ID, *validators)


def maybe_conf(conf, *validators):
    validator = cv.All(*validators)

    @jschema_extractor("maybe")
    def validate(value):
        # pylint: disable=comparison-with-callable
        if value == jschema_extractor:
            return validator

        if isinstance(value, dict):
            return validator(value)
        with cv.remove_prepend_path([conf]):
            return validator({conf: value})

    return validate


def register_action(name, action_type, schema):
    return ACTION_REGISTRY.register(name, action_type, schema)


def register_condition(name, condition_type, schema):
    return CONDITION_REGISTRY.register(name, condition_type, schema)


Action = cg.esphome_ns.class_("Action")
Trigger = cg.esphome_ns.class_("Trigger")
ACTION_REGISTRY = Registry()
Condition = cg.esphome_ns.class_("Condition")
CONDITION_REGISTRY = Registry()
validate_action = cv.validate_registry_entry("action", ACTION_REGISTRY)
validate_action_list = cv.validate_registry("action", ACTION_REGISTRY)
validate_condition = cv.validate_registry_entry("condition", CONDITION_REGISTRY)
validate_condition_list = cv.validate_registry("condition", CONDITION_REGISTRY)


def validate_potentially_and_condition(value):
    if isinstance(value, list):
        with cv.remove_prepend_path(["and"]):
            return validate_condition({"and": value})
    return validate_condition(value)


DelayAction = cg.esphome_ns.class_("DelayAction", Action, cg.Component)
LambdaAction = cg.esphome_ns.class_("LambdaAction", Action)
IfAction = cg.esphome_ns.class_("IfAction", Action)
WhileAction = cg.esphome_ns.class_("WhileAction", Action)
WaitUntilAction = cg.esphome_ns.class_("WaitUntilAction", Action, cg.Component)
UpdateComponentAction = cg.esphome_ns.class_("UpdateComponentAction", Action)
Automation = cg.esphome_ns.class_("Automation")

LambdaCondition = cg.esphome_ns.class_("LambdaCondition", Condition)
ForCondition = cg.esphome_ns.class_("ForCondition", Condition, cg.Component)


def validate_automation(extra_schema=None, extra_validators=None, single=False):
    if extra_schema is None:
        extra_schema = {}
    if isinstance(extra_schema, cv.Schema):
        extra_schema = extra_schema.schema
    schema = AUTOMATION_SCHEMA.extend(extra_schema)

    def validator_(value):
        if isinstance(value, list):
            # List of items, there are two possible options here, either a sequence of
            # actions (no then:) or a list of automations.
            try:
                # First try as a sequence of actions
                # If that succeeds, return immediately
                with cv.remove_prepend_path([CONF_THEN]):
                    return [schema({CONF_THEN: value})]
            except cv.Invalid as err:
                # Next try as a sequence of automations
                try:
                    return cv.Schema([schema])(value)
                except cv.Invalid as err2:
                    if "extra keys not allowed" in str(err2) and len(err2.path) == 2:
                        # pylint: disable=raise-missing-from
                        raise err
                    if "Unable to find action" in str(err):
                        raise err2
                    raise cv.MultipleInvalid([err, err2])
        elif isinstance(value, dict):
            if CONF_THEN in value:
                return [schema(value)]
            with cv.remove_prepend_path([CONF_THEN]):
                return [schema({CONF_THEN: value})]
        # This should only happen with invalid configs, but let's have a nice error message.
        return [schema(value)]

    @jschema_extractor("automation")
    def validator(value):
        # hack to get the schema
        # pylint: disable=comparison-with-callable
        if value == jschema_extractor:
            return schema

        value = validator_(value)
        if extra_validators is not None:
            value = cv.Schema([extra_validators])(value)
        if single:
            if len(value) != 1:
                raise cv.Invalid("Cannot have more than 1 automation for templates")
            return value[0]
        return value

    return validator


AUTOMATION_SCHEMA = cv.Schema(
    {
        cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(Trigger),
        cv.GenerateID(CONF_AUTOMATION_ID): cv.declare_id(Automation),
        cv.Required(CONF_THEN): validate_action_list,
    }
)

AndCondition = cg.esphome_ns.class_("AndCondition", Condition)
OrCondition = cg.esphome_ns.class_("OrCondition", Condition)
NotCondition = cg.esphome_ns.class_("NotCondition", Condition)


@register_condition("and", AndCondition, validate_condition_list)
async def and_condition_to_code(config, condition_id, template_arg, args):
    conditions = await build_condition_list(config, template_arg, args)
    return cg.new_Pvariable(condition_id, template_arg, conditions)


@register_condition("or", OrCondition, validate_condition_list)
async def or_condition_to_code(config, condition_id, template_arg, args):
    conditions = await build_condition_list(config, template_arg, args)
    return cg.new_Pvariable(condition_id, template_arg, conditions)


@register_condition("not", NotCondition, validate_potentially_and_condition)
async def not_condition_to_code(config, condition_id, template_arg, args):
    condition = await build_condition(config, template_arg, args)
    return cg.new_Pvariable(condition_id, template_arg, condition)


@register_condition("lambda", LambdaCondition, cv.returning_lambda)
async def lambda_condition_to_code(config, condition_id, template_arg, args):
    lambda_ = await cg.process_lambda(config, args, return_type=bool)
    return cg.new_Pvariable(condition_id, template_arg, lambda_)


@register_condition(
    "for",
    ForCondition,
    cv.Schema(
        {
            cv.Required(CONF_TIME): cv.templatable(
                cv.positive_time_period_milliseconds
            ),
            cv.Required(CONF_CONDITION): validate_potentially_and_condition,
        }
    ).extend(cv.COMPONENT_SCHEMA),
)
async def for_condition_to_code(config, condition_id, template_arg, args):
    condition = await build_condition(
        config[CONF_CONDITION], cg.TemplateArguments(), []
    )
    var = cg.new_Pvariable(condition_id, template_arg, condition)
    await cg.register_component(var, config)
    templ = await cg.templatable(config[CONF_TIME], args, cg.uint32)
    cg.add(var.set_time(templ))
    return var


@register_action(
    "delay", DelayAction, cv.templatable(cv.positive_time_period_milliseconds)
)
async def delay_action_to_code(config, action_id, template_arg, args):
    var = cg.new_Pvariable(action_id, template_arg)
    await cg.register_component(var, {})
    template_ = await cg.templatable(config, args, cg.uint32)
    cg.add(var.set_delay(template_))
    return var


@register_action(
    "if",
    IfAction,
    cv.All(
        {
            cv.Required(CONF_CONDITION): validate_potentially_and_condition,
            cv.Optional(CONF_THEN): validate_action_list,
            cv.Optional(CONF_ELSE): validate_action_list,
        },
        cv.has_at_least_one_key(CONF_THEN, CONF_ELSE),
    ),
)
async def if_action_to_code(config, action_id, template_arg, args):
    conditions = await build_condition(config[CONF_CONDITION], template_arg, args)
    var = cg.new_Pvariable(action_id, template_arg, conditions)
    if CONF_THEN in config:
        actions = await build_action_list(config[CONF_THEN], template_arg, args)
        cg.add(var.add_then(actions))
    if CONF_ELSE in config:
        actions = await build_action_list(config[CONF_ELSE], template_arg, args)
        cg.add(var.add_else(actions))
    return var


@register_action(
    "while",
    WhileAction,
    cv.Schema(
        {
            cv.Required(CONF_CONDITION): validate_potentially_and_condition,
            cv.Required(CONF_THEN): validate_action_list,
        }
    ),
)
async def while_action_to_code(config, action_id, template_arg, args):
    conditions = await build_condition(config[CONF_CONDITION], template_arg, args)
    var = cg.new_Pvariable(action_id, template_arg, conditions)
    actions = await build_action_list(config[CONF_THEN], template_arg, args)
    cg.add(var.add_then(actions))
    return var


def validate_wait_until(value):
    schema = cv.Schema(
        {
            cv.Required(CONF_CONDITION): validate_potentially_and_condition,
        }
    )
    if isinstance(value, dict) and CONF_CONDITION in value:
        return schema(value)
    return validate_wait_until({CONF_CONDITION: value})


@register_action("wait_until", WaitUntilAction, validate_wait_until)
async def wait_until_action_to_code(config, action_id, template_arg, args):
    conditions = await build_condition(config[CONF_CONDITION], template_arg, args)
    var = cg.new_Pvariable(action_id, template_arg, conditions)
    await cg.register_component(var, {})
    return var


@register_action("lambda", LambdaAction, cv.lambda_)
async def lambda_action_to_code(config, action_id, template_arg, args):
    lambda_ = await cg.process_lambda(config, args, return_type=cg.void)
    return cg.new_Pvariable(action_id, template_arg, lambda_)


@register_action(
    "component.update",
    UpdateComponentAction,
    maybe_simple_id(
        {
            cv.Required(CONF_ID): cv.use_id(cg.PollingComponent),
        }
    ),
)
async def component_update_action_to_code(config, action_id, template_arg, args):
    comp = await cg.get_variable(config[CONF_ID])
    return cg.new_Pvariable(action_id, template_arg, comp)


async def build_action(full_config, template_arg, args):
    registry_entry, config = cg.extract_registry_entry_config(
        ACTION_REGISTRY, full_config
    )
    action_id = full_config[CONF_TYPE_ID]
    builder = registry_entry.coroutine_fun
    ret = await builder(config, action_id, template_arg, args)
    return ret


async def build_action_list(config, templ, arg_type):
    actions = []
    for conf in config:
        action = await build_action(conf, templ, arg_type)
        actions.append(action)
    return actions


async def build_condition(full_config, template_arg, args):
    registry_entry, config = cg.extract_registry_entry_config(
        CONDITION_REGISTRY, full_config
    )
    action_id = full_config[CONF_TYPE_ID]
    builder = registry_entry.coroutine_fun
    ret = await builder(config, action_id, template_arg, args)
    return ret


async def build_condition_list(config, templ, args):
    conditions = []
    for conf in config:
        condition = await build_condition(conf, templ, args)
        conditions.append(condition)
    return conditions


async def build_automation(trigger, args, config):
    arg_types = [arg[0] for arg in args]
    templ = cg.TemplateArguments(*arg_types)
    obj = cg.new_Pvariable(config[CONF_AUTOMATION_ID], templ, trigger)
    actions = await build_action_list(config[CONF_THEN], templ, args)
    cg.add(obj.add_actions(actions))
    return obj