Merge branch 'rf_bridge_batch_read' into integration_batch_read

Co-authored-by: J. Nick Koston <nick@koston.org>
Co-authored-by: J. Nick Koston <nick+github@koston.org>

esphome/components/cse7766/cse7766.cpp

@@ -7,7 +7,6 @@ namespace esphome {
 namespace cse7766 {
 
 static const char *const TAG = "cse7766";
-static constexpr size_t CSE7766_RAW_DATA_SIZE = 24;
 
 void CSE7766Component::loop() {
   const uint32_t now = App.get_loop_component_start_time();
@@ -16,25 +15,39 @@ void CSE7766Component::loop() {
     this->raw_data_index_ = 0;
   }
 
-  if (this->available() == 0) {
+  // Early return prevents updating last_transmission_ when no data is available.
+  int avail = this->available();
+  if (avail <= 0) {
     return;
   }
 
   this->last_transmission_ = now;
-  while (this->available() != 0) {
-    this->read_byte(&this->raw_data_[this->raw_data_index_]);
-    if (!this->check_byte_()) {
-      this->raw_data_index_ = 0;
-      this->status_set_warning();
-      continue;
-    }
 
-    if (this->raw_data_index_ == 23) {
-      this->parse_data_();
-      this->status_clear_warning();
+  // Read all available bytes in batches to reduce UART call overhead.
+  // At 4800 baud (~480 bytes/sec) with ~122 Hz loop rate, typically ~4 bytes per call.
+  uint8_t buf[CSE7766_RAW_DATA_SIZE];
+  while (avail > 0) {
+    size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+    if (!this->read_array(buf, to_read)) {
+      break;
     }
+    avail -= to_read;
 
-    this->raw_data_index_ = (this->raw_data_index_ + 1) % 24;
+    for (size_t i = 0; i < to_read; i++) {
+      this->raw_data_[this->raw_data_index_] = buf[i];
+      if (!this->check_byte_()) {
+        this->raw_data_index_ = 0;
+        this->status_set_warning();
+        continue;
+      }
+
+      if (this->raw_data_index_ == CSE7766_RAW_DATA_SIZE - 1) {
+        this->parse_data_();
+        this->status_clear_warning();
+      }
+
+      this->raw_data_index_ = (this->raw_data_index_ + 1) % CSE7766_RAW_DATA_SIZE;
+    }
   }
 }
 
@@ -53,14 +66,15 @@ bool CSE7766Component::check_byte_() {
     return true;
   }
 
-  if (index == 23) {
+  if (index == CSE7766_RAW_DATA_SIZE - 1) {
     uint8_t checksum = 0;
-    for (uint8_t i = 2; i < 23; i++) {
+    for (uint8_t i = 2; i < CSE7766_RAW_DATA_SIZE - 1; i++) {
       checksum += this->raw_data_[i];
     }
 
-    if (checksum != this->raw_data_[23]) {
-      ESP_LOGW(TAG, "Invalid checksum from CSE7766: 0x%02X != 0x%02X", checksum, this->raw_data_[23]);
+    if (checksum != this->raw_data_[CSE7766_RAW_DATA_SIZE - 1]) {
+      ESP_LOGW(TAG, "Invalid checksum from CSE7766: 0x%02X != 0x%02X", checksum,
+               this->raw_data_[CSE7766_RAW_DATA_SIZE - 1]);
       return false;
     }
     return true;

esphome/components/cse7766/cse7766.h

@@ -8,6 +8,8 @@
 namespace esphome {
 namespace cse7766 {
 
+static constexpr size_t CSE7766_RAW_DATA_SIZE = 24;
+
 class CSE7766Component : public Component, public uart::UARTDevice {
  public:
   void set_voltage_sensor(sensor::Sensor *voltage_sensor) { voltage_sensor_ = voltage_sensor; }
@@ -33,7 +35,7 @@ class CSE7766Component : public Component, public uart::UARTDevice {
                          this->raw_data_[start_index + 2]);
   }
 
-  uint8_t raw_data_[24];
+  uint8_t raw_data_[CSE7766_RAW_DATA_SIZE];
   uint8_t raw_data_index_{0};
   uint32_t last_transmission_{0};
   sensor::Sensor *voltage_sensor_{nullptr};

esphome/components/dfplayer/dfplayer.cpp

@@ -1,4 +1,5 @@
 #include "dfplayer.h"
+#include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
 namespace esphome {
@@ -131,140 +132,149 @@ void DFPlayer::send_cmd_(uint8_t cmd, uint16_t argument) {
 }
 
 void DFPlayer::loop() {
-  // Read message
-  while (this->available()) {
-    uint8_t byte;
-    this->read_byte(&byte);
-
-    if (this->read_pos_ == DFPLAYER_READ_BUFFER_LENGTH)
-      this->read_pos_ = 0;
-
-    switch (this->read_pos_) {
-      case 0:  // Start mark
-        if (byte != 0x7E)
-          continue;
-        break;
-      case 1:  // Version
-        if (byte != 0xFF) {
-          ESP_LOGW(TAG, "Expected Version 0xFF, got %#02x", byte);
-          this->read_pos_ = 0;
-          continue;
-        }
-        break;
-      case 2:  // Buffer length
-        if (byte != 0x06) {
-          ESP_LOGW(TAG, "Expected Buffer length 0x06, got %#02x", byte);
-          this->read_pos_ = 0;
-          continue;
-        }
-        break;
-      case 9:  // End byte
-#ifdef ESPHOME_LOG_HAS_VERY_VERBOSE
-        char byte_sequence[100];
-        byte_sequence[0] = '\0';
-        for (size_t i = 0; i < this->read_pos_ + 1; ++i) {
-          snprintf(byte_sequence + strlen(byte_sequence), sizeof(byte_sequence) - strlen(byte_sequence), "%02X ",
-                   this->read_buffer_[i]);
-        }
-        ESP_LOGVV(TAG, "Received byte sequence: %s", byte_sequence);
-#endif
-        if (byte != 0xEF) {
-          ESP_LOGW(TAG, "Expected end byte 0xEF, got %#02x", byte);
-          this->read_pos_ = 0;
-          continue;
-        }
-        // Parse valid received command
-        uint8_t cmd = this->read_buffer_[3];
-        uint16_t argument = (this->read_buffer_[5] << 8) | this->read_buffer_[6];
-
-        ESP_LOGV(TAG, "Received message cmd: %#02x arg %#04x", cmd, argument);
-
-        switch (cmd) {
-          case 0x3A:
-            if (argument == 1) {
-              ESP_LOGI(TAG, "USB loaded");
-            } else if (argument == 2) {
-              ESP_LOGI(TAG, "TF Card loaded");
-            }
-            break;
-          case 0x3B:
-            if (argument == 1) {
-              ESP_LOGI(TAG, "USB unloaded");
-            } else if (argument == 2) {
-              ESP_LOGI(TAG, "TF Card unloaded");
-            }
-            break;
-          case 0x3F:
-            if (argument == 1) {
-              ESP_LOGI(TAG, "USB available");
-            } else if (argument == 2) {
-              ESP_LOGI(TAG, "TF Card available");
-            } else if (argument == 3) {
-              ESP_LOGI(TAG, "USB, TF Card available");
-            }
-            break;
-          case 0x40:
-            ESP_LOGV(TAG, "Nack");
-            this->ack_set_is_playing_ = false;
-            this->ack_reset_is_playing_ = false;
-            switch (argument) {
-              case 0x01:
-                ESP_LOGE(TAG, "Module is busy or uninitialized");
-                break;
-              case 0x02:
-                ESP_LOGE(TAG, "Module is in sleep mode");
-                break;
-              case 0x03:
-                ESP_LOGE(TAG, "Serial receive error");
-                break;
-              case 0x04:
-                ESP_LOGE(TAG, "Checksum incorrect");
-                break;
-              case 0x05:
-                ESP_LOGE(TAG, "Specified track is out of current track scope");
-                this->is_playing_ = false;
-                break;
-              case 0x06:
-                ESP_LOGE(TAG, "Specified track is not found");
-                this->is_playing_ = false;
-                break;
-              case 0x07:
-                ESP_LOGE(TAG, "Insertion error (an inserting operation only can be done when a track is being played)");
-                break;
-              case 0x08:
-                ESP_LOGE(TAG, "SD card reading failed (SD card pulled out or damaged)");
-                break;
-              case 0x09:
-                ESP_LOGE(TAG, "Entered into sleep mode");
-                this->is_playing_ = false;
-                break;
-            }
-            break;
-          case 0x41:
-            ESP_LOGV(TAG, "Ack ok");
-            this->is_playing_ |= this->ack_set_is_playing_;
-            this->is_playing_ &= !this->ack_reset_is_playing_;
-            this->ack_set_is_playing_ = false;
-            this->ack_reset_is_playing_ = false;
-            break;
-          case 0x3C:
-            ESP_LOGV(TAG, "Playback finished (USB drive)");
-            this->is_playing_ = false;
-            this->on_finished_playback_callback_.call();
-          case 0x3D:
-            ESP_LOGV(TAG, "Playback finished (SD card)");
-            this->is_playing_ = false;
-            this->on_finished_playback_callback_.call();
-            break;
-          default:
-            ESP_LOGE(TAG, "Received unknown cmd %#02x arg %#04x", cmd, argument);
-        }
-        this->sent_cmd_ = 0;
-        this->read_pos_ = 0;
-        continue;
+  // Read all available bytes in batches to reduce UART call overhead.
+  int avail = this->available();
+  uint8_t buf[64];
+  while (avail > 0) {
+    size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+    if (!this->read_array(buf, to_read)) {
+      break;
+    }
+    avail -= to_read;
+    for (size_t bi = 0; bi < to_read; bi++) {
+      uint8_t byte = buf[bi];
+
+      if (this->read_pos_ == DFPLAYER_READ_BUFFER_LENGTH)
+        this->read_pos_ = 0;
+
+      switch (this->read_pos_) {
+        case 0:  // Start mark
+          if (byte != 0x7E)
+            continue;
+          break;
+        case 1:  // Version
+          if (byte != 0xFF) {
+            ESP_LOGW(TAG, "Expected Version 0xFF, got %#02x", byte);
+            this->read_pos_ = 0;
+            continue;
+          }
+          break;
+        case 2:  // Buffer length
+          if (byte != 0x06) {
+            ESP_LOGW(TAG, "Expected Buffer length 0x06, got %#02x", byte);
+            this->read_pos_ = 0;
+            continue;
+          }
+          break;
+        case 9:  // End byte
+#ifdef ESPHOME_LOG_HAS_VERY_VERBOSE
+          char byte_sequence[100];
+          byte_sequence[0] = '\0';
+          for (size_t i = 0; i < this->read_pos_ + 1; ++i) {
+            snprintf(byte_sequence + strlen(byte_sequence), sizeof(byte_sequence) - strlen(byte_sequence), "%02X ",
+                     this->read_buffer_[i]);
+          }
+          ESP_LOGVV(TAG, "Received byte sequence: %s", byte_sequence);
+#endif
+          if (byte != 0xEF) {
+            ESP_LOGW(TAG, "Expected end byte 0xEF, got %#02x", byte);
+            this->read_pos_ = 0;
+            continue;
+          }
+          // Parse valid received command
+          uint8_t cmd = this->read_buffer_[3];
+          uint16_t argument = (this->read_buffer_[5] << 8) | this->read_buffer_[6];
+
+          ESP_LOGV(TAG, "Received message cmd: %#02x arg %#04x", cmd, argument);
+
+          switch (cmd) {
+            case 0x3A:
+              if (argument == 1) {
+                ESP_LOGI(TAG, "USB loaded");
+              } else if (argument == 2) {
+                ESP_LOGI(TAG, "TF Card loaded");
+              }
+              break;
+            case 0x3B:
+              if (argument == 1) {
+                ESP_LOGI(TAG, "USB unloaded");
+              } else if (argument == 2) {
+                ESP_LOGI(TAG, "TF Card unloaded");
+              }
+              break;
+            case 0x3F:
+              if (argument == 1) {
+                ESP_LOGI(TAG, "USB available");
+              } else if (argument == 2) {
+                ESP_LOGI(TAG, "TF Card available");
+              } else if (argument == 3) {
+                ESP_LOGI(TAG, "USB, TF Card available");
+              }
+              break;
+            case 0x40:
+              ESP_LOGV(TAG, "Nack");
+              this->ack_set_is_playing_ = false;
+              this->ack_reset_is_playing_ = false;
+              switch (argument) {
+                case 0x01:
+                  ESP_LOGE(TAG, "Module is busy or uninitialized");
+                  break;
+                case 0x02:
+                  ESP_LOGE(TAG, "Module is in sleep mode");
+                  break;
+                case 0x03:
+                  ESP_LOGE(TAG, "Serial receive error");
+                  break;
+                case 0x04:
+                  ESP_LOGE(TAG, "Checksum incorrect");
+                  break;
+                case 0x05:
+                  ESP_LOGE(TAG, "Specified track is out of current track scope");
+                  this->is_playing_ = false;
+                  break;
+                case 0x06:
+                  ESP_LOGE(TAG, "Specified track is not found");
+                  this->is_playing_ = false;
+                  break;
+                case 0x07:
+                  ESP_LOGE(TAG,
+                           "Insertion error (an inserting operation only can be done when a track is being played)");
+                  break;
+                case 0x08:
+                  ESP_LOGE(TAG, "SD card reading failed (SD card pulled out or damaged)");
+                  break;
+                case 0x09:
+                  ESP_LOGE(TAG, "Entered into sleep mode");
+                  this->is_playing_ = false;
+                  break;
+              }
+              break;
+            case 0x41:
+              ESP_LOGV(TAG, "Ack ok");
+              this->is_playing_ |= this->ack_set_is_playing_;
+              this->is_playing_ &= !this->ack_reset_is_playing_;
+              this->ack_set_is_playing_ = false;
+              this->ack_reset_is_playing_ = false;
+              break;
+            case 0x3C:
+              ESP_LOGV(TAG, "Playback finished (USB drive)");
+              this->is_playing_ = false;
+              this->on_finished_playback_callback_.call();
+            case 0x3D:
+              ESP_LOGV(TAG, "Playback finished (SD card)");
+              this->is_playing_ = false;
+              this->on_finished_playback_callback_.call();
+              break;
+            default:
+              ESP_LOGE(TAG, "Received unknown cmd %#02x arg %#04x", cmd, argument);
+          }
+          this->sent_cmd_ = 0;
+          this->read_pos_ = 0;
+          continue;
+      }
+      this->read_buffer_[this->read_pos_] = byte;
+      this->read_pos_++;
     }
-    this->read_buffer_[this->read_pos_] = byte;
-    this->read_pos_++;
   }
 }
 void DFPlayer::dump_config() {

esphome/components/dlms_meter/dlms_meter.cpp

@@ -28,15 +28,28 @@ void DlmsMeterComponent::dump_config() {
 
 void DlmsMeterComponent::loop() {
   // Read while data is available, netznoe uses two frames so allow 2x max frame length
-  while (this->available()) {
-    if (this->receive_buffer_.size() >= MBUS_MAX_FRAME_LENGTH * 2) {
+  int avail = this->available();
+  if (avail > 0) {
+    size_t remaining = MBUS_MAX_FRAME_LENGTH * 2 - this->receive_buffer_.size();
+    if (remaining == 0) {
       ESP_LOGW(TAG, "Receive buffer full, dropping remaining bytes");
-      break;
+    } else {
+      // Read all available bytes in batches to reduce UART call overhead.
+      // Cap reads to remaining buffer capacity.
+      if (static_cast<size_t>(avail) > remaining) {
+        avail = remaining;
+      }
+      uint8_t buf[64];
+      while (avail > 0) {
+        size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+        if (!this->read_array(buf, to_read)) {
+          break;
+        }
+        avail -= to_read;
+        this->receive_buffer_.insert(this->receive_buffer_.end(), buf, buf + to_read);
+        this->last_read_ = millis();
+      }
     }
-    uint8_t c;
-    this->read_byte(&c);
-    this->receive_buffer_.push_back(c);
-    this->last_read_ = millis();
   }
 
   if (!this->receive_buffer_.empty() && millis() - this->last_read_ > this->read_timeout_) {

esphome/components/dsmr/dsmr.cpp

@@ -40,9 +40,7 @@ bool Dsmr::ready_to_request_data_() {
       this->start_requesting_data_();
     }
     if (!this->requesting_data_) {
-      while (this->available()) {
-        this->read();
-      }
+      this->drain_rx_buffer_();
     }
   }
   return this->requesting_data_;
@@ -115,13 +113,21 @@ void Dsmr::stop_requesting_data_() {
     } else {
       ESP_LOGV(TAG, "Stop reading data from P1 port");
     }
-    while (this->available()) {
-      this->read();
-    }
+    this->drain_rx_buffer_();
     this->requesting_data_ = false;
   }
 }
 
+void Dsmr::drain_rx_buffer_() {
+  uint8_t buf[64];
+  int avail;
+  while ((avail = this->available()) > 0) {
+    if (!this->read_array(buf, std::min(static_cast<size_t>(avail), sizeof(buf)))) {
+      break;
+    }
+  }
+}
+
 void Dsmr::reset_telegram_() {
   this->header_found_ = false;
   this->footer_found_ = false;
@@ -133,120 +139,144 @@ void Dsmr::reset_telegram_() {
 
 void Dsmr::receive_telegram_() {
   while (this->available_within_timeout_()) {
-    const char c = this->read();
+    // Read all available bytes in batches to reduce UART call overhead.
+    uint8_t buf[64];
+    int avail = this->available();
+    while (avail > 0) {
+      size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+      if (!this->read_array(buf, to_read))
+        return;
+      avail -= to_read;
 
-    // Find a new telegram header, i.e. forward slash.
-    if (c == '/') {
-      ESP_LOGV(TAG, "Header of telegram found");
-      this->reset_telegram_();
-      this->header_found_ = true;
-    }
-    if (!this->header_found_)
-      continue;
+      for (size_t i = 0; i < to_read; i++) {
+        const char c = static_cast<char>(buf[i]);
 
-    // Check for buffer overflow.
-    if (this->bytes_read_ >= this->max_telegram_len_) {
-      this->reset_telegram_();
-      ESP_LOGE(TAG, "Error: telegram larger than buffer (%d bytes)", this->max_telegram_len_);
-      return;
-    }
+        // Find a new telegram header, i.e. forward slash.
+        if (c == '/') {
+          ESP_LOGV(TAG, "Header of telegram found");
+          this->reset_telegram_();
+          this->header_found_ = true;
+        }
+        if (!this->header_found_)
+          continue;
 
-    // Some v2.2 or v3 meters will send a new value which starts with '('
-    // in a new line, while the value belongs to the previous ObisId. For
-    // proper parsing, remove these new line characters.
-    if (c == '(') {
-      while (true) {
-        auto previous_char = this->telegram_[this->bytes_read_ - 1];
-        if (previous_char == '\n' || previous_char == '\r') {
-          this->bytes_read_--;
-        } else {
-          break;
+        // Check for buffer overflow.
+        if (this->bytes_read_ >= this->max_telegram_len_) {
+          this->reset_telegram_();
+          ESP_LOGE(TAG, "Error: telegram larger than buffer (%d bytes)", this->max_telegram_len_);
+          return;
+        }
+
+        // Some v2.2 or v3 meters will send a new value which starts with '('
+        // in a new line, while the value belongs to the previous ObisId. For
+        // proper parsing, remove these new line characters.
+        if (c == '(') {
+          while (true) {
+            auto previous_char = this->telegram_[this->bytes_read_ - 1];
+            if (previous_char == '\n' || previous_char == '\r') {
+              this->bytes_read_--;
+            } else {
+              break;
+            }
+          }
+        }
+
+        // Store the byte in the buffer.
+        this->telegram_[this->bytes_read_] = c;
+        this->bytes_read_++;
+
+        // Check for a footer, i.e. exclamation mark, followed by a hex checksum.
+        if (c == '!') {
+          ESP_LOGV(TAG, "Footer of telegram found");
+          this->footer_found_ = true;
+          continue;
+        }
+        // Check for the end of the hex checksum, i.e. a newline.
+        if (this->footer_found_ && c == '\n') {
+          // Parse the telegram and publish sensor values.
+          this->parse_telegram();
+          this->reset_telegram_();
+          return;
         }
       }
     }
-
-    // Store the byte in the buffer.
-    this->telegram_[this->bytes_read_] = c;
-    this->bytes_read_++;
-
-    // Check for a footer, i.e. exclamation mark, followed by a hex checksum.
-    if (c == '!') {
-      ESP_LOGV(TAG, "Footer of telegram found");
-      this->footer_found_ = true;
-      continue;
-    }
-    // Check for the end of the hex checksum, i.e. a newline.
-    if (this->footer_found_ && c == '\n') {
-      // Parse the telegram and publish sensor values.
-      this->parse_telegram();
-      this->reset_telegram_();
-      return;
-    }
   }
 }
 
 void Dsmr::receive_encrypted_telegram_() {
   while (this->available_within_timeout_()) {
-    const char c = this->read();
+    // Read all available bytes in batches to reduce UART call overhead.
+    uint8_t buf[64];
+    int avail = this->available();
+    while (avail > 0) {
+      size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+      if (!this->read_array(buf, to_read))
+        return;
+      avail -= to_read;
 
-    // Find a new telegram start byte.
-    if (!this->header_found_) {
-      if ((uint8_t) c != 0xDB) {
-        continue;
+      for (size_t i = 0; i < to_read; i++) {
+        const char c = static_cast<char>(buf[i]);
+
+        // Find a new telegram start byte.
+        if (!this->header_found_) {
+          if ((uint8_t) c != 0xDB) {
+            continue;
+          }
+          ESP_LOGV(TAG, "Start byte 0xDB of encrypted telegram found");
+          this->reset_telegram_();
+          this->header_found_ = true;
+        }
+
+        // Check for buffer overflow.
+        if (this->crypt_bytes_read_ >= this->max_telegram_len_) {
+          this->reset_telegram_();
+          ESP_LOGE(TAG, "Error: encrypted telegram larger than buffer (%d bytes)", this->max_telegram_len_);
+          return;
+        }
+
+        // Store the byte in the buffer.
+        this->crypt_telegram_[this->crypt_bytes_read_] = c;
+        this->crypt_bytes_read_++;
+
+        // Read the length of the incoming encrypted telegram.
+        if (this->crypt_telegram_len_ == 0 && this->crypt_bytes_read_ > 20) {
+          // Complete header + data bytes
+          this->crypt_telegram_len_ = 13 + (this->crypt_telegram_[11] << 8 | this->crypt_telegram_[12]);
+          ESP_LOGV(TAG, "Encrypted telegram length: %d bytes", this->crypt_telegram_len_);
+        }
+
+        // Check for the end of the encrypted telegram.
+        if (this->crypt_telegram_len_ == 0 || this->crypt_bytes_read_ != this->crypt_telegram_len_) {
+          continue;
+        }
+        ESP_LOGV(TAG, "End of encrypted telegram found");
+
+        // Decrypt the encrypted telegram.
+        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++)
+          this->crypt_telegram_[i] = this->crypt_telegram_[i + 4];
+        constexpr uint16_t iv_size{12};
+        gcmaes128->setIV(&this->crypt_telegram_[2], iv_size);
+        gcmaes128->decrypt(reinterpret_cast<uint8_t *>(this->telegram_),
+                           // the ciphertext start at byte 18
+                           &this->crypt_telegram_[18],
+                           // cipher size
+                           this->crypt_bytes_read_ - 17);
+        delete gcmaes128;  // NOLINT(cppcoreguidelines-owning-memory)
+
+        this->bytes_read_ = strnlen(this->telegram_, this->max_telegram_len_);
+        ESP_LOGV(TAG, "Decrypted telegram size: %d bytes", this->bytes_read_);
+        ESP_LOGVV(TAG, "Decrypted telegram: %s", this->telegram_);
+
+        // Parse the decrypted telegram and publish sensor values.
+        this->parse_telegram();
+        this->reset_telegram_();
+        return;
       }
-      ESP_LOGV(TAG, "Start byte 0xDB of encrypted telegram found");
-      this->reset_telegram_();
-      this->header_found_ = true;
     }
-
-    // Check for buffer overflow.
-    if (this->crypt_bytes_read_ >= this->max_telegram_len_) {
-      this->reset_telegram_();
-      ESP_LOGE(TAG, "Error: encrypted telegram larger than buffer (%d bytes)", this->max_telegram_len_);
-      return;
-    }
-
-    // Store the byte in the buffer.
-    this->crypt_telegram_[this->crypt_bytes_read_] = c;
-    this->crypt_bytes_read_++;
-
-    // Read the length of the incoming encrypted telegram.
-    if (this->crypt_telegram_len_ == 0 && this->crypt_bytes_read_ > 20) {
-      // Complete header + data bytes
-      this->crypt_telegram_len_ = 13 + (this->crypt_telegram_[11] << 8 | this->crypt_telegram_[12]);
-      ESP_LOGV(TAG, "Encrypted telegram length: %d bytes", this->crypt_telegram_len_);
-    }
-
-    // Check for the end of the encrypted telegram.
-    if (this->crypt_telegram_len_ == 0 || this->crypt_bytes_read_ != this->crypt_telegram_len_) {
-      continue;
-    }
-    ESP_LOGV(TAG, "End of encrypted telegram found");
-
-    // Decrypt the encrypted telegram.
-    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++)
-      this->crypt_telegram_[i] = this->crypt_telegram_[i + 4];
-    constexpr uint16_t iv_size{12};
-    gcmaes128->setIV(&this->crypt_telegram_[2], iv_size);
-    gcmaes128->decrypt(reinterpret_cast<uint8_t *>(this->telegram_),
-                       // the ciphertext start at byte 18
-                       &this->crypt_telegram_[18],
-                       // cipher size
-                       this->crypt_bytes_read_ - 17);
-    delete gcmaes128;  // NOLINT(cppcoreguidelines-owning-memory)
-
-    this->bytes_read_ = strnlen(this->telegram_, this->max_telegram_len_);
-    ESP_LOGV(TAG, "Decrypted telegram size: %d bytes", this->bytes_read_);
-    ESP_LOGVV(TAG, "Decrypted telegram: %s", this->telegram_);
-
-    // Parse the decrypted telegram and publish sensor values.
-    this->parse_telegram();
-    this->reset_telegram_();
-    return;
   }
 }
 

esphome/components/dsmr/dsmr.h

@@ -85,6 +85,7 @@ class Dsmr : public Component, public uart::UARTDevice {
   void receive_telegram_();
   void receive_encrypted_telegram_();
   void reset_telegram_();
+  void drain_rx_buffer_();
 
   /// Wait for UART data to become available within the read timeout.
   ///

esphome/components/ld2410/ld2410.cpp

@@ -275,8 +275,19 @@ void LD2410Component::restart_and_read_all_info() {
 }
 
 void LD2410Component::loop() {
-  while (this->available()) {
-    this->readline_(this->read());
+  // Read all available bytes in batches to reduce UART call overhead.
+  int avail = this->available();
+  uint8_t buf[MAX_LINE_LENGTH];
+  while (avail > 0) {
+    size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+    if (!this->read_array(buf, to_read)) {
+      break;
+    }
+    avail -= to_read;
+
+    for (size_t i = 0; i < to_read; i++) {
+      this->readline_(buf[i]);
+    }
   }
 }
 

esphome/components/ld2412/ld2412.cpp

@@ -310,8 +310,19 @@ void LD2412Component::restart_and_read_all_info() {
 }
 
 void LD2412Component::loop() {
-  while (this->available()) {
-    this->readline_(this->read());
+  // Read all available bytes in batches to reduce UART call overhead.
+  int avail = this->available();
+  uint8_t buf[MAX_LINE_LENGTH];
+  while (avail > 0) {
+    size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+    if (!this->read_array(buf, to_read)) {
+      break;
+    }
+    avail -= to_read;
+
+    for (size_t i = 0; i < to_read; i++) {
+      this->readline_(buf[i]);
+    }
   }
 }
 

esphome/components/ld2420/ld2420.cpp

@@ -335,9 +335,10 @@ void LD2420Component::revert_config_action() {
 
 void LD2420Component::loop() {
   // If there is a active send command do not process it here, the send command call will handle it.
-  while (!this->cmd_active_ && this->available()) {
-    this->readline_(this->read(), this->buffer_data_, MAX_LINE_LENGTH);
+  if (this->cmd_active_) {
+    return;
   }
+  this->read_batch_(this->buffer_data_);
 }
 
 void LD2420Component::update_radar_data(uint16_t const *gate_energy, uint8_t sample_number) {
@@ -539,6 +540,23 @@ void LD2420Component::handle_simple_mode_(const uint8_t *inbuf, int len) {
   }
 }
 
+void LD2420Component::read_batch_(std::span<uint8_t, MAX_LINE_LENGTH> buffer) {
+  // Read all available bytes in batches to reduce UART call overhead.
+  int avail = this->available();
+  uint8_t buf[MAX_LINE_LENGTH];
+  while (avail > 0) {
+    size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+    if (!this->read_array(buf, to_read)) {
+      break;
+    }
+    avail -= to_read;
+
+    for (size_t i = 0; i < to_read; i++) {
+      this->readline_(buf[i], buffer.data(), buffer.size());
+    }
+  }
+}
+
 void LD2420Component::handle_ack_data_(uint8_t *buffer, int len) {
   this->cmd_reply_.command = buffer[CMD_FRAME_COMMAND];
   this->cmd_reply_.length = buffer[CMD_FRAME_DATA_LENGTH];

esphome/components/ld2420/ld2420.h

@@ -4,6 +4,7 @@
 #include "esphome/components/uart/uart.h"
 #include "esphome/core/automation.h"
 #include "esphome/core/helpers.h"
+#include <span>
 #ifdef USE_TEXT_SENSOR
 #include "esphome/components/text_sensor/text_sensor.h"
 #endif
@@ -165,6 +166,7 @@ class LD2420Component : public Component, public uart::UARTDevice {
   void handle_energy_mode_(uint8_t *buffer, int len);
   void handle_ack_data_(uint8_t *buffer, int len);
   void readline_(int rx_data, uint8_t *buffer, int len);
+  void read_batch_(std::span<uint8_t, MAX_LINE_LENGTH> buffer);
   void set_calibration_(bool state) { this->calibration_ = state; };
   bool get_calibration_() { return this->calibration_; };
 

esphome/components/ld2450/ld2450.cpp

@@ -276,8 +276,19 @@ void LD2450Component::dump_config() {
 }
 
 void LD2450Component::loop() {
-  while (this->available()) {
-    this->readline_(this->read());
+  // Read all available bytes in batches to reduce UART call overhead.
+  int avail = this->available();
+  uint8_t buf[MAX_LINE_LENGTH];
+  while (avail > 0) {
+    size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+    if (!this->read_array(buf, to_read)) {
+      break;
+    }
+    avail -= to_read;
+
+    for (size_t i = 0; i < to_read; i++) {
+      this->readline_(buf[i]);
+    }
   }
 }
 

esphome/components/mixer/speaker/__init__.py

@@ -1,6 +1,6 @@
 from esphome import automation
 import esphome.codegen as cg
-from esphome.components import audio, esp32, speaker
+from esphome.components import audio, esp32, socket, speaker
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_BITS_PER_SAMPLE,
@@ -61,7 +61,7 @@ def _set_stream_limits(config):
 def _validate_source_speaker(config):
     fconf = fv.full_config.get()
 
-    # Get ID for the output speaker and add it to the source speakrs config to easily inherit properties
+    # Get ID for the output speaker and add it to the source speakers config to easily inherit properties
     path = fconf.get_path_for_id(config[CONF_ID])[:-3]
     path.append(CONF_OUTPUT_SPEAKER)
     output_speaker_id = fconf.get_config_for_path(path)
@@ -111,6 +111,9 @@ FINAL_VALIDATE_SCHEMA = cv.All(
 
 
 async def to_code(config):
+    # Enable wake_loop_threadsafe for immediate command processing from other tasks
+    socket.require_wake_loop_threadsafe()
+
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
 
@@ -127,6 +130,9 @@ async def to_code(config):
                 "CONFIG_SPIRAM_ALLOW_STACK_EXTERNAL_MEMORY", True
             )
 
+    # Initialize FixedVector with exact count of source speakers
+    cg.add(var.init_source_speakers(len(config[CONF_SOURCE_SPEAKERS])))
+
     for speaker_config in config[CONF_SOURCE_SPEAKERS]:
         source_speaker = cg.new_Pvariable(speaker_config[CONF_ID])
 

esphome/components/mixer/speaker/automation.h

@@ -8,8 +8,8 @@
 namespace esphome {
 namespace mixer_speaker {
 template<typename... Ts> class DuckingApplyAction : public Action<Ts...>, public Parented<SourceSpeaker> {
-  TEMPLATABLE_VALUE(uint8_t, decibel_reduction)
-  TEMPLATABLE_VALUE(uint32_t, duration)
+  TEMPLATABLE_VALUE(uint8_t, decibel_reduction);
+  TEMPLATABLE_VALUE(uint32_t, duration);
   void play(const Ts &...x) override {
     this->parent_->apply_ducking(this->decibel_reduction_.value(x...), this->duration_.value(x...));
   }

esphome/components/mixer/speaker/mixer_speaker.cpp

@@ -2,11 +2,13 @@
 
 #ifdef USE_ESP32
 
+#include "esphome/core/application.h"
 #include "esphome/core/hal.h"
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
 #include <algorithm>
+#include <array>
 #include <cstring>
 
 namespace esphome {
@@ -14,6 +16,7 @@ namespace mixer_speaker {
 
 static const UBaseType_t MIXER_TASK_PRIORITY = 10;
 
+static const uint32_t STOPPING_TIMEOUT_MS = 5000;
 static const uint32_t TRANSFER_BUFFER_DURATION_MS = 50;
 static const uint32_t TASK_DELAY_MS = 25;
 
@@ -27,21 +30,53 @@ static const char *const TAG = "speaker_mixer";
 // Gives the Q15 fixed point scaling factor to reduce by 0 dB, 1dB, ..., 50 dB
 // dB to PCM scaling factor formula: floating_point_scale_factor = 2^(-db/6.014)
 // float to Q15 fixed point formula: q15_scale_factor = floating_point_scale_factor * 2^(15)
-static const std::vector<int16_t> DECIBEL_REDUCTION_TABLE = {
+static const std::array<int16_t, 51> DECIBEL_REDUCTION_TABLE = {
     32767, 29201, 26022, 23189, 20665, 18415, 16410, 14624, 13032, 11613, 10349, 9222, 8218, 7324, 6527, 5816, 5183,
     4619,  4116,  3668,  3269,  2913,  2596,  2313,  2061,  1837,  1637,  1459,  1300, 1158, 1032, 920,  820,  731,
     651,   580,   517,   461,   411,   366,   326,   291,   259,   231,   206,   183,  163,  146,  130,  116,  103};
 
-enum MixerEventGroupBits : uint32_t {
-  COMMAND_STOP = (1 << 0),  // stops the mixer task
-  STATE_STARTING = (1 << 10),
-  STATE_RUNNING = (1 << 11),
-  STATE_STOPPING = (1 << 12),
-  STATE_STOPPED = (1 << 13),
-  ERR_ESP_NO_MEM = (1 << 19),
-  ALL_BITS = 0x00FFFFFF,  // All valid FreeRTOS event group bits
+// Event bits for SourceSpeaker command processing
+enum SourceSpeakerEventBits : uint32_t {
+  SOURCE_SPEAKER_COMMAND_START = (1 << 0),
+  SOURCE_SPEAKER_COMMAND_STOP = (1 << 1),
+  SOURCE_SPEAKER_COMMAND_FINISH = (1 << 2),
 };
 
+// Event bits for mixer task control and state
+enum MixerTaskEventBits : uint32_t {
+  MIXER_TASK_COMMAND_START = (1 << 0),
+  MIXER_TASK_COMMAND_STOP = (1 << 1),
+  MIXER_TASK_STATE_STARTING = (1 << 10),
+  MIXER_TASK_STATE_RUNNING = (1 << 11),
+  MIXER_TASK_STATE_STOPPING = (1 << 12),
+  MIXER_TASK_STATE_STOPPED = (1 << 13),
+  MIXER_TASK_ERR_ESP_NO_MEM = (1 << 19),
+  MIXER_TASK_ALL_BITS = 0x00FFFFFF,  // All valid FreeRTOS event group bits
+};
+
+static inline uint32_t atomic_subtract_clamped(std::atomic<uint32_t> &var, uint32_t amount) {
+  uint32_t current = var.load(std::memory_order_acquire);
+  uint32_t subtracted = 0;
+  if (current > 0) {
+    uint32_t new_value;
+    do {
+      subtracted = std::min(amount, current);
+      new_value = current - subtracted;
+    } while (!var.compare_exchange_weak(current, new_value, std::memory_order_release, std::memory_order_acquire));
+  }
+  return subtracted;
+}
+
+static bool create_event_group(EventGroupHandle_t &event_group, Component *component) {
+  event_group = xEventGroupCreate();
+  if (event_group == nullptr) {
+    ESP_LOGE(TAG, "Failed to create event group");
+    component->mark_failed();
+    return false;
+  }
+  return true;
+}
+
 void SourceSpeaker::dump_config() {
   ESP_LOGCONFIG(TAG,
                 "Mixer Source Speaker\n"
@@ -55,22 +90,70 @@ void SourceSpeaker::dump_config() {
 }
 
 void SourceSpeaker::setup() {
-  this->parent_->get_output_speaker()->add_audio_output_callback([this](uint32_t new_frames, int64_t write_timestamp) {
-    // The SourceSpeaker may not have included any audio in the mixed output, so verify there were pending frames
-    uint32_t speakers_playback_frames = std::min(new_frames, this->pending_playback_frames_);
-    this->pending_playback_frames_ -= speakers_playback_frames;
+  if (!create_event_group(this->event_group_, this)) {
+    return;
+  }
 
-    if (speakers_playback_frames > 0) {
-      this->audio_output_callback_(speakers_playback_frames, write_timestamp);
+  // Start with loop disabled since we begin in STATE_STOPPED with no pending commands
+  this->disable_loop();
+
+  this->parent_->get_output_speaker()->add_audio_output_callback([this](uint32_t new_frames, int64_t write_timestamp) {
+    // First, drain the playback delay (frames in pipeline before this source started contributing)
+    uint32_t delay_to_drain = atomic_subtract_clamped(this->playback_delay_frames_, new_frames);
+    uint32_t remaining_frames = new_frames - delay_to_drain;
+
+    // Then, count towards this source's pending playback frames
+    if (remaining_frames > 0) {
+      uint32_t speakers_playback_frames = atomic_subtract_clamped(this->pending_playback_frames_, remaining_frames);
+      if (speakers_playback_frames > 0) {
+        this->audio_output_callback_(speakers_playback_frames, write_timestamp);
+      }
     }
   });
 }
 
 void SourceSpeaker::loop() {
+  uint32_t event_bits = xEventGroupGetBits(this->event_group_);
+
+  // Process commands with priority: STOP > FINISH > START
+  // This ensures stop commands take precedence over conflicting start commands
+  if (event_bits & SOURCE_SPEAKER_COMMAND_STOP) {
+    if (this->state_ == speaker::STATE_RUNNING) {
+      // Clear both STOP and START bits - stop takes precedence
+      xEventGroupClearBits(this->event_group_, SOURCE_SPEAKER_COMMAND_STOP | SOURCE_SPEAKER_COMMAND_START);
+      this->enter_stopping_state_();
+    } else if (this->state_ == speaker::STATE_STOPPED) {
+      // Already stopped, just clear the command bits
+      xEventGroupClearBits(this->event_group_, SOURCE_SPEAKER_COMMAND_STOP | SOURCE_SPEAKER_COMMAND_START);
+    }
+    // Leave bits set if transitioning states (STARTING/STOPPING) - will be processed once state allows
+  } else if (event_bits & SOURCE_SPEAKER_COMMAND_FINISH) {
+    if (this->state_ == speaker::STATE_RUNNING) {
+      xEventGroupClearBits(this->event_group_, SOURCE_SPEAKER_COMMAND_FINISH);
+      this->stop_gracefully_ = true;
+    } else if (this->state_ == speaker::STATE_STOPPED) {
+      // Already stopped, just clear the command bit
+      xEventGroupClearBits(this->event_group_, SOURCE_SPEAKER_COMMAND_FINISH);
+    }
+    // Leave bit set if transitioning states - will be processed once state allows
+  } else if (event_bits & SOURCE_SPEAKER_COMMAND_START) {
+    if (this->state_ == speaker::STATE_STOPPED) {
+      xEventGroupClearBits(this->event_group_, SOURCE_SPEAKER_COMMAND_START);
+      this->state_ = speaker::STATE_STARTING;
+    } else if (this->state_ == speaker::STATE_RUNNING) {
+      // Already running, just clear the command bit
+      xEventGroupClearBits(this->event_group_, SOURCE_SPEAKER_COMMAND_START);
+    }
+    // Leave bit set if transitioning states - will be processed once state allows
+  }
+  // Process state machine
   switch (this->state_) {
     case speaker::STATE_STARTING: {
       esp_err_t err = this->start_();
       if (err == ESP_OK) {
+        this->pending_playback_frames_.store(0, std::memory_order_release);  // reset pending playback frames
+        this->playback_delay_frames_.store(0, std::memory_order_release);    // reset playback delay
+        this->has_contributed_.store(false, std::memory_order_release);      // reset contribution tracking
         this->state_ = speaker::STATE_RUNNING;
         this->stop_gracefully_ = false;
         this->last_seen_data_ms_ = millis();
@@ -78,41 +161,62 @@ void SourceSpeaker::loop() {
       } else {
         switch (err) {
           case ESP_ERR_NO_MEM:
-            this->status_set_error(LOG_STR("Failed to start mixer: not enough memory"));
+            this->status_set_error(LOG_STR("Not enough memory"));
             break;
           case ESP_ERR_NOT_SUPPORTED:
-            this->status_set_error(LOG_STR("Failed to start mixer: unsupported bits per sample"));
+            this->status_set_error(LOG_STR("Unsupported bit depth"));
             break;
           case ESP_ERR_INVALID_ARG:
-            this->status_set_error(
-                LOG_STR("Failed to start mixer: audio stream isn't compatible with the other audio stream."));
+            this->status_set_error(LOG_STR("Incompatible audio streams"));
             break;
           case ESP_ERR_INVALID_STATE:
-            this->status_set_error(LOG_STR("Failed to start mixer: mixer task failed to start"));
+            this->status_set_error(LOG_STR("Task failed"));
             break;
           default:
-            this->status_set_error(LOG_STR("Failed to start mixer"));
+            this->status_set_error(LOG_STR("Failed"));
             break;
         }
 
-        this->state_ = speaker::STATE_STOPPING;
+        this->enter_stopping_state_();
       }
       break;
     }
     case speaker::STATE_RUNNING:
-      if (!this->transfer_buffer_->has_buffered_data()) {
+      if (!this->transfer_buffer_->has_buffered_data() &&
+          (this->pending_playback_frames_.load(std::memory_order_acquire) == 0)) {
+        // No audio data in buffer waiting to get mixed and no frames are pending playback
         if ((this->timeout_ms_.has_value() && ((millis() - this->last_seen_data_ms_) > this->timeout_ms_.value())) ||
             this->stop_gracefully_) {
-          this->state_ = speaker::STATE_STOPPING;
+          // Timeout exceeded or graceful stop requested
+          this->enter_stopping_state_();
         }
       }
       break;
-    case speaker::STATE_STOPPING:
-      this->stop_();
-      this->stop_gracefully_ = false;
-      this->state_ = speaker::STATE_STOPPED;
+    case speaker::STATE_STOPPING: {
+      if ((this->parent_->get_output_speaker()->get_pause_state()) ||
+          ((millis() - this->stopping_start_ms_) > STOPPING_TIMEOUT_MS)) {
+        // If parent speaker is paused or if the stopping timeout is exceeded, force stop the output speaker
+        this->parent_->get_output_speaker()->stop();
+      }
+
+      if (this->parent_->get_output_speaker()->is_stopped() ||
+          (this->pending_playback_frames_.load(std::memory_order_acquire) == 0)) {
+        // Output speaker is stopped OR all pending playback frames have played
+        this->pending_playback_frames_.store(0, std::memory_order_release);
+        this->stop_gracefully_ = false;
+
+        this->state_ = speaker::STATE_STOPPED;
+      }
       break;
+    }
     case speaker::STATE_STOPPED:
+      // Re-check event bits for any new commands that may have arrived
+      event_bits = xEventGroupGetBits(this->event_group_);
+      if (!(event_bits &
+            (SOURCE_SPEAKER_COMMAND_START | SOURCE_SPEAKER_COMMAND_STOP | SOURCE_SPEAKER_COMMAND_FINISH))) {
+        // No pending commands, disable loop to save CPU cycles
+        this->disable_loop();
+      }
       break;
   }
 }
@@ -122,17 +226,34 @@ size_t SourceSpeaker::play(const uint8_t *data, size_t length, TickType_t ticks_
     this->start();
   }
   size_t bytes_written = 0;
-  if (this->ring_buffer_.use_count() == 1) {
-    std::shared_ptr<RingBuffer> temp_ring_buffer = this->ring_buffer_.lock();
+  std::shared_ptr<RingBuffer> temp_ring_buffer = this->ring_buffer_.lock();
+  if (temp_ring_buffer.use_count() > 0) {
+    // Only write to the ring buffer if the reference is valid
     bytes_written = temp_ring_buffer->write_without_replacement(data, length, ticks_to_wait);
     if (bytes_written > 0) {
       this->last_seen_data_ms_ = millis();
     }
+  } else {
+    // Delay to avoid repeatedly hammering while waiting for the speaker to start
+    vTaskDelay(ticks_to_wait);
   }
   return bytes_written;
 }
 
-void SourceSpeaker::start() { this->state_ = speaker::STATE_STARTING; }
+void SourceSpeaker::send_command_(uint32_t command_bit, bool wake_loop) {
+  this->enable_loop_soon_any_context();
+  uint32_t event_bits = xEventGroupGetBits(this->event_group_);
+  if (!(event_bits & command_bit)) {
+    xEventGroupSetBits(this->event_group_, command_bit);
+#if defined(USE_SOCKET_SELECT_SUPPORT) && defined(USE_WAKE_LOOP_THREADSAFE)
+    if (wake_loop) {
+      App.wake_loop_threadsafe();
+    }
+#endif
+  }
+}
+
+void SourceSpeaker::start() { this->send_command_(SOURCE_SPEAKER_COMMAND_START, true); }
 
 esp_err_t SourceSpeaker::start_() {
   const size_t ring_buffer_size = this->audio_stream_info_.ms_to_bytes(this->buffer_duration_ms_);
@@ -143,35 +264,26 @@ esp_err_t SourceSpeaker::start_() {
     if (this->transfer_buffer_ == nullptr) {
       return ESP_ERR_NO_MEM;
     }
-    std::shared_ptr<RingBuffer> temp_ring_buffer;
 
-    if (!this->ring_buffer_.use_count()) {
+    std::shared_ptr<RingBuffer> temp_ring_buffer = this->ring_buffer_.lock();
+    if (!temp_ring_buffer) {
       temp_ring_buffer = RingBuffer::create(ring_buffer_size);
       this->ring_buffer_ = temp_ring_buffer;
     }
 
-    if (!this->ring_buffer_.use_count()) {
+    if (!temp_ring_buffer) {
       return ESP_ERR_NO_MEM;
     } else {
       this->transfer_buffer_->set_source(temp_ring_buffer);
     }
   }
 
-  this->pending_playback_frames_ = 0;  // reset
   return this->parent_->start(this->audio_stream_info_);
 }
 
-void SourceSpeaker::stop() {
-  if (this->state_ != speaker::STATE_STOPPED) {
-    this->state_ = speaker::STATE_STOPPING;
-  }
-}
+void SourceSpeaker::stop() { this->send_command_(SOURCE_SPEAKER_COMMAND_STOP); }
 
-void SourceSpeaker::stop_() {
-  this->transfer_buffer_.reset();  // deallocates the transfer buffer
-}
-
-void SourceSpeaker::finish() { this->stop_gracefully_ = true; }
+void SourceSpeaker::finish() { this->send_command_(SOURCE_SPEAKER_COMMAND_FINISH); }
 
 bool SourceSpeaker::has_buffered_data() const {
   return ((this->transfer_buffer_.use_count() > 0) && this->transfer_buffer_->has_buffered_data());
@@ -191,19 +303,16 @@ void SourceSpeaker::set_volume(float volume) {
 
 float SourceSpeaker::get_volume() { return this->parent_->get_output_speaker()->get_volume(); }
 
-size_t SourceSpeaker::process_data_from_source(TickType_t ticks_to_wait) {
-  if (!this->transfer_buffer_.use_count()) {
-    return 0;
-  }
-
+size_t SourceSpeaker::process_data_from_source(std::shared_ptr<audio::AudioSourceTransferBuffer> &transfer_buffer,
+                                               TickType_t ticks_to_wait) {
   // Store current offset, as these samples are already ducked
-  const size_t current_length = this->transfer_buffer_->available();
+  const size_t current_length = transfer_buffer->available();
 
-  size_t bytes_read = this->transfer_buffer_->transfer_data_from_source(ticks_to_wait);
+  size_t bytes_read = transfer_buffer->transfer_data_from_source(ticks_to_wait);
 
   uint32_t samples_to_duck = this->audio_stream_info_.bytes_to_samples(bytes_read);
   if (samples_to_duck > 0) {
-    int16_t *current_buffer = reinterpret_cast<int16_t *>(this->transfer_buffer_->get_buffer_start() + current_length);
+    int16_t *current_buffer = reinterpret_cast<int16_t *>(transfer_buffer->get_buffer_start() + current_length);
 
     duck_samples(current_buffer, samples_to_duck, &this->current_ducking_db_reduction_,
                  &this->ducking_transition_samples_remaining_, this->samples_per_ducking_step_,
@@ -215,10 +324,13 @@ size_t SourceSpeaker::process_data_from_source(TickType_t ticks_to_wait) {
 
 void SourceSpeaker::apply_ducking(uint8_t decibel_reduction, uint32_t duration) {
   if (this->target_ducking_db_reduction_ != decibel_reduction) {
+    // Start transition from the previous target (which becomes the new current level)
     this->current_ducking_db_reduction_ = this->target_ducking_db_reduction_;
 
     this->target_ducking_db_reduction_ = decibel_reduction;
 
+    // Calculate the number of intermediate dB steps for the transition timing.
+    // Subtract 1 because the first step is taken immediately after this calculation.
     uint8_t total_ducking_steps = 0;
     if (this->target_ducking_db_reduction_ > this->current_ducking_db_reduction_) {
       // The dB reduction level is increasing (which results in quieter audio)
@@ -234,7 +346,7 @@ void SourceSpeaker::apply_ducking(uint8_t decibel_reduction, uint32_t duration)
 
       this->samples_per_ducking_step_ = this->ducking_transition_samples_remaining_ / total_ducking_steps;
       this->ducking_transition_samples_remaining_ =
-          this->samples_per_ducking_step_ * total_ducking_steps;  // Adjust for integer division rounding
+          this->samples_per_ducking_step_ * total_ducking_steps;  // adjust for integer division rounding
 
       this->current_ducking_db_reduction_ += this->db_change_per_ducking_step_;
     } else {
@@ -293,6 +405,12 @@ void SourceSpeaker::duck_samples(int16_t *input_buffer, uint32_t input_samples_t
   }
 }
 
+void SourceSpeaker::enter_stopping_state_() {
+  this->state_ = speaker::STATE_STOPPING;
+  this->stopping_start_ms_ = millis();
+  this->transfer_buffer_.reset();
+}
+
 void MixerSpeaker::dump_config() {
   ESP_LOGCONFIG(TAG,
                 "Speaker Mixer:\n"
@@ -301,42 +419,74 @@ void MixerSpeaker::dump_config() {
 }
 
 void MixerSpeaker::setup() {
-  this->event_group_ = xEventGroupCreate();
-
-  if (this->event_group_ == nullptr) {
-    ESP_LOGE(TAG, "Failed to create event group");
-    this->mark_failed();
+  if (!create_event_group(this->event_group_, this)) {
     return;
   }
+
+  // Register callback to track frames in the output pipeline
+  this->output_speaker_->add_audio_output_callback([this](uint32_t new_frames, int64_t write_timestamp) {
+    atomic_subtract_clamped(this->frames_in_pipeline_, new_frames);
+  });
+
+  // Start with loop disabled since no task is running and no commands are pending
+  this->disable_loop();
 }
 
 void MixerSpeaker::loop() {
   uint32_t event_group_bits = xEventGroupGetBits(this->event_group_);
 
-  if (event_group_bits & MixerEventGroupBits::STATE_STARTING) {
-    ESP_LOGD(TAG, "Starting speaker mixer");
-    xEventGroupClearBits(this->event_group_, MixerEventGroupBits::STATE_STARTING);
+  // Handle pending start request
+  if (event_group_bits & MIXER_TASK_COMMAND_START) {
+    // Only start the task if it's fully stopped and cleaned up
+    if (!this->status_has_error() && (this->task_handle_ == nullptr) && (this->task_stack_buffer_ == nullptr)) {
+      esp_err_t err = this->start_task_();
+      switch (err) {
+        case ESP_OK:
+          xEventGroupClearBits(this->event_group_, MIXER_TASK_COMMAND_START);
+          break;
+        case ESP_ERR_NO_MEM:
+          ESP_LOGE(TAG, "Failed to start; retrying in 1 second");
+          this->status_momentary_error("memory-failure", 1000);
+          return;
+        case ESP_ERR_INVALID_STATE:
+          ESP_LOGE(TAG, "Failed to start; retrying in 1 second");
+          this->status_momentary_error("task-failure", 1000);
+          return;
+        default:
+          ESP_LOGE(TAG, "Failed to start; retrying in 1 second");
+          this->status_momentary_error("failure", 1000);
+          return;
+      }
+    }
   }
-  if (event_group_bits & MixerEventGroupBits::ERR_ESP_NO_MEM) {
-    this->status_set_error(LOG_STR("Failed to allocate the mixer's internal buffer"));
-    xEventGroupClearBits(this->event_group_, MixerEventGroupBits::ERR_ESP_NO_MEM);
+
+  if (event_group_bits & MIXER_TASK_STATE_STARTING) {
+    ESP_LOGD(TAG, "Starting");
+    xEventGroupClearBits(this->event_group_, MIXER_TASK_STATE_STARTING);
   }
-  if (event_group_bits & MixerEventGroupBits::STATE_RUNNING) {
-    ESP_LOGD(TAG, "Started speaker mixer");
+  if (event_group_bits & MIXER_TASK_ERR_ESP_NO_MEM) {
+    this->status_set_error(LOG_STR("Not enough memory"));
+    xEventGroupClearBits(this->event_group_, MIXER_TASK_ERR_ESP_NO_MEM);
+  }
+  if (event_group_bits & MIXER_TASK_STATE_RUNNING) {
+    ESP_LOGV(TAG, "Started");
     this->status_clear_error();
-    xEventGroupClearBits(this->event_group_, MixerEventGroupBits::STATE_RUNNING);
+    xEventGroupClearBits(this->event_group_, MIXER_TASK_STATE_RUNNING);
   }
-  if (event_group_bits & MixerEventGroupBits::STATE_STOPPING) {
-    ESP_LOGD(TAG, "Stopping speaker mixer");
-    xEventGroupClearBits(this->event_group_, MixerEventGroupBits::STATE_STOPPING);
+  if (event_group_bits & MIXER_TASK_STATE_STOPPING) {
+    ESP_LOGV(TAG, "Stopping");
+    xEventGroupClearBits(this->event_group_, MIXER_TASK_STATE_STOPPING);
   }
-  if (event_group_bits & MixerEventGroupBits::STATE_STOPPED) {
+  if (event_group_bits & MIXER_TASK_STATE_STOPPED) {
     if (this->delete_task_() == ESP_OK) {
-      xEventGroupClearBits(this->event_group_, MixerEventGroupBits::ALL_BITS);
+      ESP_LOGD(TAG, "Stopped");
+      xEventGroupClearBits(this->event_group_, MIXER_TASK_ALL_BITS);
     }
   }
 
   if (this->task_handle_ != nullptr) {
+    // If the mixer task is running, check if all source speakers are stopped
+
     bool all_stopped = true;
 
     for (auto &speaker : this->source_speakers_) {
@@ -344,7 +494,15 @@ void MixerSpeaker::loop() {
     }
 
     if (all_stopped) {
-      this->stop();
+      // Send stop command signal to the mixer task since no source speakers are active
+      xEventGroupSetBits(this->event_group_, MIXER_TASK_COMMAND_STOP);
+    }
+  } else if (this->task_stack_buffer_ == nullptr) {
+    // Task is fully stopped and cleaned up, check if we can disable loop
+    event_group_bits = xEventGroupGetBits(this->event_group_);
+    if (event_group_bits == 0) {
+      // No pending events, disable loop to save CPU cycles
+      this->disable_loop();
     }
   }
 }
@@ -366,7 +524,18 @@ esp_err_t MixerSpeaker::start(audio::AudioStreamInfo &stream_info) {
     }
   }
 
-  return this->start_task_();
+  this->enable_loop_soon_any_context();  // ensure loop processes command
+
+  uint32_t event_bits = xEventGroupGetBits(this->event_group_);
+  if (!(event_bits & MIXER_TASK_COMMAND_START)) {
+    // Set MIXER_TASK_COMMAND_START bit if not already set, and then immediately wake for low latency
+    xEventGroupSetBits(this->event_group_, MIXER_TASK_COMMAND_START);
+#if defined(USE_SOCKET_SELECT_SUPPORT) && defined(USE_WAKE_LOOP_THREADSAFE)
+    App.wake_loop_threadsafe();
+#endif
+  }
+
+  return ESP_OK;
 }
 
 esp_err_t MixerSpeaker::start_task_() {
@@ -397,28 +566,31 @@ esp_err_t MixerSpeaker::start_task_() {
 }
 
 esp_err_t MixerSpeaker::delete_task_() {
-  if (!this->task_created_) {
+  if (this->task_handle_ != nullptr) {
+    // Delete the task
+    vTaskDelete(this->task_handle_);
     this->task_handle_ = nullptr;
-
-    if (this->task_stack_buffer_ != nullptr) {
-      if (this->task_stack_in_psram_) {
-        RAMAllocator<StackType_t> stack_allocator(RAMAllocator<StackType_t>::ALLOC_EXTERNAL);
-        stack_allocator.deallocate(this->task_stack_buffer_, TASK_STACK_SIZE);
-      } else {
-        RAMAllocator<StackType_t> stack_allocator(RAMAllocator<StackType_t>::ALLOC_INTERNAL);
-        stack_allocator.deallocate(this->task_stack_buffer_, TASK_STACK_SIZE);
-      }
-
-      this->task_stack_buffer_ = nullptr;
-    }
-
-    return ESP_OK;
   }
 
-  return ESP_ERR_INVALID_STATE;
-}
+  if ((this->task_handle_ == nullptr) && (this->task_stack_buffer_ != nullptr)) {
+    // Deallocate the task stack buffer
+    if (this->task_stack_in_psram_) {
+      RAMAllocator<StackType_t> stack_allocator(RAMAllocator<StackType_t>::ALLOC_EXTERNAL);
+      stack_allocator.deallocate(this->task_stack_buffer_, TASK_STACK_SIZE);
+    } else {
+      RAMAllocator<StackType_t> stack_allocator(RAMAllocator<StackType_t>::ALLOC_INTERNAL);
+      stack_allocator.deallocate(this->task_stack_buffer_, TASK_STACK_SIZE);
+    }
 
-void MixerSpeaker::stop() { xEventGroupSetBits(this->event_group_, MixerEventGroupBits::COMMAND_STOP); }
+    this->task_stack_buffer_ = nullptr;
+  }
+
+  if ((this->task_handle_ != nullptr) || (this->task_stack_buffer_ != nullptr)) {
+    return ESP_ERR_INVALID_STATE;
+  }
+
+  return ESP_OK;
+}
 
 void MixerSpeaker::copy_frames(const int16_t *input_buffer, audio::AudioStreamInfo input_stream_info,
                                int16_t *output_buffer, audio::AudioStreamInfo output_stream_info,
@@ -472,32 +644,34 @@ void MixerSpeaker::mix_audio_samples(const int16_t *primary_buffer, audio::Audio
 }
 
 void MixerSpeaker::audio_mixer_task(void *params) {
-  MixerSpeaker *this_mixer = (MixerSpeaker *) params;
+  MixerSpeaker *this_mixer = static_cast<MixerSpeaker *>(params);
 
-  xEventGroupSetBits(this_mixer->event_group_, MixerEventGroupBits::STATE_STARTING);
-
-  this_mixer->task_created_ = true;
+  xEventGroupSetBits(this_mixer->event_group_, MIXER_TASK_STATE_STARTING);
 
   std::unique_ptr<audio::AudioSinkTransferBuffer> output_transfer_buffer = audio::AudioSinkTransferBuffer::create(
       this_mixer->audio_stream_info_.value().ms_to_bytes(TRANSFER_BUFFER_DURATION_MS));
 
   if (output_transfer_buffer == nullptr) {
-    xEventGroupSetBits(this_mixer->event_group_,
-                       MixerEventGroupBits::STATE_STOPPED | MixerEventGroupBits::ERR_ESP_NO_MEM);
+    xEventGroupSetBits(this_mixer->event_group_, MIXER_TASK_STATE_STOPPED | MIXER_TASK_ERR_ESP_NO_MEM);
 
-    this_mixer->task_created_ = false;
-    vTaskDelete(nullptr);
+    vTaskSuspend(nullptr);  // Suspend this task indefinitely until the loop method deletes it
   }
 
   output_transfer_buffer->set_sink(this_mixer->output_speaker_);
 
-  xEventGroupSetBits(this_mixer->event_group_, MixerEventGroupBits::STATE_RUNNING);
+  xEventGroupSetBits(this_mixer->event_group_, MIXER_TASK_STATE_RUNNING);
 
   bool sent_finished = false;
 
+  // Pre-allocate vectors to avoid heap allocation in the loop (max 8 source speakers per schema)
+  FixedVector<SourceSpeaker *> speakers_with_data;
+  FixedVector<std::shared_ptr<audio::AudioSourceTransferBuffer>> transfer_buffers_with_data;
+  speakers_with_data.init(this_mixer->source_speakers_.size());
+  transfer_buffers_with_data.init(this_mixer->source_speakers_.size());
+
   while (true) {
     uint32_t event_group_bits = xEventGroupGetBits(this_mixer->event_group_);
-    if (event_group_bits & MixerEventGroupBits::COMMAND_STOP) {
+    if (event_group_bits & MIXER_TASK_COMMAND_STOP) {
       break;
     }
 
@@ -507,15 +681,20 @@ void MixerSpeaker::audio_mixer_task(void *params) {
     const uint32_t output_frames_free =
         this_mixer->audio_stream_info_.value().bytes_to_frames(output_transfer_buffer->free());
 
-    std::vector<SourceSpeaker *> speakers_with_data;
-    std::vector<std::shared_ptr<audio::AudioSourceTransferBuffer>> transfer_buffers_with_data;
+    speakers_with_data.clear();
+    transfer_buffers_with_data.clear();
 
     for (auto &speaker : this_mixer->source_speakers_) {
-      if (speaker->get_transfer_buffer().use_count() > 0) {
+      if (speaker->is_running() && !speaker->get_pause_state()) {
+        // Speaker is running and not paused, so it possibly can provide audio data
         std::shared_ptr<audio::AudioSourceTransferBuffer> transfer_buffer = speaker->get_transfer_buffer().lock();
-        speaker->process_data_from_source(0);  // Transfers and ducks audio from source ring buffers
+        if (transfer_buffer.use_count() == 0) {
+          // No transfer buffer allocated, so skip processing this speaker
+          continue;
+        }
+        speaker->process_data_from_source(transfer_buffer, 0);  // Transfers and ducks audio from source ring buffers
 
-        if ((transfer_buffer->available() > 0) && !speaker->get_pause_state()) {
+        if (transfer_buffer->available() > 0) {
           // Store the locked transfer buffers in their own vector to avoid releasing ownership until after the loop
           transfer_buffers_with_data.push_back(transfer_buffer);
           speakers_with_data.push_back(speaker);
@@ -547,13 +726,21 @@ void MixerSpeaker::audio_mixer_task(void *params) {
                     reinterpret_cast<int16_t *>(output_transfer_buffer->get_buffer_end()),
                     this_mixer->audio_stream_info_.value(), frames_to_mix);
 
-        // Update source speaker buffer length
-        transfer_buffers_with_data[0]->decrease_buffer_length(active_stream_info.frames_to_bytes(frames_to_mix));
-        speakers_with_data[0]->pending_playback_frames_ += frames_to_mix;
+        // Set playback delay for newly contributing source
+        if (!speakers_with_data[0]->has_contributed_.load(std::memory_order_acquire)) {
+          speakers_with_data[0]->playback_delay_frames_.store(
+              this_mixer->frames_in_pipeline_.load(std::memory_order_acquire), std::memory_order_release);
+          speakers_with_data[0]->has_contributed_.store(true, std::memory_order_release);
+        }
 
-        // Update output transfer buffer length
+        // Update source speaker pending frames
+        speakers_with_data[0]->pending_playback_frames_.fetch_add(frames_to_mix, std::memory_order_release);
+        transfer_buffers_with_data[0]->decrease_buffer_length(active_stream_info.frames_to_bytes(frames_to_mix));
+
+        // Update output transfer buffer length and pipeline frame count
         output_transfer_buffer->increase_buffer_length(
             this_mixer->audio_stream_info_.value().frames_to_bytes(frames_to_mix));
+        this_mixer->frames_in_pipeline_.fetch_add(frames_to_mix, std::memory_order_release);
       } else {
         // Speaker's stream info doesn't match the output speaker's, so it's a new source speaker
         if (!this_mixer->output_speaker_->is_stopped()) {
@@ -568,6 +755,8 @@ void MixerSpeaker::audio_mixer_task(void *params) {
                                      active_stream_info.get_sample_rate());
           this_mixer->output_speaker_->set_audio_stream_info(this_mixer->audio_stream_info_.value());
           this_mixer->output_speaker_->start();
+          // Reset pipeline frame count since we're starting fresh with a new sample rate
+          this_mixer->frames_in_pipeline_.store(0, std::memory_order_release);
           sent_finished = false;
         }
       }
@@ -596,26 +785,39 @@ void MixerSpeaker::audio_mixer_task(void *params) {
         }
       }
 
+      // Get current pipeline depth for delay calculation (before incrementing)
+      uint32_t current_pipeline_frames = this_mixer->frames_in_pipeline_.load(std::memory_order_acquire);
+
       // Update source transfer buffer lengths and add new audio durations to the source speaker pending playbacks
       for (size_t i = 0; i < transfer_buffers_with_data.size(); ++i) {
+        // Set playback delay for newly contributing sources
+        if (!speakers_with_data[i]->has_contributed_.load(std::memory_order_acquire)) {
+          speakers_with_data[i]->playback_delay_frames_.store(current_pipeline_frames, std::memory_order_release);
+          speakers_with_data[i]->has_contributed_.store(true, std::memory_order_release);
+        }
+
+        speakers_with_data[i]->pending_playback_frames_.fetch_add(frames_to_mix, std::memory_order_release);
         transfer_buffers_with_data[i]->decrease_buffer_length(
             speakers_with_data[i]->get_audio_stream_info().frames_to_bytes(frames_to_mix));
-        speakers_with_data[i]->pending_playback_frames_ += frames_to_mix;
       }
 
-      // Update output transfer buffer length
+      // Update output transfer buffer length and pipeline frame count (once, not per source)
       output_transfer_buffer->increase_buffer_length(
           this_mixer->audio_stream_info_.value().frames_to_bytes(frames_to_mix));
+      this_mixer->frames_in_pipeline_.fetch_add(frames_to_mix, std::memory_order_release);
     }
   }
 
-  xEventGroupSetBits(this_mixer->event_group_, MixerEventGroupBits::STATE_STOPPING);
+  xEventGroupSetBits(this_mixer->event_group_, MIXER_TASK_STATE_STOPPING);
+
+  // Reset pipeline frame count since the task is stopping
+  this_mixer->frames_in_pipeline_.store(0, std::memory_order_release);
 
   output_transfer_buffer.reset();
 
-  xEventGroupSetBits(this_mixer->event_group_, MixerEventGroupBits::STATE_STOPPED);
-  this_mixer->task_created_ = false;
-  vTaskDelete(nullptr);
+  xEventGroupSetBits(this_mixer->event_group_, MIXER_TASK_STATE_STOPPED);
+
+  vTaskSuspend(nullptr);  // Suspend this task indefinitely until the loop method deletes it
 }
 
 }  // namespace mixer_speaker

esphome/components/mixer/speaker/mixer_speaker.h

@@ -7,26 +7,31 @@
 #include "esphome/components/speaker/speaker.h"
 
 #include "esphome/core/component.h"
+#include "esphome/core/helpers.h"
 
-#include <freertos/event_groups.h>
 #include <freertos/FreeRTOS.h>
+#include <freertos/event_groups.h>
+
+#include <atomic>
 
 namespace esphome {
 namespace mixer_speaker {
 
 /* Classes for mixing several source speaker audio streams and writing it to another speaker component.
  *  - Volume controls are passed through to the output speaker
+ *  - Source speaker commands are signaled via event group bits and processed in its loop function to ensure thread
+ * safety
  *  - Directly handles pausing at the SourceSpeaker level; pause state is not passed through to the output speaker.
- *  - Audio sent to the SourceSpeaker's must have 16 bits per sample.
+ *  - Audio sent to the SourceSpeaker must have 16 bits per sample.
  *  - Audio sent to the SourceSpeaker can have any number of channels. They are duplicated or ignored as needed to match
  *    the number of channels required for the output speaker.
- *  - In queue mode, the audio sent to the SoureSpeakers can have different sample rates.
+ *  - In queue mode, the audio sent to the SourceSpeakers can have different sample rates.
  *  - In non-queue mode, the audio sent to the SourceSpeakers must have the same sample rates.
  *  - SourceSpeaker has an internal ring buffer. It also allocates a shared_ptr for an AudioTranserBuffer object.
  *  - Audio Data Flow:
  *      - Audio data played on a SourceSpeaker first writes to its internal ring buffer.
  *      - MixerSpeaker task temporarily takes shared ownership of each SourceSpeaker's AudioTransferBuffer.
- *      - MixerSpeaker calls SourceSpeaker's `process_data_from_source`, which tranfers audio from the SourceSpeaker's
+ *      - MixerSpeaker calls SourceSpeaker's `process_data_from_source`, which transfers audio from the SourceSpeaker's
  *        ring buffer to its AudioTransferBuffer. Audio ducking is applied at this step.
  *      - In queue mode, MixerSpeaker prioritizes the earliest configured SourceSpeaker with audio data. Audio data is
  *        sent to the output speaker.
@@ -63,13 +68,15 @@ class SourceSpeaker : public speaker::Speaker, public Component {
   bool get_pause_state() const override { return this->pause_state_; }
 
   /// @brief Transfers audio from the ring buffer into the transfer buffer. Ducks audio while transferring.
+  /// @param transfer_buffer Locked shared_ptr to the transfer buffer (must be valid, not null)
   /// @param ticks_to_wait FreeRTOS ticks to wait while waiting to read from the ring buffer.
   /// @return Number of bytes transferred from the ring buffer.
-  size_t process_data_from_source(TickType_t ticks_to_wait);
+  size_t process_data_from_source(std::shared_ptr<audio::AudioSourceTransferBuffer> &transfer_buffer,
+                                  TickType_t ticks_to_wait);
 
   /// @brief Sets the ducking level for the source speaker.
-  /// @param decibel_reduction (uint8_t) The dB reduction level. For example, 0 is no change, 10 is a reduction by 10 dB
-  /// @param duration (uint32_t) The number of milliseconds to transition from the current level to the new level
+  /// @param decibel_reduction The dB reduction level. For example, 0 is no change, 10 is a reduction by 10 dB
+  /// @param duration The number of milliseconds to transition from the current level to the new level
   void apply_ducking(uint8_t decibel_reduction, uint32_t duration);
 
   void set_buffer_duration(uint32_t buffer_duration_ms) { this->buffer_duration_ms_ = buffer_duration_ms; }
@@ -81,14 +88,15 @@ class SourceSpeaker : public speaker::Speaker, public Component {
  protected:
   friend class MixerSpeaker;
   esp_err_t start_();
-  void stop_();
+  void enter_stopping_state_();
+  void send_command_(uint32_t command_bit, bool wake_loop = false);
 
   /// @brief Ducks audio samples by a specified amount. When changing the ducking amount, it can transition gradually
   /// over a specified amount of samples.
   /// @param input_buffer buffer with audio samples to be ducked in place
   /// @param input_samples_to_duck number of samples to process in ``input_buffer``
   /// @param current_ducking_db_reduction pointer to the current dB reduction
-  /// @param ducking_transition_samples_remaining pointer to the total number of samples left before the the
+  /// @param ducking_transition_samples_remaining pointer to the total number of samples left before the
   ///         transition is finished
   /// @param samples_per_ducking_step total number of samples per ducking step for the transition
   /// @param db_change_per_ducking_step the change in dB reduction per step
@@ -114,7 +122,12 @@ class SourceSpeaker : public speaker::Speaker, public Component {
   uint32_t ducking_transition_samples_remaining_{0};
   uint32_t samples_per_ducking_step_{0};
 
-  uint32_t pending_playback_frames_{0};
+  std::atomic<uint32_t> pending_playback_frames_{0};
+  std::atomic<uint32_t> playback_delay_frames_{0};  // Frames in output pipeline when this source started contributing
+  std::atomic<bool> has_contributed_{false};        // Tracks if source has contributed during this session
+
+  EventGroupHandle_t event_group_{nullptr};
+  uint32_t stopping_start_ms_{0};
 };
 
 class MixerSpeaker : public Component {
@@ -123,10 +136,11 @@ class MixerSpeaker : public Component {
   void setup() override;
   void loop() override;
 
+  void init_source_speakers(size_t count) { this->source_speakers_.init(count); }
   void add_source_speaker(SourceSpeaker *source_speaker) { this->source_speakers_.push_back(source_speaker); }
 
   /// @brief Starts the mixer task. Called by a source speaker giving the current audio stream information
-  /// @param stream_info The calling source speakers audio stream information
+  /// @param stream_info The calling source speaker's audio stream information
   /// @return ESP_ERR_NOT_SUPPORTED if the incoming stream is incompatible due to unsupported bits per sample
   ///         ESP_ERR_INVALID_ARG if the incoming stream is incompatible to be mixed with the other input audio stream
   ///         ESP_ERR_NO_MEM if there isn't enough memory for the task's stack
@@ -134,8 +148,6 @@ class MixerSpeaker : public Component {
   ///         ESP_OK if the incoming stream is compatible and the mixer task starts
   esp_err_t start(audio::AudioStreamInfo &stream_info);
 
-  void stop();
-
   void set_output_channels(uint8_t output_channels) { this->output_channels_ = output_channels; }
   void set_output_speaker(speaker::Speaker *speaker) { this->output_speaker_ = speaker; }
   void set_queue_mode(bool queue_mode) { this->queue_mode_ = queue_mode; }
@@ -143,6 +155,9 @@ class MixerSpeaker : public Component {
 
   speaker::Speaker *get_output_speaker() const { return this->output_speaker_; }
 
+  /// @brief Returns the current number of frames in the output pipeline (written but not yet played)
+  uint32_t get_frames_in_pipeline() const { return this->frames_in_pipeline_.load(std::memory_order_acquire); }
+
  protected:
   /// @brief Copies audio frames from the input buffer to the output buffer taking into account the number of channels
   /// in each stream. If the output stream has more channels, the input samples are duplicated. If the output stream has
@@ -159,11 +174,11 @@ class MixerSpeaker : public Component {
   /// and secondary samples are duplicated or dropped as necessary to ensure the output stream has the configured number
   /// of channels. Output samples are clamped to the corresponding int16 min or max values if the mixed sample
   /// overflows.
-  /// @param primary_buffer (int16_t *) samples buffer for the primary stream
+  /// @param primary_buffer samples buffer for the primary stream
   /// @param primary_stream_info stream info for the primary stream
-  /// @param secondary_buffer (int16_t *) samples buffer for secondary stream
+  /// @param secondary_buffer samples buffer for secondary stream
   /// @param secondary_stream_info stream info for the secondary stream
-  /// @param output_buffer (int16_t *) buffer for the mixed samples
+  /// @param output_buffer buffer for the mixed samples
   /// @param output_stream_info stream info for the output buffer
   /// @param frames_to_mix number of frames in the primary and secondary buffers to mix together
   static void mix_audio_samples(const int16_t *primary_buffer, audio::AudioStreamInfo primary_stream_info,
@@ -185,20 +200,20 @@ class MixerSpeaker : public Component {
 
   EventGroupHandle_t event_group_{nullptr};
 
-  std::vector<SourceSpeaker *> source_speakers_;
+  FixedVector<SourceSpeaker *> source_speakers_;
   speaker::Speaker *output_speaker_{nullptr};
 
   uint8_t output_channels_;
   bool queue_mode_;
   bool task_stack_in_psram_{false};
 
-  bool task_created_{false};
-
   TaskHandle_t task_handle_{nullptr};
   StaticTask_t task_stack_;
   StackType_t *task_stack_buffer_{nullptr};
 
   optional<audio::AudioStreamInfo> audio_stream_info_;
+
+  std::atomic<uint32_t> frames_in_pipeline_{0};  // Frames written to output but not yet played
 };
 
 }  // namespace mixer_speaker

esphome/components/modbus/modbus.cpp

@@ -19,16 +19,25 @@ void Modbus::setup() {
 void Modbus::loop() {
   const uint32_t now = App.get_loop_component_start_time();
 
-  while (this->available()) {
-    uint8_t byte;
-    this->read_byte(&byte);
-    if (this->parse_modbus_byte_(byte)) {
-      this->last_modbus_byte_ = now;
-    } else {
-      size_t at = this->rx_buffer_.size();
-      if (at > 0) {
-        ESP_LOGV(TAG, "Clearing buffer of %d bytes - parse failed", at);
-        this->rx_buffer_.clear();
+  // Read all available bytes in batches to reduce UART call overhead.
+  int avail = this->available();
+  uint8_t buf[64];
+  while (avail > 0) {
+    size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+    if (!this->read_array(buf, to_read)) {
+      break;
+    }
+    avail -= to_read;
+
+    for (size_t i = 0; i < to_read; i++) {
+      if (this->parse_modbus_byte_(buf[i])) {
+        this->last_modbus_byte_ = now;
+      } else {
+        size_t at = this->rx_buffer_.size();
+        if (at > 0) {
+          ESP_LOGV(TAG, "Clearing buffer of %d bytes - parse failed", at);
+          this->rx_buffer_.clear();
+        }
       }
     }
   }

esphome/components/nextion/nextion.cpp

@@ -397,11 +397,17 @@ bool Nextion::remove_from_q_(bool report_empty) {
 }
 
 void Nextion::process_serial_() {
-  uint8_t d;
+  // Read all available bytes in batches to reduce UART call overhead.
+  int avail = this->available();
+  uint8_t buf[64];
+  while (avail > 0) {
+    size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+    if (!this->read_array(buf, to_read)) {
+      break;
+    }
+    avail -= to_read;
 
-  while (this->available()) {
-    read_byte(&d);
-    this->command_data_ += d;
+    this->command_data_.append(reinterpret_cast<const char *>(buf), to_read);
   }
 }
 // nextion.tech/instruction-set/

esphome/components/pipsolar/pipsolar.cpp

@@ -13,9 +13,12 @@ void Pipsolar::setup() {
 }
 
 void Pipsolar::empty_uart_buffer_() {
-  uint8_t byte;
-  while (this->available()) {
-    this->read_byte(&byte);
+  uint8_t buf[64];
+  int avail;
+  while ((avail = this->available()) > 0) {
+    if (!this->read_array(buf, std::min(static_cast<size_t>(avail), sizeof(buf)))) {
+      break;
+    }
   }
 }
 
@@ -94,32 +97,47 @@ void Pipsolar::loop() {
   }
 
   if (this->state_ == STATE_COMMAND || this->state_ == STATE_POLL) {
-    while (this->available()) {
-      uint8_t byte;
-      this->read_byte(&byte);
-
-      // make sure data and null terminator fit in buffer
-      if (this->read_pos_ >= PIPSOLAR_READ_BUFFER_LENGTH - 1) {
-        this->read_pos_ = 0;
-        this->empty_uart_buffer_();
-        ESP_LOGW(TAG, "response data too long, discarding.");
+    int avail = this->available();
+    while (avail > 0) {
+      uint8_t buf[64];
+      size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+      if (!this->read_array(buf, to_read)) {
         break;
       }
-      this->read_buffer_[this->read_pos_] = byte;
-      this->read_pos_++;
+      avail -= to_read;
+      bool done = false;
+      for (size_t i = 0; i < to_read; i++) {
+        uint8_t byte = buf[i];
 
-      // end of answer
-      if (byte == 0x0D) {
-        this->read_buffer_[this->read_pos_] = 0;
-        this->empty_uart_buffer_();
-        if (this->state_ == STATE_POLL) {
-          this->state_ = STATE_POLL_COMPLETE;
+        // make sure data and null terminator fit in buffer
+        if (this->read_pos_ >= PIPSOLAR_READ_BUFFER_LENGTH - 1) {
+          this->read_pos_ = 0;
+          this->empty_uart_buffer_();
+          ESP_LOGW(TAG, "response data too long, discarding.");
+          done = true;
+          break;
         }
-        if (this->state_ == STATE_COMMAND) {
-          this->state_ = STATE_COMMAND_COMPLETE;
+        this->read_buffer_[this->read_pos_] = byte;
+        this->read_pos_++;
+
+        // end of answer
+        if (byte == 0x0D) {
+          this->read_buffer_[this->read_pos_] = 0;
+          this->empty_uart_buffer_();
+          if (this->state_ == STATE_POLL) {
+            this->state_ = STATE_POLL_COMPLETE;
+          }
+          if (this->state_ == STATE_COMMAND) {
+            this->state_ = STATE_COMMAND_COMPLETE;
+          }
+          done = true;
+          break;
         }
       }
-    }  // available
+      if (done) {
+        break;
+      }
+    }
   }
   if (this->state_ == STATE_COMMAND) {
     if (millis() - this->command_start_millis_ > esphome::pipsolar::Pipsolar::COMMAND_TIMEOUT) {

esphome/components/pylontech/pylontech.cpp

@@ -56,17 +56,23 @@ void PylontechComponent::setup() {
 void PylontechComponent::update() { this->write_str("pwr\n"); }
 
 void PylontechComponent::loop() {
-  if (this->available() > 0) {
+  int avail = this->available();
+  if (avail > 0) {
     // pylontech sends a lot of data very suddenly
     // we need to quickly put it all into our own buffer, otherwise the uart's buffer will overflow
-    uint8_t data;
     int recv = 0;
-    while (this->available() > 0) {
-      if (this->read_byte(&data)) {
-        buffer_[buffer_index_write_] += (char) data;
-        recv++;
-        if (buffer_[buffer_index_write_].back() == static_cast<char>(ASCII_LF) ||
-            buffer_[buffer_index_write_].length() >= MAX_DATA_LENGTH_BYTES) {
+    uint8_t buf[64];
+    while (avail > 0) {
+      size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+      if (!this->read_array(buf, to_read)) {
+        break;
+      }
+      avail -= to_read;
+      recv += to_read;
+
+      for (size_t i = 0; i < to_read; i++) {
+        buffer_[buffer_index_write_] += (char) buf[i];
+        if (buf[i] == ASCII_LF || buffer_[buffer_index_write_].length() >= MAX_DATA_LENGTH_BYTES) {
           // complete line received
           buffer_index_write_ = (buffer_index_write_ + 1) % NUM_BUFFERS;
         }

esphome/components/rd03d/rd03d.cpp

@@ -1,4 +1,5 @@
 #include "rd03d.h"
+#include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 #include <cmath>
 
@@ -80,37 +81,47 @@ void RD03DComponent::dump_config() {
 }
 
 void RD03DComponent::loop() {
-  while (this->available()) {
-    uint8_t byte = this->read();
-    ESP_LOGVV(TAG, "Received byte: 0x%02X, buffer_pos: %d", byte, this->buffer_pos_);
+  // Read all available bytes in batches to reduce UART call overhead.
+  int avail = this->available();
+  uint8_t buf[64];
+  while (avail > 0) {
+    size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+    if (!this->read_array(buf, to_read)) {
+      break;
+    }
+    avail -= to_read;
+    for (size_t i = 0; i < to_read; i++) {
+      uint8_t byte = buf[i];
+      ESP_LOGVV(TAG, "Received byte: 0x%02X, buffer_pos: %d", byte, this->buffer_pos_);
 
-    // Check if we're looking for frame header
-    if (this->buffer_pos_ < FRAME_HEADER_SIZE) {
-      if (byte == FRAME_HEADER[this->buffer_pos_]) {
-        this->buffer_[this->buffer_pos_++] = byte;
-      } else if (byte == FRAME_HEADER[0]) {
-        // Start over if we see a potential new header
-        this->buffer_[0] = byte;
-        this->buffer_pos_ = 1;
-      } else {
+      // Check if we're looking for frame header
+      if (this->buffer_pos_ < FRAME_HEADER_SIZE) {
+        if (byte == FRAME_HEADER[this->buffer_pos_]) {
+          this->buffer_[this->buffer_pos_++] = byte;
+        } else if (byte == FRAME_HEADER[0]) {
+          // Start over if we see a potential new header
+          this->buffer_[0] = byte;
+          this->buffer_pos_ = 1;
+        } else {
+          this->buffer_pos_ = 0;
+        }
+        continue;
+      }
+
+      // Accumulate data bytes
+      this->buffer_[this->buffer_pos_++] = byte;
+
+      // Check if we have a complete frame
+      if (this->buffer_pos_ == FRAME_SIZE) {
+        // Validate footer
+        if (this->buffer_[FRAME_SIZE - 2] == FRAME_FOOTER[0] && this->buffer_[FRAME_SIZE - 1] == FRAME_FOOTER[1]) {
+          this->process_frame_();
+        } else {
+          ESP_LOGW(TAG, "Invalid frame footer: 0x%02X 0x%02X (expected 0x55 0xCC)", this->buffer_[FRAME_SIZE - 2],
+                   this->buffer_[FRAME_SIZE - 1]);
+        }
         this->buffer_pos_ = 0;
       }
-      continue;
-    }
-
-    // Accumulate data bytes
-    this->buffer_[this->buffer_pos_++] = byte;
-
-    // Check if we have a complete frame
-    if (this->buffer_pos_ == FRAME_SIZE) {
-      // Validate footer
-      if (this->buffer_[FRAME_SIZE - 2] == FRAME_FOOTER[0] && this->buffer_[FRAME_SIZE - 1] == FRAME_FOOTER[1]) {
-        this->process_frame_();
-      } else {
-        ESP_LOGW(TAG, "Invalid frame footer: 0x%02X 0x%02X (expected 0x55 0xCC)", this->buffer_[FRAME_SIZE - 2],
-                 this->buffer_[FRAME_SIZE - 1]);
-      }
-      this->buffer_pos_ = 0;
     }
   }
 }

esphome/components/rf_bridge/rf_bridge.cpp

@@ -136,14 +136,21 @@ void RFBridgeComponent::loop() {
     this->last_bridge_byte_ = now;
   }
 
-  while (this->available()) {
-    uint8_t byte;
-    this->read_byte(&byte);
-    if (this->parse_bridge_byte_(byte)) {
-      ESP_LOGVV(TAG, "Parsed: 0x%02X", byte);
-      this->last_bridge_byte_ = now;
-    } else {
-      this->rx_buffer_.clear();
+  int avail = this->available();
+  while (avail > 0) {
+    uint8_t buf[64];
+    size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+    if (!this->read_array(buf, to_read)) {
+      break;
+    }
+    avail -= to_read;
+    for (size_t i = 0; i < to_read; i++) {
+      if (this->parse_bridge_byte_(buf[i])) {
+        ESP_LOGVV(TAG, "Parsed: 0x%02X", buf[i]);
+        this->last_bridge_byte_ = now;
+      } else {
+        this->rx_buffer_.clear();
+      }
     }
   }
 }

esphome/components/seeed_mr24hpc1/seeed_mr24hpc1.cpp

@@ -106,12 +106,19 @@ void MR24HPC1Component::update_() {
 
 // main loop
 void MR24HPC1Component::loop() {
-  uint8_t byte;
+  // Read all available bytes in batches to reduce UART call overhead.
+  int avail = this->available();
+  uint8_t buf[64];
+  while (avail > 0) {
+    size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+    if (!this->read_array(buf, to_read)) {
+      break;
+    }
+    avail -= to_read;
 
-  // Is there data on the serial port
-  while (this->available()) {
-    this->read_byte(&byte);
-    this->r24_split_data_frame_(byte);  // split data frame
+    for (size_t i = 0; i < to_read; i++) {
+      this->r24_split_data_frame_(buf[i]);  // split data frame
+    }
   }
 
   if ((this->s_output_info_switch_flag_ == OUTPUT_SWTICH_OFF) &&

esphome/components/seeed_mr60bha2/seeed_mr60bha2.cpp

@@ -30,14 +30,21 @@ void MR60BHA2Component::dump_config() {
 
 // main loop
 void MR60BHA2Component::loop() {
-  uint8_t byte;
+  // Read all available bytes in batches to reduce UART call overhead.
+  int avail = this->available();
+  uint8_t buf[64];
+  while (avail > 0) {
+    size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+    if (!this->read_array(buf, to_read)) {
+      break;
+    }
+    avail -= to_read;
 
-  // Is there data on the serial port
-  while (this->available()) {
-    this->read_byte(&byte);
-    this->rx_message_.push_back(byte);
-    if (!this->validate_message_()) {
-      this->rx_message_.clear();
+    for (size_t i = 0; i < to_read; i++) {
+      this->rx_message_.push_back(buf[i]);
+      if (!this->validate_message_()) {
+        this->rx_message_.clear();
+      }
     }
   }
 }

esphome/components/seeed_mr60fda2/seeed_mr60fda2.cpp

@@ -49,12 +49,19 @@ void MR60FDA2Component::setup() {
 
 // main loop
 void MR60FDA2Component::loop() {
-  uint8_t byte;
+  // Read all available bytes in batches to reduce UART call overhead.
+  int avail = this->available();
+  uint8_t buf[64];
+  while (avail > 0) {
+    size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+    if (!this->read_array(buf, to_read)) {
+      break;
+    }
+    avail -= to_read;
 
-  // Is there data on the serial port
-  while (this->available()) {
-    this->read_byte(&byte);
-    this->split_frame_(byte);  // split data frame
+    for (size_t i = 0; i < to_read; i++) {
+      this->split_frame_(buf[i]);  // split data frame
+    }
   }
 }
 

esphome/components/tuya/tuya.cpp

@@ -31,10 +31,19 @@ void Tuya::setup() {
 }
 
 void Tuya::loop() {
-  while (this->available()) {
-    uint8_t c;
-    this->read_byte(&c);
-    this->handle_char_(c);
+  // Read all available bytes in batches to reduce UART call overhead.
+  int avail = this->available();
+  uint8_t buf[64];
+  while (avail > 0) {
+    size_t to_read = std::min(static_cast<size_t>(avail), sizeof(buf));
+    if (!this->read_array(buf, to_read)) {
+      break;
+    }
+    avail -= to_read;
+
+    for (size_t i = 0; i < to_read; i++) {
+      this->handle_char_(buf[i]);
+    }
   }
   process_command_queue_();
 }