[audio] Add a RingBufferAudioSource (#16314)

esphome/components/audio/audio_transfer_buffer.cpp

@@ -207,6 +207,137 @@ void ConstAudioSourceBuffer::consume(size_t bytes) {
   this->data_start_ += bytes;
 }
 
+std::unique_ptr<RingBufferAudioSource> RingBufferAudioSource::create(
+    std::shared_ptr<ring_buffer::RingBuffer> ring_buffer, size_t max_fill_bytes, uint8_t alignment_bytes) {
+  if (ring_buffer == nullptr || max_fill_bytes == 0 || alignment_bytes == 0 || alignment_bytes > MAX_ALIGNMENT_BYTES) {
+    return nullptr;
+  }
+  return std::unique_ptr<RingBufferAudioSource>(
+      new RingBufferAudioSource(std::move(ring_buffer), max_fill_bytes, alignment_bytes));
+}
+
+RingBufferAudioSource::~RingBufferAudioSource() {
+  if (this->acquired_item_ != nullptr) {
+    this->ring_buffer_->receive_release(this->acquired_item_);
+    this->acquired_item_ = nullptr;
+  }
+}
+
+void RingBufferAudioSource::release_item_() {
+  if (this->acquired_item_ == nullptr) {
+    return;
+  }
+  if (this->item_trailing_length_ > 0) {
+    // Copy the trailing sub-frame bytes into the splice buffer before returning the item; the next
+    // fill() will complete the frame from the head of the next chunk.
+    std::memcpy(this->splice_buffer_, this->item_trailing_ptr_, this->item_trailing_length_);
+    this->splice_length_ = this->item_trailing_length_;
+    this->item_trailing_ptr_ = nullptr;
+    this->item_trailing_length_ = 0;
+  }
+  this->ring_buffer_->receive_release(this->acquired_item_);
+  this->acquired_item_ = nullptr;
+}
+
+void RingBufferAudioSource::consume(size_t bytes) {
+  bytes = std::min(bytes, this->current_available_);
+  this->current_data_ += bytes;
+  this->current_available_ -= bytes;
+  // Promotion of queued data is deferred to fill() so callers see new data as a fresh return value
+  // rather than appearing silently after consume(). When the held item has nothing left depending
+  // on it (no exposed bytes and no queued region), release it now so the ring buffer can be
+  // reclaimed by writers even if fill() is never called again.
+  if (this->current_available_ == 0 && this->queued_length_ == 0) {
+    this->release_item_();
+  }
+}
+
+bool RingBufferAudioSource::has_buffered_data() const {
+  // splice_length_ is deliberately not considered here. It holds an incomplete frame whose completion
+  // bytes must still arrive through the ring buffer, which ring_buffer_->available() already reports.
+  // Counting it separately would strand a drain loop when a stream ends mid-frame and those completion
+  // bytes never come.
+  return (this->current_available_ > 0) || (this->queued_length_ > 0) || (this->ring_buffer_->available() > 0);
+}
+
+size_t RingBufferAudioSource::fill(TickType_t ticks_to_wait, bool /*pre_shift*/) {
+  if (this->current_available_ > 0) {
+    // Caller has not finished consuming the current exposure
+    return 0;
+  }
+
+  // If a queued region (the aligned remainder of the new chunk after a splice frame) is waiting,
+  // promote it to the exposed region and report its size as fresh data.
+  if (this->queued_length_ > 0) {
+    this->current_data_ = this->queued_data_;
+    this->current_available_ = this->queued_length_;
+    this->queued_data_ = nullptr;
+    this->queued_length_ = 0;
+    return this->current_available_;
+  }
+
+  // Nothing exposed and nothing queued: release the previously held item (saving any sub-frame tail
+  // to splice_buffer_) and acquire a new chunk.
+  this->release_item_();
+
+  size_t chunk_length = 0;
+  void *item = this->ring_buffer_->receive_acquire(chunk_length, this->max_fill_bytes_, ticks_to_wait);
+  if (item == nullptr) {
+    return 0;
+  }
+
+  uint8_t *chunk_data = static_cast<uint8_t *>(item);
+  bool exposing_splice_frame = false;
+
+  // Complete any pending splice frame from the head of the new chunk.
+  if (this->splice_length_ > 0) {
+    const size_t needed = static_cast<size_t>(this->alignment_bytes_) - this->splice_length_;
+    if (chunk_length < needed) {
+      // Not enough data to complete the spliced frame yet; absorb everything and wait for more.
+      std::memcpy(this->splice_buffer_ + this->splice_length_, chunk_data, chunk_length);
+      this->splice_length_ += chunk_length;
+      this->ring_buffer_->receive_release(item);
+      return 0;
+    }
+    std::memcpy(this->splice_buffer_ + this->splice_length_, chunk_data, needed);
+    chunk_data += needed;
+    chunk_length -= needed;
+    this->splice_length_ = 0;
+    exposing_splice_frame = true;
+  }
+
+  this->acquired_item_ = item;
+
+  // Split the remaining chunk into its aligned region and a (possibly zero) sub-frame trailing tail.
+  const size_t trailing = (this->alignment_bytes_ > 1) ? (chunk_length % this->alignment_bytes_) : 0;
+  const size_t aligned_bytes = chunk_length - trailing;
+  if (trailing > 0) {
+    this->item_trailing_ptr_ = chunk_data + aligned_bytes;
+    this->item_trailing_length_ = trailing;
+  }
+
+  if (exposing_splice_frame) {
+    // Expose the spliced frame from splice_buffer_, queuing the chunk's aligned region for the next
+    // fill() call.
+    this->current_data_ = this->splice_buffer_;
+    this->current_available_ = this->alignment_bytes_;
+    this->queued_data_ = chunk_data;
+    this->queued_length_ = aligned_bytes;
+    return this->alignment_bytes_;
+  }
+
+  if (aligned_bytes == 0) {
+    // The entire chunk is a sub-frame tail (only possible when alignment exceeds chunk size). Save it
+    // to the splice buffer and release the item so the next fill() can complete the frame.
+    this->release_item_();
+    return 0;
+  }
+
+  this->current_data_ = chunk_data;
+  this->current_available_ = aligned_bytes;
+  return aligned_bytes;
+}
+
 }  // namespace esphome::audio
 
 #endif