[i2s_audio] Publish an unknown capture time when it cannot be trusted

Dropped buffers are now counted instead of subtracted, so a drop between the read and the snapshot can be detected; that read gets capture time 0, as does every read when the callbacks failed to register.

esphome/components/i2s_audio/i2s_audio.cpp

@@ -20,6 +20,7 @@ void I2SAudioComponent::setup() {
   this->audio_out_->build_full_duplex_config(chan_cfg, tx_cfg);
   this->audio_in_->build_full_duplex_config(rx_cfg);
 
+  this->dma_desc_num_ = chan_cfg.dma_desc_num;
   esp_err_t err = i2s_new_channel(&chan_cfg, &this->tx_handle_, &this->rx_handle_);
   if (err == ESP_OK) {
     // The first initialized channel drives the shared clocks. Use RX so the microphone runs on its own.
@@ -79,6 +80,9 @@ bool I2SAudioComponent::update_rx_channel_() {
   if (needed == this->rx_enabled_)
     return true;
 
+  if (needed) {
+    this->audio_in_->full_duplex_rx_enabling(this->rx_handle_, this->dma_desc_num_);
+  }
   esp_err_t err = needed ? i2s_channel_enable(this->rx_handle_) : i2s_channel_disable(this->rx_handle_);
   if (err != ESP_OK) {
     ESP_LOGE(TAG, "Failed to %s RX channel: %s", needed ? LOG_STR_LITERAL("enable") : LOG_STR_LITERAL("disable"),

esphome/components/i2s_audio/i2s_audio.h

@@ -46,6 +46,9 @@ class I2SAudioIn : public I2SAudioBase {
  public:
   /// @brief Builds the RX configuration the parent uses to set up a full duplex channel pair.
   virtual void build_full_duplex_config(i2s_std_config_t &std_cfg) = 0;
+  /// @brief Called by the parent on the main loop just before it enables the full duplex RX channel, which clears
+  /// its receive queue. The channel is still disabled, so event callbacks can be registered on it.
+  virtual void full_duplex_rx_enabling(i2s_chan_handle_t rx_handle, uint32_t dma_desc_num) = 0;
 #endif
 };
 
@@ -126,6 +129,7 @@ class I2SAudioComponent final : public Component {
   I2SAudioOut *audio_out_{nullptr};
   i2s_chan_handle_t rx_handle_{nullptr};
   i2s_chan_handle_t tx_handle_{nullptr};
+  uint32_t dma_desc_num_{0};
   bool rx_in_use_{false};
   bool tx_in_use_{false};
   bool rx_enabled_{false};

esphome/components/i2s_audio/microphone/i2s_audio_microphone.cpp

@@ -4,18 +4,24 @@
 
 #include <driver/i2s_std.h>
 #include <driver/i2s_pdm.h>
+#include <esp_timer.h>
 
 #include "esphome/core/hal.h"
 #include "esphome/core/log.h"
 
 #include "esphome/components/audio/audio.h"
 
+#include <algorithm>
+
 namespace esphome::i2s_audio {
 
 static const UBaseType_t MAX_LISTENERS = 16;
 
 static const uint32_t READ_DURATION_MS = 16;
 
+static constexpr uint32_t DMA_BUFFERS_COUNT = 4;
+static constexpr uint32_t DMA_FRAME_NUM = 256;
+
 static const size_t TASK_STACK_SIZE = 4096;
 static const ssize_t TASK_PRIORITY = 23;
 
@@ -124,8 +130,43 @@ void I2SAudioMicrophone::build_std_config_(i2s_std_config_t &std_cfg) const {
 
 #ifdef USE_I2S_AUDIO_FULL_DUPLEX
 void I2SAudioMicrophone::build_full_duplex_config(i2s_std_config_t &std_cfg) { this->build_std_config_(std_cfg); }
+
+void I2SAudioMicrophone::full_duplex_rx_enabling(i2s_chan_handle_t rx_handle, uint32_t dma_desc_num) {
+  // A failure only loses capture times; the speaker may still need RX enabled, so it is not passed on
+  this->start_capture_count_(rx_handle, dma_desc_num);
+}
 #endif
 
+bool I2SAudioMicrophone::start_capture_count_(i2s_chan_handle_t rx_handle, uint32_t dma_desc_num) {
+  // Neither the interrupt nor the mic task runs while the channel is disabled
+  this->dma_bytes_received_ = 0;
+  this->dma_buffers_dropped_ = 0;
+  this->dma_receive_time_us_ = 0;
+  this->dma_bytes_read_ = 0;
+  this->dma_desc_num_ = dma_desc_num;
+  const i2s_event_callbacks_t callbacks = {
+      .on_recv = i2s_on_recv_cb,
+      .on_recv_q_ovf = i2s_on_recv_q_ovf_cb,
+  };
+  esp_err_t err = i2s_channel_register_event_callback(rx_handle, &callbacks, this);
+  if (err != ESP_OK) {
+    ESP_LOGE(TAG, "Registering callbacks failed: %s", esp_err_to_name(err));
+    return false;
+  }
+  return true;
+}
+
+bool I2SAudioMicrophone::read_dma_buffer_size_() {
+  i2s_chan_info_t chan_info{};
+  esp_err_t err = i2s_channel_get_info(this->rx_handle_, &chan_info);
+  if ((err != ESP_OK) || (chan_info.total_dma_buf_size == 0) || (this->dma_desc_num_ == 0)) {
+    ESP_LOGE(TAG, "Getting the DMA buffer size failed");
+    return false;
+  }
+  this->dma_buffer_bytes_ = chan_info.total_dma_buf_size / this->dma_desc_num_;
+  return true;
+}
+
 bool I2SAudioMicrophone::start_driver_() {
 #ifdef USE_I2S_AUDIO_FULL_DUPLEX
   if (this->parent_->is_full_duplex()) {
@@ -135,7 +176,7 @@ bool I2SAudioMicrophone::start_driver_() {
       return false;
     }
     this->configure_stream_settings_();
-    return true;
+    return this->read_dma_buffer_size_();
   }
 #endif
 
@@ -148,8 +189,8 @@ bool I2SAudioMicrophone::start_driver_() {
   i2s_chan_config_t chan_cfg = {
       .id = this->parent_->get_port(),
       .role = this->i2s_role_,
-      .dma_desc_num = 4,
-      .dma_frame_num = 256,
+      .dma_desc_num = DMA_BUFFERS_COUNT,
+      .dma_frame_num = DMA_FRAME_NUM,
       .auto_clear = false,
   };
   /* Allocate a new RX channel and get the handle of this channel */
@@ -210,6 +251,10 @@ bool I2SAudioMicrophone::start_driver_() {
     return false;
   }
 
+  if (!this->start_capture_count_(this->rx_handle_, DMA_BUFFERS_COUNT)) {
+    return false;
+  }
+
   /* Before reading data, start the RX channel first */
   err = i2s_channel_enable(this->rx_handle_);
   if (err != ESP_OK) {
@@ -219,7 +264,7 @@ bool I2SAudioMicrophone::start_driver_() {
 
   this->configure_stream_settings_();  // redetermine the settings in case some settings were changed after compilation
 
-  return true;
+  return this->read_dma_buffer_size_();
 }
 
 void I2SAudioMicrophone::stop() {
@@ -272,7 +317,16 @@ void I2SAudioMicrophone::mic_task(void *params) {
 
     // Copied so the format cannot change while the task runs
     const audio::AudioStreamInfo stream_info = this_microphone->audio_stream_info_;
-    const size_t bytes_to_read = stream_info.ms_to_bytes(READ_DURATION_MS);
+
+    // Read whole DMA buffers. When its queue is nearly full, the driver discards the unread rest of a buffer it is
+    // partway through without reporting it, which would throw off the capture time count.
+    const size_t dma_buffer_bytes = this_microphone->dma_buffer_bytes_;
+    const size_t bytes_to_read =
+        std::max<size_t>(1, (stream_info.ms_to_bytes(READ_DURATION_MS) + dma_buffer_bytes / 2) / dma_buffer_bytes) *
+        dma_buffer_bytes;
+    const uint32_t read_duration_ms =
+        (stream_info.bytes_to_frames(bytes_to_read) * 1000 + stream_info.get_sample_rate() - 1) /
+        stream_info.get_sample_rate();
     std::vector<uint8_t> samples;
     samples.reserve(bytes_to_read);
 
@@ -280,13 +334,44 @@ void I2SAudioMicrophone::mic_task(void *params) {
 
     while (!(xEventGroupGetBits(this_microphone->event_group_) & MicrophoneEventGroupBits::COMMAND_STOP)) {
       if (this_microphone->has_data_callbacks_()) {
+        portENTER_CRITICAL(&this_microphone->dma_lock_);
+        const uint32_t drops_before_read = this_microphone->dma_buffers_dropped_;
+        portEXIT_CRITICAL(&this_microphone->dma_lock_);
+
         samples.resize(bytes_to_read);
-        size_t bytes_read = this_microphone->read_(samples.data(), bytes_to_read, 2 * READ_DURATION_MS);
+        size_t bytes_read = this_microphone->read_(samples.data(), bytes_to_read, 2 * read_duration_ms);
+        if (bytes_read == 0) {
+          continue;
+        }
+
+        portENTER_CRITICAL(&this_microphone->dma_lock_);
+        const uint32_t bytes_received = this_microphone->dma_bytes_received_;
+        const uint32_t drops = this_microphone->dma_buffers_dropped_;
+        const int64_t receive_time_us = this_microphone->dma_receive_time_us_;
+        portEXIT_CRITICAL(&this_microphone->dma_lock_);
+
+        // A drop between the two snapshots may have removed buffers newer than this read's first frame, which the count
+        // back would skip, so the capture time is unknown. It is also unknown if no buffer was recorded because the
+        // callbacks failed to register.
+        int64_t capture_time_us = 0;
+        if ((drops == drops_before_read) && (receive_time_us != 0)) {
+          // Count back from the newest DMA buffer to this read's first frame. At most dma_desc_num_ buffers follow the
+          // read: a full queue plus one the driver has recorded but not queued yet.
+          const uint32_t bytes_queued = bytes_received - drops * static_cast<uint32_t>(dma_buffer_bytes);
+          const uint32_t bytes_after_first =
+              std::min<uint32_t>(bytes_queued - this_microphone->dma_bytes_read_,
+                                 bytes_read + this_microphone->dma_desc_num_ * dma_buffer_bytes);
+          const uint32_t frames_after_first = stream_info.bytes_to_frames(bytes_after_first);
+          capture_time_us =
+              receive_time_us - static_cast<int64_t>(frames_after_first) * 1000000 / stream_info.get_sample_rate();
+        }
+        this_microphone->dma_bytes_read_ += bytes_read;
+
         samples.resize(bytes_read);
         if (this_microphone->correct_dc_offset_) {
           this_microphone->fix_dc_offset_(samples);
         }
-        this_microphone->publish_audio_(samples, stream_info);
+        this_microphone->publish_audio_(samples, stream_info, capture_time_us);
       } else {
         vTaskDelay(pdMS_TO_TICKS(READ_DURATION_MS));
       }
@@ -300,6 +385,25 @@ void I2SAudioMicrophone::mic_task(void *params) {
   }
 }
 
+bool IRAM_ATTR I2SAudioMicrophone::i2s_on_recv_cb(i2s_chan_handle_t handle, i2s_event_data_t *event, void *user_ctx) {
+  const int64_t now = esp_timer_get_time();
+  auto *this_microphone = static_cast<I2SAudioMicrophone *>(user_ctx);
+  portENTER_CRITICAL_ISR(&this_microphone->dma_lock_);
+  this_microphone->dma_bytes_received_ += event->size;
+  this_microphone->dma_receive_time_us_ = now;
+  portEXIT_CRITICAL_ISR(&this_microphone->dma_lock_);
+  return false;
+}
+
+bool IRAM_ATTR I2SAudioMicrophone::i2s_on_recv_q_ovf_cb(i2s_chan_handle_t handle, i2s_event_data_t *event,
+                                                        void *user_ctx) {
+  auto *this_microphone = static_cast<I2SAudioMicrophone *>(user_ctx);
+  portENTER_CRITICAL_ISR(&this_microphone->dma_lock_);
+  ++this_microphone->dma_buffers_dropped_;
+  portEXIT_CRITICAL_ISR(&this_microphone->dma_lock_);
+  return false;
+}
+
 void I2SAudioMicrophone::fix_dc_offset_(std::vector<uint8_t> &data) {
   /**
    * From https://www.musicdsp.org/en/latest/Filters/135-dc-filter.html:
@@ -364,13 +468,12 @@ size_t I2SAudioMicrophone::read_(uint8_t *buf, size_t len, uint32_t timeout_ms)
       ESP_LOGW(TAG, "Read error: %s", esp_err_to_name(err));
     }
     this->status_set_warning();
-    return 0;
-  }
-  if ((bytes_read == 0) && (timeout_ms > 0)) {
+  } else if ((bytes_read == 0) && (timeout_ms > 0)) {
     this->status_set_warning();
-    return 0;
+  } else {
+    this->status_clear_warning();
   }
-  this->status_clear_warning();
+  // Even after an error, any bytes read have left the DMA buffers and are valid audio
 #ifdef USE_ESP32_VARIANT_ESP32
   // For ESP32 16-bit standard mono mode, adjacent samples need to be swapped.
   if (this->slot_mode_ == I2S_SLOT_MODE_MONO && this->slot_bit_width_ == I2S_SLOT_BIT_WIDTH_16BIT && !this->pdm_) {

esphome/components/i2s_audio/microphone/i2s_audio_microphone.h

@@ -35,6 +35,7 @@ class I2SAudioMicrophone final : public I2SAudioIn, public microphone::Microphon
 
 #ifdef USE_I2S_AUDIO_FULL_DUPLEX
   void build_full_duplex_config(i2s_std_config_t &std_cfg) override;
+  void full_duplex_rx_enabling(i2s_chan_handle_t rx_handle, uint32_t dma_desc_num) override;
 #endif
 
  protected:
@@ -48,6 +49,14 @@ class I2SAudioMicrophone final : public I2SAudioIn, public microphone::Microphon
   /// @brief Stops the I2S driver.
   void stop_driver_();
 
+  /// @brief Registers the DMA callbacks on a disabled RX channel and restarts the capture time counts.
+  /// @return True if successful, false otherwise
+  bool start_capture_count_(i2s_chan_handle_t rx_handle, uint32_t dma_desc_num);
+
+  /// @brief Reads the size of one DMA buffer from a started RX channel.
+  /// @return True if successful, false otherwise
+  bool read_dma_buffer_size_();
+
   /// @brief Attempts to correct a microphone DC offset; e.g., a microphones silent level is offset from 0. Applies a
   /// correction offset that is updated using an exponential moving average for all samples away from 0.
   /// @param data
@@ -60,11 +69,32 @@ class I2SAudioMicrophone final : public I2SAudioIn, public microphone::Microphon
 
   static void mic_task(void *params);
 
+  /// @brief Records the byte count and time of each DMA buffer the driver queues for reading.
+  static bool i2s_on_recv_cb(i2s_chan_handle_t handle, i2s_event_data_t *event, void *user_ctx);
+  /// @brief Counts a buffer the driver dropped because the task fell behind; it will never be read.
+  static bool i2s_on_recv_q_ovf_cb(i2s_chan_handle_t handle, i2s_event_data_t *event, void *user_ctx);
+
   SemaphoreHandle_t active_listeners_semaphore_{nullptr};
   EventGroupHandle_t event_group_{nullptr};
 
   TaskHandle_t task_handle_{nullptr};
 
+  // Written by the DMA interrupt and read by the mic task, under dma_lock_. Since the channel was enabled: the bytes
+  // received and the buffers dropped from the driver's full queue (both wrapping), and the time the newest buffer was
+  // received, or 0 if none has been.
+  portMUX_TYPE dma_lock_ = portMUX_INITIALIZER_UNLOCKED;
+  uint32_t dma_bytes_received_{0};
+  uint32_t dma_buffers_dropped_{0};
+  int64_t dma_receive_time_us_{0};
+
+  // Bytes the mic task has read since the channel was enabled, wrapping. A full duplex channel can stay enabled
+  // across microphone sessions, so this outlives the task.
+  uint32_t dma_bytes_read_{0};
+
+  // DMA layout of the RX channel, set when the driver starts
+  size_t dma_buffer_bytes_{0};
+  uint32_t dma_desc_num_{0};
+
   gpio_num_t din_pin_{I2S_GPIO_UNUSED};
   i2s_chan_handle_t rx_handle_{nullptr};
 #if SOC_I2S_SUPPORTS_PDM_RX