ebus implementation, first version

esphome/components/ebus/__init__.py

@@ -0,0 +1,72 @@
+import voluptuous as vol
+import esphome.codegen as cg
+import esphome.config_validation as cv
+
+from esphome.const import (
+    CONF_ID,
+    CONF_TX_PIN,
+    CONF_RX_PIN,
+)
+
+ebus_ns = cg.esphome_ns.namespace("ebus")
+
+CONF_EBUS_ID = "ebus_id"
+CONF_PRIMARY_ADDRESS = "primary_address"
+CONF_MAX_TRIES = "max_tries"
+CONF_MAX_LOCK_COUNTER = "max_lock_counter"
+CONF_HISTORY_QUEUE_SIZE = "history_queue_size"
+CONF_COMMAND_QUEUE_SIZE = "command_queue_size"
+CONF_POLL_INTERVAL = "poll_interval"
+CONF_UART = "uart"
+CONF_NUM = "num"
+
+
+def is_primary_nibble(value):
+    return (value & 0x0F) in {0x00, 0x01, 0x03, 0x07, 0x0F}
+
+
+def validate_primary_address(value):
+    """Validate that the config option is a valid ebus primary address."""
+    if is_primary_nibble(value) and is_primary_nibble(value >> 4):
+        return value & 0xFF
+    raise vol.Invalid(f"'0x{value:02x}' is an invalid ebus primary address")
+
+
+EbusComponent = ebus_ns.class_("EbusComponent", cg.Component)
+
+
+CONFIG_SCHEMA = cv.All(
+    cv.Schema(
+        {
+            cv.GenerateID(): cv.declare_id(EbusComponent),
+            cv.Required(CONF_PRIMARY_ADDRESS): validate_primary_address,
+            cv.Optional(CONF_MAX_TRIES, default=2): cv.hex_uint8_t,
+            cv.Optional(CONF_MAX_LOCK_COUNTER, default=4): cv.hex_uint8_t,
+            cv.Optional(CONF_HISTORY_QUEUE_SIZE, default=20): cv.uint8_t,
+            cv.Optional(CONF_COMMAND_QUEUE_SIZE, default=10): cv.uint8_t,
+            cv.Optional(CONF_POLL_INTERVAL, default="30s"): cv.time_period,
+            cv.Required(CONF_UART): cv.Schema(
+                {
+                    cv.Required(CONF_NUM): cv.uint8_t,
+                    cv.Required(CONF_TX_PIN): cv.uint8_t,
+                    cv.Required(CONF_RX_PIN): cv.uint8_t,
+                }
+            ),
+        }
+    )
+)
+
+
+async def to_code(config):
+    var = cg.new_Pvariable(config[CONF_ID])
+    await cg.register_component(var, config)
+
+    cg.add(var.set_primary_address(config[CONF_PRIMARY_ADDRESS]))
+    cg.add(var.set_max_tries(config[CONF_MAX_TRIES]))
+    cg.add(var.set_max_lock_counter(config[CONF_MAX_LOCK_COUNTER]))
+    cg.add(var.set_uart_num(config[CONF_UART][CONF_NUM]))
+    cg.add(var.set_uart_tx_pin(config[CONF_UART][CONF_TX_PIN]))
+    cg.add(var.set_uart_rx_pin(config[CONF_UART][CONF_RX_PIN]))
+    cg.add(var.set_history_queue_size(config[CONF_HISTORY_QUEUE_SIZE]))
+    cg.add(var.set_command_queue_size(config[CONF_COMMAND_QUEUE_SIZE]))
+    cg.add(var.set_update_interval(config[CONF_POLL_INTERVAL].total_milliseconds))

esphome/components/ebus/ebus.cpp

@@ -0,0 +1,334 @@
+#include "ebus.h"
+
+namespace esphome {
+namespace ebus {
+
+Ebus::Ebus(ebus_config_t &config) {
+  this->primary_address_ = config.primary_address;
+  this->max_tries_ = config.max_tries;
+  this->max_lock_counter_ = config.max_lock_counter;
+}
+
+void Ebus::set_uart_send_function(std::function<void(const char *, int16_t)> uart_send) {
+  this->uart_send_ = uart_send;
+}
+
+void Ebus::set_queue_received_telegram_function(std::function<void(Telegram &telegram)> queue_received_telegram) {
+  this->queue_received_telegram_ = queue_received_telegram;
+}
+
+void Ebus::set_dequeue_command_function(std::function<bool(void *const)> dequeue_command) {
+  this->dequeue_command_ = dequeue_command;
+}
+
+uint8_t Ebus::uart_send_char(uint8_t cr, bool esc, bool run_crc, uint8_t crc_init) {
+  char buffer[2];
+  uint8_t crc = 0;
+  uint8_t len = 1;
+  if (esc && cr == ESC) {
+    buffer[0] = ESC;
+    buffer[1] = 0x00;
+    len = 2;
+  } else if (esc && cr == SYN) {
+    buffer[0] = ESC;
+    buffer[1] = 0x01;
+    len = 2;
+  } else {
+    buffer[0] = cr;
+  }
+  uart_send_(buffer, len);
+  if (!run_crc) {
+    return 0;
+  }
+  crc = Elf::crc8_calc(buffer[0], crc_init);
+  if (len == 1) {
+    return crc;
+  }
+  return Elf::crc8_calc(buffer[1], crc);
+}
+
+void Ebus::uart_send_char(uint8_t cr, bool esc) {
+  this->uart_send_char(cr, esc, false, 0);
+}
+
+void Ebus::uart_send_remaining_request_part(SendCommand &command) {
+  this->uart_send_char(command.getZZ());
+  this->uart_send_char(command.getPB());
+  this->uart_send_char(command.getSB());
+  this->uart_send_char(command.getNN());
+  for (int i = 0; i < command.getNN(); i++) {
+    this->uart_send_char((uint8_t)command.get_request_byte(i));
+  }
+  this->uart_send_char(command.get_crc());
+}
+
+void Ebus::process_received_char(unsigned char received_byte) {
+  // keep track of number of character between last 2 SYN chars
+  // this is needed in case of arbitration
+  if (received_byte == SYN) {
+    this->state = this->char_count_since_last_syn_ == 1 ? EbusState::arbitration : EbusState::normal;
+    this->char_count_since_last_syn_ = 0;
+
+    if (this->lock_counter_ > 0 && this->state == EbusState::normal) {
+      this->lock_counter_--;
+    }
+
+  } else {
+    this->char_count_since_last_syn_++;
+  }
+
+  if (this->receiving_telegram_.is_finished()) {
+    if (this->queue_received_telegram_) {
+      this->queue_received_telegram_(this->receiving_telegram_);
+    }
+    this->receiving_telegram_ = Telegram();
+  }
+
+  if (this->active_command_.is_finished() && this->dequeue_command_) {
+    SendCommand dequeued;
+    if (this->dequeue_command_(&dequeued)) {
+      this->active_command_ = dequeued;
+    }
+  }
+
+  switch (this->receiving_telegram_.get_state()) {
+  case TelegramState::waitForSyn:
+    if (received_byte == SYN) {
+      this->receiving_telegram_.set_state(TelegramState::waitForArbitration);
+    }
+    break;
+  case TelegramState::waitForArbitration:
+    if (received_byte != SYN) {
+      this->receiving_telegram_.push_req_data(received_byte);
+      this->receiving_telegram_.set_state(TelegramState::waitForRequestData);
+    }
+    break;
+  case TelegramState::waitForRequestData:
+    if (received_byte == SYN) {
+      if (this->receiving_telegram_.getZZ() == ESC) {
+        this->receiving_telegram_.set_state(TelegramState::endArbitration);
+      } else {
+        this->receiving_telegram_.set_state(TelegramState::endErrorUnexpectedSyn);
+      }
+    } else {
+      this->receiving_telegram_.push_req_data(received_byte);
+      if (this->receiving_telegram_.is_request_complete()) {
+        this->receiving_telegram_.set_state(this->receiving_telegram_.is_ack_expected() ? TelegramState::waitForRequestAck : TelegramState::endCompleted);
+      }
+    }
+    break;
+  case TelegramState::waitForRequestAck:
+    switch (received_byte) {
+    case ACK:
+      this->receiving_telegram_.set_state(this->receiving_telegram_.is_response_expected() ? TelegramState::waitForResponseData : TelegramState::endCompleted);
+      break;
+    case NACK:
+      this->receiving_telegram_.set_state(TelegramState::endErrorRequestNackReceived);
+      break;
+    default:
+      this->receiving_telegram_.set_state(TelegramState::endErrorRequestNoAck);
+    }
+    break;
+  case TelegramState::waitForResponseData:
+    if (received_byte == SYN) {
+      this->receiving_telegram_.set_state(TelegramState::endErrorUnexpectedSyn);
+    } else {
+      this->receiving_telegram_.push_response_data(received_byte);
+      if (this->receiving_telegram_.is_response_complete()) {
+        this->receiving_telegram_.set_state(TelegramState::waitForResponseAck);
+      }
+    }
+    break;
+  case TelegramState::waitForResponseAck:
+    switch (received_byte) {
+    case ACK:
+      this->receiving_telegram_.set_state(TelegramState::endCompleted);
+      break;
+    case NACK:
+      this->receiving_telegram_.set_state(TelegramState::endErrorResponseNackReceived);
+      break;
+    default:
+      this->receiving_telegram_.set_state(TelegramState::endErrorResponseNoAck);
+    }
+    break;
+  default:
+    break;
+  }
+
+  switch (this->active_command_.get_state()) {
+  case TelegramState::waitForSend:
+    if (received_byte == SYN && state == EbusState::normal && this->lock_counter_ == 0) {
+      this->active_command_.set_state(TelegramState::waitForArbitration);
+      this->uart_send_char(this->active_command_.getQQ());
+    }
+    break;
+  case TelegramState::waitForArbitration:
+    if (received_byte == this->active_command_.getQQ()) {
+      // we won arbitration
+      this->uart_send_remaining_request_part(this->active_command_);
+      if (this->active_command_.is_ack_expected()) {
+        this->active_command_.set_state(TelegramState::waitForCommandAck);
+      } else {
+        this->active_command_.set_state(TelegramState::endCompleted);
+        this->lock_counter_ = this->max_lock_counter_;
+      }
+    } else if (Elf::get_priority_class(received_byte) == Elf::get_priority_class(this->active_command_.getQQ())) {
+      // eligible for round 2
+      this->active_command_.set_state(TelegramState::waitForArbitration2nd);
+    } else {
+      // lost arbitration, try again later if retries left
+      this->active_command_.set_state(this->active_command_.can_retry(this->max_tries_) ? TelegramState::waitForSend : TelegramState::endSendFailed);
+    }
+    break;
+  case TelegramState::waitForArbitration2nd:
+    if (received_byte == SYN) {
+      this->uart_send_char(this->active_command_.getQQ());
+    } else if (received_byte == this->active_command_.getQQ()) {
+      // won round 2
+      this->uart_send_remaining_request_part(this->active_command_);
+      if (this->active_command_.is_ack_expected()) {
+        this->active_command_.set_state(TelegramState::waitForCommandAck);
+      } else {
+        this->active_command_.set_state(TelegramState::endCompleted);
+        this->lock_counter_ = this->max_lock_counter_;
+      }
+    } else {
+      // try again later if retries left
+      this->active_command_.set_state(this->active_command_.can_retry(this->max_tries_) ? TelegramState::waitForSend : TelegramState::endSendFailed);
+    }
+    break;
+  case TelegramState::waitForCommandAck:
+    if (received_byte == ACK) {
+      this->active_command_.set_state(TelegramState::endCompleted);
+      this->lock_counter_ = this->max_lock_counter_;
+    } else if (received_byte == SYN) { // timeout waiting for ACK signaled by AUTO-SYN
+      this->active_command_.set_state(this->active_command_.can_retry(this->max_tries_) ? TelegramState::waitForSend : TelegramState::endSendFailed);
+    }
+    break;
+  default:
+    break;
+  }
+
+  // responses to our commands are stored in receiving_telegram_
+  // when response is completed send ACK or NACK when we were the primary
+  if (this->receiving_telegram_.get_state() == TelegramState::waitForResponseAck &&
+      this->receiving_telegram_.getQQ() == this->primary_address_) {
+    if (this->receiving_telegram_.is_response_valid()) {
+      this->uart_send_char(ACK);
+      this->uart_send_char(SYN, false);
+    } else {
+      this->uart_send_char(NACK);
+    }
+  }
+
+  // Handle our responses
+  this->handle_response(this->receiving_telegram_);
+
+}
+
+void Ebus::add_send_response_handler(std::function<uint8_t(Telegram &, uint8_t *)> send_response_handler) {
+  send_response_handlers_.push_back(send_response_handler);
+}
+
+void Ebus::handle_response(Telegram &telegram) {
+  if (telegram.get_state() != TelegramState::waitForRequestAck ||
+      telegram.getZZ() != Elf::to_secondary(this->primary_address_)) {
+    return;
+  }
+  if (!telegram.is_request_valid()) {
+    uart_send_char(NACK);
+    return;
+  }
+
+  // response buffer
+  uint8_t buf[RESPONSE_BUFFER_SIZE] = {0};
+  int len = 0;
+
+  // find response
+  for (auto const& handler : send_response_handlers_) {
+    len = handler(telegram, buf);
+    if (len != 0) {
+      break;
+    }
+  }
+
+  // we found no reponse to send
+  if (len == 0) {
+    uart_send_char(NACK);
+    return;
+  }
+
+  uart_send_char(ACK);
+  uint8_t crc = Elf::crc8_calc(len, 0);
+  uart_send_char(len);
+  for (int i = 0; i < len; i++) {
+    crc = uart_send_char(buf[i], true, true, crc);
+  }
+  uart_send_char(crc);
+}
+
+unsigned char Elf::crc8_calc(unsigned char data, unsigned char crc_init) {
+  unsigned char crc;
+  unsigned char polynom;
+
+  crc = crc_init;
+  for (int i = 0; i < 8; i++) {
+    if (crc & 0x80) {
+      polynom = (unsigned char)0x9B;
+    } else {
+      polynom = (unsigned char)0;
+    }
+    crc = (unsigned char)((crc & ~0x80) << 1);
+    if (data & 0x80) {
+      crc = (unsigned char)(crc | 1);
+    }
+    crc = (unsigned char)(crc ^ polynom);
+    data = (unsigned char)(data << 1);
+  }
+  return (crc);
+}
+
+unsigned char Elf::crc8_array(unsigned char data[], unsigned int length) {
+  unsigned char uc_crc;
+  uc_crc = (unsigned char)0;
+  for (int i = 0; i < length; i++) {
+    uc_crc = crc8_calc(data[i], uc_crc);
+  }
+  return (uc_crc);
+}
+
+bool Elf::is_primary(uint8_t address) {
+  return is_primary_nibble(get_priority_class(address)) &&  //
+         is_primary_nibble(get_sub_address(address));
+}
+
+int Elf::is_primary_nibble(uint8_t nibble) {
+  switch (nibble) {
+  case 0b0000:
+  case 0b0001:
+  case 0b0011:
+  case 0b0111:
+  case 0b1111:
+    return true;
+  default:
+    return false;
+  }
+}
+
+uint8_t Elf::get_priority_class(uint8_t address) {
+  return (address & 0x0F);
+}
+
+uint8_t Elf::get_sub_address(uint8_t address) {
+  return (address >> 4);
+}
+
+uint8_t Elf::to_secondary(uint8_t address) {
+  if (is_primary(address)) {
+    return (address + 5) % 0xFF;
+  }
+  return address;
+}
+
+}  // namespace ebus
+}  // namespace esphome

esphome/components/ebus/ebus.h

@@ -0,0 +1,62 @@
+#pragma once
+
+#include <stdint.h>
+#include <functional>
+#include <list>
+
+#include "telegram.h"
+
+namespace esphome {
+namespace ebus {
+
+typedef struct {
+  uint8_t primary_address;
+  uint8_t max_tries;
+  uint8_t max_lock_counter;
+} ebus_config_t;
+
+
+class Elf {
+public:
+  static unsigned char crc8_calc(unsigned char data, unsigned char crc_init);
+  static unsigned char crc8_array(unsigned char data[], unsigned int length);
+  static bool is_primary(uint8_t address);
+  static int is_primary_nibble(uint8_t nibble);
+  static uint8_t get_priority_class(uint8_t address);
+  static uint8_t get_sub_address(uint8_t address);
+  static uint8_t to_secondary(uint8_t address);
+};
+
+
+class Ebus {
+public:
+  explicit Ebus(ebus_config_t &config);
+  void set_uart_send_function(std::function<void(const char *, int16_t)> uart_send);
+  void set_queue_received_telegram_function(std::function<void(Telegram &telegram)> queue_received_telegram);
+  void set_dequeue_command_function(std::function<bool(void *const)> dequeue_command);
+  void process_received_char(unsigned char receivedByte);
+  void add_send_response_handler(std::function<uint8_t(Telegram &, uint8_t *)>);
+
+protected:
+  uint8_t primary_address_;
+  uint8_t max_tries_;
+  uint8_t max_lock_counter_;
+  uint8_t lock_counter_ = 0;
+  uint8_t char_count_since_last_syn_ = 0;
+  EbusState state = EbusState::arbitration;
+  Telegram receiving_telegram_;
+  SendCommand active_command_;
+  std::list<std::function<uint8_t(Telegram &, uint8_t *)>> send_response_handlers_;
+
+  std::function<void(const char *, int16_t)> uart_send_;
+  std::function<void(Telegram &)> queue_received_telegram_;
+  std::function<bool(void *const&)> dequeue_command_;
+  uint8_t uart_send_char(uint8_t cr, bool esc, bool run_crc, uint8_t crc_init);
+  void uart_send_char(uint8_t cr, bool esc = true);
+  void uart_send_remaining_request_part(SendCommand &command);
+  void handle_response(Telegram &telegram);
+
+};
+
+}  // namespace ebus
+}  // namespace esphome

esphome/components/ebus/ebus_component.cpp

@@ -0,0 +1,186 @@
+
+#include "ebus_component.h"
+
+namespace esphome {
+namespace ebus {
+
+void EbusComponent::dump_config() {
+  ESP_LOGCONFIG(TAG, "EbusComponent");
+  ESP_LOGCONFIG(TAG, "  primary_addres: 0x%02x", this->primary_address_);
+  ESP_LOGCONFIG(TAG, "  max_tries: %d", this->max_tries_);
+  ESP_LOGCONFIG(TAG, "  max_lock_counter: %d", this->max_lock_counter_);
+  ESP_LOGCONFIG(TAG, "  history_queue_size: %d", this->history_queue_size_);
+  ESP_LOGCONFIG(TAG, "  command_queue_size: %d", this->command_queue_size_);
+  ESP_LOGCONFIG(TAG, "  poll_interval (ms): %d", this->update_interval_);
+  ESP_LOGCONFIG(TAG, "  uart:");
+  ESP_LOGCONFIG(TAG, "    num: %d", this->uart_num_);
+  ESP_LOGCONFIG(TAG, "    tx_pin: %d", this->uart_tx_pin_);
+  ESP_LOGCONFIG(TAG, "    rx_pin: %d", this->uart_rx_pin_);
+}
+
+void EbusComponent::setup() {
+  this->setup_queues();
+  this->setup_ebus();
+  this->setup_uart();
+  this->setup_tasks();
+}
+
+void EbusComponent::set_primary_address(uint8_t primary_address) {
+  this->primary_address_ = primary_address;
+}
+void EbusComponent::set_max_tries(uint8_t max_tries) {
+  this->max_tries_ = max_tries;
+}
+void EbusComponent::set_max_lock_counter(uint8_t max_lock_counter) {
+  this->max_lock_counter_ = max_lock_counter;
+}
+void EbusComponent::set_uart_num(uint8_t uart_num) {
+  this->uart_num_ = uart_num;
+}
+void EbusComponent::set_uart_tx_pin(uint8_t uart_tx_pin) {
+  this->uart_tx_pin_ = uart_tx_pin;
+}
+void EbusComponent::set_uart_rx_pin(uint8_t uart_rx_pin) {
+  this->uart_rx_pin_ = uart_rx_pin;
+}
+void EbusComponent::set_history_queue_size(uint8_t history_queue_size) {
+  this->history_queue_size_ = history_queue_size;
+}
+void EbusComponent::set_command_queue_size(uint8_t command_queue_size) {
+  this->command_queue_size_ = command_queue_size;
+}
+
+void EbusComponent::add_sender(EbusSender *sender) {
+  sender->set_primary_address(this->primary_address_);
+  this->senders_.push_back(sender);
+}
+void EbusComponent::add_receiver(EbusReceiver *receiver) {
+  this->receivers_.push_back(receiver);
+}
+
+void EbusComponent::setup_queues() {
+  this->history_queue_ = xQueueCreate(this->history_queue_size_, sizeof(Telegram));
+  this->command_queue_ = xQueueCreate(this->command_queue_size_, sizeof(Telegram));
+}
+void EbusComponent::setup_ebus() {
+  ebus_config_t ebus_config = ebus_config_t {
+    .primary_address = this->primary_address_,
+    .max_tries = this->max_tries_,
+    .max_lock_counter = this->max_lock_counter_,
+  };
+  this->ebus = new Ebus(ebus_config);
+
+  this->ebus->set_uart_send_function( [&](const char * buffer, int16_t length) {
+    return uart_write_bytes(this->uart_num_, buffer, length);
+  } );
+
+  this->ebus->set_queue_received_telegram_function( [&](Telegram &telegram) {
+    BaseType_t xHigherPriorityTaskWoken;
+    xHigherPriorityTaskWoken = pdFALSE;
+    xQueueSendToBackFromISR(this->history_queue_, &telegram, &xHigherPriorityTaskWoken);
+    if (xHigherPriorityTaskWoken) {
+      portYIELD_FROM_ISR();
+    }
+  } );
+
+  this->ebus->set_dequeue_command_function( [&](void *const command) {
+    BaseType_t xHigherPriorityTaskWoken = pdFALSE;
+    if (xQueueReceiveFromISR(this->command_queue_, command, &xHigherPriorityTaskWoken)) {
+      if (xHigherPriorityTaskWoken) {
+        portYIELD_FROM_ISR();
+      }
+      return true;
+    }
+    return false;
+  } );
+
+}
+
+void EbusComponent::setup_uart() {
+  portMUX_TYPE mux = portMUX_INITIALIZER_UNLOCKED;
+  portENTER_CRITICAL(&mux);
+
+  uart_config_t uart_config = {
+    .baud_rate = 2400,
+    .data_bits = UART_DATA_8_BITS,
+    .parity = UART_PARITY_DISABLE,
+    .stop_bits = UART_STOP_BITS_1,
+    .flow_ctrl = UART_HW_FLOWCTRL_DISABLE,
+    .rx_flow_ctrl_thresh = 2,
+    .use_ref_tick = true,
+  };
+
+  ESP_ERROR_CHECK(uart_param_config(this->uart_num_, &uart_config));
+
+  ESP_ERROR_CHECK(uart_set_pin(
+      this->uart_num_,
+      this->uart_tx_pin_,
+      this->uart_rx_pin_,
+      UART_PIN_NO_CHANGE,
+      UART_PIN_NO_CHANGE));
+
+  ESP_ERROR_CHECK(uart_driver_install(this->uart_num_, 256, 0, 0, NULL, 0));
+
+  portEXIT_CRITICAL(&mux);
+}
+
+void EbusComponent::setup_tasks() {
+  xTaskCreate(&process_received_bytes, "ebus_process_received_bytes", 2048, (void*) this, 10, NULL);
+  xTaskCreate(&process_received_messages, "ebus_process_received_messages", 2560, (void*) this, 5, NULL);
+}
+
+void EbusComponent::process_received_bytes(void *pvParameter) {
+  EbusComponent* instance = static_cast<EbusComponent*>(pvParameter);
+
+  while (1) {
+    uint8_t receivedByte;
+    int len = uart_read_bytes(instance->uart_num_, &receivedByte, 1, 20 / portTICK_PERIOD_MS);
+    if (len) {
+      instance->ebus->process_received_char(receivedByte);
+      taskYIELD();
+    }
+  }
+}
+
+void EbusComponent::process_received_messages(void *pvParameter) {
+  EbusComponent* instance = static_cast<EbusComponent*>(pvParameter);
+
+  Telegram telegram;
+  while (1) {
+    if (xQueueReceive(instance->history_queue_, &telegram, pdMS_TO_TICKS(1000))) {
+      instance->handle_message(telegram);
+      // TODO: this comment is kept as reference on how to debug stack overflows. Could be generalized.
+      // ESP_LOGD(TAG, "Task: %s, Stack Highwater Mark: %d", pcTaskGetTaskName(NULL), uxTaskGetStackHighWaterMark(NULL));
+      taskYIELD();
+    }
+  }
+}
+
+void EbusComponent::handle_message(Telegram &telegram) {
+  if (telegram.get_state() != TelegramState::endCompleted) {
+    ESP_LOGD(TAG, "Message received with invalid state: %s, QQ:%02X, ZZ:%02X, Command:%02X%02X",
+             telegram.get_state_string(),
+             telegram.getQQ(),
+             telegram.getZZ(),
+             telegram.getPB(),
+             telegram.getSB());
+    return;
+  }
+
+  for (auto const& receiver : this->receivers_) {
+    receiver->process_received(telegram);
+  }
+}
+
+void EbusComponent::update() {
+  for (auto const& sender : this->senders_) {
+    optional<SendCommand> command = sender->prepare_command();
+    if (command.has_value()) {
+      xQueueSendToBack(this->command_queue_, &command.value(), portMAX_DELAY);
+    }
+  }
+
+}
+
+} // namespace ebus
+} // namespace esphome

esphome/components/ebus/ebus_component.h

@@ -0,0 +1,82 @@
+#pragma once
+
+#include "esphome/core/log.h"
+#include "esphome/core/component.h"
+
+#include "ebus.h"
+
+#include <driver/uart.h>
+#include <soc/uart_reg.h>
+
+
+namespace esphome {
+namespace ebus {
+
+  static const char *const TAG = "ebus";
+
+class EbusReceiver {
+public:
+  EbusReceiver() {}
+  virtual void process_received(Telegram) = 0;
+};
+
+class EbusSender {
+public:
+  EbusSender() {}
+  virtual void set_primary_address(uint8_t) = 0;
+  virtual optional<SendCommand> prepare_command() = 0;
+};
+
+class EbusComponent : public PollingComponent {
+public:
+  EbusComponent() {
+  }
+
+  void dump_config() override;
+  void setup() override;
+
+  void set_primary_address(uint8_t);
+  void set_max_tries(uint8_t);
+  void set_max_lock_counter(uint8_t);
+  void set_uart_num(uint8_t);
+  void set_uart_tx_pin(uint8_t);
+  void set_uart_rx_pin(uint8_t);
+  void set_history_queue_size(uint8_t);
+  void set_command_queue_size(uint8_t);
+
+  void add_sender(EbusSender *);
+  void add_receiver(EbusReceiver *);
+
+  void update();
+
+protected:
+  uint8_t primary_address_;
+  uint8_t max_tries_;
+  uint8_t max_lock_counter_;
+  uint8_t history_queue_size_;
+  uint8_t command_queue_size_;
+  uint8_t uart_num_;
+  uint8_t uart_tx_pin_;
+  uint8_t uart_rx_pin_;
+
+  QueueHandle_t history_queue_;
+  QueueHandle_t command_queue_;
+
+  std::list<EbusSender *> senders_;
+  std::list<EbusReceiver *> receivers_;
+
+  Ebus* ebus;
+
+  void setup_queues();
+  void setup_ebus();
+  void setup_uart();
+  void setup_tasks();
+
+  static void process_received_bytes(void *);
+  static void process_received_messages(void *);
+  void handle_message(Telegram &);
+
+};
+
+} // namespace ebus
+} // namespace esphome

esphome/components/ebus/sensor/__init__.py

@@ -0,0 +1,99 @@
+import voluptuous as vol
+import esphome.codegen as cg
+import esphome.config_validation as cv
+from esphome.components import sensor
+
+from esphome.const import (
+    CONF_SENSORS,
+    CONF_SOURCE,
+    CONF_COMMAND,
+    CONF_PAYLOAD,
+    CONF_POSITION,
+    CONF_BYTES,
+)
+
+from .. import EbusComponent, CONF_EBUS_ID, ebus_ns
+
+AUTO_LOAD = ["ebus"]
+
+EbusSensor = ebus_ns.class_("EbusSensor", sensor.Sensor, cg.Component)
+
+CONF_TELEGRAM = "telegram"
+CONF_DESTINATION = "destination"
+CONF_DECODE = "decode"
+CONF_DIVIDER = "divider"
+
+
+SYN = 0xAA
+ESC = 0xA9
+
+
+def validate_address(destination):
+    if destination == SYN:
+        raise vol.Invalid("SYN symbol (0xAA) is not a valid address")
+    if destination == ESC:
+        raise vol.Invalid("ESC symbol (0xA9) is not a valid address")
+    return cv.hex_uint8_t(destination)
+
+
+CONFIG_SCHEMA = cv.Schema(
+    {
+        cv.GenerateID(): cv.declare_id(EbusSensor),
+        cv.GenerateID(CONF_EBUS_ID): cv.use_id(EbusComponent),
+        cv.Required(CONF_SENSORS): cv.ensure_list(
+            sensor.sensor_schema().extend(
+                {
+                    cv.GenerateID(): cv.declare_id(EbusSensor),
+                    cv.Required(CONF_TELEGRAM): cv.Schema(
+                        {
+                            cv.Optional(CONF_SOURCE): validate_address,
+                            cv.Optional(CONF_DESTINATION): validate_address,
+                            cv.Required(CONF_COMMAND): cv.hex_uint16_t,
+                            cv.Required(CONF_PAYLOAD): cv.Schema([cv.hex_uint8_t]),
+                            cv.Optional(CONF_DECODE): cv.Schema(
+                                {
+                                    cv.Optional(CONF_POSITION, default=0): cv.int_range(
+                                        0, 15
+                                    ),
+                                    cv.Required(CONF_BYTES): cv.int_range(1, 4),
+                                    cv.Required(CONF_DIVIDER): cv.float_,
+                                }
+                            ),
+                        }
+                    ),
+                }
+            )
+        ),
+    }
+)
+
+
+async def to_code(config):
+    ebus = await cg.get_variable(config[CONF_EBUS_ID])
+
+    for i, conf in enumerate(config[CONF_SENSORS]):
+        print(f"Sensor: {i}, {conf}")
+        sens = await sensor.new_sensor(conf)
+        if CONF_SOURCE in conf[CONF_TELEGRAM]:
+            cg.add(sens.set_source(conf[CONF_TELEGRAM][CONF_SOURCE]))
+
+        if CONF_DESTINATION in conf[CONF_TELEGRAM]:
+            cg.add(sens.set_destination(conf[CONF_TELEGRAM][CONF_DESTINATION]))
+        cg.add(sens.set_command(conf[CONF_TELEGRAM][CONF_COMMAND]))
+        cg.add(sens.set_payload(conf[CONF_TELEGRAM][CONF_PAYLOAD]))
+        cg.add(
+            sens.set_response_read_position(
+                conf[CONF_TELEGRAM][CONF_DECODE][CONF_POSITION]
+            )
+        )
+        cg.add(
+            sens.set_response_read_bytes(conf[CONF_TELEGRAM][CONF_DECODE][CONF_BYTES])
+        )
+        cg.add(
+            sens.set_response_read_divider(
+                conf[CONF_TELEGRAM][CONF_DECODE][CONF_DIVIDER]
+            )
+        )
+
+        cg.add(ebus.add_receiver(sens))
+        cg.add(ebus.add_sender(sens))

esphome/components/ebus/sensor/ebus_sensor.cpp

@@ -0,0 +1,100 @@
+
+#include "ebus_sensor.h"
+
+// TODO: remove
+#define GET_BYTE(CMD, I) ((uint8_t) ((CMD >> 8 * I) & 0XFF))
+
+namespace esphome {
+namespace ebus {
+
+void EbusSensor::dump_config() {
+  ESP_LOGCONFIG(TAG, "EbusSensor");
+  ESP_LOGCONFIG(TAG, "  message:");
+  if (this->source_ == SYN) {
+    ESP_LOGCONFIG(TAG, "    source: N/A");
+  } else {
+    ESP_LOGCONFIG(TAG, "    source: 0x%02x", this->source_);
+  }
+  if (this->destination_ == SYN) {
+    ESP_LOGCONFIG(TAG, "    destination: N/A");
+  } else {
+    ESP_LOGCONFIG(TAG, "    destination: 0x%02x", this->destination_);
+  }
+  ESP_LOGCONFIG(TAG, "    command: 0x%04x", this->command_);
+};
+
+void EbusSensor::set_primary_address(uint8_t primary_address) {
+  this->primary_address_ = primary_address;
+}
+void EbusSensor::set_source(uint8_t source) {
+  this->source_ = source;
+}
+void EbusSensor::set_destination(uint8_t destination) {
+  this->destination_ = destination;
+}
+void EbusSensor::set_command(uint16_t command) {
+  this->command_ = command;
+}
+void EbusSensor::set_payload(const std::vector<uint8_t> &payload) {
+  this->payload_ = payload;
+}
+void EbusSensor::set_response_read_position(uint8_t response_position) {
+  this->response_position_ = response_position;
+}
+void EbusSensor::set_response_read_bytes(uint8_t response_bytes) {
+  this->response_bytes_ = response_bytes;
+}
+void EbusSensor::set_response_read_divider(float response_divider) {
+  this->response_divider_ = response_divider;
+}
+
+optional<SendCommand> EbusSensor::prepare_command() {
+  optional<SendCommand> command;
+  if (this->destination_ != SYN) {
+    command = SendCommand(  //
+         this->primary_address_,
+         Elf::to_secondary(this->destination_),
+         GET_BYTE(this->command_, 1),
+         GET_BYTE(this->command_, 0),
+         this->payload_.size(),
+         &this->payload_[0]);
+  }
+  return command;
+}
+
+void EbusSensor::process_received(Telegram telegram) {
+  if (!is_mine(telegram)) {
+    return;
+  }
+  this->publish_state(to_float(telegram, this->response_position_, this->response_bytes_, this->response_divider_));
+}
+
+uint32_t EbusSensor::get_response_bytes(Telegram &telegram, uint8_t start, uint8_t length) {
+  uint32_t result = 0;
+  for (uint8_t i = 0; i < 4 && i < length; i++) {
+    result = result | (telegram.get_response_byte(start + i) << (i * 8));
+  }
+  return result;
+}
+
+float EbusSensor::to_float(Telegram &telegram, uint8_t start, uint8_t length, float divider) {
+  return get_response_bytes(telegram, start, length) / divider;
+}
+
+bool EbusSensor::is_mine(Telegram &telegram) {
+  if (this->source_ != SYN && this->source_ != telegram.getZZ()) {
+    return false;
+  }
+  if (telegram.getCommand() != this->command_) {
+    return false;
+  }
+  for (int i = 0; i < this->payload_.size(); i++) {
+    if (this->payload_[i] != telegram.get_request_byte(i)) {
+      return false;
+    }
+  }
+  return true;
+}
+
+}  // namespace ebus
+}  // namespace esphome

esphome/components/ebus/sensor/ebus_sensor.h

@@ -0,0 +1,46 @@
+#pragma once
+
+#include "../ebus_component.h"
+#include "esphome/components/sensor/sensor.h"
+
+namespace esphome {
+namespace ebus {
+
+class EbusSensor : public EbusReceiver, public EbusSender, public sensor::Sensor, public Component {
+public:
+  EbusSensor() {
+  }
+
+  void dump_config() override;
+
+  void set_primary_address(uint8_t) override;
+  void set_source(uint8_t);
+  void set_destination(uint8_t);
+  void set_command(uint16_t);
+  void set_payload(const std::vector<uint8_t> &);
+
+  void set_response_read_position(uint8_t);
+  void set_response_read_bytes(uint8_t);
+  void set_response_read_divider(float);
+
+  void process_received(Telegram) override;
+  optional<SendCommand> prepare_command() override;
+
+  // TODO: refactor these
+  uint32_t get_response_bytes(Telegram &telegram, uint8_t start, uint8_t length);
+  float to_float(Telegram &telegram, uint8_t start, uint8_t length, float divider);
+  bool is_mine(Telegram &telegram);
+
+protected:
+  uint8_t primary_address_;
+  uint8_t source_ = SYN;
+  uint8_t destination_ = SYN;
+  uint16_t command_;
+  std::vector<uint8_t> payload_{};
+  uint8_t response_position_;
+  uint8_t response_bytes_;
+  float response_divider_;
+};
+
+}  // namespace ebus
+}  // namespace esphome

esphome/components/ebus/telegram.cpp

@@ -0,0 +1,148 @@
+#include "ebus.h"
+
+namespace esphome {
+namespace ebus {
+
+TelegramBase::TelegramBase() {
+}
+
+void TelegramBase::set_state(TelegramState new_state) {
+  this->state = new_state;
+}
+
+TelegramState TelegramBase::get_state() {
+  return this->state;
+}
+
+#define X(name, int) case int: return ""#name"";
+const char * TelegramBase::get_state_string() {
+  switch((int8_t) this->state) {
+    TELEGRAM_STATE_TABLE
+    default:
+      return "[INVALID STATE]";
+  }
+}
+#undef X
+
+
+void TelegramBase::push_buffer(uint8_t cr, uint8_t *buffer, uint8_t *pos, uint8_t *crc, int max_pos) {
+  if (*pos < max_pos) {
+    *crc = Elf::crc8_calc(cr, *crc);
+  }
+  if (this->wait_for_escaped_char_) {
+    buffer[(*pos)] = (cr == 0x0 ? ESC : SYN);
+    this->wait_for_escaped_char_ = false;
+  } else {
+    buffer[(*pos)++] = cr;
+    this->wait_for_escaped_char_ = (cr == ESC);
+  }
+}
+
+TelegramType TelegramBase::get_type() {
+  if (this->getZZ() == ESC) {
+    return TelegramType::Unknown;
+  }
+  if (this->getZZ() == BROADCAST_ADDRESS) {
+    return TelegramType::Broadcast;
+  }
+  if (Elf::is_primary(this->getZZ())) {
+    return TelegramType::PrimaryPrimary;
+  }
+  return TelegramType::PrimarySecondary;
+}
+
+int16_t TelegramBase::get_request_byte(uint8_t pos) {
+  if (pos > this->getNN() || pos >= MAX_DATA_LENGTH) {
+    return -1;
+  }
+  return this->request_buffer[OFFSET_DATA + pos];
+}
+
+uint8_t TelegramBase::get_request_crc() {
+  return this->request_buffer[OFFSET_DATA + this->getNN()];
+}
+
+void TelegramBase::push_req_data(uint8_t cr) {
+  this->push_buffer(cr, request_buffer, &request_buffer_pos, &request_rolling_crc, OFFSET_DATA + getNN());
+}
+
+bool TelegramBase::is_ack_expected() {
+  return (this->get_type() != TelegramType::Broadcast);
+}
+
+bool TelegramBase::is_response_expected() {
+  return (this->get_type() == TelegramType::PrimarySecondary);
+}
+
+bool TelegramBase::is_finished() {
+  return this->state < TelegramState::unknown;
+}
+
+
+Telegram::Telegram() {
+  this->state = TelegramState::waitForSyn;
+}
+
+int16_t Telegram::get_response_byte(uint8_t pos) {
+  if (pos > this->getResponseNN() || pos >= MAX_DATA_LENGTH) {
+    return INVALID_RESPONSE_BYTE;
+  }
+  return this->response_buffer[RESPONSE_OFFSET + pos];
+}
+
+uint8_t Telegram::get_response_crc() {
+  return this->response_buffer[RESPONSE_OFFSET + this->getResponseNN()];
+}
+
+void Telegram::push_response_data(uint8_t cr) {
+  this->push_buffer(cr, response_buffer, &response_buffer_pos, &response_rolling_crc, RESPONSE_OFFSET + getResponseNN());
+}
+
+bool Telegram::is_response_complete() {
+  return (this->state > TelegramState::waitForSyn || this->state == TelegramState::endCompleted) &&
+         (this->response_buffer_pos > RESPONSE_OFFSET) &&
+         (this->response_buffer_pos == (RESPONSE_OFFSET + this->getResponseNN() + 1)) &&
+         !this->wait_for_escaped_char_;
+}
+
+bool Telegram::is_response_valid() {
+  return this->is_response_complete() && this->get_response_crc() == this->response_rolling_crc;
+}
+
+bool Telegram::is_request_complete() {
+  return (this->state > TelegramState::waitForSyn || this->state == TelegramState::endCompleted) &&
+         (this->request_buffer_pos > OFFSET_DATA) &&
+         (this->request_buffer_pos == (OFFSET_DATA + this->getNN() + 1)) && !this->wait_for_escaped_char_;
+}
+bool Telegram::is_request_valid() {
+  return this->is_request_complete() && this->get_request_crc() == this->request_rolling_crc;
+}
+
+
+SendCommand::SendCommand() {
+  this->state = TelegramState::endCompleted;
+}
+
+SendCommand::SendCommand(uint8_t QQ, uint8_t ZZ, uint8_t PB, uint8_t SB, uint8_t NN, uint8_t *data) {
+  this->state = TelegramState::waitForSend;
+  this->push_req_data(QQ);
+  this->push_req_data(ZZ);
+  this->push_req_data(PB);
+  this->push_req_data(SB);
+  this->push_req_data(NN);
+  for (int i = 0; i < NN; i++) {
+    this->push_req_data(data[i]);
+  }
+  this->push_req_data(this->request_rolling_crc);
+}
+
+bool SendCommand::can_retry(int8_t max_tries) {
+  return this->tries_count_++ < max_tries;
+}
+
+uint8_t SendCommand::get_crc() {
+  return this->request_rolling_crc;
+}
+
+}  // namespace ebus
+}  // namespace esphome

esphome/components/ebus/telegram.h

@@ -0,0 +1,161 @@
+#pragma once
+
+namespace esphome {
+namespace ebus {
+
+enum EbusState : int8_t {
+  normal,
+  arbitration,
+};
+
+enum TelegramType : int8_t {
+  Unknown = -1,
+  Broadcast = 0,
+  PrimaryPrimary = 1,
+  PrimarySecondary = 2,
+};
+
+#define TELEGRAM_STATE_TABLE \
+X(waitForSyn, 1)                    \
+X(waitForSend, 2)                   \
+X(waitForRequestData, 3)            \
+X(waitForRequestAck, 4)             \
+X(waitForResponseData, 5)           \
+X(waitForResponseAck, 6)            \
+X(waitForArbitration, 7)            \
+X(waitForArbitration2nd, 8)         \
+X(waitForCommandAck, 9)             \
+X(unknown, 0)                       \
+X(endErrorUnexpectedSyn, -1)        \
+X(endErrorRequestNackReceived, -2)  \
+X(endErrorResponseNackReceived, -3) \
+X(endErrorResponseNoAck, -4)        \
+X(endErrorRequestNoAck, -5)         \
+X(endArbitration, -6)               \
+X(endCompleted, -16)                \
+X(endSendFailed, -17)
+
+#define X(name, int) name = int,
+enum TelegramState : int8_t {
+  TELEGRAM_STATE_TABLE
+};
+#undef X
+
+  const uint8_t SYN = 0xAA;
+  const uint8_t ESC = 0xA9;
+  const uint8_t ACK = 0x00;
+  const uint8_t NACK = 0xFF;
+
+  const uint8_t BROADCAST_ADDRESS = 0xFE;
+
+  /* Specification says:
+     1. In primary and secondary telegram part, standardised commands must be limited to 10 used data bytes.
+     2. In primary and secondary telegram part, the sum of mfr.-specific telegram used data bytes must not exceed 14.
+     We use 16 to be on the safe side for now.
+  */
+  const uint8_t MAX_DATA_LENGTH = 16;
+  const uint8_t OFFSET_QQ = 0;
+  const uint8_t OFFSET_ZZ = 1;
+  const uint8_t OFFSET_PB = 2;
+  const uint8_t OFFSET_SB = 3;
+  const uint8_t OFFSET_NN = 4;
+  const uint8_t OFFSET_DATA = 5;
+  const uint8_t REQUEST_BUFFER_SIZE = (OFFSET_DATA + MAX_DATA_LENGTH + 1);
+  const uint8_t RESPONSE_BUFFER_SIZE = (MAX_DATA_LENGTH + 2);
+  const uint8_t RESPONSE_OFFSET = 1;
+  const uint8_t INVALID_RESPONSE_BYTE = -1;
+
+
+class TelegramBase {
+public:
+  TelegramBase();
+
+  uint8_t getQQ() {
+    return this->request_buffer[OFFSET_QQ];
+  }
+  uint8_t getZZ() {
+    return this->request_buffer[OFFSET_ZZ];
+  }
+  uint8_t getPB() {
+    return this->request_buffer[OFFSET_PB];
+  }
+  uint8_t getSB() {
+    return this->request_buffer[OFFSET_SB];
+  }
+  uint16_t getCommand() {
+    return ((uint16_t) getPB()) << 8 | getSB();
+  }
+  uint8_t getNN() {
+    uint8_t nn = this->request_buffer[OFFSET_NN];
+    if (nn >= MAX_DATA_LENGTH) {
+      return 0;
+    }
+    return nn;
+  }
+
+  void set_state(TelegramState new_state);
+  TelegramState get_state();
+  const char * get_state_string();
+
+  TelegramType get_type();
+  int16_t get_request_byte(uint8_t pos);
+  uint8_t get_request_crc();
+  void push_req_data(uint8_t cr);
+  bool is_ack_expected();
+  bool is_response_expected();
+  bool is_finished();
+
+protected:
+  TelegramState state;
+  uint8_t request_buffer[REQUEST_BUFFER_SIZE] = {ESC, ESC};  // initialize QQ and ZZ with ESC char to distinguish from valid primary 0
+  uint8_t request_buffer_pos = 0;
+  uint8_t request_rolling_crc = 0;
+  bool wait_for_escaped_char_ = false;
+  void push_buffer(uint8_t cr, uint8_t *buffer, uint8_t *pos, uint8_t *crc, int max_pos);
+
+};
+
+
+class Telegram : public TelegramBase {
+public:
+  Telegram();
+
+  uint8_t getResponseNN() {
+    uint8_t nn = response_buffer[0];
+    if (nn >= MAX_DATA_LENGTH) {
+      return 0;
+    }
+    return nn;
+  }
+
+  int16_t get_response_byte(uint8_t pos);
+  uint8_t get_response_crc();
+
+  void push_response_data(uint8_t cr);
+  bool is_response_complete();
+  bool is_response_valid();
+  bool is_request_complete();
+  bool is_request_valid();
+
+protected:
+  uint8_t response_buffer[RESPONSE_BUFFER_SIZE] = {0};
+  uint8_t response_buffer_pos = 0;
+  uint8_t response_rolling_crc = 0;
+
+};
+
+class SendCommand : public TelegramBase {
+public:
+  SendCommand();
+  SendCommand(uint8_t QQ, uint8_t ZZ, uint8_t PB, uint8_t SB, uint8_t NN, uint8_t *data);
+  bool can_retry(int8_t max_tries);
+  uint8_t get_crc();
+
+protected:
+  uint8_t tries_count_ = 0;
+
+};
+
+
+}  // namespace ebus
+}  // namespace esphome