sascha/esphome
1
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2025-10-31 07:03:55 +00:00
Code Issues Releases Wiki Activity

Dsmr component (#1881)

Browse Source 
Co-authored-by: Otto winter <otto@otto-winter.com>
This commit is contained in:
Guillermo Ruffino
2021-08-10 05:32:16 -03:00
committed by GitHub
parent 98d32876b5
commit f26767b65e
8 changed files with 665 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
esphome/components/dsmr/__init__.py Normal file
View File

@@ -0,0 +1,59 @@
import esphome.codegen as cg
import esphome.config_validation as cv
from esphome.components import uart
from esphome.const import (
CONF_ID,
CONF_UART_ID,
)
CODEOWNERS = ["@glmnet", "@zuidwijk"]
DEPENDENCIES = ["uart"]
AUTO_LOAD = ["sensor", "text_sensor"]
CONF_DSMR_ID = "dsmr_id"
CONF_DECRYPTION_KEY = "decryption_key"
# Hack to prevent compile error due to ambiguity with lib namespace
dsmr_ns = cg.esphome_ns.namespace("esphome::dsmr")
Dsmr = dsmr_ns.class_("Dsmr", cg.Component, uart.UARTDevice)
def _validate_key(value):
value = cv.string_strict(value)
parts = [value[i : i + 2] for i in range(0, len(value), 2)]
if len(parts) != 16:
raise cv.Invalid("Decryption key must consist of 16 hexadecimal numbers")
parts_int = []
if any(len(part) != 2 for part in parts):
raise cv.Invalid("Decryption key must be format XX")
for part in parts:
try:
parts_int.append(int(part, 16))
except ValueError:
# pylint: disable=raise-missing-from
raise cv.Invalid("Decryption key must be hex values from 00 to FF")
return "".join(f"{part:02X}" for part in parts_int)
CONFIG_SCHEMA = cv.Schema(
{
cv.GenerateID(): cv.declare_id(Dsmr),
cv.Optional(CONF_DECRYPTION_KEY): _validate_key,
}
).extend(uart.UART_DEVICE_SCHEMA)
async def to_code(config):
uart_component = await cg.get_variable(config[CONF_UART_ID])
var = cg.new_Pvariable(config[CONF_ID], uart_component)
if CONF_DECRYPTION_KEY in config:
cg.add(var.set_decryption_key(config[CONF_DECRYPTION_KEY]))
await cg.register_component(var, config)
# DSMR Parser
cg.add_library("glmnet/Dsmr", "0.3")
# Crypto
cg.add_library("rweather/Crypto", "0.2.0")

182
esphome/components/dsmr/dsmr.cpp Normal file
View File

@@ -0,0 +1,182 @@
#include "dsmr.h"
#include "esphome/core/log.h"
#include <AES.h>
#include <Crypto.h>
#include <GCM.h>
namespace esphome {
namespace dsmr {
static const char *const TAG = "dsmr";
void Dsmr::loop() {
if (this->decryption_key_.empty())
this->receive_telegram_();
else
this->receive_encrypted_();
}
void Dsmr::receive_telegram_() {
while (available()) {
const char c = read();
if (c == '/') { // header: forward slash
ESP_LOGV(TAG, "Header found");
header_found_ = true;
footer_found_ = false;
telegram_len_ = 0;
}
if (!header_found_)
continue;
if (telegram_len_ >= MAX_TELEGRAM_LENGTH) { // Buffer overflow
header_found_ = false;
footer_found_ = false;
ESP_LOGE(TAG, "Error: Message larger than buffer");
return;
}
telegram_[telegram_len_] = c;
telegram_len_++;
if (c == '!') { // footer: exclamation mark
ESP_LOGV(TAG, "Footer found");
footer_found_ = true;
} else {
if (footer_found_ && c == 10) { // last \n after footer
header_found_ = false;
// Parse message
if (parse_telegram())
return;
}
}
}
}
void Dsmr::receive_encrypted_() {
// Encrypted buffer
uint8_t buffer[MAX_TELEGRAM_LENGTH];
size_t buffer_length = 0;
size_t packet_size = 0;
while (available()) {
const char c = read();
if (!header_found_) {
if ((uint8_t) c == 0xdb) {
ESP_LOGV(TAG, "Start byte 0xDB found");
header_found_ = true;
}
}
// Sanity check
if (!header_found_ || buffer_length >= MAX_TELEGRAM_LENGTH) {
if (buffer_length == 0) {
ESP_LOGE(TAG, "First byte of encrypted telegram should be 0xDB, aborting.");
} else {
ESP_LOGW(TAG, "Unexpected data");
}
this->status_momentary_warning("unexpected_data");
this->flush();
while (available())
read();
return;
}
buffer[buffer_length++] = c;
if (packet_size == 0 && buffer_length > 20) {
// Complete header + a few bytes of data
packet_size = buffer[11] << 8 | buffer[12];
}
if (buffer_length == packet_size + 13 && packet_size > 0) {
ESP_LOGV(TAG, "Encrypted data: %d bytes", buffer_length);
GCM<AES128> *gcmaes128{new GCM<AES128>()};
gcmaes128->setKey(this->decryption_key_.data(), gcmaes128->keySize());
// the iv is 8 bytes of the system title + 4 bytes frame counter
// system title is at byte 2 and frame counter at byte 15
for (int i = 10; i < 14; i++)
buffer[i] = buffer[i + 4];
constexpr uint16_t iv_size{12};
gcmaes128->setIV(&buffer[2], iv_size);
gcmaes128->decrypt(reinterpret_cast<uint8_t *>(this->telegram_),
// the ciphertext start at byte 18
&buffer[18],
// cipher size
buffer_length - 17);
delete gcmaes128;
telegram_len_ = strnlen(this->telegram_, sizeof(this->telegram_));
ESP_LOGV(TAG, "Decrypted data length: %d", telegram_len_);
ESP_LOGVV(TAG, "Decrypted data %s", this->telegram_);
parse_telegram();
telegram_len_ = 0;
return;
}
if (!available()) {
// baud rate is 115200 for encrypted data, this means a few byte should arrive every time
// program runs faster than buffer loading then available() might return false in the middle
delay(4); // Wait for data
}
}
if (buffer_length > 0)
ESP_LOGW(TAG, "Timeout while waiting for encrypted data or invalid data received.");
}
bool Dsmr::parse_telegram() {
MyData data;
ESP_LOGV(TAG, "Trying to parse");
::dsmr::ParseResult<void> res =
::dsmr::P1Parser::parse(&data, telegram_, telegram_len_,
false); // Parse telegram according to data definition. Ignore unknown values.
if (res.err) {
// Parsing error, show it
auto err_str = res.fullError(telegram_, telegram_ + telegram_len_);
ESP_LOGE(TAG, "%s", err_str.c_str());
return false;
} else {
this->status_clear_warning();
publish_sensors(data);
return true;
}
}
void Dsmr::dump_config() {
ESP_LOGCONFIG(TAG, "dsmr:");
#define DSMR_LOG_SENSOR(s) LOG_SENSOR(" ", #s, this->s_##s##_);
DSMR_SENSOR_LIST(DSMR_LOG_SENSOR, )
#define DSMR_LOG_TEXT_SENSOR(s) LOG_TEXT_SENSOR(" ", #s, this->s_##s##_);
DSMR_TEXT_SENSOR_LIST(DSMR_LOG_TEXT_SENSOR, )
}
void Dsmr::set_decryption_key(const std::string &decryption_key) {
if (decryption_key.length() == 0) {
ESP_LOGI(TAG, "Disabling decryption");
this->decryption_key_.clear();
return;
}
if (decryption_key.length() != 32) {
ESP_LOGE(TAG, "Error, decryption key must be 32 character long.");
return;
}
this->decryption_key_.clear();
ESP_LOGI(TAG, "Decryption key is set.");
// Verbose level prints decryption key
ESP_LOGV(TAG, "Using decryption key: %s", decryption_key.c_str());
char temp[3] = {0};
for (int i = 0; i < 16; i++) {
strncpy(temp, &(decryption_key.c_str()[i * 2]), 2);
decryption_key_.push_back(std::strtoul(temp, nullptr, 16));
}
}
} // namespace dsmr
} // namespace esphome

104
esphome/components/dsmr/dsmr.h Normal file
View File

@@ -0,0 +1,104 @@
#pragma once
#include "esphome/core/component.h"
#include "esphome/components/sensor/sensor.h"
#include "esphome/components/text_sensor/text_sensor.h"
#include "esphome/components/uart/uart.h"
#include "esphome/core/log.h"
#include "esphome/core/defines.h"
// don't include <dsmr.h> because it puts everything in global namespace
#include <dsmr/parser.h>
#include <dsmr/fields.h>
namespace esphome {
namespace dsmr {
static constexpr uint32_t MAX_TELEGRAM_LENGTH = 1500;
static constexpr uint32_t POLL_TIMEOUT = 1000;
using namespace ::dsmr::fields;
// DSMR_**_LIST generated by ESPHome and written in esphome/core/defines
#if !defined(DSMR_SENSOR_LIST) && !defined(DSMR_TEXT_SENSOR_LIST)
// Neither set, set it to a dummy value to not break build
#define DSMR_TEXT_SENSOR_LIST(F, SEP) F(identification)
#endif
#if defined(DSMR_SENSOR_LIST) && defined(DSMR_TEXT_SENSOR_LIST)
#define DSMR_BOTH ,
#else
#define DSMR_BOTH
#endif
#ifndef DSMR_SENSOR_LIST
#define DSMR_SENSOR_LIST(F, SEP)
#endif
#ifndef DSMR_TEXT_SENSOR_LIST
#define DSMR_TEXT_SENSOR_LIST(F, SEP)
#endif
#define DSMR_DATA_SENSOR(s) s
#define DSMR_COMMA ,
using MyData = ::dsmr::ParsedData<DSMR_TEXT_SENSOR_LIST(DSMR_DATA_SENSOR, DSMR_COMMA)
DSMR_BOTH DSMR_SENSOR_LIST(DSMR_DATA_SENSOR, DSMR_COMMA)>;
class Dsmr : public Component, public uart::UARTDevice {
public:
Dsmr(uart::UARTComponent *uart) : uart::UARTDevice(uart) {}
void loop() override;
bool parse_telegram();
void publish_sensors(MyData &data) {
#define DSMR_PUBLISH_SENSOR(s) \
if (data.s##_present && this->s_##s##_ != nullptr) \
s_##s##_->publish_state(data.s);
DSMR_SENSOR_LIST(DSMR_PUBLISH_SENSOR, )
#define DSMR_PUBLISH_TEXT_SENSOR(s) \
if (data.s##_present && this->s_##s##_ != nullptr) \
s_##s##_->publish_state(data.s.c_str());
DSMR_TEXT_SENSOR_LIST(DSMR_PUBLISH_TEXT_SENSOR, )
};
void dump_config() override;
void set_decryption_key(const std::string &decryption_key);
// Sensor setters
#define DSMR_SET_SENSOR(s) \
void set_##s(sensor::Sensor *sensor) { s_##s##_ = sensor; }
DSMR_SENSOR_LIST(DSMR_SET_SENSOR, )
#define DSMR_SET_TEXT_SENSOR(s) \
void set_##s(text_sensor::TextSensor *sensor) { s_##s##_ = sensor; }
DSMR_TEXT_SENSOR_LIST(DSMR_SET_TEXT_SENSOR, )
protected:
void receive_telegram_();
void receive_encrypted_();
// Telegram buffer
char telegram_[MAX_TELEGRAM_LENGTH];
int telegram_len_{0};
// Serial parser
bool header_found_{false};
bool footer_found_{false};
// Sensor member pointers
#define DSMR_DECLARE_SENSOR(s) sensor::Sensor *s_##s##_{nullptr};
DSMR_SENSOR_LIST(DSMR_DECLARE_SENSOR, )
#define DSMR_DECLARE_TEXT_SENSOR(s) text_sensor::TextSensor *s_##s##_{nullptr};
DSMR_TEXT_SENSOR_LIST(DSMR_DECLARE_TEXT_SENSOR, )
std::vector<uint8_t> decryption_key_{};
};
} // namespace dsmr
} // namespace esphome

210
esphome/components/dsmr/sensor.py Normal file
View File

@@ -0,0 +1,210 @@
import esphome.codegen as cg
import esphome.config_validation as cv
from esphome.components import sensor
from esphome.const import (
DEVICE_CLASS_CURRENT,
DEVICE_CLASS_EMPTY,
DEVICE_CLASS_ENERGY,
DEVICE_CLASS_POWER,
DEVICE_CLASS_VOLTAGE,
ICON_EMPTY,
LAST_RESET_TYPE_NEVER,
STATE_CLASS_MEASUREMENT,
STATE_CLASS_NONE,
UNIT_AMPERE,
UNIT_EMPTY,
UNIT_VOLT,
UNIT_WATT,
)
from . import Dsmr, CONF_DSMR_ID
AUTO_LOAD = ["dsmr"]
CONFIG_SCHEMA = cv.Schema(
{
cv.GenerateID(CONF_DSMR_ID): cv.use_id(Dsmr),
cv.Optional("energy_delivered_lux"): sensor.sensor_schema(
"kWh",
ICON_EMPTY,
3,
DEVICE_CLASS_ENERGY,
STATE_CLASS_MEASUREMENT,
LAST_RESET_TYPE_NEVER,
),
cv.Optional("energy_delivered_tariff1"): sensor.sensor_schema(
"kWh",
ICON_EMPTY,
3,
DEVICE_CLASS_ENERGY,
STATE_CLASS_MEASUREMENT,
LAST_RESET_TYPE_NEVER,
),
cv.Optional("energy_delivered_tariff2"): sensor.sensor_schema(
"kWh",
ICON_EMPTY,
3,
DEVICE_CLASS_ENERGY,
STATE_CLASS_MEASUREMENT,
LAST_RESET_TYPE_NEVER,
),
cv.Optional("energy_returned_lux"): sensor.sensor_schema(
"kWh",
ICON_EMPTY,
3,
DEVICE_CLASS_ENERGY,
STATE_CLASS_MEASUREMENT,
LAST_RESET_TYPE_NEVER,
),
cv.Optional("energy_returned_tariff1"): sensor.sensor_schema(
"kWh",
ICON_EMPTY,
3,
DEVICE_CLASS_ENERGY,
STATE_CLASS_MEASUREMENT,
LAST_RESET_TYPE_NEVER,
),
cv.Optional("energy_returned_tariff2"): sensor.sensor_schema(
"kWh",
ICON_EMPTY,
3,
DEVICE_CLASS_ENERGY,
STATE_CLASS_MEASUREMENT,
LAST_RESET_TYPE_NEVER,
),
cv.Optional("total_imported_energy"): sensor.sensor_schema(
"kvarh", ICON_EMPTY, 3, DEVICE_CLASS_ENERGY, STATE_CLASS_NONE
),
cv.Optional("total_exported_energy"): sensor.sensor_schema(
"kvarh", ICON_EMPTY, 3, DEVICE_CLASS_ENERGY, STATE_CLASS_NONE
),
cv.Optional("power_delivered"): sensor.sensor_schema(
UNIT_WATT, ICON_EMPTY, 3, DEVICE_CLASS_POWER, STATE_CLASS_MEASUREMENT
),
cv.Optional("power_returned"): sensor.sensor_schema(
UNIT_WATT, ICON_EMPTY, 3, DEVICE_CLASS_POWER, STATE_CLASS_MEASUREMENT
),
cv.Optional("reactive_power_delivered"): sensor.sensor_schema(
"kvar", ICON_EMPTY, 3, DEVICE_CLASS_ENERGY, STATE_CLASS_NONE
),
cv.Optional("reactive_power_returned"): sensor.sensor_schema(
"kvar", ICON_EMPTY, 3, DEVICE_CLASS_ENERGY, STATE_CLASS_MEASUREMENT
),
cv.Optional("electricity_threshold"): sensor.sensor_schema(
UNIT_EMPTY, ICON_EMPTY, 3, DEVICE_CLASS_EMPTY, STATE_CLASS_NONE
),
cv.Optional("electricity_switch_position"): sensor.sensor_schema(
UNIT_EMPTY, ICON_EMPTY, 3, DEVICE_CLASS_EMPTY, STATE_CLASS_NONE
),
cv.Optional("electricity_failures"): sensor.sensor_schema(
UNIT_EMPTY, ICON_EMPTY, 0, DEVICE_CLASS_EMPTY, STATE_CLASS_NONE
),
cv.Optional("electricity_long_failures"): sensor.sensor_schema(
UNIT_EMPTY, ICON_EMPTY, 0, DEVICE_CLASS_EMPTY, STATE_CLASS_NONE
),
cv.Optional("electricity_sags_l1"): sensor.sensor_schema(
UNIT_EMPTY, ICON_EMPTY, 0, DEVICE_CLASS_EMPTY, STATE_CLASS_NONE
),
cv.Optional("electricity_sags_l2"): sensor.sensor_schema(
UNIT_EMPTY, ICON_EMPTY, 0, DEVICE_CLASS_EMPTY, STATE_CLASS_MEASUREMENT
),
cv.Optional("electricity_sags_l3"): sensor.sensor_schema(
UNIT_EMPTY, ICON_EMPTY, 0, DEVICE_CLASS_EMPTY, STATE_CLASS_NONE
),
cv.Optional("electricity_swells_l1"): sensor.sensor_schema(
UNIT_EMPTY, ICON_EMPTY, 0, DEVICE_CLASS_EMPTY, STATE_CLASS_MEASUREMENT
),
cv.Optional("electricity_swells_l2"): sensor.sensor_schema(
UNIT_EMPTY, ICON_EMPTY, 0, DEVICE_CLASS_EMPTY, STATE_CLASS_NONE
),
cv.Optional("electricity_swells_l3"): sensor.sensor_schema(
UNIT_EMPTY, ICON_EMPTY, 0, DEVICE_CLASS_EMPTY, STATE_CLASS_NONE
),
cv.Optional("current_l1"): sensor.sensor_schema(
UNIT_AMPERE, ICON_EMPTY, 1, DEVICE_CLASS_CURRENT, STATE_CLASS_MEASUREMENT
),
cv.Optional("current_l2"): sensor.sensor_schema(
UNIT_AMPERE, ICON_EMPTY, 1, DEVICE_CLASS_CURRENT, STATE_CLASS_MEASUREMENT
),
cv.Optional("current_l3"): sensor.sensor_schema(
UNIT_AMPERE, ICON_EMPTY, 1, DEVICE_CLASS_CURRENT, STATE_CLASS_MEASUREMENT
),
cv.Optional("power_delivered_l1"): sensor.sensor_schema(
UNIT_WATT, ICON_EMPTY, 3, DEVICE_CLASS_POWER, STATE_CLASS_MEASUREMENT
),
cv.Optional("power_delivered_l2"): sensor.sensor_schema(
UNIT_WATT, ICON_EMPTY, 3, DEVICE_CLASS_POWER, STATE_CLASS_MEASUREMENT
),
cv.Optional("power_delivered_l3"): sensor.sensor_schema(
UNIT_WATT, ICON_EMPTY, 3, DEVICE_CLASS_POWER, STATE_CLASS_MEASUREMENT
),
cv.Optional("power_returned_l1"): sensor.sensor_schema(
UNIT_WATT, ICON_EMPTY, 3, DEVICE_CLASS_POWER, STATE_CLASS_MEASUREMENT
),
cv.Optional("power_returned_l2"): sensor.sensor_schema(
UNIT_WATT, ICON_EMPTY, 3, DEVICE_CLASS_POWER, STATE_CLASS_MEASUREMENT
),
cv.Optional("power_returned_l3"): sensor.sensor_schema(
UNIT_WATT, ICON_EMPTY, 3, DEVICE_CLASS_POWER, STATE_CLASS_MEASUREMENT
),
cv.Optional("reactive_power_delivered_l1"): sensor.sensor_schema(
UNIT_WATT, ICON_EMPTY, 3, DEVICE_CLASS_POWER, STATE_CLASS_MEASUREMENT
),
cv.Optional("reactive_power_delivered_l2"): sensor.sensor_schema(
UNIT_WATT, ICON_EMPTY, 3, DEVICE_CLASS_POWER, STATE_CLASS_MEASUREMENT
),
cv.Optional("reactive_power_delivered_l3"): sensor.sensor_schema(
UNIT_WATT, ICON_EMPTY, 3, DEVICE_CLASS_POWER, STATE_CLASS_MEASUREMENT
),
cv.Optional("reactive_power_returned_l1"): sensor.sensor_schema(
UNIT_WATT, ICON_EMPTY, 3, DEVICE_CLASS_POWER, STATE_CLASS_MEASUREMENT
),
cv.Optional("reactive_power_returned_l2"): sensor.sensor_schema(
UNIT_WATT, ICON_EMPTY, 3, DEVICE_CLASS_POWER, STATE_CLASS_MEASUREMENT
),
cv.Optional("reactive_power_returned_l3"): sensor.sensor_schema(
UNIT_WATT, ICON_EMPTY, 3, DEVICE_CLASS_POWER, STATE_CLASS_MEASUREMENT
),
cv.Optional("voltage_l1"): sensor.sensor_schema(
UNIT_VOLT, ICON_EMPTY, 1, DEVICE_CLASS_VOLTAGE, STATE_CLASS_NONE
),
cv.Optional("voltage_l2"): sensor.sensor_schema(
UNIT_VOLT, ICON_EMPTY, 1, DEVICE_CLASS_VOLTAGE, STATE_CLASS_NONE
),
cv.Optional("voltage_l3"): sensor.sensor_schema(
UNIT_VOLT, ICON_EMPTY, 1, DEVICE_CLASS_VOLTAGE, STATE_CLASS_NONE
),
cv.Optional("gas_delivered"): sensor.sensor_schema(
"m³",
ICON_EMPTY,
3,
DEVICE_CLASS_EMPTY,
STATE_CLASS_MEASUREMENT,
LAST_RESET_TYPE_NEVER,
),
cv.Optional("gas_delivered_be"): sensor.sensor_schema(
"m³",
ICON_EMPTY,
3,
DEVICE_CLASS_EMPTY,
STATE_CLASS_MEASUREMENT,
LAST_RESET_TYPE_NEVER,
),
}
).extend(cv.COMPONENT_SCHEMA)
async def to_code(config):
hub = await cg.get_variable(config[CONF_DSMR_ID])
sensors = []
for key, conf in config.items():
if not isinstance(conf, dict):
continue
id = conf.get("id")
if id and id.type == sensor.Sensor:
s = await sensor.new_sensor(conf)
cg.add(getattr(hub, f"set_{key}")(s))
sensors.append(f"F({key})")
cg.add_define("DSMR_SENSOR_LIST(F, sep)", cg.RawExpression(" sep ".join(sensors)))

94
esphome/components/dsmr/text_sensor.py Normal file
View File

@@ -0,0 +1,94 @@
import esphome.codegen as cg
import esphome.config_validation as cv
from esphome.components import text_sensor
from esphome.const import (
CONF_ID,
)
from . import Dsmr, CONF_DSMR_ID
AUTO_LOAD = ["dsmr"]
CONFIG_SCHEMA = cv.Schema(
{
cv.GenerateID(CONF_DSMR_ID): cv.use_id(Dsmr),
cv.Optional("identification"): text_sensor.TEXT_SENSOR_SCHEMA.extend(
{
cv.GenerateID(): cv.declare_id(text_sensor.TextSensor),
}
),
cv.Optional("p1_version"): text_sensor.TEXT_SENSOR_SCHEMA.extend(
{
cv.GenerateID(): cv.declare_id(text_sensor.TextSensor),
}
),
cv.Optional("p1_version_be"): text_sensor.TEXT_SENSOR_SCHEMA.extend(
{
cv.GenerateID(): cv.declare_id(text_sensor.TextSensor),
}
),
cv.Optional("timestamp"): text_sensor.TEXT_SENSOR_SCHEMA.extend(
{
cv.GenerateID(): cv.declare_id(text_sensor.TextSensor),
}
),
cv.Optional("electricity_tariff"): text_sensor.TEXT_SENSOR_SCHEMA.extend(
{
cv.GenerateID(): cv.declare_id(text_sensor.TextSensor),
}
),
cv.Optional("electricity_failure_log"): text_sensor.TEXT_SENSOR_SCHEMA.extend(
{
cv.GenerateID(): cv.declare_id(text_sensor.TextSensor),
}
),
cv.Optional("message_short"): text_sensor.TEXT_SENSOR_SCHEMA.extend(
{
cv.GenerateID(): cv.declare_id(text_sensor.TextSensor),
}
),
cv.Optional("message_long"): text_sensor.TEXT_SENSOR_SCHEMA.extend(
{
cv.GenerateID(): cv.declare_id(text_sensor.TextSensor),
}
),
cv.Optional("gas_equipment_id"): text_sensor.TEXT_SENSOR_SCHEMA.extend(
{
cv.GenerateID(): cv.declare_id(text_sensor.TextSensor),
}
),
cv.Optional("thermal_equipment_id"): text_sensor.TEXT_SENSOR_SCHEMA.extend(
{
cv.GenerateID(): cv.declare_id(text_sensor.TextSensor),
}
),
cv.Optional("water_equipment_id"): text_sensor.TEXT_SENSOR_SCHEMA.extend(
{
cv.GenerateID(): cv.declare_id(text_sensor.TextSensor),
}
),
cv.Optional("sub_equipment_id"): text_sensor.TEXT_SENSOR_SCHEMA.extend(
{
cv.GenerateID(): cv.declare_id(text_sensor.TextSensor),
}
),
}
).extend(cv.COMPONENT_SCHEMA)
async def to_code(config):
hub = await cg.get_variable(config[CONF_DSMR_ID])
text_sensors = []
for key, conf in config.items():
if not isinstance(conf, dict):
continue
id = conf.get("id")
if id and id.type == text_sensor.TextSensor:
var = cg.new_Pvariable(conf[CONF_ID])
await text_sensor.register_text_sensor(var, conf)
cg.add(getattr(hub, f"set_{key}")(var))
text_sensors.append(f"F({key})")
cg.add_define(
"DSMR_TEXT_SENSOR_LIST(F, sep)", cg.RawExpression(" sep ".join(text_sensors))
)