8 files changed, 153 insertions, 107 deletions

diff --git a/src/audio/include/audio_decoder.hpp b/src/audio/include/audio_decoder.hpp
index 47642469..be8daf99 100644
--- a/src/audio/include/audio_decoder.hpp
+++ b/src/audio/include/audio_decoder.hpp
@@ -3,6 +3,7 @@
 #include <cstddef>
 #include <cstdint>
 #include <memory>
+#include <vector>
 
 #include "chunk.hpp"
 #include "ff.h"
@@ -10,6 +11,7 @@
 
 #include "audio_element.hpp"
 #include "codec.hpp"
+#include "stream_info.hpp"
 
 namespace audio {
 
@@ -22,28 +24,21 @@ class AudioDecoder : public IAudioElement {
   AudioDecoder();
   ~AudioDecoder();
 
-  auto StackSizeBytes() const -> std::size_t override { return 10 * 1024; };
-
-  auto HasUnprocessedInput() -> bool override;
-  auto IsOverBuffered() -> bool override;
-
-  auto ProcessStreamInfo(const StreamInfo& info) -> void override;
-  auto ProcessChunk(const cpp::span<std::byte>& chunk) -> void override;
-  auto ProcessEndOfStream() -> void override;
-  auto Process() -> void override;
+  auto Process(std::vector<Stream>* inputs, MutableStream* output)
+      -> void override;
 
   AudioDecoder(const AudioDecoder&) = delete;
   AudioDecoder& operator=(const AudioDecoder&) = delete;
 
  private:
-  memory::Arena arena_;
   std::unique_ptr<codecs::ICodec> current_codec_;
   std::optional<StreamInfo> stream_info_;
-  std::optional<ChunkReader> chunk_reader_;
 
-  bool has_sent_stream_info_;
+  bool has_set_stream_info_;
   bool has_samples_to_send_;
   bool needs_more_input_;
+
+  auto ProcessStreamInfo(const StreamInfo& info) -> bool;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/audio_element.hpp b/src/audio/include/audio_element.hpp
index 91036348..c9192e4a 100644
--- a/src/audio/include/audio_element.hpp
+++ b/src/audio/include/audio_element.hpp
@@ -40,62 +40,8 @@ class IAudioElement {
   IAudioElement();
   virtual ~IAudioElement();
 
-  /*
-   * Returns the stack size in bytes that this element requires. This should
-   * be tuned according to the observed stack size of each element, as different
-   * elements have fairly different stack requirements (particular decoders).
-   */
-  virtual auto StackSizeBytes() const -> std::size_t { return 4096; };
-
-  /* Returns this element's input buffer. */
-  auto InputEventQueue() const -> QueueHandle_t { return input_events_; }
-  /* Returns this element's output buffer. */
-  auto OutputEventQueue() const -> QueueHandle_t { return output_events_; }
-  auto OutputEventQueue(const QueueHandle_t q) -> void { output_events_ = q; }
-
-  virtual auto HasUnprocessedInput() -> bool = 0;
-
-  virtual auto IsOverBuffered() -> bool { return false; }
-
-  auto HasUnflushedOutput() -> bool { return !buffered_output_.empty(); }
-  auto FlushBufferedOutput() -> bool;
-
-  /*
-   * Called when a StreamInfo message is received. Used to configure this
-   * element in preperation for incoming chunks.
-   */
-  virtual auto ProcessStreamInfo(const StreamInfo& info) -> void = 0;
-
-  /*
-   * Called when a ChunkHeader message is received. Includes the data associated
-   * with this chunk of stream data. This method should return the number of
-   * bytes in this chunk that were actually used; leftover bytes will be
-   * prepended to the next call.
-   */
-  virtual auto ProcessChunk(const cpp::span<std::byte>& chunk) -> void = 0;
-
-  virtual auto ProcessEndOfStream() -> void = 0;
-
-  virtual auto ProcessLogStatus() -> void {}
-
-  /*
-   * Called when there has been no data received over the input buffer for some
-   * time. This could be used to synthesize output, or to save memory by
-   * releasing unused resources.
-   */
-  virtual auto Process() -> void = 0;
-
- protected:
-  auto SendOrBufferEvent(std::unique_ptr<StreamEvent> event) -> bool;
-
-  // Queue for events coming into this element. Owned by us.
-  QueueHandle_t input_events_;
-  // Queue for events going into the next element. Not owned by us, may be null
-  // if we're not yet in a pipeline.
-  // FIXME: it would be nicer if this was non-nullable.
-  QueueHandle_t output_events_;
-  // Output events that have been generated, but are yet to be sent downstream.
-  std::deque<std::unique_ptr<StreamEvent>> buffered_output_;
+  virtual auto Process(std::vector<Stream>* inputs, MutableStream* output)
+      -> void = 0;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/audio_playback.hpp b/src/audio/include/audio_playback.hpp
index a1bea64f..507e6f73 100644
--- a/src/audio/include/audio_playback.hpp
+++ b/src/audio/include/audio_playback.hpp
@@ -5,7 +5,9 @@
 #include <string>
 #include <vector>
 
+#include "audio_task.hpp"
 #include "esp_err.h"
+#include "fatfs_audio_input.hpp"
 #include "result.hpp"
 #include "span.hpp"
 
@@ -23,12 +25,10 @@ namespace audio {
 class AudioPlayback {
  public:
   enum Error { ERR_INIT_ELEMENT, ERR_MEM };
-  static auto create(drivers::GpioExpander* expander,
-                     std::shared_ptr<drivers::SdStorage> storage)
+  static auto create(drivers::GpioExpander* expander)
       -> cpp::result<std::unique_ptr<AudioPlayback>, Error>;
 
-  // TODO(jacqueline): configure on the fly once we have things to configure.
-  AudioPlayback();
+  AudioPlayback(FatfsAudioInput *file_input);
   ~AudioPlayback();
 
   /*
@@ -44,9 +44,10 @@ class AudioPlayback {
   AudioPlayback& operator=(const AudioPlayback&) = delete;
 
  private:
-  auto ConnectElements(IAudioElement* src, IAudioElement* sink) -> void;
+  FatfsAudioInput *file_source;
 
-  QueueHandle_t input_handle_;
+  std::vector<std::unique_ptr<IAudioElement>> all_elements_;
+  std::unique_ptr<task::Handle> pipeline_;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/audio_task.hpp b/src/audio/include/audio_task.hpp
index df70ebaa..8db99850 100644
--- a/src/audio/include/audio_task.hpp
+++ b/src/audio/include/audio_task.hpp
@@ -6,17 +6,38 @@
 
 #include "audio_element.hpp"
 #include "freertos/portmacro.h"
+#include "pipeline.hpp"
 
 namespace audio {
 
+namespace task {
 struct AudioTaskArgs {
-  std::shared_ptr<IAudioElement>& element;
+  Pipeline* pipeline;
+  QueueHandle_t input;
 };
 
-auto StartAudioTask(const std::string& name,
-                    std::optional<BaseType_t> core_id,
-                    std::shared_ptr<IAudioElement> element) -> void;
+extern "C" void AudioTaskMain(void* args);
 
-void AudioTaskMain(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*;
+
+}  // namespace task
 
 }  // namespace audio
diff --git a/src/audio/include/fatfs_audio_input.hpp b/src/audio/include/fatfs_audio_input.hpp
index 9f2d676c..b3a6d843 100644
--- a/src/audio/include/fatfs_audio_input.hpp
+++ b/src/audio/include/fatfs_audio_input.hpp
@@ -3,41 +3,36 @@
 #include <cstdint>
 #include <memory>
 #include <string>
+#include <vector>
 
 #include "arena.hpp"
 #include "chunk.hpp"
 #include "freertos/FreeRTOS.h"
 
+#include "ff.h"
 #include "freertos/message_buffer.h"
 #include "freertos/queue.h"
 #include "span.hpp"
 
 #include "audio_element.hpp"
-#include "storage.hpp"
 #include "stream_buffer.hpp"
 
 namespace audio {
 
 class FatfsAudioInput : public IAudioElement {
  public:
-  explicit FatfsAudioInput(std::shared_ptr<drivers::SdStorage> storage);
+  explicit FatfsAudioInput();
   ~FatfsAudioInput();
 
-  auto HasUnprocessedInput() -> bool override;
-  auto IsOverBuffered() -> bool override;
+  auto OpenFile(const std::string& path) -> void;
 
-  auto ProcessStreamInfo(const StreamInfo& info) -> void override;
-  auto ProcessChunk(const cpp::span<std::byte>& chunk) -> void override;
-  auto ProcessEndOfStream() -> void override;
-  auto Process() -> void override;
+  auto Process(std::vector<Stream>* inputs, MutableStream* output)
+      -> void override;
 
   FatfsAudioInput(const FatfsAudioInput&) = delete;
   FatfsAudioInput& operator=(const FatfsAudioInput&) = delete;
 
  private:
-  memory::Arena arena_;
-  std::shared_ptr<drivers::SdStorage> storage_;
-
   FIL current_file_;
   bool is_file_open_;
 };
diff --git a/src/audio/include/i2s_audio_output.hpp b/src/audio/include/i2s_audio_output.hpp
index 2bea091b..57881b35 100644
--- a/src/audio/include/i2s_audio_output.hpp
+++ b/src/audio/include/i2s_audio_output.hpp
@@ -2,6 +2,7 @@
 
 #include <cstdint>
 #include <memory>
+#include <vector>
 
 #include "audio_element.hpp"
 #include "chunk.hpp"
@@ -9,6 +10,7 @@
 
 #include "dac.hpp"
 #include "gpio_expander.hpp"
+#include "stream_info.hpp"
 
 namespace audio {
 
@@ -22,14 +24,8 @@ class I2SAudioOutput : public IAudioElement {
                  std::unique_ptr<drivers::AudioDac> dac);
   ~I2SAudioOutput();
 
-  auto HasUnprocessedInput() -> bool override;
-  auto IsOverBuffered() -> bool override;
-
-  auto ProcessStreamInfo(const StreamInfo& info) -> void override;
-  auto ProcessChunk(const cpp::span<std::byte>& chunk) -> void override;
-  auto ProcessEndOfStream() -> void override;
-  auto ProcessLogStatus() -> void override;
-  auto Process() -> void override;
+  auto Process(std::vector<Stream>* inputs, MutableStream* output)
+      -> void override;
 
   I2SAudioOutput(const I2SAudioOutput&) = delete;
   I2SAudioOutput& operator=(const I2SAudioOutput&) = delete;
@@ -40,8 +36,9 @@ class I2SAudioOutput : public IAudioElement {
   drivers::GpioExpander* expander_;
   std::unique_ptr<drivers::AudioDac> dac_;
 
-  std::optional<ChunkReader> chunk_reader_;
-  cpp::span<std::byte> latest_chunk_;
+  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
new file mode 100644
index 00000000..42f70828
--- /dev/null
+++ b/src/audio/include/pipeline.hpp
@@ -0,0 +1,42 @@
+#pragma once
+
+#include <memory>
+#include <optional>
+#include <string>
+
+#include "freertos/portmacro.h"
+
+#include "audio_element.hpp"
+#include "himem.hpp"
+#include "stream_info.hpp"
+
+namespace audio {
+
+static const std::size_t kPipelineBufferSize = 32 * 1024;
+
+class Pipeline {
+ public:
+  Pipeline(IAudioElement* output);
+  ~Pipeline();
+  auto AddInput(IAudioElement* input) -> Pipeline*;
+
+  auto OutputElement() const -> IAudioElement*;
+
+  auto NumInputs() const -> std::size_t;
+
+  auto InStreams(std::vector<MappableRegion<kPipelineBufferSize>>*,
+                 std::vector<MutableStream>*) -> void;
+
+  auto OutStream(MappableRegion<kPipelineBufferSize>*) -> MutableStream;
+
+  auto GetIterationOrder() -> std::vector<Pipeline*>;
+
+ private:
+  IAudioElement* root_;
+  std::vector<std::unique_ptr<Pipeline>> subtrees_;
+
+  HimemAlloc<kPipelineBufferSize> output_buffer_;
+  StreamInfo output_info_;
+};
+
+}  // namespace audio
diff --git a/src/audio/include/stream_info.hpp b/src/audio/include/stream_info.hpp
index bf67364f..47f65649 100644
--- a/src/audio/include/stream_info.hpp
+++ b/src/audio/include/stream_info.hpp
@@ -4,19 +4,68 @@
 #include <optional>
 #include <string>
 #include <string_view>
+#include <variant>
 
-#include "cbor.h"
 #include "result.hpp"
-#include "sys/_stdint.h"
+#include "span.hpp"
+#include "types.hpp"
 
 namespace audio {
 
 struct StreamInfo {
-  std::optional<std::string> path;
-  std::optional<uint8_t> channels;
-  std::optional<uint8_t> bits_per_sample;
-  std::optional<uint16_t> sample_rate;
-  std::optional<size_t> chunk_size;
+  // The number of bytes that are available for consumption within this
+  // stream's buffer.
+  std::size_t bytes_in_stream;
+
+  // The total length of this stream, in case its source is finite (e.g. a
+  // file on disk). May be absent for endless streams (internet streams,
+  // generated audio, etc.)
+  std::optional<std::size_t> length_bytes;
+
+  struct Encoded {
+    // The codec that this stream is associated with.
+    codecs::StreamType type;
+
+    bool operator==(const Encoded&) const = default;
+  };
+
+  struct Pcm {
+    // Number of channels in this stream.
+    uint8_t channels;
+    // Number of bits per sample.
+    uint8_t bits_per_sample;
+    // The sample rate.
+    uint16_t sample_rate;
+
+    bool operator==(const Pcm&) const = default;
+  };
+
+  std::variant<Encoded, Pcm> data;
+
+  bool operator==(const StreamInfo&) const = default;
+};
+
+class MutableStream {
+ public:
+  StreamInfo* info;
+  cpp::span<std::byte> data;
+
+  MutableStream(StreamInfo* i, cpp::span<std::byte> d)
+      : info(i), data(d) {}
+};
+
+/*
+ * A byte buffer + associated metadata, which is not allowed to modify any of
+ * the underlying data.
+ */
+class Stream {
+ public:
+  explicit Stream(const MutableStream& s) : info(*s.info), data(s.data) {}
+
+  const StreamInfo& info;
+  // `data` itself left mutable for signalling how much of the stream was
+  // consumed
+  cpp::span<const std::byte> data;
 };
 
 }  // namespace audio