chenjintao/lucky-cat
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
6d103938ab371c399088847a657650de910e143b
lucky-cat/tools/test_speech_endpoint.py
T

477 lines
26 KiB
Python
Raw Blame History

#!/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, snapshot_error;
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) {
if(snapshot_error) return snapshot_error;
*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) {
test_activation();
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.active=true; snapshot_error=ESP_ERR_TIMEOUT;
assert(control_plane_service_ptt_lifecycle() && deactivated==3 && !strcmp(last_reason,"capture_timeout"));
snapshot_error=0;
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");
}
"""
CP_ACTIVATION_STUB = r"""
#define s_ptt_runtime ptt
#define CONTROL_PLANE_NETWORK_ACTION_CONVERSATION 1
static int s_runtime_lock, pauses, resumes, expressions, notifications, readiness_calls;
static bool s_ws_start_requested, ota_allowed=true;
static esp_err_t readiness_error;
static void (*readiness_hook)(void);
static bool control_plane_ota_business_interactions_allowed(void) {return ota_allowed;}
static void control_plane_local_wake_pause_for_foreground(void) {++pauses;}
static void control_plane_local_wake_resume_after_foreground(void) {++resumes;}
static esp_err_t control_plane_network_feedback_require_ready(int action) {
(void)action; ++readiness_calls; if(readiness_hook) readiness_hook(); return readiness_error;
}
static void control_plane_ptt_transition_locked(ptt_state_t state,const char *reason) {
(void)reason; ptt.state=state; ptt.state_enter_tick=ticks;
}
static void control_plane_ptt_log_transition(ptt_state_t a,ptt_state_t b,uint32_t gen,const char *why) {
(void)a;(void)b;(void)gen;(void)why;
}
static void control_plane_request_ptt_expression(domain_expression_id_t id,domain_expression_trigger_t trigger) {
(void)id;(void)trigger;++expressions;
}
static void control_plane_notify_wifi_state_task(void) {++notifications;}
"""
CP_ACTIVATION_TEST = r"""
static void concurrent_activation(void) {
readiness_hook=NULL;
control_plane_set_ptt_active_with_trigger(true,DOMAIN_EXPRESSION_TRIGGER_LOCAL_WAKE);
}
static void test_activation(void) {
const ptt_state_t stages[]={PTT_STATE_WAIT_WIFI,PTT_STATE_WAIT_WS,PTT_STATE_WAIT_START_SENT,
PTT_STATE_WAIT_LISTENING,PTT_STATE_WAIT_WAKE_PROMPT,PTT_STATE_WAIT_SPEECH_STARTED,
PTT_STATE_STREAMING};
ticks=100; ptt=(ptt_runtime_t){.state=PTT_STATE_IDLE};
ota_allowed=false;
control_plane_set_ptt_active_with_trigger(true,DOMAIN_EXPRESSION_TRIGGER_PTT_BUTTON);
assert(!ptt.active && !pauses && !readiness_calls);
ota_allowed=true; readiness_error=ESP_ERR_INVALID_STATE;
control_plane_set_ptt_active_with_trigger(true,DOMAIN_EXPRESSION_TRIGGER_PTT_BUTTON);
assert(!ptt.active && resumes==1);
readiness_error=ESP_OK;
control_plane_set_ptt_active_with_trigger(true,DOMAIN_EXPRESSION_TRIGGER_PTT_BUTTON);
assert(ptt.active && ptt.generation==1 && ptt.press_tick==100 && expressions==1);
assert(ptt.state==PTT_STATE_WAIT_WIFI && s_ws_start_requested && notifications==1);
for(unsigned i=0;i<sizeof(stages)/sizeof(stages[0]);++i) {
ptt.state=stages[i]; ptt.endpoint_start_attempted=true;
ptt.start_action_sent=true; ptt.speech_action_sent=true;
ptt_runtime_t before=ptt;
int previous_pauses=pauses, previous_ready=readiness_calls;
ticks+=1000;
control_plane_set_ptt_active_with_trigger(true,DOMAIN_EXPRESSION_TRIGGER_LOCAL_WAKE);
assert(!memcmp(&before,&ptt,sizeof(ptt)));
assert(pauses==previous_pauses && readiness_calls==previous_ready && expressions==1);
}
/* 重复触发后仍接收原代次端点；窗口也不能被续期。 */
endpoint=(domain_speech_endpoint_snapshot_t){.active=true,.ended=true,.generation=ptt.generation};
assert(control_plane_service_ptt_lifecycle() && !strcmp(last_reason,"vad_endpoint"));
ptt=(ptt_runtime_t){.active=true,.state=PTT_STATE_STREAMING,.generation=9,.press_tick=100};
endpoint=(domain_speech_endpoint_snapshot_t){0}; ticks=8099;
control_plane_set_ptt_active_with_trigger(true,DOMAIN_EXPRESSION_TRIGGER_PTT_BUTTON);
assert(!control_plane_service_ptt_lifecycle()); ++ticks;
assert(control_plane_service_ptt_lifecycle() && !strcmp(last_reason,"capture_timeout"));
const ptt_state_t teardown[]={PTT_STATE_TEARDOWN_LOCAL,PTT_STATE_TEARDOWN_REMOTE};
for(unsigned i=0;i<2;++i) {
ptt=(ptt_runtime_t){.state=teardown[i],.generation=10};
control_plane_set_ptt_active_with_trigger(true,DOMAIN_EXPRESSION_TRIGGER_PTT_BUTTON);
assert(ptt.active && ptt.generation==11 && ptt.pending_press_after_teardown);
assert(ptt.state==teardown[i] && !ptt.endpoint_start_attempted);
}
ptt=(ptt_runtime_t){.state=PTT_STATE_IDLE,.generation=20};
readiness_hook=concurrent_activation;
int previous_expressions=expressions;
control_plane_set_ptt_active_with_trigger(true,DOMAIN_EXPRESSION_TRIGGER_PTT_BUTTON);
assert(ptt.generation==21 && expressions==previous_expressions+1);
assert(ptt.listening_ui_trigger==DOMAIN_EXPRESSION_TRIGGER_LOCAL_WAKE);
deactivated=0;
puts("PASS: actual PTT activation C ignores repeats, preserves endpoint/window, queues teardown and rejects concurrent duplicate");
}
"""
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_')
cp_lifecycle=extract(cp_source, 'control_plane_service_ptt_lifecycle')
cp_activation='\n'.join(extract(cp_source, n) for n in [
'control_plane_ptt_has_active_turn_locked', 'control_plane_set_ptt_active_with_trigger'])
compile_run(tmp, 'control_plane', CP_STUB + cp_types + cp_test + CP_ACTIVATION_STUB +
cp_activation + cp_lifecycle + CP_ACTIVATION_TEST + CP_MAIN)
compile_run(tmp, 'control_plane_disabled', CP_STUB.replace(
'#define CONFIG_TQ_SPEECH_ENDPOINT_ENABLE 1', '#define CONFIG_TQ_SPEECH_ENDPOINT_ENABLE 0') +
cp_types + cp_test + CP_ACTIVATION_STUB + cp_activation + cp_lifecycle + r'''
int main(void) {
ticks=100; control_plane_set_ptt_active_with_trigger(true,DOMAIN_EXPRESSION_TRIGGER_PTT_BUTTON);
ptt.state=PTT_STATE_STREAMING;
ticks=8100; control_plane_set_ptt_active_with_trigger(true,DOMAIN_EXPRESSION_TRIGGER_LOCAL_WAKE);
assert(ptt.generation==1 && ptt.press_tick==100);
assert(control_plane_service_ptt_lifecycle() && !strcmp(last_reason,"capture_timeout"));
puts("PASS: VAD disabled control-plane C preserves generation and 8s fallback on repeated activation");
}
''')
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')
Reference in New Issue View Git Blame Copy Permalink