AudioIO.h

Go to the documentation of this file.

#pragma once
#include "AudioTools/CoreAudio/AudioOutput.h"
#include "AudioTools/CoreAudio/AudioStreams.h"
 
#ifndef MAX_ZERO_READ_COUNT
#define MAX_ZERO_READ_COUNT 3
#endif
 
#ifndef CHANNEL_SELECT_BUFFER_SIZE
#define CHANNEL_SELECT_BUFFER_SIZE 256
#endif
 
namespace audio_tools {
template <class T>
class TransformationReader {
 public:
  void begin(T* transform, Stream* source) {
    TRACED();
    active = true;
    is_eof = false;
    total_bytes_read = 0;
    last_setup_buffer_size = 0;
    p_stream = source;
    p_transform = transform;
    if (transform == nullptr) {
      LOGE("transform is NULL");
      active = false;
    }
    if (p_stream == nullptr) {
      LOGE("p_stream is NULL");
      active = false;
    }
  }
 
  size_t readBytes(uint8_t* data, size_t len) {
    LOGD("TransformationReader::readBytes: %d", (int)len);
    if (!active) {
      LOGE("inactive");
      return 0;
    }
    if (p_stream == nullptr) {
      LOGE("p_stream is NULL");
      return 0;
    }
 
    setupBuffers(len);
 
    if (!is_eof) {
      fillResultQueue(len);
    }
 
    int result_len = min((int)len, result_queue.available());
    result_len = result_queue.readBytes(data, result_len);
    LOGD("TransformationReader::readBytes: %d -> %d", (int)len, result_len);
    total_bytes_read += result_len;
    return result_len;
  }
 
  int available() {
    if (!active || p_stream == nullptr || p_transform == nullptr) return 0;
    if (is_eof) return result_queue.available();
    setupBuffers(max_read_size);
    fillResultQueue(max_read_size);
    int available_bytes = result_queue.available();
    LOGD("TransformationReader::available: %d", available_bytes);
    return available_bytes > max_read_size ? max_read_size : available_bytes;
  }
 
  void end() {
    result_queue_buffer.resize(0);
    buffer.resize(0);
    total_bytes_read = 0;
    active = false;
  }
 
  void setResultQueueFactor(int factor) { result_queue_factor = factor; }
 
  void resizeResultQueue(int size) {
    result_queue_buffer.resize(size);
    result_queue.begin();
  }
 
  void setMaxReadSize(int size) { max_read_size = size; }
 
  size_t getTotalBytesRead() const { return total_bytes_read; }
 
  void setEofOnZeroReads(bool flag) { eof_on_zero_reads = flag; }
 
 protected:
  RingBuffer<uint8_t> result_queue_buffer{0};
  QueueStream<uint8_t> result_queue{result_queue_buffer};  //
  Stream* p_stream = nullptr;
  Vector<uint8_t> buffer{0};  // we allocate memory only when needed
  T* p_transform = nullptr;
  bool active = false;
  int result_queue_factor = 5;
  int max_read_size = DEFAULT_BUFFER_SIZE;
  int last_setup_buffer_size = 0;
  float last_byte_factor = 0.0f;
  bool is_eof = false;
  bool eof_on_zero_reads = true;
  size_t total_bytes_read = 0;
 
  void resizeReadBuffer(int size) { buffer.resize(size); }
 
  void setupBuffers(size_t len) {
    float byte_factor = p_transform->getByteFactor();
    if (byte_factor <= 0.0f) {
      LOGE("Invalid byte factor: %f", byte_factor);
      byte_factor = 1.0f;
    }
    // Recompute if the requested length changed, or if the transform's byte
    // factor drifted meaningfully since the read chunk size was last sized
    // (e.g. a live-adjusted resampling step size). For a transform with a
    // constant byte factor (the common decoder/encoder case) this check
    // never re-triggers after the first call, so behavior there is
    // unchanged; it only matters for transforms whose byte factor changes
    // at runtime, where a stale chunk size would otherwise silently distort
    // the consumption/production ratio.
    bool byte_factor_changed =
        fabsf(byte_factor - last_byte_factor) > 0.01f * last_byte_factor;
    if (len == last_setup_buffer_size && !byte_factor_changed) return;
    LOGD("setupBuffers: %d", (int)len);
    last_byte_factor = byte_factor;
 
    // we read half the necessary bytes
    int size = (0.5f / byte_factor * len);
    // process full samples/frames
    size = size / 4 * 4;
    if (size <= 0) size = 4;
    if (buffer.size() < size) {
      LOGI("read size: %d", size);
      buffer.resize(size);
    }
 
    if (result_queue_buffer.size() == 0) {
      // make sure that the ring buffer is big enough
      int rb_size = len * result_queue_factor;
      LOGI("buffer size: %d", rb_size);
      result_queue_buffer.resize(rb_size);
      result_queue.begin();
    }
    last_setup_buffer_size = len;
  }
 
  void fillResultQueue(size_t len) {
    if (is_eof) return;
    if (result_queue.available() >= len) return;
    LOGD("fillResultQueue: %d", (int)len);
 
    // Detect misconfigured buffer: if the ring buffer capacity is smaller than
    // the requested len bytes we can never satisfy the condition and will loop
    // forever. Issue an error and bail out early.
    if ((int)len > result_queue_buffer.size()) {
      LOGE("fillResultQueue: result_queue_buffer too small: %d < %d. "
           "Increase result_queue_factor or call resizeReadResultQueue().",
           result_queue_buffer.size(), (int)len);
      return;
    }
 
    Print* tmp = setupOutput();
    int zero_count = 0;
    while (result_queue.available() < len) {
      // Detect buffer-full stall: if we can't write any more data but the
      // queue is still below len, we must stop to avoid an endless loop.
      if (result_queue.available() >= result_queue_buffer.size()) {
        LOGE("fillResultQueue: result_queue full (%d) but target not reached "
             "(%d/%d). Increase result_queue_factor or call "
             "resizeReadResultQueue().",
             result_queue_buffer.size(), result_queue.available(), (int)len);
        break;
      }
      int read_size = buffer.size();
      int read_eff = p_stream->readBytes(buffer.data(), read_size);
      LOGD("readBytes from source: %d -> %d", read_size,read_eff);
      if (read_eff > 0) {
        zero_count = 0;  // reset 0 count
        if (read_eff != buffer.size()) {
          LOGD("readBytes %d -> %d", buffer.size(), read_eff);
        }
        int write_eff = p_transform->write(buffer.data(), read_eff);
        if (write_eff != read_eff) {
          LOGE("TransformationReader::write %d -> %d", read_eff, write_eff);
        }
      } else {
        // limit the number of reads which provide 0;
        if (++zero_count > MAX_ZERO_READ_COUNT) {
          if (eof_on_zero_reads) {
            is_eof = true;
            // Flush any buffered/final encoder bytes into result_queue.
            p_transform->flush();
          }
          // Otherwise the source is just a momentarily empty live
          // producer (buffer underflow, not end of stream): stop trying
          // for this call, but leave is_eof false so the next call
          // retries once more data has arrived.
          break;
        }
        // wait for some more data
        delay(5);
      }
    }
    LOGD("fillResultQueue available: %d", result_queue.available());
    restoreOutput(tmp);
  }
 
  Print* setupOutput() {
    Print* result = p_transform->getPrint();
    p_transform->setOutput((Print&)result_queue);
 
    return result;
  }
  void restoreOutput(Print* out) {
    if (out) p_transform->setOutput(*out);
  }
};
 
class ReformatBaseStream : public ModifyingStream {
 public:
  virtual void setStream(Stream& stream) override {
    TRACED();
    p_stream = &stream;
    p_print = &stream;
  }
 
  virtual void setStream(AudioStream& stream) {
    TRACED();
    p_stream = &stream;
    p_print = &stream;
    // setNotifyOnOutput(stream);
    addNotifyAudioChange(stream);
  }
 
  virtual void setOutput(AudioOutput& print) {
    TRACED();
    p_print = &print;
    addNotifyAudioChange(print);
  }
 
  virtual void setOutput(Print& print) override {
    TRACED();
    p_print = &print;
  }
 
  virtual Print* getPrint() { return p_print; }
 
  virtual Stream* getStream() { return p_stream; }
 
  size_t readBytes(uint8_t* data, size_t len) override {
    LOGD("ReformatBaseStream::readBytes: %d", (int)len);
    return reader.readBytes(data, len);
  }
 
  int available() override {
    return reader.available();
  }
 
  int availableForWrite() override {
    return DEFAULT_BUFFER_SIZE;  // reader.availableForWrite();
  }
 
  virtual float getByteFactor() = 0;
 
  virtual void flush() {}
 
  void end() override {
    TRACED();
    AudioStream::end();
    reader.end();
  }
 
  void resizeReadResultQueue(int size) { reader.resizeResultQueue(size); }
 
  void setReadResultQueueSize(int size) { reader.resizeResultQueue(size); }
 
  void setMaxReadSize(int size) { reader.setMaxReadSize(size); }
 
  virtual TransformationReader<ReformatBaseStream>& transformationReader() {
    return reader;
  }
 
 
 protected:
  TransformationReader<ReformatBaseStream> reader;
  Stream* p_stream = nullptr;
  Print* p_print = nullptr;
 
  void setupReader() {
    if (getStream() != nullptr) {
      reader.begin(this, getStream());
    }
  }
};
 
class AudioOutputAdapter : public AudioOutput {};
 
class AdapterPrintToAudioOutput : public AudioOutputAdapter {
 public:
  AdapterPrintToAudioOutput() = default;
  AdapterPrintToAudioOutput(Print& print) { setStream(print); }
  void setStream(Print& out) { p_print = &out; }
  void setAudioInfo(AudioInfo info) { cfg = info; }
  size_t write(const uint8_t* data, size_t len) {
    return p_print->write(data, len);
  }
  virtual bool isDeletable() { return true; }
 
  AudioInfo audioInfo() { return cfg; }
 
 protected:
  Print* p_print = nullptr;
  AudioInfo cfg;
};
 
class AdapterAudioStreamToAudioOutput : public AudioOutputAdapter {
 public:
  AdapterAudioStreamToAudioOutput() = default;
 
  AdapterAudioStreamToAudioOutput(AudioStream& stream) { setStream(stream); }
 
  void setStream(AudioStream& stream) { p_stream = &stream; }
 
  void setAudioInfo(AudioInfo info) override { p_stream->setAudioInfo(info); }
 
  AudioInfo audioInfo() override { return p_stream->audioInfo(); }
 
  size_t write(const uint8_t* data, size_t len) override {
    return p_stream->write(data, len);
  }
 
  int availableForWrite() override { return p_stream->availableForWrite(); }
 
  bool begin() override { return p_stream->begin(); }
 
  void end() override { p_stream->end(); }
 
  virtual bool isDeletable() override { return true; }
 
  operator bool() override { return *p_stream; }
 
 protected:
  AudioStream* p_stream = nullptr;
};
 
class AdapterAudioOutputToAudioStream : public AudioStream {
 public:
  AdapterAudioOutputToAudioStream() = default;
 
  AdapterAudioOutputToAudioStream(AudioOutput& stream) { setOutput(stream); }
 
  void setOutput(AudioOutput& stream) { p_stream = &stream; }
 
  void setAudioInfo(AudioInfo info) override { p_stream->setAudioInfo(info); }
 
  AudioInfo audioInfo() override { return p_stream->audioInfo(); }
 
  size_t write(const uint8_t* data, size_t len) override {
    return p_stream->write(data, len);
  }
 
  bool begin() override { return p_stream->begin(); }
 
  void end() override { p_stream->end(); }
 
  virtual bool isDeletable() { return true; }
 
  operator bool() override { return *p_stream; }
 
 protected:
  AudioOutput* p_stream = nullptr;
};
 
class MultiOutput : public ModifyingOutput {
 public:
  MultiOutput() = default;
 
  MultiOutput(Print& out) { add(out); }
 
  MultiOutput(AudioOutput& out) { add(out); }
 
  MultiOutput(AudioStream& out) { add(out); }
 
  MultiOutput(AudioOutput& out1, AudioOutput& out2) {
    add(out1);
    add(out2);
  }
 
  MultiOutput(AudioStream& out1, AudioStream& out2) {
    add(out1);
    add(out2);
  }
 
  MultiOutput(Print& out1, Print& out2) {
    add(out1);
    add(out2);
  }
 
  virtual ~MultiOutput() { clear(); }
 
  void add(AudioOutput& out) { vector.push_back(&out); }
 
  void add(AudioStream& stream) {
    AdapterAudioStreamToAudioOutput* out =
        new AdapterAudioStreamToAudioOutput(stream);
    vector.push_back(out);
  }
 
  void add(Print& print) {
    AdapterPrintToAudioOutput* out = new AdapterPrintToAudioOutput(print);
    vector.push_back(out);
  }
 
  void flush() {
    for (int j = 0; j < vector.size(); j++) {
      vector[j]->flush();
    }
  }
 
  void setAudioInfo(AudioInfo info) {
    for (int j = 0; j < vector.size(); j++) {
      vector[j]->setAudioInfo(info);
    }
  }
 
  size_t write(const uint8_t* data, size_t len) override {
    for (auto& out : vector) {
      int open = len;
      int start = 0;
      // create copy of data to avoid that one output changes the data for the
      // other outputs
      uint8_t copy[len];
      memcpy(copy, data, len);
      while (open > 0) {
        int written = out->write(copy + start, open);
        open -= written;
        start += written;
      }
    }
    return len;
  }
 
  size_t write(uint8_t ch) override {
    for (int j = 0; j < vector.size(); j++) {
      int open = 1;
      while (open > 0) {
        open -= vector[j]->write(ch);
      }
    }
    return 1;
  }
 
  void clear() {
    for (auto& tmp : vector) {
      if (tmp != nullptr && tmp->isDeletable()) {
        delete tmp;
      }
    }
    vector.clear();
  }
 
 protected:
  Vector<AudioOutput*> vector;
 
  void setOutput(Print& out) { add(out); }
};
 
class TimedStream : public ModifyingStream {
 public:
  TimedStream() = default;
 
  TimedStream(AudioStream& io, long startSeconds = 0, long endSeconds = -1) {
    p_stream = &io;
    p_print = &io;
    p_info = &io;
    setStartSec(startSeconds);
    setEndSec(endSeconds);
  }
 
  TimedStream(AudioOutput& o, long startSeconds = 0, long endSeconds = -1) {
    p_print = &o;
    p_info = &o;
    setStartSec(startSeconds);
    setEndSec(endSeconds);
  }
 
  TimedStream(Stream& io, long startSeconds = 0, long endSeconds = -1) {
    p_stream = &io;
    p_print = &io;
    setStartSec(startSeconds);
    setEndSec(endSeconds);
  }
 
  TimedStream(Print& o, long startSeconds = 0, long endSeconds = -1) {
    p_print = &o;
    setStartSec(startSeconds);
    setEndSec(endSeconds);
  }
 
  void setStartSec(uint32_t startSeconds) {
    start_ms = startSeconds * 1000;
    calculateByteLimits();
  }
 
  void setStartMs(uint32_t ms) {
    start_ms = ms;
    calculateByteLimits();
  }
 
  void setEndSec(uint32_t endSeconds) {
    end_ms = endSeconds * 1000;
    calculateByteLimits();
  }
 
  void setEndMs(uint32_t ms) {
    end_ms = ms;
    calculateByteLimits();
  }
 
  bool isPlaying() {
    if (current_bytes < start_bytes) return false;
    if (end_bytes > 0 && current_bytes > end_bytes) return false;
    return true;
  }
 
  bool isActive() {
    return (current_bytes < end_bytes && current_bytes >= start_bytes);
  }
 
  bool begin(AudioInfo info) {
    setAudioInfo(info);
    return begin();
  }
 
  bool begin() override {
    calculateByteLimits();
    current_bytes = 0;
    LOGI("byte range %u - %u", (unsigned)start_bytes, (unsigned)end_bytes);
    return true;
  }
 
  operator bool() override { return isActive(); }
 
  size_t readBytes(uint8_t* data, size_t len) override {
    // if reading is not supported we stop
    if (p_stream == nullptr) return 0;
    // Positioin to start
    if (start_bytes > current_bytes) {
      consumeBytes(start_bytes - current_bytes);
    }
    // if we are past the end we stop
    if (!isActive()) return 0;
    // read the data now
    size_t result = 0;
    do {
      result = p_stream->readBytes(data, len);
      current_bytes += len;
      // ignore data before start time
    } while (result > 0 && current_bytes < start_bytes);
    return isPlaying() ? result : 0;
  }
 
  size_t write(const uint8_t* data, size_t len) override {
    if (current_bytes >= end_bytes) return 0;
    current_bytes += len;
    if (current_bytes < start_bytes) return len;
    return p_print->write(data, len);
  }
 
  int available() override {
    if (p_stream == nullptr) return 0;
    return current_bytes < end_bytes ? p_stream->available() : 0;
  }
 
  void setAudioInfo(AudioInfo info) override {
    AudioStream::setAudioInfo(info);
    if (p_info) p_info->setAudioInfo(info);
    calculateByteLimits();
  }
 
  int availableForWrite() override {
    return current_bytes < end_bytes ? p_print->availableForWrite() : 0;
  }
 
  void setCompressionRatio(float ratio) { compression_ratio = ratio; }
 
  int bytesPerSecond() {
    return info.sample_rate * info.channels * info.bits_per_sample / 8;
  }
 
  void setOutput(Print& out) override { p_print = &out; }
 
  void setStream(Stream& stream) override {
    p_print = &stream;
    p_stream = &stream;
  }
 
  void setOutput(AudioOutput& out) {
    p_print = &out;
    p_info = &out;
  }
 
  void setStream(AudioOutput& out) {
    p_print = &out;
    p_info = &out;
  }
 
  void setStream(AudioStream& stream) {
    p_print = &stream;
    p_stream = &stream;
    p_info = &stream;
  }
 
  size_t size() { return end_bytes - start_bytes; }
 
 protected:
  Stream* p_stream = nullptr;
  Print* p_print = nullptr;
  AudioInfoSupport* p_info = nullptr;
  uint32_t start_ms = 0;
  uint32_t end_ms = UINT32_MAX;
  uint32_t start_bytes = 0;
  uint32_t end_bytes = UINT32_MAX;
  uint32_t current_bytes = 0;
  float compression_ratio = 1.0;
 
  void consumeBytes(uint32_t len) {
    int open = len;
    uint8_t buffer[1024];
    while (open > 0) {
      int toread = min(1024, open);
      p_stream->readBytes(buffer, toread);
      open -= toread;
    }
    current_bytes += len;
    LOGD("consumed %u -> %u", (unsigned)len, (unsigned)current_bytes);
  }
 
  void calculateByteLimits() {
    float bytes_per_second = bytesPerSecond();
    if (bytes_per_second > 0) {
      start_bytes = bytes_per_second * start_ms / compression_ratio / 1000;
      end_bytes = bytes_per_second * end_ms / compression_ratio / 1000;
    } else {
      LOGE("AudioInfo not defined");
    }
  }
};
 
class ChannelsSelectOutput : public AudioOutput {
 public:
  ChannelsSelectOutput() = default;
 
  bool begin(AudioInfo info) override {
    setAudioInfo(info);
    return begin();
  }
 
  bool begin() override {
    AudioOutput::begin();
    // make sure that selected channels are valid
    for (auto& out : out_channels) {
      for (auto& ch : out.channels) {
        if (ch > cfg.channels - 1) {
          LOGE("Channel '%d' not valid for max %d channels", ch, cfg.channels);
          return false;
        }
      }
    }
    return true;
  }
 
  void addOutput(AudioOutput& out, uint16_t channel) {
    Vector<uint16_t> channels;
    channels.push_back(channel);
    ChannelSelectionOutputDef def;
    def.channels = channels;
    def.p_out = &out;
    def.p_audio_info = &out;
    out_channels.push_back(def);
  }
 
  void addOutput(AudioStream& out, uint16_t channel) {
    Vector<uint16_t> channels;
    channels.push_back(channel);
    ChannelSelectionOutputDef def;
    def.channels = channels;
    def.p_out = &out;
    def.p_audio_info = &out;
    out_channels.push_back(def);
  }
 
  void addOutput(Print& out, uint16_t channel) {
    Vector<uint16_t> channels;
    channels.push_back(channel);
    ChannelSelectionOutputDef def;
    def.channels = channels;
    def.p_out = &out;
    out_channels.push_back(def);
  }
 
  void addOutput(Print& out, uint16_t left, uint16_t right) {
    Vector<uint16_t> channels;
    channels.push_back(left);
    channels.push_back(right);
    ChannelSelectionOutputDef def;
    def.channels = channels;
    def.p_out = &out;
    out_channels.push_back(def);
  }
 
  void addOutput(AudioOutput& out, uint16_t left, uint16_t right) {
    Vector<uint16_t> channels;
    channels.push_back(left);
    channels.push_back(right);
    ChannelSelectionOutputDef def;
    def.channels = channels;
    def.p_out = &out;
    def.p_audio_info = &out;
    out_channels.push_back(def);
  }
 
  void addOutput(AudioStream& out, uint16_t left, uint16_t right) {
    Vector<uint16_t> channels;
    channels.push_back(left);
    channels.push_back(right);
    ChannelSelectionOutputDef def;
    def.channels = channels;
    def.p_out = &out;
    def.p_audio_info = &out;
    out_channels.push_back(def);
  }
 
  size_t write(const uint8_t* data, size_t len) override {
    if (!is_active) return false;
    LOGD("write %d", (int)len);
    switch (cfg.bits_per_sample) {
      case 16:
        return writeT<int16_t>(data, len);
      case 24:
        return writeT<int24_t>(data, len);
      case 32:
        return writeT<int32_t>(data, len);
      default:
        return 0;
    }
  }
 
  void setAudioInfo(AudioInfo ai) override {
    this->cfg = ai;
    // notifyAudioChange(ai);
    for (auto& info : out_channels) {
      auto p_notify = info.p_audio_info;
      if (p_notify != nullptr) {
        AudioInfo result{ai};
        result.channels = info.channels.size();
        p_notify->setAudioInfo(result);
      }
    }
  }
 
 protected:
  struct ChannelSelectionOutputDef {
    Print* p_out = nullptr;
    AudioInfoSupport* p_audio_info = nullptr;
    SingleBuffer<uint8_t> buffer{CHANNEL_SELECT_BUFFER_SIZE};
    Vector<uint16_t> channels{0};
  };
  Vector<ChannelSelectionOutputDef> out_channels{0};
 
  template <typename T>
  size_t writeT(const uint8_t* buffer, size_t size) {
    if (!is_active) return 0;
    int sample_count = size / sizeof(T);
    // int result_size = sample_count / cfg.channels;
    T* data = (T*)buffer;
 
    for (int i = 0; i < sample_count; i += cfg.channels) {
      T* frame = data + i;
      for (auto& out : out_channels) {
        T out_frame[out.channels.size()];
        int ch_out = 0;
        for (auto& ch : out.channels) {
          // make sure we have a valid channel
          int channel = (ch < cfg.channels) ? ch : cfg.channels - 1;
          out_frame[ch_out++] = frame[channel];
        }
        // write to buffer
        size_t written = out.buffer.writeArray((const uint8_t*)&out_frame,
                                               sizeof(out_frame));
        // write buffer to final output
        if (out.buffer.availableForWrite() < sizeof(out_frame)) {
          out.p_out->write(out.buffer.data(), out.buffer.available());
          out.buffer.reset();
        }
        // if (written != sizeof(out_frame)) {
        //   LOGW("Could not write all samples %d -> %d", sizeof(out_frame),
        //   written);
        // }
      }
    }
    return size;
  }
 
  int getChannels(Print* out, int defaultChannels) {
    for (auto& channels_select : out_channels) {
      if (channels_select.p_out == out) return channels_select.channels.size();
    }
    return defaultChannels;
  }
};
 
}  // namespace audio_tools