Create RingBuffer for VoiceAssistant (#6102)

esphome/components/voice_assistant/voice_assistant.cpp

@@ -86,14 +86,14 @@ void VoiceAssistant::setup() {
 
 #ifdef USE_ESP_ADF
   this->vad_instance_ = vad_create(VAD_MODE_4);
+#endif
 
-  this->ring_buffer_ = rb_create(BUFFER_SIZE, sizeof(int16_t));
+  this->ring_buffer_ = RingBuffer::create(BUFFER_SIZE * sizeof(int16_t));
   if (this->ring_buffer_ == nullptr) {
     ESP_LOGW(TAG, "Could not allocate ring buffer");
     this->mark_failed();
     return;
   }
-#endif
 
   ExternalRAMAllocator<uint8_t> send_allocator(ExternalRAMAllocator<uint8_t>::ALLOW_FAILURE);
   this->send_buffer_ = send_allocator.allocate(SEND_BUFFER_SIZE);
@@ -112,14 +112,8 @@ int VoiceAssistant::read_microphone_() {
       memset(this->input_buffer_, 0, INPUT_BUFFER_SIZE * sizeof(int16_t));
       return 0;
     }
-#ifdef USE_ESP_ADF
     // Write audio into ring buffer
-    int available = rb_bytes_available(this->ring_buffer_);
-    if (available < bytes_read) {
-      rb_read(this->ring_buffer_, nullptr, bytes_read - available, 0);
-    }
-    rb_write(this->ring_buffer_, (char *) this->input_buffer_, bytes_read, 0);
-#endif
+    this->ring_buffer_->write((void *) this->input_buffer_, bytes_read);
   } else {
     ESP_LOGD(TAG, "microphone not running");
   }
@@ -141,9 +135,9 @@ void VoiceAssistant::loop() {
   switch (this->state_) {
     case State::IDLE: {
       if (this->continuous_ && this->desired_state_ == State::IDLE) {
+        this->ring_buffer_->reset();
 #ifdef USE_ESP_ADF
         if (this->use_wake_word_) {
-          rb_reset(this->ring_buffer_);
           this->set_state_(State::START_MICROPHONE, State::WAIT_FOR_VAD);
         } else
 #endif
@@ -236,19 +230,13 @@ void VoiceAssistant::loop() {
       break;  // State changed when udp server port received
     }
     case State::STREAMING_MICROPHONE: {
-      size_t bytes_read = this->read_microphone_();
-#ifdef USE_ESP_ADF
-      if (rb_bytes_filled(this->ring_buffer_) >= SEND_BUFFER_SIZE) {
-        rb_read(this->ring_buffer_, (char *) this->send_buffer_, SEND_BUFFER_SIZE, 0);
+      this->read_microphone_();
+      if (this->ring_buffer_->available() >= SEND_BUFFER_SIZE) {
+        this->ring_buffer_->read((void *) this->send_buffer_, SEND_BUFFER_SIZE, 0);
         this->socket_->sendto(this->send_buffer_, SEND_BUFFER_SIZE, 0, (struct sockaddr *) &this->dest_addr_,
                               sizeof(this->dest_addr_));
       }
-#else
-      if (bytes_read > 0) {
-        this->socket_->sendto(this->input_buffer_, bytes_read, 0, (struct sockaddr *) &this->dest_addr_,
-                              sizeof(this->dest_addr_));
-      }
-#endif
+
       break;
     }
     case State::STOP_MICROPHONE: {
@@ -473,9 +461,9 @@ void VoiceAssistant::request_start(bool continuous, bool silence_detection) {
   if (this->state_ == State::IDLE) {
     this->continuous_ = continuous;
     this->silence_detection_ = silence_detection;
+    this->ring_buffer_->reset();
 #ifdef USE_ESP_ADF
     if (this->use_wake_word_) {
-      rb_reset(this->ring_buffer_);
       this->set_state_(State::START_MICROPHONE, State::WAIT_FOR_VAD);
     } else
 #endif
@@ -618,9 +606,9 @@ void VoiceAssistant::on_event(const api::VoiceAssistantEventResponse &msg) {
     case api::enums::VOICE_ASSISTANT_RUN_END: {
       ESP_LOGD(TAG, "Assist Pipeline ended");
       if (this->state_ == State::STREAMING_MICROPHONE) {
+        this->ring_buffer_->reset();
 #ifdef USE_ESP_ADF
         if (this->use_wake_word_) {
-          rb_reset(this->ring_buffer_);
           // No need to stop the microphone since we didn't use the speaker
           this->set_state_(State::WAIT_FOR_VAD, State::WAITING_FOR_VAD);
         } else

esphome/components/voice_assistant/voice_assistant.h

@@ -7,6 +7,7 @@
 #include "esphome/core/automation.h"
 #include "esphome/core/component.h"
 #include "esphome/core/helpers.h"
+#include "esphome/core/ring_buffer.h"
 
 #include "esphome/components/api/api_connection.h"
 #include "esphome/components/api/api_pb2.h"
@@ -21,7 +22,6 @@
 
 #ifdef USE_ESP_ADF
 #include <esp_vad.h>
-#include <ringbuf.h>
 #endif
 
 namespace esphome {
@@ -177,10 +177,10 @@ class VoiceAssistant : public Component {
 
 #ifdef USE_ESP_ADF
   vad_handle_t vad_instance_;
-  ringbuf_handle_t ring_buffer_;
   uint8_t vad_threshold_{5};
   uint8_t vad_counter_{0};
 #endif
+  std::unique_ptr<RingBuffer> ring_buffer_;
 
   bool use_wake_word_;
   uint8_t noise_suppression_level_;

esphome/core/ring_buffer.cpp

@@ -0,0 +1,49 @@
+#include "ring_buffer.h"
+
+#include "esphome/core/helpers.h"
+#include "esphome/core/log.h"
+
+#ifdef USE_ESP32
+
+#include "helpers.h"
+
+namespace esphome {
+
+static const char *const TAG = "ring_buffer";
+
+std::unique_ptr<RingBuffer> RingBuffer::create(size_t len) {
+  std::unique_ptr<RingBuffer> rb = make_unique<RingBuffer>();
+
+  ExternalRAMAllocator<uint8_t> allocator(ExternalRAMAllocator<uint8_t>::ALLOW_FAILURE);
+  rb->storage_ = allocator.allocate(len);
+  if (rb->storage_ == nullptr) {
+    return nullptr;
+  }
+
+  rb->handle_ = xStreamBufferCreateStatic(len, 0, rb->storage_, &rb->structure_);
+  return rb;
+}
+
+size_t RingBuffer::read(void *data, size_t size, TickType_t ticks_to_wait) {
+  return xStreamBufferReceive(this->handle_, data, size, ticks_to_wait);
+}
+
+size_t RingBuffer::write(void *data, size_t len) {
+  size_t free = this->free();
+  if (free < len) {
+    size_t needed = len - free;
+    uint8_t discard[needed];
+    xStreamBufferReceive(this->handle_, discard, needed, 0);
+  }
+  return xStreamBufferSend(this->handle_, data, len, 0);
+}
+
+size_t RingBuffer::available() const { return xStreamBufferBytesAvailable(this->handle_); }
+
+size_t RingBuffer::free() const { return xStreamBufferSpacesAvailable(this->handle_); }
+
+BaseType_t RingBuffer::reset() { return xStreamBufferReset(this->handle_); }
+
+}  // namespace esphome
+
+#endif

esphome/core/ring_buffer.h

@@ -0,0 +1,34 @@
+#pragma once
+
+#ifdef USE_ESP32
+
+#include <freertos/FreeRTOS.h>
+#include <freertos/stream_buffer.h>
+
+#include <cinttypes>
+#include <memory>
+
+namespace esphome {
+
+class RingBuffer {
+ public:
+  size_t read(void *data, size_t size, TickType_t ticks_to_wait = 0);
+
+  size_t write(void *data, size_t len);
+
+  size_t available() const;
+  size_t free() const;
+
+  BaseType_t reset();
+
+  static std::unique_ptr<RingBuffer> create(size_t len);
+
+ protected:
+  StreamBufferHandle_t handle_;
+  StaticStreamBuffer_t structure_;
+  uint8_t *storage_;
+};
+
+}  // namespace esphome
+
+#endif