esphome/esphome/components/rotary_encoder/sensor.py
Otto Winter a33bb32874
Convert components to async-def syntax (#1823)
* Convert components to async-def syntax

* Remove stray @coroutine

* Manual part

* Convert complexer components code to async-def

* Manual cleanup

* More manual cleanup
2021-05-24 21:45:31 +02:00

135 lines
4.6 KiB
Python

import esphome.codegen as cg
import esphome.config_validation as cv
from esphome import pins, automation
from esphome.components import sensor
from esphome.const import (
CONF_ID,
CONF_RESOLUTION,
CONF_MIN_VALUE,
CONF_MAX_VALUE,
DEVICE_CLASS_EMPTY,
UNIT_STEPS,
ICON_ROTATE_RIGHT,
CONF_VALUE,
CONF_PIN_A,
CONF_PIN_B,
CONF_TRIGGER_ID,
)
rotary_encoder_ns = cg.esphome_ns.namespace("rotary_encoder")
RotaryEncoderResolution = rotary_encoder_ns.enum("RotaryEncoderResolution")
RESOLUTIONS = {
1: RotaryEncoderResolution.ROTARY_ENCODER_1_PULSE_PER_CYCLE,
2: RotaryEncoderResolution.ROTARY_ENCODER_2_PULSES_PER_CYCLE,
4: RotaryEncoderResolution.ROTARY_ENCODER_4_PULSES_PER_CYCLE,
}
CONF_PIN_RESET = "pin_reset"
CONF_ON_CLOCKWISE = "on_clockwise"
CONF_ON_ANTICLOCKWISE = "on_anticlockwise"
RotaryEncoderSensor = rotary_encoder_ns.class_(
"RotaryEncoderSensor", sensor.Sensor, cg.Component
)
RotaryEncoderSetValueAction = rotary_encoder_ns.class_(
"RotaryEncoderSetValueAction", automation.Action
)
RotaryEncoderClockwiseTrigger = rotary_encoder_ns.class_(
"RotaryEncoderClockwiseTrigger", automation.Trigger
)
RotaryEncoderAnticlockwiseTrigger = rotary_encoder_ns.class_(
"RotaryEncoderAnticlockwiseTrigger", automation.Trigger
)
def validate_min_max_value(config):
if CONF_MIN_VALUE in config and CONF_MAX_VALUE in config:
min_val = config[CONF_MIN_VALUE]
max_val = config[CONF_MAX_VALUE]
if min_val >= max_val:
raise cv.Invalid(
"Max value {} must be smaller than min value {}"
"".format(max_val, min_val)
)
return config
CONFIG_SCHEMA = cv.All(
sensor.sensor_schema(UNIT_STEPS, ICON_ROTATE_RIGHT, 0, DEVICE_CLASS_EMPTY)
.extend(
{
cv.GenerateID(): cv.declare_id(RotaryEncoderSensor),
cv.Required(CONF_PIN_A): cv.All(
pins.internal_gpio_input_pin_schema, pins.validate_has_interrupt
),
cv.Required(CONF_PIN_B): cv.All(
pins.internal_gpio_input_pin_schema, pins.validate_has_interrupt
),
cv.Optional(CONF_PIN_RESET): pins.internal_gpio_output_pin_schema,
cv.Optional(CONF_RESOLUTION, default=1): cv.enum(RESOLUTIONS, int=True),
cv.Optional(CONF_MIN_VALUE): cv.int_,
cv.Optional(CONF_MAX_VALUE): cv.int_,
cv.Optional(CONF_ON_CLOCKWISE): automation.validate_automation(
{
cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(
RotaryEncoderClockwiseTrigger
),
}
),
cv.Optional(CONF_ON_ANTICLOCKWISE): automation.validate_automation(
{
cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(
RotaryEncoderAnticlockwiseTrigger
),
}
),
}
)
.extend(cv.COMPONENT_SCHEMA),
validate_min_max_value,
)
async def to_code(config):
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
await sensor.register_sensor(var, config)
pin_a = await cg.gpio_pin_expression(config[CONF_PIN_A])
cg.add(var.set_pin_a(pin_a))
pin_b = await cg.gpio_pin_expression(config[CONF_PIN_B])
cg.add(var.set_pin_b(pin_b))
if CONF_PIN_RESET in config:
pin_i = await cg.gpio_pin_expression(config[CONF_PIN_RESET])
cg.add(var.set_reset_pin(pin_i))
cg.add(var.set_resolution(config[CONF_RESOLUTION]))
if CONF_MIN_VALUE in config:
cg.add(var.set_min_value(config[CONF_MIN_VALUE]))
if CONF_MAX_VALUE in config:
cg.add(var.set_max_value(config[CONF_MAX_VALUE]))
for conf in config.get(CONF_ON_CLOCKWISE, []):
trigger = cg.new_Pvariable(conf[CONF_TRIGGER_ID], var)
await automation.build_automation(trigger, [], conf)
for conf in config.get(CONF_ON_ANTICLOCKWISE, []):
trigger = cg.new_Pvariable(conf[CONF_TRIGGER_ID], var)
await automation.build_automation(trigger, [], conf)
@automation.register_action(
"sensor.rotary_encoder.set_value",
RotaryEncoderSetValueAction,
cv.Schema(
{
cv.Required(CONF_ID): cv.use_id(sensor.Sensor),
cv.Required(CONF_VALUE): cv.templatable(cv.int_),
}
),
)
async def sensor_template_publish_to_code(config, action_id, template_arg, args):
paren = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, paren)
template_ = await cg.templatable(config[CONF_VALUE], args, int)
cg.add(var.set_value(template_))
return var