added textsensor mode

This commit is contained in:
j0ta29 2023-08-26 18:48:36 +00:00
parent d999d6dc28
commit f1df196687
8 changed files with 115 additions and 74 deletions

View file

@ -1,14 +1,18 @@
from esphome import core
from esphome import pins
import esphome.codegen as cg
from esphome.components import text_sensor as ts
import esphome.config_validation as cv
from esphome.const import (
CONF_ADDRESS,
CONF_DIV_RATIO,
CONF_ID,
CONF_LOGGER,
CONF_PROTOCOL,
CONF_RX_PIN,
CONF_STATE,
CONF_TX_PIN,
CONF_UPDATE_INTERVAL,
)
from esphome.core import CORE
@ -31,6 +35,20 @@ DeviceInfoSensor = optolink_ns.class_(
)
DEVICE_INFO_SENSOR_ID = "device_info_sensor_id"
CONF_OPTOLINK_ID = "optolink_id"
SENSOR_BASE_SCHEMA = cv.Schema(
{
cv.GenerateID(CONF_OPTOLINK_ID): cv.use_id(OptolinkComponent),
cv.Optional(CONF_UPDATE_INTERVAL, default="10s"): cv.All(
cv.positive_time_period_milliseconds,
cv.Range(min=core.TimePeriod(seconds=1), max=core.TimePeriod(seconds=1800)),
),
cv.Required(CONF_ADDRESS): cv.hex_uint32_t,
# cv.Required(CONF_BYTES): cv.one_of(1, 2, 4, int=True),
cv.Optional(CONF_DIV_RATIO, default=1): cv.one_of(1, 10, 100, 3600, int=True),
}
)
def required_on_esp32(attribute):
"""Validate that this option can only be specified on the given target platforms."""

View file

@ -1,22 +1,13 @@
from esphome import core
import esphome.codegen as cg
import esphome.config_validation as cv
from esphome.components import binary_sensor
from esphome.const import CONF_ID, CONF_ADDRESS, CONF_UPDATE_INTERVAL
from . import OptolinkComponent, optolink_ns, CONF_OPTOLINK_ID
from esphome.const import CONF_ADDRESS, CONF_ID
from . import SENSOR_BASE_SCHEMA, optolink_ns, CONF_OPTOLINK_ID
OptolinkBinarySensor = optolink_ns.class_(
"OptolinkBinarySensor", binary_sensor.BinarySensor, cg.PollingComponent
)
CONFIG_SCHEMA = binary_sensor.binary_sensor_schema(OptolinkBinarySensor).extend(
{
cv.GenerateID(CONF_OPTOLINK_ID): cv.use_id(OptolinkComponent),
cv.Required(CONF_ADDRESS): cv.hex_uint32_t,
cv.Optional(CONF_UPDATE_INTERVAL, default="10s"): cv.All(
cv.positive_time_period_milliseconds,
cv.Range(min=core.TimePeriod(seconds=1), max=core.TimePeriod(seconds=1800)),
),
}
SENSOR_BASE_SCHEMA
)

View file

@ -1,4 +1,3 @@
from esphome import core
import esphome.codegen as cg
import esphome.config_validation as cv
from esphome.components import number
@ -10,10 +9,9 @@ from esphome.const import (
CONF_MAX_VALUE,
CONF_MIN_VALUE,
CONF_STEP,
CONF_UPDATE_INTERVAL,
)
from .sensor import SENSOR_BASE_SCHEMA
from . import OptolinkComponent, optolink_ns, CONF_OPTOLINK_ID
from . import optolink_ns, CONF_OPTOLINK_ID
OptolinkNumber = optolink_ns.class_(
"OptolinkNumber", number.Number, cg.PollingComponent
@ -22,17 +20,11 @@ OptolinkNumber = optolink_ns.class_(
CONFIG_SCHEMA = (
number.NUMBER_SCHEMA.extend(
{
cv.GenerateID(CONF_OPTOLINK_ID): cv.use_id(OptolinkComponent),
cv.GenerateID(): cv.declare_id(OptolinkNumber),
cv.Required(CONF_MAX_VALUE): cv.float_,
cv.Required(CONF_MIN_VALUE): cv.float_range(min=0.0),
cv.Required(CONF_STEP): cv.float_,
cv.Optional(CONF_UPDATE_INTERVAL, default="10s"): cv.All(
cv.positive_time_period_milliseconds,
cv.Range(
min=core.TimePeriod(seconds=1), max=core.TimePeriod(seconds=1800)
),
),
cv.Required(CONF_BYTES): cv.one_of(1, 2, 4, int=True),
}
)
.extend(cv.COMPONENT_SCHEMA)

View file

@ -8,18 +8,37 @@ namespace esphome {
namespace optolink {
void OptolinkTextSensor::setup() {
if (!raw_) {
setup_datapoint_();
} else {
if (mode_ == RAW) {
datapoint_ = new Datapoint<convRaw>(get_sensor_name().c_str(), "optolink", address_, writeable_);
datapoint_->setLength(bytes_);
datapoint_->setCallback([this](const IDatapoint &dp, DPValue dp_value) {
ESP_LOGD("OptolinkSensorBase", "Datapoint %s - %s: <raw>", dp.getGroup(), dp.getName());
uint8_t buffer[bytes_ + 1];
dp_value.getRaw(buffer);
buffer[bytes_] = 0x0;
ESP_LOGD("OptolinkTextSensor", "Datapoint %s - %s: %s", dp.getGroup(), dp.getName(), buffer);
publish_state((char *) buffer);
});
} else if (mode_ == DAY_SCHEDULE) {
datapoint_ = new Datapoint<convRaw>(get_sensor_name().c_str(), "optolink", address_ + 8 * dow_, writeable_);
datapoint_->setLength(8);
datapoint_->setCallback([this](const IDatapoint &dp, DPValue dp_value) {
uint8_t data[8];
dp_value.getRaw(data);
ESP_LOGD("OptolinkTextSensor", "Datapoint %s - %s", dp.getGroup(), dp.getName());
char buffer[100];
for (int i = 0; i < 8; i++) {
if (data[i] != 0xFF) {
int hour = data[i] >> 3;
int minute = (data[i] & 0b111) * 10;
sprintf(buffer + i * 6, "%02d:%02d ", hour, minute);
} else {
sprintf(buffer + i * 6, " ");
}
}
publish_state(buffer);
});
} else {
setup_datapoint_();
}
};

View file

@ -10,23 +10,27 @@
namespace esphome {
namespace optolink {
enum TextSensorMode { MAP, RAW, DAY_SCHEDULE };
class OptolinkTextSensor : public OptolinkSensorBase,
public esphome::text_sensor::TextSensor,
public esphome::PollingComponent {
public:
OptolinkTextSensor(Optolink *optolink) : OptolinkSensorBase(optolink) {}
void set_raw(bool raw) { raw_ = raw; }
void set_mode(TextSensorMode mode) { mode_ = mode; }
void set_day_of_week(int dow) { dow_ = dow; }
protected:
void setup() override;
void update() override { optolink_->read_value(datapoint_); }
const StringRef &get_sensor_name() override { return get_name(); }
void value_changed(float state) override { publish_state(std::to_string(state)); };
void value_changed(float state) override { publish_state(std::to_string((uint32_t) state)); };
private:
bool raw_ = false;
TextSensorMode mode_ = MAP;
int dow_ = 0;
};
} // namespace optolink

View file

@ -1,4 +1,3 @@
from esphome import core
import esphome.codegen as cg
import esphome.config_validation as cv
from esphome.components import select
@ -9,9 +8,8 @@ from esphome.const import (
CONF_FROM,
CONF_ID,
CONF_TO,
CONF_UPDATE_INTERVAL,
)
from . import OptolinkComponent, optolink_ns, CONF_OPTOLINK_ID
from . import optolink_ns, CONF_OPTOLINK_ID
from .sensor import SENSOR_BASE_SCHEMA
OptolinkSelect = optolink_ns.class_(
@ -37,18 +35,12 @@ MAP_ID = "mappings"
CONFIG_SCHEMA = (
select.SELECT_SCHEMA.extend(
{
cv.GenerateID(CONF_OPTOLINK_ID): cv.use_id(OptolinkComponent),
cv.GenerateID(): cv.declare_id(OptolinkSelect),
cv.GenerateID(MAP_ID): cv.declare_id(
cg.std_ns.class_("map").template(cg.std_string, cg.std_string)
),
cv.Required(CONF_MAP): cv.ensure_list(validate_mapping),
cv.Optional(CONF_UPDATE_INTERVAL, default="10s"): cv.All(
cv.positive_time_period_milliseconds,
cv.Range(
min=core.TimePeriod(seconds=1), max=core.TimePeriod(seconds=1800)
),
),
cv.Required(CONF_BYTES): cv.one_of(1, 2, 4, int=True),
}
)
.extend(cv.COMPONENT_SCHEMA)

View file

@ -1,38 +1,22 @@
from esphome import core
import esphome.codegen as cg
import esphome.config_validation as cv
from esphome.components import sensor
from esphome.const import (
CONF_ID,
CONF_ADDRESS,
CONF_BYTES,
CONF_DIV_RATIO,
CONF_UPDATE_INTERVAL,
CONF_ID,
)
from . import optolink_ns, OptolinkComponent
from . import CONF_OPTOLINK_ID, SENSOR_BASE_SCHEMA, optolink_ns
OptolinkSensor = optolink_ns.class_(
"OptolinkSensor", sensor.Sensor, cg.PollingComponent
)
CONF_OPTOLINK_ID = "optolink_id"
SENSOR_BASE_SCHEMA = cv.Schema(
{
cv.Required(CONF_ADDRESS): cv.hex_uint32_t,
cv.Required(CONF_BYTES): cv.one_of(1, 2, 4, int=True),
cv.Optional(CONF_DIV_RATIO, default=1): cv.one_of(1, 10, 100, 3600, int=True),
}
)
CONFIG_SCHEMA = (
sensor.sensor_schema(OptolinkSensor)
.extend(
{
cv.GenerateID(CONF_OPTOLINK_ID): cv.use_id(OptolinkComponent),
cv.Optional(CONF_UPDATE_INTERVAL, default="10s"): cv.All(
cv.positive_time_period_milliseconds,
cv.Range(
min=core.TimePeriod(seconds=1), max=core.TimePeriod(seconds=1800)
),
),
cv.Required(CONF_BYTES): cv.one_of(1, 2, 4, int=True),
}
)
.extend(SENSOR_BASE_SCHEMA)

View file

@ -1,4 +1,3 @@
from esphome import core
import esphome.codegen as cg
import esphome.config_validation as cv
from esphome.components import text_sensor
@ -7,32 +6,71 @@ from esphome.const import (
CONF_BYTES,
CONF_DIV_RATIO,
CONF_ID,
CONF_RAW,
CONF_UPDATE_INTERVAL,
CONF_MODE,
)
from . import optolink_ns, OptolinkComponent, CONF_OPTOLINK_ID
from . import optolink_ns, CONF_OPTOLINK_ID
from .sensor import SENSOR_BASE_SCHEMA
OptolinkTextSensor = optolink_ns.class_(
"OptolinkTextSensor", text_sensor.TextSensor, cg.PollingComponent
)
TextSensorMode = optolink_ns.enum("TextSensorMode")
MODE = {
"MAP": TextSensorMode.MAP,
"RAW": TextSensorMode.RAW,
"DAY_SCHEDULE": TextSensorMode.DAY_SCHEDULE,
}
DAY_OF_WEEK = {
"MONDAY": 0,
"TUESDAY": 1,
"WEDNESDAY": 2,
"THURSDAY": 3,
"FRIDAY": 4,
"SATURDAY": 5,
"SUNDAY": 6,
}
CONF_DOW = "day_of_week"
def check_bytes():
def validator_(config):
bytes_needed = config[CONF_MODE] in ["MAP", "RAW"]
bytes_defined = CONF_BYTES in config
if bytes_needed and not bytes_defined:
raise cv.Invalid(f"{CONF_BYTES} is required in mode MAP or RAW")
if not bytes_needed and bytes_defined:
raise cv.Invalid(f"{CONF_BYTES} is not allowed in mode DAY_SCHEDULE")
return config
return validator_
def check_dow():
def validator_(config):
if config[CONF_MODE] == "DAY_SCHEDULE" and CONF_DOW not in config:
raise cv.Invalid(f"{CONF_DOW} is required in mode DAY_SCHEDULE")
if config[CONF_MODE] != "DAY_SCHEDULE" and CONF_DOW in config:
raise cv.Invalid(f"{CONF_DOW} is only allowed in mode DAY_SCHEDULE")
return config
return validator_
CONFIG_SCHEMA = cv.All(
text_sensor.text_sensor_schema(OptolinkTextSensor)
.extend(
{
cv.GenerateID(CONF_OPTOLINK_ID): cv.use_id(OptolinkComponent),
cv.Optional(CONF_UPDATE_INTERVAL, default="10s"): cv.All(
cv.positive_time_period_milliseconds,
cv.Range(
min=core.TimePeriod(seconds=1), max=core.TimePeriod(seconds=1800)
),
),
cv.Optional(CONF_RAW, default=False): cv.boolean,
cv.Optional(CONF_MODE, default="MAP"): cv.enum(MODE, upper=True),
cv.Optional(CONF_BYTES): cv.int_range(min=1, max=9),
cv.Optional(CONF_DOW): cv.enum(DAY_OF_WEEK, upper=True),
}
)
.extend(SENSOR_BASE_SCHEMA)
.extend({cv.Required(CONF_BYTES): cv.int_}),
.extend(SENSOR_BASE_SCHEMA),
check_bytes(),
check_dow(),
)
@ -43,7 +81,10 @@ async def to_code(config):
await cg.register_component(var, config)
await text_sensor.register_text_sensor(var, config)
cg.add(var.set_raw(config[CONF_RAW]))
cg.add(var.set_mode(config[CONF_MODE]))
cg.add(var.set_address(config[CONF_ADDRESS]))
cg.add(var.set_bytes(config[CONF_BYTES]))
cg.add(var.set_div_ratio(config[CONF_DIV_RATIO]))
if CONF_BYTES in config:
cg.add(var.set_bytes(config[CONF_BYTES]))
if CONF_DOW in config:
cg.add(var.set_day_of_week(config[CONF_DOW]))