from esphome.components import sensor import esphome.config_validation as cv import esphome.codegen as cg from esphome.const import CONF_ID, CONF_ADDRESS, CONF_LAMBDA, CONF_OFFSET from .. import ( SensorItem, modbus_controller_ns, ModbusController, MODBUS_REGISTER_TYPE, SENSOR_VALUE_TYPE, ) from ..const import ( CONF_BITMASK, CONF_BYTE_OFFSET, CONF_FORCE_NEW_RANGE, CONF_MODBUS_CONTROLLER_ID, CONF_REGISTER_COUNT, CONF_REGISTER_TYPE, CONF_SKIP_UPDATES, CONF_VALUE_TYPE, ) DEPENDENCIES = ["modbus_controller"] CODEOWNERS = ["@martgras"] ModbusSensor = modbus_controller_ns.class_( "ModbusSensor", cg.Component, sensor.Sensor, SensorItem ) TYPE_REGISTER_MAP = { "RAW": 1, "U_WORD": 1, "S_WORD": 1, "U_DWORD": 2, "U_DWORD_R": 2, "S_DWORD": 2, "S_DWORD_R": 2, "U_QWORD": 4, "U_QWORDU_R": 4, "S_QWORD": 4, "U_QWORD_R": 4, "FP32": 2, "FP32_R": 2, } CONFIG_SCHEMA = cv.All( sensor.SENSOR_SCHEMA.extend( { cv.GenerateID(): cv.declare_id(ModbusSensor), cv.GenerateID(CONF_MODBUS_CONTROLLER_ID): cv.use_id(ModbusController), cv.Required(CONF_ADDRESS): cv.positive_int, cv.Required(CONF_REGISTER_TYPE): cv.enum(MODBUS_REGISTER_TYPE), cv.Optional(CONF_OFFSET, default=0): cv.positive_int, cv.Optional(CONF_BYTE_OFFSET): cv.positive_int, cv.Optional(CONF_BITMASK, default=0xFFFFFFFF): cv.hex_uint32_t, cv.Optional(CONF_VALUE_TYPE, default="U_WORD"): cv.enum(SENSOR_VALUE_TYPE), cv.Optional(CONF_REGISTER_COUNT, default=0): cv.positive_int, cv.Optional(CONF_SKIP_UPDATES, default=0): cv.positive_int, cv.Optional(CONF_FORCE_NEW_RANGE, default=False): cv.boolean, cv.Optional(CONF_LAMBDA): cv.returning_lambda, } ).extend(cv.COMPONENT_SCHEMA), ) async def to_code(config): byte_offset = 0 if CONF_OFFSET in config: byte_offset = config[CONF_OFFSET] # A CONF_BYTE_OFFSET setting overrides CONF_OFFSET if CONF_BYTE_OFFSET in config: byte_offset = config[CONF_BYTE_OFFSET] value_type = config[CONF_VALUE_TYPE] reg_count = config[CONF_REGISTER_COUNT] if reg_count == 0: reg_count = TYPE_REGISTER_MAP[value_type] var = cg.new_Pvariable( config[CONF_ID], config[CONF_REGISTER_TYPE], config[CONF_ADDRESS], byte_offset, config[CONF_BITMASK], config[CONF_VALUE_TYPE], reg_count, config[CONF_SKIP_UPDATES], config[CONF_FORCE_NEW_RANGE], ) await cg.register_component(var, config) await sensor.register_sensor(var, config) paren = await cg.get_variable(config[CONF_MODBUS_CONTROLLER_ID]) cg.add(paren.add_sensor_item(var)) if CONF_LAMBDA in config: template_ = await cg.process_lambda( config[CONF_LAMBDA], [ (ModbusSensor.operator("ptr"), "item"), (cg.float_, "x"), ( cg.std_vector.template(cg.uint8).operator("const").operator("ref"), "data", ), ], return_type=cg.optional.template(float), ) cg.add(var.set_template(template_))