Как я сделал обычную посудомоечную машину умной, с Home Assistant-ESPHome

import esphome.codegen as cg
import esphome.config_validation as cv
from esphome.components import text_sensor, binary_sensor, sensor
from esphome.const import CONF_ID, UNIT_MINUTE, ICON_TIMER

CONF_PROGRAM = "program"
CONF_DELAY_TIMER = "delay_timer"
CONF_POWER = "power"
CONF_SALT_MISSING = "salt_missing"
CONF_HALF_LOAD = "half_load"
CONF_RUNNING = "running"
CONF_REMAINING_MINUTES = "remaining_minutes"

AUTO_LOAD = ["text_sensor", "binary_sensor", "sensor"]
CODEOWNERS = [""]

tm1638_sniffer_ns = cg.esphome_ns.namespace("tm1638_sniffer")
TM1638Sniffer = tm1638_sniffer_ns.class_("TM1638Sniffer", cg.Component)

CONFIG_SCHEMA = cv.Schema({
    cv.GenerateID(): cv.declare_id(TM1638Sniffer),

    cv.Optional(CONF_PROGRAM): text_sensor.text_sensor_schema(),
    cv.Optional(CONF_DELAY_TIMER): text_sensor.text_sensor_schema(),

    cv.Optional(CONF_POWER): binary_sensor.binary_sensor_schema(),
    cv.Optional(CONF_SALT_MISSING): binary_sensor.binary_sensor_schema(),
    cv.Optional(CONF_HALF_LOAD): binary_sensor.binary_sensor_schema(),
    cv.Optional(CONF_RUNNING): binary_sensor.binary_sensor_schema(),

    cv.Optional(CONF_REMAINING_MINUTES): sensor.sensor_schema(
        unit_of_measurement=UNIT_MINUTE,
        icon=ICON_TIMER,
        accuracy_decimals=0,
    ),
}).extend(cv.COMPONENT_SCHEMA)

async def to_code(config):
    var = cg.new_Pvariable(config[CONF_ID])
    await cg.register_component(var, config)

    if CONF_PROGRAM in config:
        sens = await text_sensor.new_text_sensor(config[CONF_PROGRAM])
        cg.add(var.set_program_sensor(sens))

    if CONF_DELAY_TIMER in config:
        sens = await text_sensor.new_text_sensor(config[CONF_DELAY_TIMER])
        cg.add(var.set_delay_timer_sensor(sens))

    if CONF_POWER in config:
        sens = await binary_sensor.new_binary_sensor(config[CONF_POWER])
        cg.add(var.set_power_sensor(sens))

    if CONF_SALT_MISSING in config:
        sens = await binary_sensor.new_binary_sensor(config[CONF_SALT_MISSING])
        cg.add(var.set_salt_missing_sensor(sens))

    if CONF_HALF_LOAD in config:
        sens = await binary_sensor.new_binary_sensor(config[CONF_HALF_LOAD])
        cg.add(var.set_half_load_sensor(sens))

    if CONF_RUNNING in config:
        sens = await binary_sensor.new_binary_sensor(config[CONF_RUNNING])
        cg.add(var.set_running_sensor(sens))

    if CONF_REMAINING_MINUTES in config:
        sens = await sensor.new_sensor(config[CONF_REMAINING_MINUTES])
        cg.add(var.set_remaining_minutes_sensor(sens))

#include "tm1638_sniffer.h"
#include "esphome/core/log.h"
#include <Arduino.h>

namespace esphome {
namespace tm1638_sniffer {

static const char *const TAG = "tm1638_sniffer";

#define PIN_DIO D5
#define PIN_CLK D6
#define PIN_STB D7

volatile bool TM1638Sniffer::active_ = false;
volatile uint8_t TM1638Sniffer::bit_count_ = 0;
volatile uint8_t TM1638Sniffer::cur_byte_ = 0;
volatile uint8_t TM1638Sniffer::buf_[128];
volatile uint8_t TM1638Sniffer::len_ = 0;
volatile bool TM1638Sniffer::ready_ = false;

uint8_t TM1638Sniffer::last_buf_[14];
uint8_t TM1638Sniffer::last_len_ = 0;

void TM1638Sniffer::setup() {
  pinMode(PIN_DIO, INPUT_PULLUP);
  pinMode(PIN_CLK, INPUT_PULLUP);
  pinMode(PIN_STB, INPUT_PULLUP);

  attachInterrupt(digitalPinToInterrupt(PIN_STB), TM1638Sniffer::isr_stb, CHANGE);
  attachInterrupt(digitalPinToInterrupt(PIN_CLK), TM1638Sniffer::isr_clk, RISING);

  ESP_LOGI(TAG, "TM1638 dishwasher sniffer started");
}

void TM1638Sniffer::loop() {
  if (!ready_) return;

  noInterrupts();

  uint8_t local_len = len_;

  if (local_len < 4 || local_len > 16) {
    ready_ = false;
    interrupts();
    return;
  }

  uint8_t local_buf[128];

  for (uint8_t i = 0; i < local_len; i++) {
    local_buf[i] = buf_[i];
  }

  ready_ = false;
  interrupts();

  if (local_buf[0] != 0xC0) return;
  if (local_len < 14) return;

  local_len = 14;

  bool same = last_len_ == local_len;

  if (same) {
    for (uint8_t i = 0; i < local_len; i++) {
      if (local_buf[i] != last_buf_[i]) {
        same = false;
        break;
      }
    }
  }

  if (!same) {
    last_len_ = local_len;
    for (uint8_t i = 0; i < local_len; i++) {
      last_buf_[i] = local_buf[i];
    }

    std::string out = "display changed:";
    char tmp[8];

    for (uint8_t i = 0; i < local_len; i++) {
      snprintf(tmp, sizeof(tmp), " %02X", local_buf[i]);
      out += tmp;
    }

    ESP_LOGD(TAG, "%s", out.c_str());

    decode_packet_(local_buf, local_len);
  } else {
    // Даже если дисплей не менялся, таймер остатка нужно обновлять примерно раз в минуту.
    static uint32_t last_timer_update_ms = 0;
    uint32_t now = millis();

    if (was_running_ && selected_duration_min_ > 0 && now - last_timer_update_ms > 60000) {
      last_timer_update_ms = now;

      uint32_t elapsed_min = (now - started_at_ms_) / 60000;
      int remaining = selected_duration_min_ - elapsed_min;
      if (remaining < 0) remaining = 0;

      if (remaining_minutes_sensor_ != nullptr) {
        remaining_minutes_sensor_->publish_state(remaining);
      }
    }
  }
}

int TM1638Sniffer::program_duration_minutes_(const std::string &program) {
  if (program == "Эко") return 175;
  if (program == "Деликатная") return 110;
  if (program == "90 мин") return 90;
  if (program == "Быстрая") return 40;
  if (program == "Интенсивная") return 130;
  if (program == "Стандартная") return 155;
  return 0;
}

void TM1638Sniffer::decode_packet_(uint8_t *d, uint8_t len) {
  if (len < 14) return;

  bool power = d[2] & 0x01;
  bool salt_missing = d[6] & 0x01;
  bool half_load = d[8] & 0x01;

  bool any_program_led =
      (d[6] & 0x02) ||
      (d[8] & 0x02) ||
      (d[10] & 0x02) ||
      (d[12] & 0x02) ||
      (d[2] & 0x02) ||
      (d[4] & 0x02);

  std::string program = "Не выбрана";

  if (d[6] & 0x02) {
    program = "Эко";
  } else if (d[8] & 0x02) {
    program = "Деликатная";
  } else if (d[10] & 0x02) {
    program = "90 мин";
  } else if (d[12] & 0x02) {
    program = "Быстрая";
  } else if (d[2] & 0x02) {
    program = "Интенсивная";
  } else if (d[4] & 0x02) {
    program = "Стандартная";
  }

  if (any_program_led) {
    program_was_selected_ = true;
    selected_program_ = program;
    selected_duration_min_ = program_duration_minutes_(program);
  }

  bool running = power && program_was_selected_ && !any_program_led;

  if (!power) {
    program_was_selected_ = false;
    was_running_ = false;
    started_at_ms_ = 0;
    selected_duration_min_ = 0;
    selected_program_ = "Не выбрана";
  }

  if (running && !was_running_) {
    was_running_ = true;
    started_at_ms_ = millis();

    if (selected_duration_min_ == 0) {
      selected_duration_min_ = program_duration_minutes_(selected_program_);
    }

    ESP_LOGI(TAG, "Dishwasher started: program=%s duration=%d min",
             selected_program_.c_str(), selected_duration_min_);
  }

  if (!running && was_running_) {
    was_running_ = false;
    ESP_LOGI(TAG, "Dishwasher stopped or finished");
  }

  int remaining = 0;

  if (running && selected_duration_min_ > 0) {
    uint32_t elapsed_min = (millis() - started_at_ms_) / 60000;
    remaining = selected_duration_min_ - elapsed_min;
    if (remaining < 0) remaining = 0;
  }

  std::string delay = "Выключен";

  if (d[1] & 0x80) {
    delay = "3 часа";
  } else if (d[3] & 0x80) {
    delay = "6 часов";
  } else if (d[5] & 0x80) {
    delay = "9 часов";
  } else if (d[7] & 0x80) {
    delay = "12 часов";
  }

  if (program_sensor_ != nullptr) {
    program_sensor_->publish_state(any_program_led ? program : selected_program_);
  }

  if (delay_timer_sensor_ != nullptr) {
    delay_timer_sensor_->publish_state(delay);
  }

  if (power_sensor_ != nullptr) {
    power_sensor_->publish_state(power);
  }

  if (salt_missing_sensor_ != nullptr) {
    salt_missing_sensor_->publish_state(salt_missing);
  }

  if (half_load_sensor_ != nullptr) {
    half_load_sensor_->publish_state(half_load);
  }

  if (running_sensor_ != nullptr) {
    running_sensor_->publish_state(running);
  }

  if (remaining_minutes_sensor_ != nullptr) {
    remaining_minutes_sensor_->publish_state(remaining);
  }

  ESP_LOGD(TAG,
           "program=%s selected_program=%s delay=%s power=%s salt_missing=%s half_load=%s running=%s remaining=%d",
           program.c_str(),
           selected_program_.c_str(),
           delay.c_str(),
           power ? "ON" : "OFF",
           salt_missing ? "ON" : "OFF",
           half_load ? "ON" : "OFF",
           running ? "ON" : "OFF",
           remaining);
}

void ICACHE_RAM_ATTR TM1638Sniffer::isr_stb() {
  bool stb = digitalRead(PIN_STB);

  if (!stb) {
    active_ = true;
    bit_count_ = 0;
    cur_byte_ = 0;
    len_ = 0;
  } else {
    active_ = false;
    ready_ = true;
  }
}

void ICACHE_RAM_ATTR TM1638Sniffer::isr_clk() {
  if (!active_) return;
  if (len_ >= sizeof(buf_)) return;

  uint8_t bit = digitalRead(PIN_DIO) ? 1 : 0;

  cur_byte_ |= bit << bit_count_;
  bit_count_++;

  if (bit_count_ == 8) {
    buf_[len_++] = cur_byte_;
    cur_byte_ = 0;
    bit_count_ = 0;
  }
}

}  // namespace tm1638_sniffer
}  // namespace esphome

#pragma once

#include "esphome/core/component.h"
#include "esphome/components/text_sensor/text_sensor.h"
#include "esphome/components/binary_sensor/binary_sensor.h"
#include "esphome/components/sensor/sensor.h"
#include <Arduino.h>

namespace esphome {
namespace tm1638_sniffer {

class TM1638Sniffer : public Component {
 public:
  void setup() override;
  void loop() override;

  void set_program_sensor(text_sensor::TextSensor *sensor) { program_sensor_ = sensor; }
  void set_delay_timer_sensor(text_sensor::TextSensor *sensor) { delay_timer_sensor_ = sensor; }

  void set_power_sensor(binary_sensor::BinarySensor *sensor) { power_sensor_ = sensor; }
  void set_salt_missing_sensor(binary_sensor::BinarySensor *sensor) { salt_missing_sensor_ = sensor; }
  void set_half_load_sensor(binary_sensor::BinarySensor *sensor) { half_load_sensor_ = sensor; }
  void set_running_sensor(binary_sensor::BinarySensor *sensor) { running_sensor_ = sensor; }

  void set_remaining_minutes_sensor(sensor::Sensor *sensor) { remaining_minutes_sensor_ = sensor; }

 protected:
  static void ICACHE_RAM_ATTR isr_stb();
  static void ICACHE_RAM_ATTR isr_clk();

  void decode_packet_(uint8_t *data, uint8_t len);
  int program_duration_minutes_(const std::string &program);

  text_sensor::TextSensor *program_sensor_{nullptr};
  text_sensor::TextSensor *delay_timer_sensor_{nullptr};

  binary_sensor::BinarySensor *power_sensor_{nullptr};
  binary_sensor::BinarySensor *salt_missing_sensor_{nullptr};
  binary_sensor::BinarySensor *half_load_sensor_{nullptr};
  binary_sensor::BinarySensor *running_sensor_{nullptr};

  sensor::Sensor *remaining_minutes_sensor_{nullptr};

  bool program_was_selected_{false};
  bool was_running_{false};
  uint32_t started_at_ms_{0};
  int selected_duration_min_{0};
  std::string selected_program_{"Не выбрана"};

  static volatile bool active_;
  static volatile uint8_t bit_count_;
  static volatile uint8_t cur_byte_;
  static volatile uint8_t buf_[128];
  static volatile uint8_t len_;
  static volatile bool ready_;

  static uint8_t last_buf_[14];
  static uint8_t last_len_;
};

}  // namespace tm1638_sniffer
}  // namespace esphome

esphome:
  name: washmashine
  friendly_name: Washmashine

esp8266:
  board: d1_mini

external_components:
  - source:
      type: local
      path: components

logger:
  level: DEBUG
  baud_rate: 0

api:
  encryption:
    key: ""

ota:
  - platform: esphome
    password: ""

wifi:
  ssid: !secret wifi_ssid
  password: !secret wifi_password
  min_auth_mode: WPA2
  ap:
    ssid: "Washmashine Fallback"
    password: ""

captive_portal:

tm1638_sniffer:
  program:
    name: "Dishwasher Program"
  delay_timer:
    name: "Dishwasher Delay Timer"
  power:
    name: "Dishwasher Power"
  salt_missing:
    name: "Dishwasher Salt Missing"
  half_load:
    name: "Dishwasher Half Load"
  running:
    name: "Dishwasher Running"
  remaining_minutes:
    name: "Dishwasher Remaining Minutes"