[tcp_uart][uart_tcp] Add a disconnects sensor (#20096)

esphome/components/tcp_uart/__init__.py

@@ -1,5 +1,5 @@
 import esphome.codegen as cg
-from esphome.components import binary_sensor, socket, uart
+from esphome.components import binary_sensor, sensor, socket, uart
 from esphome.components.const import (
     CONF_ALLOWED_IPS,
     CONF_CONNECTED,
@@ -17,14 +17,17 @@ from esphome.const import (
     CONF_PORT,
     DEVICE_CLASS_CONNECTIVITY,
     ENTITY_CATEGORY_DIAGNOSTIC,
+    STATE_CLASS_TOTAL_INCREASING,
 )
 from esphome.types import ConfigType
 
 CODEOWNERS = ["@Bascht74"]
 DEPENDENCIES = ["network"]
-AUTO_LOAD = ["uart", "binary_sensor", "socket"]
+AUTO_LOAD = ["uart", "binary_sensor", "sensor", "socket"]
 MULTI_CONF = True
 
+CONF_DISCONNECTS = "disconnects"
+
 tcp_uart_ns = cg.esphome_ns.namespace("tcp_uart")
 TcpUart = tcp_uart_ns.class_("TcpUart", uart.UARTComponent, cg.Component)
 
@@ -46,6 +49,11 @@ BASE_SCHEMA = cv.Schema(
             device_class=DEVICE_CLASS_CONNECTIVITY,
             entity_category=ENTITY_CATEGORY_DIAGNOSTIC,
         ),
+        cv.Optional(CONF_DISCONNECTS): sensor.sensor_schema(
+            accuracy_decimals=0,
+            state_class=STATE_CLASS_TOTAL_INCREASING,
+            entity_category=ENTITY_CATEGORY_DIAGNOSTIC,
+        ),
     }
 ).extend(cv.COMPONENT_SCHEMA)
 
@@ -93,3 +101,5 @@ async def to_code(config: ConfigType) -> None:
         cg.add(var.set_host(host))
     binary_sensors = binary_sensor.sub_binary_sensors(config)
     await binary_sensors(CONF_CONNECTED, var.set_connected_sensor)
+    sensors = sensor.sub_sensors(config)
+    await sensors(CONF_DISCONNECTS, var.set_disconnects_sensor)

esphome/components/tcp_uart/tcp_uart.cpp

@@ -20,6 +20,11 @@ void TcpUart::setup() {
   if (this->connected_sensor_ != nullptr) {
     this->connected_sensor_->publish_state(false);
   }
+#ifdef USE_SENSOR
+  if (this->disconnects_sensor_ != nullptr) {
+    this->disconnects_sensor_->publish_state(0);
+  }
+#endif
 }
 
 void TcpUart::dump_config() {
@@ -34,6 +39,9 @@ void TcpUart::dump_config() {
   this->listener_.dump_config();
 #endif
   LOG_BINARY_SENSOR("  ", "Connected", this->connected_sensor_);
+#ifdef USE_SENSOR
+  LOG_SENSOR("  ", "Disconnects", this->disconnects_sensor_);
+#endif
 }
 
 void TcpUart::on_shutdown() {
@@ -53,6 +61,13 @@ void TcpUart::sync_link_() {
   if (this->connected_sensor_ != nullptr) {
     this->connected_sensor_->publish_state(up);
   }
+#ifdef USE_SENSOR
+  // Only edges get here, so down is the falling edge.
+  if (!up && this->disconnects_sensor_ != nullptr) {
+    this->disconnects_++;
+    this->disconnects_sensor_->publish_state(this->disconnects_);
+  }
+#endif
 }
 
 void TcpUart::read_socket_() {

esphome/components/tcp_uart/tcp_uart.h

@@ -1,6 +1,9 @@
 #pragma once
 
 #include "esphome/components/binary_sensor/binary_sensor.h"
+#ifdef USE_SENSOR
+#include "esphome/components/sensor/sensor.h"
+#endif
 #include "esphome/components/socket/tcp_client_link.h"
 #ifdef USE_SOCKET_TCP_LISTENER
 #include "esphome/components/socket/tcp_listener.h"
@@ -21,6 +24,9 @@ class TcpUart : public uart::UARTComponent, public Component {
   void set_port(uint16_t port) { this->link_.set_port(port); }
   void set_reconnect_interval(uint32_t ms) { this->link_.set_reconnect_interval(ms); }
   void set_connected_sensor(binary_sensor::BinarySensor *sensor) { this->connected_sensor_ = sensor; }
+#ifdef USE_SENSOR
+  void set_disconnects_sensor(sensor::Sensor *sensor) { this->disconnects_sensor_ = sensor; }
+#endif
 #ifdef USE_SOCKET_TCP_LISTENER
   void set_server(bool server) { this->server_ = server; }
 #ifdef USE_SOCKET_IPV4_ALLOW
@@ -59,6 +65,10 @@ class TcpUart : public uart::UARTComponent, public Component {
   socket::TcpListener listener_;
 #endif
   binary_sensor::BinarySensor *connected_sensor_{nullptr};
+#ifdef USE_SENSOR
+  sensor::Sensor *disconnects_sensor_{nullptr};
+  uint32_t disconnects_{0};
+#endif
   uint32_t last_drop_log_ms_{0};
   // rx_[rx_start_, rx_end_) holds unread bytes; read_socket_() compacts to the front.
   uint16_t rx_start_{0};

esphome/components/uart_tcp/__init__.py

@@ -1,5 +1,5 @@
 import esphome.codegen as cg
-from esphome.components import binary_sensor, socket, uart
+from esphome.components import binary_sensor, sensor, socket, uart
 from esphome.components.const import (
     CONF_ALLOWED_IPS,
     CONF_CONNECTED,
@@ -16,6 +16,7 @@ from esphome.const import (
     CONF_UART_ID,
     DEVICE_CLASS_CONNECTIVITY,
     ENTITY_CATEGORY_DIAGNOSTIC,
+    STATE_CLASS_TOTAL_INCREASING,
 )
 from esphome.core import CORE
 import esphome.final_validate as fv
@@ -24,9 +25,11 @@ from esphome.types import ConfigType
 CODEOWNERS = ["@Bascht74"]
 DOMAIN = "uart_tcp"
 DEPENDENCIES = ["network", "uart"]
-AUTO_LOAD = ["binary_sensor", "socket"]
+AUTO_LOAD = ["binary_sensor", "sensor", "socket"]
 MULTI_CONF = True
 
+CONF_DISCONNECTS = "disconnects"
+
 uart_tcp_ns = cg.esphome_ns.namespace("uart_tcp")
 UartTcp = uart_tcp_ns.class_("UartTcp", cg.Component, uart.UARTDevice)
 
@@ -43,6 +46,11 @@ BASE_SCHEMA = cv.Schema(
             device_class=DEVICE_CLASS_CONNECTIVITY,
             entity_category=ENTITY_CATEGORY_DIAGNOSTIC,
         ),
+        cv.Optional(CONF_DISCONNECTS): sensor.sensor_schema(
+            accuracy_decimals=0,
+            state_class=STATE_CLASS_TOTAL_INCREASING,
+            entity_category=ENTITY_CATEGORY_DIAGNOSTIC,
+        ),
     }
 ).extend(cv.COMPONENT_SCHEMA)
 
@@ -149,3 +157,5 @@ async def to_code(config: ConfigType) -> None:
         cg.add(var.set_host(host))
     binary_sensors = binary_sensor.sub_binary_sensors(config)
     await binary_sensors(CONF_CONNECTED, var.set_connected_sensor)
+    sensors = sensor.sub_sensors(config)
+    await sensors(CONF_DISCONNECTS, var.set_disconnects_sensor)

esphome/components/uart_tcp/uart_tcp.cpp

@@ -22,6 +22,11 @@ void UartTcp::setup() {
   if (this->connected_sensor_ != nullptr) {
     this->connected_sensor_->publish_state(false);
   }
+#ifdef USE_SENSOR
+  if (this->disconnects_sensor_ != nullptr) {
+    this->disconnects_sensor_->publish_state(0);
+  }
+#endif
 }
 
 void UartTcp::dump_config() {
@@ -36,6 +41,9 @@ void UartTcp::dump_config() {
   this->listener_.dump_config();
 #endif
   LOG_BINARY_SENSOR("  ", "Connected", this->connected_sensor_);
+#ifdef USE_SENSOR
+  LOG_SENSOR("  ", "Disconnects", this->disconnects_sensor_);
+#endif
 }
 
 void UartTcp::on_shutdown() {
@@ -55,6 +63,13 @@ void UartTcp::sync_link_() {
   if (this->connected_sensor_ != nullptr) {
     this->connected_sensor_->publish_state(up);
   }
+#ifdef USE_SENSOR
+  // Only edges get here, so down is the falling edge.
+  if (!up && this->disconnects_sensor_ != nullptr) {
+    this->disconnects_++;
+    this->disconnects_sensor_->publish_state(this->disconnects_);
+  }
+#endif
 }
 
 void UartTcp::read_socket_() {

esphome/components/uart_tcp/uart_tcp.h

@@ -1,6 +1,9 @@
 #pragma once
 
 #include "esphome/components/binary_sensor/binary_sensor.h"
+#ifdef USE_SENSOR
+#include "esphome/components/sensor/sensor.h"
+#endif
 #include "esphome/components/socket/tcp_client_link.h"
 #ifdef USE_SOCKET_TCP_LISTENER
 #include "esphome/components/socket/tcp_listener.h"
@@ -20,6 +23,9 @@ class UartTcp final : public Component, public uart::UARTDevice {
   void set_port(uint16_t port) { this->link_.set_port(port); }
   void set_reconnect_interval(uint32_t ms) { this->link_.set_reconnect_interval(ms); }
   void set_connected_sensor(binary_sensor::BinarySensor *sensor) { this->connected_sensor_ = sensor; }
+#ifdef USE_SENSOR
+  void set_disconnects_sensor(sensor::Sensor *sensor) { this->disconnects_sensor_ = sensor; }
+#endif
 #ifdef USE_SOCKET_TCP_LISTENER
   void set_server(bool server) { this->server_ = server; }
 #ifdef USE_SOCKET_IPV4_ALLOW
@@ -50,6 +56,10 @@ class UartTcp final : public Component, public uart::UARTDevice {
   binary_sensor::BinarySensor *connected_sensor_{nullptr};
   // Loop start time of the last socket-to-UART write; sizes the next paced write.
   uint32_t last_write_ms_{0};
+#ifdef USE_SENSOR
+  sensor::Sensor *disconnects_sensor_{nullptr};
+  uint32_t disconnects_{0};
+#endif
   bool server_{false};
   // The link state loop() saw last; edges clear the buffer and publish the sensor.
   bool link_was_up_{false};

tests/components/tcp_uart/benchmark.yaml

@@ -0,0 +1,3 @@
+# The disconnects counter is compiled only with USE_SENSOR; declaring the
+# sensor domain makes the C++ unit test build define it.
+sensor:

tests/components/tcp_uart/common.yaml

@@ -12,6 +12,8 @@ tcp_uart:
       - 192.0.2.0/24
     connected:
       name: TCP UART Server Connected
+    disconnects:
+      name: TCP UART Server Disconnects
 
 interval:
   - interval: 60s

tests/components/tcp_uart/test-client.esp32-idf.yaml

@@ -9,3 +9,5 @@ tcp_uart:
     reconnect_interval: 10s
     connected:
       name: TCP UART Connected
+    disconnects:
+      name: TCP UART Disconnects

tests/components/tcp_uart/test_disconnect_host.cpp

@@ -0,0 +1,100 @@
+#include <gtest/gtest.h>
+
+#include <cmath>
+#include <csignal>
+#include <memory>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include "esphome/components/sensor/sensor.h"
+#include "esphome/components/tcp_uart/tcp_uart.h"
+
+#ifdef USE_HOST
+
+namespace esphome::tcp_uart::testing {
+
+class TcpUartDisconnectUnderTest : public TcpUart {
+ public:
+  TcpUartDisconnectUnderTest() {
+    this->set_host("peer");
+    this->set_port(1);
+    this->link_.begin("disconnect_test");
+  }
+  socket::TcpClientLink &link() { return this->link_; }
+};
+
+class TcpUartDisconnect : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    signal(SIGPIPE, SIG_IGN);
+    this->uart_.set_disconnects_sensor(&this->disconnects_);
+    this->connect_peer();
+  }
+  void TearDown() override {
+    this->close_peer();
+    this->uart_.link().close();
+  }
+  // Hands the UART a fresh session, as an accept or a reconnect would.
+  void connect_peer() {
+    int fds[2];
+    ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0);
+    this->peer_fd_ = fds[1];
+    this->uart_.link().adopt(std::make_unique<socket::Socket>(fds[0]));
+  }
+  void close_peer() {
+    if (this->peer_fd_ >= 0) {
+      ::close(this->peer_fd_);
+      this->peer_fd_ = -1;
+    }
+  }
+  void loops(int count) {
+    for (int i = 0; i < count; i++) {
+      this->uart_.loop();
+    }
+  }
+
+  TcpUartDisconnectUnderTest uart_;
+  sensor::Sensor disconnects_;
+  int peer_fd_{-1};
+};
+
+TEST_F(TcpUartDisconnect, CountsTheFallingEdgeOnce) {
+  this->uart_.loop();
+  EXPECT_TRUE(this->uart_.is_connected());
+  EXPECT_TRUE(std::isnan(this->disconnects_.state));
+
+  this->uart_.link().close();
+  this->uart_.loop();
+  EXPECT_FALSE(this->uart_.is_connected());
+  EXPECT_FLOAT_EQ(this->disconnects_.state, 1);
+
+  this->uart_.loop();
+  EXPECT_FLOAT_EQ(this->disconnects_.state, 1);
+}
+
+TEST_F(TcpUartDisconnect, CountsACloseByThePeer) {
+  this->loops(1);
+  this->close_peer();
+  // Read the EOF, then run the down edge, then idle.
+  this->loops(3);
+  EXPECT_FALSE(this->uart_.is_connected());
+  EXPECT_FLOAT_EQ(this->disconnects_.state, 1);
+}
+
+TEST_F(TcpUartDisconnect, CountsEverySessionAfterAReconnect) {
+  for (int session = 1; session <= 3; session++) {
+    if (session > 1) {
+      this->connect_peer();
+    }
+    this->loops(1);
+    ASSERT_TRUE(this->uart_.is_connected());
+    this->close_peer();
+    this->loops(3);
+    ASSERT_FALSE(this->uart_.is_connected());
+    EXPECT_FLOAT_EQ(this->disconnects_.state, session);
+  }
+}
+
+}  // namespace esphome::tcp_uart::testing
+
+#endif

tests/components/uart_tcp/benchmark.yaml

@@ -0,0 +1,3 @@
+# The disconnects counter is compiled only with USE_SENSOR; declaring the
+# sensor domain makes the C++ unit test build define it.
+sensor:

tests/components/uart_tcp/common.yaml

@@ -12,3 +12,5 @@ uart_tcp:
       - 192.168.1.0/24
     connected:
       name: UART TCP Connected
+    disconnects:
+      name: UART TCP Disconnects

tests/components/uart_tcp/test-client.esp32-idf.yaml

@@ -14,3 +14,5 @@ uart_tcp:
     reconnect_interval: 10s
     connected:
       name: UART TCP Connected
+    disconnects:
+      name: UART TCP Disconnects

tests/components/uart_tcp/test.host.yaml

@@ -10,3 +10,5 @@ uart_tcp:
     port: 44502
     connected:
       name: UART TCP Connected
+    disconnects:
+      name: UART TCP Disconnects

tests/components/uart_tcp/test_disconnect_host.cpp

@@ -0,0 +1,123 @@
+#include <gtest/gtest.h>
+
+#include <arpa/inet.h>
+#include <cmath>
+#include <csignal>
+#include <cstdint>
+#include <fcntl.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include "esphome/components/binary_sensor/binary_sensor.h"
+#include "esphome/components/sensor/sensor.h"
+#include "esphome/components/uart_tcp/uart_tcp.h"
+#include "esphome/core/application.h"
+#include "esphome/core/wake.h"
+
+#ifdef USE_HOST
+
+namespace esphome::uart_tcp::testing {
+
+// An idle UART: nothing to read, unknown free TX space.
+class IdleUart : public uart::UARTComponent {
+ public:
+  IdleUart() { this->set_baud_rate(9600); }
+  void write_array(const uint8_t *data, size_t len) override {}
+  bool peek_byte(uint8_t *data) override { return false; }
+  bool read_array(uint8_t *data, size_t len) override { return len == 0; }
+  size_t available() override { return 0; }
+  size_t available_for_write() override { return SIZE_MAX; }
+  uart::UARTFlushResult flush() override { return uart::UARTFlushResult::UART_FLUSH_RESULT_ASSUMED_SUCCESS; }
+  void check_logger_conflict() override {}
+};
+
+// Client role against a loopback listener the test owns.
+class UartTcpDisconnect : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    signal(SIGPIPE, SIG_IGN);
+    this->listen_fd_ = ::socket(AF_INET, SOCK_STREAM, 0);
+    ASSERT_GE(this->listen_fd_, 0);
+    struct sockaddr_in addr {};
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    ASSERT_EQ(::bind(this->listen_fd_, reinterpret_cast<struct sockaddr *>(&addr), sizeof(addr)), 0);
+    ASSERT_EQ(::listen(this->listen_fd_, 1), 0);
+    ASSERT_EQ(::fcntl(this->listen_fd_, F_SETFL, O_NONBLOCK), 0);
+    socklen_t len = sizeof(addr);
+    ASSERT_EQ(::getsockname(this->listen_fd_, reinterpret_cast<struct sockaddr *>(&addr), &len), 0);
+
+    this->bridge_.set_uart_parent(&this->uart_);
+    this->bridge_.set_host("127.0.0.1");
+    this->bridge_.set_port(ntohs(addr.sin_port));
+    // A zero interval lets a dropped link retry on the next pass.
+    this->bridge_.set_reconnect_interval(0);
+    this->bridge_.set_connected_sensor(&this->connected_);
+    this->bridge_.set_disconnects_sensor(&this->disconnects_);
+    this->tick();
+    this->bridge_.setup();
+  }
+  void TearDown() override {
+    this->bridge_.on_shutdown();
+    this->close_peer();
+    ::close(this->listen_fd_);
+  }
+
+  // Publish the next loop start time, as Application::loop() does.
+  void tick() {
+    this->now_ += 16;
+    LoopBlockingGuard dispatch{nullptr, nullptr, this->now_};
+  }
+  // One main loop pass: select() marks readable sockets, then the component runs.
+  void pass() {
+    internal::wakeable_delay(5);
+    this->tick();
+    this->bridge_.loop();
+  }
+  void connect() {
+    for (int i = 0; i < 50 && this->peer_fd_ < 0; i++) {
+      this->pass();
+      this->peer_fd_ = ::accept(this->listen_fd_, nullptr, nullptr);
+    }
+    ASSERT_GE(this->peer_fd_, 0);
+    for (int i = 0; i < 50 && !this->connected_.state; i++)
+      this->pass();
+    ASSERT_TRUE(this->connected_.state);
+  }
+  // The peer closes; pass until the bridge has run its down edge.
+  void peer_closes() {
+    this->close_peer();
+    for (int i = 0; i < 50 && this->connected_.state; i++)
+      this->pass();
+    ASSERT_FALSE(this->connected_.state);
+  }
+  void close_peer() {
+    if (this->peer_fd_ >= 0) {
+      ::close(this->peer_fd_);
+      this->peer_fd_ = -1;
+    }
+  }
+
+  IdleUart uart_;
+  UartTcp bridge_;
+  binary_sensor::BinarySensor connected_;
+  sensor::Sensor disconnects_;
+  int listen_fd_{-1};
+  int peer_fd_{-1};
+  uint32_t now_{0};
+};
+
+TEST_F(UartTcpDisconnect, StartsAtZero) { EXPECT_FLOAT_EQ(this->disconnects_.state, 0); }
+
+TEST_F(UartTcpDisconnect, CountsEveryCloseByThePeerAcrossReconnects) {
+  for (int session = 1; session <= 3; session++) {
+    this->connect();
+    this->peer_closes();
+    EXPECT_FLOAT_EQ(this->disconnects_.state, session);
+  }
+}
+
+}  // namespace esphome::uart_tcp::testing
+
+#endif