Merge branch 'main' into leveldb

9 files changed, 228 insertions, 123 deletions

diff --git a/src/audio/include/audio_decoder.hpp b/src/audio/include/audio_decoder.hpp
index 47642469..6a1b5177 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,19 @@ 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(const std::vector<InputStream>& inputs, OutputStream* 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_;
+  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;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/audio_element.hpp b/src/audio/include/audio_element.hpp
index 91036348..5884f7b2 100644
--- a/src/audio/include/audio_element.hpp
+++ b/src/audio/include/audio_element.hpp
@@ -37,65 +37,11 @@ static const size_t kEventQueueSize = 8;
  */
 class IAudioElement {
  public:
-  IAudioElement();
-  virtual ~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(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 a1bea64f..cd4be3e7 100644
--- a/src/audio/include/audio_playback.hpp
+++ b/src/audio/include/audio_playback.hpp
@@ -5,7 +5,11 @@
 #include <string>
 #include <vector>
 
+#include "audio_task.hpp"
+#include "driver_cache.hpp"
 #include "esp_err.h"
+#include "fatfs_audio_input.hpp"
+#include "i2s_audio_output.hpp"
 #include "result.hpp"
 #include "span.hpp"
 
@@ -22,13 +26,7 @@ namespace audio {
  */
 class AudioPlayback {
  public:
-  enum Error { ERR_INIT_ELEMENT, ERR_MEM };
-  static auto create(drivers::GpioExpander* expander,
-                     std::shared_ptr<drivers::SdStorage> storage)
-      -> cpp::result<std::unique_ptr<AudioPlayback>, Error>;
-
-  // TODO(jacqueline): configure on the fly once we have things to configure.
-  AudioPlayback();
+  explicit AudioPlayback(drivers::DriverCache* drivers);
   ~AudioPlayback();
 
   /*
@@ -44,9 +42,9 @@ class AudioPlayback {
   AudioPlayback& operator=(const AudioPlayback&) = delete;
 
  private:
-  auto ConnectElements(IAudioElement* src, IAudioElement* sink) -> void;
-
-  QueueHandle_t input_handle_;
+  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..7a535c35
--- /dev/null
+++ b/src/audio/include/audio_sink.hpp
@@ -0,0 +1,44 @@
+#pragma once
+
+#include <stdint.h>
+#include "audio_element.hpp"
+#include "esp_heap_caps.h"
+#include "freertos/FreeRTOS.h"
+#include "stream_info.hpp"
+namespace audio {
+
+class IAudioSink {
+ private:
+  // TODO: tune. at least about 12KiB seems right for mp3
+  static const std::size_t kDrainBufferSize = 48 * 1024;
+  uint8_t* buffer_;
+  StaticStreamBuffer_t* metadata_;
+  StreamBufferHandle_t handle_;
+
+ public:
+  IAudioSink()
+      : buffer_(reinterpret_cast<uint8_t*>(
+            heap_caps_malloc(kDrainBufferSize,
+                             MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT))),
+        metadata_(reinterpret_cast<StaticStreamBuffer_t*>(
+            heap_caps_malloc(sizeof(StaticStreamBuffer_t),
+                             MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT))),
+        handle_(xStreamBufferCreateStatic(kDrainBufferSize,
+                                          1,
+                                          buffer_,
+                                          metadata_)) {}
+
+  virtual ~IAudioSink() {
+    vStreamBufferDelete(handle_);
+    free(buffer_);
+    free(metadata_);
+  }
+
+  virtual auto Configure(const StreamInfo::Format& format) -> bool = 0;
+  virtual auto Send(const cpp::span<std::byte>& data) -> void = 0;
+  virtual auto Log() -> void {}
+
+  auto buffer() -> StreamBufferHandle_t { return handle_; }
+};
+
+}  // namespace audio
diff --git a/src/audio/include/audio_task.hpp b/src/audio/include/audio_task.hpp
index df70ebaa..a7b7a0fa 100644
--- a/src/audio/include/audio_task.hpp
+++ b/src/audio/include/audio_task.hpp
@@ -5,18 +5,33 @@
 #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 {
-  std::shared_ptr<IAudioElement>& element;
+  Pipeline* pipeline;
+  IAudioSink* sink;
 };
+struct AudioDrainArgs {
+  IAudioSink* sink;
+  std::atomic<Command>* command;
+};
+
+extern "C" void AudioTaskMain(void* args);
+extern "C" void AudioDrainMain(void* args);
 
-auto StartAudioTask(const std::string& name,
-                    std::optional<BaseType_t> core_id,
-                    std::shared_ptr<IAudioElement> element) -> void;
+auto StartPipeline(Pipeline* pipeline, IAudioSink* sink) -> void;
+auto StartDrain(IAudioSink* sink) -> void;
 
-void AudioTaskMain(void* args);
+}  // namespace task
 
 }  // namespace audio
diff --git a/src/audio/include/fatfs_audio_input.hpp b/src/audio/include/fatfs_audio_input.hpp
index 9f2d676c..24f62e3c 100644
--- a/src/audio/include/fatfs_audio_input.hpp
+++ b/src/audio/include/fatfs_audio_input.hpp
@@ -3,41 +3,37 @@
 #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"
+#include "stream_info.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(const std::vector<InputStream>& inputs, OutputStream* 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..07430777 100644
--- a/src/audio/include/i2s_audio_output.hpp
+++ b/src/audio/include/i2s_audio_output.hpp
@@ -2,34 +2,28 @@
 
 #include <cstdint>
 #include <memory>
+#include <vector>
 
 #include "audio_element.hpp"
+#include "audio_sink.hpp"
 #include "chunk.hpp"
 #include "result.hpp"
 
 #include "dac.hpp"
 #include "gpio_expander.hpp"
+#include "stream_info.hpp"
 
 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>;
-
   I2SAudioOutput(drivers::GpioExpander* expander,
-                 std::unique_ptr<drivers::AudioDac> dac);
+                 std::shared_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 Configure(const StreamInfo::Format& format) -> bool override;
+  auto Send(const cpp::span<std::byte>& data) -> void override;
+  auto Log() -> void override;
 
   I2SAudioOutput(const I2SAudioOutput&) = delete;
   I2SAudioOutput& operator=(const I2SAudioOutput&) = delete;
@@ -38,10 +32,9 @@ class I2SAudioOutput : public IAudioElement {
   auto SetVolume(uint8_t volume) -> void;
 
   drivers::GpioExpander* expander_;
-  std::unique_ptr<drivers::AudioDac> dac_;
+  std::shared_ptr<drivers::AudioDac> dac_;
 
-  std::optional<ChunkReader> chunk_reader_;
-  cpp::span<std::byte> latest_chunk_;
+  std::optional<StreamInfo::Pcm> current_config_;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/pipeline.hpp b/src/audio/include/pipeline.hpp
new file mode 100644
index 00000000..7a658a38
--- /dev/null
+++ b/src/audio/include/pipeline.hpp
@@ -0,0 +1,43 @@
+#pragma once
+
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include "freertos/portmacro.h"
+
+#include "audio_element.hpp"
+#include "himem.hpp"
+#include "stream_info.hpp"
+
+namespace audio {
+
+static const std::size_t kPipelineBufferSize = 64 * 1024;
+
+class Pipeline {
+ public:
+  explicit 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<RawStream>*) -> void;
+
+  auto OutStream(MappableRegion<kPipelineBufferSize>*) -> RawStream;
+
+  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..28095935 100644
--- a/src/audio/include/stream_info.hpp
+++ b/src/audio/include/stream_info.hpp
@@ -4,19 +4,96 @@
 #include <optional>
 #include <string>
 #include <string_view>
+#include <type_traits>
+#include <utility>
+#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{0};
+
+  // 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.
+    uint32_t sample_rate;
+
+    bool operator==(const Pcm&) const = default;
+  };
+
+  typedef std::variant<std::monostate, Encoded, Pcm> Format;
+  Format format{};
+
+  bool operator==(const StreamInfo&) const = default;
+};
+
+class RawStream {
+ public:
+  StreamInfo* info;
+  cpp::span<std::byte> data;
+  bool is_incomplete;
+
+  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 InputStream {
+ public:
+  explicit InputStream(RawStream* s) : raw_(s) {}
+
+  void consume(std::size_t bytes) const;
+
+  void mark_incomplete() const;
+
+  const StreamInfo& info() const;
+
+  cpp::span<const std::byte> data() const;
+
+ private:
+  RawStream* raw_;
+};
+
+class OutputStream {
+ public:
+  explicit OutputStream(RawStream* s) : raw_(s) {}
+
+  void add(std::size_t bytes) const;
+
+  bool prepare(const StreamInfo::Format& new_format);
+
+  const StreamInfo& info() const;
+
+  cpp::span<std::byte> data() const;
+
+  bool is_incomplete() const;
+
+ private:
+  RawStream* raw_;
 };
 
 }  // namespace audio