Move microphone to use rtos task to read

Use afe_fr for noise suppression (not working)

esphome/components/esp_adf/esp_adf.h

@@ -7,6 +7,33 @@
 
 namespace esphome {
 namespace esp_adf {
+
+static const size_t BUFFER_SIZE = 1024;
+
+enum class TaskEventType : uint8_t {
+  STARTING = 0,
+  STARTED,
+  RUNNING,
+  STOPPING,
+  STOPPED,
+  WARNING = 255,
+};
+
+struct TaskEvent {
+  TaskEventType type;
+  esp_err_t err;
+};
+
+struct CommandEvent {
+  bool stop;
+};
+
+struct DataEvent {
+  bool stop;
+  size_t len;
+  uint8_t data[BUFFER_SIZE];
+};
+
 class ESPADF;
 
 class ESPADFPipeline : public Parented<ESPADF> {};

esphome/components/esp_adf/microphone/esp_adf_microphone.cpp

@@ -11,14 +11,56 @@
 #include <filter_resample.h>
 #include <i2s_stream.h>
 #include <raw_stream.h>
+#include <recorder_sr.h>
 
 namespace esphome {
 namespace esp_adf {
 
-static const size_t BUFFER_SIZE = 1024;
-
 static const char *const TAG = "esp_adf.microphone";
 
+void ESPADFMicrophone::setup() {
+  this->ring_buffer_ = rb_create(BUFFER_SIZE, sizeof(int16_t));
+  if (this->ring_buffer_ == nullptr) {
+    ESP_LOGW(TAG, "Could not allocate ring buffer.");
+    this->mark_failed();
+    return;
+  }
+  this->feed_event_queue_ = xQueueCreate(20, sizeof(TaskEvent));
+  this->feed_command_queue_ = xQueueCreate(20, sizeof(CommandEvent));
+  this->fetch_command_queue_ = xQueueCreate(20, sizeof(CommandEvent));
+
+  afe_config_t cfg_afe = {
+      .aec_init = false,
+      .se_init = true,
+      .vad_init = true,
+      .wakenet_init = false,
+      .voice_communication_init = false,
+      .voice_communication_agc_init = false,
+      .voice_communication_agc_gain = 15,
+      .vad_mode = VAD_MODE_3,
+      .wakenet_model_name = nullptr,
+      .wakenet_mode = DET_MODE_2CH_90,
+      .afe_mode = SR_MODE_HIGH_PERF,
+      .afe_perferred_core = 1,
+      .afe_perferred_priority = 5,
+      .afe_ringbuf_size = 50,
+      .memory_alloc_mode = AFE_MEMORY_ALLOC_MORE_PSRAM,
+      .agc_mode = AFE_MN_PEAK_AGC_MODE_3,
+      .pcm_config =
+          {
+              .total_ch_num = 1,
+              .mic_num = 1,
+              .ref_num = 0,
+              .sample_rate = 16000,
+          },
+      .debug_init = false,
+      .debug_hook = {{AFE_DEBUG_HOOK_MASE_TASK_IN, NULL}, {AFE_DEBUG_HOOK_FETCH_TASK_IN, NULL}},
+  };
+
+  this->afe_data_ = this->afe_handle_->create_from_config(&cfg_afe);
+  this->afe_chunk_size_ = this->afe_handle_->get_feed_chunksize(this->afe_data_);
+}
+
 void ESPADFMicrophone::start() {
   if (this->is_failed())
     return;
@@ -32,6 +74,33 @@ void ESPADFMicrophone::start_() {
   if (!this->parent_->try_lock()) {
     return;
   }
+  this->state_ = microphone::STATE_RUNNING;
+
+  xTaskCreate(ESPADFMicrophone::feed_task, "feed_task", 8192, (void *) this, 0, &this->feed_task_handle_);
+  xTaskCreate(ESPADFMicrophone::fetch_task, "fetch_task", 8192, (void *) this, 0, &this->fetch_task_handle_);
+}
+
+void ESPADFMicrophone::feed_task(void *params) {
+  ESPADFMicrophone *this_mic = (ESPADFMicrophone *) params;
+  TaskEvent event;
+
+  event.type = TaskEventType::STARTING;
+  xQueueSend(this_mic->feed_event_queue_, &event, portMAX_DELAY);
+
+  size_t buffer_size = this_mic->afe_chunk_size_ * sizeof(int16_t);
+
+  ExternalRAMAllocator<int16_t> allocator(ExternalRAMAllocator<int16_t>::ALLOW_FAILURE);
+  int16_t *afe_buffer = allocator.allocate(this_mic->afe_chunk_size_);
+  if (afe_buffer == nullptr) {
+    event.type = TaskEventType::STOPPED;
+    xQueueSend(this_mic->feed_event_queue_, &event, portMAX_DELAY);
+
+    while (true) {
+      delay(10);
+    };
+    return;
+  }
+
   i2s_driver_config_t i2s_config = {
       .mode = (i2s_mode_t) (I2S_MODE_MASTER | I2S_MODE_RX),
       .sample_rate = 44100,
@@ -48,13 +117,11 @@ void ESPADFMicrophone::start_() {
       .bits_per_chan = I2S_BITS_PER_CHAN_DEFAULT,
   };
 
-  ESP_LOGI(TAG, "Init pipeline");
   audio_pipeline_cfg_t pipeline_cfg = {
       .rb_size = 8 * 1024,
   };
-  this->pipeline_ = audio_pipeline_init(&pipeline_cfg);
+  audio_pipeline_handle_t pipeline = audio_pipeline_init(&pipeline_cfg);
 
-  ESP_LOGI(TAG, "Init i2s stream");
   i2s_stream_cfg_t i2s_cfg = {
       .type = AUDIO_STREAM_READER,
       .i2s_config = i2s_config,
@@ -71,9 +138,8 @@ void ESPADFMicrophone::start_() {
       .need_expand = false,
       .expand_src_bits = I2S_BITS_PER_SAMPLE_16BIT,
   };
-  this->i2s_stream_reader_ = i2s_stream_init(&i2s_cfg);
+  audio_element_handle_t i2s_stream_reader = i2s_stream_init(&i2s_cfg);
 
-  ESP_LOGI(TAG, "Init filter");
   rsp_filter_cfg_t rsp_cfg = {
       .src_rate = 44100,
       .src_ch = 2,
@@ -94,27 +160,115 @@ void ESPADFMicrophone::start_() {
       .task_prio = RSP_FILTER_TASK_PRIO,
       .stack_in_ext = true,
   };
-  this->filter_ = rsp_filter_init(&rsp_cfg);
+  audio_element_handle_t filter = rsp_filter_init(&rsp_cfg);
 
-  ESP_LOGI(TAG, "Init raw stream");
   raw_stream_cfg_t raw_cfg = {
       .type = AUDIO_STREAM_READER,
       .out_rb_size = 8 * 1024,
   };
-  this->raw_read_ = raw_stream_init(&raw_cfg);
+  audio_element_handle_t raw_read = raw_stream_init(&raw_cfg);
 
-  ESP_LOGI(TAG, "Register all elements to audio pipeline");
+  audio_pipeline_register(pipeline, i2s_stream_reader, "i2s");
-  audio_pipeline_register(this->pipeline_, this->i2s_stream_reader_, "i2s");
+  audio_pipeline_register(pipeline, filter, "filter");
-  audio_pipeline_register(this->pipeline_, this->filter_, "filter");
+  audio_pipeline_register(pipeline, raw_read, "raw");
-  audio_pipeline_register(this->pipeline_, this->raw_read_, "raw");
 
   const char *link_tag[3] = {"i2s", "filter", "raw"};
-  audio_pipeline_link(this->pipeline_, &link_tag[0], 3);
+  audio_pipeline_link(pipeline, &link_tag[0], 3);
 
-  ESP_LOGI(TAG, "Starting pipeline");
+  audio_pipeline_run(pipeline);
-  audio_pipeline_run(this->pipeline_);
 
-  this->state_ = microphone::STATE_RUNNING;
+  event.type = TaskEventType::STARTED;
+  xQueueSend(this_mic->feed_event_queue_, &event, portMAX_DELAY);
+
+  CommandEvent command_event;
+  size_t fill_count = 0;
+
+  while (true) {
+    if (xQueueReceive(this_mic->feed_command_queue_, &command_event, 0) == pdTRUE) {
+      if (command_event.stop) {
+        // Stop signal from main thread
+        break;
+      }
+    }
+
+    int bytes_read = raw_stream_read(raw_read, (char *) (afe_buffer + fill_count), buffer_size - fill_count);
+
+    if (bytes_read == -2 || bytes_read == 0) {
+      // No data in buffers to read.
+      continue;
+    } else if (bytes_read < 0) {
+      event.type = TaskEventType::WARNING;
+      event.err = bytes_read;
+      xQueueSend(this_mic->feed_event_queue_, &event, 0);
+      continue;
+    }
+
+    event.type = TaskEventType::RUNNING;
+    event.err = ESP_OK;
+    xQueueSend(this_mic->feed_event_queue_, &event, 0);
+
+    fill_count += bytes_read;
+
+    if (fill_count == buffer_size) {
+      this_mic->afe_handle_->feed(this_mic->afe_data_, afe_buffer);
+      fill_count -= buffer_size;
+    }
+  }
+
+  allocator.deallocate(afe_buffer, this_mic->afe_chunk_size_);
+
+  audio_pipeline_stop(pipeline);
+  audio_pipeline_wait_for_stop(pipeline);
+  audio_pipeline_terminate(pipeline);
+
+  event.type = TaskEventType::STOPPING;
+  xQueueSend(this_mic->feed_event_queue_, &event, portMAX_DELAY);
+
+  audio_pipeline_unregister(pipeline, i2s_stream_reader);
+  audio_pipeline_unregister(pipeline, filter);
+  audio_pipeline_unregister(pipeline, raw_read);
+
+  audio_pipeline_deinit(pipeline);
+  audio_element_deinit(i2s_stream_reader);
+  audio_element_deinit(filter);
+  audio_element_deinit(raw_read);
+
+  event.type = TaskEventType::STOPPED;
+  xQueueSend(this_mic->feed_event_queue_, &event, portMAX_DELAY);
+
+  while (true) {
+    delay(10);
+  }
+}
+
+void ESPADFMicrophone::fetch_task(void *params) {
+  ESPADFMicrophone *this_mic = (ESPADFMicrophone *) params;
+
+  CommandEvent event;
+
+  while (true) {
+    if (xQueueReceive(this_mic->fetch_command_queue_, &event, 0) == pdTRUE) {
+      if (event.stop) {
+        // Stop signal from main thread
+        break;
+      }
+    }
+    afe_fetch_result_t *result = this_mic->afe_handle_->fetch(this_mic->afe_data_);
+
+    if (result == nullptr) {
+      continue;
+    }
+
+    int available = rb_bytes_available(this_mic->ring_buffer_);
+    if (available < result->data_size) {
+      rb_read(this_mic->ring_buffer_, nullptr, result->data_size - available, 0);
+    }
+    rb_write(this_mic->ring_buffer_, (char *) result->data, result->data_size, 0);
+  }
+
+  while (true) {
+    delay(10);
+  }
 }
 
 void ESPADFMicrophone::stop() {
@@ -129,26 +283,14 @@ void ESPADFMicrophone::stop() {
 
 void ESPADFMicrophone::stop_() {
   ESP_LOGD(TAG, "Stopping microphone");
-  audio_pipeline_stop(this->pipeline_);
+  CommandEvent command_event;
-  audio_pipeline_wait_for_stop(this->pipeline_);
+  command_event.stop = true;
-  audio_pipeline_terminate(this->pipeline_);
+  xQueueSendToFront(this->feed_command_queue_, &command_event, portMAX_DELAY);
-
+  xQueueSendToFront(this->fetch_command_queue_, &command_event, portMAX_DELAY);
-  audio_pipeline_unregister(this->pipeline_, this->i2s_stream_reader_);
-  audio_pipeline_unregister(this->pipeline_, this->filter_);
-  audio_pipeline_unregister(this->pipeline_, this->raw_read_);
-
-  audio_pipeline_deinit(this->pipeline_);
-  audio_element_deinit(this->i2s_stream_reader_);
-  audio_element_deinit(this->filter_);
-  audio_element_deinit(this->raw_read_);
-
-  this->parent_->unlock();
-  this->state_ = microphone::STATE_STOPPED;
-  ESP_LOGD(TAG, "Microphone stopped");
 }
 
 size_t ESPADFMicrophone::read(int16_t *buf, size_t len) {
-  int bytes_read = raw_stream_read(this->raw_read_, (char *) buf, len);
+  int bytes_read = rb_read(this->ring_buffer_, (char *) buf, len, 0);
 
   if (bytes_read == -2 || bytes_read == 0) {
     // No data in buffers to read.
@@ -171,6 +313,34 @@ void ESPADFMicrophone::read_() {
   this->data_callbacks_.call(samples);
 }
 
+void ESPADFMicrophone::watch_() {
+  TaskEvent event;
+  if (xQueueReceive(this->feed_event_queue_, &event, 0) == pdTRUE) {
+    switch (event.type) {
+      case TaskEventType::STARTING:
+      case TaskEventType::STARTED:
+      case TaskEventType::STOPPING:
+        break;
+      case TaskEventType::RUNNING:
+        this->status_clear_warning();
+        break;
+      case TaskEventType::STOPPED:
+        this->parent_->unlock();
+        this->state_ = microphone::STATE_STOPPED;
+        vTaskDelete(this->feed_task_handle_);
+        vTaskDelete(this->fetch_task_handle_);
+        this->feed_task_handle_ = nullptr;
+        this->fetch_task_handle_ = nullptr;
+        ESP_LOGD(TAG, "Microphone stopped");
+        break;
+      case TaskEventType::WARNING:
+        ESP_LOGW(TAG, "Error writing to pipeline: %s", esp_err_to_name(event.err));
+        this->status_set_warning();
+        break;
+    }
+  }
+}
+
 void ESPADFMicrophone::loop() {
   switch (this->state_) {
     case microphone::STATE_STOPPED:

esphome/components/esp_adf/microphone/esp_adf_microphone.h

@@ -9,12 +9,16 @@
 
 #include <audio_element.h>
 #include <audio_pipeline.h>
+#include <esp_afe_sr_iface.h>
+#include <esp_afe_sr_models.h>
+#include <ringbuf.h>
 
 namespace esphome {
 namespace esp_adf {
 
 class ESPADFMicrophone : public ESPADFPipeline, public microphone::Microphone, public Component {
  public:
+  void setup() override;
   void start() override;
   void stop() override;
 
@@ -26,9 +30,23 @@ class ESPADFMicrophone : public ESPADFPipeline, public microphone::Microphone, p
   void start_();
   void stop_();
   void read_();
+  void watch_();
 
-  audio_pipeline_handle_t pipeline_;
+  static void feed_task(void *params);
-  audio_element_handle_t i2s_stream_reader_, filter_, raw_read_;
+  static void fetch_task(void *params);
+
+  const esp_afe_sr_iface_t *afe_handle_{&ESP_AFE_SR_HANDLE};
+  esp_afe_sr_data_t *afe_data_{nullptr};
+  size_t afe_chunk_size_{0};
+
+  ringbuf_handle_t ring_buffer_;
+
+  TaskHandle_t feed_task_handle_{nullptr};
+  QueueHandle_t feed_event_queue_;
+  QueueHandle_t feed_command_queue_;
+
+  TaskHandle_t fetch_task_handle_{nullptr};
+  QueueHandle_t fetch_command_queue_;
 };
 
 }  // namespace esp_adf

esphome/components/esp_adf/speaker/esp_adf_speaker.cpp

@@ -165,7 +165,7 @@ void ESPADFSpeaker::player_task(void *params) {
       current += bytes_written;
     }
 
-    event.type = TaskEventType::PLAYING;
+    event.type = TaskEventType::RUNNING;
     xQueueSend(this_speaker->event_queue_, &event, 0);
   }
 
@@ -214,7 +214,7 @@ void ESPADFSpeaker::watch_() {
       case TaskEventType::STARTED:
       case TaskEventType::STOPPING:
         break;
-      case TaskEventType::PLAYING:
+      case TaskEventType::RUNNING:
         this->status_clear_warning();
         break;
       case TaskEventType::STOPPED:

esphome/components/esp_adf/speaker/esp_adf_speaker.h

@@ -17,28 +17,6 @@
 namespace esphome {
 namespace esp_adf {
 
-static const size_t BUFFER_SIZE = 1024;
-
-enum class TaskEventType : uint8_t {
-  STARTING = 0,
-  STARTED,
-  PLAYING,
-  STOPPING,
-  STOPPED,
-  WARNING = 255,
-};
-
-struct TaskEvent {
-  TaskEventType type;
-  esp_err_t err;
-};
-
-struct DataEvent {
-  bool stop;
-  size_t len;
-  uint8_t data[BUFFER_SIZE];
-};
-
 class ESPADFSpeaker : public ESPADFPipeline, public speaker::Speaker, public Component {
  public:
   float get_setup_priority() const override { return esphome::setup_priority::LATE; }