mirror of
https://github.com/esphome/esphome.git
synced 2024-12-24 16:58:07 +01:00
325 lines
11 KiB
Python
325 lines
11 KiB
Python
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.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
|