mostly single task pipeline

1 files changed, 68 insertions, 85 deletions

diff --git a/src/audio/audio_decoder.cpp b/src/audio/audio_decoder.cpp
index 9879b042..f8614478 100644
--- a/src/audio/audio_decoder.cpp
+++ b/src/audio/audio_decoder.cpp
@@ -2,13 +2,16 @@
 
 #include <string.h>
 
+#include <algorithm>
 #include <cstddef>
 #include <cstdint>
 #include <memory>
 
+#include "cbor/tinycbor/src/cborinternal_p.h"
 #include "freertos/FreeRTOS.h"
 
 #include "esp_heap_caps.h"
+#include "esp_log.h"
 #include "freertos/message_buffer.h"
 #include "freertos/portmacro.h"
 
@@ -21,126 +24,106 @@ namespace audio {
 
 static const char* kTag = "DEC";
 
-static const std::size_t kChunkSize = 1024;
-static const std::size_t kReadahead = 8;
-
 AudioDecoder::AudioDecoder()
     : IAudioElement(),
-      arena_(kChunkSize, kReadahead, MALLOC_CAP_SPIRAM),
       stream_info_({}),
       has_samples_to_send_(false),
       needs_more_input_(true) {}
 
 AudioDecoder::~AudioDecoder() {}
 
-auto AudioDecoder::HasUnprocessedInput() -> bool {
-  return !needs_more_input_ || has_samples_to_send_;
-}
-
-auto AudioDecoder::IsOverBuffered() -> bool {
-  return arena_.BlocksFree() == 0;
-}
-
-auto AudioDecoder::ProcessStreamInfo(const StreamInfo& info) -> void {
-  stream_info_ = info;
-
-  if (info.chunk_size) {
-    chunk_reader_.emplace(info.chunk_size.value());
-  } else {
-    ESP_LOGE(kTag, "no chunk size given");
-    return;
+auto AudioDecoder::ProcessStreamInfo(const StreamInfo& info) -> bool {
+  if (!std::holds_alternative<StreamInfo::Encoded>(info.data)) {
+    return false;
   }
+  const auto& encoded = std::get<StreamInfo::Encoded>(info.data);
 
   // Reuse the existing codec if we can. This will help with gapless playback,
   // since we can potentially just continue to decode as we were before,
   // without any setup overhead.
+  // TODO: use audio type from stream
   if (current_codec_ != nullptr &&
-      current_codec_->CanHandleFile(info.path.value_or(""))) {
+      current_codec_->CanHandleType(encoded.type)) {
     current_codec_->ResetForNewStream();
-    return;
+    return true;
   }
 
-  auto result = codecs::CreateCodecForFile(info.path.value_or(""));
+  // TODO: use audio type from stream
+  auto result = codecs::CreateCodecForType(encoded.type);
   if (result.has_value()) {
     current_codec_ = std::move(result.value());
   } else {
     ESP_LOGE(kTag, "no codec for this file");
-    return;
-  }
-
-  stream_info_ = info;
-  has_sent_stream_info_ = false;
-}
-
-auto AudioDecoder::ProcessChunk(const cpp::span<std::byte>& chunk) -> void {
-  if (current_codec_ == nullptr || !chunk_reader_) {
-    // Should never happen, but fail explicitly anyway.
-    ESP_LOGW(kTag, "received chunk without chunk size or codec");
-    return;
+    return false;
   }
 
-  ESP_LOGI(kTag, "received new chunk (size %u)", chunk.size());
-  current_codec_->SetInput(chunk_reader_->HandleNewData(chunk));
-  needs_more_input_ = false;
+  return true;
 }
 
-auto AudioDecoder::ProcessEndOfStream() -> void {
-  has_samples_to_send_ = false;
-  needs_more_input_ = true;
-  current_codec_.reset();
+auto AudioDecoder::Process(std::vector<Stream>* inputs, MutableStream* output)
+    -> void {
+  // We don't really expect multiple inputs, so just pick the first that
+  // contains data. If none of them contain data, then we can still flush
+  // pending samples.
+  auto input =
+      std::find_if(inputs->begin(), inputs->end(),
+                   [](const Stream& s) { return s.data.size_bytes() > 0; });
+
+  if (input != inputs->end()) {
+    const StreamInfo* info = input->info;
+    if (!stream_info_ || *stream_info_ != *info) {
+      // The input stream has changed! Immediately throw everything away and
+      // start from scratch.
+      // TODO: special case gapless playback? needs thought.
+      stream_info_ = *info;
+      has_samples_to_send_ = false;
+      has_set_stream_info_ = false;
+
+      ProcessStreamInfo(*info);
+    }
 
-  SendOrBufferEvent(std::unique_ptr<StreamEvent>(
-      StreamEvent::CreateEndOfStream(input_events_)));
-}
+    current_codec_->SetInput(input->data);
+  }
 
-auto AudioDecoder::Process() -> void {
-  if (has_samples_to_send_) {
-    // Writing samples is relatively quick (it's just a bunch of memcopy's), so
-    // do them all at once.
-    while (has_samples_to_send_ && !IsOverBuffered()) {
-      if (!has_sent_stream_info_) {
-        has_sent_stream_info_ = true;
+  while (true) {
+    if (has_samples_to_send_) {
+      if (!has_set_stream_info_) {
+        has_set_stream_info_ = true;
         auto format = current_codec_->GetOutputFormat();
-        stream_info_->bits_per_sample = format.bits_per_sample;
-        stream_info_->sample_rate = format.sample_rate_hz;
-        stream_info_->channels = format.num_channels;
-        stream_info_->chunk_size = kChunkSize;
-
-        auto event =
-            StreamEvent::CreateStreamInfo(input_events_, *stream_info_);
-        SendOrBufferEvent(std::unique_ptr<StreamEvent>(event));
-      }
-
-      auto block = arena_.Acquire();
-      if (!block) {
-        return;
+        output->info->data.emplace<StreamInfo::Pcm>(
+            format.bits_per_sample, format.sample_rate_hz, format.num_channels);
       }
 
-      auto write_res =
-          current_codec_->WriteOutputSamples({block->start, block->size});
-      block->used_size = write_res.first;
+      auto write_res = current_codec_->WriteOutputSamples(
+          output->data.subspan(output->info->bytes_in_stream));
+      output->info->bytes_in_stream += write_res.first;
       has_samples_to_send_ = !write_res.second;
 
-      auto chunk = std::unique_ptr<StreamEvent>(
-          StreamEvent::CreateArenaChunk(input_events_, *block));
-      if (!SendOrBufferEvent(std::move(chunk))) {
-        return;
+      if (has_samples_to_send_) {
+        // We weren't able to fit all the generated samples into the output
+        // buffer. Stop trying; we'll finish up during the next pass.
+        break;
       }
     }
-    // We will process the next frame during the next call to this method.
-  }
-
-  if (!needs_more_input_) {
-    auto res = current_codec_->ProcessNextFrame();
-    if (res.has_error()) {
-      // TODO(jacqueline): Handle errors.
-      return;
-    }
-    needs_more_input_ = res.value();
-    has_samples_to_send_ = true;
 
-    if (needs_more_input_) {
-      chunk_reader_->HandleBytesUsed(current_codec_->GetInputPosition());
+    if (input != inputs->end()) {
+      auto res = current_codec_->ProcessNextFrame();
+      if (res.has_error()) {
+        // TODO(jacqueline): Handle errors.
+        return;
+      }
+      input->data = input->data.subspan(current_codec_->GetInputPosition());
+
+      if (res.value()) {
+        // We're out of data in this buffer. Finish immediately; there's nothing
+        // to send.
+        break;
+      } else {
+        has_samples_to_send_ = true;
+      }
+    } else {
+      // No input; nothing to do.
+      break;
     }
   }
 }