From 61c91b3cdb2c9dd655f3adf0f461f5cefb3b2e9b Mon Sep 17 00:00:00 2001 From: jacqueline Date: Fri, 10 Feb 2023 15:32:21 +1100 Subject: Mostly working pipeline, including proper EOF signalling --- src/audio/audio_decoder.cpp | 12 +++- src/audio/audio_playback.cpp | 8 ++- src/audio/audio_task.cpp | 10 ++- src/audio/chunk.cpp | 11 +++- src/audio/fatfs_audio_input.cpp | 10 +++ src/audio/i2s_audio_output.cpp | 107 +++++++++++++++++++++++++++++--- src/audio/include/audio_decoder.hpp | 1 + src/audio/include/audio_element.hpp | 2 + src/audio/include/audio_task.hpp | 4 ++ src/audio/include/chunk.hpp | 5 +- src/audio/include/fatfs_audio_input.hpp | 1 + src/audio/include/i2s_audio_output.hpp | 12 +++- src/audio/include/stream_event.hpp | 2 + src/audio/stream_event.cpp | 11 ++++ 14 files changed, 177 insertions(+), 19 deletions(-) (limited to 'src/audio') diff --git a/src/audio/audio_decoder.cpp b/src/audio/audio_decoder.cpp index 27b9cca3..d90ca496 100644 --- a/src/audio/audio_decoder.cpp +++ b/src/audio/audio_decoder.cpp @@ -84,6 +84,16 @@ auto AudioDecoder::ProcessChunk(const cpp::span& chunk) return {}; } +auto AudioDecoder::ProcessEndOfStream() -> void { + has_samples_to_send_ = false; + needs_more_input_ = true; + current_codec_.reset(); + + SendOrBufferEvent( + std::unique_ptr( + StreamEvent::CreateEndOfStream(input_events_))); +} + auto AudioDecoder::Process() -> cpp::result { if (has_samples_to_send_) { ESP_LOGI(kTag, "sending samples"); @@ -132,7 +142,7 @@ auto AudioDecoder::Process() -> cpp::result { has_samples_to_send_ = true; if (needs_more_input_) { - chunk_reader_->HandleLeftovers(current_codec_->GetInputPosition()); + chunk_reader_->HandleBytesUsed(current_codec_->GetInputPosition()); } } diff --git a/src/audio/audio_playback.cpp b/src/audio/audio_playback.cpp index 4df598c5..504a2a4e 100644 --- a/src/audio/audio_playback.cpp +++ b/src/audio/audio_playback.cpp @@ -38,9 +38,9 @@ auto AudioPlayback::create(drivers::GpioExpander* expander, playback->ConnectElements(codec.get(), sink.get()); // Launch! - playback->element_handles_.push_back(StartAudioTask("src", source)); - playback->element_handles_.push_back(StartAudioTask("dec", codec)); - playback->element_handles_.push_back(StartAudioTask("sink", sink)); + playback->element_handles_.push_back(StartAudioTask("src", {}, source)); + playback->element_handles_.push_back(StartAudioTask("dec", {}, codec)); + playback->element_handles_.push_back(StartAudioTask("sink", 0, sink)); playback->input_handle_ = source->InputEventQueue(); @@ -60,6 +60,8 @@ auto AudioPlayback::Play(const std::string& filename) -> void { info.path = filename; auto event = StreamEvent::CreateStreamInfo(input_handle_, info); xQueueSend(input_handle_, &event, portMAX_DELAY); + event = StreamEvent::CreateEndOfStream(input_handle_); + xQueueSend(input_handle_, &event, portMAX_DELAY); } auto AudioPlayback::ConnectElements(IAudioElement* src, IAudioElement* sink) diff --git a/src/audio/audio_task.cpp b/src/audio/audio_task.cpp index 47b115cc..078aa461 100644 --- a/src/audio/audio_task.cpp +++ b/src/audio/audio_task.cpp @@ -28,6 +28,7 @@ namespace audio { static const char* kTag = "task"; auto StartAudioTask(const std::string& name, + std::optional core_id, std::shared_ptr element) -> std::unique_ptr { auto task_handle = std::make_unique(); @@ -36,8 +37,13 @@ auto StartAudioTask(const std::string& name, AudioTaskArgs* args = new AudioTaskArgs{.element = element}; ESP_LOGI(kTag, "starting audio task %s", name.c_str()); - xTaskCreate(&AudioTaskMain, name.c_str(), element->StackSizeBytes(), args, - kTaskPriorityAudio, task_handle.get()); + if (core_id) { + xTaskCreatePinnedToCore(&AudioTaskMain, name.c_str(), element->StackSizeBytes(), args, + kTaskPriorityAudio, task_handle.get(), *core_id); + } else { + xTaskCreate(&AudioTaskMain, name.c_str(), element->StackSizeBytes(), args, + kTaskPriorityAudio, task_handle.get()); + } return std::make_unique(std::move(task_handle), element); } diff --git a/src/audio/chunk.cpp b/src/audio/chunk.cpp index 61c0dc2b..01760a84 100644 --- a/src/audio/chunk.cpp +++ b/src/audio/chunk.cpp @@ -41,8 +41,11 @@ auto ChunkReader::HandleNewData(cpp::span data) return last_data_in_working_buffer_; } -auto ChunkReader::HandleLeftovers(std::size_t bytes_used) -> void { - leftover_bytes_ = last_data_in_working_buffer_.size() - bytes_used; +auto ChunkReader::HandleBytesUsed(std::size_t bytes_used) -> void { + HandleBytesLeftOver(last_data_in_working_buffer_.size() - bytes_used); +} +auto ChunkReader::HandleBytesLeftOver(std::size_t bytes_left) -> void { + leftover_bytes_ = bytes_left; // Ensure that we don't have more than a chunk of leftever bytes. This is // bad, because we probably won't have enough data to store the next chunk. @@ -55,4 +58,8 @@ auto ChunkReader::HandleLeftovers(std::size_t bytes_used) -> void { } } +auto ChunkReader::GetLeftovers() -> cpp::span { + return working_buffer_.first(leftover_bytes_); +} + } // namespace audio diff --git a/src/audio/fatfs_audio_input.cpp b/src/audio/fatfs_audio_input.cpp index 08f39347..8990bf4f 100644 --- a/src/audio/fatfs_audio_input.cpp +++ b/src/audio/fatfs_audio_input.cpp @@ -69,6 +69,16 @@ auto FatfsAudioInput::ProcessChunk(const cpp::span& chunk) return cpp::fail(UNSUPPORTED_STREAM); } +auto FatfsAudioInput::ProcessEndOfStream() -> void { + if (is_file_open_) { + f_close(¤t_file_); + is_file_open_ = false; + SendOrBufferEvent( + std::unique_ptr( + StreamEvent::CreateEndOfStream(input_events_))); + } +} + auto FatfsAudioInput::Process() -> cpp::result { if (is_file_open_) { auto dest_event = std::unique_ptr( diff --git a/src/audio/i2s_audio_output.cpp b/src/audio/i2s_audio_output.cpp index b00e31d3..3bed15b5 100644 --- a/src/audio/i2s_audio_output.cpp +++ b/src/audio/i2s_audio_output.cpp @@ -7,6 +7,7 @@ #include "audio_element.hpp" #include "dac.hpp" +#include "freertos/projdefs.h" #include "gpio_expander.hpp" #include "result.hpp" @@ -15,6 +16,8 @@ static const char* kTag = "I2SOUT"; namespace audio { +static const std::size_t kDmaQueueLength = 8; + auto I2SAudioOutput::create(drivers::GpioExpander* expander) -> cpp::result, Error> { // First, we need to perform initial configuration of the DAC chip. @@ -38,12 +41,26 @@ I2SAudioOutput::I2SAudioOutput(drivers::GpioExpander* expander, : expander_(expander), dac_(std::move(dac)), volume_(255), - is_soft_muted_(false) {} + is_soft_muted_(false), + chunk_reader_(), + latest_chunk_(), + dma_size_(), + dma_queue_(nullptr) {} I2SAudioOutput::~I2SAudioOutput() { + if (dma_queue_ != nullptr) { + ClearDmaQueue(); + } // TODO: power down the DAC. } +auto I2SAudioOutput::HasUnprocessedInput() -> bool { + if (dma_queue_ == nullptr || !dma_size_) { + return false; + } + return latest_chunk_.size() >= *dma_size_; +} + auto I2SAudioOutput::ProcessStreamInfo(const StreamInfo& info) -> cpp::result { // TODO(jacqueline): probs do something with the channel hey @@ -53,6 +70,12 @@ auto I2SAudioOutput::ProcessStreamInfo(const StreamInfo& info) return cpp::fail(UNSUPPORTED_STREAM); } + if (!info.chunk_size) { + ESP_LOGE(kTag, "audio stream missing chunk size"); + return cpp::fail(UNSUPPORTED_STREAM); + } + chunk_reader_.emplace(*info.chunk_size); + ESP_LOGI(kTag, "incoming audio stream: %u bpp @ %u Hz", *info.bits_per_sample, *info.sample_rate); @@ -85,23 +108,78 @@ auto I2SAudioOutput::ProcessStreamInfo(const StreamInfo& info) return cpp::fail(UNSUPPORTED_STREAM); } - dac_->Reconfigure(bps, sample_rate); + QueueHandle_t new_dma_queue = + xQueueCreate(kDmaQueueLength, sizeof(std::byte*)); + + dma_size_ = dac_->Reconfigure(bps, sample_rate, new_dma_queue); + + if (dma_queue_ != nullptr) { + ClearDmaQueue(); + } + dma_queue_ = new_dma_queue; return {}; } auto I2SAudioOutput::ProcessChunk(const cpp::span& chunk) -> cpp::result { - ESP_LOGI(kTag, "playing samples"); - SetSoftMute(false); - // TODO(jacqueline): write smaller parts with a small delay so that we can - // be responsive to pause and seek commands. - dac_->WriteData(chunk, portMAX_DELAY); + ESP_LOGI(kTag, "received new samples"); + latest_chunk_ = chunk_reader_->HandleNewData(chunk); return 0; } +auto I2SAudioOutput::ProcessEndOfStream() -> void { + if (chunk_reader_ && dma_size_) { + auto leftovers = chunk_reader_->GetLeftovers(); + if (leftovers.size() > 0 && leftovers.size() < *dma_size_) { + std::byte* dest = static_cast(malloc(*dma_size_)); + cpp::span dest_span(dest, *dma_size_); + + std::copy(leftovers.begin(), leftovers.end(), dest_span.begin()); + std::fill(dest_span.begin() + leftovers.size(), dest_span.end(), static_cast(0)); + + xQueueSend(dma_queue_, &dest, portMAX_DELAY); + } + } + + SendOrBufferEvent( + std::unique_ptr( + StreamEvent::CreateEndOfStream(input_events_))); + + chunk_reader_.reset(); + dma_size_.reset(); +} + auto I2SAudioOutput::Process() -> cpp::result { - // TODO(jacqueline): Play the stream in smaller sections + std::size_t spaces_available = uxQueueSpacesAvailable(dma_queue_); + if (spaces_available == 0) { + // TODO: think about this more. can this just be the output event queue? + vTaskDelay(pdMS_TO_TICKS(100)); + return {}; + } + + // Fill the queue as much as possible, since we need to be able to stream + // FAST. + while (latest_chunk_.size() >= *dma_size_ && spaces_available > 0) { + // TODO: small memory arena for this? + std::byte* dest = static_cast(malloc(*dma_size_)); + cpp::span dest_span(dest, *dma_size_); + cpp::span src_span = latest_chunk_.first(*dma_size_); + std::copy(src_span.begin(), src_span.end(), dest_span.begin()); + if (!xQueueSend(dma_queue_, &dest, 0)) { + // TODO: calculate how often we expect this to happen. + free(dest); + break; + } + latest_chunk_ = latest_chunk_.subspan(*dma_size_); + ESP_LOGI(kTag, "wrote dma buffer of size %u", *dma_size_); + } + if (latest_chunk_.size() < *dma_size_) { + // TODO: if this is the end of the stream, then we should be sending this + // with zero padding. hmm. i guess we need an explicit EOF event? + chunk_reader_->HandleBytesLeftOver(latest_chunk_.size()); + ESP_LOGI(kTag, "not enough samples for dma buffer"); + } return {}; } @@ -124,4 +202,17 @@ auto I2SAudioOutput::SetSoftMute(bool enabled) -> void { } } +auto I2SAudioOutput::ClearDmaQueue() -> void { + // Ensure we don't leak any memory from events leftover in the queue. + while (uxQueueSpacesAvailable(dma_queue_) < kDmaQueueLength) { + std::byte* data = nullptr; + if (xQueueReceive(input_events_, &data, 0)) { + free(data); + } else { + break; + } + } + vQueueDelete(dma_queue_); +} + } // namespace audio diff --git a/src/audio/include/audio_decoder.hpp b/src/audio/include/audio_decoder.hpp index d0f7469b..aa83825f 100644 --- a/src/audio/include/audio_decoder.hpp +++ b/src/audio/include/audio_decoder.hpp @@ -37,6 +37,7 @@ class AudioDecoder : public IAudioElement { -> cpp::result override; auto ProcessChunk(const cpp::span& chunk) -> cpp::result override; + auto ProcessEndOfStream() -> void override; auto Process() -> cpp::result override; AudioDecoder(const AudioDecoder&) = delete; diff --git a/src/audio/include/audio_element.hpp b/src/audio/include/audio_element.hpp index c6453d4d..0c80524c 100644 --- a/src/audio/include/audio_element.hpp +++ b/src/audio/include/audio_element.hpp @@ -105,6 +105,8 @@ class IAudioElement { virtual auto ProcessChunk(const cpp::span& chunk) -> cpp::result = 0; + virtual auto ProcessEndOfStream() -> void = 0; + /* * Called when there has been no data received over the input buffer for some * time. This could be used to synthesize output, or to save memory by diff --git a/src/audio/include/audio_task.hpp b/src/audio/include/audio_task.hpp index 9a76ea7e..399ad679 100644 --- a/src/audio/include/audio_task.hpp +++ b/src/audio/include/audio_task.hpp @@ -1,9 +1,12 @@ #pragma once #include +#include +#include #include "audio_element.hpp" #include "audio_element_handle.hpp" +#include "freertos/portmacro.h" namespace audio { @@ -12,6 +15,7 @@ struct AudioTaskArgs { }; auto StartAudioTask(const std::string& name, + std::optional core_id, std::shared_ptr element) -> std::unique_ptr; diff --git a/src/audio/include/chunk.hpp b/src/audio/include/chunk.hpp index 0ece1ed6..31e1f969 100644 --- a/src/audio/include/chunk.hpp +++ b/src/audio/include/chunk.hpp @@ -27,7 +27,8 @@ class ChunkReader { explicit ChunkReader(std::size_t chunk_size); ~ChunkReader(); - auto HandleLeftovers(std::size_t bytes_used) -> void; + auto HandleBytesLeftOver(std::size_t bytes_left) -> void; + auto HandleBytesUsed(std::size_t bytes_used) -> void; /* * Reads chunks of data from the given input stream, and invokes the given @@ -43,6 +44,8 @@ class ChunkReader { */ auto HandleNewData(cpp::span data) -> cpp::span; + auto GetLeftovers() -> cpp::span; + ChunkReader(const ChunkReader&) = delete; ChunkReader& operator=(const ChunkReader&) = delete; diff --git a/src/audio/include/fatfs_audio_input.hpp b/src/audio/include/fatfs_audio_input.hpp index 5625d941..883441c2 100644 --- a/src/audio/include/fatfs_audio_input.hpp +++ b/src/audio/include/fatfs_audio_input.hpp @@ -28,6 +28,7 @@ class FatfsAudioInput : public IAudioElement { -> cpp::result override; auto ProcessChunk(const cpp::span& chunk) -> cpp::result override; + auto ProcessEndOfStream() -> void override; auto Process() -> cpp::result override; FatfsAudioInput(const FatfsAudioInput&) = delete; diff --git a/src/audio/include/i2s_audio_output.hpp b/src/audio/include/i2s_audio_output.hpp index 02ac7a16..b4fd4c59 100644 --- a/src/audio/include/i2s_audio_output.hpp +++ b/src/audio/include/i2s_audio_output.hpp @@ -4,6 +4,7 @@ #include #include "audio_element.hpp" +#include "chunk.hpp" #include "result.hpp" #include "dac.hpp" @@ -21,13 +22,13 @@ class I2SAudioOutput : public IAudioElement { std::unique_ptr dac); ~I2SAudioOutput(); - // TODO. - auto HasUnprocessedInput() -> bool override { return false; } + auto HasUnprocessedInput() -> bool override; auto ProcessStreamInfo(const StreamInfo& info) -> cpp::result override; auto ProcessChunk(const cpp::span& chunk) -> cpp::result override; + auto ProcessEndOfStream() -> void override; auto Process() -> cpp::result override; I2SAudioOutput(const I2SAudioOutput&) = delete; @@ -37,11 +38,18 @@ class I2SAudioOutput : public IAudioElement { auto SetVolume(uint8_t volume) -> void; auto SetSoftMute(bool enabled) -> void; + auto ClearDmaQueue() -> void; + drivers::GpioExpander* expander_; std::unique_ptr dac_; uint8_t volume_; bool is_soft_muted_; + + std::optional chunk_reader_; + cpp::span latest_chunk_; + std::optional dma_size_; + QueueHandle_t dma_queue_; }; } // namespace audio diff --git a/src/audio/include/stream_event.hpp b/src/audio/include/stream_event.hpp index 4f441fa5..e84c8388 100644 --- a/src/audio/include/stream_event.hpp +++ b/src/audio/include/stream_event.hpp @@ -16,6 +16,7 @@ struct StreamEvent { static auto CreateChunkData(QueueHandle_t source, std::size_t chunk_size) -> StreamEvent*; static auto CreateChunkNotification(QueueHandle_t source) -> StreamEvent*; + static auto CreateEndOfStream(QueueHandle_t source) -> StreamEvent*; StreamEvent(); ~StreamEvent(); @@ -28,6 +29,7 @@ struct StreamEvent { STREAM_INFO, CHUNK_DATA, CHUNK_NOTIFICATION, + END_OF_STREAM, } tag; union { diff --git a/src/audio/stream_event.cpp b/src/audio/stream_event.cpp index e3228680..6efebbca 100644 --- a/src/audio/stream_event.cpp +++ b/src/audio/stream_event.cpp @@ -37,6 +37,13 @@ auto StreamEvent::CreateChunkNotification(QueueHandle_t source) return event; } +auto StreamEvent::CreateEndOfStream(QueueHandle_t source) -> StreamEvent* { + auto event = new StreamEvent; + event->tag = StreamEvent::END_OF_STREAM; + event->source = source; + return event; +} + StreamEvent::StreamEvent() : tag(StreamEvent::UNINITIALISED) {} StreamEvent::~StreamEvent() { @@ -51,6 +58,8 @@ StreamEvent::~StreamEvent() { break; case CHUNK_NOTIFICATION: break; + case END_OF_STREAM: + break; } } @@ -70,6 +79,8 @@ StreamEvent::StreamEvent(StreamEvent&& other) { break; case CHUNK_NOTIFICATION: break; + case END_OF_STREAM: + break; } other.tag = StreamEvent::UNINITIALISED; } -- cgit v1.2.3