3#include "AudioTools/CoreAudio/AudioBasic/Collections.h"
4#include "AudioTools/CoreAudio/AudioBasic/Str.h"
5#include "AudioTools/CoreAudio/AudioLogger.h"
30 virtual bool read(T &result) = 0;
34 if (data ==
nullptr) {
39 for (
int j = 0; j < lenResult; j++) {
42 LOGD(
"readArray %d -> %d", len, lenResult);
60 for (
int j = 0; j < len; j++) {
61 if (!
write(data[j])) {
67 LOGD(
"writeArray %d -> %d", len, result);
81 virtual bool peek(T &result) = 0;
86 bool isEmpty() {
return available() == 0; }
106 virtual size_t size() = 0;
111 if (size() == 0)
return 0.0f;
112 return 100.0f *
static_cast<float>(
available()) /
113 static_cast<float>(size());
118 LOGE(
"resize not implemented for this buffer");
132 LOGD(
"%s: %d", LOG_METHOD, len);
134 int result = min(len, p_buffer->available());
135 for (
int j = 0; j < result; j++) {
137 p_buffer->read(sample);
145 template <
int rows,
int channels>
147 int lenResult = min(rows, p_buffer->available());
148 for (
int j = 0; j < lenResult; j++) {
150 p_buffer->read(sample);
151 for (
int i = 0; i < channels; i++) {
160 BaseBuffer<T> *p_buffer =
nullptr;
194 this->owns_buffer =
false;
195 this->buffer = (uint8_t *)
data;
196 this->current_read_pos = 0;
197 this->current_write_pos = len;
201 if (size() == 0)
resize(len);
207 if (current_write_pos < buffer.size()) {
208 buffer[current_write_pos++] = sample;
214 bool read(T &result)
override {
215 bool success =
false;
216 if (current_read_pos < current_write_pos) {
217 result = buffer[current_read_pos++];
223 bool peek(T &result)
override {
224 bool success =
false;
225 if (current_read_pos < current_write_pos) {
226 result = buffer[current_read_pos];
233 int result = current_write_pos - current_read_pos;
234 return max(result, 0);
241 int peekArray(uint8_t *
data,
int len) {
243 if (len > len_available) {
246 memcpy(
data, buffer.data() + current_read_pos, len);
255 return len_available;
257 current_read_pos += len;
258 len_available -= len;
259 memmove(buffer.data(), buffer.data() + current_read_pos, len_available *
sizeof(T));
260 current_read_pos = 0;
261 current_write_pos = len_available;
263 if (is_clear_with_zero) {
264 memset(buffer.data() + current_write_pos, 0,
265 buffer.size() - current_write_pos);
274 memmove(buffer.data(), buffer.data() + current_read_pos, av *
sizeof(T));
275 current_write_pos = av;
276 current_read_pos = 0;
280 T *
address()
override {
return buffer.data(); }
283 T *
data() {
return buffer.data() + current_read_pos; }
286 current_read_pos = 0;
287 current_write_pos = 0;
288 if (is_clear_with_zero) {
289 memset(buffer.data(), 0, buffer.size());
296 size_t result = min(available_size, (
size_t)buffer.size());
297 current_read_pos = 0;
298 current_write_pos = result;
302 size_t size()
override {
return buffer.size(); }
305 if (buffer.size() < size) {
326 int current_read_pos = 0;
327 int current_write_pos = 0;
328 bool owns_buffer =
true;
329 bool is_clear_with_zero =
false;
346 bool read(T &result)
override {
351 result = _aucBuffer[_iTail];
352 _iTail = nextIndex(_iTail);
359 bool peek(T &result)
override {
364 result = _aucBuffer[_iTail];
368 virtual int peekArray(T *data,
int n) {
369 if (isEmpty())
return -1;
371 int count = _numElems;
373 for (
int j = 0; j < n; j++) {
374 data[j] = _aucBuffer[tail];
375 tail = nextIndex(tail);
378 if (count == 0)
break;
386 bool isEmpty() {
return available() == 0; }
389 virtual bool write(T data)
override {
392 _aucBuffer[_iHead] = data;
393 _iHead = nextIndex(_iHead);
414 virtual T *
address()
override {
return _aucBuffer.data(); }
417 if (max_size != len && len > 0) {
418 LOGI(
"resize: %d", len);
419 _aucBuffer.resize(len);
426 virtual size_t size()
override {
return max_size; }
435 int nextIndex(
int index) {
436 if (max_size == 0)
return 0;
437 return (uint32_t)(index + 1) % max_size; }
448template <
class File,
typename T>
457 if (p_file) p_file->close();
463 p_file = &bufferFile;
465 LOGE(
"file is not valid");
475 if (p_file ==
nullptr)
return 0;
476 int read_count = min(count,
available());
479 if (!
file_seek(offset.pos))
return false;
481 if (offset.len1 > 0) {
483 n +=
file_read(data + offset.len, offset.len1);
484 read_pos = offset.len1;
486 read_pos += read_count;
489 for (
int i = 0; i < count; i++) {
490 LOGI(
"read #%d value %d", offset.pos, (
int)data[i]);
493 element_count -= read_count;
498 bool peek(T &result)
override {
499 if (p_file ==
nullptr || isEmpty()) {
510 if (p_file ==
nullptr)
return 0;
511 int read_count = min(count,
available());
514 if (!
file_seek(offset.pos))
return false;
516 if (offset.len1 > 0) {
518 n +=
file_read(data + offset.len, offset.len1);
520 assert(n == read_count);
529 if (p_file ==
nullptr)
return 0;
530 for (
int i = 0; i < len; i++) {
531 LOGI(
"write #%d value %d", write_pos, (
int)data[i]);
537 if (!
file_seek(offset.pos))
return false;
539 if (offset.len1 > 0) {
541 n +=
file_write(data + offset.len, offset.len1);
542 write_pos = offset.len1;
544 write_pos += write_count;
546 element_count += write_count;
553 bool isEmpty() {
return available() == 0; }
570 size_t size()
override {
return max_size; }
582 File *p_file =
nullptr;
585 int element_count = 0;
599 int overflow = (pos + len) - max_size;
606 result.len = len - overflow;
607 result.len1 = overflow;
614 int file_pos = pos *
sizeof(T);
615 if (p_file->position() != file_pos) {
616 LOGD(
"file_seek: %d", pos);
617 if (!p_file->seek(file_pos)) {
618 LOGE(
"seek %d", file_pos);
627 LOGD(
"file_write: %d", count);
628 if (p_file ==
nullptr)
return 0;
629 int to_write_bytes =
sizeof(T) * count;
630 int bytes_written = p_file->write((
const uint8_t *)data, to_write_bytes);
632 int elements_written = bytes_written /
sizeof(T);
633 if (bytes_written != to_write_bytes) {
634 LOGE(
"write: %d -> %d bytes", to_write_bytes, bytes_written);
636 return elements_written;
641 LOGD(
"file_read: %d", count);
642 int read_bytes = count *
sizeof(T);
643 int result_bytes = p_file->readBytes((
char *)result, read_bytes);
644 int result_count = result_bytes /
sizeof(T);
645 if (result_count != count) {
646 LOGE(
"readBytes: %d -> %d", read_bytes, result_bytes);
664 virtual ~NBuffer() { freeMemory(); }
667 bool read(T &result)
override {
669 return actual_read_buffer->read(result);
673 bool peek(T &result)
override {
675 return actual_read_buffer->peek(result);
684 if (actual_write_buffer ==
nullptr) {
685 actual_write_buffer = getNextAvailableBuffer();
687 if (actual_write_buffer !=
nullptr) {
688 result = actual_write_buffer->write(data);
690 if (actual_write_buffer->isFull()) {
691 addFilledBuffer(actual_write_buffer);
692 actual_write_buffer = getNextAvailableBuffer();
696 if (start_time == 0l) {
699 if (result) sample_count++;
706 if (actual_read_buffer ==
nullptr) {
707 actual_read_buffer = getNextFilledBuffer();
709 if (actual_read_buffer ==
nullptr) {
712 int result = actual_read_buffer->available();
717 (actual_read_buffer ==
nullptr) ? 0 : actual_read_buffer->available();
724 if (actual_write_buffer ==
nullptr) {
725 actual_write_buffer = getNextAvailableBuffer();
728 if (actual_write_buffer ==
nullptr) {
732 if (actual_write_buffer->isFull()) {
735 addFilledBuffer(actual_write_buffer);
736 actual_write_buffer = getNextAvailableBuffer();
738 return actual_write_buffer->availableForWrite();
744 while (actual_read_buffer !=
nullptr) {
745 actual_read_buffer->reset();
746 addAvailableBuffer(actual_read_buffer);
748 actual_read_buffer = getNextFilledBuffer();
754 unsigned long run_time = (
millis() - start_time);
755 return run_time == 0 ? 0 : sample_count * 1000 / run_time;
760 return actual_read_buffer ==
nullptr ? nullptr
761 : actual_read_buffer->address();
771 int count = bytes / buffer_size;
772 return resize(buffer_size, count);
777 if (buffer_size ==
size && buffer_count == count)
return true;
779 filled_buffers.resize(count);
780 available_buffers.resize(count);
784 buffer_count = count;
786 for (
int j = 0; j < count; j++) {
788 LOGD(
"new buffer %p", buffer);
789 available_buffers.enqueue(buffer);
795 size_t size() {
return buffer_size * buffer_count; }
798 int buffer_size = 1024;
799 uint16_t buffer_count = 0;
803 QueueFromVector<BaseBuffer<T> *> filled_buffers{0,
nullptr};
804 unsigned long start_time = 0;
805 unsigned long sample_count = 0;
811 if (actual_write_buffer) {
812 LOGD(
"deleting %p", actual_write_buffer);
813 delete actual_write_buffer;
814 actual_write_buffer =
nullptr;
816 if (actual_read_buffer) {
817 LOGD(
"deleting %p", actual_read_buffer);
818 delete actual_read_buffer;
819 actual_read_buffer =
nullptr;
823 while (ptr !=
nullptr) {
824 LOGD(
"deleting %p", ptr);
826 ptr = getNextAvailableBuffer();
829 ptr = getNextFilledBuffer();
830 while (ptr !=
nullptr) {
831 LOGD(
"deleting %p", ptr);
833 ptr = getNextFilledBuffer();
837 void resetCurrent() {
838 if (actual_read_buffer !=
nullptr) {
839 actual_read_buffer->
reset();
840 addAvailableBuffer(actual_read_buffer);
843 actual_read_buffer = getNextFilledBuffer();
846 virtual BaseBuffer<T> *getNextAvailableBuffer() {
847 if (available_buffers.empty())
return nullptr;
848 BaseBuffer<T> *result =
nullptr;
849 available_buffers.dequeue(result);
853 virtual bool addAvailableBuffer(BaseBuffer<T> *buffer) {
854 return available_buffers.enqueue(buffer);
857 virtual BaseBuffer<T> *getNextFilledBuffer() {
858 if (filled_buffers.empty())
return nullptr;
859 BaseBuffer<T> *result =
nullptr;
860 filled_buffers.dequeue(result);
864 virtual bool addFilledBuffer(BaseBuffer<T> *buffer) {
865 return filled_buffers.enqueue(buffer);
883 actual_write_buffer = getNextAvailableBuffer();
884 if (actual_write_buffer !=
nullptr) {
885 addFilledBuffer(actual_write_buffer);
900 for (
auto &buffer : this->filled_buffers.toVector()) {
902 if (sbuffer->
id ==
id) {
912 using NBuffer<T>::resetCurrent;
913 using NBuffer<T>::addFilledBuffer;
914 using NBuffer<T>::getNextAvailableBuffer;
915 using NBuffer<T>::actual_write_buffer;
916 using NBuffer<T>::actual_read_buffer;
929template <
class File,
typename T>
933 NBufferFile(
int fileSize) { number_of_objects_per_file = fileSize; }
939 next_file_name.
set(
"buffer-");
941 snprintf(number, 40,
"%d", file_count);
942 next_file_name.
add(number);
943 next_file_name.
add(
".tmp");
944 return next_file_name.
c_str();
950 if (!file)
return false;
951 empty_files.enqueue(file);
961 if (!filled_files.dequeue(read_file)) {
968 int result = read_file.readBytes((
char *)data, len *
sizeof(T)) /
sizeof(T);
973 empty_files.enqueue(read_file);
980 size_t pos = read_file.position();
981 bool result =
read(data);
989 if (!write_file || write_file.size() + len > number_of_objects_per_file) {
993 filled_files.enqueue(write_file);
996 if (!empty_files.dequeue(write_file))
return false;
998 int result = write_file.write((uint8_t *)data, len *
sizeof(T));
999 return result /
sizeof(T);
1003 return filled_files.size() * number_of_objects_per_file +
1004 (read_file.available() /
sizeof(T));
1010 number_of_objects_per_file - (write_file.available() /
sizeof(T));
1011 return empty_files.size() * number_of_objects_per_file +
1012 write_file.available() + open_current;
1015 size_t size()
override {
return number_of_objects_per_file * file_count; }
1019 cleanupFile(read_file);
1020 cleanupFile(write_file);
1022 while (empty_files.dequeue(file)) cleanupFile(file);
1023 while (filled_files.dequeue(file)) cleanupFile(file);
1028 file_delete_callback = cb;
1034 empty_files.enqueue(read_file);
1039 empty_files.enqueue(write_file);
1043 while (filled_files.dequeue(file)) {
1045 empty_files.enqueue(file);
1057 int number_of_objects_per_file = 0;
1059 const uint16_t max_file_name = 256;
1061 void (*file_delete_callback)(
const char *filename);
1063 void cleanupFile(
File &file) {
1066 int len = strlen(file.name());
1067 char file_name[len + 1];
1068 strncpy(file_name, file.name(), len);
1070 file_delete_callback(file_name);
1083template <
typename T>
1087 LOGI(
"BufferedArray(%d)", len);
1092 int16_t *getValues(
size_t offset,
size_t length) {
1093 LOGD(
"getValues(%d,%d) - max %d", offset, length, array.size());
1097 actual_end = length;
1100 last_end = actual_end >= 0 ? actual_end : offset;
1102 actual_end = offset + length > actual_end ? offset + length : actual_end;
1104 int size = actual_end - last_end;
1106 LOGD(
"readBytes(%d,%d)", last_end, size);
1107 assert(last_end + size <= array.size());
1108 p_stream->readBytes((uint8_t *)(&array[last_end]), size * 2);
1110 assert(offset < actual_end);
1111 return &array[offset];
1115 int actual_end = -1;
1118 Stream *p_stream =
nullptr;