github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-06-24 16:20:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

[resampler] Refactor for stability and to support Sendspin (#12254)

Browse Source 
Co-authored-by: J. Nick Koston <nick@home-assistant.io>
This commit is contained in:
Kevin Ahrendt
2026-02-10 08:56:31 -06:00
committed by GitHub
parent d4ccc64dc0
commit 298efb5340
3 changed files with 204 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
esphome/components/resampler/speaker/__init__.py
View File

@@ -1,5 +1,5 @@
import esphome.codegen as cg
from esphome.components import audio, esp32, speaker
from esphome.components import audio, esp32, socket, speaker
import esphome.config_validation as cv
from esphome.const import (
CONF_BITS_PER_SAMPLE,
@@ -34,7 +34,7 @@ def _set_stream_limits(config):
return config
def _validate_audio_compatability(config):
def _validate_audio_compatibility(config):
inherit_property_from(CONF_BITS_PER_SAMPLE, CONF_OUTPUT_SPEAKER)(config)
inherit_property_from(CONF_NUM_CHANNELS, CONF_OUTPUT_SPEAKER)(config)
inherit_property_from(CONF_SAMPLE_RATE, CONF_OUTPUT_SPEAKER)(config)
@@ -73,10 +73,13 @@ CONFIG_SCHEMA = cv.All(
)
FINAL_VALIDATE_SCHEMA = _validate_audio_compatability
FINAL_VALIDATE_SCHEMA = _validate_audio_compatibility
async def to_code(config):
# Enable wake_loop_threadsafe for immediate command processing from other tasks
socket.require_wake_loop_threadsafe()
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
await speaker.register_speaker(var, config)
@@ -86,12 +89,11 @@ async def to_code(config):
cg.add(var.set_buffer_duration(config[CONF_BUFFER_DURATION]))
if task_stack_in_psram := config.get(CONF_TASK_STACK_IN_PSRAM):
cg.add(var.set_task_stack_in_psram(task_stack_in_psram))
if task_stack_in_psram and config[CONF_TASK_STACK_IN_PSRAM]:
esp32.add_idf_sdkconfig_option(
"CONFIG_SPIRAM_ALLOW_STACK_EXTERNAL_MEMORY", True
)
if config.get(CONF_TASK_STACK_IN_PSRAM):
cg.add(var.set_task_stack_in_psram(True))
esp32.add_idf_sdkconfig_option(
"CONFIG_SPIRAM_ALLOW_STACK_EXTERNAL_MEMORY", True
)
cg.add(var.set_target_bits_per_sample(config[CONF_BITS_PER_SAMPLE]))
cg.add(var.set_target_sample_rate(config[CONF_SAMPLE_RATE]))

247
esphome/components/resampler/speaker/resampler_speaker.cpp
View File

@@ -4,6 +4,8 @@
#include "esphome/components/audio/audio_resampler.h"
#include "esphome/core/application.h"
#include "esphome/core/defines.h"
#include "esphome/core/helpers.h"
#include "esphome/core/log.h"
@@ -17,13 +19,17 @@ static const UBaseType_t RESAMPLER_TASK_PRIORITY = 1;
static const uint32_t TRANSFER_BUFFER_DURATION_MS = 50;
static const uint32_t TASK_DELAY_MS = 20;
static const uint32_t TASK_STACK_SIZE = 3072;
static const uint32_t STATE_TRANSITION_TIMEOUT_MS = 5000;
static const char *const TAG = "resampler_speaker";
enum ResamplingEventGroupBits : uint32_t {
COMMAND_STOP = (1 << 0), // stops the resampler task
COMMAND_STOP = (1 << 0), // signals stop request
COMMAND_START = (1 << 1), // signals start request
COMMAND_FINISH = (1 << 2), // signals finish request (graceful stop)
TASK_COMMAND_STOP = (1 << 5), // signals the task to stop
STATE_STARTING = (1 << 10),
STATE_RUNNING = (1 << 11),
STATE_STOPPING = (1 << 12),
@@ -34,9 +40,16 @@ enum ResamplingEventGroupBits : uint32_t {
ALL_BITS = 0x00FFFFFF, // All valid FreeRTOS event group bits
};
void ResamplerSpeaker::dump_config() {
ESP_LOGCONFIG(TAG,
"Resampler Speaker:\n"
" Target Bits Per Sample: %u\n"
" Target Sample Rate: %" PRIu32 " Hz",
this->target_bits_per_sample_, this->target_sample_rate_);
}
void ResamplerSpeaker::setup() {
this->event_group_ = xEventGroupCreate();
if (this->event_group_ == nullptr) {
ESP_LOGE(TAG, "Failed to create event group");
this->mark_failed();
@@ -55,81 +68,155 @@ void ResamplerSpeaker::setup() {
this->audio_output_callback_(new_frames, write_timestamp);
}
});
// Start with loop disabled since no task is running and no commands are pending
this->disable_loop();
}
void ResamplerSpeaker::loop() {
uint32_t event_group_bits = xEventGroupGetBits(this->event_group_);
// Process commands with priority: STOP > FINISH > START
// This ensures stop commands take precedence over conflicting start commands
if (event_group_bits & ResamplingEventGroupBits::COMMAND_STOP) {
if (this->state_ == speaker::STATE_RUNNING || this->state_ == speaker::STATE_STARTING) {
// Clear STOP, START, and FINISH bits - stop takes precedence
xEventGroupClearBits(this->event_group_, ResamplingEventGroupBits::COMMAND_STOP |
ResamplingEventGroupBits::COMMAND_START |
ResamplingEventGroupBits::COMMAND_FINISH);
this->waiting_for_output_ = false;
this->enter_stopping_state_();
} else if (this->state_ == speaker::STATE_STOPPED) {
// Already stopped, just clear the command bits
xEventGroupClearBits(this->event_group_, ResamplingEventGroupBits::COMMAND_STOP |
ResamplingEventGroupBits::COMMAND_START |
ResamplingEventGroupBits::COMMAND_FINISH);
}
// Leave bits set if STATE_STOPPING - will be processed once stopped
} else if (event_group_bits & ResamplingEventGroupBits::COMMAND_FINISH) {
if (this->state_ == speaker::STATE_RUNNING) {
xEventGroupClearBits(this->event_group_, ResamplingEventGroupBits::COMMAND_FINISH);
this->output_speaker_->finish();
} else if (this->state_ == speaker::STATE_STOPPED) {
// Already stopped, just clear the command bit
xEventGroupClearBits(this->event_group_, ResamplingEventGroupBits::COMMAND_FINISH);
}
// Leave bit set if transitioning states - will be processed once state allows
} else if (event_group_bits & ResamplingEventGroupBits::COMMAND_START) {
if (this->state_ == speaker::STATE_STOPPED) {
xEventGroupClearBits(this->event_group_, ResamplingEventGroupBits::COMMAND_START);
this->state_ = speaker::STATE_STARTING;
} else if (this->state_ == speaker::STATE_RUNNING) {
// Already running, just clear the command bit
xEventGroupClearBits(this->event_group_, ResamplingEventGroupBits::COMMAND_START);
}
// Leave bit set if transitioning states - will be processed once state allows
}
// Re-read bits after command processing (enter_stopping_state_ may have set task bits)
event_group_bits = xEventGroupGetBits(this->event_group_);
if (event_group_bits & ResamplingEventGroupBits::STATE_STARTING) {
ESP_LOGD(TAG, "Starting resampler task");
ESP_LOGD(TAG, "Starting");
xEventGroupClearBits(this->event_group_, ResamplingEventGroupBits::STATE_STARTING);
}
if (event_group_bits & ResamplingEventGroupBits::ERR_ESP_NO_MEM) {
this->status_set_error(LOG_STR("Resampler task failed to allocate the internal buffers"));
this->status_set_error(LOG_STR("Not enough memory"));
xEventGroupClearBits(this->event_group_, ResamplingEventGroupBits::ERR_ESP_NO_MEM);
this->state_ = speaker::STATE_STOPPING;
this->enter_stopping_state_();
}
if (event_group_bits & ResamplingEventGroupBits::ERR_ESP_NOT_SUPPORTED) {
this->status_set_error(LOG_STR("Cannot resample due to an unsupported audio stream"));
this->status_set_error(LOG_STR("Unsupported stream"));
xEventGroupClearBits(this->event_group_, ResamplingEventGroupBits::ERR_ESP_NOT_SUPPORTED);
this->state_ = speaker::STATE_STOPPING;
this->enter_stopping_state_();
}
if (event_group_bits & ResamplingEventGroupBits::ERR_ESP_FAIL) {
this->status_set_error(LOG_STR("Resampler task failed"));
this->status_set_error(LOG_STR("Resampler failure"));
xEventGroupClearBits(this->event_group_, ResamplingEventGroupBits::ERR_ESP_FAIL);
this->state_ = speaker::STATE_STOPPING;
this->enter_stopping_state_();
}
if (event_group_bits & ResamplingEventGroupBits::STATE_RUNNING) {
ESP_LOGD(TAG, "Started resampler task");
ESP_LOGV(TAG, "Started");
this->status_clear_error();
xEventGroupClearBits(this->event_group_, ResamplingEventGroupBits::STATE_RUNNING);
}
if (event_group_bits & ResamplingEventGroupBits::STATE_STOPPING) {
ESP_LOGD(TAG, "Stopping resampler task");
ESP_LOGV(TAG, "Stopping");
xEventGroupClearBits(this->event_group_, ResamplingEventGroupBits::STATE_STOPPING);
}
if (event_group_bits & ResamplingEventGroupBits::STATE_STOPPED) {
if (this->delete_task_() == ESP_OK) {
ESP_LOGD(TAG, "Stopped resampler task");
xEventGroupClearBits(this->event_group_, ResamplingEventGroupBits::ALL_BITS);
}
this->delete_task_();
ESP_LOGD(TAG, "Stopped");
xEventGroupClearBits(this->event_group_, ResamplingEventGroupBits::ALL_BITS);
}
switch (this->state_) {
case speaker::STATE_STARTING: {
esp_err_t err = this->start_();
if (err == ESP_OK) {
this->status_clear_error();
this->state_ = speaker::STATE_RUNNING;
} else {
switch (err) {
case ESP_ERR_INVALID_STATE:
this->status_set_error(LOG_STR("Failed to start resampler: resampler task failed to start"));
break;
case ESP_ERR_NO_MEM:
this->status_set_error(LOG_STR("Failed to start resampler: not enough memory for task stack"));
default:
this->status_set_error(LOG_STR("Failed to start resampler"));
break;
if (!this->waiting_for_output_) {
esp_err_t err = this->start_();
if (err == ESP_OK) {
this->callback_remainder_ = 0; // reset callback remainder
this->status_clear_error();
this->waiting_for_output_ = true;
this->state_start_ms_ = App.get_loop_component_start_time();
} else {
this->set_start_error_(err);
this->waiting_for_output_ = false;
this->enter_stopping_state_();
}
} else {
if (this->output_speaker_->is_running()) {
this->state_ = speaker::STATE_RUNNING;
this->waiting_for_output_ = false;
} else if ((App.get_loop_component_start_time() - this->state_start_ms_) > STATE_TRANSITION_TIMEOUT_MS) {
// Timed out waiting for the output speaker to start
this->waiting_for_output_ = false;
this->enter_stopping_state_();
}
this->state_ = speaker::STATE_STOPPING;
}
break;
}
case speaker::STATE_RUNNING:
if (this->output_speaker_->is_stopped()) {
this->state_ = speaker::STATE_STOPPING;
this->enter_stopping_state_();
}
break;
case speaker::STATE_STOPPING: {
if ((this->output_speaker_->get_pause_state()) ||
((App.get_loop_component_start_time() - this->state_start_ms_) > STATE_TRANSITION_TIMEOUT_MS)) {
// If output speaker is paused or stopping timeout exceeded, force stop
this->output_speaker_->stop();
}
if (this->output_speaker_->is_stopped() && (this->task_handle_ == nullptr)) {
// Only transition to stopped state once the output speaker and resampler task are fully stopped
this->waiting_for_output_ = false;
this->state_ = speaker::STATE_STOPPED;
}
break;
case speaker::STATE_STOPPING:
this->stop_();
this->state_ = speaker::STATE_STOPPED;
break;
}
case speaker::STATE_STOPPED:
event_group_bits = xEventGroupGetBits(this->event_group_);
if (event_group_bits == 0) {
// No pending events, disable loop to save CPU cycles
this->disable_loop();
}
break;
}
}
void ResamplerSpeaker::set_start_error_(esp_err_t err) {
switch (err) {
case ESP_ERR_INVALID_STATE:
this->status_set_error(LOG_STR("Task failed to start"));
break;
case ESP_ERR_NO_MEM:
this->status_set_error(LOG_STR("Not enough memory"));
break;
default:
this->status_set_error(LOG_STR("Failed to start"));
break;
}
}
@@ -143,16 +230,33 @@ size_t ResamplerSpeaker::play(const uint8_t *data, size_t length, TickType_t tic
if ((this->output_speaker_->is_running()) && (!this->requires_resampling_())) {
bytes_written = this->output_speaker_->play(data, length, ticks_to_wait);
} else {
if (this->ring_buffer_.use_count() == 1) {
std::shared_ptr<RingBuffer> temp_ring_buffer = this->ring_buffer_.lock();
std::shared_ptr<RingBuffer> temp_ring_buffer = this->ring_buffer_.lock();
if (temp_ring_buffer) {
// Only write to the ring buffer if the reference is valid
bytes_written = temp_ring_buffer->write_without_replacement(data, length, ticks_to_wait);
} else {
// Delay to avoid repeatedly hammering while waiting for the speaker to start
vTaskDelay(ticks_to_wait);
}
}
return bytes_written;
}
void ResamplerSpeaker::start() { this->state_ = speaker::STATE_STARTING; }
void ResamplerSpeaker::send_command_(uint32_t command_bit, bool wake_loop) {
this->enable_loop_soon_any_context();
uint32_t event_bits = xEventGroupGetBits(this->event_group_);
if (!(event_bits & command_bit)) {
xEventGroupSetBits(this->event_group_, command_bit);
#if defined(USE_SOCKET_SELECT_SUPPORT) && defined(USE_WAKE_LOOP_THREADSAFE)
if (wake_loop) {
App.wake_loop_threadsafe();
}
#endif
}
}
void ResamplerSpeaker::start() { this->send_command_(ResamplingEventGroupBits::COMMAND_START, true); }
esp_err_t ResamplerSpeaker::start_() {
this->target_stream_info_ = audio::AudioStreamInfo(
@@ -185,7 +289,7 @@ esp_err_t ResamplerSpeaker::start_task_() {
}
if (this->task_handle_ == nullptr) {
this->task_handle_ = xTaskCreateStatic(resample_task, "sample", TASK_STACK_SIZE, (void *) this,
this->task_handle_ = xTaskCreateStatic(resample_task, "resampler", TASK_STACK_SIZE, (void *) this,
RESAMPLER_TASK_PRIORITY, this->task_stack_buffer_, &this->task_stack_);
}
@@ -196,43 +300,47 @@ esp_err_t ResamplerSpeaker::start_task_() {
return ESP_OK;
}
void ResamplerSpeaker::stop() { this->state_ = speaker::STATE_STOPPING; }
void ResamplerSpeaker::stop() { this->send_command_(ResamplingEventGroupBits::COMMAND_STOP); }
void ResamplerSpeaker::stop_() {
void ResamplerSpeaker::enter_stopping_state_() {
this->state_ = speaker::STATE_STOPPING;
this->state_start_ms_ = App.get_loop_component_start_time();
if (this->task_handle_ != nullptr) {
xEventGroupSetBits(this->event_group_, ResamplingEventGroupBits::COMMAND_STOP);
xEventGroupSetBits(this->event_group_, ResamplingEventGroupBits::TASK_COMMAND_STOP);
}
this->output_speaker_->stop();
}
esp_err_t ResamplerSpeaker::delete_task_() {
if (!this->task_created_) {
void ResamplerSpeaker::delete_task_() {
if (this->task_handle_ != nullptr) {
// Delete the suspended task
vTaskDelete(this->task_handle_);
this->task_handle_ = nullptr;
if (this->task_stack_buffer_ != nullptr) {
if (this->task_stack_in_psram_) {
RAMAllocator<StackType_t> stack_allocator(RAMAllocator<StackType_t>::ALLOC_EXTERNAL);
stack_allocator.deallocate(this->task_stack_buffer_, TASK_STACK_SIZE);
} else {
RAMAllocator<StackType_t> stack_allocator(RAMAllocator<StackType_t>::ALLOC_INTERNAL);
stack_allocator.deallocate(this->task_stack_buffer_, TASK_STACK_SIZE);
}
this->task_stack_buffer_ = nullptr;
}
return ESP_OK;
}
return ESP_ERR_INVALID_STATE;
if (this->task_stack_buffer_ != nullptr) {
// Deallocate the task stack buffer
if (this->task_stack_in_psram_) {
RAMAllocator<StackType_t> stack_allocator(RAMAllocator<StackType_t>::ALLOC_EXTERNAL);
stack_allocator.deallocate(this->task_stack_buffer_, TASK_STACK_SIZE);
} else {
RAMAllocator<StackType_t> stack_allocator(RAMAllocator<StackType_t>::ALLOC_INTERNAL);
stack_allocator.deallocate(this->task_stack_buffer_, TASK_STACK_SIZE);
}
this->task_stack_buffer_ = nullptr;
}
}
void ResamplerSpeaker::finish() { this->output_speaker_->finish(); }
void ResamplerSpeaker::finish() { this->send_command_(ResamplingEventGroupBits::COMMAND_FINISH); }
bool ResamplerSpeaker::has_buffered_data() const {
bool has_ring_buffer_data = false;
if (this->requires_resampling_() && (this->ring_buffer_.use_count() > 0)) {
has_ring_buffer_data = (this->ring_buffer_.lock()->available() > 0);
if (this->requires_resampling_()) {
std::shared_ptr<RingBuffer> temp_ring_buffer = this->ring_buffer_.lock();
if (temp_ring_buffer) {
has_ring_buffer_data = (temp_ring_buffer->available() > 0);
}
}
return (has_ring_buffer_data || this->output_speaker_->has_buffered_data());
}
@@ -253,9 +361,8 @@ bool ResamplerSpeaker::requires_resampling_() const {
}
void ResamplerSpeaker::resample_task(void *params) {
ResamplerSpeaker *this_resampler = (ResamplerSpeaker *) params;
ResamplerSpeaker *this_resampler = static_cast<ResamplerSpeaker *>(params);
this_resampler->task_created_ = true;
xEventGroupSetBits(this_resampler->event_group_, ResamplingEventGroupBits::STATE_STARTING);
std::unique_ptr<audio::AudioResampler> resampler =
@@ -269,7 +376,7 @@ void ResamplerSpeaker::resample_task(void *params) {
std::shared_ptr<RingBuffer> temp_ring_buffer =
RingBuffer::create(this_resampler->audio_stream_info_.ms_to_bytes(this_resampler->buffer_duration_ms_));
if (temp_ring_buffer.use_count() == 0) {
if (!temp_ring_buffer) {
err = ESP_ERR_NO_MEM;
} else {
this_resampler->ring_buffer_ = temp_ring_buffer;
@@ -291,7 +398,7 @@ void ResamplerSpeaker::resample_task(void *params) {
while (err == ESP_OK) {
uint32_t event_bits = xEventGroupGetBits(this_resampler->event_group_);
if (event_bits & ResamplingEventGroupBits::COMMAND_STOP) {
if (event_bits & ResamplingEventGroupBits::TASK_COMMAND_STOP) {
break;
}
@@ -310,8 +417,8 @@ void ResamplerSpeaker::resample_task(void *params) {
xEventGroupSetBits(this_resampler->event_group_, ResamplingEventGroupBits::STATE_STOPPING);
resampler.reset();
xEventGroupSetBits(this_resampler->event_group_, ResamplingEventGroupBits::STATE_STOPPED);
this_resampler->task_created_ = false;
vTaskDelete(nullptr);
vTaskSuspend(nullptr); // Suspend this task indefinitely until the loop method deletes it
}
} // namespace resampler

24
esphome/components/resampler/speaker/resampler_speaker.h
View File

@@ -8,14 +8,16 @@
#include "esphome/core/component.h"
#include <freertos/event_groups.h>
#include <freertos/FreeRTOS.h>
#include <freertos/event_groups.h>
namespace esphome {
namespace resampler {
class ResamplerSpeaker : public Component, public speaker::Speaker {
public:
float get_setup_priority() const override { return esphome::setup_priority::DATA; }
void dump_config() override;
void setup() override;
void loop() override;
@@ -65,13 +67,18 @@ class ResamplerSpeaker : public Component, public speaker::Speaker {
/// ESP_ERR_INVALID_STATE if the task wasn't created
esp_err_t start_task_();
/// @brief Stops the output speaker. If the resampling task is running, it sends the stop command.
void stop_();
/// @brief Transitions to STATE_STOPPING, records the stopping timestamp, sends the task stop command if the task is
/// running, and stops the output speaker.
void enter_stopping_state_();
/// @brief Deallocates the task stack and resets the pointers.
/// @return ESP_OK if successful
/// ESP_ERR_INVALID_STATE if the task hasn't stopped itself
esp_err_t delete_task_();
/// @brief Sets the appropriate status error based on the start failure reason.
void set_start_error_(esp_err_t err);
/// @brief Deletes the resampler task if suspended, deallocates the task stack, and resets the related pointers.
void delete_task_();
/// @brief Sends a command via event group bits, enables the loop, and optionally wakes the main loop.
void send_command_(uint32_t command_bit, bool wake_loop = false);
inline bool requires_resampling_() const;
static void resample_task(void *params);
@@ -83,7 +90,7 @@ class ResamplerSpeaker : public Component, public speaker::Speaker {
speaker::Speaker *output_speaker_{nullptr};
bool task_stack_in_psram_{false};
bool task_created_{false};
bool waiting_for_output_{false};
TaskHandle_t task_handle_{nullptr};
StaticTask_t task_stack_;
@@ -98,6 +105,7 @@ class ResamplerSpeaker : public Component, public speaker::Speaker {
uint32_t target_sample_rate_;
uint32_t buffer_duration_ms_;
uint32_t state_start_ms_{0};
uint64_t callback_remainder_{0};
};