Add very limited resampling (it's slow as shit)

16 files changed, 544 insertions, 63 deletions

diff --git a/src/audio/CMakeLists.txt b/src/audio/CMakeLists.txt
index 22a0160c..9e50f8ff 100644
--- a/src/audio/CMakeLists.txt
+++ b/src/audio/CMakeLists.txt
@@ -5,7 +5,7 @@
 idf_component_register(
   SRCS "audio_task.cpp" "chunk.cpp" "fatfs_audio_input.cpp"
   "stream_message.cpp" "i2s_audio_output.cpp" "stream_buffer.cpp" "track_queue.cpp"
-  "stream_event.cpp" "stream_info.cpp" "audio_fsm.cpp"
+  "stream_event.cpp" "stream_info.cpp" "audio_fsm.cpp" "sink_mixer.cpp"
   INCLUDE_DIRS "include"
   REQUIRES "codecs" "drivers" "cbor" "result" "tasks" "span" "memory" "tinyfsm" "database" "system_fsm" "playlist" "libsamplerate")
 
diff --git a/src/audio/audio_task.cpp b/src/audio/audio_task.cpp
index 7e0fb207..c3498965 100644
--- a/src/audio/audio_task.cpp
+++ b/src/audio/audio_task.cpp
@@ -34,6 +34,7 @@
 #include "freertos/queue.h"
 #include "freertos/ringbuf.h"
 #include "pipeline.hpp"
+#include "sink_mixer.hpp"
 #include "span.hpp"
 
 #include "arena.hpp"
@@ -115,14 +116,12 @@ AudioTask::AudioTask(IAudioSource* source, IAudioSink* sink)
     : source_(source),
       sink_(sink),
       codec_(),
+      mixer_(new SinkMixer(sink->stream())),
       timer_(),
       has_begun_decoding_(false),
       current_input_format_(),
       current_output_format_(),
-      sample_buffer_(reinterpret_cast<std::byte*>(
-          heap_caps_malloc(kSampleBufferSize,
-                           MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT))),
-      sample_buffer_len_(kSampleBufferSize) {}
+      codec_buffer_(new RawStream(kSampleBufferSize)) {}
 
 void AudioTask::Main() {
   for (;;) {
@@ -246,13 +245,17 @@ auto AudioTask::BeginDecoding(InputStream& stream) -> bool {
     return false;
   }
 
+  OutputStream writer{codec_buffer_.get()};
+  writer.prepare(new_format, {});
+
   return true;
 }
 
 auto AudioTask::ContinueDecoding(InputStream& stream) -> bool {
   while (!stream.data().empty()) {
-    auto res = codec_->ContinueStream(stream.data(),
-                                      {sample_buffer_, sample_buffer_len_});
+    OutputStream writer{codec_buffer_.get()};
+
+    auto res = codec_->ContinueStream(stream.data(), writer.data());
 
     stream.consume(res.first);
 
@@ -263,9 +266,10 @@ auto AudioTask::ContinueDecoding(InputStream& stream) -> bool {
         return false;
       }
     } else {
-      xStreamBufferSend(sink_->stream(), sample_buffer_,
-                        res.second->bytes_written, portMAX_DELAY);
-      timer_->AddBytes(res.second->bytes_written);
+      writer.add(res.second->bytes_written);
+
+      InputStream reader{codec_buffer_.get()};
+      SendToSink(reader);
     }
   }
   return true;
@@ -284,21 +288,22 @@ auto AudioTask::FinishDecoding(InputStream& stream) -> void {
     std::unique_ptr<RawStream> mad_buffer;
     mad_buffer.reset(new RawStream(stream.data().size_bytes() + 8));
 
-    OutputStream writer{mad_buffer.get()};
+    OutputStream mad_writer{mad_buffer.get()};
     std::copy(stream.data().begin(), stream.data().end(),
-              writer.data().begin());
-    std::fill(writer.data().begin(), writer.data().end(), std::byte{0});
+              mad_writer.data().begin());
+    std::fill(mad_writer.data().begin(), mad_writer.data().end(), std::byte{0});
     InputStream padded_stream{mad_buffer.get()};
 
-    auto res = codec_->ContinueStream(stream.data(),
-                                      {sample_buffer_, sample_buffer_len_});
+    OutputStream writer{codec_buffer_.get()};
+    auto res = codec_->ContinueStream(stream.data(), writer.data());
     if (res.second.has_error()) {
       return;
     }
 
-    xStreamBufferSend(sink_->stream(), sample_buffer_,
-                      res.second->bytes_written, portMAX_DELAY);
-    timer_->AddBytes(res.second->bytes_written);
+    writer.add(res.second->bytes_written);
+
+    InputStream reader{codec_buffer_.get()};
+    SendToSink(reader);
   }
 }
 
@@ -319,24 +324,31 @@ auto AudioTask::ForwardPcmStream(StreamInfo::Pcm& format,
   xStreamBufferSend(sink_->stream(), samples.data(), samples.size_bytes(),
                     portMAX_DELAY);
   timer_->AddBytes(samples.size_bytes());
+  InputStream reader{codec_buffer_.get()};
+  SendToSink(reader);
+
   return true;
 }
 
 auto AudioTask::ConfigureSink(const StreamInfo::Pcm& format,
                               const Duration& duration) -> bool {
   if (format != current_output_format_) {
-    // The new format is different to the old one. Wait for the sink to drain
-    // before continuing.
-    while (!xStreamBufferIsEmpty(sink_->stream())) {
-      ESP_LOGI(kTag, "waiting for sink stream to drain...");
-      // TODO(jacqueline): Get the sink drain ISR to notify us of this
-      // via semaphore instead of busy-ish waiting.
-      vTaskDelay(pdMS_TO_TICKS(100));
-    }
+    current_output_format_ = format;
+    StreamInfo::Pcm new_sink_format = sink_->PrepareFormat(format);
+    if (new_sink_format != current_sink_format_) {
+      current_sink_format_ = new_sink_format;
+
+      // The new format is different to the old one. Wait for the sink to drain
+      // before continuing.
+      while (!xStreamBufferIsEmpty(sink_->stream())) {
+        ESP_LOGI(kTag, "waiting for sink stream to drain...");
+        // TODO(jacqueline): Get the sink drain ISR to notify us of this
+        // via semaphore instead of busy-ish waiting.
+        vTaskDelay(pdMS_TO_TICKS(10));
+      }
 
-    ESP_LOGI(kTag, "configuring sink");
-    if (!sink_->Configure(format)) {
-      return false;
+      ESP_LOGI(kTag, "configuring sink");
+      sink_->Configure(new_sink_format);
     }
   }
 
@@ -345,4 +357,17 @@ auto AudioTask::ConfigureSink(const StreamInfo::Pcm& format,
   return true;
 }
 
+auto AudioTask::SendToSink(InputStream& stream) -> void {
+  std::size_t bytes_to_send = stream.data().size_bytes();
+  std::size_t bytes_sent;
+  if (stream.info().format_as<StreamInfo::Pcm>() == current_sink_format_) {
+    bytes_sent = xStreamBufferSend(sink_->stream(), stream.data().data(),
+                                   bytes_to_send, portMAX_DELAY);
+    stream.consume(bytes_sent);
+  } else {
+    bytes_sent = mixer_->MixAndSend(stream, current_sink_format_.value());
+  }
+  timer_->AddBytes(bytes_sent);
+}
+
 }  // namespace audio
diff --git a/src/audio/i2s_audio_output.cpp b/src/audio/i2s_audio_output.cpp
index bb413b38..e53dbe2a 100644
--- a/src/audio/i2s_audio_output.cpp
+++ b/src/audio/i2s_audio_output.cpp
@@ -115,10 +115,19 @@ auto I2SAudioOutput::AdjustVolumeDown() -> bool {
   return true;
 }
 
-auto I2SAudioOutput::Configure(const StreamInfo::Pcm& pcm) -> bool {
+auto I2SAudioOutput::PrepareFormat(const StreamInfo::Pcm& orig)
+    -> StreamInfo::Pcm {
+  return StreamInfo::Pcm{
+      .channels = std::min<uint8_t>(orig.channels, 2),
+      .bits_per_sample = std::clamp<uint8_t>(orig.bits_per_sample, 16, 32),
+      .sample_rate = std::clamp<uint32_t>(orig.sample_rate, 8000, 96000),
+  };
+}
+
+auto I2SAudioOutput::Configure(const StreamInfo::Pcm& pcm) -> void {
   if (current_config_ && pcm == *current_config_) {
     ESP_LOGI(kTag, "ignoring unchanged format");
-    return true;
+    return;
   }
 
   ESP_LOGI(kTag, "incoming audio stream: %u ch %u bpp @ %lu Hz", pcm.channels,
@@ -134,7 +143,7 @@ auto I2SAudioOutput::Configure(const StreamInfo::Pcm& pcm) -> bool {
       break;
     default:
       ESP_LOGE(kTag, "dropping stream with out of bounds channels");
-      return false;
+      return;
   }
 
   drivers::I2SDac::BitsPerSample bps;
@@ -150,30 +159,36 @@ auto I2SAudioOutput::Configure(const StreamInfo::Pcm& pcm) -> bool {
       break;
     default:
       ESP_LOGE(kTag, "dropping stream with unknown bps");
-      return false;
+      return;
   }
 
   drivers::I2SDac::SampleRate sample_rate;
   switch (pcm.sample_rate) {
+    case 8000:
+      sample_rate = drivers::I2SDac::SAMPLE_RATE_8;
+      break;
+    case 32000:
+      sample_rate = drivers::I2SDac::SAMPLE_RATE_32;
+      break;
     case 44100:
       sample_rate = drivers::I2SDac::SAMPLE_RATE_44_1;
       break;
     case 48000:
       sample_rate = drivers::I2SDac::SAMPLE_RATE_48;
       break;
+    case 88200:
+      sample_rate = drivers::I2SDac::SAMPLE_RATE_88_2;
+      break;
+    case 96000:
+      sample_rate = drivers::I2SDac::SAMPLE_RATE_96;
+      break;
     default:
       ESP_LOGE(kTag, "dropping stream with unknown rate");
-      return false;
+      return;
   }
 
   dac_->Reconfigure(ch, bps, sample_rate);
   current_config_ = pcm;
-
-  return true;
-}
-
-auto I2SAudioOutput::Send(const cpp::span<std::byte>& data) -> void {
-  dac_->WriteData(data);
 }
 
 }  // namespace audio
diff --git a/src/audio/include/audio_sink.hpp b/src/audio/include/audio_sink.hpp
index 261f7c79..20bb6da6 100644
--- a/src/audio/include/audio_sink.hpp
+++ b/src/audio/include/audio_sink.hpp
@@ -38,8 +38,8 @@ class IAudioSink {
   virtual auto AdjustVolumeUp() -> bool = 0;
   virtual auto AdjustVolumeDown() -> bool = 0;
 
-  virtual auto Configure(const StreamInfo::Pcm& format) -> bool = 0;
-  virtual auto Send(const cpp::span<std::byte>& data) -> void = 0;
+  virtual auto PrepareFormat(const StreamInfo::Pcm&) -> StreamInfo::Pcm = 0;
+  virtual auto Configure(const StreamInfo::Pcm& format) -> void = 0;
 
   auto stream() -> StreamBufferHandle_t { return stream_; }
 };
diff --git a/src/audio/include/audio_task.hpp b/src/audio/include/audio_task.hpp
index f6e9789b..b27aa039 100644
--- a/src/audio/include/audio_task.hpp
+++ b/src/audio/include/audio_task.hpp
@@ -14,6 +14,7 @@
 #include "audio_source.hpp"
 #include "codec.hpp"
 #include "pipeline.hpp"
+#include "sink_mixer.hpp"
 #include "stream_info.hpp"
 
 namespace audio {
@@ -63,18 +64,20 @@ class AudioTask {
   auto ForwardPcmStream(StreamInfo::Pcm&, cpp::span<const std::byte>) -> bool;
 
   auto ConfigureSink(const StreamInfo::Pcm&, const Duration&) -> bool;
+  auto SendToSink(InputStream&) -> void;
 
   IAudioSource* source_;
   IAudioSink* sink_;
   std::unique_ptr<codecs::ICodec> codec_;
+  std::unique_ptr<SinkMixer> mixer_;
   std::unique_ptr<Timer> timer_;
 
   bool has_begun_decoding_;
   std::optional<StreamInfo::Format> current_input_format_;
   std::optional<StreamInfo::Pcm> current_output_format_;
+  std::optional<StreamInfo::Pcm> current_sink_format_;
 
-  std::byte* sample_buffer_;
-  std::size_t sample_buffer_len_;
+  std::unique_ptr<RawStream> codec_buffer_;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/i2s_audio_output.hpp b/src/audio/include/i2s_audio_output.hpp
index 43155711..e0f791c5 100644
--- a/src/audio/include/i2s_audio_output.hpp
+++ b/src/audio/include/i2s_audio_output.hpp
@@ -35,8 +35,8 @@ class I2SAudioOutput : public IAudioSink {
   auto AdjustVolumeUp() -> bool override;
   auto AdjustVolumeDown() -> bool override;
 
-  auto Configure(const StreamInfo::Pcm& format) -> bool override;
-  auto Send(const cpp::span<std::byte>& data) -> void override;
+  auto PrepareFormat(const StreamInfo::Pcm&) -> StreamInfo::Pcm override;
+  auto Configure(const StreamInfo::Pcm& format) -> void override;
 
   I2SAudioOutput(const I2SAudioOutput&) = delete;
   I2SAudioOutput& operator=(const I2SAudioOutput&) = delete;
diff --git a/src/audio/include/sink_mixer.hpp b/src/audio/include/sink_mixer.hpp
new file mode 100644
index 00000000..632ffa2e
--- /dev/null
+++ b/src/audio/include/sink_mixer.hpp
@@ -0,0 +1,88 @@
+/*
+ * Copyright 2023 jacqueline <me@jacqueline.id.au>
+ *
+ * SPDX-License-Identifier: GPL-3.0-only
+ */
+
+#pragma once
+
+#include <sys/_stdint.h>
+#include <cstdint>
+#include <memory>
+
+#include "samplerate.h"
+
+#include "audio_decoder.hpp"
+#include "audio_sink.hpp"
+#include "audio_source.hpp"
+#include "codec.hpp"
+#include "pipeline.hpp"
+#include "stream_info.hpp"
+
+namespace audio {
+
+/*
+ * Handles the final downmix + resample + quantisation stage of audio,
+ * generation sending the result directly to an IAudioSink.
+ */
+class SinkMixer {
+ public:
+  SinkMixer(StreamBufferHandle_t dest);
+  ~SinkMixer();
+
+  auto MixAndSend(InputStream&, const StreamInfo::Pcm&) -> std::size_t;
+
+ private:
+  auto Main() -> void;
+
+  auto SetTargetFormat(const StreamInfo::Pcm& format) -> void;
+  auto HandleBytes() -> void;
+
+  template <typename T>
+  auto ConvertFixedToFloating(InputStream&, OutputStream&) -> void;
+  auto Resample(float, int, InputStream&, OutputStream&) -> void;
+  template <typename T>
+  auto Quantise(InputStream&) -> std::size_t;
+
+  enum class Command {
+    kReadBytes,
+    kSetSourceFormat,
+    kSetTargetFormat,
+  };
+
+  struct Args {
+    Command cmd;
+    StreamInfo::Pcm format;
+  };
+
+  QueueHandle_t commands_;
+  SemaphoreHandle_t is_idle_;
+
+  SRC_STATE* resampler_;
+
+  std::unique_ptr<RawStream> input_stream_;
+  std::unique_ptr<RawStream> floating_point_stream_;
+  std::unique_ptr<RawStream> resampled_stream_;
+
+  cpp::span<std::byte> quantisation_buffer_;
+  cpp::span<short> quantisation_buffer_as_shorts_;
+  cpp::span<int> quantisation_buffer_as_ints_;
+
+  StreamInfo::Pcm target_format_;
+  StreamBufferHandle_t source_;
+  StreamBufferHandle_t sink_;
+};
+
+template <>
+auto SinkMixer::ConvertFixedToFloating<short>(InputStream&, OutputStream&)
+    -> void;
+template <>
+auto SinkMixer::ConvertFixedToFloating<int>(InputStream&, OutputStream&)
+    -> void;
+
+template <>
+auto SinkMixer::Quantise<short>(InputStream&) -> std::size_t;
+template <>
+auto SinkMixer::Quantise<int>(InputStream&) -> std::size_t;
+
+}  // namespace audio
diff --git a/src/audio/include/stream_info.hpp b/src/audio/include/stream_info.hpp
index d48c39a8..d31e035c 100644
--- a/src/audio/include/stream_info.hpp
+++ b/src/audio/include/stream_info.hpp
@@ -56,6 +56,18 @@ class StreamInfo {
     bool operator==(const Encoded&) const = default;
   };
 
+  /*
+   * Two-channel, interleaved, 32-bit floating point pcm samples.
+   */
+  struct FloatingPointPcm {
+    // Number of channels in this stream.
+    uint8_t channels;
+    // The sample rate.
+    uint32_t sample_rate;
+
+    bool operator==(const FloatingPointPcm&) const = default;
+  };
+
   struct Pcm {
     // Number of channels in this stream.
     uint8_t channels;
@@ -64,10 +76,14 @@ class StreamInfo {
     // The sample rate.
     uint32_t sample_rate;
 
+    auto real_bytes_per_sample() const -> uint8_t {
+      return bits_per_sample == 16 ? 2 : 4;
+    }
+
     bool operator==(const Pcm&) const = default;
   };
 
-  typedef std::variant<std::monostate, Encoded, Pcm> Format;
+  typedef std::variant<std::monostate, Encoded, FloatingPointPcm, Pcm> Format;
   auto format() const -> const Format& { return format_; }
   auto set_format(Format f) -> void { format_ = f; }
 
@@ -98,6 +114,12 @@ class RawStream {
 
   auto info() -> StreamInfo& { return info_; }
   auto data() -> cpp::span<std::byte>;
+  template <typename T>
+  auto data_as() -> cpp::span<T> {
+    auto orig = data();
+    return {reinterpret_cast<T*>(orig.data()), orig.size_bytes() / sizeof(T)};
+  }
+  auto empty() const -> bool { return info_.bytes_in_stream() == 0; }
 
  private:
   StreamInfo info_;
@@ -114,6 +136,12 @@ class InputStream {
   const StreamInfo& info() const;
 
   cpp::span<const std::byte> data() const;
+  template <typename T>
+  auto data_as() const -> cpp::span<const T> {
+    auto orig = data();
+    return {reinterpret_cast<const T*>(orig.data()),
+            orig.size_bytes() / sizeof(T)};
+  }
 
  private:
   RawStream* raw_;
@@ -131,6 +159,11 @@ class OutputStream {
   const StreamInfo& info() const;
 
   cpp::span<std::byte> data() const;
+  template <typename T>
+  auto data_as() const -> cpp::span<T> {
+    auto orig = data();
+    return {reinterpret_cast<T*>(orig.data()), orig.size_bytes() / sizeof(T)};
+  }
 
  private:
   RawStream* raw_;
diff --git a/src/audio/sink_mixer.cpp b/src/audio/sink_mixer.cpp
new file mode 100644
index 00000000..072dc9b7
--- /dev/null
+++ b/src/audio/sink_mixer.cpp
@@ -0,0 +1,301 @@
+/*
+ * Copyright 2023 jacqueline <me@jacqueline.id.au>
+ *
+ * SPDX-License-Identifier: GPL-3.0-only
+ */
+
+#include "sink_mixer.hpp"
+
+#include <stdint.h>
+#include <cmath>
+
+#include "esp_heap_caps.h"
+#include "esp_log.h"
+#include "freertos/portmacro.h"
+#include "freertos/projdefs.h"
+#include "samplerate.h"
+
+#include "stream_info.hpp"
+#include "tasks.hpp"
+
+static constexpr char kTag[] = "mixer";
+
+static constexpr std::size_t kSourceBufferLength = 4 * 1024;
+static constexpr std::size_t kInputBufferLength = 4 * 1024;
+static constexpr std::size_t kReformatBufferLength = 4 * 1024;
+static constexpr std::size_t kResampleBufferLength = kReformatBufferLength;
+static constexpr std::size_t kQuantisedBufferLength = 2 * 1024;
+
+namespace audio {
+
+SinkMixer::SinkMixer(StreamBufferHandle_t dest)
+    : commands_(xQueueCreate(1, sizeof(Args))),
+      is_idle_(xSemaphoreCreateBinary()),
+      resampler_(nullptr),
+      source_(xStreamBufferCreate(kSourceBufferLength, 1)),
+      sink_(dest) {
+  input_stream_.reset(new RawStream(kInputBufferLength));
+  floating_point_stream_.reset(new RawStream(kReformatBufferLength));
+  resampled_stream_.reset(new RawStream(kResampleBufferLength));
+
+  quantisation_buffer_ = {
+      reinterpret_cast<std::byte*>(heap_caps_malloc(
+          kQuantisedBufferLength, MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT)),
+      kQuantisedBufferLength};
+  quantisation_buffer_as_ints_ = {
+      reinterpret_cast<int*>(quantisation_buffer_.data()),
+      quantisation_buffer_.size_bytes() / 4};
+  quantisation_buffer_as_shorts_ = {
+      reinterpret_cast<short*>(quantisation_buffer_.data()),
+      quantisation_buffer_.size_bytes() / 2};
+
+  tasks::StartPersistent<tasks::Type::kMixer>([&]() { Main(); });
+}
+
+SinkMixer::~SinkMixer() {
+  vQueueDelete(commands_);
+  vSemaphoreDelete(is_idle_);
+  vStreamBufferDelete(source_);
+  heap_caps_free(quantisation_buffer_.data());
+  if (resampler_ != nullptr) {
+    src_delete(resampler_);
+  }
+}
+
+auto SinkMixer::MixAndSend(InputStream& input, const StreamInfo::Pcm& target)
+    -> std::size_t {
+  if (input.info().format_as<StreamInfo::Pcm>() !=
+      input_stream_->info().format_as<StreamInfo::Pcm>()) {
+    xSemaphoreTake(is_idle_, portMAX_DELAY);
+    Args args{
+        .cmd = Command::kSetSourceFormat,
+        .format = input.info().format_as<StreamInfo::Pcm>().value(),
+    };
+    xQueueSend(commands_, &args, portMAX_DELAY);
+    xSemaphoreGive(is_idle_);
+  }
+  if (target_format_ != target) {
+    xSemaphoreTake(is_idle_, portMAX_DELAY);
+    Args args{
+        .cmd = Command::kSetTargetFormat,
+        .format = target,
+    };
+    xQueueSend(commands_, &args, portMAX_DELAY);
+    xSemaphoreGive(is_idle_);
+  }
+
+  Args args{
+      .cmd = Command::kReadBytes,
+      .format = {},
+  };
+  xQueueSend(commands_, &args, portMAX_DELAY);
+
+  auto buf = input.data();
+  std::size_t bytes_sent =
+      xStreamBufferSend(source_, buf.data(), buf.size_bytes(), portMAX_DELAY);
+  input.consume(bytes_sent);
+  return bytes_sent;
+}
+
+auto SinkMixer::Main() -> void {
+  OutputStream input_receiver{input_stream_.get()};
+  xSemaphoreGive(is_idle_);
+
+  for (;;) {
+    Args args;
+    while (!xQueueReceive(commands_, &args, portMAX_DELAY)) {
+    }
+    switch (args.cmd) {
+      case Command::kSetSourceFormat:
+        ESP_LOGI(kTag, "setting source format");
+        input_receiver.prepare(args.format, {});
+        break;
+      case Command::kSetTargetFormat:
+        ESP_LOGI(kTag, "setting target format");
+        target_format_ = args.format;
+        break;
+      case Command::kReadBytes:
+        xSemaphoreTake(is_idle_, 0);
+        while (!xStreamBufferIsEmpty(source_)) {
+          auto buf = input_receiver.data();
+          std::size_t bytes_received = xStreamBufferReceive(
+              source_, buf.data(), buf.size_bytes(), portMAX_DELAY);
+          input_receiver.add(bytes_received);
+          HandleBytes();
+        }
+        xSemaphoreGive(is_idle_);
+        break;
+    }
+  }
+}
+
+auto SinkMixer::HandleBytes() -> void {
+  InputStream input{input_stream_.get()};
+  auto pcm = input.info().format_as<StreamInfo::Pcm>();
+  if (!pcm) {
+    ESP_LOGE(kTag, "mixer got unsupported data");
+    return;
+  }
+
+  if (*pcm == target_format_) {
+    // The happiest possible case: the input format matches the output
+    // format already. Streams like this should probably have bypassed the
+    // mixer.
+    // TODO(jacqueline): Make this an error; it's slow to use the mixer in this
+    // case, compared to just writing directly to the sink.
+    auto buf = input.data();
+    std::size_t bytes_sent =
+        xStreamBufferSend(sink_, buf.data(), buf.size_bytes(), portMAX_DELAY);
+    input.consume(bytes_sent);
+    return;
+  }
+
+  // Work out the resampling ratio using floating point arithmetic, since
+  // relying on the FPU for this will be much faster, and the difference in
+  // accuracy is unlikely to be noticeable.
+  float src_ratio = static_cast<float>(target_format_.sample_rate) /
+                    static_cast<float>(pcm->sample_rate);
+
+  // Loop until we don't have any complete frames left in the input stream,
+  // where a 'frame' is one complete sample per channel.
+  while (!input_stream_->empty()) {
+    // The first step of both resampling and requantising is to convert the
+    // fixed point pcm input data into 32 bit floating point samples.
+    OutputStream floating_writer{floating_point_stream_.get()};
+    if (pcm->bits_per_sample == 16) {
+      ConvertFixedToFloating<short>(input, floating_writer);
+    } else {
+      // FIXME: We should consider treating 24 bit and 32 bit samples
+      // differently.
+      ConvertFixedToFloating<int>(input, floating_writer);
+    }
+
+    InputStream floating_reader{floating_point_stream_.get()};
+
+    while (!floating_point_stream_->empty()) {
+      RawStream* quantisation_source;
+      if (pcm->sample_rate != target_format_.sample_rate) {
+        // The input data needs to be resampled before being sent to the sink.
+        OutputStream resample_writer{resampled_stream_.get()};
+        Resample(src_ratio, pcm->channels, floating_reader, resample_writer);
+        quantisation_source = resampled_stream_.get();
+      } else {
+        // The input data already has an acceptable sample rate. All we need to
+        // do is quantise it.
+        quantisation_source = floating_point_stream_.get();
+      }
+
+      InputStream quantise_reader{quantisation_source};
+      while (!quantisation_source->empty()) {
+        std::size_t samples_available;
+        if (target_format_.bits_per_sample == 16) {
+          samples_available = Quantise<short>(quantise_reader);
+        } else {
+          samples_available = Quantise<int>(quantise_reader);
+        }
+
+        assert(samples_available * target_format_.real_bytes_per_sample() <=
+               quantisation_buffer_.size_bytes());
+
+        std::size_t bytes_sent = xStreamBufferSend(
+            sink_, quantisation_buffer_.data(),
+            samples_available * target_format_.real_bytes_per_sample(),
+            portMAX_DELAY);
+        assert(bytes_sent ==
+               samples_available * target_format_.real_bytes_per_sample());
+      }
+    }
+  }
+}
+
+template <>
+auto SinkMixer::ConvertFixedToFloating<short>(InputStream& in_str,
+                                              OutputStream& out_str) -> void {
+  auto in = in_str.data_as<short>();
+  auto out = out_str.data_as<float>();
+  std::size_t samples_converted = std::min(in.size(), out.size());
+
+  src_short_to_float_array(in.data(), out.data(), samples_converted);
+
+  in_str.consume(samples_converted * sizeof(short));
+  out_str.add(samples_converted * sizeof(float));
+}
+
+template <>
+auto SinkMixer::ConvertFixedToFloating<int>(InputStream& in_str,
+                                            OutputStream& out_str) -> void {
+  auto in = in_str.data_as<int>();
+  auto out = out_str.data_as<float>();
+  std::size_t samples_converted = std::min(in.size(), out.size());
+
+  src_int_to_float_array(in.data(), out.data(), samples_converted);
+
+  in_str.consume(samples_converted * sizeof(int));
+  out_str.add(samples_converted * sizeof(float));
+}
+
+auto SinkMixer::Resample(float src_ratio,
+                         int channels,
+                         InputStream& in,
+                         OutputStream& out) -> void {
+  if (resampler_ == nullptr || src_get_channels(resampler_) != channels) {
+    if (resampler_ != nullptr) {
+      src_delete(resampler_);
+    }
+
+    ESP_LOGI(kTag, "creating new resampler with %u channels", channels);
+
+    int err = 0;
+    resampler_ = src_new(SRC_LINEAR, channels, &err);
+    assert(resampler_ != NULL);
+    assert(err == 0);
+  }
+
+  auto in_buf = in.data_as<float>();
+  auto out_buf = out.data_as<float>();
+
+  src_set_ratio(resampler_, src_ratio);
+  SRC_DATA args{
+      .data_in = in_buf.data(),
+      .data_out = out_buf.data(),
+      .input_frames = static_cast<long>(in_buf.size()),
+      .output_frames = static_cast<long>(out_buf.size()),
+      .input_frames_used = 0,
+      .output_frames_gen = 0,
+      .end_of_input = 0,
+      .src_ratio = src_ratio,
+  };
+  int err = src_process(resampler_, &args);
+  if (err != 0) {
+    ESP_LOGE(kTag, "resampler error: %s", src_strerror(err));
+  }
+
+  in.consume(args.input_frames_used * sizeof(float));
+  out.add(args.output_frames_gen * sizeof(float));
+}
+
+template <>
+auto SinkMixer::Quantise<short>(InputStream& in) -> std::size_t {
+  auto src = in.data_as<float>();
+  cpp::span<short> dest = quantisation_buffer_as_shorts_;
+  dest = dest.first(std::min(src.size(), dest.size()));
+
+  src_float_to_short_array(src.data(), dest.data(), dest.size());
+
+  in.consume(dest.size() * sizeof(float));
+  return dest.size();
+}
+
+template <>
+auto SinkMixer::Quantise<int>(InputStream& in) -> std::size_t {
+  auto src = in.data_as<float>();
+  cpp::span<int> dest = quantisation_buffer_as_ints_;
+  dest = dest.first(std::min<int>(src.size(), dest.size()));
+
+  src_float_to_int_array(src.data(), dest.data(), dest.size());
+
+  in.consume(dest.size() * sizeof(float));
+  return dest.size();
+}
+
+}  // namespace audio
diff --git a/src/drivers/bluetooth.cpp b/src/drivers/bluetooth.cpp
index 9748522f..a02fa620 100644
--- a/src/drivers/bluetooth.cpp
+++ b/src/drivers/bluetooth.cpp
@@ -5,7 +5,7 @@
 namespace drivers {
 
 auto Bluetooth::Enable() -> Bluetooth* {
-    return nullptr;
+  return nullptr;
 }
 
 }  // namespace drivers
diff --git a/src/drivers/i2s_dac.cpp b/src/drivers/i2s_dac.cpp
index c835fb1f..885321d1 100644
--- a/src/drivers/i2s_dac.cpp
+++ b/src/drivers/i2s_dac.cpp
@@ -161,7 +161,6 @@ auto I2SDac::Reconfigure(Channels ch, BitsPerSample bps, SampleRate rate)
       word_length = 0b10;
       break;
     case BPS_32:
-      // TODO(jacqueline): Error on this? It's not supported anymore.
       slot_config_.data_bit_width = I2S_DATA_BIT_WIDTH_32BIT;
       slot_config_.ws_width = 32;
       word_length = 0b11;
diff --git a/src/drivers/include/bluetooth.hpp b/src/drivers/include/bluetooth.hpp
index f3a4b2ac..22b58c8b 100644
--- a/src/drivers/include/bluetooth.hpp
+++ b/src/drivers/include/bluetooth.hpp
@@ -6,14 +6,15 @@
 namespace drivers {
 
 class Bluetooth {
-    public:
-    static auto Enable() -> Bluetooth*;
-    Bluetooth();
-    ~Bluetooth();
+ public:
+  static auto Enable() -> Bluetooth*;
+  Bluetooth();
+  ~Bluetooth();
 
-    struct Device {};
-    auto Scan() -> std::vector<Device>;
-    private:
-  };
+  struct Device {};
+  auto Scan() -> std::vector<Device>;
 
-}
+ private:
+};
+
+}  // namespace drivers
diff --git a/src/drivers/include/i2s_dac.hpp b/src/drivers/include/i2s_dac.hpp
index 06c0dc16..889ba68c 100644
--- a/src/drivers/include/i2s_dac.hpp
+++ b/src/drivers/include/i2s_dac.hpp
@@ -51,14 +51,12 @@ class I2SDac {
     BPS_32 = I2S_DATA_BIT_WIDTH_32BIT,
   };
   enum SampleRate {
-    SAMPLE_RATE_11_025 = 11025,
-    SAMPLE_RATE_16 = 16000,
-    SAMPLE_RATE_22_05 = 22050,
+    SAMPLE_RATE_8 = 8000,
     SAMPLE_RATE_32 = 32000,
     SAMPLE_RATE_44_1 = 44100,
     SAMPLE_RATE_48 = 48000,
+    SAMPLE_RATE_88_2 = 88200,
     SAMPLE_RATE_96 = 96000,
-    SAMPLE_RATE_192 = 192000,
   };
 
   auto Reconfigure(Channels ch, BitsPerSample bps, SampleRate rate) -> void;
diff --git a/src/system_fsm/include/system_fsm.hpp b/src/system_fsm/include/system_fsm.hpp
index dc188780..d30a712c 100644
--- a/src/system_fsm/include/system_fsm.hpp
+++ b/src/system_fsm/include/system_fsm.hpp
@@ -13,10 +13,10 @@
 #include "database.hpp"
 #include "display.hpp"
 #include "gpios.hpp"
+#include "nvs.hpp"
 #include "relative_wheel.hpp"
 #include "samd.hpp"
 #include "storage.hpp"
-#include "nvs.hpp"
 #include "tag_parser.hpp"
 #include "tinyfsm.hpp"
 #include "touchwheel.hpp"
diff --git a/src/tasks/tasks.cpp b/src/tasks/tasks.cpp
index 861c7bf0..34c690f3 100644
--- a/src/tasks/tasks.cpp
+++ b/src/tasks/tasks.cpp
@@ -34,6 +34,10 @@ auto Name<Type::kAudio>() -> std::string {
   return "AUDIO";
 }
 template <>
+auto Name<Type::kMixer>() -> std::string {
+  return "MIXER";
+}
+template <>
 auto Name<Type::kDatabase>() -> std::string {
   return "DB";
 }
@@ -77,6 +81,14 @@ auto AllocateStack<Type::kFileStreamer>() -> cpp::span<StackType_t> {
           size};
 }
 
+template <>
+auto AllocateStack<Type::kMixer>() -> cpp::span<StackType_t> {
+  std::size_t size = 4 * 1024;
+  return {static_cast<StackType_t*>(
+              heap_caps_malloc(size, MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT)),
+          size};
+}
+
 // Leveldb is designed for non-embedded use cases, where stack space isn't so
 // much of a concern. It therefore uses an eye-wateringly large amount of stack.
 template <>
@@ -105,6 +117,10 @@ auto Priority() -> UBaseType_t;
 // Realtime audio is the entire point of this device, so give this task the
 // highest priority.
 template <>
+auto Priority<Type::kMixer>() -> UBaseType_t {
+  return 12;
+}
+template <>
 auto Priority<Type::kAudio>() -> UBaseType_t {
   return 11;
 }
diff --git a/src/tasks/tasks.hpp b/src/tasks/tasks.hpp
index 742bb3cc..1321aab8 100644
--- a/src/tasks/tasks.hpp
+++ b/src/tasks/tasks.hpp
@@ -36,6 +36,8 @@ enum class Type {
   kFileStreamer,
   // The main audio pipeline task.
   kAudio,
+  // TODO
+  kMixer,
   // Task for running database queries.
   kDatabase,
   // Task for internal database operations