AudioStreams.h

Go to the documentation of this file.

#pragma once
#include "AudioToolsConfig.h"
#include "AudioTools/CoreAudio/AudioEffects/SoundGenerator.h"
#include "AudioTools/CoreAudio/AudioLogger.h"
#include "AudioTools/CoreAudio/AudioOutput.h"
#include "AudioTools/CoreAudio/AudioTimer/AudioTimer.h"
#include "AudioTools/CoreAudio/AudioTypes.h"
#include "AudioTools/CoreAudio/BaseConverter.h"
#include "AudioTools/CoreAudio/BaseStream.h"
#include "AudioTools/CoreAudio/Buffers.h"
 
#ifndef IRAM_ATTR
#define IRAM_ATTR
#endif
 
namespace audio_tools {

class AudioStreamWrapper : public AudioStream {
 public:
  AudioStreamWrapper(Stream &s) {
    TRACED();
    p_stream = &s;
    p_stream->setTimeout(clientTimeout);
  }
 
  virtual bool begin() { return true; }
  virtual void end() {}
 
  virtual size_t readBytes(uint8_t *data, size_t len) {
    // Serial.print("Timeout audiostream: ");
    // Serial.println(p_stream->getTimeout());
    return p_stream->readBytes(data, len);
  }
 
  int read() { return p_stream->read(); }
 
  int peek() { return p_stream->peek(); }
 
  int available() { return p_stream->available(); }
 
  virtual size_t write(uint8_t c) { return p_stream->write(c); }
 
  virtual size_t write(const uint8_t *data, size_t len) {
    return p_stream->write(data, len);
  }
 
  virtual int availableForWrite() { return p_stream->availableForWrite(); }
 
  virtual void flush() { p_stream->flush(); }
 
 protected:
  Stream *p_stream;
  int32_t clientTimeout = URL_CLIENT_TIMEOUT;  // 10000;
};

 
class ModifyingStream : public AudioStream {
 public:
  virtual void setStream(Stream &in) = 0;
  virtual void setOutput(Print &out) = 0;

  virtual void setStream(AudioStream &io) {
    setStream(static_cast<Stream&>(io));
    addNotifyAudioChange(io);
  }

  virtual void setOutput(AudioOutput &out) {
    setOutput(static_cast<Print&>(out));
    addNotifyAudioChange(out);
  }
};

class MemoryStream : public AudioStream {
 public:
  // Default constructor
  MemoryStream() = default;
  MemoryStream(int buffer_size, MemoryType memoryType) {
    LOGD("MemoryStream: %d", buffer_size);
    this->buffer_size = buffer_size;
    this->memory_type = memoryType;
    resize(buffer_size);
    info.clear();  // mark audio info as unknown
  }

  MemoryStream(const uint8_t *buffer, int buffer_size, bool isActive = true,
               MemoryType memoryType = FLASH_RAM) {
    LOGD("MemoryStream: %d", buffer_size);
    owns_memory = false;
    setValue(buffer, buffer_size, memoryType);
    is_active = isActive;
    info.clear();  // mark audio info as unknown
  }

  MemoryStream(MemoryStream &source) : AudioStream() { copy(source); }

  MemoryStream(MemoryStream &&source) {
    setValue(source.buffer, source.buffer_size, source.memory_type);
    // clear source data
    source.setValue(nullptr, 0, source.memory_type);
  }
 
  ~MemoryStream() {
    TRACED();
    if (memoryCanChange() && owns_memory && buffer != nullptr) free(buffer);
  }

  MemoryStream &operator=(MemoryStream &other) {
    copy(other);
    return *this;
  }

  operator bool() override { return available() > 0; }

  bool begin(AudioInfo info) {
    this->info = info;
    return begin();
  }

  bool begin() override {
    TRACED();
    write_pos = memoryCanChange() ? 0 : buffer_size;
    if (this->buffer == nullptr && owns_memory && memoryCanChange()) {
      resize(buffer_size);
    }
    read_pos = 0;
    is_active = this->buffer != nullptr;
    return is_active;
  }

  void rewind() {
    if (buffer != nullptr && buffer_size > 0) {
      read_pos = 0;
    }
  }
 
  virtual size_t write(uint8_t byte) override {
    if (!is_active) return 0;
    if (memory_type == FLASH_RAM) return 0;
    if (buffer == nullptr) return 0;
    int result = 0;
    if (write_pos < buffer_size) {
      result = 1;
      buffer[write_pos] = byte;
      write_pos++;
    }
    return result;
  }
 
  virtual size_t write(const uint8_t *data, size_t len) override {
    if (!is_active) return 0;
    if (memory_type == FLASH_RAM) return 0;
    size_t result = 0;
    for (size_t j = 0; j < len; j++) {
      if (!write(data[j])) {
        break;
      }
      result = j + 1;
    }
    return result;
  }
 
  virtual int available() override {
    if (!is_active) return 0;
    if (buffer == nullptr) return 0;
    int result = write_pos - read_pos;
    if (result <= 0 && is_loop) {
      // rewind to start
      read_pos = rewind_pos;
      result = write_pos - read_pos;
      // call callback
      if (rewind_cb != nullptr) rewind_cb();
    }
    return is_loop ? DEFAULT_BUFFER_SIZE : result;
  }
 
  virtual int availableForWrite() override {
    if (!is_active) return 0;
    if (memory_type == FLASH_RAM) return 0;
    return buffer_size - write_pos;
  }
 
  virtual int read() override {
    int result = peek();
    if (result >= 0) {
      read_pos++;
    }
    return result;
  }
 
  virtual size_t readBytes(uint8_t *data, size_t len) override {
    if (!is_active) return 0;
    size_t count = 0;
    while (count < len) {
      int c = read();
      if (c < 0) break;
      *data++ = (char)c;
      count++;
    }
    return count;
  }
 
  virtual int peek() override {
    if (!is_active) return -1;
    int result = -1;
    if (available() > 0) {
      result = buffer[read_pos];
    }
    return result;
  }
 
  virtual void flush() override {}
 
  virtual void end() override {
    read_pos = 0;
    is_active = false;
  }

  virtual void clear(bool reset = false) {
    if (memoryCanChange()) {
      write_pos = 0;
      read_pos = 0;
      if (owns_memory && buffer == nullptr) {
        resize(buffer_size);
      }
      if (reset) {
        // we clear the buffer data
        memset(buffer, 0, buffer_size);
      }
    } else {
      read_pos = 0;
      LOGW("data is read only");
    }
  }

  virtual void setLoop(bool loop) {
    is_loop = loop;
    rewind_pos = 0;
    if (buffer != nullptr && buffer_size > 12) {
      if (memcmp("WAVE", buffer + 8, 4) == 0) {
        rewind_pos = 44;
      }
    }
  }

  virtual void setLoop(bool loop, int rewindPos) {
    is_loop = loop;
    rewind_pos = rewindPos;
  }

  virtual bool resize(size_t size) {
    if (!memoryCanChange()) return false;
    if (!owns_memory) return false;
 
    buffer_size = size;
    switch (memory_type) {
#if defined(USE_PSRAM) && defined(ARDUINO)
      case PS_RAM:
        buffer = (buffer == nullptr) ? (uint8_t *)ps_calloc(size, 1)
                                     : (uint8_t *)ps_realloc(buffer, size);
        break;
#endif
      default:
        buffer = (buffer == nullptr) ? (uint8_t *)calloc(size, 1)
                                     : (uint8_t *)realloc(buffer, size);
        break;
    }
    return buffer != nullptr;
  }

  virtual uint8_t *data() { return buffer; }

  virtual bool setAvailable(size_t len) { 
    if (len <= buffer_size) {
      write_pos = len;
      if (read_pos > write_pos) read_pos = write_pos;
      return true;
    }
    return false;
  }

  void setRewindCallback(void (*cb)()) { this->rewind_cb = cb; }

  void setValue(const uint8_t *buffer, int buffer_size,
                MemoryType memoryType = FLASH_RAM) {
    this->buffer_size = buffer_size;
    this->read_pos = 0;
    this->write_pos = buffer_size;
    this->buffer = (uint8_t *)buffer;
    this->memory_type = memoryType;
  }
 
 protected:
  int write_pos = 0;
  int read_pos = 0;
  int buffer_size = 0;
  int rewind_pos = 0;
  uint8_t *buffer = nullptr;
  MemoryType memory_type = RAM;
  bool is_loop = false;
  void (*rewind_cb)() = nullptr;
  bool is_active = false;
  bool owns_memory = true;
 
  bool memoryCanChange() { return memory_type != FLASH_RAM; }
 
  void copy(MemoryStream &source) {
    if (this == &source) return;
 
    // Release currently owned mutable buffer.
    if (memoryCanChange() && owns_memory && buffer != nullptr) {
      free(buffer);
      buffer = nullptr;
    }
 
    // Shallow copy for FLASH memory source.
    if (source.memory_type == FLASH_RAM) {
      owns_memory = false;
      setValue(source.buffer, source.buffer_size, source.memory_type);
      is_active = source.is_active;
      is_loop = source.is_loop;
      rewind_pos = source.rewind_pos;
      return;
    }
 
    // Deep copy for mutable source memory.
    owns_memory = true;
    setValue(nullptr, source.buffer_size, source.memory_type);
    if (!resize(source.buffer_size)) {
      setValue(nullptr, 0, source.memory_type);
      is_active = false;
      return;
    }
 
    if (source.buffer != nullptr && source.buffer_size > 0) {
      memcpy(buffer, source.buffer, source.buffer_size);
    }
 
    write_pos = source.write_pos;
    read_pos = source.read_pos;
    if (write_pos > buffer_size) write_pos = buffer_size;
    if (read_pos > write_pos) read_pos = write_pos;
 
    is_active = source.is_active;
    is_loop = source.is_loop;
    rewind_pos = source.rewind_pos;
  }
};

class RingBufferStream : public AudioStream {
 public:
  RingBufferStream(int size = DEFAULT_BUFFER_SIZE) { resize(size); }
 
  virtual int available() override {
    // LOGD("RingBufferStream::available: %zu",buffer->available());
    return buffer.available();
  }
 
  virtual int availableForWrite() override {
    return buffer.availableForWrite();
  }
 
  virtual void flush() override {}
  virtual int peek() override {
    uint8_t data = 0;
    if (!buffer.peek(data)) return -1;
    return data;
  }
  virtual int read() override {
    uint8_t data = 0;
    if (!buffer.read(data)) return -1;
    return data;
  }
 
  virtual size_t readBytes(uint8_t *data, size_t len) override {
    return buffer.readArray(data, len);
  }
 
  virtual size_t write(const uint8_t *data, size_t len) override {
    // LOGD("RingBufferStream::write: %zu",len);
    return buffer.writeArray(data, len);
  }
 
  virtual size_t write(uint8_t c) override { return buffer.write(c); }
 
  void resize(int size) { buffer.resize(size); }
 
  size_t size() { return buffer.size(); }
 
 protected:
  RingBuffer<uint8_t> buffer{0};
};

 
template <class T = int16_t>
class GeneratedSoundStream : public AudioStream {
 public:
  GeneratedSoundStream() = default;
 
  GeneratedSoundStream(SoundGenerator<T> &generator) {
    TRACED();
    setInput(generator);
  }
 
  void setInput(SoundGenerator<T> &generator) {
    this->p_generator = &generator;
  }
 
  AudioInfo defaultConfig() {
    if (p_generator == nullptr) {
      LOGW("%s", source_not_defined_error);
      return AudioInfo();
    }
    return this->p_generator->defaultConfig();
  }
 
  void setAudioInfo(AudioInfo newInfo) override {
    if (newInfo.bits_per_sample != sizeof(T) * 8) {
      LOGE("Wrong bits_per_sample: %d", newInfo.bits_per_sample);
    }
    if (p_generator != nullptr) {
      p_generator->setAudioInfo(newInfo);
    } 
    AudioStream::setAudioInfo(newInfo);
  }

  bool begin() override {
    TRACED();
    if (p_generator == nullptr) {
      LOGE("%s", source_not_defined_error);
      return false;
    }
    p_generator->begin();
    notifyAudioChange(p_generator->audioInfo());
    active = true;
    return active;
  }

  bool begin(AudioInfo cfg) {
    TRACED();
    if (p_generator == nullptr) {
      LOGE("%s", source_not_defined_error);
      return false;
    }
    p_generator->begin(cfg);
    notifyAudioChange(p_generator->audioInfo());
    active = true;
    return active;
  }

  void end() override {
    TRACED();
    if (p_generator != nullptr) {
      p_generator->end();
    }
    active = true;  // legacy support - most sketches do not call begin
  }
 
  AudioInfo audioInfo() override {
    if (p_generator == nullptr) {
      LOGE("%s", source_not_defined_error);
      return AudioStream::audioInfo();
    }
    return p_generator->audioInfo();
  }

  virtual int available() override { return active ? buffer_size : 0; }

  size_t readBytes(uint8_t *data, size_t len) override {
    if (!active) return 0;
    if (p_generator == nullptr) {
      return 0;
    }
    LOGD("GeneratedSoundStream::readBytes: %u", (unsigned int)len);
    return p_generator->readBytes(data, len);
  }
 
  bool isActive() { 
    if (p_generator == nullptr) return false;
    return active && p_generator->isActive(); 
  }
 
  operator bool() override { return isActive(); }
 
  void flush() override {}

  void resize(int maxReadSize) { buffer_size = maxReadSize; }
 
 protected:
  bool active = true;  // support for legacy sketches
  SoundGenerator<T> *p_generator = nullptr;
  int buffer_size =
      DEFAULT_BUFFER_SIZE * 100;  // there is no reason to limit this
  const char *source_not_defined_error = "Source not defined";
};

class BufferedStream : public ModifyingStream {
 public:
  BufferedStream(size_t buffer_size) {
    TRACED();
    if (buffer_size > 0) resize(buffer_size);
  }
 
  BufferedStream(Print &out, size_t buffer_size = 1024) {
    TRACED();
    setOutput(out);
    if (buffer_size > 0) resize(buffer_size);
  }
 
  BufferedStream(Stream &io, size_t buffer_size = 1024) {
    TRACED();
    setStream(io);
    if (buffer_size > 0) resize(buffer_size);
  }
 
  BufferedStream(size_t buffer_size, Print &out) {
    TRACED();
    setOutput(out);
    if (buffer_size > 0) resize(buffer_size);
  }
 
  BufferedStream(size_t buffer_size, Stream &io) {
    TRACED();
    setStream(io);
    if (buffer_size > 0) resize(buffer_size);
  }
 
  void setOutput(Print &out) override { p_out = &out; }
  void setStream(Print &out) { setOutput(out); }
  void setStream(Stream &io) override {
    p_in = &io;
    p_out = &io;
  }

  size_t write(uint8_t c) override {
    if (buffer.isFull()) {
      flush();
    }
    return buffer.write(c);
  }

  size_t write(const uint8_t *data, size_t len) override {
    LOGD("%s: %zu", LOG_METHOD, len);
    int result = 0;
    for (int j = 0; j < len; j++) {
      result += write(data[j]);
    }
    return result;
  }

  void flush() override {
    // just dump the memory of the buffer and clear it
    if (buffer.available() > 0) {
      writeExt(buffer.address(), buffer.available());
      buffer.reset();
    }
  }

  int read() override {
    if (buffer.isEmpty()) {
      refill();
    }
    uint8_t result = 0;
    if (!buffer.read(result)) return -1;
    return result;
  }

  int peek() override {
    if (buffer.isEmpty()) {
      refill();
    }
    uint8_t result = 0;
    if (!buffer.peek(result)) return -1;
    return result;
  };

  size_t readBytes(uint8_t *data, size_t len) override {
    if (buffer.isEmpty() && len >= minReadBufferSize) {
      return readExt(data, len);
    } else {
      refill(len);
      return buffer.readArray(data, len);
    }
  }

  size_t peekBytes(uint8_t *data, size_t len) {
    if (buffer.isEmpty()) {
      refill();
    }
    return buffer.peekArray(data, len);
  }

  int available() override {
    if (p_in == nullptr) return 0;
    return buffer.available() + p_in->available();
  }

  void clear() { buffer.reset(); }

  void resize(int size) { buffer.resize(size); }

  void setMinUnbufferedReadSize(size_t size) { minReadBufferSize = size; }
 
 protected:
  SingleBuffer<uint8_t> buffer{0};
  Print *p_out = nullptr;
  Stream *p_in = nullptr;
  size_t minReadBufferSize = 1024;

  void refill() {
    // Preserve any existing unread data then append new bytes
    buffer.trim();
    size_t already_available = buffer.available();
    size_t free_space = buffer.availableForWrite();
    if (free_space == 0) {
      // Buffer full – nothing we can append
      return;
    }
    // Read new data directly behind existing bytes
    size_t added = readExt(buffer.address() + already_available, free_space);
    buffer.setAvailable(already_available + added);
  }

  void refill(size_t len) {
    if (buffer.available() >= len) return;
    refill();
  }
 
  virtual size_t writeExt(const uint8_t *data, size_t len) {
    return p_out == nullptr ? 0 : p_out->write(data, len);
  }
  virtual size_t readExt(uint8_t *data, size_t len) {
    return p_in == nullptr ? 0 : p_in->readBytes(data, len);
  }
};

template <typename T = int16_t>
class ConverterStream : public ModifyingStream {
 public:
  ConverterStream() = default;
 
  ConverterStream(BaseConverter &converter) { setConverter(converter); }
 
  ConverterStream(Stream &stream, BaseConverter &converter) {
    setConverter(converter);
    setStream(stream);
  }
 
  ConverterStream(Print &out, BaseConverter &converter) {
    setConverter(converter);
    setOutput(out);
  }
 
  void setStream(Stream &stream) {
    TRACEI();
    p_stream = &stream;
    p_out = &stream;
  }
 
  void setOutput(Print &out) {
    TRACEI();
    p_out = &out;
  }
 
  void setConverter(BaseConverter &cnv) { p_converter = &cnv; }
 
  virtual int availableForWrite() { return p_out->availableForWrite(); }
 
  virtual size_t write(const uint8_t *data, size_t len) {
    size_t result = p_converter->convert((uint8_t *)data, len);
    if (result > 0) {
      size_t result_written = p_out->write(data, result);
      return len * result_written / result;
    }
    return 0;
  }
 
  size_t readBytes(uint8_t *data, size_t len) override {
    if (p_stream == nullptr) return 0;
    size_t result = p_stream->readBytes(data, len);
    return p_converter->convert(data, result);
  }

  virtual int available() override {
    if (p_stream == nullptr) return 0;
    return p_stream->available();
  }
 
 protected:
  Stream *p_stream = nullptr;
  Print *p_out = nullptr;
  BaseConverter *p_converter;
};

class ProgressStreamInfo : public AudioInfo {
 public:
  size_t total_size = 0;
};

class ProgressStream : public ModifyingStream {
 public:
  ProgressStream() = default;
 
  ProgressStream(Print &print) { setPrint(print); }
 
  ProgressStream(Stream &stream) { setStream(stream); }
 
  ProgressStream(AudioStream &stream) {
    setStream(stream);
    p_info_from = &stream;
  }
 
  ProgressStreamInfo &defaultConfig() { return progress_info; }
 
  void setAudioInfo(AudioInfo info) override {
    AudioStream::setAudioInfo(info);
    progress_info.copyFrom(info);
  }
 
  void setStream(Stream &stream) override {
    p_stream = &stream;
    p_print = &stream;
  }
 
  void setStream(Print &print) { p_print = &print; }
 
  void setPrint(Print &print) { p_print = &print; }
 
  bool begin() override {
    if (p_info_from != nullptr) {
      setAudioInfo(p_info_from->audioInfo());
    }
    return AudioStream::begin();
  }

  bool begin(size_t len) {
    setSize(len);
    return begin();
  }
 
  bool begin(ProgressStreamInfo info) {
    progress_info = info;
    setAudioInfo(info);
    return begin();
  }

  void setSize(size_t len) {
    total_processed = 0;
    progress_info.total_size = len;
  }

  size_t size() { return progress_info.total_size; }

  size_t processedBytes() { return total_processed; }

  size_t processedSecs() { return total_processed / byteRate(); }

  size_t totalBytes() { return progress_info.total_size; }

  size_t totalSecs() { return totalBytes() / byteRate(); }

  float percentage() {
    if (progress_info.total_size == 0) return 0;
    return 100.0 * total_processed / progress_info.total_size;
  }

  size_t readBytes(uint8_t *data, size_t len) override {
    if (p_stream == nullptr) return 0;
    return measure(p_stream->readBytes(data, len));
  }
 
  int available() override {
    if (p_stream == nullptr) return 0;
    return p_stream->available();
  }

  virtual size_t write(const uint8_t *data, size_t len) override {
    if (p_print == nullptr) return 0;
    return measure(p_print->write(data, len));
  }

  virtual int availableForWrite() override {
    if (p_print == nullptr) return 0;
    return p_print->availableForWrite();
  }
 
 protected:
  ProgressStreamInfo progress_info;
  Stream *p_stream = nullptr;
  Print *p_print = nullptr;
  AudioInfoSupport *p_info_from = nullptr;
  size_t total_processed = 0;
 
  size_t measure(size_t len) {
    total_processed += len;
    return len;
  }
 
  size_t byteRate() {
    AudioInfo info = audioInfo();
    int byte_rate = info.sample_rate * info.bits_per_sample * info.channels / 8;
    if (byte_rate == 0) {
      LOGE("Audio Info not defined");
      return 0;
    }
    return byte_rate;
  }
};

struct ThrottleConfig : public AudioInfo {
  ThrottleConfig() {
    sample_rate = 44100;
    bits_per_sample = 16;
    channels = 2;
  }
  int correction_us = 0;
};

class Throttle : public ModifyingStream {
 public:
  Throttle() = default;
  Throttle(Print &out) { setOutput(out); }
  Throttle(Stream &io) { setStream(io); }

  void setStream(Stream &io) override {
    p_out = &io;
    p_in = &io;
  };

  void setOutput(Print &out) override { p_out = &out; }
 
  ThrottleConfig defaultConfig() {
    ThrottleConfig c;
    return c;
  }
 
  bool begin(ThrottleConfig cfg) {
    LOGI("begin sample_rate: %d, channels: %d, bits: %d", (int)info.sample_rate,
         (int)info.channels, (int)info.bits_per_sample);
    this->info = cfg;
    this->cfg = cfg;
    return begin();
  }
 
  bool begin(AudioInfo info) {
    LOGI("begin sample_rate: %d, channels: %d, bits: %d", (int)info.sample_rate,
         (int)info.channels, (int)info.bits_per_sample);
    this->info = info;
    this->cfg.copyFrom(info);
    return begin();
  }
 
  bool begin() override {
    frame_size = cfg.bits_per_sample / 8 * cfg.channels;
    startDelay();
    return true;
  }
 
  // (re)starts the timing
  void startDelay() {
    start_time = micros();
    sum_frames = 0;
  }
 
  int availableForWrite() override {
    if (p_out) {
      return p_out->availableForWrite();
    }
    return DEFAULT_BUFFER_SIZE;
  }
 
  size_t write(const uint8_t *data, size_t len) override {
    size_t result = p_out->write(data, len);
    delayBytes(len);
    return result;
  }
 
  int available() override {
    if (p_in == nullptr) return 0;
    return p_in->available();
  }
 
  size_t readBytes(uint8_t *data, size_t len) override {
    if (p_in == nullptr) {
      delayBytes(len);
      return 0;
    }
    size_t result = p_in->readBytes(data, len);
    delayBytes(len);
    return result;
  }
 
  // delay
  void delayBytes(size_t bytes) { delayFrames(bytes / frame_size); }
 
  // delay
  void delayFrames(size_t frames) {
    sum_frames += frames;
    uint64_t durationUsEff = micros() - start_time;
    uint64_t durationUsToBe = getDelayUs(sum_frames);
    int64_t waitUs = durationUsToBe - durationUsEff + cfg.correction_us;
    LOGD("wait us: %ld", static_cast<long>(waitUs));
    if (waitUs > 0) {
      int64_t waitMs = waitUs / 1000;
      if (waitMs > 0) delay(waitMs);
      delayMicroseconds(waitUs - (waitMs * 1000));
    } else {
      LOGD("negative delay!")
    }
  }
 
  inline int64_t getDelayUs(uint64_t frames) {
    return (frames * 1000000) / cfg.sample_rate;
  }
 
  inline int64_t getDelayMs(uint64_t frames) {
    return getDelayUs(frames) / 1000;
  }
 
  inline int64_t getDelaySec(uint64_t frames) {
    return getDelayUs(frames) / 1000000l;
  }
 
 protected:
  uint32_t start_time = 0;
  uint32_t sum_frames = 0;
  ThrottleConfig cfg;
  int frame_size = 0;
  Print *p_out = nullptr;
  Stream *p_in = nullptr;
};

 
template <typename T=int16_t, typename SumT=float>
class InputMixer : public AudioStream {
 public:
  InputMixer() = default;

  int add(Stream &in, int weight = 100) {
    streams.push_back(&in);
    weights.push_back(weight);
    recalculateWeights();
    return streams.indexOf(&in);
  }

  bool set(int index, Stream &in) {
    if (index < size()) {
      streams[index] = &in;
      return true;
    } else {
      LOGE("Invalid index %d - max is %d", index, size() - 1);
      return false;
    }
  }
 
  virtual bool begin(AudioInfo info) {
    setAudioInfo(info);
    frame_size = info.bits_per_sample / 8 * info.channels;
    LOGI("frame_size: %d", frame_size);
    return frame_size > 0;
  }

  void setWeight(int index, int weight) {
    if (index < streams.size()) {
      weights[index] = weight;
      recalculateWeights();
    } else {
      LOGE("Invalid index %d - max is %d", index, size() - 1);
    }
  }

  void end() override {
    streams.clear();
    weights.clear();
    gains.clear();
    result_vect.clear();
    current_vect.clear();
    total_weights = 0.0;
  }

  int size() { return streams.size(); }

  size_t readBytes(uint8_t *data, size_t len) override {
    if (total_weights == 0 || frame_size == 0 || len == 0) {
      LOGW("readBytes: %d", (int)len);
      return 0;
    }
 
    if (limit_available_data) {
      len = min((int)len, availableBytes());
    }
 
    int result_len = 0;
 
    if (len > 0) {
      // result_len must be full frames
      result_len = (len / frame_size) * frame_size;;
      // replace sample based with vector based implementation
      // readBytesSamples((T*)data, result_len));
      result_len = readBytesVector((T *)data, result_len);
    }
    return result_len;
  }

  int available() override {
    if (total_weights == 0 || streams.empty()) return 0;
    return limit_available_data ? availableBytes() : DEFAULT_BUFFER_SIZE;
  }

  void setLimitToAvailableData(bool flag) { limit_available_data = flag; }

  void setRetryCount(int retry) { retry_count = retry; }

  bool remove(int idx) {
    if (idx < 0 || idx >= size()) {
      return false;
    }
    streams.erase(idx);
    weights.erase(idx);
    recalculateWeights();
    return true;
  }

  bool remove() {
    bool rc = false;
    int idx = nextEmptyIndex();
    while (idx >= 0) {
      rc = true;
      streams.erase(idx);
      weights.erase(idx);
      idx = nextEmptyIndex();
    }
    recalculateWeights();
    return rc;
  }

  int indexOf(Stream &stream) { return streams.indexOf(&stream); }

  Stream *operator[](int idx) {
    if (idx < 0 || idx >= size()) return nullptr;
    return streams[idx];
  }

  int nextEmptyIndex() {
    for (int i = 0; i < streams.size(); i++) {
      if (streams[i]->available() == 0) {
        return i;
      }
    }
    return -1;
  }
 
 protected:
  Vector<Stream *> streams{0};
  Vector<int> weights{0};
  Vector<float> gains{0};
  int total_weights = 0;
  int frame_size = 4;
  bool limit_available_data = false;
  int retry_count = 5;
  Vector<SumT> result_vect;
  Vector<T> current_vect;

  void recalculateWeights() {
    int total = 0;
    for (int j = 0; j < weights.size(); j++) {
      total += weights[j];
    }
    total_weights = total;
    gains.resize(weights.size());
    for (int j = 0; j < weights.size(); j++) {
      gains[j] = total_weights == 0 ? 0.0f : static_cast<float>(weights[j]) / total_weights;
    }
  }

  int readBytesVector(T *p_data, int byteCount) {
    int samples = byteCount / sizeof(T);
    if (result_vect.size() < samples) result_vect.resize(samples);
    if (current_vect.size() < samples) current_vect.resize(samples);
    int stream_count = size();
    resultClear(samples);
    int samples_eff_max = 0;
    for (int j = 0; j < stream_count; j++) {
      if (weights[j] > 0) {
        int samples_eff =
            readSamples(streams[j], current_vect.data(), samples, retry_count);
        if (samples_eff > samples_eff_max) samples_eff_max = samples_eff;
        resultAdd(MixGain<T>(gains[j]), samples_eff);
      }
    }
    // copy result
    for (int j = 0; j < samples_eff_max; j++) {
      p_data[j] = static_cast<T>(result_vect[j]);
    }
    return samples_eff_max * sizeof(T);
  }

  int availableBytes() {
    int result = DEFAULT_BUFFER_SIZE;
    for (int j = 0; j < size(); j++) {
      result = min(result, streams[j]->available());
    }
    return result;
  }
 
  void resultAdd(MixGain<T> factor, int samples_eff) {
    // only accumulate samples that were actually read; tail is already zeroed
    for (int j = 0; j < samples_eff; j++) {
      result_vect[j] += static_cast<SumT>(factor.scale(current_vect[j]));
    }
  }
 
  void resultClear(int samples) {
    memset(result_vect.data(), 0, sizeof(SumT) * samples);
  }
};

template <typename T>
class InputMerge : public AudioStream {
 public:
  InputMerge() = default;

  InputMerge(Stream &left, Stream &right) {
    add(left, 1);
    add(right, 1);
  };

  AudioInfo audioInfo() override {
    AudioInfo info = AudioStream::audioInfo();
    info.channels = total_channel_count;
    return info;
  }
 
  virtual bool begin(AudioInfo info) {
    setAudioInfo(info);
    return begin();
  }
 
  virtual bool begin() override {
    // make sure that we use the correct channel count
    info.channels = channelCount();
    return AudioStream::begin();
  }

  size_t readBytes(uint8_t *data, size_t len) override {
    LOGD("readBytes: %d", (int)len);
    T *p_data = (T *)data;
    int result_len = min((size_t)available(), len);
    int frames = result_len / (sizeof(T) * total_channel_count);
    int result_idx = 0;
    for (int j = 0; j < frames; j++) {
      for (int i = 0; i < records.size(); i++) {
        for (int ch = 0; ch < records[i].channels; ch++) {
          p_data[result_idx++] =
              records[i].weight * readSample<T>(records[i].stream);
        }
      }
    }
    return result_idx * sizeof(T);
  }

  void add(Stream &in, int channelCount, float weight = 1.0) {
    MergeRecord rec(&in, channelCount, weight);
    records.push_back(rec);
    total_channel_count += channelCount;
  }

  void setWeight(int channel, float weight) {
    if (channel < channelCount()) {
      records[channel].weight = weight;
    } else {
      LOGE("Invalid channel %d - max is %d", channel, channelCount() - 1);
    }
  }

  void end() override { records.clear(); }

  int channelCount() { return total_channel_count; }

  int available() override {
    int result = records[0].stream->available();
    for (int j = 1; j < channelCount(); j++) {
      int tmp = records[j].stream->available();
      if (tmp < result) {
        result = tmp;
      }
    }
    return result;
  }
 
 protected:
  struct MergeRecord {
    Stream *stream = nullptr;
    int channels = 0;
    float weight = 1.0;
    MergeRecord() = default;
    MergeRecord(Stream *str, int ch, float w) {
      stream = str;
      channels = ch;
      weight = w;
    }
  };
  Vector<MergeRecord> records;
  int total_channel_count = 0;
};

class CallbackStream : public ModifyingStream {
 public:
  CallbackStream() = default;

  CallbackStream(Stream &io, size_t (*cb_update)(uint8_t *data, size_t len)) {
    p_stream = &io;
    p_out = &io;
    setUpdateCallback(cb_update);
  }

  CallbackStream(Print &out, size_t (*cb_update)(uint8_t *data, size_t len)) {
    p_out = &out;
    setUpdateCallback(cb_update);
  }
 
  CallbackStream(size_t (*cb_read)(uint8_t *data, size_t len),
                 size_t (*cb_write)(const uint8_t *data, size_t len)) {
    setWriteCallback(cb_write);
    setReadCallback(cb_read);
  }

  void setWriteCallback(size_t (*cb_write)(const uint8_t *data, size_t len)) {
    this->cb_write = cb_write;
  }

  void setReadCallback(size_t (*cb_read)(uint8_t *data, size_t len)) {
    this->cb_read = cb_read;
  }

  void setUpdateCallback(size_t (*cb_update)(uint8_t *data, size_t len)) {
    this->cb_update = cb_update;
  }
 
  // callback result negative -> no change; callbeack result >=0 provides the
  // result
  void setAvailableCallback(int (*cb)()) { this->cb_available = cb; }

  void setAudioInfoCallback(void (*cb)(AudioInfo info)) {
    this->cb_audio_info = cb;
  }

  void setAudioInfo(AudioInfo info) override {
    ModifyingStream::setAudioInfo(info);
    if (cb_audio_info != nullptr) {
      cb_audio_info(info);
    }
  }
 
  virtual bool begin(AudioInfo info) {
    setAudioInfo(info);
    return begin();
  }
  virtual bool begin() override {
    active = true;
    return true;
  }
 
  void end() override { active = false; }
 
  int available() override {
    int result = AudioStream::available();
    // determine value from opional variable
    if (available_bytes >= 0) return available_bytes;
    // check if there is a callback
    if (cb_available == nullptr) return result;
    // determine value from callback
    int tmp_available = cb_available();
    if (tmp_available < 0) return result;
 
    return tmp_available;
  }
 
  size_t readBytes(uint8_t *data, size_t len) override {
    if (!active) return 0;
    // provide data from callback
    if (cb_read) {
      return cb_read(data, len);
    }
    // provide data from source
    size_t result = 0;
    if (p_stream) {
      result = p_stream->readBytes(data, len);
    }
    if (cb_update) {
      result = cb_update(data, result);
    }
    return result;
  }
 
  size_t write(const uint8_t *data, size_t len) override {
    if (!active) return 0;
    // write to callback
    if (cb_write) {
      return cb_write(data, len);
    }
    // write to output
    if (p_out) {
      size_t result = len;
      if (cb_update) {
        result = cb_update((uint8_t *)data, len);
      }
      return p_out->write(data, result);
    }
    // no processing possible
    return 0;
  }

  void setStream(Stream &in) override {
    p_stream = &in;
    p_out = &in;
  }

  void setOutput(Print &out) override { p_out = &out; }

  void setOutput(Stream &in) {
    p_stream = &in;
    p_out = &in;
  }

  void setStream(Print &out) { p_out = &out; }

  void setAvailable(int val) { available_bytes = val; }
 
 protected:
  bool active = true;
  size_t (*cb_write)(const uint8_t *data, size_t len) = nullptr;
  size_t (*cb_read)(uint8_t *data, size_t len) = nullptr;
  size_t (*cb_update)(uint8_t *data, size_t len) = nullptr;
  void (*cb_audio_info)(AudioInfo info) = nullptr;
  int (*cb_available)() = nullptr;
  Stream *p_stream = nullptr;
  Print *p_out = nullptr;
  int available_bytes = -1;
};

template <typename T = int16_t, class TF = float>
class FilteredStream : public ModifyingStream {
 public:
  FilteredStream() = default;
  FilteredStream(Stream &stream) : ModifyingStream() { setStream(stream); }
  FilteredStream(Stream &stream, int channels) : ModifyingStream() {
    this->channels = channels;
    setStream(stream);
    converter.setChannels(channels);
  }
  FilteredStream(Print &stream) : ModifyingStream() { setOutput(stream); }
  FilteredStream(Print &stream, int channels) : ModifyingStream() {
    this->channels = channels;
    setOutput(stream);
    converter.setChannels(channels);
  }
 
  virtual ~FilteredStream() { end(); }
 
  void setStream(Stream &stream) override {
    p_io = &stream;
    p_out = &stream;
  }
 
  void setOutput(Print &stream) override { p_out = &stream; }
 
  bool begin(AudioInfo info) {
    setAudioInfo(info);
    this->channels = info.channels;
    return begin();
  }
 
  bool begin() override {
    if (channels == 0) {
      LOGE("channels must not be 0");
      return false;
    }
    converter.setChannels(channels);
    return AudioStream::begin();
  }
 
  void end() override {
    ModifyingStream::end();
  }
 
  virtual size_t write(const uint8_t *data, size_t len) override {
    if (p_out == nullptr) return 0;
    size_t result = converter.convert((uint8_t *)data, len);
    return p_out->write(data, result);
  }
 
  size_t readBytes(uint8_t *data, size_t len) override {
    if (p_io == nullptr) return 0;
    size_t result = p_io->readBytes(data, len);
    result = converter.convert(data, result);
    return result;
  }
 
  virtual int available() override {
    if (p_io == nullptr) return 0;
    return p_io->available();
  }
 
  virtual int availableForWrite() override {
    if (p_out == nullptr) return 0;
    return p_out->availableForWrite();
  }
 
  virtual void flush() override {
    if (p_out != nullptr) p_out->flush();
  }

  void setFilter(int channel, Filter<TF> *filter) {
    if (channel >= converter.getChannels()) {
      converter.setChannels(channel + 1);
      if (channels < channel + 1) channels = channel + 1;
    }
    converter.setFilter(channel, filter);
  }

  void setFilter(int channel, Filter<TF> &filter) {
    setFilter(channel, &filter);
  }
 
 protected:
  int channels = 0;
  Stream *p_io = nullptr;
  Print *p_out = nullptr;
  ConverterNChannels<T, TF> converter{0};
};

using ActivityCallback = void (*)(bool isActive);

 
class VolumeMeter : public ModifyingStream {
 public:
  VolumeMeter() = default;
  VolumeMeter(AudioStream &as) { setStream(as); }
  VolumeMeter(AudioOutput &ao) { setOutput(ao); }
  VolumeMeter(Print &print) { setOutput(print); }
  VolumeMeter(Stream &stream) { setStream(stream); }
 
  bool begin(AudioInfo info) {
    setAudioInfo(info);
    return begin();
  }
 
  bool begin() override {
    setAudioInfo(audioInfo());
    return true;
  }
 
  void setAudioInfo(AudioInfo info) override {
    int channels = info.channels;
    LOGI("VolumeMeter::setAudioInfo: channels %d", channels);
    ModifyingStream::setAudioInfo(info);
    if (channels > 0) {
      volumes.resize(channels);
      volumes_tmp.resize(channels);
      sum.resize(channels);
      sum_tmp.resize(channels);
    }
  }
 
  size_t write(const uint8_t *data, size_t len) override {
    updateVolumes(data, len);
    size_t result = len;
    if (p_out != nullptr) {
      result = p_out->write(data, len);
    }
    return result;
  }
 
  size_t readBytes(uint8_t *data, size_t len) override {
    if (p_stream == nullptr) return 0;
    size_t result = p_stream->readBytes(data, len);
    updateVolumes((const uint8_t *)data, len);
    return result;
  }

  float volume() { return f_volume; }

  float volume(int channel) {
    if (volumes.size() == 0) {
      LOGE("begin not called!");
      return 0.0f;
    }
    if (channel >= volumes.size()) {
      LOGE("invalid channel %d", channel);
      return 0.0f;
    }
    return volumes[channel];
  }

  float volumeRatio() {
    return volume() / NumberConverter::maxValue(info.bits_per_sample);
  }

  float volumeRatio(int channel) {
    return volume(channel) / NumberConverter::maxValue(info.bits_per_sample);
  }

  float volumeDB() {
    // prevent infinite value
    if (volumeRatio() == 0) return -1000;
    return 20.0f * log10(volumeRatio());
  }

  float volumeDB(int channel) {
    // prevent infinite value
    if (volumeRatio(channel) == 0) return -1000;
    return 20.0f * log10(volumeRatio(channel));
  }

  float volumePercent() { return 100.0f * volumeRatio(); }

  float volumePercent(int channel) { return 100.0f * volumeRatio(channel); }

  float volumeAvg() {
    float total = 0;
    size_t count = 0;
    for (int j = 0; j < info.channels && j < (int)sum.size(); j++) {
      total += sum[j];
      count += sample_count_per_channel;
    }
    return count > 0 ? total / count : 0.0f;
  }

  float volumeAvg(int channel) {
    if (channel < 0 || channel >= (int)sum.size() || sample_count_per_channel == 0)
      return 0.0f;
    return sum[channel] / sample_count_per_channel;
  }

  void clear() {
    f_volume_tmp = 0;
    if (volumes_tmp.size() > 0 && sum_tmp.size() > 0) {
      for (int j = 0; j < info.channels; j++) {
        volumes_tmp[j] = 0;
        sum_tmp[j] = 0;
      }
    }
  }

  void setActivityCallback(ActivityCallback callback, float threshold = 0.2, unsigned long duration_ms = 2000) {
    activity_callback = callback;
    activity_threshold = threshold;
    activity_duration_ms = duration_ms;
    activity_monitoring_enabled = (callback != nullptr);
  }

  bool isActive() const {
    return is_active;
  }
 
  void setOutput(AudioOutput &out) {
    addNotifyAudioChange(out);
    setOutput((Print &)out);
  }
  void setStream(AudioStream &io) {
    addNotifyAudioChange(io);
    setStream((Stream &)io);
  }
  void setOutput(Print &out) override { p_out = &out; }
  void setStream(Stream &io) override {
    p_out = &io;
    p_stream = &io;
  }
 
 protected:
  float f_volume_tmp = 0;
  float f_volume = 0;
  Vector<float> volumes{0};
  Vector<float> volumes_tmp{0};
  Vector<float> sum{0};
  Vector<float> sum_tmp{0};
  Print *p_out = nullptr;
  Stream *p_stream = nullptr;
  size_t sample_count_per_channel = 0;
  
  // Activity monitoring
  ActivityCallback activity_callback = nullptr;
  float activity_threshold = 0.0f;
  unsigned long activity_duration_ms = 0;
  bool activity_monitoring_enabled = false;
  bool is_active = false;
  unsigned long inactive_start_time = 0;
 
  void updateVolumes(const uint8_t *data, size_t len) {
    if (data == nullptr || len == 0) return;
    clear();
    switch (info.bits_per_sample) {
      case 8:
        updateVolumesT<int8_t>(data, len);
        break;
      case 16:
        updateVolumesT<int16_t>(data, len);
        break;
      case 24:
        updateVolumesT<int24_t>(data, len);
        break;
      case 32:
        updateVolumesT<int32_t>(data, len);
        break;
      default:
        LOGE("Unsupported bits_per_sample: %d", info.bits_per_sample);
        break;
    }
  }
 
  template <typename T>
  void updateVolumesT(const uint8_t *buffer, size_t size) {
    T *bufferT = (T *)buffer;
    int samplesCount = size / sizeof(T);
    sample_count_per_channel = samplesCount / info.channels;
    for (int j = 0; j < samplesCount; j++) {
      float tmp = abs(static_cast<float>(bufferT[j]));
      updateVolume(tmp, j);
    }
    commit();
  }
 
  void updateVolume(float tmp, int j) {
    if (tmp > f_volume_tmp) {
      f_volume_tmp = tmp;
    }
    if (volumes_tmp.size() > 0 && info.channels > 0) {
      int ch = j % info.channels;
      if (tmp > volumes_tmp[ch]) {
        volumes_tmp[ch] = tmp;
      }
      // sum of all samples for volumeAvg()
      sum_tmp[ch] += tmp;
    }
  }
 
  void commit() {
    f_volume = f_volume_tmp;
    if (volumes.size() > 0 && sum.size() > 0) {
      for (int j = 0; j < info.channels; j++) {
        volumes[j] = volumes_tmp[j];
        sum[j] = sum_tmp[j];
      }
    }
    updateActivityState();
  }
 
  void updateActivityState() {
    if (!activity_monitoring_enabled || activity_callback == nullptr) return;
 
    float current_volume_ratio = volumeRatio();
    bool above_threshold = current_volume_ratio > activity_threshold;
    unsigned long current_time = millis();
 
    if (above_threshold) {
      // Volume is above threshold - should be active
      if (!is_active) {
        is_active = true;
        activity_callback(true);
      }
      inactive_start_time = 0; // Reset inactive timer
    } else {
      // Volume is below threshold
      if (is_active) {
        // Currently active, check if we should transition to inactive
        if (inactive_start_time == 0) {
          // Start timing the inactive period
          inactive_start_time = current_time;
        } else if (current_time - inactive_start_time >= activity_duration_ms) {
          // Been below threshold long enough
          is_active = false;
          activity_callback(false);
          inactive_start_time = 0;
        }
      }
    }
  }
};
 
// legacy names
using VolumePrint = VolumeMeter;

using VolumeOutput = VolumeMeter;

class AudioInputMonitor : public ModifyingStream {
 public:
  AudioInputMonitor(uint8_t limitPercent = 20) {
    setLimitPercent(limitPercent);
    volume_meter.addNotifyAudioChange(*this);
  }

  AudioInputMonitor(AudioStream &as, uint8_t limitPercent = 20)
      : AudioInputMonitor(limitPercent) {
    setStream(as);
  }

  AudioInputMonitor(AudioOutput &ao, uint8_t limitPercent = 20)
      : AudioInputMonitor(limitPercent) {
    setOutput(ao);
  }

  AudioInputMonitor(Print &print, uint8_t limitPercent = 20)
      : AudioInputMonitor(limitPercent) {
    setOutput(print);
  }

  AudioInputMonitor(Stream &stream, uint8_t limitPercent = 20)
      : AudioInputMonitor(limitPercent) {
    setStream(stream);
  }

  void setLimitPercent(uint8_t percent) { limit_percent = percent; }

  uint8_t limitPercent() const { return limit_percent; }

  bool begin(AudioInfo info) {
    setAudioInfo(info);
    return begin();
  }

  bool begin() override {
    volume_meter.begin(audioInfo());
    return true;
  }

  void setAudioInfo(AudioInfo info) override {
    ModifyingStream::setAudioInfo(info);
    volume_meter.setAudioInfo(info);
  }

  size_t write(const uint8_t *data, size_t len) override {
    size_t result = volume_meter.write(data, len);
    if (result > 0 && volume_meter.volumePercent() > limit_percent) {
      time_over_last_volume_limit = millis();
    }
    return result;
  }

  size_t readBytes(uint8_t *data, size_t len) override {
    size_t result = volume_meter.readBytes(data, len);
    if (result > 0 && volume_meter.volumePercent() > limit_percent) {
      time_over_last_volume_limit = millis();
    }
    return result;
  }

  void setOutput(Print &out) override { volume_meter.setOutput(out); }
  void setOutput(AudioOutput &out) { volume_meter.setOutput(out); }
  void setStream(Stream &io) override { volume_meter.setStream(io); }
  void setStream(AudioStream &io) { volume_meter.setStream(io); }

  VolumeMeter &getVolumeMeter() { return volume_meter; }

  bool isActive(uint16_t time_ms = 1000) {
    return (millis() - time_over_last_volume_limit) < time_ms;
  }
 
protected:
  VolumeMeter volume_meter; 
  uint8_t limit_percent = 20; 
  uint64_t time_over_last_volume_limit = 0; 
};
 
#ifdef USE_TIMER
struct TimerCallbackAudioStreamInfo : public AudioInfo {
  RxTxMode rx_tx_mode = RX_MODE;
  uint16_t buffer_size = DEFAULT_BUFFER_SIZE;
  bool use_timer = true;
  int timer_id = -1;
  TimerFunction timer_function = DirectTimerCallback;
  bool adapt_sample_rate = false;
  uint16_t (*callback)(uint8_t *data, uint16_t len) = nullptr;
};
 
// forward declaration: relevant only if use_timer == true
static void timerCallback(void *obj);
class TimerCallbackAudioStream : public BufferedStream {
  friend void timerCallback(void *obj);
 
 public:
  TimerCallbackAudioStream() : BufferedStream(80) { TRACED(); }
 
  ~TimerCallbackAudioStream() {
    TRACED();
    if (timer != nullptr) delete timer;
    if (buffer != nullptr) delete buffer;
    if (frame != nullptr) delete[] frame;
  }

  TimerCallbackAudioStreamInfo defaultConfig() {
    TimerCallbackAudioStreamInfo def;
    return def;
  }

  virtual void setAudioInfo(AudioInfo info) {
    TRACED();
    if (cfg.sample_rate != info.sample_rate || cfg.channels != info.channels ||
        cfg.bits_per_sample != info.bits_per_sample) {
      bool do_restart = active;
      if (do_restart) end();
      cfg.sample_rate = info.sample_rate;
      cfg.channels = info.channels;
      cfg.bits_per_sample = info.bits_per_sample;
      if (do_restart) begin(cfg);
    }
  }

  TimerCallbackAudioStreamInfo audioInfoExt() { return cfg; }
  AudioInfo audioInfo() { return cfg; }
 
  void begin(TimerCallbackAudioStreamInfo config) {
    LOGD("%s:  %s", LOG_METHOD,
         config.rx_tx_mode == RX_MODE ? "RX_MODE" : "TX_MODE");
    this->cfg = config;
    this->frameCallback = config.callback;
    if (cfg.use_timer) {
      frameSize = cfg.bits_per_sample * cfg.channels / 8;
      frame = new uint8_t[frameSize];
      buffer = new RingBuffer<uint8_t>(cfg.buffer_size);
      timer = new AudioTimer();
      timer->setTimerFunction(cfg.timer_function);
      if (cfg.timer_id >= 0) {
        timer->setTimer(cfg.timer_id);
      }
      time = AudioTime::toTimeUs(cfg.sample_rate);
      LOGI("sample_rate: %u -> time: %u milliseconds",
           (unsigned int)cfg.sample_rate, (unsigned int)time);
      timer->setCallbackParameter(this);
      timer->begin(timerCallback, time, TimeUnit::US);
    }
 
    notifyAudioChange(cfg);
    active = true;
  }

  bool begin() {
    TRACED();
    if (this->frameCallback != nullptr) {
      if (cfg.use_timer) {
        timer->begin(timerCallback, time, TimeUnit::US);
      }
      active = true;
    }
    return active;
  }

  void end() {
    TRACED();
    if (cfg.use_timer) {
      timer->end();
    }
    active = false;
  }

  uint16_t currentSampleRate() { return currentRateValue; }
 
 protected:
  TimerCallbackAudioStreamInfo cfg;
  bool active = false;
  uint16_t (*frameCallback)(uint8_t *data, uint16_t len);
  // below only relevant with timer
  AudioTimer *timer = nullptr;
  RingBuffer<uint8_t> *buffer = nullptr;
  uint8_t *frame = nullptr;
  uint16_t frameSize = 0;
  uint32_t time = 0;
  unsigned long lastTimestamp = 0u;
  uint32_t currentRateValue = 0;
  uint32_t printCount = 0;
 
  // used for audio sink
  virtual size_t writeExt(const uint8_t *data, size_t len) override {
    if (!active) return 0;
    TRACED();
    size_t result = 0;
    if (!cfg.use_timer) {
      result = frameCallback((uint8_t *)data, len);
    } else {
      result = buffer->writeArray((uint8_t *)data, len);
    }
    if (++printCount % 10000 == 0) printSampleRate();
    return result;
  }
 
  // used for audio source
  virtual size_t readExt(uint8_t *data, size_t len) override {
    if (!active) return 0;
    TRACED();
 
    size_t result = 0;
    if (!cfg.use_timer) {
      result = frameCallback(data, len);
    } else {
      result = buffer->readArray(data, len);
    }
    if (++printCount % 10000 == 0) printSampleRate();
    return result;
  }

  virtual void measureSampleRate() {
    unsigned long ms = millis();
    if (lastTimestamp > 0u) {
      uint32_t diff = ms - lastTimestamp;
      if (diff > 0) {
        uint16_t rate = 1 * 1000 / diff;
 
        if (currentRateValue == 0) {
          currentRateValue = rate;
        } else {
          currentRateValue = (currentRateValue + rate) / 2;
        }
      }
    }
    lastTimestamp = ms;
  }

  virtual void printSampleRate() {
    LOGI("effective sample rate: %u", (unsigned int)currentRateValue);
    if (cfg.adapt_sample_rate &&
        abs((int)currentRateValue - (int)cfg.sample_rate) > 200) {
      cfg.sample_rate = currentRateValue;
      notifyAudioChange(cfg);
    }
  }
};
 
// relevant only if use_timer == true
void IRAM_ATTR timerCallback(void *obj) {
  TimerCallbackAudioStream *src = (TimerCallbackAudioStream *)obj;
  if (src != nullptr) {
    // LOGD("%s:  %s", LOG_METHOD, src->cfg.rx_tx_mode==RX_MODE ?
    // "RX_MODE":"TX_MODE");
    if (src->cfg.rx_tx_mode == RX_MODE) {
      // input
      uint16_t available_bytes = src->frameCallback(src->frame, src->frameSize);
      uint16_t buffer_available = src->buffer->availableForWrite();
      if (buffer_available < available_bytes) {
        // if buffer is full make space
        uint16_t to_clear = available_bytes - buffer_available;
        uint8_t tmp[to_clear];
        src->buffer->readArray(tmp, to_clear);
      }
      if (src->buffer->writeArray(src->frame, available_bytes) !=
          available_bytes) {
        assert(false);
      }
    } else {
      // output
      if (src->buffer != nullptr && src->frame != nullptr &&
          src->frameSize > 0) {
        uint16_t available_bytes =
            src->buffer->readArray(src->frame, src->frameSize);
        if (available_bytes !=
            src->frameCallback(src->frame, available_bytes)) {
          LOGE("data underflow");
        }
      }
    }
    src->measureSampleRate();
  }
}
 
#endif
 
}  // namespace audio_tools