New pipeline building, still needs proper control

9 files changed, 117 insertions, 55 deletions

diff --git a/src/audio/include/audio_decoder.hpp b/src/audio/include/audio_decoder.hpp
index be8daf99..6a1b5177 100644
--- a/src/audio/include/audio_decoder.hpp
+++ b/src/audio/include/audio_decoder.hpp
@@ -24,7 +24,7 @@ class AudioDecoder : public IAudioElement {
   AudioDecoder();
   ~AudioDecoder();
 
-  auto Process(std::vector<Stream>* inputs, MutableStream* output)
+  auto Process(const std::vector<InputStream>& inputs, OutputStream* output)
       -> void override;
 
   AudioDecoder(const AudioDecoder&) = delete;
@@ -32,11 +32,9 @@ class AudioDecoder : public IAudioElement {
 
  private:
   std::unique_ptr<codecs::ICodec> current_codec_;
-  std::optional<StreamInfo> stream_info_;
-
-  bool has_set_stream_info_;
+  std::optional<StreamInfo::Format> current_input_format_;
+  std::optional<StreamInfo::Format> current_output_format_;
   bool has_samples_to_send_;
-  bool needs_more_input_;
 
   auto ProcessStreamInfo(const StreamInfo& info) -> bool;
 };
diff --git a/src/audio/include/audio_element.hpp b/src/audio/include/audio_element.hpp
index c9192e4a..5884f7b2 100644
--- a/src/audio/include/audio_element.hpp
+++ b/src/audio/include/audio_element.hpp
@@ -37,11 +37,11 @@ static const size_t kEventQueueSize = 8;
  */
 class IAudioElement {
  public:
-  IAudioElement();
-  virtual ~IAudioElement();
+  IAudioElement() {}
+  virtual ~IAudioElement() {}
 
-  virtual auto Process(std::vector<Stream>* inputs, MutableStream* output)
-      -> void = 0;
+  virtual auto Process(const std::vector<InputStream>& inputs,
+                       OutputStream* output) -> void = 0;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/audio_playback.hpp b/src/audio/include/audio_playback.hpp
index 507e6f73..88dc29aa 100644
--- a/src/audio/include/audio_playback.hpp
+++ b/src/audio/include/audio_playback.hpp
@@ -8,6 +8,7 @@
 #include "audio_task.hpp"
 #include "esp_err.h"
 #include "fatfs_audio_input.hpp"
+#include "i2s_audio_output.hpp"
 #include "result.hpp"
 #include "span.hpp"
 
@@ -28,7 +29,7 @@ class AudioPlayback {
   static auto create(drivers::GpioExpander* expander)
       -> cpp::result<std::unique_ptr<AudioPlayback>, Error>;
 
-  AudioPlayback(FatfsAudioInput *file_input);
+  explicit AudioPlayback(std::unique_ptr<I2SAudioOutput> output);
   ~AudioPlayback();
 
   /*
@@ -44,10 +45,9 @@ class AudioPlayback {
   AudioPlayback& operator=(const AudioPlayback&) = delete;
 
  private:
-  FatfsAudioInput *file_source;
-
-  std::vector<std::unique_ptr<IAudioElement>> all_elements_;
-  std::unique_ptr<task::Handle> pipeline_;
+  std::unique_ptr<FatfsAudioInput> file_source_;
+  std::unique_ptr<I2SAudioOutput> i2s_output_;
+  std::vector<std::unique_ptr<IAudioElement>> elements_;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/audio_sink.hpp b/src/audio/include/audio_sink.hpp
new file mode 100644
index 00000000..ed7eb02b
--- /dev/null
+++ b/src/audio/include/audio_sink.hpp
@@ -0,0 +1,22 @@
+#pragma once
+
+#include "audio_element.hpp"
+#include "stream_info.hpp"
+namespace audio {
+
+class IAudioSink {
+ private:
+  static const std::size_t kDrainBufferSize = 8 * 1024;
+  StreamBufferHandle_t buffer_;
+
+ public:
+  IAudioSink() : buffer_(xStreamBufferCreate(kDrainBufferSize, 1)) {}
+  virtual ~IAudioSink() { vStreamBufferDelete(buffer_); }
+
+  virtual auto Configure(const StreamInfo::Format& format) -> bool = 0;
+  virtual auto Send(const cpp::span<std::byte>& data) -> void = 0;
+
+  auto buffer() const -> StreamBufferHandle_t { return buffer_; }
+};
+
+}  // namespace audio
diff --git a/src/audio/include/audio_task.hpp b/src/audio/include/audio_task.hpp
index 8db99850..a7b7a0fa 100644
--- a/src/audio/include/audio_task.hpp
+++ b/src/audio/include/audio_task.hpp
@@ -5,38 +5,32 @@
 #include <string>
 
 #include "audio_element.hpp"
+#include "audio_sink.hpp"
+#include "dac.hpp"
 #include "freertos/portmacro.h"
 #include "pipeline.hpp"
+#include "stream_buffer.hpp"
 
 namespace audio {
 
 namespace task {
+
+enum Command { PLAY, PAUSE, QUIT };
+
 struct AudioTaskArgs {
   Pipeline* pipeline;
-  QueueHandle_t input;
+  IAudioSink* sink;
+};
+struct AudioDrainArgs {
+  IAudioSink* sink;
+  std::atomic<Command>* command;
 };
 
 extern "C" void AudioTaskMain(void* args);
+extern "C" void AudioDrainMain(void* args);
 
-enum Command { PLAY, PAUSE, QUIT };
-
-class Handle {
- public:
-  explicit Handle(QueueHandle_t input);
-  ~Handle();
-
-  auto SetStreamInfo() -> void;
-  auto Play() -> void;
-  auto Pause() -> void;
-  auto Quit() -> void;
-
-  auto OutputBuffer() -> StreamBufferHandle_t;
-
- private:
-  QueueHandle_t input;
-};
-
-auto Start(Pipeline* pipeline) -> Handle*;
+auto StartPipeline(Pipeline* pipeline, IAudioSink* sink) -> void;
+auto StartDrain(IAudioSink* sink) -> void;
 
 }  // namespace task
 
diff --git a/src/audio/include/fatfs_audio_input.hpp b/src/audio/include/fatfs_audio_input.hpp
index b3a6d843..24f62e3c 100644
--- a/src/audio/include/fatfs_audio_input.hpp
+++ b/src/audio/include/fatfs_audio_input.hpp
@@ -16,6 +16,7 @@
 
 #include "audio_element.hpp"
 #include "stream_buffer.hpp"
+#include "stream_info.hpp"
 
 namespace audio {
 
@@ -26,7 +27,7 @@ class FatfsAudioInput : public IAudioElement {
 
   auto OpenFile(const std::string& path) -> void;
 
-  auto Process(std::vector<Stream>* inputs, MutableStream* output)
+  auto Process(const std::vector<InputStream>& inputs, OutputStream* output)
       -> void override;
 
   FatfsAudioInput(const FatfsAudioInput&) = delete;
diff --git a/src/audio/include/i2s_audio_output.hpp b/src/audio/include/i2s_audio_output.hpp
index 57881b35..77019228 100644
--- a/src/audio/include/i2s_audio_output.hpp
+++ b/src/audio/include/i2s_audio_output.hpp
@@ -5,6 +5,7 @@
 #include <vector>
 
 #include "audio_element.hpp"
+#include "audio_sink.hpp"
 #include "chunk.hpp"
 #include "result.hpp"
 
@@ -14,18 +15,18 @@
 
 namespace audio {
 
-class I2SAudioOutput : public IAudioElement {
+class I2SAudioOutput : public IAudioSink {
  public:
   enum Error { DAC_CONFIG, I2S_CONFIG, STREAM_INIT };
   static auto create(drivers::GpioExpander* expander)
-      -> cpp::result<std::shared_ptr<I2SAudioOutput>, Error>;
+      -> cpp::result<std::unique_ptr<I2SAudioOutput>, Error>;
 
   I2SAudioOutput(drivers::GpioExpander* expander,
                  std::unique_ptr<drivers::AudioDac> dac);
   ~I2SAudioOutput();
 
-  auto Process(std::vector<Stream>* inputs, MutableStream* output)
-      -> void override;
+  auto Configure(const StreamInfo::Format& format) -> bool override;
+  auto Send(const cpp::span<std::byte>& data) -> void override;
 
   I2SAudioOutput(const I2SAudioOutput&) = delete;
   I2SAudioOutput& operator=(const I2SAudioOutput&) = delete;
@@ -37,8 +38,6 @@ class I2SAudioOutput : public IAudioElement {
   std::unique_ptr<drivers::AudioDac> dac_;
 
   std::optional<StreamInfo::Pcm> current_config_;
-
-  auto ProcessStreamInfo(const StreamInfo& info) -> bool;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/pipeline.hpp b/src/audio/include/pipeline.hpp
index 42f70828..2e9247bc 100644
--- a/src/audio/include/pipeline.hpp
+++ b/src/audio/include/pipeline.hpp
@@ -3,6 +3,7 @@
 #include <memory>
 #include <optional>
 #include <string>
+#include <vector>
 
 #include "freertos/portmacro.h"
 
@@ -16,7 +17,7 @@ static const std::size_t kPipelineBufferSize = 32 * 1024;
 
 class Pipeline {
  public:
-  Pipeline(IAudioElement* output);
+  explicit Pipeline(IAudioElement* output);
   ~Pipeline();
   auto AddInput(IAudioElement* input) -> Pipeline*;
 
@@ -25,9 +26,9 @@ class Pipeline {
   auto NumInputs() const -> std::size_t;
 
   auto InStreams(std::vector<MappableRegion<kPipelineBufferSize>>*,
-                 std::vector<MutableStream>*) -> void;
+                 std::vector<RawStream>*) -> void;
 
-  auto OutStream(MappableRegion<kPipelineBufferSize>*) -> MutableStream;
+  auto OutStream(MappableRegion<kPipelineBufferSize>*) -> RawStream;
 
   auto GetIterationOrder() -> std::vector<Pipeline*>;
 
diff --git a/src/audio/include/stream_info.hpp b/src/audio/include/stream_info.hpp
index 47f65649..5622517f 100644
--- a/src/audio/include/stream_info.hpp
+++ b/src/audio/include/stream_info.hpp
@@ -4,6 +4,8 @@
 #include <optional>
 #include <string>
 #include <string_view>
+#include <type_traits>
+#include <utility>
 #include <variant>
 
 #include "result.hpp"
@@ -35,37 +37,82 @@ struct StreamInfo {
     // Number of bits per sample.
     uint8_t bits_per_sample;
     // The sample rate.
-    uint16_t sample_rate;
+    uint32_t sample_rate;
 
     bool operator==(const Pcm&) const = default;
   };
 
-  std::variant<Encoded, Pcm> data;
+  typedef std::variant<Encoded, Pcm> Format;
+  Format format;
 
   bool operator==(const StreamInfo&) const = default;
 };
 
-class MutableStream {
+class RawStream {
  public:
   StreamInfo* info;
   cpp::span<std::byte> data;
+  bool is_incomplete;
 
-  MutableStream(StreamInfo* i, cpp::span<std::byte> d)
-      : info(i), data(d) {}
+  RawStream(StreamInfo* i, cpp::span<std::byte> d)
+      : info(i), data(d), is_incomplete(false) {}
 };
 
 /*
  * A byte buffer + associated metadata, which is not allowed to modify any of
  * the underlying data.
  */
-class Stream {
+class InputStream {
  public:
-  explicit Stream(const MutableStream& s) : info(*s.info), data(s.data) {}
+  explicit InputStream(RawStream* s) : raw_(s) {}
 
-  const StreamInfo& info;
-  // `data` itself left mutable for signalling how much of the stream was
-  // consumed
-  cpp::span<const std::byte> data;
+  void consume(std::size_t bytes) const {
+    auto new_data = raw_->data.subspan(bytes);
+    std::move(new_data.begin(), new_data.end(), raw_->data.begin());
+    raw_->info->bytes_in_stream = new_data.size_bytes();
+  }
+
+  void mark_incomplete() const { raw_->is_incomplete = true; }
+
+  const StreamInfo& info() const { return *raw_->info; }
+
+  cpp::span<const std::byte> data() const {
+    return raw_->data.first(raw_->info->bytes_in_stream);
+  }
+
+ private:
+  RawStream* raw_;
+};
+
+class OutputStream {
+ public:
+  explicit OutputStream(RawStream* s) : raw_(s) {}
+
+  void add(std::size_t bytes) const { raw_->info->bytes_in_stream += bytes; }
+
+  bool prepare(const StreamInfo::Format& new_format) {
+    if (new_format == raw_->info->format) {
+      raw_->info->format = new_format;
+      return true;
+    }
+    if (raw_->is_incomplete) {
+      raw_->info->format = new_format;
+      raw_->info->bytes_in_stream = 0;
+      return true;
+    }
+    return false;
+  }
+
+  const StreamInfo& info() const { return *raw_->info; }
+
+  cpp::span<std::byte> data() const {
+    return raw_->data.subspan(raw_->info->bytes_in_stream);
+  }
+
+  bool is_incomplete() const { return raw_->is_incomplete; }
+
+ private:
+  RawStream* raw_;
 };
 
 }  // namespace audio