chenjintao/lucky-cat
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(voice): add local VAD speech endpoint detection

Browse Source
This commit is contained in:
chenjintao
2026-10-23 10:59:26 +08:00
parent ee32b4d63f
commit f17ce857ff
17 changed files with 1183 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AGENTS.md
+1
View File
@@ -49,6 +49,7 @@ The following matrix is authoritative for all new and refactored cross-layer con
| Battery sensing and battery-state policy (`platform_battery_*`, `domain_battery_*`) | `domain` for SOC/level semantics; `platform` for ADC, charger GPIO, and load-quality primitives | `components/domain/include/domain.h` for semantic state; `components/platform/include/platform.h` for hardware samples | `control_plane` consumes domain status only; platform load tracking stays internal and upper layers must not infer SOC from raw ADC values. |
| LAN HTTP transport contract (bounded request/response structs, handler delegate, server lifecycle/state) | `platform` | `components/platform/include/platform.h` | `domain` owns route semantics and maps them in implementation; upper layers must not expose platform HTTP types. |
| Multimodal session transport contract exposed to orchestration (connected/disconnected/error/text/binary event family, audio uplink callback family) | `domain` | `components/domain/include/domain.h` | `platform` internal transport types stay internal; cross-layer public contract is declared once at `domain` facade. |
| Local speech endpoint detection (`domain_speech_endpoint_*`, VAD state and generation-bound snapshot) | `domain` | `components/domain/include/domain.h` | `platform` owns generic PCM subscriptions only; `control_plane` queries domain snapshots and owns turn termination policy. See ADR 0006. |
| Device-auth persistence port (opaque auth key/state record slots and legacy migration targets) | `domain` | `components/domain/include/domain.h` | `control_plane` owns serialized auth record schemas and consumes typed auth slots through `domain`; `platform` remains an opaque blob KV owner only. |
| User birth profile and daily fortune result (`domain_birth_profile_*`, `domain_fortune_result_*`) | `domain` | `components/domain/include/domain.h` | `control_plane` consumes and persists these contracts through the domain facade; platform storage remains opaque and semantics-free. |
| LAN user configuration web application contract (`domain_user_config_server_*`) | `domain` | `components/domain/include/domain.h` | `control_plane` owns connectivity-driven start/stop policy; page routes and user/fortune JSON remain domain-owned. |
components/control_plane/internal/control_plane_internal.h
+1 -1
View File
@@ -141,7 +141,7 @@ esp_err_t control_plane_multimodal_register_responding_text_callback(
void *user_data);
esp_err_t control_plane_multimodal_start_capture(void);
esp_err_t control_plane_multimodal_start_uplink(void);
void control_plane_multimodal_stop_uplink(void);
esp_err_t control_plane_multimodal_stop_uplink(void);
bool control_plane_multimodal_is_uplink_running(void);
esp_err_t control_plane_multimodal_play_downstream_audio(const uint8_t *data, size_t len);
esp_err_t control_plane_multimodal_handle_ptt_press_step(void);
components/control_plane/src/multimodal/control_plane_multimodal_runtime.c
+4 -1
View File
@@ -682,7 +682,10 @@ esp_err_t control_plane_multimodal_handle_ptt_press_step(void)
esp_err_t control_plane_multimodal_handle_ptt_release(void)
{
control_plane_multimodal_stop_uplink();
esp_err_t err = control_plane_multimodal_stop_uplink();
if (err != ESP_OK) {
return err;
}
return control_plane_multimodal_send_speech_stop();
}
components/control_plane/src/multimodal/control_plane_multimodal_session_service.c
+6 -4
View File
@@ -2356,6 +2356,9 @@ esp_err_t control_plane_multimodal_send_speech_stop(void)
esp_err_t err = ESP_OK;
if (transport_available() && has_active_task) {
err = send_directive("continue-task", "StopSpeech", true, false);
if (err == ESP_OK) {
ESP_LOGI(TAG, "StopSpeech sent after uplink drain");
}
}
if (lock_state()) {
@@ -2678,15 +2681,14 @@ esp_err_t control_plane_multimodal_start_uplink(void)
return domain_runtime_multimodal_audio_start_uplink(control_plane_audio_uplink_send_cb, NULL);
}
void control_plane_multimodal_stop_uplink(void)
esp_err_t control_plane_multimodal_stop_uplink(void)
{
if (!domain_runtime_multimodal_audio_is_uplink_running()) {
return;
}
/* 即便采集已暂停，上次 stop 超时的队列/在途发送仍需要重试排空。 */
esp_err_t err = domain_runtime_multimodal_audio_stop_uplink(CONTROL_PLANE_UPLINK_READY_WAIT_MS);
if (err != ESP_OK) {
ESP_LOGW(TAG, "Mic uplink stop ended with %s", esp_err_to_name(err));
}
return err;
}
bool control_plane_multimodal_is_uplink_running(void)
components/control_plane/src/runtime/control_plane_runtime_workers.c
+55 -1
View File
@@ -244,6 +244,7 @@ typedef struct {
bool pending_press_after_teardown;
bool start_action_sent;
bool speech_action_sent;
bool endpoint_start_attempted;
bool listening_ui_active;
domain_expression_trigger_t listening_ui_trigger;
} ptt_runtime_t;
@@ -754,6 +755,7 @@ static void control_plane_set_ptt_active_with_trigger(
s_ptt_runtime.foreground_attempts = 0;
s_ptt_runtime.start_action_sent = false;
s_ptt_runtime.speech_action_sent = false;
s_ptt_runtime.endpoint_start_attempted = false;
s_ptt_runtime.listening_ui_active = true;
s_ptt_runtime.listening_ui_trigger = listening_ui_trigger;
if (s_ptt_runtime.state == PTT_STATE_TEARDOWN_LOCAL ||
@@ -1792,8 +1794,18 @@ static void control_plane_handle_ptt_teardown(void)
{
ptt_runtime_t ptt = control_plane_ptt_snapshot();
if (ptt.state == PTT_STATE_TEARDOWN_LOCAL) {
control_plane_multimodal_stop_uplink();
esp_err_t endpoint_err = domain_speech_endpoint_stop();
esp_err_t stop_err = endpoint_err == ESP_OK
? control_plane_multimodal_stop_uplink() : endpoint_err;
if (stop_err != ESP_OK) {
/* 有界失败：废弃整轮并关闭连接，不无限重试或宣告输入成功结束。 */
ESP_LOGW(TAG, "Abort voice turn after incomplete drain: %s", esp_err_to_name(stop_err));
control_plane_multimodal_mark_transport_unavailable(
CONTROL_PLANE_TRANSPORT_LOSS_PROTOCOL_ERROR, "uplink_drain_failed");
control_plane_ptt_set_state(PTT_STATE_TEARDOWN_REMOTE, "uplink_aborted");
} else {
control_plane_ptt_set_state(PTT_STATE_TEARDOWN_REMOTE, "uplink_stopped");
}
ptt = control_plane_ptt_snapshot();
}
@@ -1833,6 +1845,7 @@ static void control_plane_handle_ptt_teardown(void)
s_ptt_runtime.press_tick = xTaskGetTickCount();
s_ptt_runtime.start_action_sent = false;
s_ptt_runtime.speech_action_sent = false;
s_ptt_runtime.endpoint_start_attempted = false;
control_plane_ptt_transition_locked(PTT_STATE_WAIT_WIFI, "pending_press");
}
generation = s_ptt_runtime.generation;
@@ -1851,6 +1864,7 @@ static bool control_plane_service_ptt_lifecycle(void)
{
ptt_runtime_t ptt = control_plane_ptt_snapshot();
if (ptt.state == PTT_STATE_IDLE) {
(void)domain_speech_endpoint_stop();
if (ptt.active || ptt.listening_ui_active) {
ESP_LOGW(TAG,
"Recover inconsistent PTT idle state: generation=%u active=%d listening_ui=%d",
@@ -1878,6 +1892,20 @@ static bool control_plane_service_ptt_lifecycle(void)
TickType_t now = xTaskGetTickCount();
bool local_wake_prompt_gate = control_plane_local_wake_prompt_gate_active();
#if CONFIG_TQ_SPEECH_ENDPOINT_ENABLE
if (!local_wake_prompt_gate && ptt.state == PTT_STATE_STREAMING) {
domain_speech_endpoint_snapshot_t endpoint = {0};
if (domain_speech_endpoint_get_snapshot(&endpoint) == ESP_OK &&
endpoint.active && endpoint.ended && endpoint.generation == ptt.generation) {
ESP_LOGI(TAG, "VAD endpoint: generation=%u audio_ms=%u silence_ms=%u max_process_us=%u gaps=%u",
(unsigned)endpoint.generation, (unsigned)endpoint.audio_ms,
(unsigned)endpoint.silence_ms, (unsigned)endpoint.max_process_us,
(unsigned)endpoint.discontinuities);
(void)control_plane_deactivate_ptt(true, ptt.generation, "vad_endpoint");
return true;
}
}
#endif
#if CONFIG_TQ_PTT_MODE_CLICK_TO_TALK
TickType_t capture_timeout_ticks =
pdMS_TO_TICKS(CONFIG_TQ_PTT_CLICK_CAPTURE_TIMEOUT_MS);
@@ -2037,6 +2065,30 @@ static void control_plane_drive_ptt_state_machine(TickType_t *last_ws_start_atte
esp_err_t press_err = control_plane_multimodal_handle_ptt_press_step();
if (press_err == ESP_OK) {
#if CONFIG_TQ_SPEECH_ENDPOINT_ENABLE
/* 开麦成功后才启用，不把提示音、联网等待或本地唤醒采集当成用户输入。 */
bool arm_endpoint = false;
portENTER_CRITICAL(&s_runtime_lock);
if (s_ptt_runtime.active && s_ptt_runtime.generation == ptt.generation &&
!s_ptt_runtime.endpoint_start_attempted &&
s_ptt_runtime.state == PTT_STATE_STREAMING) {
s_ptt_runtime.endpoint_start_attempted = true;
arm_endpoint = true;
}
portEXIT_CRITICAL(&s_runtime_lock);
if (arm_endpoint) {
const domain_speech_endpoint_config_t config = {
.generation = ptt.generation,
.min_speech_ms = CONFIG_TQ_SPEECH_ENDPOINT_MIN_SPEECH_MS,
.end_silence_ms = CONFIG_TQ_SPEECH_ENDPOINT_SILENCE_MS,
};
esp_err_t endpoint_err = domain_speech_endpoint_start(&config);
if (endpoint_err != ESP_OK) {
ESP_LOGW(TAG, "VAD unavailable; retain capture timeout: %s",
esp_err_to_name(endpoint_err));
}
}
#endif
portENTER_CRITICAL(&s_runtime_lock);
if (s_ptt_runtime.state == PTT_STATE_WAIT_START_SENT) {
s_ptt_runtime.start_action_sent = true;
@@ -2345,6 +2397,7 @@ static void control_plane_wifi_state_task(void *arg)
was_connected = connected;
}
(void)domain_speech_endpoint_stop();
control_plane_stop_user_config_server();
control_plane_set_task_handle(&s_wifi_state_task, NULL);
return;
@@ -2852,6 +2905,7 @@ void control_plane_stop_runtime_workers(void)
if (!control_plane_wait_task_exit(&s_wifi_state_task, CONTROL_PLANE_TASK_STOP_WAIT_MS)) {
control_plane_force_delete_task(&s_wifi_state_task, "cp_wifi_state");
}
(void)domain_speech_endpoint_stop();
#if CONFIG_TQ_TASK_MONITOR_ENABLE
if (!control_plane_wait_task_exit(&s_task_monitor_task, CONTROL_PLANE_TASK_STOP_WAIT_MS)) {
control_plane_force_delete_task_with_caps(&s_task_monitor_task, "cp_task_monitor");
components/domain/CMakeLists.txt
+1
View File
@@ -11,6 +11,7 @@ idf_component_register(
"src/runtime/domain_runtime_adapters.c"
"src/runtime/domain_ota.c"
"src/runtime/domain_local_command.c"
"src/runtime/domain_speech_endpoint.c"
"src/runtime/domain_mini_program_provisioning.c"
INCLUDE_DIRS
"include"
components/domain/include/domain.h
+31 -3
View File
@@ -297,16 +297,44 @@ esp_err_t domain_runtime_multimodal_audio_start_uplink(
void *user_data);
/**
* @brief 有界停止多模态上行采集和流发送。 / Stop uplink capture/stream with bounded teardown.
* @brief 暂停采集、补齐尾帧并限时排空发送；常驻任务由 deinit 回收。
*
* `timeout_ms < 0` is invalid, zero requests asynchronous stop, and positive
* values wait up to the bound before lower layers apply deterministic fallback.
* 负超时无效；零尝试立即完成，否则 ESP_ERR_TIMEOUT；正数为完整调用预算。
* 成功后没有在途采集/发送；超时保留资源供重试，不后台清理，不应发送 StopSpeech。
*/
esp_err_t domain_runtime_multimodal_audio_stop_uplink(int timeout_ms);
bool domain_runtime_multimodal_audio_is_uplink_running(void);
bool domain_runtime_multimodal_audio_is_initialized(void);
/** 保留麦克风采集，仅控制扬声器输出。 / Gate speaker output while preserving capture. */
esp_err_t domain_runtime_multimodal_audio_set_speaker_output_enabled(bool enabled);
typedef struct {
uint32_t generation;
uint32_t min_speech_ms;
uint32_t end_silence_ms;
} domain_speech_endpoint_config_t;
typedef struct {
uint32_t generation;
bool active;
bool speech_confirmed;
bool ended;
uint32_t audio_ms;
uint32_t silence_ms;
uint32_t max_process_us;
uint32_t discontinuities;
} domain_speech_endpoint_snapshot_t;
/**
* 本地端点检测，无新 worker。start 对相同配置幂等，不同配置须先 stop。
* start/stop/get_snapshot 最多等锁 100 ms；stop 另以同一总预算等待 PCM 回调。
* stop 超时保持资源可重试，绝不后台释放；成功移除订阅并销毁 VAD。
* PCM 回调只用零等待锁及固定 20 ms 分帧，不执行协议或阻塞式收尾。
* 配置关闭时 start 返回 ESP_ERR_NOT_SUPPORTED，stop 幂等，快照为 inactive。
*/
esp_err_t domain_speech_endpoint_start(const domain_speech_endpoint_config_t *config);
esp_err_t domain_speech_endpoint_stop(void);
esp_err_t domain_speech_endpoint_get_snapshot(domain_speech_endpoint_snapshot_t *out);
/**
* @brief 启动“喵掌柜”本地 MultiNet 检测。 / Start local MultiNet detection for "喵掌柜".
*
components/domain/src/runtime/domain_speech_endpoint.c
+241
View File
@@ -0,0 +1,241 @@
#include "domain.h"
#include <string.h>
#include "sdkconfig.h"
#if CONFIG_TQ_SPEECH_ENDPOINT_ENABLE
#include "esp_heap_caps.h"
#include "esp_log.h"
#include "esp_timer.h"
#include "esp_vad.h"
#include "freertos/FreeRTOS.h"
#include "freertos/semphr.h"
#include "freertos/task.h"
#include "platform.h"
#define ENDPOINT_LOCK_MS 100U
#define ENDPOINT_RATE_HZ 16000U
#define ENDPOINT_FRAME_MS 20U
#define ENDPOINT_FRAME_SAMPLES 320U
#define ENDPOINT_MAX_INPUT_SAMPLES 512U
#define ENDPOINT_GAP_US 100000LL
static const char *TAG = "domain_endpoint";
static portMUX_TYPE s_init_lock = portMUX_INITIALIZER_UNLOCKED;
static StaticSemaphore_t s_mutex_storage;
static SemaphoreHandle_t s_mutex;
static bool s_initializing;
static bool s_gap_pending;
static vad_handle_t s_vad;
static int16_t s_pcm[ENDPOINT_FRAME_SAMPLES];
static size_t s_fill;
static uint32_t s_speech_run_ms;
static int64_t s_last_pcm_us;
static domain_speech_endpoint_config_t s_config;
static domain_speech_endpoint_snapshot_t s_snapshot;
static bool endpoint_lock(void)
{
portENTER_CRITICAL(&s_init_lock);
SemaphoreHandle_t mutex = s_mutex;
bool create = mutex == NULL && !s_initializing;
if (create) {
s_initializing = true;
}
portEXIT_CRITICAL(&s_init_lock);
if (create) {
mutex = xSemaphoreCreateMutexStatic(&s_mutex_storage);
portENTER_CRITICAL(&s_init_lock);
s_mutex = mutex;
s_initializing = false;
portEXIT_CRITICAL(&s_init_lock);
}
return mutex != NULL &&
xSemaphoreTake(mutex, pdMS_TO_TICKS(ENDPOINT_LOCK_MS)) == pdTRUE;
}
static void endpoint_reset_continuity(void)
{
s_fill = 0U;
s_speech_run_ms = 0U;
s_snapshot.silence_ms = 0U;
++s_snapshot.discontinuities;
}
static void endpoint_pcm_observer(const int16_t *pcm, size_t samples,
uint32_t sample_rate_hz, void *user_data)
{
(void)user_data;
/* 注册发生在锁创建后。回调绝不等待控制线程，也不调用停止/发送接口。 */
if (xSemaphoreTake(s_mutex, 0U) != pdTRUE) {
portENTER_CRITICAL(&s_init_lock);
s_gap_pending = true;
portEXIT_CRITICAL(&s_init_lock);
return;
}
if (!s_snapshot.active || s_snapshot.ended) {
xSemaphoreGive(s_mutex);
return;
}
const int64_t started_us = esp_timer_get_time();
portENTER_CRITICAL(&s_init_lock);
bool gap = s_gap_pending;
s_gap_pending = false;
portEXIT_CRITICAL(&s_init_lock);
if (gap || (s_last_pcm_us != 0 && started_us - s_last_pcm_us > ENDPOINT_GAP_US)) {
endpoint_reset_continuity();
}
s_last_pcm_us = started_us;
if (pcm == NULL || samples == 0U || samples > ENDPOINT_MAX_INPUT_SAMPLES ||
sample_rate_hz != ENDPOINT_RATE_HZ) {
endpoint_reset_continuity();
xSemaphoreGive(s_mutex);
return;
}
while (samples > 0U && !s_snapshot.ended) {
size_t count = ENDPOINT_FRAME_SAMPLES - s_fill;
if (count > samples) {
count = samples;
}
memcpy(s_pcm + s_fill, pcm, count * sizeof(int16_t));
s_fill += count;
pcm += count;
samples -= count;
if (s_fill != ENDPOINT_FRAME_SAMPLES) {
continue;
}
s_fill = 0U;
vad_state_t activity = vad_process(s_vad, s_pcm,
ENDPOINT_RATE_HZ, ENDPOINT_FRAME_MS);
s_snapshot.audio_ms += ENDPOINT_FRAME_MS;
if (activity == VAD_SPEECH) {
s_snapshot.silence_ms = 0U;
if (!s_snapshot.speech_confirmed) {
s_speech_run_ms += ENDPOINT_FRAME_MS;
s_snapshot.speech_confirmed = s_speech_run_ms >= s_config.min_speech_ms;
}
} else {
s_speech_run_ms = 0U;
if (s_snapshot.speech_confirmed) {
s_snapshot.silence_ms += ENDPOINT_FRAME_MS;
s_snapshot.ended = s_snapshot.silence_ms >= s_config.end_silence_ms;
}
}
}
uint32_t elapsed_us = (uint32_t)(esp_timer_get_time() - started_us);
if (elapsed_us > s_snapshot.max_process_us) {
s_snapshot.max_process_us = elapsed_us;
}
xSemaphoreGive(s_mutex);
}
esp_err_t domain_speech_endpoint_start(const domain_speech_endpoint_config_t *config)
{
if (config == NULL || config->generation == 0U ||
config->min_speech_ms < 100U || config->min_speech_ms > 1000U ||
config->end_silence_ms < 400U || config->end_silence_ms > 2000U) {
return ESP_ERR_INVALID_ARG;
}
if (!endpoint_lock()) {
return ESP_ERR_TIMEOUT;
}
if (s_vad != NULL) {
bool same = s_snapshot.active && s_config.generation == config->generation &&
s_config.min_speech_ms == config->min_speech_ms &&
s_config.end_silence_ms == config->end_silence_ms;
xSemaphoreGive(s_mutex);
return same ? ESP_OK : ESP_ERR_INVALID_STATE;
}
s_vad = vad_create(VAD_MODE_1);
if (s_vad == NULL) {
xSemaphoreGive(s_mutex);
return ESP_ERR_NO_MEM;
}
s_config = *config;
s_snapshot = (domain_speech_endpoint_snapshot_t) {
.generation = config->generation,
.active = true,
};
s_fill = 0U;
s_speech_run_ms = 0U;
s_last_pcm_us = 0;
portENTER_CRITICAL(&s_init_lock);
s_gap_pending = false;
portEXIT_CRITICAL(&s_init_lock);
esp_err_t err = platform_multimodal_audio_register_pcm_observer(endpoint_pcm_observer, NULL);
if (err != ESP_OK) {
s_snapshot.active = false;
vad_destroy(s_vad);
s_vad = NULL;
}
xSemaphoreGive(s_mutex);
if (err == ESP_OK) {
ESP_LOGI(TAG, "VAD armed: generation=%u speech_ms=%u silence_ms=%u internal_free=%u psram_free=%u",
(unsigned)config->generation, (unsigned)config->min_speech_ms,
(unsigned)config->end_silence_ms,
(unsigned)heap_caps_get_free_size(MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT),
(unsigned)heap_caps_get_free_size(MALLOC_CAP_SPIRAM));
}
return err;
}
esp_err_t domain_speech_endpoint_stop(void)
{
const TickType_t started = xTaskGetTickCount();
if (!endpoint_lock()) {
return ESP_ERR_TIMEOUT;
}
s_snapshot.active = false;
if (s_vad == NULL) {
xSemaphoreGive(s_mutex);
return ESP_OK;
}
uint32_t elapsed_ms = (uint32_t)pdTICKS_TO_MS(xTaskGetTickCount() - started);
int remaining_ms = elapsed_ms < ENDPOINT_LOCK_MS ? (int)(ENDPOINT_LOCK_MS - elapsed_ms) : 0;
/* 回调仅 try-lock，所以持有状态锁等待注销不会形成互等。 */
esp_err_t err = platform_multimodal_audio_unregister_pcm_observer(
endpoint_pcm_observer, NULL, remaining_ms);
if (err == ESP_OK) {
vad_destroy(s_vad);
s_vad = NULL;
s_fill = 0U;
}
xSemaphoreGive(s_mutex);
return err;
}
esp_err_t domain_speech_endpoint_get_snapshot(domain_speech_endpoint_snapshot_t *out)
{
if (out == NULL) {
return ESP_ERR_INVALID_ARG;
}
if (!endpoint_lock()) {
return ESP_ERR_TIMEOUT;
}
*out = s_snapshot;
xSemaphoreGive(s_mutex);
return ESP_OK;
}
#else
esp_err_t domain_speech_endpoint_start(const domain_speech_endpoint_config_t *config)
{
(void)config;
return ESP_ERR_NOT_SUPPORTED;
}
esp_err_t domain_speech_endpoint_stop(void)
{
return ESP_OK;
}
esp_err_t domain_speech_endpoint_get_snapshot(domain_speech_endpoint_snapshot_t *out)
{
if (out == NULL) {
return ESP_ERR_INVALID_ARG;
}
memset(out, 0, sizeof(*out));
return ESP_OK;
}
#endif
components/platform/include/platform.h
+11 -10
View File
@@ -548,15 +548,13 @@ esp_err_t platform_multimodal_audio_start_uplink(
void *user_data);
/**
* @brief 以有界方式停止多模态上行采集与流式发送任务。
* Stop multimodal uplink capture/stream worker with bounded teardown.
* @brief 暂停上行、补齐尾帧并限时排空发送；常驻 worker 留待 deinit 回收。
* Pause capture and drain the padded final frame; deinit owns workers.
*
* 负超时无效；零只请求协作停止；正超时结束时强制清理所属任务；重复调用幂等。
* Semantics:
* - `timeout_ms < 0`: invalid argument.
* - `timeout_ms == 0`: request cooperative stop and return immediately.
* - `timeout_ms > 0`: wait up to the bound, then force-delete owned workers.
* - Repeated calls are idempotent.
* 负超时无效；零只尝试立即完成，未完成返回 ESP_ERR_TIMEOUT。
* 正数预算包含等待当前采集迭代和尾包发送；成功后没有在途采集/发送。
* 超时保留队列、回调和缓冲以便重试，不后台清理；不可据此发送输入结束指令。
* 重复调用幂等；不可在 PCM observer / uplink sender 回调中调用。
*/
esp_err_t platform_multimodal_audio_stop_uplink(int timeout_ms);
bool platform_multimodal_audio_is_uplink_running(void);
@@ -571,12 +569,14 @@ bool platform_multimodal_audio_is_initialized(void);
*/
esp_err_t platform_multimodal_audio_set_speaker_output_enabled(bool enabled);
/**
* @brief 为当前采集路径注册唯一的单声道 PCM 观察者。
* Register the single mono PCM observer for the active capture path.
* @brief 为当前采集路径注册单声道 PCM 观察者（固定两个独立槽）。
* Register a mono PCM observer in one of two fixed slots.
*
* 回调在 platform 采集任务中同步执行，但不持有注册锁；回调必须有界且不可阻塞。
* The callback runs synchronously in the platform capture task, outside the
* registration lock, and must remain bounded and non-blocking.
* 相同 cb/user_data 重复注册幂等；槽位满返回 ESP_ERR_NO_MEM。
* 不得在观察者回调中注册/注销或调用音频 start/stop。
*/
esp_err_t platform_multimodal_audio_register_pcm_observer(
platform_multimodal_audio_pcm_observer_t observer,
@@ -587,6 +587,7 @@ esp_err_t platform_multimodal_audio_register_pcm_observer(
*
* 负超时无效，零只执行非阻塞的排空检查。
* Negative timeout is invalid. Zero performs a non-blocking drain check.
* 超时后槽位保持禁用且保留在途计数；必须重试成功后才能释放 user_data。
*/
esp_err_t platform_multimodal_audio_unregister_pcm_observer(
platform_multimodal_audio_pcm_observer_t observer,
components/platform/src/multimodal/audio_pipeline.c
+19 -4
View File
@@ -388,7 +388,10 @@ typedef struct {
TaskHandle_t uplink_send_task;
volatile bool uplink_recording_active;
volatile bool uplink_task_exit_requested;
volatile bool uplink_send_in_flight;
bool uplink_send_in_flight;
bool uplink_tail_flush_requested;
bool uplink_stop_incomplete;
esp_err_t uplink_tail_flush_error;
bool uplink_first_network_pcm_pending;
bool uplink_first_network_packet_pending;
TickType_t uplink_network_start_tick;
@@ -412,9 +415,21 @@ static bool s_downstream_last_terminal_valid;
static uint32_t s_downstream_last_terminal_generation;
static multimodal_audio_downstream_playback_event_type_t
s_downstream_last_terminal_type;
static multimodal_audio_pcm_observer_t s_pcm_observer;
static void *s_pcm_observer_ctx;
static uint32_t s_pcm_observer_in_flight;
#define MULTIMODAL_AUDIO_PCM_OBSERVER_SLOTS 2U
typedef struct {
multimodal_audio_pcm_observer_t callback;
void *ctx;
uint32_t in_flight;
uint32_t registration;
bool enabled;
} audio_pcm_observer_slot_t;
static audio_pcm_observer_slot_t s_pcm_observers[MULTIMODAL_AUDIO_PCM_OBSERVER_SLOTS];
static uint32_t s_pcm_observer_registration;
static StaticSemaphore_t s_uplink_capture_mutex_storage;
static SemaphoreHandle_t s_uplink_capture_mutex;
static StaticSemaphore_t s_uplink_lifecycle_mutex_storage;
static SemaphoreHandle_t s_uplink_lifecycle_mutex;
static bool s_uplink_capture_mutex_initializing;
static portMUX_TYPE s_pcm_observer_lock = portMUX_INITIALIZER_UNLOCKED;
static bool s_uplink_stack_warned;
static bool s_uplink_send_stack_warned;
components/platform/src/multimodal/audio_pipeline/audio_pipeline_runtime_impl.c
+255 -73
View File
@@ -206,31 +206,73 @@ static bool audio_consume_first_network_packet(multimodal_audio_ctx_t *ctx,
static void notify_pcm_observer(const int16_t *pcm, size_t samples)
{
/* in_flight 计数让 unregister 能等待正在执行的观察者，同时回调在锁外运行。 */
multimodal_audio_pcm_observer_t observer = NULL;
void *observer_ctx = NULL;
/* 固定两个订阅，无分配；先计入在途，再在锁外回调，支持独立有界注销。 */
audio_pcm_observer_slot_t observers[MULTIMODAL_AUDIO_PCM_OBSERVER_SLOTS] = {0};
portENTER_CRITICAL(&s_pcm_observer_lock);
observer = s_pcm_observer;
observer_ctx = s_pcm_observer_ctx;
if (observer != NULL) {
++s_pcm_observer_in_flight;
for (size_t i = 0; i < MULTIMODAL_AUDIO_PCM_OBSERVER_SLOTS; ++i) {
if (s_pcm_observers[i].enabled) {
observers[i] = s_pcm_observers[i];
++s_pcm_observers[i].in_flight;
}
}
portEXIT_CRITICAL(&s_pcm_observer_lock);
if (observer == NULL) {
return;
for (size_t i = 0; i < MULTIMODAL_AUDIO_PCM_OBSERVER_SLOTS; ++i) {
if (observers[i].callback == NULL) {
continue;
}
observer(pcm,
samples,
observers[i].callback(pcm, samples,
(uint32_t)CONFIG_MULTIMODAL_AUDIO_CODEC_SAMPLE_RATE,
observer_ctx);
observers[i].ctx);
portENTER_CRITICAL(&s_pcm_observer_lock);
if (s_pcm_observer_in_flight > 0U) {
--s_pcm_observer_in_flight;
}
--s_pcm_observers[i].in_flight;
portEXIT_CRITICAL(&s_pcm_observer_lock);
}
}
static bool ensure_uplink_mutexes(void)
{
portENTER_CRITICAL(&s_audio_runtime_lock);
SemaphoreHandle_t mutex = s_uplink_capture_mutex;
bool create = mutex == NULL && !s_uplink_capture_mutex_initializing;
if (create) {
s_uplink_capture_mutex_initializing = true;
}
portEXIT_CRITICAL(&s_audio_runtime_lock);
if (create) {
mutex = xSemaphoreCreateMutexStatic(&s_uplink_capture_mutex_storage);
SemaphoreHandle_t lifecycle = xSemaphoreCreateMutexStatic(&s_uplink_lifecycle_mutex_storage);
portENTER_CRITICAL(&s_audio_runtime_lock);
s_uplink_capture_mutex = mutex;
s_uplink_lifecycle_mutex = lifecycle;
s_uplink_capture_mutex_initializing = false;
portEXIT_CRITICAL(&s_audio_runtime_lock);
}
return mutex != NULL;
}
static bool lock_uplink_capture(TickType_t ticks)
{
return ensure_uplink_mutexes() &&
xSemaphoreTake(s_uplink_capture_mutex, ticks) == pdTRUE;
}
static void audio_set_uplink_send_in_flight(multimodal_audio_ctx_t *ctx, bool active)
{
portENTER_CRITICAL(&s_audio_runtime_lock);
ctx->uplink_send_in_flight = active;
portEXIT_CRITICAL(&s_audio_runtime_lock);
}
static bool audio_uplink_send_pending(multimodal_audio_ctx_t *ctx)
{
/* 必须先读队列，再读 busy；发送者先置 busy 再出队，不能漏掉两者交界。 */
if (ctx->uplink_send_queue != NULL && uxQueueMessagesWaiting(ctx->uplink_send_queue) > 0U) {
return true;
}
portENTER_CRITICAL(&s_audio_runtime_lock);
bool in_flight = ctx->uplink_send_in_flight;
portEXIT_CRITICAL(&s_audio_runtime_lock);
return in_flight;
}
static void audio_snapshot_uplink_tasks(const multimodal_audio_ctx_t *ctx,
@@ -950,6 +992,11 @@ static esp_err_t encode_and_queue_mono_pcm(multimodal_audio_ctx_t *ctx,
}
track_uplink_packet_queued(packet.len);
TaskHandle_t send_task = NULL;
audio_snapshot_uplink_tasks(ctx, NULL, &send_task);
if (send_task != NULL) {
xTaskNotifyGive(send_task);
}
return ESP_OK;
}
@@ -1013,11 +1060,15 @@ static void uplink_send_task(void *arg)
continue;
}
audio_set_uplink_send_in_flight(ctx, true);
uplink_packet_t packet = {0};
if (xQueueReceive(ctx->uplink_send_queue, &packet, pdMS_TO_TICKS(100)) != pdTRUE) {
if (xQueueReceive(ctx->uplink_send_queue, &packet, 0U) != pdTRUE) {
audio_set_uplink_send_in_flight(ctx, false);
ulTaskNotifyTake(pdTRUE, pdMS_TO_TICKS(20));
continue;
}
if (packet.len == 0U) {
audio_set_uplink_send_in_flight(ctx, false);
continue;
}
@@ -1025,12 +1076,12 @@ static void uplink_send_task(void *arg)
void *cb_ctx = NULL;
audio_snapshot_uplink_callback(ctx, &cb, &cb_ctx);
if (cb == NULL) {
audio_set_uplink_send_in_flight(ctx, false);
continue;
}
ctx->uplink_send_in_flight = true;
esp_err_t send_err = cb(packet.data, packet.len, cb_ctx);
ctx->uplink_send_in_flight = false;
audio_set_uplink_send_in_flight(ctx, false);
uint32_t first_packet_elapsed_ms = 0U;
if (send_err == ESP_OK &&
audio_consume_first_network_packet(ctx, &first_packet_elapsed_ms)) {
@@ -1055,6 +1106,38 @@ static void uplink_send_task(void *arg)
delete_audio_task_with_caps(NULL);
}
static bool audio_uplink_tail_pending(multimodal_audio_ctx_t *ctx, esp_err_t *out_error)
{
portENTER_CRITICAL(&s_audio_runtime_lock);
bool pending = ctx->uplink_tail_flush_requested;
if (out_error != NULL) {
*out_error = ctx->uplink_tail_flush_error;
}
portEXIT_CRITICAL(&s_audio_runtime_lock);
return pending;
}
static void audio_flush_uplink_tail(multimodal_audio_ctx_t *ctx)
{
if (!audio_uplink_tail_pending(ctx, NULL) || !lock_uplink_capture(pdMS_TO_TICKS(20))) {
return;
}
esp_err_t err = ESP_OK;
if (ctx->uplink_mono_fill_bytes > 0U) {
memset((uint8_t *)ctx->uplink_mono_buf + ctx->uplink_mono_fill_bytes, 0,
ctx->uplink_mono_bytes - ctx->uplink_mono_fill_bytes);
err = encode_and_queue_mono_pcm(ctx, ctx->uplink_mono_buf, ctx->uplink_mono_bytes);
if (err == ESP_OK) {
ctx->uplink_mono_fill_bytes = 0U;
}
}
portENTER_CRITICAL(&s_audio_runtime_lock);
ctx->uplink_tail_flush_error = err;
ctx->uplink_tail_flush_requested = false;
portEXIT_CRITICAL(&s_audio_runtime_lock);
xSemaphoreGive(s_uplink_capture_mutex);
}
static void uplink_capture_task(void *arg)
{
/* 采集循环先完成声道映射/AEC，再通知 PCM 观察者并编码网络音频。 */
@@ -1091,6 +1174,8 @@ static void uplink_capture_task(void *arg)
while (!audio_get_uplink_task_exit_requested(ctx)) {
if (!audio_get_uplink_recording_active(ctx)) {
/* Opus 编码必须留在原 32 KB 采集栈，不能挪到控制线程的小栈。 */
audio_flush_uplink_tail(ctx);
ulTaskNotifyTake(pdTRUE, pdMS_TO_TICKS(100));
continue;
}
@@ -1100,9 +1185,17 @@ static void uplink_capture_task(void *arg)
continue;
}
if (!lock_uplink_capture(pdMS_TO_TICKS(20))) {
continue;
}
if (!audio_get_uplink_recording_active(ctx)) {
xSemaphoreGive(s_uplink_capture_mutex);
continue;
}
int ret = esp_codec_dev_read(ctx->codec_dev, ctx->uplink_codec_buf, (int)ctx->uplink_codec_read_bytes);
if (ret != ESP_CODEC_DEV_OK) {
ESP_LOGW(TAG, "Mic read failed: 0x%x", ret);
xSemaphoreGive(s_uplink_capture_mutex);
vTaskDelay(pdMS_TO_TICKS(20));
continue;
}
@@ -1117,6 +1210,7 @@ static void uplink_capture_task(void *arg)
(unsigned)clean_bytes,
(unsigned)ctx->uplink_aec_frame_bytes);
ctx->uplink_mono_fill_bytes = 0U;
xSemaphoreGive(s_uplink_capture_mutex);
continue;
}
size_t uplink_frames = ctx->uplink_aec_frame_samples;
@@ -1145,6 +1239,7 @@ static void uplink_capture_task(void *arg)
uplink_frames,
ctx->uplink_aec_clean_buf,
clean_bytes);
xSemaphoreGive(s_uplink_capture_mutex);
if (err != ESP_OK && err != ESP_ERR_INVALID_SIZE && err != ESP_ERR_TIMEOUT) {
ESP_LOGW(TAG, "Encode uplink PCM failed: %s", esp_err_to_name(err));
vTaskDelay(pdMS_TO_TICKS(10));
@@ -1368,6 +1463,11 @@ void multimodal_audio_deinit(void)
audio_set_downstream_playback_stopping(&s_audio, true);
audio_debug_dump_deinit();
(void)multimodal_audio_stop_uplink(2000);
if (s_uplink_lifecycle_mutex == NULL ||
xSemaphoreTake(s_uplink_lifecycle_mutex, pdMS_TO_TICKS(1000)) != pdTRUE) {
ESP_LOGE(TAG, "Uplink lifecycle busy during deinit; retain resources for retry");
return;
}
audio_set_uplink_task_exit_requested(&s_audio, true);
TaskHandle_t notify_capture_task = NULL;
audio_snapshot_uplink_tasks(&s_audio, &notify_capture_task, NULL);
@@ -1401,6 +1501,13 @@ void multimodal_audio_deinit(void)
cleanup_codec_path(&s_audio);
deinit_buses(&s_audio);
memset(&s_audio, 0, sizeof(s_audio));
/* 两个 worker 已退出；force-delete 可能遗留被持有的 capture mutex。 */
vSemaphoreDelete(s_uplink_capture_mutex);
SemaphoreHandle_t capture_mutex = xSemaphoreCreateMutexStatic(&s_uplink_capture_mutex_storage);
portENTER_CRITICAL(&s_audio_runtime_lock);
s_uplink_capture_mutex = capture_mutex;
portEXIT_CRITICAL(&s_audio_runtime_lock);
xSemaphoreGive(s_uplink_lifecycle_mutex);
}
bool multimodal_audio_is_initialized(void)
@@ -1457,15 +1564,29 @@ esp_err_t multimodal_audio_register_pcm_observer(
if (observer == NULL) {
return ESP_ERR_INVALID_ARG;
}
esp_err_t result = ESP_OK;
esp_err_t result = ESP_ERR_NO_MEM;
portENTER_CRITICAL(&s_pcm_observer_lock);
if (s_pcm_observer != NULL &&
(s_pcm_observer != observer || s_pcm_observer_ctx != observer_ctx)) {
result = ESP_ERR_INVALID_STATE;
} else {
s_pcm_observer = observer;
s_pcm_observer_ctx = observer_ctx;
size_t free_slot = MULTIMODAL_AUDIO_PCM_OBSERVER_SLOTS;
for (size_t i = 0; i < MULTIMODAL_AUDIO_PCM_OBSERVER_SLOTS; ++i) {
audio_pcm_observer_slot_t *slot = &s_pcm_observers[i];
if (slot->callback == observer && slot->ctx == observer_ctx) {
result = slot->enabled ? ESP_OK : ESP_ERR_INVALID_STATE;
portEXIT_CRITICAL(&s_pcm_observer_lock);
return result;
}
if (slot->callback == NULL && free_slot == MULTIMODAL_AUDIO_PCM_OBSERVER_SLOTS) {
free_slot = i;
}
}
if (free_slot < MULTIMODAL_AUDIO_PCM_OBSERVER_SLOTS) {
if (++s_pcm_observer_registration == 0U) {
++s_pcm_observer_registration;
}
s_pcm_observers[free_slot] = (audio_pcm_observer_slot_t) {
.callback = observer, .ctx = observer_ctx, .enabled = true,
.registration = s_pcm_observer_registration,
};
result = ESP_OK;
}
portEXIT_CRITICAL(&s_pcm_observer_lock);
return result;
@@ -1479,42 +1600,44 @@ esp_err_t multimodal_audio_unregister_pcm_observer(
if (observer == NULL || timeout_ms < 0) {
return ESP_ERR_INVALID_ARG;
}
portENTER_CRITICAL(&s_pcm_observer_lock);
if (s_pcm_observer == NULL) {
portEXIT_CRITICAL(&s_pcm_observer_lock);
return ESP_OK;
}
if (s_pcm_observer != observer || s_pcm_observer_ctx != observer_ctx) {
portEXIT_CRITICAL(&s_pcm_observer_lock);
return ESP_ERR_INVALID_STATE;
}
s_pcm_observer = NULL;
s_pcm_observer_ctx = NULL;
portEXIT_CRITICAL(&s_pcm_observer_lock);
TickType_t started = xTaskGetTickCount();
TickType_t timeout_ticks = pdMS_TO_TICKS((uint32_t)timeout_ms);
if (timeout_ms > 0 && timeout_ticks == 0U) {
timeout_ticks = 1U;
}
uint32_t registration = 0U;
for (;;) {
uint32_t in_flight = 0U;
bool pending = false;
portENTER_CRITICAL(&s_pcm_observer_lock);
in_flight = s_pcm_observer_in_flight;
for (size_t i = 0; i < MULTIMODAL_AUDIO_PCM_OBSERVER_SLOTS; ++i) {
audio_pcm_observer_slot_t *slot = &s_pcm_observers[i];
if (slot->callback != observer || slot->ctx != observer_ctx) {
continue;
}
if (registration != 0U && slot->registration != registration) {
break; /* 另一注销者已完成；不能移除其后注册的新订阅。 */
}
registration = slot->registration;
slot->enabled = false;
pending = slot->in_flight != 0U;
if (!pending) {
memset(slot, 0, sizeof(*slot));
}
break;
}
portEXIT_CRITICAL(&s_pcm_observer_lock);
if (in_flight == 0U) {
if (!pending) {
return ESP_OK;
}
if (timeout_ms == 0 ||
(xTaskGetTickCount() - started) >= timeout_ticks) {
if (timeout_ms == 0 || (xTaskGetTickCount() - started) >= timeout_ticks) {
/* 保留禁用的槽位与在途计数，重试前不能复用或释放 user_data。 */
return ESP_ERR_TIMEOUT;
}
vTaskDelay(1U);
}
}
esp_err_t multimodal_audio_start_uplink(multimodal_audio_uplink_cb_t cb, void *cb_ctx)
static esp_err_t audio_start_uplink_locked(multimodal_audio_uplink_cb_t cb, void *cb_ctx)
{
/* 回调在任务启动前发布；任一启动失败都会撤销回调，避免半启动状态。 */
if (!s_audio.initialized || !s_audio.codec_dev || !s_audio.opus_enc ||
@@ -1559,42 +1682,101 @@ esp_err_t multimodal_audio_start_uplink(multimodal_audio_uplink_cb_t cb, void *c
return ESP_OK;
}
esp_err_t multimodal_audio_start_uplink(multimodal_audio_uplink_cb_t cb, void *cb_ctx)
{
/* 初始化两个 mutex；锁序始终 lifecycle -> capture，采集任务只拿 capture。 */
if (!ensure_uplink_mutexes()) {
return ESP_ERR_NO_MEM;
}
if (xSemaphoreTake(s_uplink_lifecycle_mutex, pdMS_TO_TICKS(100)) != pdTRUE) {
return ESP_ERR_TIMEOUT;
}
if (!lock_uplink_capture(pdMS_TO_TICKS(100))) {
xSemaphoreGive(s_uplink_lifecycle_mutex);
return ESP_ERR_TIMEOUT;
}
bool unfinished = audio_get_deinit_in_progress(&s_audio) || s_audio.uplink_stop_incomplete ||
audio_uplink_tail_pending(&s_audio, NULL) ||
(!audio_get_uplink_recording_active(&s_audio) && audio_uplink_send_pending(&s_audio));
esp_err_t err = unfinished ? ESP_ERR_INVALID_STATE : audio_start_uplink_locked(cb, cb_ctx);
xSemaphoreGive(s_uplink_capture_mutex);
xSemaphoreGive(s_uplink_lifecycle_mutex);
return err;
}
esp_err_t multimodal_audio_stop_uplink(int timeout_ms)
{
/* 先协作请求退出并等待两个任务；超时后保留明确错误而不伪装成已停止。 */
if (timeout_ms < 0) {
return ESP_ERR_INVALID_ARG;
}
TickType_t started = xTaskGetTickCount();
TickType_t budget = pdMS_TO_TICKS((uint32_t)timeout_ms);
if (timeout_ms > 0 && budget == 0U) {
budget = 1U;
}
/* 未初始化时先安全建立锁；此处不操作任何 PCM。 */
if (!ensure_uplink_mutexes()) {
return ESP_ERR_NO_MEM;
}
TickType_t elapsed = xTaskGetTickCount() - started;
if (xSemaphoreTake(s_uplink_lifecycle_mutex, elapsed < budget ? budget - elapsed : 0U) != pdTRUE) {
return ESP_ERR_TIMEOUT;
}
s_audio.uplink_stop_incomplete = true;
audio_set_uplink_recording_active(&s_audio, false);
elapsed = xTaskGetTickCount() - started;
if (!lock_uplink_capture(elapsed < budget ? budget - elapsed : 0U)) {
xSemaphoreGive(s_uplink_lifecycle_mutex);
return ESP_ERR_TIMEOUT;
}
multimodal_audio_uplink_cb_t cb = NULL;
audio_snapshot_uplink_callback(&s_audio, &cb, NULL);
esp_err_t err = ESP_OK;
if (cb != NULL && s_audio.uplink_mono_fill_bytes > 0U &&
!audio_uplink_tail_pending(&s_audio, NULL)) {
if (timeout_ms == 0) {
err = ESP_ERR_TIMEOUT;
} else {
portENTER_CRITICAL(&s_audio_runtime_lock);
s_audio.uplink_tail_flush_error = ESP_OK;
s_audio.uplink_tail_flush_requested = true;
portEXIT_CRITICAL(&s_audio_runtime_lock);
}
}
xSemaphoreGive(s_uplink_capture_mutex);
if (audio_uplink_tail_pending(&s_audio, NULL)) {
TaskHandle_t capture_task = NULL;
audio_snapshot_uplink_tasks(&s_audio, &capture_task, NULL);
if (capture_task != NULL) {
xTaskNotifyGive(capture_task);
}
}
while (err == ESP_OK) {
bool tail_pending = audio_uplink_tail_pending(&s_audio, &err);
if (err != ESP_OK || (!tail_pending && !audio_uplink_send_pending(&s_audio))) {
break;
}
if ((xTaskGetTickCount() - started) >= budget) {
err = ESP_ERR_TIMEOUT;
break;
}
vTaskDelay(1U);
}
if (err == ESP_OK) {
s_audio.uplink_stop_incomplete = false;
audio_set_uplink_callback(&s_audio, NULL, NULL);
/* 非网络采集的残帧也已被 capture 锁确认静止。 */
s_audio.uplink_mono_fill_bytes = 0U;
portENTER_CRITICAL(&s_audio_runtime_lock);
s_audio.uplink_first_network_pcm_pending = false;
s_audio.uplink_first_network_packet_pending = false;
portEXIT_CRITICAL(&s_audio_runtime_lock);
TickType_t deadline = xTaskGetTickCount() + pdMS_TO_TICKS((uint32_t)timeout_ms);
while (((s_audio.uplink_send_queue != NULL &&
uxQueueMessagesWaiting(s_audio.uplink_send_queue) > 0U) ||
s_audio.uplink_send_in_flight) &&
xTaskGetTickCount() < deadline) {
vTaskDelay(pdMS_TO_TICKS(20));
}
if ((s_audio.uplink_send_queue != NULL &&
uxQueueMessagesWaiting(s_audio.uplink_send_queue) > 0U) ||
s_audio.uplink_send_in_flight) {
ESP_LOGW(TAG,
"Mic uplink stop timed out with %u queued packet(s), in_flight=%d",
(unsigned)((s_audio.uplink_send_queue != NULL) ?
uxQueueMessagesWaiting(s_audio.uplink_send_queue) : 0U),
s_audio.uplink_send_in_flight ? 1 : 0);
reset_uplink_send_queue(&s_audio);
xSemaphoreGive(s_uplink_lifecycle_mutex);
if (err != ESP_OK) {
ESP_LOGW(TAG, "Mic uplink stop incomplete: %s", esp_err_to_name(err));
}
audio_set_uplink_callback(&s_audio, NULL, NULL);
s_audio.uplink_mono_fill_bytes = 0U;
log_heap_snapshot("after_uplink_stop");
return ESP_OK;
return err;
}
bool multimodal_audio_is_uplink_running(void)
docs/adr/0006-local-speech-endpoint.md
+30
View File
@@ -0,0 +1,30 @@
# ADR 0006: 本地语音端点检测所有权
## 状态
Accepted（实现与主机/构建验证；声学阈值待实机验收）
## 决策
- `domain` 独占本地语音端点契约、VAD 实例、分帧和连续人声/静音状态，声明于
`domain.h`。不向其他 facade 复制这些契约。
- `platform` 保持通用 PCM 观察者原语，将单订阅扩为固定两个槽位，供本地命令
和端点检测共享同一份 AEC 后 PCM。回调同步执行、不可阻塞，不新增采集任务。
- `control_plane` 在真实网络拾音启动后按 PTT generation 启用检测，查询领域快照，
只结束匹配代次的活动输入；保留 8 秒兜底和现有状态机的唯一收尾路径。
- 端点检测不在回调中停止采集或发送协议。平台停止先与在途采集串行，再由原采集任务补齐
最后一个 Opus 帧、限时排空发送；失败时废弃传输，不宣告正常结束或无限重试。
- 停止端点检测先禁用订阅并等待在途回调，再释放 VAD。超时保留资源供重试，
不安排后台释放。快照按值返回，所有 VAD 状态由互斥锁保护。
## 迁移与资源
1. 先扩展通用 PCM 订阅和明确音频停止语义，维持旧调用兼容。
2. 新增领域检测器和可关闭配置，再接入现有 PTT 推进器。
3. 主机验证实际 C 的分帧、代次、失败释放、订阅隔离和尾包顺序，构建启用/关闭配置。
4. 实机检查慢语速/停顿/噪声/远场/提示音、唤醒和按键、取消/断网/OTA，记录
端点到 StopSpeech/首个下行/播放的分段延迟及内部 RAM 最低余量。
固定检测帧为 320 个 int16 样本（640 B），不新增任务、队列或 DMA 缓冲。
WebRTC VAD 的库内分配及采集任务栈峰值需日志和设备测量，不能用帧大小冒充总开销。
检测初始化失败回退原输入窗口，不影响流式上行。
docs/knowledge vault/voice-uplink-flow-zh.md
+31 -38
View File
@@ -1,5 +1,10 @@
# 语音对话上行流程详解
> 2026-10-15 更新：默认单击/本地唤醒输入增加本地 VAD（确认 200 ms 人声后，
> 连续 900 ms VAD 静音提前收尾），保留 8 秒兜底。参数、所有权和验收见
> [本地语音端点检测](../local-speech-endpoint.zh-CN.md)。以下松键流程也适用于
> VAD/超时触发的统一 PTT 收尾。
## 1. 文档目的
本文结合当前工程代码，完整说明设备从用户按下语音键开始，到麦克风音频被编码为 Opus 并通过 WebSocket 发送给服务端为止的上行流程。
@@ -72,9 +77,11 @@ WebSocket Binary 发送
```text
用户松开 PTT
↓
停止产生新的 PCM/Opus 包
停止新增 PCM，等待在途采集迭代完成
↓
限时等待发送队列排空
采集任务为不足一帧的 PCM 补零编码
↓
限时等待发送队列和在途包排空
↓
清除上行发送回调
↓
@@ -458,6 +465,7 @@ while (!audio_get_uplink_task_exit_requested(ctx)) {
```c
if (!audio_get_uplink_recording_active(ctx)) {
audio_flush_uplink_tail(ctx);
ulTaskNotifyTake(pdTRUE, pdMS_TO_TICKS(100));
continue;
}
@@ -780,12 +788,15 @@ delete_audio_task_with_caps(NULL);
## 8. `uplink_send_task()` 如何发送 Opus
发送任务循环等待队列：
发送任务先在锁内标记 busy，再零等待出队；空队列时清 busy，并等待任务通知
（最多 20 ms）。正常入队后立即通知发送任务，不额外等待轮询周期：
```c
xQueueReceive(ctx->uplink_send_queue, &packet, pdMS_TO_TICKS(100));
xQueueReceive(ctx->uplink_send_queue, &packet, 0U);
```
停止方先查队列再查 busy，避免包刚出队、尚未标记在途时误认为已经排空。
取到有效包后，快照保存好的回调：
```c
@@ -874,51 +885,33 @@ WebSocket Binary 的 payload 是原始 Opus 包，不使用 RTP 封装。
## 11. 停止上行流程
用户松开 PTT 后，Control Plane 执行：
VAD、8 秒兜底或松键将活动输入送入同一 PTT teardown。只有平台停止成功后
才发送 `StopSpeech`；失败则废弃该轮并关闭传输，不能把未排空当成成功。
```c
control_plane_multimodal_stop_uplink();
control_plane_multimodal_send_speech_stop();
```
### 11.1 停止采集与尾帧补齐
### 11.1 停止产生新包
生命周期锁串行化 start/stop。停止方将 `uplink_recording_active` 设为 false，
再获取 capture 锁，等待当前 I2S read/AEC/编码迭代完成。
不足 Opus 帧长的 PCM 交回原采集任务补零编码，不挪到控制线程的小栈。
采集任务与调用方通过受锁保护的尾帧请求/结果握手，不新增编码 worker。
平台层首先执行：
### 11.2 有界排空
```c
audio_set_uplink_recording_active(&s_audio, false);
```
停止预算覆盖 capture 锁等待、尾帧处理、发送队列和在途包等待。
负超时无效；零超时只尝试立即完成，无法完成返回 `ESP_ERR_TIMEOUT`。
超时保留缓冲、回调和待发送数据供重试，不安排后台资源销毁；旧轮未收尾前
新 start 返回失败，不得复用旧数据。常驻采集/发送任务只在 deinit 回收。
采集任务下次循环时进入通知等待，不再读取 Codec，也不再产生新的 Opus 包。
### 11.3 清除回调与协议结束
### 11.2 等待队列排空
停止函数会在给定超时内等待：
- `uplink_send_queue` 为空；
- 没有正在发送的包。
这样可以尽量保证用户松键之前已经采集完成的最后几帧不会立即丢失。
如果超时仍未排空，代码记录警告并清空剩余队列，避免停止流程无限阻塞。
### 11.3 清除发送回调
```c
audio_set_uplink_callback(&s_audio, NULL, NULL);
```
这可以阻止旧队列数据在下一轮会话中误用上一轮发送上下文。
### 11.4 发送 StopSpeech
本地停止采集后，Control Plane 向服务端发送：
成功确认尾帧完成、队列为空且没有正在发送的包后，清除上行回调，随后发送：
```c
send_directive("continue-task", "StopSpeech", true, false);
```
服务端据此结束本轮语音输入，并继续执行 ASR、LLM 和 TTS。
`StopSpeech sent after uplink drain` 日志可与 VAD endpoint 日志对齐。
如果平台停止失败，PTT 收尾走有界传输废弃路径，不无限重试或继续正常提交输入。
## 12. 其他会触发停止上行的情况
docs/local-speech-endpoint.zh-CN.md
+90
View File
@@ -0,0 +1,90 @@
# 本地 VAD 语音端点检测
更新：2026-10-15。适用于 lucky-cat；实现所有权见 [ADR 0006](adr/0006-local-speech-endpoint.md)。
## 当前行为
音频仍然边采集边上传。服务端保持 `push2talk`，设备用本地 VAD 更早决定何时
发送 `StopSpeech`，不切换云端识别模式或修改语音回复语义。
| 配置 | 默认值 | 含义 |
| --- | --- | --- |
| `CONFIG_TQ_SPEECH_ENDPOINT_ENABLE` | y | 仅单击对话模式自动判停，可关闭回退 |
| `CONFIG_TQ_SPEECH_ENDPOINT_MIN_SPEECH_MS` | 200 | 开始判停前需要连续有效人声 |
| `CONFIG_TQ_SPEECH_ENDPOINT_SILENCE_MS` | 900 | 确认人声后，连续 VAD 静音达到该值结束输入 |
| `CONFIG_TQ_PTT_CLICK_CAPTURE_TIMEOUT_MS` | 8000 | 保留原最长窗口兜底 |
本地唤醒仍先完成联网和本地回应提示音，再开始网络拾音和 VAD。按键入口复用
原窗口起点，可能包含连接等待；本次不改变这一点。持续静音或未达到人声阈值的
短促声音不触发提前结束，继续走原超时。阈值以 20 ms 检测帧向上取整。
900 ms 指 VAD 输出连续静音，不等于从物理声音结束到 StopSpeech 的精确时长；
库内判定延迟、20 ms 分帧、现有 80 ms 控制轮询、调度及尾包发送都会影响时间。
句中停顿超过阈值可能被截断；慢语速、轻声、儿童、远场及噪声需实机调参。
## 所有权、并发与收尾
- `platform` 提供固定两个 PCM 观察者槽位，同一份 AEC 后 16 kHz 单声道 PCM
分别交给已有本地命令观察者与领域端点检测器。无第二条采集链或新音频队列。
- `domain_speech_endpoint.c` 使用 WebRTC `vad_create(VAD_MODE_1)` 和 `vad_process()`，
将输入重分帧为 320 样本。PCM 回调零等待取锁，最多处理 512 样本，不执行
协议、等待或资源销毁；状态查询和生命周期使用有界互斥锁。
- 丢失回调、非法输入或超过 100 ms 的输入间断重置连续人声/静音计时，保持保守
判停。只保留“曾确认有人声”，不能把缺失数据当静音补算。
- 控制层只接受活动 `STREAMING`、提示音门禁关闭且 generation 匹配的结束快照。
旧事件不会结束新一轮。初始化失败记录日志，保留原超时。
- 每轮仅尝试启动一次检测。收尾/取消/正常停机注销订阅并释放 VAD；注销超时
禁用订阅、保留实例和在途计数以便重试，不后台释放。
- 平台停止先暂停并等待在途采集，再由原采集任务补零编码尾帧。编码留在原
32 KB 采集任务栈，控制线程只等待结果。发送队列与在途包全部排空后才正常
发送 `StopSpeech`。停止失败走传输废弃路径，不提交正常结束、不无限重试。
- stop 超时后仍可能完成之前请求的尾帧和发送；资源保留，禁止启动新一轮，必须
重试收尾或走 deinit。deinit 才拥有常驻 worker 的退出/join/强制删除和资源销毁。
## 日志与资源
- `VAD armed`：代次、阈值及启用后的内部 RAM/PSRAM 空闲快照。
- `VAD endpoint`：代次、处理音频时长、连续静音、最大回调处理墙钟时间与间断数。
- `StopSpeech sent after uplink drain`：正常输入结束指令已发出。
- `VAD unavailable; retain capture timeout`：本轮回退原输入窗口。
- `Abort voice turn after incomplete drain`：失败废弃，不能记作成功回复。
固定 PCM 重分帧数组为 640 B，另有领域状态/互斥锁和平台第二个观察者槽、
采集与生命周期锁。不新增 worker、任务栈、DMA 或模型分区资源。
VAD 库内部动态分配、原采集栈高水位和联网后的最小内部 RAM 必须实机测量；
`internal_free` 是并发系统的空闲快照，不能直接当成 VAD 的精确分配量。
`max_process_us` 为墙钟时间，含抢占，不是纯 CPU 时间。PSRAM 不增加固定像素/
音频队列缓冲，但库分配采用的实际堆区域仍以运行日志为准。
## 验证与验收
`python3 tools/test_speech_endpoint.py` 编译并执行生产 C，使用可控 VAD/RTOS 桩：
覆盖分帧、纯静音、短噪声、句中停顿、代次、输入间断、初始化失败、注销超时、
订阅并发替换、队列出队与 busy 竞态、原采集上下文尾帧补零、排空与失败重试。
这些测试验证控制逻辑，不验证真实声学识别率或 ESP32 时序。
实机至少检查：
1. 按键和唤醒各测试短句、长句、未说话、轻声、噪声和不同长度句中停顿。
2. 连续多轮、快速取消/重新触发、服务端提前结束、TTS 播放及唤醒提示音门禁。
3. 拾音期间断网、OTA/待机切换、发送拥塞、内存不足与重复初始化。
4. 记录最后人声 → endpoint → StopSpeech → 首个下行音频 → 实际播放的分段延迟；
同时统计误截断、未判停、队列丢包、回调耗时、栈高水位和内部 RAM 最低余量。
本次构建/主机验证结果见下方记录；没有实机日志前，不承诺提速秒数或声学阈值合适。
### 2026-10-15 本次验证记录
- 主机：`test_speech_endpoint.py` 全部通过，启用/关闭分支均编译并执行；
五个现有 `test_lcd_{status_hud,emote_transport,emote_timeline,listening_gfx,fortune_gfx}.py`
回归脚本通过（脚本自带的 profiling 开/关路径也通过）。
- 构建：ESP-IDF 5.5.4，VAD 开启和关闭两个独立 sdkconfig 均构建成功；
项目默认配置保持开启。关闭配置与构建在临时目录，未覆盖项目 sdkconfig。
- 固件：开启 3,167,888 B，关闭 3,157,248 B，开关差异 +10,640 B；
同源 MAP 的内部静态 data +16 B、BSS +832 B，合计 +848 B，IRAM 代码无增量。
这是同一份修改中 VAD 开关的对比，双方均包含共享 PCM/尾包收尾修正，
不能当成整个变更相对旧提交的净成本。模型分区未变，没有新增固定 PSRAM/DMA 缓冲。
- 架构：原检查脚本在已有 `control_plane_battery_service.c` 的 `portMAX_DELAY`
处失败，已对照 HEAD 确认不是本次引入。仅排除该既有检查后，其余架构检查通过；
本次新增路径无无限等待或跨层内部头依赖。`git diff --check` 通过。
- 设备：未烧录，尚无实机误截断率、延迟收益、VAD 动态堆和栈峰值数据。
main/Kconfig.projbuild
+22
View File
@@ -619,6 +619,28 @@ menu "TQ Cat Configuration"
Maximum microphone capture window started by one PTT short press.
When it expires, uplink capture is stopped and StopSpeech is sent.
config TQ_SPEECH_ENDPOINT_ENABLE
bool "End click-to-talk input using local VAD"
default y
depends on TQ_KEY_INPUT_ENABLE && TQ_PTT_MODE_CLICK_TO_TALK
help
Run WebRTC VAD on AEC-clean PCM while streaming audio. After
confirmed speech and continuous silence, finish the current
input through the existing PTT teardown. Keep the capture
timeout as fallback. No additional audio worker is created.
config TQ_SPEECH_ENDPOINT_MIN_SPEECH_MS
int "Minimum continuous speech before endpoint detection (ms)"
default 200
range 100 1000
depends on TQ_SPEECH_ENDPOINT_ENABLE
config TQ_SPEECH_ENDPOINT_SILENCE_MS
int "Continuous silence to end confirmed speech (ms)"
default 900
range 400 2000
depends on TQ_SPEECH_ENDPOINT_ENABLE
endmenu
menu "Display & Screen"
sdkconfig.defaults
+3
View File
@@ -92,6 +92,9 @@ CONFIG_TQ_KEY_POST_DOUBLE_CLICK_SUPPRESS_MS=450
CONFIG_TQ_PTT_MODE_CLICK_TO_TALK=y
# CONFIG_TQ_PTT_MODE_HOLD_TO_TALK is not set
CONFIG_TQ_PTT_CLICK_CAPTURE_TIMEOUT_MS=8000
CONFIG_TQ_SPEECH_ENDPOINT_ENABLE=y
CONFIG_TQ_SPEECH_ENDPOINT_MIN_SPEECH_MS=200
CONFIG_TQ_SPEECH_ENDPOINT_SILENCE_MS=900
CONFIG_TQ_LOCAL_WAKE_ENABLE=y
CONFIG_TQ_OTA_ENABLE=y
CONFIG_TQ_OTA_REPORT_URL="https://gz.api.talkingq.com/ota/api/ota/versionInfoReport"
tools/test_speech_endpoint.py
+376
View File
@@ -0,0 +1,376 @@
#!/usr/bin/env python3
"""Host-test production C with deterministic VAD/RTOS stubs; not acoustic validation."""
from pathlib import Path
import re
import subprocess
import tempfile
ROOT = Path(__file__).resolve().parents[1]
DOMAIN = ROOT / 'components/domain/src/runtime/domain_speech_endpoint.c'
AUDIO = (ROOT / 'components/platform/src/multimodal/audio_pipeline/audio_pipeline_runtime_impl.c').read_text()
def extract(src, name):
m = re.search(r'(?:static\s+)?[\w* ]+\b' + name + r'\([^;{}]*\)\s*\{', src)
assert m, name
start = src.index('{', m.start())
end, depth = start + 1, 1
while depth:
depth += (src[end] == '{') - (src[end] == '}')
end += 1
return src[m.start():end]
COMMON = r'''
#pragma once
#include <assert.h>
#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
typedef int esp_err_t;
#define ESP_OK 0
#define ESP_ERR_INVALID_ARG 1
#define ESP_ERR_INVALID_STATE 2
#define ESP_ERR_NO_MEM 3
#define ESP_ERR_TIMEOUT 4
#define ESP_ERR_NOT_SUPPORTED 5
#define ESP_LOGI(...) ((void)0)
#define ESP_LOGW(...) ((void)0)
#define MALLOC_CAP_INTERNAL 1
#define MALLOC_CAP_8BIT 2
#define MALLOC_CAP_SPIRAM 4
#define pdTRUE 1
#define pdMS_TO_TICKS(x) (x)
#define pdTICKS_TO_MS(x) (x)
#define portENTER_CRITICAL(x) ((void)(x))
#define portEXIT_CRITICAL(x) ((void)(x))
#define portMUX_INITIALIZER_UNLOCKED 0
typedef int portMUX_TYPE;
typedef uint32_t TickType_t;
typedef struct { bool taken; } StaticSemaphore_t;
typedef StaticSemaphore_t *SemaphoreHandle_t;
static TickType_t ticks;
static int64_t now_us = 1000;
static int fail_lock;
static SemaphoreHandle_t xSemaphoreCreateMutexStatic(StaticSemaphore_t *p) { return p; }
static int xSemaphoreTake(SemaphoreHandle_t p, unsigned wait) {
(void)wait;
if(fail_lock || p->taken) return 0;
p->taken=true; return 1;
}
static void xSemaphoreGive(SemaphoreHandle_t p) { assert(p->taken); p->taken=false; }
static TickType_t xTaskGetTickCount(void) { return ticks; }
static void (*delay_hook)(void);
static void vTaskDelay(unsigned t) { ticks+=t; if(delay_hook) delay_hook(); }
static int64_t esp_timer_get_time(void) { return ++now_us; }
static size_t heap_caps_get_free_size(int caps) { (void)caps; return 12345; }
static const char *esp_err_to_name(int e) { (void)e; return "stub"; }
'''
DOMAIN_STUB = r'''
#include "common.h"
typedef void (*platform_multimodal_audio_pcm_observer_t)(const int16_t *, size_t, uint32_t, void *);
static platform_multimodal_audio_pcm_observer_t registered;
static int register_error, unregister_error, vad_alloc_fail, live_vad, vad_calls;
static esp_err_t platform_multimodal_audio_register_pcm_observer(platform_multimodal_audio_pcm_observer_t cb, void *ctx) {
(void)ctx; if(register_error) return register_error; assert(!registered); registered=cb; return 0;
}
static esp_err_t platform_multimodal_audio_unregister_pcm_observer(platform_multimodal_audio_pcm_observer_t cb, void *ctx, int ms) {
(void)ctx; (void)ms; assert(cb==registered); if(unregister_error) return unregister_error;
registered=NULL; return 0;
}
typedef void *vad_handle_t;
typedef enum {VAD_SILENCE, VAD_SPEECH} vad_state_t;
#define VAD_MODE_1 1
static vad_handle_t vad_create(int m) { (void)m; if(vad_alloc_fail) return NULL; ++live_vad; return malloc(1); }
static void vad_destroy(vad_handle_t h) { assert(h); --live_vad; free(h); }
static vad_state_t vad_process(vad_handle_t h, int16_t *pcm, int rate, int ms) {
assert(h && rate==16000 && ms==20); ++vad_calls;
return pcm[0] ? VAD_SPEECH : VAD_SILENCE;
}
'''
DOMAIN_TEST = r'''
static domain_speech_endpoint_snapshot_t snap(void) {
domain_speech_endpoint_snapshot_t s;
assert(domain_speech_endpoint_get_snapshot(&s)==0); return s;
}
static void feed(unsigned samples, bool voice, unsigned chunk) {
int16_t pcm[512]; for(unsigned i=0;i<512;i++) pcm[i]=voice;
while(samples) { unsigned n=samples<chunk?samples:chunk; now_us+=n*1000000LL/16000;
registered(pcm,n,16000,NULL); samples-=n; }
}
static void reset(unsigned generation) {
assert(domain_speech_endpoint_stop()==0 && live_vad==0);
domain_speech_endpoint_config_t c={generation,200,900};
assert(domain_speech_endpoint_start(&c)==0 && live_vad==1);
}
int main(void) {
domain_speech_endpoint_config_t c={1,200,900};
assert(domain_speech_endpoint_start(NULL)==ESP_ERR_INVALID_ARG);
c.generation=0; assert(domain_speech_endpoint_start(&c)==ESP_ERR_INVALID_ARG); c.generation=1;
vad_alloc_fail=1; assert(domain_speech_endpoint_start(&c)==ESP_ERR_NO_MEM); vad_alloc_fail=0;
register_error=ESP_ERR_NO_MEM; assert(domain_speech_endpoint_start(&c)==ESP_ERR_NO_MEM && live_vad==0); register_error=0;
assert(domain_speech_endpoint_start(&c)==0);
assert(domain_speech_endpoint_start(&c)==0 && live_vad==1);
c.generation=2; assert(domain_speech_endpoint_start(&c)==ESP_ERR_INVALID_STATE);
feed(16000*8,false,256); assert(!snap().speech_confirmed && !snap().ended);
reset(2); feed(1600,true,17); feed(16000,false,256); assert(!snap().ended);
reset(3); feed(3200,true,256); assert(snap().speech_confirmed);
feed(14080,false,256); assert(!snap().ended && snap().silence_ms==880);
feed(320,false,160); assert(snap().ended && snap().generation==3);
reset(4); assert(!snap().ended); feed(3200,true,320); feed(9600,false,320);
feed(320,true,320); assert(snap().silence_ms==0); feed(14400,false,256); assert(snap().ended);
reset(5); feed(3200,true,320); feed(12800,false,320);
fail_lock=1; feed(320,false,320); fail_lock=0;
feed(1600,false,320); assert(!snap().ended && snap().silence_ms==100 && snap().discontinuities==1);
now_us+=200000; feed(320,false,320); assert(snap().silence_ms==20 && snap().discontinuities==2);
int16_t bad[512]={0}; registered(bad,512,8000,NULL); assert(snap().silence_ms==0);
unregister_error=ESP_ERR_TIMEOUT; assert(domain_speech_endpoint_stop()==ESP_ERR_TIMEOUT && live_vad==1);
assert(!snap().active); registered(bad,320,16000,NULL); assert(live_vad==1);
assert(domain_speech_endpoint_start(&c)==ESP_ERR_INVALID_STATE);
unregister_error=0; assert(domain_speech_endpoint_stop()==0 && live_vad==0 && !registered);
assert(domain_speech_endpoint_stop()==0);
puts("PASS: actual domain C framing, silence, short noise, pauses, generations, gaps, init failure and timeout retention");
}
'''
OBSERVER_STUB = r'''
#include "common.h"
#define CONFIG_MULTIMODAL_AUDIO_CODEC_SAMPLE_RATE 16000
#define MULTIMODAL_AUDIO_PCM_OBSERVER_SLOTS 2U
typedef void (*multimodal_audio_pcm_observer_t)(const int16_t *,size_t,uint32_t,void *);
typedef struct {multimodal_audio_pcm_observer_t callback; void *ctx; uint32_t in_flight; uint32_t registration; bool enabled;} audio_pcm_observer_slot_t;
static audio_pcm_observer_slot_t s_pcm_observers[2];
static uint32_t s_pcm_observer_registration;
static int s_pcm_observer_lock;
'''
OBSERVER_TEST = r'''
static int calls[3], mode;
static void replace_registration(void);
static void observer(const int16_t *p,size_t n,uint32_t rate,void *ctx) {
assert(n==1 && rate==16000 && p); int id=(int)(intptr_t)ctx; ++calls[id];
if(mode && id==0) {
assert(multimodal_audio_unregister_pcm_observer(observer,(void *)1,0)==ESP_ERR_TIMEOUT);
assert(multimodal_audio_register_pcm_observer(observer,(void *)1)==ESP_ERR_INVALID_STATE);
}
}
static void replace_registration(void) {
delay_hook=NULL;
for(unsigned i=0;i<2;i++) if(s_pcm_observers[i].ctx==(void *)2) s_pcm_observers[i].in_flight=0;
assert(multimodal_audio_unregister_pcm_observer(observer,(void *)2,0)==0);
assert(multimodal_audio_register_pcm_observer(observer,(void *)2)==0);
}
int main(void) {
int16_t pcm=0;
assert(multimodal_audio_register_pcm_observer(NULL,NULL)==ESP_ERR_INVALID_ARG);
assert(multimodal_audio_register_pcm_observer(observer,NULL)==0);
assert(multimodal_audio_register_pcm_observer(observer,NULL)==0);
assert(multimodal_audio_register_pcm_observer(observer,(void *)1)==0);
assert(multimodal_audio_register_pcm_observer(observer,(void *)2)==ESP_ERR_NO_MEM);
notify_pcm_observer(&pcm,1); assert(calls[0]==1 && calls[1]==1);
mode=1; notify_pcm_observer(&pcm,1); assert(calls[0]==2 && calls[1]==2);
mode=0; notify_pcm_observer(&pcm,1); assert(calls[0]==3 && calls[1]==2);
assert(multimodal_audio_unregister_pcm_observer(observer,(void *)1,10)==0);
assert(multimodal_audio_register_pcm_observer(observer,(void *)2)==0);
notify_pcm_observer(&pcm,1); assert(calls[0]==4 && calls[2]==1);
assert(multimodal_audio_unregister_pcm_observer(observer,NULL,10)==0);
notify_pcm_observer(&pcm,1); assert(calls[0]==4 && calls[2]==2);
for(unsigned i=0;i<2;i++) if(s_pcm_observers[i].ctx==(void *)2) s_pcm_observers[i].in_flight=1;
delay_hook=replace_registration;
assert(multimodal_audio_unregister_pcm_observer(observer,(void *)2,10)==0);
notify_pcm_observer(&pcm,1); assert(calls[2]==3);
puts("PASS: actual PCM observer C two slots, in-flight timeout, independent cleanup and concurrent replacement");
}
'''
STOP_STUB = r'''
#include "common.h"
typedef int (*multimodal_audio_uplink_cb_t)(const uint8_t *,size_t,void *);
typedef struct { bool active, uplink_tail_flush_requested, uplink_stop_incomplete; esp_err_t uplink_tail_flush_error; size_t uplink_mono_fill_bytes,uplink_mono_bytes; int16_t *uplink_mono_buf;
bool uplink_first_network_pcm_pending,uplink_first_network_packet_pending; } multimodal_audio_ctx_t;
static multimodal_audio_ctx_t s_audio;
static int s_audio_runtime_lock;
static StaticSemaphore_t capture_mutex, lifecycle_mutex;
static SemaphoreHandle_t s_uplink_lifecycle_mutex=&lifecycle_mutex;
static bool ensure_uplink_mutexes(void) { return true; }
typedef void *TaskHandle_t;
static void audio_snapshot_uplink_tasks(multimodal_audio_ctx_t *c,TaskHandle_t *task,void *send) {(void)c;(void)send;*task=(void *)1;}
static void xTaskNotifyGive(TaskHandle_t task) {(void)task;}
static int encode_context;
static bool deinitializing;
static bool audio_get_deinit_in_progress(multimodal_audio_ctx_t *c) {(void)c;return deinitializing;}
static bool audio_get_uplink_recording_active(multimodal_audio_ctx_t *c) {return c->active;}
static SemaphoreHandle_t s_uplink_capture_mutex=&capture_mutex;
static int pending, drain_after, encode_error, encoded, cleared, acquire_fail, finish_capture;
static int dummy_send(const uint8_t *p,size_t n,void *c) {(void)p;(void)n;(void)c;return 0;}
static multimodal_audio_uplink_cb_t callback=dummy_send;
static bool lock_uplink_capture(unsigned ticks) {
(void)ticks; if(acquire_fail) return false;
if(finish_capture) {s_audio.uplink_mono_fill_bytes=6;finish_capture=0;}
capture_mutex.taken=true; return true;
}
static void audio_set_uplink_recording_active(multimodal_audio_ctx_t *c,bool b) {c->active=b;}
static void audio_snapshot_uplink_callback(multimodal_audio_ctx_t *c,multimodal_audio_uplink_cb_t *cb,void *ctx) {(void)c;(void)ctx;*cb=callback;}
static bool audio_uplink_send_pending(multimodal_audio_ctx_t *c) {
(void)c; if(pending && ticks>=(unsigned)drain_after && drain_after>=0) pending=0; return pending;
}
static int encode_and_queue_mono_pcm(multimodal_audio_ctx_t *c,const int16_t *pcm,size_t bytes) {
(void)c; assert(encode_context && capture_mutex.taken && bytes==8 && pcm[0]==11);
assert(pcm[1]==0 && pcm[2]==0 && pcm[3]==0);
++encoded; if(encode_error) return encode_error; pending=1; return 0;
}
static void audio_set_uplink_callback(multimodal_audio_ctx_t *c,multimodal_audio_uplink_cb_t cb,void *ctx) {
(void)c;(void)ctx;assert(!pending);callback=cb;++cleared;
}
'''
STOP_START_STUB = r'''
static int audio_start_uplink_locked(multimodal_audio_uplink_cb_t cb,void *ctx) {(void)ctx;s_audio.active=true;callback=cb;return 0;}
'''
STOP_TEST = r'''
static void run_capture(void) {encode_context=1; audio_flush_uplink_tail(&s_audio); encode_context=0;}
int main(void) {
delay_hook=run_capture;
int16_t pcm[4]={11,0,0,99}; s_audio.uplink_mono_buf=pcm; s_audio.uplink_mono_bytes=8;
assert(multimodal_audio_stop_uplink(-1)==ESP_ERR_INVALID_ARG);
s_audio.uplink_mono_fill_bytes=2; s_audio.active=true;
assert(multimodal_audio_stop_uplink(0)==ESP_ERR_TIMEOUT && encoded==0 && !cleared);
acquire_fail=1; assert(multimodal_audio_stop_uplink(10)==ESP_ERR_TIMEOUT && callback); acquire_fail=0;
finish_capture=1; drain_after=5;
assert(multimodal_audio_stop_uplink(20)==0 && encoded==1 && cleared==1 && ticks==5);
assert(!callback && s_audio.uplink_mono_fill_bytes==0 && !s_audio.active);
assert(multimodal_audio_stop_uplink(20)==0); finish_capture=0;
callback=dummy_send; pending=1; drain_after=-1;
assert(multimodal_audio_stop_uplink(10)==ESP_ERR_TIMEOUT && callback && pending);
pending=0; assert(multimodal_audio_stop_uplink(10)==0 && !callback);
callback=dummy_send; s_audio.uplink_mono_fill_bytes=2; encode_error=ESP_ERR_NO_MEM;
assert(multimodal_audio_stop_uplink(10)==ESP_ERR_NO_MEM && callback && s_audio.uplink_mono_fill_bytes==2);
assert(multimodal_audio_start_uplink(dummy_send,NULL)==ESP_ERR_INVALID_STATE);
encode_error=0; drain_after=ticks+2;
assert(multimodal_audio_stop_uplink(10)==0 && !s_audio.uplink_stop_incomplete);
assert(multimodal_audio_start_uplink(dummy_send,NULL)==0 && s_audio.active);
assert(multimodal_audio_stop_uplink(10)==0);
deinitializing=true; assert(multimodal_audio_start_uplink(dummy_send,NULL)==ESP_ERR_INVALID_STATE);
puts("PASS: actual start/stop C capture handshake, capture-context padding, drain order, budgets, retry and reuse guards");
}
'''
CP_STUB = r"""
#include "common.h"
#include "domain.h"
#define CONFIG_TQ_SPEECH_ENDPOINT_ENABLE 1
#define CONFIG_TQ_PTT_MODE_CLICK_TO_TALK 1
#define CONFIG_TQ_PTT_CLICK_CAPTURE_TIMEOUT_MS 8000
#define CONTROL_PLANE_PTT_DENIED_TIMEOUT_MS 15000
"""
CP_TEST = r"""
static ptt_runtime_t ptt;
static domain_speech_endpoint_snapshot_t endpoint;
static bool gate, change_generation;
static int deactivated;
static const char *last_reason;
static ptt_runtime_t control_plane_ptt_snapshot(void) {return ptt;}
static int domain_speech_endpoint_stop(void) {endpoint.active=false;return 0;}
static int domain_speech_endpoint_get_snapshot(domain_speech_endpoint_snapshot_t *out) {
*out=endpoint; if(change_generation) {++ptt.generation;change_generation=false;} return 0;
}
static bool control_plane_local_wake_prompt_gate_active(void) {return gate;}
static bool control_plane_deactivate_ptt(bool check,uint32_t gen,const char *reason) {
assert(check); if(gen!=ptt.generation) return false;
++deactivated; last_reason=reason; ptt.active=false; ++ptt.generation;return true;
}
static void control_plane_ptt_set_state(ptt_state_t state,const char *why) {(void)why;ptt.state=state;}
"""
CP_MAIN = r"""
int main(void) {
ticks=100; ptt=(ptt_runtime_t){.active=true,.state=PTT_STATE_STREAMING,.generation=7,.press_tick=1};
endpoint=(domain_speech_endpoint_snapshot_t){.active=true,.ended=true,.generation=6};
assert(!control_plane_service_ptt_lifecycle() && deactivated==0);
endpoint.generation=7; gate=true; assert(!control_plane_service_ptt_lifecycle() && !deactivated);
gate=false; change_generation=true; assert(control_plane_service_ptt_lifecycle() && !deactivated && ptt.active);
endpoint.generation=ptt.generation;
assert(control_plane_service_ptt_lifecycle() && deactivated==1 && !strcmp(last_reason,"vad_endpoint"));
ptt.active=true; ptt.state=PTT_STATE_STREAMING; endpoint.active=false;
assert(!control_plane_service_ptt_lifecycle()); ticks=8001;
assert(control_plane_service_ptt_lifecycle() && deactivated==2 && !strcmp(last_reason,"capture_timeout"));
ptt=(ptt_runtime_t){.state=PTT_STATE_IDLE}; endpoint.active=true;
assert(!control_plane_service_ptt_lifecycle() && !endpoint.active);
puts("PASS: actual control-plane C stale-generation rejection, prompt gate, concurrent cancel and 8s fallback");
}
"""
PENDING_TEST = r"""
#include "common.h"
typedef struct {bool uplink_send_in_flight;void *uplink_send_queue;} multimodal_audio_ctx_t;
static int s_audio_runtime_lock, queue_count;
static bool dequeue_race;
static multimodal_audio_ctx_t ctx;
static unsigned uxQueueMessagesWaiting(void *queue) {
(void)queue;
if(dequeue_race) {ctx.uplink_send_in_flight=true;return 0;}
return queue_count;
}
"""
PENDING_MAIN = r"""
int main(void) {
ctx.uplink_send_queue=(void *)1;
assert(!audio_uplink_send_pending(&ctx));
queue_count=1;assert(audio_uplink_send_pending(&ctx));
queue_count=0;dequeue_race=true;assert(audio_uplink_send_pending(&ctx));
puts("PASS: actual drain query closes queue-empty versus in-flight dequeue race");
}
"""
def compile_run(tmp, name, code):
path = tmp / (name + '.c')
path.write_text(code)
binary = tmp / name
subprocess.run(['cc', '-std=c11', '-Wall', '-Wextra', '-Werror',
'-Wno-unused-function', '-Wno-unused-variable',
'-I' + str(tmp), '-I' + str(ROOT / 'components/domain/include'),
'-I' + str(ROOT / 'components/platform/include'), str(path), '-o', str(binary)], check=True)
subprocess.run([str(binary)], check=True)
with tempfile.TemporaryDirectory(prefix='speech-endpoint-') as directory:
tmp = Path(directory)
(tmp / 'common.h').write_text(COMMON)
(tmp / 'freertos').mkdir()
for header in ['esp_err.h', 'esp_log.h', 'esp_timer.h', 'esp_heap_caps.h',
'freertos/FreeRTOS.h', 'freertos/semphr.h', 'freertos/task.h']:
(tmp / header).write_text('#include "common.h"\n')
(tmp / 'esp_vad.h').write_text(DOMAIN_STUB)
(tmp / 'sdkconfig.h').write_text('#define CONFIG_TQ_SPEECH_ENDPOINT_ENABLE 1\n')
compile_run(tmp, 'domain', '#include "common.h"\n' + DOMAIN.read_text() + DOMAIN_TEST)
(tmp / 'sdkconfig.h').write_text('#define CONFIG_TQ_SPEECH_ENDPOINT_ENABLE 0\n')
compile_run(tmp, 'disabled', '#include "common.h"\n' + DOMAIN.read_text() + r'''
int main(void) { domain_speech_endpoint_snapshot_t s;
assert(domain_speech_endpoint_start(NULL)==ESP_ERR_NOT_SUPPORTED);
assert(domain_speech_endpoint_stop()==0);
assert(domain_speech_endpoint_get_snapshot(&s)==0 && !s.active);
puts("PASS: disabled domain C fallback"); }
''')
compile_run(tmp, 'observers', OBSERVER_STUB + '\n'.join(extract(AUDIO, n) for n in [
'multimodal_audio_register_pcm_observer', 'multimodal_audio_unregister_pcm_observer',
'notify_pcm_observer']) + OBSERVER_TEST)
compile_run(tmp, 'pending', PENDING_TEST + extract(AUDIO, 'audio_uplink_send_pending') + PENDING_MAIN)
cp_source=(ROOT / 'components/control_plane/src/runtime/control_plane_runtime_workers.c').read_text()
cp_types=re.search(r'typedef enum \{\s+PTT_STATE_IDLE.*?} ptt_runtime_t;',cp_source,re.S).group(0)
# Hide facade declarations only for test doubles; production function is unchanged.
cp_test=CP_TEST.replace('static int domain_speech_endpoint_', 'int domain_speech_endpoint_')
compile_run(tmp, 'control_plane', CP_STUB + cp_types + cp_test + extract(cp_source, 'control_plane_service_ptt_lifecycle') + CP_MAIN)
compile_run(tmp, 'stop', STOP_STUB + '\n'.join(extract(AUDIO, n) for n in ['audio_uplink_tail_pending', 'audio_flush_uplink_tail', 'multimodal_audio_stop_uplink']) + STOP_START_STUB + extract(AUDIO, 'multimodal_audio_start_uplink') + STOP_TEST)
# Integration guards supplement C execution; they do not replace device timing tests.
cp = (ROOT / 'components/control_plane/src/runtime/control_plane_runtime_workers.c').read_text()
assert 'endpoint.generation == ptt.generation' in cp
assert 'endpoint_start_attempted' in cp and '"vad_endpoint"' in cp
teardown = extract(cp, 'control_plane_handle_ptt_teardown')
assert teardown.index('domain_speech_endpoint_stop') < teardown.index('control_plane_multimodal_stop_uplink')
assert teardown.index('control_plane_multimodal_stop_uplink') < teardown.index('control_plane_multimodal_send_speech_stop')
assert 'CONTROL_PLANE_TRANSPORT_LOSS_PROTOCOL_ERROR' in teardown
print('PASS: generation-gated integration and bounded abort path')