[emontx] Add pause/resume actions to let another component take over the UART (#18756)

Co-authored-by: Claude <noreply@anthropic.com>

esphome/components/emontx/__init__.py

@@ -157,3 +157,16 @@ automation.register_apply_action(
         CONF_COMMAND, "send_command", cg.std_string, const_fn=_plain_literal
     ),
 )
+
+
+# Let another component (e.g. serial_proxy) take exclusive ownership of the UART.
+
+EMONTX_ID_SCHEMA = automation.maybe_simple_id({cv.GenerateID(): cv.use_id(EmonTx)})
+
+automation.register_apply_action(
+    "emontx.pause", EMONTX_ID_SCHEMA, automation.ApplyCall("set_paused(true)")
+)
+automation.register_apply_action(
+    "emontx.resume", EMONTX_ID_SCHEMA, automation.ApplyCall("set_paused(false)")
+)
+automation.register_apply_condition("emontx.is_paused", EMONTX_ID_SCHEMA, "is_paused()")

esphome/components/emontx/emontx.cpp

@@ -16,6 +16,20 @@ void EmonTx::setup() { this->buffer_pos_ = 0; }
  * 2. If line starts with '{', parse as JSON and update sensors/callbacks
  */
 void EmonTx::loop() {
+  if (this->is_paused()) {
+    this->was_paused_ = true;
+    return;
+  }
+  if (this->was_paused_) {
+    // Drop what arrived during the pause and resync at the next line start
+    this->was_paused_ = false;
+    this->buffer_pos_ = 0;
+    this->discarding_line_ = true;
+    uint8_t dropped;
+    while (this->available() > 0 && this->read_byte(&dropped))
+      this->discarding_line_ = dropped != '\n';
+  }
+
   // Read all available data to prevent UART buffer overflow
   while (this->available() > 0) {
     uint8_t received;
@@ -45,7 +59,7 @@ void EmonTx::loop() {
         }
       }
     } else if (received == '\r' || this->discarding_line_) {
-      // Ignore CR, and the rest of a line that overflowed the buffer
+      // Ignore CR, and the rest of a line that overflowed or was interrupted by a pause
     } else if (this->buffer_pos_ >= MAX_LINE_LENGTH) {
       ESP_LOGW(TAG, "Buffer overflow (>%zu bytes), discarding line", MAX_LINE_LENGTH);
       this->buffer_pos_ = 0;
@@ -100,6 +114,10 @@ void EmonTx::dump_config() {
  * @param command The command string to send (LF will be appended automatically).
  */
 void EmonTx::send_command(const char *command) {
+  if (this->is_paused()) {
+    ESP_LOGW(TAG, "Not sending command, UART is paused: %s", command);
+    return;
+  }
   ESP_LOGD(TAG, "Sending command to emonTx: %s", command);
   this->write_str(command);
   this->write_byte('\n');

esphome/components/emontx/emontx.h

@@ -8,6 +8,7 @@
 #include "esphome/components/json/json_util.h"
 
 #include <array>
+#include <atomic>
 
 #ifdef USE_SENSOR
 #include "esphome/components/sensor/sensor.h"
@@ -41,6 +42,12 @@ class EmonTx final : public Component, public uart::UARTDevice {
   void send_command(const char *command);
   void send_command(const std::string &command) { this->send_command(command.c_str()); }
 
+  /// Stop reading the UART and sending commands, so another component can use it.
+  /// Atomic, so it can be called from any task. Hold the pause for at least one loop() pass.
+  void set_paused(bool paused) { this->paused_.store(paused, std::memory_order_relaxed); }
+
+  bool is_paused() const { return this->paused_.load(std::memory_order_relaxed); }
+
 #ifdef USE_SENSOR
   void init_sensors(size_t count) { this->sensors_.init(count); }
   void register_sensor(const char *tag_name, sensor::Sensor *sensor);
@@ -55,7 +62,9 @@ class EmonTx final : public Component, public uart::UARTDevice {
   LazyCallbackManager<void(JsonObject, StringRef)> json_callbacks_;
   LazyCallbackManager<void(StringRef)> data_callbacks_;
   uint16_t buffer_pos_{0};
-  bool discarding_line_{false};  // Set after an overflow until the end of that line
+  std::atomic<bool> paused_{false};
+  bool was_paused_{false};
+  bool discarding_line_{false};  // Set after an overflow or a pause until the end of that line
   std::array<char, MAX_LINE_LENGTH + 1> buffer_{};
 };
 

tests/components/emontx/common-actions.yaml

@@ -5,3 +5,21 @@ button:
       - emontx.send_command:
           id: test_emontx
           command: !lambda return "l";
+  - platform: template
+    name: Pause emontx
+    on_press:
+      - if:
+          condition:
+            emontx.is_paused: test_emontx
+          then:
+            - emontx.resume: test_emontx
+          else:
+            - emontx.pause: test_emontx
+  - platform: template
+    name: Pause emontx explicit id
+    on_press:
+      - emontx.pause:
+          id: test_emontx
+      - delay: 1s
+      - emontx.resume:
+          id: test_emontx

tests/components/emontx/emontx_test.cpp

@@ -1,5 +1,6 @@
 #include <gtest/gtest.h>
 
+#include <cstring>
 #include <string>
 #include <vector>
 
@@ -91,4 +92,70 @@ TEST_F(EmonTxTest, OverflowAcrossLoopCalls) {
   EXPECT_EQ(this->lines_[0], "MSG:3");
 }
 
+TEST_F(EmonTxTest, LoopLeavesBytesUntouchedWhenPaused) {
+  this->uart_.push_rx("{\"P1\":123}\n");
+  this->sut_.set_paused(true);
+  this->sut_.loop();
+
+  EXPECT_EQ(this->uart_.available(), std::strlen("{\"P1\":123}\n"));
+  EXPECT_TRUE(this->lines_.empty());
+}
+
+TEST_F(EmonTxTest, SendCommandIsSuppressedWhilePaused) {
+  this->sut_.set_paused(true);
+  this->sut_.send_command("x");
+  EXPECT_TRUE(this->uart_.tx.empty());
+
+  this->sut_.set_paused(false);
+  this->sut_.send_command("x");
+  EXPECT_EQ(std::string(this->uart_.tx.begin(), this->uart_.tx.end()), "x\n");
+}
+
+// Neither the part before the pause nor the part after it may reach on_data
+TEST_F(EmonTxTest, ResumeDiscardsInterruptedLine) {
+  this->uart_.push_rx("{\"P1\":4");
+  this->sut_.loop();
+  this->sut_.set_paused(true);
+  this->sut_.loop();
+  this->sut_.set_paused(false);
+
+  this->uart_.push_rx("56}\n");
+  this->sut_.loop();
+  EXPECT_TRUE(this->lines_.empty());
+
+  this->uart_.push_rx("{\"P1\":789}\n");
+  this->sut_.loop();
+  ASSERT_EQ(this->lines_.size(), 1u);
+  EXPECT_EQ(this->lines_[0], "{\"P1\":789}");
+}
+
+TEST_F(EmonTxTest, ResumeDropsBacklogReceivedWhilePaused) {
+  this->sut_.set_paused(true);
+  this->sut_.loop();
+  this->uart_.push_rx("{\"P1\":1}\n{\"P1\":2}\n{\"P1\":");
+  this->sut_.set_paused(false);
+
+  this->sut_.loop();
+  EXPECT_EQ(this->uart_.available(), 0u);
+  EXPECT_TRUE(this->lines_.empty());
+
+  this->uart_.push_rx("3}\n{\"P1\":4}\n");
+  this->sut_.loop();
+  ASSERT_EQ(this->lines_.size(), 1u);
+  EXPECT_EQ(this->lines_[0], "{\"P1\":4}");
+}
+
+TEST_F(EmonTxTest, ResumeAfterCompleteBacklogKeepsNextLine) {
+  this->sut_.set_paused(true);
+  this->sut_.loop();
+  this->uart_.push_rx("{\"P1\":1}\n");
+  this->sut_.set_paused(false);
+
+  this->sut_.loop();
+  this->uart_.push_rx("{\"P1\":2}\n");
+  this->sut_.loop();
+  ASSERT_EQ(this->lines_.size(), 1u);
+  EXPECT_EQ(this->lines_[0], "{\"P1\":2}");
+}
+
 }  // namespace esphome::emontx::testing