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;
191 this->owns_buffer =
false;
192 this->buffer = (uint8_t *)
data;
193 this->current_read_pos = 0;
194 this->current_write_pos = len;
198 if (size() == 0)
resize(len);
204 if (current_write_pos < buffer.size()) {
205 buffer[current_write_pos++] = sample;
211 bool read(T &result)
override {
212 bool success =
false;
213 if (current_read_pos < current_write_pos) {
214 result = buffer[current_read_pos++];
220 bool peek(T &result)
override {
221 bool success =
false;
222 if (current_read_pos < current_write_pos) {
223 result = buffer[current_read_pos];
230 int result = current_write_pos - current_read_pos;
231 return max(result, 0);
238 int peekArray(uint8_t *
data,
int len) {
240 if (len > len_available) {
243 memcpy(
data, buffer.data() + current_read_pos, len);
252 return len_available;
254 current_read_pos += len;
255 len_available -= len;
256 memmove(buffer.data(), buffer.data() + current_read_pos, len_available *
sizeof(T));
257 current_read_pos = 0;
258 current_write_pos = len_available;
260 if (is_clear_with_zero) {
261 memset(buffer.data() + current_write_pos, 0,
262 buffer.size() - current_write_pos);
271 memmove(buffer.data(), buffer.data() + current_read_pos, av *
sizeof(T));
272 current_write_pos = av;
273 current_read_pos = 0;
277 T *
address()
override {
return buffer.data(); }
280 T *
data() {
return buffer.data() + current_read_pos; }
283 current_read_pos = 0;
284 current_write_pos = 0;
285 if (is_clear_with_zero) {
286 memset(buffer.data(), 0, buffer.size());
293 size_t result = min(available_size, (
size_t)buffer.size());
294 current_read_pos = 0;
295 current_write_pos = result;
299 size_t size()
override {
return buffer.size(); }
302 if (buffer.size() < size) {
323 int current_read_pos = 0;
324 int current_write_pos = 0;
325 bool owns_buffer =
true;
326 bool is_clear_with_zero =
false;
343 bool read(T &result)
override {
348 result = _aucBuffer[_iTail];
349 _iTail = nextIndex(_iTail);
356 bool peek(T &result)
override {
361 result = _aucBuffer[_iTail];
365 virtual int peekArray(T *data,
int n) {
366 if (isEmpty())
return -1;
368 int count = _numElems;
370 for (
int j = 0; j < n; j++) {
371 data[j] = _aucBuffer[tail];
372 tail = nextIndex(tail);
375 if (count == 0)
break;
383 bool isEmpty() {
return available() == 0; }
386 virtual bool write(T data)
override {
389 _aucBuffer[_iHead] = data;
390 _iHead = nextIndex(_iHead);
411 virtual T *
address()
override {
return _aucBuffer.data(); }
414 if (max_size != len) {
415 LOGI(
"resize: %d", len);
416 _aucBuffer.resize(len);
423 virtual size_t size()
override {
return max_size; }
432 int nextIndex(
int index) {
433 if (max_size == 0)
return 0;
434 return (uint32_t)(index + 1) % max_size; }
445template <
class File,
typename T>
454 if (p_file) p_file->close();
460 p_file = &bufferFile;
462 LOGE(
"file is not valid");
472 if (p_file ==
nullptr)
return 0;
473 int read_count = min(count,
available());
476 if (!
file_seek(offset.pos))
return false;
478 if (offset.len1 > 0) {
480 n +=
file_read(data + offset.len, offset.len1);
481 read_pos = offset.len1;
483 read_pos += read_count;
486 for (
int i = 0; i < count; i++) {
487 LOGI(
"read #%d value %d", offset.pos, (
int)data[i]);
490 element_count -= read_count;
495 bool peek(T &result)
override {
496 if (p_file ==
nullptr || isEmpty()) {
507 if (p_file ==
nullptr)
return 0;
508 int read_count = min(count,
available());
511 if (!
file_seek(offset.pos))
return false;
513 if (offset.len1 > 0) {
515 n +=
file_read(data + offset.len, offset.len1);
517 assert(n == read_count);
526 if (p_file ==
nullptr)
return 0;
527 for (
int i = 0; i < len; i++) {
528 LOGI(
"write #%d value %d", write_pos, (
int)data[i]);
534 if (!
file_seek(offset.pos))
return false;
536 if (offset.len1 > 0) {
538 n +=
file_write(data + offset.len, offset.len1);
539 write_pos = offset.len1;
541 write_pos += write_count;
543 element_count += write_count;
550 bool isEmpty() {
return available() == 0; }
567 size_t size()
override {
return max_size; }
579 File *p_file =
nullptr;
582 int element_count = 0;
596 int overflow = (pos + len) - max_size;
603 result.len = len - overflow;
604 result.len1 = overflow;
611 int file_pos = pos *
sizeof(T);
612 if (p_file->position() != file_pos) {
613 LOGD(
"file_seek: %d", pos);
614 if (!p_file->seek(file_pos)) {
615 LOGE(
"seek %d", file_pos);
624 LOGD(
"file_write: %d", count);
625 if (p_file ==
nullptr)
return 0;
626 int to_write_bytes =
sizeof(T) * count;
627 int bytes_written = p_file->write((
const uint8_t *)data, to_write_bytes);
629 int elements_written = bytes_written /
sizeof(T);
630 if (bytes_written != to_write_bytes) {
631 LOGE(
"write: %d -> %d bytes", to_write_bytes, bytes_written);
633 return elements_written;
638 LOGD(
"file_read: %d", count);
639 int read_bytes = count *
sizeof(T);
640 int result_bytes = p_file->readBytes((
char *)result, read_bytes);
641 int result_count = result_bytes /
sizeof(T);
642 if (result_count != count) {
643 LOGE(
"readBytes: %d -> %d", read_bytes, result_bytes);
661 virtual ~NBuffer() { freeMemory(); }
664 bool read(T &result)
override {
666 return actual_read_buffer->read(result);
670 bool peek(T &result)
override {
672 return actual_read_buffer->peek(result);
681 if (actual_write_buffer ==
nullptr) {
682 actual_write_buffer = getNextAvailableBuffer();
684 if (actual_write_buffer !=
nullptr) {
685 result = actual_write_buffer->write(data);
687 if (actual_write_buffer->isFull()) {
688 addFilledBuffer(actual_write_buffer);
689 actual_write_buffer = getNextAvailableBuffer();
693 if (start_time == 0l) {
696 if (result) sample_count++;
703 if (actual_read_buffer ==
nullptr) {
704 actual_read_buffer = getNextFilledBuffer();
706 if (actual_read_buffer ==
nullptr) {
709 int result = actual_read_buffer->available();
714 (actual_read_buffer ==
nullptr) ? 0 : actual_read_buffer->available();
721 if (actual_write_buffer ==
nullptr) {
722 actual_write_buffer = getNextAvailableBuffer();
725 if (actual_write_buffer ==
nullptr) {
729 if (actual_write_buffer->isFull()) {
732 addFilledBuffer(actual_write_buffer);
733 actual_write_buffer = getNextAvailableBuffer();
735 return actual_write_buffer->availableForWrite();
741 while (actual_read_buffer !=
nullptr) {
742 actual_read_buffer->reset();
743 addAvailableBuffer(actual_read_buffer);
745 actual_read_buffer = getNextFilledBuffer();
751 unsigned long run_time = (
millis() - start_time);
752 return run_time == 0 ? 0 : sample_count * 1000 / run_time;
757 return actual_read_buffer ==
nullptr ? nullptr
758 : actual_read_buffer->address();
768 int count = bytes / buffer_size;
769 return resize(buffer_size, count);
774 if (buffer_size ==
size && buffer_count == count)
return true;
776 filled_buffers.resize(count);
777 available_buffers.resize(count);
781 buffer_count = count;
783 for (
int j = 0; j < count; j++) {
785 LOGD(
"new buffer %p", buffer);
786 available_buffers.enqueue(buffer);
792 size_t size() {
return buffer_size * buffer_count; }
795 int buffer_size = 1024;
796 uint16_t buffer_count = 0;
800 QueueFromVector<BaseBuffer<T> *> filled_buffers{0,
nullptr};
801 unsigned long start_time = 0;
802 unsigned long sample_count = 0;
808 if (actual_write_buffer) {
809 LOGD(
"deleting %p", actual_write_buffer);
810 delete actual_write_buffer;
811 actual_write_buffer =
nullptr;
813 if (actual_read_buffer) {
814 LOGD(
"deleting %p", actual_read_buffer);
815 delete actual_read_buffer;
816 actual_read_buffer =
nullptr;
820 while (ptr !=
nullptr) {
821 LOGD(
"deleting %p", ptr);
823 ptr = getNextAvailableBuffer();
826 ptr = getNextFilledBuffer();
827 while (ptr !=
nullptr) {
828 LOGD(
"deleting %p", ptr);
830 ptr = getNextFilledBuffer();
834 void resetCurrent() {
835 if (actual_read_buffer !=
nullptr) {
836 actual_read_buffer->
reset();
837 addAvailableBuffer(actual_read_buffer);
840 actual_read_buffer = getNextFilledBuffer();
843 virtual BaseBuffer<T> *getNextAvailableBuffer() {
844 if (available_buffers.empty())
return nullptr;
845 BaseBuffer<T> *result =
nullptr;
846 available_buffers.dequeue(result);
850 virtual bool addAvailableBuffer(BaseBuffer<T> *buffer) {
851 return available_buffers.enqueue(buffer);
854 virtual BaseBuffer<T> *getNextFilledBuffer() {
855 if (filled_buffers.empty())
return nullptr;
856 BaseBuffer<T> *result =
nullptr;
857 filled_buffers.dequeue(result);
861 virtual bool addFilledBuffer(BaseBuffer<T> *buffer) {
862 return filled_buffers.enqueue(buffer);
880 actual_write_buffer = getNextAvailableBuffer();
881 if (actual_write_buffer !=
nullptr) {
882 addFilledBuffer(actual_write_buffer);
897 for (
auto &buffer : this->filled_buffers.toVector()) {
899 if (sbuffer->
id ==
id) {
909 using NBuffer<T>::resetCurrent;
910 using NBuffer<T>::addFilledBuffer;
911 using NBuffer<T>::getNextAvailableBuffer;
912 using NBuffer<T>::actual_write_buffer;
913 using NBuffer<T>::actual_read_buffer;
926template <
class File,
typename T>
930 NBufferFile(
int fileSize) { number_of_objects_per_file = fileSize; }
936 next_file_name.
set(
"buffer-");
938 snprintf(number, 40,
"%d", file_count);
939 next_file_name.
add(number);
940 next_file_name.
add(
".tmp");
941 return next_file_name.
c_str();
947 if (!file)
return false;
948 empty_files.enqueue(file);
958 if (!filled_files.dequeue(read_file)) {
965 int result = read_file.readBytes((
char *)data, len *
sizeof(T)) /
sizeof(T);
970 empty_files.enqueue(read_file);
977 size_t pos = read_file.position();
978 bool result =
read(data);
986 if (!write_file || write_file.size() + len > number_of_objects_per_file) {
990 filled_files.enqueue(write_file);
993 if (!empty_files.dequeue(write_file))
return false;
995 int result = write_file.write((uint8_t *)data, len *
sizeof(T));
996 return result /
sizeof(T);
1000 return filled_files.size() * number_of_objects_per_file +
1001 (read_file.available() /
sizeof(T));
1007 number_of_objects_per_file - (write_file.available() /
sizeof(T));
1008 return empty_files.size() * number_of_objects_per_file +
1009 write_file.available() + open_current;
1012 size_t size()
override {
return number_of_objects_per_file * file_count; }
1016 cleanupFile(read_file);
1017 cleanupFile(write_file);
1019 while (empty_files.dequeue(file)) cleanupFile(file);
1020 while (filled_files.dequeue(file)) cleanupFile(file);
1025 file_delete_callback = cb;
1031 empty_files.enqueue(read_file);
1036 empty_files.enqueue(write_file);
1040 while (filled_files.dequeue(file)) {
1042 empty_files.enqueue(file);
1054 int number_of_objects_per_file = 0;
1056 const uint16_t max_file_name = 256;
1058 void (*file_delete_callback)(
const char *filename);
1060 void cleanupFile(
File &file) {
1063 int len = strlen(file.name());
1064 char file_name[len + 1];
1065 strncpy(file_name, file.name(), len);
1067 file_delete_callback(file_name);
1080template <
typename T>
1084 LOGI(
"BufferedArray(%d)", len);
1089 int16_t *getValues(
size_t offset,
size_t length) {
1090 LOGD(
"getValues(%d,%d) - max %d", offset, length, array.size());
1094 actual_end = length;
1097 last_end = actual_end >= 0 ? actual_end : offset;
1099 actual_end = offset + length > actual_end ? offset + length : actual_end;
1101 int size = actual_end - last_end;
1103 LOGD(
"readBytes(%d,%d)", last_end, size);
1104 assert(last_end + size <= array.size());
1105 p_stream->readBytes((uint8_t *)(&array[last_end]), size * 2);
1107 assert(offset < actual_end);
1108 return &array[offset];
1112 int actual_end = -1;
1115 Stream *p_stream =
nullptr;