Add Zira spoken button confirmations and opt-in G726 speech tests

This commit is contained in:
mikecarper committed 2026-10-03 03:03:02 -07:00
1 parent 5833ac2f1f
commit 8519638367
33 files changed
+11977 -30

No files matched your search

+854
View File
@@ -0,0 +1,854 @@
"""Run production spoken-button mappings, courtesy audio, and UI transactions natively."""
from pathlib import Path
import unittest
import test_gps_voice_buttons as gps_tests
from test_radio_receive_contract import method
ROOT = Path(__file__).resolve().parents[1]
PROMPTS = r'''
#include <cassert>
#include <initializer_list>
#define PIN_BUZZER 25
#define PIN_BUZZER_EN 37
#define MESH_BUTTON_AUDIO_HIL 1
#define HAS_DRV2605 1
#define MESH_GPS_VOICE 1
#include <helpers/ui/buzzer.cpp>
uint32_t now_ms=0;
int main(){
using namespace mesh::audio;
struct Mapping {ButtonVoicePrompt prompt;VoiceClip clip;const char* tone;};
const Mapping mappings[]={
{ButtonVoicePrompt::Ready,readyClip,nullptr},
{ButtonVoicePrompt::NotificationCleared,notificationClearedClip,nullptr},
{ButtonVoicePrompt::AdvertQueued,advertQueuedClip,nullptr},
{ButtonVoicePrompt::AdvertFailed,advertFailedClip,nullptr},
{ButtonVoicePrompt::SoundOn,soundOnClip,nullptr},
{ButtonVoicePrompt::SoundOff,soundOffClip,nullptr},
{ButtonVoicePrompt::GpsOn,gpsOnClip,"Startup:d=4,o=5,b=160:16c6,16e6,8g6"},
{ButtonVoicePrompt::GpsOff,gpsOffClip,"Shutdown:d=4,o=5,b=100:8g5,16e5,16c5"},
{ButtonVoicePrompt::ActionFailed,actionFailedClip,nullptr},
{ButtonVoicePrompt::UsbSetup,usbSetupClip,nullptr},
{ButtonVoicePrompt::ShuttingDown,shuttingDownClip,"Shutdown:d=4,o=5,b=100:8g5,16e5,16c5"},
{ButtonVoicePrompt::Restarting,restartingClip,"Shutdown:d=4,o=5,b=100:8g5,16e5,16c5"},
#ifdef HAS_DRV2605
{ButtonVoicePrompt::AlertsSoundVibration,alertsSoundVibrationClip,nullptr},
{ButtonVoicePrompt::AlertsSoundOnly,alertsSoundOnlyClip,nullptr},
{ButtonVoicePrompt::AlertsVibrationOnly,alertsVibrationOnlyClip,nullptr},
{ButtonVoicePrompt::AlertsSilent,alertsSilentClip,nullptr},
#endif
};
genericBuzzer buzzer;buzzer.begin();
auto finish=[&]{
assert(!registers.ENABLE);
rtttl::finish();buzzer.loop();
now_ms+=99;buzzer.loop();assert(!registers.ENABLE);
++now_ms;buzzer.loop();
assert(HwPWM3.owned&&HwPWM3.pin&&HwPWM3.bound_pin==PIN_BUZZER);
unsigned sequence=0,events=0;
while(registers.ENABLE){
assert(++events<2000);
registers.EVENTS_SEQEND[sequence]=1;PWM3_IRQHandler();sequence^=1;
}
buzzer.loop();assert(!buzzer.isPlaying());
assert(!HwPWM3.owned&&!HwPWM3.pin&&!irq_enabled);
assert(pin_states[PIN_BUZZER]==LOW);
};
for(const auto& entry:mappings){
auto actual=buttonVoiceClip(entry.prompt);
assert(actual.data==entry.clip.data&&actual.bytes==entry.clip.bytes);
assert(actual.samples==entry.clip.samples&&actual.sample_rate==entry.clip.sample_rate);
assert(actual.data&&actual.bytes==(actual.samples+1)/2);
assert(actual.sample_rate==8000||actual.sample_rate==16000);
VoiceDecoder decoder;decoder.begin(actual);
unsigned count=0,nonzero=0;
while(!decoder.done()){nonzero+=decoder.next()!=0;++count;}
assert(count==actual.samples&&nonzero>count/2);
auto started=gps_voice.started(),completed=gps_voice.completed(),tones=rtttl::begins;
assert(buzzer.speakButton(entry.prompt)&&buzzer.isPlaying());
if(entry.tone){assert(rtttl::melody==entry.tone&&rtttl::begins==tones+1);}
else {assert(rtttl::begins==tones&&!rtttl::isPlaying());}
finish();
assert(gps_voice.started()==started+1&&gps_voice.completed()==completed+1);
}
// Invalid values must reject without preempting a valid tone or active voice.
const auto invalid=static_cast<ButtonVoicePrompt>(255);
assert(!buttonVoiceClip(invalid).data);
assert(buzzer.speakGps(true));
auto tones=rtttl::begins,started=gps_voice.started();
assert(!buzzer.speakButton(invalid));
assert(buzzer.isPlaying()&&rtttl::isPlaying()&&rtttl::begins==tones);
#ifndef HAS_DRV2605
for(auto prompt:{ButtonVoicePrompt::AlertsSoundVibration,ButtonVoicePrompt::AlertsSoundOnly,
ButtonVoicePrompt::AlertsVibrationOnly,ButtonVoicePrompt::AlertsSilent}){
assert(!buttonVoiceClip(prompt).data&&!buzzer.speakButton(prompt,true));
assert(buzzer.isPlaying()&&rtttl::isPlaying()&&rtttl::begins==tones);
}
#endif
finish();
assert(gps_voice.started()==started+1);
assert(buzzer.speakButton(ButtonVoicePrompt::Ready));
buzzer.loop();now_ms+=100;buzzer.loop();
assert(registers.ENABLE&&buzzer.isPlaying());
started=gps_voice.started();
assert(!buzzer.speakButton(invalid,true));
assert(registers.ENABLE&&gps_voice.started()==started);
#ifndef HAS_DRV2605
for(auto prompt:{ButtonVoicePrompt::AlertsSoundVibration,ButtonVoicePrompt::AlertsSoundOnly,
ButtonVoicePrompt::AlertsVibrationOnly,ButtonVoicePrompt::AlertsSilent}){
assert(!buzzer.speakButton(prompt,true));
assert(registers.ENABLE&&gps_voice.started()==started);
}
#endif
unsigned sequence=0,events=0;
while(registers.ENABLE){
assert(++events<2000);registers.EVENTS_SEQEND[sequence]=1;
PWM3_IRQHandler();sequence^=1;
}
buzzer.loop();assert(!buzzer.isPlaying());
// A busy PWM fails the deferred voice start, without leaking ownership.
HwPWM3.reject=true;
assert(buzzer.speakButton(ButtonVoicePrompt::Ready));
buzzer.loop();now_ms+=100;buzzer.loop();
assert(!buzzer.isPlaying()&&!registers.ENABLE&&!HwPWM3.owned&&!irq_enabled);
}
'''
COURTESY = r'''
#include <cassert>
#define PIN_BUZZER 25
#define PIN_BUZZER_EN 37
#define MESH_BUTTON_AUDIO_HIL 1
#define MESH_GPS_VOICE 1
#include <helpers/ui/buzzer.cpp>
uint32_t now_ms=0;
int main(){
genericBuzzer buzzer;buzzer.begin();buzzer.quiet(true);
assert(buzzer.isQuiet()&&pin_states[PIN_BUZZER_EN]==LOW);
auto before=gps_voice.started();
assert(buzzer.speakButton(ButtonVoicePrompt::Ready));
assert(!buzzer.isPlaying()&&gps_voice.started()==before);
assert(pin_states[PIN_BUZZER_EN]==LOW);
// The explicit sound-off confirmation is one-shot: it never unmutes alerts.
assert(buzzer.speakButton(ButtonVoicePrompt::SoundOff,true));
assert(buzzer.isQuiet()&&buzzer.isPlaying()&&pin_states[PIN_BUZZER_EN]==HIGH);
buzzer.loop();now_ms+=100;buzzer.loop();
assert(registers.ENABLE&&gps_voice.started()==before+1);
unsigned sequence=0,events=0;
while(registers.ENABLE){
assert(++events<2000);registers.EVENTS_SEQEND[sequence]=1;
PWM3_IRQHandler();sequence^=1;
}
assert(buzzer.isPlaying()); // IRQ cleanup is deferred to the foreground.
buzzer.loop();
assert(buzzer.isQuiet()&&!buzzer.isPlaying()&&pin_states[PIN_BUZZER_EN]==LOW);
assert(pin_states[PIN_BUZZER]==LOW&&!HwPWM3.owned&&!HwPWM3.pin&&!irq_enabled);
auto tones=rtttl::begins;
buzzer.play("normal:d=4,o=5,b=120:c");
assert(!buzzer.isPlaying()&&rtttl::begins==tones&&pin_states[PIN_BUZZER_EN]==LOW);
// Every cancellation path releases the courtesy output without changing mute.
assert(buzzer.speakButton(ButtonVoicePrompt::SoundOff,true));
buzzer.stop();assert(buzzer.isQuiet()&&!buzzer.isPlaying());
assert(pin_states[PIN_BUZZER_EN]==LOW&&pin_states[PIN_BUZZER]==LOW);
assert(buzzer.speakButton(ButtonVoicePrompt::SoundOff,true));
buzzer.quiet(true);assert(!buzzer.isPlaying()&&pin_states[PIN_BUZZER_EN]==LOW);
// Muted ordinary notifications must preserve queued and active courtesy voice.
assert(buzzer.speakButton(ButtonVoicePrompt::SoundOff,true));
before=gps_voice.started();auto cancelled=gps_voice.cancelled();tones=rtttl::begins;
buzzer.play("normal:d=4,o=5,b=120:c");
assert(buzzer.isQuiet()&&buzzer.isPlaying()&&pin_states[PIN_BUZZER_EN]==HIGH);
assert(!registers.ENABLE&&gps_voice.started()==before&&rtttl::begins==tones);
buzzer.loop();now_ms+=100;buzzer.loop();
assert(registers.ENABLE&&gps_voice.started()==before+1);
buzzer.play("normal:d=4,o=5,b=120:c");
assert(buzzer.isQuiet()&&buzzer.isPlaying()&&registers.ENABLE);
assert(gps_voice.cancelled()==cancelled&&rtttl::begins==tones&&pin_states[PIN_BUZZER_EN]==HIGH);
// Ordinary sound still preempts voice as soon as the user actually unmutes.
buzzer.quiet(false);buzzer.play("normal:d=4,o=5,b=120:c");
assert(!registers.ENABLE&&gps_voice.cancelled()==cancelled+1);
assert(rtttl::isPlaying()&&rtttl::begins==tones+1&&!HwPWM3.owned&&!irq_enabled);
rtttl::finish();buzzer.loop();assert(!buzzer.isPlaying());buzzer.quiet(true);
// Starting a courtesy direction tone also respects cancellation in its gap.
assert(buzzer.speakButton(ButtonVoicePrompt::GpsOff,true));
assert(buzzer.isQuiet()&&rtttl::isPlaying()&&pin_states[PIN_BUZZER_EN]==HIGH);
rtttl::finish();buzzer.loop();buzzer.stop();
now_ms+=1000;buzzer.loop();
assert(!buzzer.isPlaying()&&pin_states[PIN_BUZZER_EN]==LOW&&!registers.ENABLE);
// An unavailable player must restore the sleeping amplifier immediately.
HwPWM3.reject=true;
assert(buzzer.speakButton(ButtonVoicePrompt::SoundOff,true));
buzzer.loop();now_ms+=100;buzzer.loop();
assert(buzzer.isQuiet()&&!buzzer.isPlaying()&&pin_states[PIN_BUZZER_EN]==LOW);
}
'''
LEGACY_ORDINARY_PLAY = r'''
#include <cassert>
#define PIN_BUZZER 25
#define PIN_BUZZER_EN 37
#define MESH_GPS_VOICE 0
#include <helpers/ui/buzzer.cpp>
uint32_t now_ms=0;
int main(){
genericBuzzer buzzer;buzzer.begin();buzzer.quiet(true);
assert(buzzer.speakButton(ButtonVoicePrompt::GpsOff,true)&&rtttl::isPlaying());
auto tones=rtttl::begins;
// Non-voice builds retain the original cancel-before-quiet behavior.
buzzer.play("normal:d=4,o=5,b=120:c");
assert(buzzer.isQuiet()&&!buzzer.isPlaying()&&rtttl::begins==tones);
assert(pin_states[PIN_BUZZER_EN]==LOW);
buzzer.quiet(false);assert(buzzer.speakGps(true));tones=rtttl::begins;
buzzer.play("normal:d=4,o=5,b=120:c");
assert(!buzzer.isQuiet()&&buzzer.isPlaying()&&rtttl::begins==tones+1);
assert(rtttl::melody=="normal:d=4,o=5,b=120:c");
}
'''
UI_ACTIONS = r'''
#include <cassert>
#include <algorithm>
#include <string>
#include <vector>
#include <helpers/ui/ButtonVoice.h>
#include "examples/companion_radio/ui-orig/Button.h"
#define PIN_BUZZER 25
#define MESH_GPS_VOICE 1
#define ENV_INCLUDE_GPS 1
#define MESH_DEBUG_PRINTLN(...) ((void)0)
uint32_t now_ms=0;
static std::vector<std::string> events;
enum class UIEventType{contactMessage,channelMessage,ack,roomMessage,newContactMessage,none};
namespace mesh {
namespace notify {enum {Sound=1};}
namespace ui {
enum class DisplayWake {Button};
inline uint8_t radioProfileDisplayPageCount(bool,uint8_t){return 4;}
}
}
struct Display {
bool on=true;unsigned wakes=0;
bool isOn(){return on;}
void wake(mesh::ui::DisplayWake){++wakes;on=true;}
};
static uint8_t radioProfileStatusPageCount(Display&,bool){return 2;}
struct CompanionNodePrefs {uint8_t gps_enabled=0,buzzer_quiet=0;};
struct SensorManager {
bool enabled=false,available=true,reject=false;
bool setSettingValue(const char*,const char* value){
events.push_back("gps");
if(!available||reject)return false;
enabled=value[0]=='1';return true;
}
const char* getSettingByKey(const char*){return available?(enabled?"1":"0"):nullptr;}
};
static SensorManager sensors;
struct MyMesh {
CompanionNodePrefs _prefs;
bool save_ok=true,advert_ok=true,notification_active=false,output_muted=false,dual_radio=false;
unsigned saves=0,adverts=0,notification_buttons=0,mute_calls=0,cli_entries=0;
std::vector<bool> saved_quiet;
bool savePrefs(){++saves;events.push_back("save");saved_quiet.push_back(_prefs.buzzer_quiet);return save_ok;}
template<class T> bool savePreference(T& p,T v){T old=p;p=v;if(savePrefs())return true;p=old;return false;}
void applyGpsPrefs(){sensors.setSettingValue("gps",_prefs.gps_enabled?"1":"0");}
bool setGpsEnabled(bool);
bool advert(){++adverts;events.push_back("advert");return advert_ok;}
bool notificationButton(){++notification_buttons;bool cleared=notification_active;notification_active=false;return cleared;}
void setNotificationOutputMute(int,bool muted){++mute_calls;output_muted=muted;events.push_back("mute");}
bool isDualRadioActive(){return dual_radio;}
void enterCLIRescue(){++cli_entries;events.push_back("cli");}
};
static MyMesh the_mesh;
struct Buzzer {
bool muted=false,playing=false,accept=true;
unsigned plays=0,notification_plays=0,stops=0,notification_stops=0,shutdowns=0;
std::vector<ButtonVoicePrompt> prompts;
std::vector<bool> courtesy;
bool speakButton(ButtonVoicePrompt prompt,bool allowQuiet=false){
events.push_back("speak");prompts.push_back(prompt);courtesy.push_back(allowQuiet);
playing=accept&&(!muted||allowQuiet);return accept;
}
bool speakGps(bool value){return speakButton(value?ButtonVoicePrompt::GpsOn:ButtonVoicePrompt::GpsOff);}
void quiet(bool q){muted=q;events.push_back("quiet");}
bool isQuiet(){return muted;}
bool isPlaying(){return playing;}
void stop(){++stops;playing=false;events.push_back("stop");}
void stopNotification(){++notification_stops;playing=false;events.push_back("notification_stop");}
void shutdown(){++shutdowns;playing=!muted;}
void play(const char*){++plays;}
void playNotification(const char*){++notification_plays;}
void loop(){}
};
struct Board {
unsigned offs=0,reboots=0;
void powerOff(){++offs;events.push_back("off");}
void reboot(){++reboots;events.push_back("reboot");}
};
struct UITask {
Display* _display=nullptr;SensorManager* _sensors=&sensors;
CompanionNodePrefs* _node_prefs=&the_mesh._prefs;
Buzzer buzzer;Board board;Board* _board=&board;
char _alert[80]={},_origin[80]={},_msg[80]={};
bool _need_refresh=false,_displayWasOn=false,_button_notification_cleared=false;
bool _shutdown_pending=false,_shutdown_restart=false;
uint32_t _shutdown_started_at=0,ui_started_at=0;
uint8_t _radio_profile_display_page=0,_notification_outputs=mesh::notify::Sound;
bool guard_ok=true,guard_reject_drain=false;unsigned guards=0,preview_clears=0;
bool prepareForShutdown(){++guards;events.push_back("guard");return guard_ok&&(!guard_reject_drain||guards==1);}
bool shouldPlayMessageTone()const{return true;}
bool isPairingScreenActive()const{return false;}
void resetRadioProfileDisplayPage(){_radio_profile_display_page=0;}
void clearMsgPreview(){++preview_clears;_origin[0]=_msg[0]=0;}
void notify(UIEventType);void notificationMelody(const char*);
void handleButtonAnyPress();void handleButtonShortPress();
void handleButtonDoublePress();void handleButtonTriplePress();
void handleButtonQuadruplePress();void handleButtonLongPress();
void shutdown(bool restart=false);void servicePendingShutdown();
};
@METHODS@
int main(){
auto reset=[](){the_mesh=MyMesh{};sensors=SensorManager{};events.clear();now_ms=10000;};
auto last=[](const UITask& ui,ButtonVoicePrompt expected){assert(!ui.buzzer.prompts.empty()&&ui.buzzer.prompts.back()==expected);};
reset();UITask single;
single.handleButtonAnyPress();single.handleButtonShortPress();
last(single,ButtonVoicePrompt::Ready);
the_mesh.notification_active=true;
single.handleButtonAnyPress();single.handleButtonAnyPress(); // The first click really acknowledged it.
single.handleButtonShortPress();last(single,ButtonVoicePrompt::NotificationCleared);
single.handleButtonAnyPress();single.handleButtonShortPress();last(single,ButtonVoicePrompt::Ready);
assert(single.buzzer.prompts.size()==3);
// Multi-click actions consume the acknowledgement latch instead of leaking it to a later single.
the_mesh.notification_active=true;
single.handleButtonAnyPress();single.handleButtonDoublePress();last(single,ButtonVoicePrompt::AdvertQueued);
single.handleButtonAnyPress();single.handleButtonShortPress();last(single,ButtonVoicePrompt::Ready);
// An attached display retains its existing preview/wake behavior, with no Ready chatter.
Display display;single._display=&display;single._displayWasOn=true;
strcpy(single._origin,"sender");strcpy(single._msg,"message");
auto prompt_count=single.buzzer.prompts.size();
single.handleButtonAnyPress();single.handleButtonShortPress();
assert(single.preview_clears==1&&single.buzzer.prompts.size()==prompt_count&&display.wakes==1);
reset();UITask advert;
advert.handleButtonDoublePress();last(advert,ButtonVoicePrompt::AdvertQueued);
assert((events==std::vector<std::string>{"advert","speak"}));
the_mesh.advert_ok=false;events.clear();
advert.handleButtonDoublePress();last(advert,ButtonVoicePrompt::AdvertFailed);
assert(the_mesh.adverts==2&&strstr(advert._alert,"failed"));
assert((events==std::vector<std::string>{"advert","speak"}));
reset();UITask mute;
mute.handleButtonTriplePress();last(mute,ButtonVoicePrompt::SoundOff);
assert(mute.buzzer.muted&&the_mesh._prefs.buzzer_quiet&&the_mesh.output_muted);
assert(mute.buzzer.courtesy.back()&&mute.buzzer.playing);
assert((events==std::vector<std::string>{"save","quiet","mute","speak"}));
events.clear();mute.handleButtonTriplePress();last(mute,ButtonVoicePrompt::SoundOn);
assert(!mute.buzzer.muted&&!the_mesh._prefs.buzzer_quiet&&!the_mesh.output_muted);
assert(mute.buzzer.courtesy.back()&&the_mesh.saves==2&&the_mesh.mute_calls==2);
// Failed commits preserve runtime quiet, stored preference, and output routing separately.
for(bool old_quiet:{false,true}){
mute.buzzer.muted=old_quiet;the_mesh._prefs.buzzer_quiet=old_quiet;
the_mesh.output_muted=old_quiet;the_mesh.save_ok=false;events.clear();
auto mutes=the_mesh.mute_calls;
mute.handleButtonTriplePress();last(mute,ButtonVoicePrompt::ActionFailed);
assert(mute.buzzer.muted==old_quiet&&the_mesh._prefs.buzzer_quiet==old_quiet);
assert(the_mesh.output_muted==old_quiet&&the_mesh.mute_calls==mutes);
assert(mute.buzzer.courtesy.back()&&mute.buzzer.playing&&strstr(mute._alert,"failed"));
assert(std::find(events.begin(),events.end(),"mute")==events.end());
}
// Even an initially inconsistent runtime/preference pair is not rewritten on failure.
mute.buzzer.muted=false;the_mesh._prefs.buzzer_quiet=1;
mute.handleButtonTriplePress();assert(!mute.buzzer.muted&&the_mesh._prefs.buzzer_quiet==1);
reset();UITask gps;
#if ENV_INCLUDE_GPS == 1
gps.handleButtonQuadruplePress();last(gps,ButtonVoicePrompt::GpsOn);
assert(sensors.enabled&&the_mesh._prefs.gps_enabled&&the_mesh.saves==1);
assert((events==std::vector<std::string>{"gps","save","speak"}));
events.clear();gps.handleButtonQuadruplePress();last(gps,ButtonVoicePrompt::GpsOff);
assert(!sensors.enabled&&!the_mesh._prefs.gps_enabled&&the_mesh.saves==2);
the_mesh.save_ok=false;events.clear();
gps.handleButtonQuadruplePress();last(gps,ButtonVoicePrompt::ActionFailed);
assert(!sensors.enabled&&!the_mesh._prefs.gps_enabled&&strstr(gps._alert,"failed"));
assert((events==std::vector<std::string>{"gps","save","gps","speak"}));
the_mesh.save_ok=true;sensors.reject=true;
auto saves=the_mesh.saves;gps.handleButtonQuadruplePress();last(gps,ButtonVoicePrompt::ActionFailed);
assert(the_mesh.saves==saves&&!sensors.enabled&&!the_mesh._prefs.gps_enabled);
sensors.reject=false;sensors.available=false;
gps.handleButtonQuadruplePress();last(gps,ButtonVoicePrompt::ActionFailed);
assert(the_mesh.saves==saves);
#else
gps.handleButtonQuadruplePress();last(gps,ButtonVoicePrompt::ActionFailed);
assert(the_mesh.saves==0&&!sensors.enabled&&!the_mesh._prefs.gps_enabled);
#endif
gps._sensors=nullptr;gps.handleButtonQuadruplePress();last(gps,ButtonVoicePrompt::ActionFailed);
reset();UITask early;
now_ms=7999;early.handleButtonLongPress();last(early,ButtonVoicePrompt::UsbSetup);
assert(the_mesh.cli_entries==1&&early.guards==0&&!early._shutdown_pending);
assert((events==std::vector<std::string>{"cli","speak"}));
// Uptime arithmetic is unsigned even when boot began just before clock wrap.
early.ui_started_at=0xfffffff0;now_ms=7982;
early.handleButtonLongPress();last(early,ButtonVoicePrompt::UsbSetup);
assert(the_mesh.cli_entries==2);
reset();UITask rejected;rejected.guard_ok=false;
rejected.shutdown();last(rejected,ButtonVoicePrompt::ActionFailed);
assert(!rejected._shutdown_pending&&!rejected.board.offs&&!rejected.board.reboots);
rejected.servicePendingShutdown();assert(rejected.guards==1&&!rejected.board.offs);
reset();UITask shutdown;
now_ms=8000;shutdown.handleButtonLongPress();last(shutdown,ButtonVoicePrompt::ShuttingDown);
assert(shutdown._shutdown_pending&&shutdown.guards==1&&!shutdown.board.offs);
assert((events==std::vector<std::string>{"guard","speak"}));
// Shutdown callbacks return immediately; draining is foreground service, not a busy wait.
assert(now_ms==8000);
auto before=shutdown.buzzer.prompts.size();
auto adverts=the_mesh.adverts,shutdown_saves=the_mesh.saves;
shutdown.handleButtonAnyPress();shutdown.handleButtonShortPress();shutdown.handleButtonDoublePress();
shutdown.handleButtonTriplePress();shutdown.handleButtonQuadruplePress();shutdown.handleButtonLongPress();
shutdown.shutdown(true);shutdown.notify(UIEventType::ack);shutdown.notificationMelody("notification");
assert(shutdown.buzzer.prompts.size()==before&&the_mesh.adverts==adverts&&the_mesh.saves==shutdown_saves);
assert(!shutdown.buzzer.plays&&!shutdown.buzzer.notification_plays&&shutdown.guards==1);
now_ms+=4999;shutdown.servicePendingShutdown();assert(!shutdown.board.offs&&shutdown._shutdown_pending);
shutdown.buzzer.playing=false;shutdown.servicePendingShutdown();
assert(shutdown.board.offs==1&&!shutdown._shutdown_pending&&shutdown.guards==2&&shutdown.buzzer.stops==1);
shutdown.servicePendingShutdown();assert(shutdown.board.offs==1);
reset();UITask restart;
now_ms=0xfffffff0;restart.shutdown(true);last(restart,ButtonVoicePrompt::Restarting);
now_ms+=4999;restart.servicePendingShutdown();assert(!restart.board.reboots&&restart._shutdown_pending);
++now_ms;restart.servicePendingShutdown();
assert(restart.board.reboots==1&&!restart.board.offs&&!restart._shutdown_pending&&!restart.buzzer.playing);
reset();UITask drain_rejected;drain_rejected.guard_reject_drain=true;
drain_rejected.shutdown();drain_rejected.buzzer.playing=false;drain_rejected.servicePendingShutdown();
last(drain_rejected,ButtonVoicePrompt::ActionFailed);
assert(!drain_rejected._shutdown_pending&&!drain_rejected.board.offs&&drain_rejected.guards==2);
// Use the actual click decoder to establish that holding does not turn into a Ready/advert.
reset();UITask held;now_ms=100;
Button button(6,HIGH);button.begin();
static_assert(BUTTON_LONG_PRESS_TIME_MS==3000,"three-second hold");
button.onAnyPress([&]{held.handleButtonAnyPress();});
button.onShortPress([&]{held.handleButtonShortPress();});
button.onDoublePress([&]{held.handleButtonDoublePress();});
button.onLongPress([&]{held.handleButtonLongPress();});
assert(button.injectPresses(1,3500));
for(unsigned i=0;i<500;++i){now_ms+=10;button.update();}
assert(held.buzzer.prompts.size()==1&&the_mesh.adverts==0&&the_mesh.cli_entries==1);
last(held,ButtonVoicePrompt::UsbSetup);
}
'''
NOTIFICATION_ACK = r'''
#include <cassert>
#include <initializer_list>
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wmisleading-indentation"
#include <helpers/CompanionNotificationPolicy.h>
#pragma GCC diagnostic pop
uint32_t now_ms=0;
using namespace mesh::notify;
struct Probe : Sink,Store {
unsigned stops=0;
uint8_t capabilities()const override{return Sound;}
bool gpioAvailable(uint8_t)const override{return false;}
void pulse(Output,bool,int8_t)override{}
void melody(const char* value)override{if(!value)++stops;}
void screen(int8_t)override{}
bool save(const Settings&)override{return true;}
};
struct MyMesh {
Controller _notifications;
explicit MyMesh(Probe& probe):_notifications(probe,probe){}
@WRAPPER@
};
int main(){
Probe probe;MyMesh mesh(probe);
assert(!mesh.notificationButton()&&probe.stops==0);
auto& settings=mesh._notifications.settings();settings.enabled=1;settings.outputs=Sound;
auto& rule=settings.rules[0];rule.used=1;
assert(setField(rule,FSound,"test:d=4,o=5,b=120:c"));
assert(setField(rule,FStop,"button"));
assert(mesh._notifications.message(nullptr,nullptr,false,100));
assert(mesh.notificationButton()&&!mesh._notifications.active()&&probe.stops==1);
assert(!mesh.notificationButton()&&probe.stops==1);
// An active alert with another stop policy was not actually cleared by input.
for(const char* policy:{"connected","never"}){
assert(setField(rule,FStop,policy));
assert(mesh._notifications.message(nullptr,nullptr,false,200));
auto stops=probe.stops;
assert(!mesh.notificationButton()&&mesh._notifications.active()&&probe.stops==stops);
mesh._notifications.stop();
}
}
'''
NOTIFICATION_MELODY = r'''
#include <cassert>
#include <cstdint>
#define PIN_BUZZER 25
#define MESH_GPS_VOICE @VOICE@
uint32_t now_ms=0;
struct Buzzer {
bool muted=true,playing=true,control_voice=true;unsigned melodies=0,stops=0,notification_stops=0;
bool isQuiet(){return muted;}
void playNotification(const char*){++melodies;playing=true;control_voice=false;}
void stop(){++stops;playing=false;}
void stopNotification(){++notification_stops;if(!MESH_GPS_VOICE||!control_voice)playing=false;}
};
struct UITask {
Buzzer buzzer;bool _shutdown_pending=false;
void notificationMelody(const char*);
};
@METHOD@
int main(){
UITask ui;
// A muted programmable alert must not preempt the final courtesy confirmation.
ui.notificationMelody("ordinary");
#if MESH_GPS_VOICE
assert(ui.buzzer.melodies==0&&ui.buzzer.playing&&ui.buzzer.stops==0);
#else
assert(ui.buzzer.melodies==1&&ui.buzzer.playing&&ui.buzzer.stops==0);
#endif
// Background stop requests retain control speech while quiet.
ui.notificationMelody(nullptr);
assert(ui.buzzer.stops==0&&ui.buzzer.notification_stops==1);
#if MESH_GPS_VOICE
assert(ui.buzzer.playing);
#else
assert(!ui.buzzer.playing);
#endif
ui.buzzer.muted=false;ui.notificationMelody("ordinary");
#if MESH_GPS_VOICE
assert(ui.buzzer.melodies==1&&ui.buzzer.playing);
#else
assert(ui.buzzer.melodies==2&&ui.buzzer.playing);
#endif
ui._shutdown_pending=true;
ui.notificationMelody("ordinary");ui.notificationMelody(nullptr);
#if MESH_GPS_VOICE
assert(ui.buzzer.melodies==1&&ui.buzzer.stops==0&&ui.buzzer.notification_stops==1&&ui.buzzer.playing);
#else
assert(ui.buzzer.melodies==3&&ui.buzzer.stops==0&&ui.buzzer.notification_stops==2&&!ui.buzzer.playing);
#endif
}
'''
NOTIFICATION_STOPS = r'''
#include <cassert>
#include <initializer_list>
#define PIN_BUZZER 25
#define PIN_BUZZER_EN 37
#define MESH_BUTTON_AUDIO_HIL 1
#define MESH_GPS_VOICE @VOICE@
#include <helpers/ui/buzzer.cpp>
uint32_t now_ms=0;
int main(){
genericBuzzer buzzer;buzzer.begin();
buzzer.playNotification("ordinary:d=4,o=5,b=120:c");
assert(buzzer.isPlaying()&&rtttl::isPlaying());
buzzer.stopNotification();assert(!buzzer.isPlaying()&&!rtttl::isPlaying());
assert(buzzer.speakGps(true)&&rtttl::isPlaying());
buzzer.stopNotification();
#if MESH_GPS_VOICE
// A stale notification stop cannot cut off the direction tone or its gap.
assert(buzzer.isPlaying()&&rtttl::isPlaying()&&!registers.ENABLE);
auto starts=gps_voice.started(),cancelled=gps_voice.cancelled();
rtttl::finish();buzzer.loop();
assert(buzzer.isPlaying()&&!rtttl::isPlaying()&&!registers.ENABLE);
buzzer.stopNotification();assert(buzzer.isPlaying());
now_ms+=99;buzzer.loop();buzzer.stopNotification();
assert(buzzer.isPlaying()&&!registers.ENABLE);
++now_ms;buzzer.loop();assert(registers.ENABLE&&gps_voice.started()==starts+1);
buzzer.stopNotification();
assert(registers.ENABLE&&HwPWM3.owned&&buzzer.isPlaying()&&gps_voice.cancelled()==cancelled);
auto drain=[&]{
unsigned sequence=0,events=0;
while(registers.ENABLE){
assert(++events<2000);buzzer.stopNotification();
assert(registers.ENABLE&&HwPWM3.owned&&gps_voice.cancelled()==cancelled);
registers.EVENTS_SEQEND[sequence]=1;PWM3_IRQHandler();sequence^=1;
}
// Completed DMA still owns its hardware until foreground cleanup.
assert(buzzer.isPlaying()&&HwPWM3.owned);
buzzer.stopNotification();assert(HwPWM3.owned&&gps_voice.cancelled()==cancelled);
buzzer.loop();assert(!buzzer.isPlaying()&&!HwPWM3.owned&&!HwPWM3.pin&&!irq_enabled);
};
drain();
// The same protection covers a courtesy confirmation while alerts are muted.
buzzer.quiet(true);assert(buzzer.speakButton(ButtonVoicePrompt::ActionFailed,true));
buzzer.stopNotification();assert(buzzer.isPlaying()&&buzzer.isQuiet());
buzzer.loop();now_ms+=100;buzzer.loop();assert(registers.ENABLE);
drain();
assert(pin_states[PIN_BUZZER_EN]==LOW&&pin_states[PIN_BUZZER]==LOW);
#else
// Without voice, this API remains an ordinary tone stop.
assert(!buzzer.isPlaying()&&!rtttl::isPlaying());
buzzer.quiet(true);
#endif
buzzer.playNotification("ordinary:d=4,o=5,b=120:c");
assert(buzzer.isPlaying()&&pin_states[PIN_BUZZER_EN]==HIGH);
buzzer.stopNotification();
assert(!buzzer.isPlaying()&&!rtttl::isPlaying()&&buzzer.isQuiet());
assert(pin_states[PIN_BUZZER_EN]==LOW);
}
'''
X1_ALERT_MODES = r'''
#include <cassert>
#include <cstdio>
#include <cstring>
#include <string>
#include <vector>
#include <helpers/ui/ButtonVoice.h>
#define PIN_BUZZER 25
#define HAS_DRV2605 1
#define MESH_GPS_VOICE @VOICE@
#define MESH_DEBUG_PRINTLN(...) ((void)0)
uint32_t now_ms=0;
static std::vector<std::string> mode_events;
enum class UIEventType{ack};
namespace mesh{namespace notify{enum{Sound=1,Vibration=2};}}
struct CompanionNodePrefs{uint8_t buzzer_quiet=0,vibe_quiet=0;};
struct MyMesh {
CompanionNodePrefs _prefs;
bool save_ok=true,mask_save_ok=true;
unsigned saves=0,mutes=0;
bool sound_muted=false,vibration_muted=false;
std::vector<unsigned> saved_modes;
bool savePrefs(){
mode_events.push_back("save");
++saves;saved_modes.push_back((_prefs.buzzer_quiet?2:0)|(_prefs.vibe_quiet?1:0));
return save_ok;
}
void setNotificationOutputMute(int output,bool quiet){
++mutes;
mode_events.push_back("mask");
if(!mask_save_ok)return;
if(output==mesh::notify::Sound)sound_muted=quiet;
else if(output==mesh::notify::Vibration)vibration_muted=quiet;
else assert(false);
}
};
static MyMesh the_mesh;
struct Buzzer {
bool muted=false;
std::vector<ButtonVoicePrompt> prompts;
std::vector<bool> courtesy;
void quiet(bool value){muted=value;}
bool isQuiet(){return muted;}
bool speakButton(ButtonVoicePrompt prompt,bool allowQuiet=false){
mode_events.push_back("voice");
prompts.push_back(prompt);courtesy.push_back(allowQuiet);return true;
}
};
struct Vibration {
bool muted=false,active=false;unsigned triggers=0,stops=0,pulses=0;
void quiet(bool value){muted=value;}
bool isQuiet(){return muted;}
void trigger(bool confirmation=false){assert(confirmation);++triggers;}
void stop(){++stops;active=false;mode_events.push_back("stop");}
void pulse(bool on){++pulses;active=on;}
};
struct UITask {
CompanionNodePrefs* _node_prefs=&the_mesh._prefs;
Buzzer buzzer;Vibration vibration;
bool _shutdown_pending=false,_button_notification_cleared=false,_need_refresh=false;
char _alert[80]={};unsigned acknowledgements=0;
void notify(UIEventType){++acknowledgements;}
void handleButtonTriplePress();
void notificationVibration(bool);
};
@METHOD@
@VIBRATION_METHOD@
int main(){
UITask ui;
const ButtonVoicePrompt expected[]={
ButtonVoicePrompt::AlertsSoundVibration,ButtonVoicePrompt::AlertsSoundOnly,
ButtonVoicePrompt::AlertsVibrationOnly,ButtonVoicePrompt::AlertsSilent,
};
unsigned vibration_triggers=0,vibration_stops=0;
for(unsigned step=1;step<=8;++step){
const auto mode=step&3;
ui._button_notification_cleared=true;
auto saves=the_mesh.saves;
mode_events.clear();
ui.vibration.active=true;
ui.handleButtonTriplePress();
assert(the_mesh.saves==saves+1&&the_mesh.saved_modes.back()==mode);
assert(the_mesh._prefs.buzzer_quiet==!!(mode&2));
assert(the_mesh._prefs.vibe_quiet==!!(mode&1));
assert(ui.buzzer.muted==!!(mode&2)&&ui.vibration.muted==!!(mode&1));
assert(the_mesh.sound_muted==!!(mode&2)&&the_mesh.vibration_muted==!!(mode&1));
if(!(mode&1))++vibration_triggers;
assert(ui.vibration.triggers==vibration_triggers&&ui._need_refresh);
#if MESH_GPS_VOICE
if(mode&1){
++vibration_stops;assert(!ui.vibration.active);
assert(mode_events.size()>=2&&mode_events[0]=="save"&&mode_events[1]=="stop");
}
assert(ui.vibration.stops==vibration_stops);
assert(ui.buzzer.prompts.size()==step&&ui.buzzer.prompts.back()==expected[mode]);
// Selecting silent or vibration-only gets a single final confirmation,
// without unmuting saved alerts or notification routing.
assert(ui.buzzer.courtesy.back()&&!ui._button_notification_cleared);
assert(ui.acknowledgements==0);
#else
assert(ui.buzzer.prompts.empty());
#endif
}
#if MESH_GPS_VOICE
// Each possible persisted mode is restored on failure. Deliberately divergent
// runtime and routing states are preserved too; a failed save is not a toggle.
the_mesh.save_ok=false;
for(unsigned mode=0;mode<4;++mode){
the_mesh._prefs.buzzer_quiet=!!(mode&2);the_mesh._prefs.vibe_quiet=!!(mode&1);
ui.buzzer.muted=!(mode&2);ui.vibration.muted=!(mode&1);
the_mesh.sound_muted=!(mode&2);the_mesh.vibration_muted=!(mode&1);
const auto old_buzzer=ui.buzzer.muted,old_vibration=ui.vibration.muted;
const auto old_sound=the_mesh.sound_muted,old_vibe=the_mesh.vibration_muted;
const auto mutes=the_mesh.mutes,triggers=ui.vibration.triggers,saves=the_mesh.saves;
const auto stops=ui.vibration.stops;
mode_events.clear();
ui.handleButtonTriplePress();
assert(the_mesh.saves==saves+1&&the_mesh.saved_modes.back()==((mode+1)&3));
assert(the_mesh._prefs.buzzer_quiet==!!(mode&2)&&the_mesh._prefs.vibe_quiet==!!(mode&1));
assert(ui.buzzer.muted==old_buzzer&&ui.vibration.muted==old_vibration);
assert(the_mesh.sound_muted==old_sound&&the_mesh.vibration_muted==old_vibe);
assert(the_mesh.mutes==mutes&&ui.vibration.triggers==triggers);
assert(ui.vibration.stops==stops);
assert((mode_events==std::vector<std::string>{"save","voice"}));
assert(ui.buzzer.prompts.back()==ButtonVoicePrompt::ActionFailed&&strstr(ui._alert,"failed"));
assert(ui.buzzer.courtesy.back());
}
// A notification-mask write failure cannot undo the committed physical mode.
the_mesh.save_ok=true;the_mesh.mask_save_ok=false;
the_mesh._prefs.buzzer_quiet=the_mesh._prefs.vibe_quiet=0;
the_mesh.sound_muted=the_mesh.vibration_muted=false;
ui.vibration.active=true;
ui.handleButtonTriplePress();
assert(the_mesh._prefs.vibe_quiet==1&&!the_mesh.vibration_muted&&!ui.vibration.active);
auto pulses=ui.vibration.pulses,stops=ui.vibration.stops;
ui.notificationVibration(true);
assert(ui.vibration.pulses==pulses&&ui.vibration.stops==stops+1&&!ui.vibration.active);
ui.notificationVibration(false);
assert(ui.vibration.pulses==pulses+1&&!ui.vibration.active);
the_mesh._prefs.vibe_quiet=0;ui.notificationVibration(true);
assert(ui.vibration.pulses==pulses+2&&ui.vibration.active);
ui.notificationVibration(false);assert(!ui.vibration.active);
ui._node_prefs=nullptr;ui.notificationVibration(true);assert(ui.vibration.active);
ui._node_prefs=&the_mesh._prefs;
// Shutdown quiesces multi-click changes instead of preempting final speech.
ui._shutdown_pending=true;
const auto saves=the_mesh.saves;
const auto prompts=ui.buzzer.prompts.size();
ui.handleButtonTriplePress();
assert(the_mesh.saves==saves&&ui.buzzer.prompts.size()==prompts);
#else
(void)expected;
(void)vibration_stops;
assert(ui.acknowledgements==4);
// The opt-out retains its historical driver forwarding despite mute prefs.
ui._node_prefs->vibe_quiet=1;
ui.notificationVibration(true);assert(ui.vibration.active);
ui.notificationVibration(false);assert(!ui.vibration.active);
#endif
}
'''
class Tests(unittest.TestCase):
build = gps_tests.Tests.build
def test_every_prompt_uses_its_own_clip_and_preserves_direction_tones(self):
self.build(PROMPTS, rtttl=True)
def test_each_new_board_default_uses_its_actual_buzzer_pin_and_releases_pwm(self):
for board, pin, enable in (
('RAK_WISMESH_TAG', 21, None),
('MESH_TRACKER_X1', 25, None),
('R1Neo', 3, None),
('THINKNODE_M3', 23, 36),
):
with self.subTest(board=board):
source = PROMPTS.replace('#define PIN_BUZZER 25', f'#define PIN_BUZZER {pin}')
source = source.replace('#define PIN_BUZZER_EN 37',
'' if enable is None else f'#define PIN_BUZZER_EN {enable}')
source = source.replace('#define MESH_GPS_VOICE 1',
'#define NRF52_PLATFORM\n#define COMPANION_RADIO_FULL 1\n'
+f'#define {board}\n')
if board != 'MESH_TRACKER_X1':
source = source.replace('#define HAS_DRV2605 1', '')
self.build(source, rtttl=True)
def test_quiet_courtesy_completion_and_cancellation_release_both_pins(self):
self.build(COURTESY, rtttl=True)
def test_muted_x1_mode_announcements_never_unmute_notifications_and_release_pwm(self):
for prompt in ('AlertsSoundVibration', 'AlertsSoundOnly',
'AlertsVibrationOnly', 'AlertsSilent'):
with self.subTest(prompt=prompt):
self.build(COURTESY.replace('ButtonVoicePrompt::SoundOff',
'ButtonVoicePrompt::'+prompt)
.replace('#define MESH_GPS_VOICE 1',
'#define HAS_DRV2605 1\n#define MESH_GPS_VOICE 1'), rtttl=True)
def test_x1_alert_modes_report_committed_result_and_legacy_tones_remain(self):
ui = (ROOT / 'examples/companion_radio/ui-orig/UITask.cpp').read_text()
triple = method(ui, 'void UITask::handleButtonTriplePress()')
vibration = method(ui, 'void UITask::notificationVibration(bool on)')
for voice in (0, 1):
with self.subTest(voice=voice):
self.build(X1_ALERT_MODES.replace('@METHOD@', triple)
.replace('@VIBRATION_METHOD@', vibration)
.replace('@VOICE@', str(voice)))
def test_voice_disabled_ordinary_play_keeps_its_legacy_preemption(self):
self.build(LEGACY_ORDINARY_PLAY, rtttl=True)
def test_notification_cleared_requires_an_actual_button_acknowledgement(self):
mesh = (ROOT / 'examples/companion_radio/MyMesh.h').read_text()
wrapper = method(mesh, 'bool notificationButton()')
self.build(NOTIFICATION_ACK.replace('@WRAPPER@', wrapper))
def test_muted_programmable_notifications_do_not_cancel_courtesy_audio(self):
ui = (ROOT / 'examples/companion_radio/ui-orig/UITask.cpp').read_text()
melody = method(ui, 'void UITask::notificationMelody(const char* text)')
for voice in (0, 1):
with self.subTest(voice=voice):
self.build(NOTIFICATION_MELODY.replace('@METHOD@', melody)
.replace('@VOICE@', str(voice)))
def test_notification_stop_preserves_control_tone_gap_and_dma_but_cancels_alerts(self):
for voice in (0, 1):
with self.subTest(voice=voice):
self.build(NOTIFICATION_STOPS.replace('@VOICE@', str(voice)), rtttl=True)
def test_button_actions_report_committed_results_and_shutdown_without_blocking(self):
mesh = (ROOT / 'examples/companion_radio/MyMesh.cpp').read_text()
ui = (ROOT / 'examples/companion_radio/ui-orig/UITask.cpp').read_text()
methods = method(mesh, 'bool MyMesh::setGpsEnabled(bool enabled)')
for signature in (
'void UITask::notify(UIEventType t)',
'void UITask::notificationMelody(const char* text)',
'void UITask::handleButtonAnyPress()',
'void UITask::handleButtonShortPress()',
'void UITask::handleButtonDoublePress()',
'void UITask::handleButtonTriplePress()',
'void UITask::handleButtonQuadruplePress()',
'void UITask::handleButtonLongPress()',
'void UITask::shutdown(bool restart)',
'void UITask::servicePendingShutdown()',
):
methods += '\n' + method(ui, signature)
self.build(UI_ACTIONS.replace('@METHODS@', methods), button=True)
self.build(UI_ACTIONS.replace('@METHODS@', methods)
.replace('#define ENV_INCLUDE_GPS 1', '#define ENV_INCLUDE_GPS 0'), button=True)
loop = method(ui, 'void UITask::loop()')
self.assertLess(loop.index('buzzer.loop()'), loop.index('servicePendingShutdown()'))
if __name__ == '__main__':
unittest.main()
+340
View File
@@ -0,0 +1,340 @@
"""Validate the fixed prerecorded catalog without speech synthesis or hardware."""
from pathlib import Path
import hashlib
import re
import shutil
import struct
import subprocess
import sys
import tempfile
import unittest
import wave
ROOT = Path(__file__).resolve().parents[1]
ENCODER = ROOT / "scripts/generate_button_voice.py"
SYNTHESIZER = ROOT / "scripts/generate_button_voice.ps1"
POWERSHELL = shutil.which("pwsh") or shutil.which("powershell")
CATALOG = (
("ready", "Ready"),
("notificationCleared", "Notification cleared"),
("advertQueued", "Advert queued"),
("advertFailed", "Advert failed"),
("soundOn", "Sound on"),
("soundOff", "Sound off"),
("actionFailed", "Action failed"),
("usbSetup", "USB setup"),
("shuttingDown", "Shutting down"),
("restarting", "Restarting"),
("alertsSoundVibration", "Sound and vibration"),
("alertsSoundOnly", "Sound only"),
("alertsVibrationOnly", "Vibration only"),
("alertsSilent", "Silent"),
)
# Approved Microsoft Zira Desktop, rate -1 / compressed / peak-0.12 / 8 kHz.
# Fingerprints were derived from the freshly synthesized WAVs, then checked
# against the selected header. They must not be computed from that header at
# test runtime: a Zira default/label does not establish the recordings' identity.
ZIRA_RECORDINGS = {
'ready': ('1b7823a41160abedaebb87d185a60c53bd11aba167797210c1dc37486e7c2fe0', 3147),
'notificationCleared': ('86ef9cdaaeee5e56254fef71512fe0de11119f732b2ab3125d2efda1c6dd538d', 10984),
'advertQueued': ('93db629d81e9d029a46f49df068dc0d8b8182687b348ec6123b43b8910ab2abb', 7062),
'advertFailed': ('2465649d7259fed3283915e5e56d23ae9f43772f10a419a84d654ab0360c778e', 7429),
'soundOn': ('a644478dbe1ef01d126de862b2153b077b468acaa4e452dd26b12ec4878faf1e', 6229),
'soundOff': ('43565c13fab68fa5e0ca20c55a7213e10109e6289e4fa35dc7396acdda905f78', 5393),
'actionFailed': ('971c92c5150931ffe63504573996ee931d77d77c2a888dfe35f8ba389a8c32d8', 7470),
'usbSetup': ('f42ae082f7eab15bc7e2bd7b25f1ed50c6ed264099e5b7814897828d5b8ef051', 8957),
'shuttingDown': ('973104cd1e48c6f584ec1e9eb3169bfa78bedc9399dbede88e364ef5571c91bb', 7461),
'restarting': ('49425018482bdee20b8140cd2c33419778b591ba93e0d1137f06a0ab18a3f035', 5866),
'alertsSoundVibration': ('8fedaa6d711c8623f67719214bc4d040ad43ad17736bcac82a516b8fc9c5aa93', 10915),
'alertsSoundOnly': ('6f5d4f043d53c9ff40537a13df600f7c0c71abb462ddc97f357fceda8e5fd565', 7108),
'alertsVibrationOnly': ('bd2e5d34d79655c721774db8a6e76aaf26c54b8e075fea3ddd21a427ac6f3f2e', 8723),
'alertsSilent': ('fcdef5047e616f6454aae42c90bb238303654b04e21fc9e5378693d91233e2a8', 5717),
}
def write_wav(path, samples, *, rate=8000, channels=1, width=2):
with wave.open(str(path), "wb") as wav:
wav.setnchannels(channels)
wav.setsampwidth(width)
wav.setframerate(rate)
values = samples * channels
payload = (struct.pack("<" + "h" * len(values), *values)
if width == 2 else bytes(len(values)))
wav.writeframes(payload)
def parse_header(source):
arrays = {
name: bytes(int(value, 16) for value in re.findall(r"0x([0-9a-fA-F]{2})", body))
for name, body in re.findall(
r"static const uint8_t (\w+)Data\[\] = \{(.*?)\};", source, re.S)
}
clips = {
name: (data, size, int(samples), int(rate))
for name, data, size, samples, rate in re.findall(
r"static constexpr VoiceClip (\w+)Clip = \{\s*"
r"(\w+)Data, sizeof\((\w+)Data\), (\d+), (\d+)\s*\};", source)
}
return arrays, clips
class EncoderTests(unittest.TestCase):
def fixtures(self, directory):
samples = {}
for index, (name, _) in enumerate(CATALOG):
# This known prefix encodes to 00 91 a1 b3 from predictor/index 0.
# Odd and even lengths exercise the padded final nibble separately.
values = [0, 0, 1, -1, 2, -2, 3, -3]
values += [((offset * 997 + index * 1733) % 20001) - 10000
for offset in range(9 + index * 7)]
write_wav(directory / (name + ".wav"), values)
samples[name] = values
return samples
def generate(self, directory, output, *arguments):
return subprocess.run(
[sys.executable, str(ENCODER), "--directory", str(directory),
"--output", str(output), *arguments], cwd=ROOT, capture_output=True,
text=True, timeout=30,
)
def test_fixed_catalog_sizes_independent_streams_and_reproducibility(self):
with tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary)
samples = self.fixtures(directory)
first, second = directory / "first.h", directory / "second.h"
result = self.generate(directory, first)
self.assertEqual(result.returncode, 0, result.stderr)
repeated = self.generate(directory, second)
self.assertEqual(repeated.returncode, 0, repeated.stderr)
self.assertEqual(first.read_bytes(), second.read_bytes())
self.assertEqual(result.stdout, repeated.stdout)
source = first.read_text(encoding="utf-8")
self.assertNotIn('// Voice:', source)
self.assert_x1_only_recordings_are_guarded(source)
arrays, clips = parse_header(source)
self.assertEqual(tuple(arrays), tuple(name for name, _ in CATALOG))
self.assertEqual(tuple(clips), tuple(arrays))
for name, phrase in CATALOG:
with self.subTest(phrase=phrase):
count = len(samples[name])
self.assertEqual(clips[name], (name, name, count, 8000))
self.assertEqual(len(arrays[name]), (count + 1) // 2)
self.assertEqual(arrays[name][:4], bytes.fromhex("00 91 a1 b3"))
if count & 1:
self.assertEqual(arrays[name][-1] >> 4, 0)
self.assertIn('// "' + phrase + '"', source)
self.assertIn(f"{phrase}: {count / 8000:.3f}s", result.stdout)
self.assertIn(
f"Button confirmations: {sum(len(data) for data in arrays.values())} audio bytes",
result.stdout,
)
def test_checked_in_assets_match_fixed_catalog_and_clip_bounds(self):
source = (ROOT / "src/helpers/ui/ButtonVoiceData.h").read_text(encoding="utf-8")
self.assert_x1_only_recordings_are_guarded(source)
arrays, clips = parse_header(source)
self.assertEqual(tuple(arrays), tuple(name for name, _ in CATALOG))
self.assertEqual(tuple(clips), tuple(arrays))
for name, phrase in CATALOG:
with self.subTest(phrase=phrase):
data_name, size_name, count, rate = clips[name]
self.assertEqual((data_name, size_name, rate), (name, name, 8000))
self.assertGreater(count, 0)
self.assertLessEqual(count, 32000)
self.assertEqual(len(arrays[name]), (count + 1) // 2)
self.assertTrue(any(arrays[name]))
if count & 1:
self.assertEqual(arrays[name][-1] >> 4, 0)
self.assertIn('// "' + phrase + '"', source)
def test_optional_voice_metadata_does_not_change_encoded_fixture_bytes(self):
with tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary)
self.fixtures(directory)
plain, selected = directory / 'plain.h', directory / 'zira.h'
baseline = self.generate(directory, plain)
voiced = self.generate(directory, selected, '--voice', 'Microsoft Zira Desktop')
self.assertEqual(baseline.returncode, 0, baseline.stderr)
self.assertEqual(voiced.returncode, 0, voiced.stderr)
source = selected.read_text(encoding='utf-8')
self.assertIn('// Voice: Microsoft Zira Desktop\n', source)
self.assertEqual(parse_header(source), parse_header(plain.read_text(encoding='utf-8')))
def test_selected_production_recordings_are_the_approved_zira_catalog_not_just_zira_labels(self):
source = (ROOT / 'src/helpers/ui/ButtonVoiceData.h').read_text(encoding='utf-8')
self.assertEqual([line for line in source.splitlines() if line.startswith('// Voice:')],
['// Voice: Microsoft Zira Desktop'])
arrays, clips = parse_header(source)
self.assertEqual(tuple(arrays), tuple(ZIRA_RECORDINGS))
self.assertEqual(tuple(clips), tuple(ZIRA_RECORDINGS))
for name, (fingerprint, samples) in ZIRA_RECORDINGS.items():
with self.subTest(recording=name):
self.assertEqual(hashlib.sha256(arrays[name]).hexdigest(), fingerprint)
self.assertEqual(clips[name], (name, name, samples, 8000))
def assert_x1_only_recordings_are_guarded(self, source):
start = source.index('#ifdef HAS_DRV2605')
end = source.index('#endif', start)
self.assertLess(source.index('restartingClip'), start)
for name, _ in CATALOG[10:]:
self.assertLess(start, source.index(name+'Data'))
self.assertLess(source.index(name+'Clip'), end)
def test_changed_source_wav_replaces_only_its_matching_clip(self):
with tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary)
self.fixtures(directory)
output = directory / "selected.h"
result = self.generate(directory, output)
self.assertEqual(result.returncode, 0, result.stderr)
previous, previous_clips = parse_header(output.read_text(encoding="utf-8"))
write_wav(directory / "ready.wav", [1000, -1000] * 16)
result = self.generate(directory, output)
self.assertEqual(result.returncode, 0, result.stderr)
current, current_clips = parse_header(output.read_text(encoding="utf-8"))
self.assertNotEqual(current["ready"], previous["ready"])
self.assertEqual(current_clips["ready"], ("ready", "ready", 32, 8000))
for name, _ in CATALOG[1:]:
self.assertEqual(current[name], previous[name])
self.assertEqual(current_clips[name], previous_clips[name])
def test_empty_and_overlong_clips_do_not_replace_existing_header(self):
for count in (0, 32001):
with self.subTest(samples=count), tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary)
self.fixtures(directory)
# A late failure must not leave a partially regenerated catalog.
write_wav(directory / (CATALOG[-1][0] + ".wav"), [0] * count)
output = directory / "selected.h"
output.write_bytes(b"existing selected header\n")
result = self.generate(directory, output)
self.assertNotEqual(result.returncode, 0)
self.assertIn("nonempty clip under four seconds", result.stderr)
self.assertEqual(output.read_bytes(), b"existing selected header\n")
def test_four_second_boundary_is_accepted(self):
with tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary)
self.fixtures(directory)
write_wav(directory / "restarting.wav", [0] * 32000)
output = directory / "boundary.h"
result = self.generate(directory, output)
self.assertEqual(result.returncode, 0, result.stderr)
arrays, clips = parse_header(output.read_text(encoding="utf-8"))
self.assertEqual(len(arrays["restarting"]), 16000)
self.assertEqual(clips["restarting"][2:], (32000, 8000))
def test_wrong_wav_formats_and_missing_catalog_clip_preserve_header(self):
for format_options in ({"rate": 16000}, {"channels": 2}, {"width": 1}, None):
with self.subTest(format=format_options), tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary)
self.fixtures(directory)
wav = directory / "restarting.wav"
if format_options is None:
wav.unlink()
else:
write_wav(wav, [0] * 32, **format_options)
output = directory / "selected.h"
output.write_bytes(b"selected header\n")
result = self.generate(directory, output)
self.assertNotEqual(result.returncode, 0)
if format_options is not None:
self.assertIn("Expected mono 16-bit PCM at 8000 Hz", result.stderr)
self.assertEqual(output.read_bytes(), b"selected header\n")
def test_powershell_and_encoder_have_the_same_fixed_phrase_catalog(self):
source = SYNTHESIZER.read_text(encoding="utf-8")
body = source.split("$buttonPhrases = [ordered]@{", 1)[1].split("}", 1)[0]
self.assertEqual(tuple(re.findall(r"(\w+) = '([^']+)'", body)), CATALOG)
self.assertIn("[string]$Voice = 'Microsoft Zira Desktop'", source)
self.assertIn("$buttonVoice.SelectVoice($Voice)", source)
invocation = next(line for line in source.splitlines()
if line.startswith('& python $buttonEncoder '))
self.assertIn('--voice $Voice', invocation)
self.assertIn("$buttonVoices -notcontains $Voice", source)
self.assertIn("$buttonVoice.Rate = -1", source)
self.assertIn("acompressor=threshold=0.08:ratio=6", source)
self.assertIn("aresample=8000,alimiter=limit=0.12:level=disabled", source)
@unittest.skipUnless(POWERSHELL, "PowerShell is needed for synthesis plan tests")
class PlanTests(unittest.TestCase):
def plan(self, directory=None, header=None, *arguments):
targets = []
if directory is not None:
targets += ["-OutputDirectory", str(directory)]
if header is not None:
targets += ["-OutputHeader", str(header)]
return subprocess.run(
[POWERSHELL, "-NoProfile", "-NonInteractive", "-File", str(SYNTHESIZER),
*targets, *arguments, "-WhatIf", "-Verbose"], cwd=ROOT, capture_output=True,
text=True, timeout=30,
)
def test_defaults_keep_zira_and_the_selected_production_header(self):
result = self.plan()
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn("Synthesize and encode Microsoft Zira Desktop button confirmations", result.stdout)
self.assertIn(str(ROOT / "src/helpers/ui/ButtonVoiceData.h"), result.stdout)
def test_whatif_creates_neither_audio_directory_nor_header_directory(self):
with tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary) / "audio"
header = Path(temporary) / "headers" / "button.h"
result = self.plan(directory, header)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn("Synthesize and encode Microsoft Zira Desktop button confirmations", result.stdout)
self.assertIn(str(header), result.stdout)
self.assertFalse(directory.exists())
self.assertFalse(header.parent.exists())
def test_whatif_does_not_overwrite_existing_header(self):
for voice in ("Microsoft Mark", "Microsoft Zira Desktop"):
with self.subTest(voice=voice), tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary) / "audio"
header = Path(temporary) / "button.h"
header.write_bytes(b"selected prerecorded catalog\n")
result = self.plan(directory, header, "-Voice", voice)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn(f"Synthesize and encode {voice} button confirmations", result.stdout)
self.assertEqual(header.read_bytes(), b"selected prerecorded catalog\n")
self.assertFalse(directory.exists())
def test_onlymissing_whatif_preserves_existing_recordings_and_header(self):
with tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary)
header = directory / "selected.h"
recording = directory / "soundOff.wav"
header.write_bytes(b"selected prerecorded catalog\n")
write_wav(recording, [0, 1, -1, 2, -2])
original = recording.read_bytes()
result = self.plan(directory, header, "-OnlyMissing")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(recording.read_bytes(), original)
self.assertEqual(header.read_bytes(), b"selected prerecorded catalog\n")
self.assertFalse((directory / "alertsSilent.wav").exists())
def test_mark_variant_with_explicit_header_plans_without_writes(self):
with tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary) / "mark"
header = Path(temporary) / "headers" / "mark.h"
result = self.plan(directory, header, "-Voice", "Microsoft Mark")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn("Synthesize and encode Microsoft Mark button confirmations", result.stdout)
self.assertIn(str(header), result.stdout)
self.assertFalse(directory.exists())
self.assertFalse(header.parent.exists())
def test_non_default_voice_requires_an_explicit_header(self):
with tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary) / "mark"
result = self.plan(directory, None, "-Voice", "Microsoft Mark")
self.assertNotEqual(result.returncode, 0)
self.assertIn("Voice variants require an explicit -OutputHeader", result.stderr)
self.assertFalse(directory.exists())
if __name__ == "__main__":
unittest.main()
+199
View File
@@ -0,0 +1,199 @@
"""Run the production laboratory audio-preview dispatch without any hardware actions."""
from pathlib import Path
import shutil
import subprocess
import tempfile
import unittest
from test_radio_receive_contract import method
ROOT = Path(__file__).resolve().parents[1]
COMPILER = shutil.which("g++")
SIGNATURE = "bool UITask::handleButtonAudioTest(const char* command, char* reply, size_t size)"
STUBS = r'''
#include <cassert>
#include <cstddef>
#include <cstdio>
#include <cstring>
#include <string>
#include <helpers/ui/ButtonVoice.h>
#define PIN_USER_BTN 6
struct Button {
unsigned calls=0;
bool injectPresses(uint8_t,uint32_t){++calls;return true;}
};
struct Buzzer {
bool quiet=true,playing=true,available=true;
unsigned calls=0,status_calls=0,gain_calls=0;
ButtonVoicePrompt last=ButtonVoicePrompt::Ready;
bool allow_quiet=false;
bool speakButton(ButtonVoicePrompt prompt,bool allowQuiet=false) {
++calls;last=prompt;allow_quiet=allowQuiet;
if(!available)return false;
playing=true;return true;
}
bool voiceGain(unsigned){++gain_calls;return true;}
void voiceStatus(char* reply,size_t size){++status_calls;snprintf(reply,size,"voice stub");}
};
struct UITask {
Button button;
Button* _userButton=&button;
#ifdef PIN_BUZZER
Buzzer buzzer;
#endif
#if defined(PIN_BUZZER) && MESH_GPS_VOICE
bool _shutdown_pending=false;
#endif
// No button/action/persistence APIs exist in this fixture: preview code must
// compile and execute without calling any of those real-world operations.
unsigned settings_marker=12345;
#ifdef MESH_BUTTON_AUDIO_HIL
bool handleButtonAudioTest(const char* command,char* reply,size_t size);
#endif
};
'''
ENABLED_MAIN = r'''
int main(){
UITask ui;char reply[180];
struct Mapping {const char* name;ButtonVoicePrompt prompt;};
const Mapping mappings[]={
{"ready",ButtonVoicePrompt::Ready},
{"notificationCleared",ButtonVoicePrompt::NotificationCleared},
{"advertQueued",ButtonVoicePrompt::AdvertQueued},
{"advertFailed",ButtonVoicePrompt::AdvertFailed},
{"soundOn",ButtonVoicePrompt::SoundOn},
{"soundOff",ButtonVoicePrompt::SoundOff},
{"gpsOn",ButtonVoicePrompt::GpsOn},
{"gpsOff",ButtonVoicePrompt::GpsOff},
{"actionFailed",ButtonVoicePrompt::ActionFailed},
{"usbSetup",ButtonVoicePrompt::UsbSetup},
{"shuttingDown",ButtonVoicePrompt::ShuttingDown},
{"restarting",ButtonVoicePrompt::Restarting},
#ifdef HAS_DRV2605
{"alertsSoundVibration",ButtonVoicePrompt::AlertsSoundVibration},
{"alertsSoundOnly",ButtonVoicePrompt::AlertsSoundOnly},
{"alertsVibrationOnly",ButtonVoicePrompt::AlertsVibrationOnly},
{"alertsSilent",ButtonVoicePrompt::AlertsSilent},
#endif
};
for(const auto& mapping:mappings){
auto calls=ui.buzzer.calls;
std::string command=std::string("hil voice play ")+mapping.name;
assert(ui.handleButtonAudioTest(command.c_str(),reply,sizeof(reply)));
assert(std::string(reply)==std::string("OK - preview only: ")+mapping.name);
assert(ui.buzzer.calls==calls+1&&ui.buzzer.last==mapping.prompt);
assert(ui.buzzer.allow_quiet&&ui.buzzer.quiet&&ui.buzzer.playing);
assert(!ui._shutdown_pending&&ui.settings_marker==12345&&ui.button.calls==0);
}
#ifndef HAS_DRV2605
// Other screenless boards omit the X1-only recordings and reject their
// preview names before changing any currently playing confirmation.
for(const char* name:{"alertsSoundVibration","alertsSoundOnly","alertsVibrationOnly","alertsSilent"}){
auto calls=ui.buzzer.calls;auto last=ui.buzzer.last;
std::string command=std::string("hil voice play ")+name;
assert(ui.handleButtonAudioTest(command.c_str(),reply,sizeof(reply)));
assert(std::string(reply)=="Error: unknown voice preview name");
assert(ui.buzzer.calls==calls&&ui.buzzer.last==last&&ui.buzzer.playing);
}
#endif
// Unknown/case-mismatched/extended names may not preempt active playback.
for(const char* name:{"","unknown","Ready","soundOn extra","shuttingdown"}){
auto calls=ui.buzzer.calls;auto last=ui.buzzer.last;
std::string command=std::string("hil voice play ")+name;
assert(ui.handleButtonAudioTest(command.c_str(),reply,sizeof(reply)));
assert(std::string(reply)=="Error: unknown voice preview name");
assert(ui.buzzer.calls==calls&&ui.buzzer.last==last&&ui.buzzer.playing);
}
ui._shutdown_pending=true;
for(const auto& mapping:mappings){
auto calls=ui.buzzer.calls;auto last=ui.buzzer.last;
std::string command=std::string("hil voice play ")+mapping.name;
assert(ui.handleButtonAudioTest(command.c_str(),reply,sizeof(reply)));
assert(std::string(reply)=="Error: shutdown pending; preview refused");
assert(ui.buzzer.calls==calls&&ui.buzzer.last==last&&ui.buzzer.playing);
assert(ui._shutdown_pending&&ui.settings_marker==12345&&ui.button.calls==0);
}
ui._shutdown_pending=false;ui.buzzer.available=false;
assert(ui.handleButtonAudioTest("hil voice play ready",reply,sizeof(reply)));
assert(std::string(reply)=="Error: voice preview unavailable");
assert(ui.buzzer.quiet&&!ui._shutdown_pending&&ui.settings_marker==12345);
ui.buzzer.available=true;
char tiny[12];memset(tiny,'x',sizeof(tiny));
assert(ui.handleButtonAudioTest("hil voice play restarting",tiny,9));
assert(tiny[8]==0&&tiny[9]=='x'&&tiny[10]=='x'&&tiny[11]=='x');
char zero='x';
assert(ui.handleButtonAudioTest("hil voice play ready",&zero,0)&&zero=='x');
// Existing laboratory commands and unrelated command routing stay intact.
assert(ui.handleButtonAudioTest("hil voice gain 8",reply,sizeof(reply)));
assert(ui.buzzer.gain_calls==1&&std::string(reply)=="OK - speech gain 8");
assert(ui.handleButtonAudioTest("hil voice status",reply,sizeof(reply)));
assert(ui.buzzer.status_calls==1&&std::string(reply)=="voice stub");
assert(!ui.handleButtonAudioTest("get gps",reply,sizeof(reply)));
}
'''
DISABLED_MAIN = r'''
int main(){
UITask ui;char reply[180];
assert(ui.handleButtonAudioTest("hil voice play shuttingDown",reply,sizeof(reply)));
assert(std::string(reply)=="Error: voice preview unsupported");
assert(ui.settings_marker==12345&&ui.button.calls==0);
#ifdef PIN_BUZZER
assert(ui.buzzer.calls==0&&ui.buzzer.quiet&&ui.buzzer.playing);
#endif
}
'''
@unittest.skipUnless(COMPILER, "A native C++ compiler is needed for preview dispatch tests")
class Tests(unittest.TestCase):
def build(self, flags, main):
source = (ROOT / "examples/companion_radio/ui-orig/UITask.cpp").read_text()
handler = method(source, SIGNATURE)
# Keep the actual production method's outer laboratory-only gate.
start = source.index(SIGNATURE)
self.assertEqual(source[:start].rstrip().splitlines()[-1], "#ifdef MESH_BUTTON_AUDIO_HIL")
end = start + len(handler)
self.assertTrue(source[end:].lstrip().startswith("#endif"))
unit = (flags + STUBS + "\n#ifdef MESH_BUTTON_AUDIO_HIL\n"
+ handler + "\n#endif\n" + main)
with tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary)
path = directory / "preview.cpp"
binary = directory / "preview"
path.write_text(unit)
subprocess.run(
[COMPILER, "-std=c++17", "-Wall", "-Wextra", "-Werror",
"-I" + str(ROOT / "src"), str(path), "-o", str(binary)],
check=True, timeout=30,
)
subprocess.run([str(binary)], check=True, timeout=30)
def test_all_prompt_previews_are_courtesy_audio_only_and_reject_preemption(self):
self.build(
"#include <initializer_list>\n#define MESH_BUTTON_AUDIO_HIL 1\n"
"#define PIN_BUZZER 25\n#define HAS_DRV2605 1\n#define MESH_GPS_VOICE 1\n", ENABLED_MAIN)
def test_non_vibrating_board_omits_and_rejects_x1_only_mode_previews(self):
self.build(
"#include <initializer_list>\n#define MESH_BUTTON_AUDIO_HIL 1\n"
"#define PIN_BUZZER 21\n#define MESH_GPS_VOICE 1\n", ENABLED_MAIN)
def test_voice_disabled_build_rejects_preview_without_preempting(self):
self.build(
"#define MESH_BUTTON_AUDIO_HIL 1\n#define PIN_BUZZER 25\n"
"#define MESH_GPS_VOICE 0\n", DISABLED_MAIN)
def test_no_buzzer_build_rejects_preview(self):
self.build("#define MESH_BUTTON_AUDIO_HIL 1\n#define MESH_GPS_VOICE 0\n", DISABLED_MAIN)
def test_production_build_has_no_preview_handler(self):
self.build(
"#define PIN_BUZZER 25\n#define MESH_GPS_VOICE 1\n",
"int main(){UITask ui;assert(ui.settings_marker==12345);}\n")
if __name__ == "__main__":
unittest.main()
+294
View File
@@ -0,0 +1,294 @@
"""Exercise opt-in G.726 speech against independent FFmpeg MSB-packed vectors."""
from pathlib import Path
import hashlib
import json
import os
import shutil
import struct
import subprocess
import tempfile
import unittest
import test_gps_voice_buttons as shared
from test_button_voice_generator import ZIRA_RECORDINGS
from test_gps_voice_generator import ZIRA_GPS_RECORDINGS
ROOT = Path(__file__).resolve().parents[1]
CC = shutil.which('gcc')
CXX = shutil.which('g++')
DECODER = ROOT / 'src/helpers/ui/g726/g726_decoder.c'
SANITIZERS = (['-fsanitize=address,undefined', '-fno-omit-frame-pointer']
if os.environ.get('MESH_G726_TEST_SANITIZERS') == '1' else [])
# FFmpeg 7.1 g726 encoder/decoder, 8 kHz mono, continuous MSB-first packing.
# Input: 149 samples of int(23000*sin(i*.43)+7000*sin(i*1.97)). The reference
# PCM below is pinned, not computed with the decoder being tested. Odd sample
# counts exercise both cross-byte extraction and ignored final padding bits.
VECTORS = {
16: (
'1555aaae557fafd05c2fa3d41cebf7170be8f4133aedc41ffaf1d70ee83113caf9c402fa3d00',
(12,60,68,80,100,120,144,192,-152,-260,-460,-808,-1348,-2240,-1496,-3228,
2668,6388,11088,18032,28296,3608,-2328,-5448,-17708,-30576,-21068,-16732,
-13524,7960,9392,10368,19852,31136,7272,9872,9056,-11444,-11928,-10604,
-19392,-31324,-7528,-9496,-9344,11660,29504,21868,17484,28932,6588,6040,
-2032,-15800,-30028,-23620,-19428,-15560,8012,1476,7584,18856,30580,8036,
11692,10804,-11724,-14448,-12284,-20856,32136,-8856,-9784,-9688,12472,
15444,12360,21112,18280,4004,3612,-1752,-11740,-20788,-16104,-24696,
-20612,7188,-204,8240,23980,20804,14996,23660,8108,-3212,-8576,-10660,
-18848,-25496,-18976,-15376,-5852,10232,7664,20152,-32364,13088,10528,
10836,-900,-16984,-15196,-22620,28652,-15884,1424,-152,8760,23344,17788,
21364,20676,8416,-2200,-2692,-13772,-24704,-19472,-14280,-21680,4836,
4692,9024,25744,25352,17892,15096,11768,-7900,-14896,-14800,-21052,
-29392,-11948,-6336,-6580,11280,18152),
),
24: (
'edb6db92493e6d25ef976a976b93679bfcd7677f2fb972d7467d37a512cf3a6d39a9f2bd367d4aed72adf29d93e96e65f6afdae8dda5f732',
(0,60,76,92,108,132,180,252,-272,-476,-908,-1840,-3640,-7076,-1780,-3756,
6724,16012,19764,21560,22432,8192,-8500,-3184,-18772,-24144,-17696,
-14340,-18324,4024,13468,6724,19724,25136,11348,10740,9836,-5588,
-19216,-9292,-28780,-23492,-10576,-4544,-6324,13428,24408,12108,27792,
23912,2564,-912,-1384,-18568,-27700,-13868,-18952,-17316,3560,5296,
7024,18520,25224,12844,15316,13484,-10176,-10292,-11676,-26980,-26004,
-12020,-16696,-4148,14284,17648,15116,22948,21032,8440,5108,-2084,
-17480,-21536,-18780,-25992,-16528,2392,-3592,9136,29948,21828,17336,
23512,13388,-9504,-7892,-12828,-27144,-19100,-13100,-17900,-5200,
11896,11460,20468,30592,19264,10800,12696,-3676,-18696,-14372,-20972,
-30060,-11752,-3936,-5604,8180,25072,17120,19576,26832,9028,-3872,
-1548,-15656,-28576,-16560,-15880,-21400,204,9884,6276,17700,27396,
14016,12568,13016,-5332,-15028,-13156,-20244,-25892,-13184,-9964,
-9684,11280,23016),
),
}
PREAMBLE = r'''
#include <cassert>
#include <cstdint>
#include <cstring>
#include <initializer_list>
#include <limits>
#include <vector>
#include <helpers/ui/GpsVoice.h>
#define PIN_BUZZER 25
#include <helpers/ui/nrf52/GpsVoicePlayer.h>
uint32_t now_ms=0;
using namespace mesh::audio;
'''
DECODE_MAIN = r'''
int main(){
constexpr auto codec=VoiceCodec::G726@RATE@;
constexpr unsigned bits=@BITS@;
const uint8_t data[]={@DATA@};
const int16_t expected[]={@PCM@};
constexpr unsigned count=sizeof(expected)/sizeof(expected[0]);
const VoiceClip full{data,sizeof(data),count,8000,codec};
assert(full.valid()&&full.bits()==bits);
assert(full.requiredBytes(count)==sizeof(data));
VoiceDecoder decoder;
auto check=[&](VoiceClip clip,unsigned samples){
decoder.begin(clip);assert(decoder.position()==0);
for(unsigned i=0;i<samples;++i){assert(!decoder.done());assert(decoder.next()==expected[i]);}
assert(decoder.done()&&decoder.position()==samples);
for(unsigned i=0;i<8;++i)assert(decoder.next()==0&&decoder.position()==samples);
};
// Each clip resets adaptive state, including after partial playback.
for(unsigned repetition=0;repetition<3;++repetition){
check(full,count);decoder.begin(full);decoder.next();check(full,count);
}
for(unsigned n:{1u,2u,3u,4u,5u,7u,8u,9u,31u,32u,33u,63u,64u,65u,count}){
VoiceClip short_clip{data,full.requiredBytes(n),n,8000,codec};
assert(short_clip.valid());check(short_clip,n);
}
std::vector<uint8_t> padding(data,data+sizeof(data));
const unsigned unused=(8-(count*bits)%8)%8;
assert(unused);padding.back()^=(1u<<unused)-1;
check({padding.data(),padding.size(),count,8000,codec},count);
// Truncated metadata may decode only complete codes; it never reads padding
// or a second byte beyond the declared span when a 3-bit code straddles it.
for(size_t bytes=0;bytes<sizeof(data);++bytes){
std::vector<uint8_t> bounded(data,data+bytes);
VoiceClip short_clip{bounded.data(),bytes,count,8000,codec};
assert(!short_clip.valid());check(short_clip,static_cast<unsigned>(bytes*8/bits));
}
check({nullptr,sizeof(data),count,8000,codec},0);
check({data,sizeof(data),0,8000,codec},0);
check({data,0,count,8000,codec},0);
VoiceClip huge{data,sizeof(data)-1,std::numeric_limits<uint32_t>::max(),8000,codec};
assert(!huge.valid());check(huge,static_cast<unsigned>((sizeof(data)-1)*8/bits));
VoiceClip all_bytes{data,std::numeric_limits<size_t>::max(),
std::numeric_limits<uint32_t>::max(),8000,codec};
assert(all_bytes.requiredBytes(all_bytes.samples)==
(static_cast<uint64_t>(all_bytes.samples)*bits+7)/8);
assert(all_bytes.valid()); // Validation only: actual pointer length is caller-owned.
auto other=codec==VoiceCodec::G72616?VoiceCodec::G72624:VoiceCodec::G72616;
for(auto invalid:{other,static_cast<VoiceCodec>(255)}){
VoiceClip bad{data,sizeof(data),count,8000,invalid};
assert(!bad.valid());check(bad,0);
}
assert(!VoiceClip(data,sizeof(data),count,16000,codec).valid());
// IMA remains selectable in an experiment, with its original 4-argument API.
const uint8_t ima[]={0x77};
VoiceClip original{ima,sizeof(ima),2};assert(original.codec==VoiceCodec::Ima&&original.valid());
decoder.begin(original);assert(decoder.next()==11&&decoder.next()==41&&decoder.done());
check(full,count);decoder.begin(original);assert(decoder.next()==11);
}
'''
PLAYER_MAIN = r'''
int main(){
constexpr auto codec=VoiceCodec::G726@RATE@;
const uint8_t data[]={@DATA@};
const int16_t expected[]={@PCM@};
constexpr unsigned count=sizeof(expected)/sizeof(expected[0]);
GpsVoicePlayer player;
assert(player.setGain(2));
for(unsigned n:{1u,3u,31u,32u,33u,63u,64u,65u,count}){
VoiceClip clip{data,sizeof(data),n,8000,codec};
assert(player.start(clip)&&player.active()&&player.playing());
assert(HwPWM3.owned&&HwPWM3.bound_pin==PIN_BUZZER&&irq_enabled);
assert(registers.COUNTERTOP==500&&registers.SEQ[0].REFRESH==0);
VoiceDecoder reference;reference.begin(clip);
VoiceInterpolator reconstruction;reconstruction.begin();
for(unsigned i=0;i<2;++i){
auto values=reinterpret_cast<const uint16_t*>(registers.SEQ[i].PTR);
assert(registers.SEQ[i].CNT==128);
for(unsigned j=0;j<128;++j){
int32_t pcm=reconstruction.next(reference)*2;
if(pcm>32767)pcm=32767;
if(pcm < -32768)pcm=-32768;
assert(values[j]==(0x8000|static_cast<uint16_t>((pcm+32768)*500/65536)));
}
}
unsigned sequence=0,events=0;
while(player.playing()){
assert(++events<100);registers.EVENTS_SEQEND[sequence]=1;player.interrupt();sequence^=1;
}
assert(player.played()==n&&events==(n*4+127)/128&&player.active());
player.service();assert(!player.active()&&!HwPWM3.owned&&!HwPWM3.pin&&!irq_enabled);
assert(!registers.ENABLE&&pin_states[PIN_BUZZER]==LOW);
}
assert(player.started()==9&&player.completed()==9&&player.cancelled()==0);
VoiceClip full{data,sizeof(data),count,8000,codec};
assert(player.start(full));player.stop();
assert(player.cancelled()==1&&!HwPWM3.owned&&!irq_enabled);
assert(!player.start({data,sizeof(data)-1,count,8000,codec}));
assert(!player.start({data,sizeof(data),count,16000,codec}));
assert(!player.start({nullptr,sizeof(data),count,8000,codec}));
assert(!player.start({data,sizeof(data),0,8000,codec}));
assert(!player.start({data,sizeof(data),count,8000,static_cast<VoiceCodec>(255)}));
auto other=codec==VoiceCodec::G72616?VoiceCodec::G72624:VoiceCodec::G72616;
assert(!player.start({data,sizeof(data),count,8000,other}));
HwPWM3.reject=true;assert(!player.start(full));
assert(!player.active()&&!HwPWM3.owned&&!irq_enabled&&!registers.ENABLE);
}
'''
@unittest.skipUnless(CC and CXX, 'Native C and C++ compilers are required')
class Tests(unittest.TestCase):
def build(self, source, bitrate, *, run=True, expected_error=None):
with tempfile.TemporaryDirectory() as temporary:
directory=Path(temporary)
(directory/'Arduino.h').write_text(shared.ARDUINO)
(directory/'HardwarePWM.h').write_text(shared.PWM)
(directory/'NonBlockingRtttl.h').write_text(shared.RTTTL)
path=directory/'test.cpp';path.write_text(source)
binary=directory/'test'
flags=[] if bitrate is None else [f'-DMESH_GPS_VOICE_G726_BITRATE={bitrate}',
'-DMESH_BUTTON_AUDIO_HIL=1']
if expected_error is not None:
result=subprocess.run([CXX,'-std=c++17','-I'+str(ROOT/'src'),
'-c',str(path),'-o',str(directory/'gate.o')],
capture_output=True,text=True,timeout=30)
self.assertNotEqual(result.returncode,0)
self.assertIn(expected_error,result.stderr)
return
objects=[]
if bitrate is not None:
obj=directory/'decoder.o'
subprocess.run([CC,'-std=c99','-Wall','-Wextra','-Werror',*SANITIZERS,*flags,
'-I'+str(ROOT/'src'),'-c',str(DECODER),'-o',str(obj)],
check=True,timeout=30)
objects.append(str(obj))
subprocess.run([CXX,'-std=c++17','-Wall','-Wextra','-Werror',*SANITIZERS,*flags,
'-I'+str(directory),'-I'+str(ROOT/'src'),'-I'+str(ROOT),
str(path),*objects,'-o',str(binary)],check=True,timeout=30)
if run:subprocess.run([str(binary)],check=True,timeout=30)
def fixture(self, source, bitrate):
data,pcm=VECTORS[bitrate]
return (PREAMBLE+source.replace('@RATE@',str(bitrate))
.replace('@BITS@',str(bitrate//8))
.replace('@DATA@',','.join('0x'+data[i:i+2] for i in range(0,len(data),2)))
.replace('@PCM@',','.join(map(str,pcm))))
def test_msb_reference_vectors_reset_padding_odd_lengths_and_bounds(self):
for bitrate in VECTORS:
with self.subTest(bitrate=bitrate):self.build(self.fixture(DECODE_MAIN,bitrate),bitrate)
def test_g726_shares_existing_pwm_interpolation_and_dma_lifecycle(self):
for bitrate in VECTORS:
with self.subTest(bitrate=bitrate):self.build(self.fixture(PLAYER_MAIN,bitrate),bitrate)
def test_invalid_bitrates_or_missing_laboratory_gate_cannot_compile(self):
for flags,error in (
('#define MESH_GPS_VOICE_G726_BITRATE 16\n','opt-in MESH_BUTTON_AUDIO_HIL'),
('#define MESH_BUTTON_AUDIO_HIL 0\n#define MESH_GPS_VOICE_G726_BITRATE 24\n',
'opt-in MESH_BUTTON_AUDIO_HIL'),
*[(f'#define MESH_BUTTON_AUDIO_HIL 1\n#define MESH_GPS_VOICE_G726_BITRATE {rate}\n',
'bitrate must be 16 or 24') for rate in (0,8,32,40)],
):
with self.subTest(flags=flags):
self.build(flags+'#include <helpers/ui/GpsVoice.h>\n',None,expected_error=error)
def test_default_ima_has_no_experimental_decoder_dependency(self):
self.build(PREAMBLE+r'''
int main(){
const uint8_t data[]={0x77};VoiceClip clip{data,sizeof(data),2};
VoiceDecoder decoder;decoder.begin(clip);
assert(decoder.next()==11&&decoder.next()==41&&decoder.done());
}
''',None)
REFERENCE_DIR = Path(os.environ.get('G726_REFERENCE_DIR',
str(ROOT/'out/gps-voice/g726-test')))
@unittest.skipUnless(CC and CXX and (REFERENCE_DIR/'report.json').is_file(),
'Optional independently decoded voice-catalog fixtures are unavailable')
class LocalVoiceCatalogTests(unittest.TestCase):
build = Tests.build
def test_all_32_actual_zira_g726_recordings_match_independent_ffmpeg_pcm(self):
report=json.loads((REFERENCE_DIR/'report.json').read_text(encoding='utf-8'))
expected_samples={name:recording[1] for name,recording in
{**ZIRA_RECORDINGS,**ZIRA_GPS_RECORDINGS}.items()}
self.assertEqual({clip['clip'] for clip in report['clips']}, set(expected_samples))
for bitrate in (16,24):
statements=[]
for clip in report['clips']:
name=clip['clip'];samples=clip['samples']
self.assertEqual(samples,expected_samples[name])
data=(REFERENCE_DIR/f'{name}.g726{bitrate}.be').read_bytes()
pcm_bytes=(REFERENCE_DIR/f'{name}.g726{bitrate}.decoded8k.s16le').read_bytes()
self.assertEqual(len(data),(samples*(bitrate//8)+7)//8)
self.assertEqual(len(pcm_bytes),samples*2)
self.assertEqual(hashlib.sha256(data).hexdigest(),
clip['codecs'][f'g726{bitrate}']['payload_sha256'])
pcm=struct.unpack(f'<{samples}h',pcm_bytes)
statements.append('''{
const uint8_t data[]={'''+','.join(map(str,data))+'''};
const int16_t expected[]={'''+','.join(map(str,pcm))+'''};
VoiceClip clip{data,sizeof(data),'''+str(samples)+''',8000,VoiceCodec::G726'''+str(bitrate)+'''};
assert(clip.valid());VoiceDecoder decoder;decoder.begin(clip);
for(unsigned i=0;i<clip.samples;++i){assert(!decoder.done());assert(decoder.next()==expected[i]);}
assert(decoder.done()&&decoder.position()==clip.samples&&decoder.next()==0);
}
''')
with self.subTest(bitrate=bitrate):
self.build(PREAMBLE+'int main(){\n'+''.join(statements)+'}\n',bitrate)
if __name__=='__main__':
unittest.main()
+199
View File
@@ -0,0 +1,199 @@
"""Validate the experimental G.726 catalog without FFmpeg or speech synthesis."""
from pathlib import Path
import hashlib
import io
import importlib.util
import re
import struct
import subprocess
import sys
import tempfile
from types import SimpleNamespace
import unittest
from unittest import mock
from test_button_voice_generator import CATALOG, ZIRA_RECORDINGS
from test_gps_voice_generator import ZIRA_GPS_RECORDINGS
ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT/'scripts/generate_g726_voice.py'
sys.path.insert(0,str(ROOT/'scripts'))
try:
spec=importlib.util.spec_from_file_location('g726_generator_under_test',SCRIPT)
generator=importlib.util.module_from_spec(spec)
spec.loader.exec_module(generator)
finally:
sys.path.remove(str(ROOT/'scripts'))
# Independent FFmpeg reference payloads from approved Zira WAVs; fixed here
# rather than inferred from the generated header being tested. Tuple order is
# 16 kbps then 24 kbps, always raw MSB-first g726, not g726le.
GOLDENS = {
'actionFailed': ('33663298be60ec0e2add5ebde561e69e503f222a0ab8c37323498eb1d1714365', '7e5e12f8372c473055bd07e169b3438c6f816dd68f0cd1b3ef09048f6fe51afc'),
'advertFailed': ('135d7e4ca80bcc34c02f51d72220aa17f4fc75f0d23fdac3f7eb46c53b2ecdc1', '92ce6a28b2daca7754f2b845320da0520fab447fdff46ac89ff59d283b5f28a6'),
'advertQueued': ('6f0820ccb7823100f9ad0987f45c82db1b80eb2ab3cbefdd5ebff223e59bf94b', '776a6c07f48cae3f9038f811585d1d99da35f2c203333556ad4dd9b72b9e30da'),
'alertsSilent': ('3392d2847dfcbdc8a3851653a777e32eedbd748b0588fbb1e8af8bbc75f78516', 'f01749532c1edba71079594f58dca6888cc37b9ac3cb9ec9ec12ffd813834a69'),
'alertsSoundOnly': ('1488c87772a6868b621eed3144ec46e2bda551265919cd5dd881f2640628096a', '801e8c93c52d6b219557aaff305ccdbd3c59caa5dc4bce4a1f41cc90c08b5da9'),
'alertsSoundVibration': ('90506e3b6d2626021f34c93337bb4460b77360547dbd51b821c24b23e84c2634', '4255e55a8e6be96ac8da1e8b36476896f758b27db3f88bef045f0af180f7edf8'),
'alertsVibrationOnly': ('b4e62f14ddb4c1f6783018feeae3a732bd97e73275659d4106b4ad4eca83ac70', '9ea16e3f209a544d86bab44bc36bb811d79630134fa1ccc455a68ff06edde980'),
'notificationCleared': ('734d1eaa542201c3138962647672f076a317bef044bf25ead2c4648797ed6550', '867ef876f2804b8d2767afddd9f7d922c129e6f47dfbe08d4ed5966ae6d3d24b'),
'ready': ('2a266c9f56176376eb2322a2b72e4df868a35926f4b520e13acf337e792534e3', '8eaa38100aa5a482ad72504ce40995979de45843794ffd198b2d1685089e17be'),
'restarting': ('9c204820af9bb40de30edac13bea43249884ba82753b9f6706b311abd9a93786', '2a6f1c6fcbaa26a12c9381f5ace6c7cda68f89ed7651dc506fff27d8a41aeae6'),
'shuttingDown': ('cef5b599fa81aef97459becae5e2b8e85287f153a7b5d6c5467959ee8facc130', '11fde2f8561194d823252cf957f044dc24900ebdb0debb0dae83d043b81ee627'),
'soundOff': ('20a8242182bead0359ea4920c99963ce324285a4e1d28bfcb3d8324d10d7e8c4', 'ccbf34a368af7d89892c96b8b8b59361e7671e717354d41c3c365c011a5a1261'),
'soundOn': ('e346e20ea7e69b5f40f39235d57d9e5805353a4a7abbf52800f46a9670d25fd1', '4a5f531ce35dafdd1c7735bb3d67018bae73c724cce1bf8fc0dba10a15b4d201'),
'usbSetup': ('fe99f1b81aff17db4afd6c1b9f21e70a78e983bdf1b61de99bb5ff90f01cda75', 'a714e239ebc3c2c373f246a627334db974b6d197e096e1025d10748e01f40498'),
'gpsOff': ('2d85181600ccd769a2be90393244cbd31ca8bdc3f0a919c8f24be6ed6b1792e2', 'c7ca2f32062ce0f16531d1e11c42faba2e3486b609a96c5160d0ab7d87917587'),
'gpsOn': ('8d198c025d6bc123e59d890a718835f27c75bf404f5c5288bbf08780e514aff0', 'fc44dccf32f0fb94a4d3b8b4b114eea19c5210e515d3b300fce847e5fc788b4a'),
}
class EncodeTests(unittest.TestCase):
def test_encoder_uses_exact_msb_first_ffmpeg_command_pcm_and_ceil_length(self):
for bitrate in (16,24):
for samples in ([0,-1,1], [32767,-32768,0,127,-128,1024,-1024,7,-7]):
with self.subTest(bitrate=bitrate,samples=len(samples)):
size=(len(samples)*(bitrate//8)+7)//8
payload=bytes(range(size))
with mock.patch.object(generator.subprocess,'run',
return_value=SimpleNamespace(stdout=payload)) as run:
self.assertEqual(generator.encode(samples,bitrate,'chosen-ffmpeg'),payload)
self.assertEqual(run.call_args.args[0],[
'chosen-ffmpeg','-hide_banner','-loglevel','error','-f','s16le',
'-ar','8000','-ac','1','-i','pipe:0','-c:a','g726',
'-b:a',str(bitrate*1000),'-f','g726','pipe:1'])
self.assertEqual(run.call_args.kwargs['input'],
struct.pack('<'+'h'*len(samples),*samples))
self.assertTrue(run.call_args.kwargs['check'])
self.assertEqual(run.call_args.kwargs['stdout'],subprocess.PIPE)
self.assertEqual(run.call_args.kwargs['stderr'],subprocess.PIPE)
def test_bad_bitrates_are_rejected_before_spawning_ffmpeg(self):
for bitrate in (-1,0,8,32,40,None,'16'):
with self.subTest(bitrate=bitrate), mock.patch.object(generator.subprocess,'run') as run:
with self.assertRaisesRegex(ValueError,'bitrate must be 16 or 24'):
generator.encode([0,1,-1],bitrate)
run.assert_not_called()
def test_short_long_ffmpeg_output_and_failed_process_are_rejected(self):
for bitrate in (16,24):
expected=(5*(bitrate//8)+7)//8
for length in (expected-1,expected+1):
with self.subTest(bitrate=bitrate,length=length):
with mock.patch.object(generator.subprocess,'run',
return_value=SimpleNamespace(stdout=bytes(length))):
with self.assertRaisesRegex(ValueError,f'expected {expected} bytes, got {length}'):
generator.encode([0]*5,bitrate)
with mock.patch.object(generator.subprocess,'run',
side_effect=subprocess.CalledProcessError(1,['ffmpeg'])):
with self.assertRaises(subprocess.CalledProcessError):generator.encode([0],16)
class CatalogTests(unittest.TestCase):
def test_mock_generation_emits_both_bitrate_catalogs_and_exact_source_lengths(self):
names=tuple(name for name,_ in CATALOG)+('gpsOn','gpsOff')
source_samples={name:[index]*((index+1)*3) for index,name in enumerate(names)}
with tempfile.TemporaryDirectory() as temporary:
directory=Path(temporary);output=directory/'generated.h'
arguments=self.argv(directory,output,'Microsoft Zira Desktop')
def read(path):
name={'missing-on.wav':'gpsOn','missing-off.wav':'gpsOff'}.get(path.name,path.stem)
return source_samples[name]
def encode(samples,bitrate,ffmpeg):
self.assertEqual(ffmpeg,'ffmpeg')
return bytes([bitrate])*((len(samples)*(bitrate//8)+7)//8)
with mock.patch.object(sys,'argv',arguments), mock.patch.object(generator,'read',side_effect=read), \
mock.patch.object(generator,'encode',side_effect=encode) as encoder, \
mock.patch.object(sys,'stdout',io.StringIO()):
generator.main()
self.assertEqual(encoder.call_count,32)
source=output.read_text(encoding='utf-8')
for bitrate in (16,24):
body=source.split(f'#if MESH_GPS_VOICE_G726_BITRATE == {bitrate}\n',1)[1]
body=body.split(f'#endif // {bitrate} kbps',1)[0]
for name in names:
count=len(source_samples[name])
self.assertIn(f'{name}Data, sizeof({name}Data), {count}, 8000, VoiceCodec::G726{bitrate}',body)
self.assertEqual(body.count('#ifdef HAS_DRV2605'),4)
self.assertEqual(body.count('#endif // HAS_DRV2605'),4)
def test_invalid_source_lengths_preserve_existing_experimental_header(self):
for count in (0,32001):
with self.subTest(samples=count), tempfile.TemporaryDirectory() as temporary:
directory=Path(temporary);output=directory/'generated.h'
output.write_bytes(b'existing experiment\n')
def read(path):return [0]*count if path.name=='missing-off.wav' else [0,1,-1]
with mock.patch.object(sys,'argv',self.argv(directory,output,'Microsoft Zira Desktop')), \
mock.patch.object(generator,'read',side_effect=read), \
mock.patch.object(generator,'encode') as encoder:
with self.assertRaisesRegex(ValueError,'nonempty clip under four seconds'):
generator.main()
encoder.assert_not_called()
self.assertEqual(output.read_bytes(),b'existing experiment\n')
def test_generated_16_and_24_catalogs_match_independent_zira_reference_hashes_and_lengths(self):
source=(ROOT/'src/helpers/ui/G726VoiceData.h').read_text(encoding='utf-8')
self.assertIn('// Voice: Microsoft Zira Desktop. 8 kHz mono, MSB-first raw G.726',source)
self.assertIn('EXPERIMENT ONLY',source)
self.assertIn('#if !defined(MESH_GPS_VOICE_G726_BITRATE)',source)
names=tuple(name for name,_ in CATALOG)+('gpsOn','gpsOff')
samples={name:value[1] for name,value in {**ZIRA_RECORDINGS,**ZIRA_GPS_RECORDINGS}.items()}
self.assertEqual(set(names),set(GOLDENS))
for index,bitrate in enumerate((16,24)):
with self.subTest(bitrate=bitrate):
body=source.split(f'#if MESH_GPS_VOICE_G726_BITRATE == {bitrate}\n',1)[1]
body=body.split(f'#endif // {bitrate} kbps',1)[0]
arrays={name:bytes(int(value,16) for value in re.findall(r'0x([0-9a-fA-F]{2})',data))
for name,data in re.findall(r'static const uint8_t (\w+)Data\[\] = \{(.*?)\};',body,re.S)}
clips={name:(data,size,int(count),int(rate),int(codec))
for name,data,size,count,rate,codec in re.findall(
r'static constexpr VoiceClip (\w+)Clip = \{\s*'
r'(\w+)Data, sizeof\((\w+)Data\), (\d+), (\d+), VoiceCodec::G726(\d+)\s*\};',body)}
self.assertEqual(tuple(arrays),names)
self.assertEqual(tuple(clips),names)
for name in names:
self.assertEqual(clips[name],(name,name,samples[name],8000,bitrate))
self.assertEqual(len(arrays[name]),(samples[name]*(bitrate//8)+7)//8)
self.assertEqual(hashlib.sha256(arrays[name]).hexdigest(),GOLDENS[name][index],name)
# Each optional mode must be inside a HAS_DRV2605 guard; the
# common twelve clips must remain outside those guards.
guarded=False
for line in body.splitlines():
if line=='#ifdef HAS_DRV2605':
self.assertFalse(guarded);guarded=True
elif line=='#endif // HAS_DRV2605':
self.assertTrue(guarded);guarded=False
else:
match=re.match(r'static const uint8_t (\w+)Data',line)
if match:self.assertEqual(guarded,match[1].startswith('alerts'),match[1])
self.assertFalse(guarded)
def test_cli_refuses_production_filenames_and_invalid_voice_without_touching_existing_files(self):
with tempfile.TemporaryDirectory() as temporary:
directory=Path(temporary)
for name in ('ButtonVoiceData.h','GpsVoiceData.h','buttonvoicedata.h','GPSVOICEDATA.H'):
with self.subTest(name=name):
target=directory/name;target.write_bytes(b'existing selected IMA audio\n')
result=self.cli(directory,target,'Microsoft Zira Desktop')
self.assertNotEqual(result.returncode,0)
self.assertIn('must not overwrite production IMA headers',result.stderr)
self.assertEqual(target.read_bytes(),b'existing selected IMA audio\n')
for voice in ('',' ','name\nsecondline','name\rsecondline'):
with self.subTest(voice=voice):
target=directory/'experimental.h';target.write_bytes(b'existing experiment\n')
result=self.cli(directory,target,voice)
self.assertNotEqual(result.returncode,0)
self.assertIn('nonempty single-line label',result.stderr)
self.assertEqual(target.read_bytes(),b'existing experiment\n')
def cli(self,directory,output,voice):
return subprocess.run([sys.executable,*self.argv(directory,output,voice)],
capture_output=True,text=True,timeout=30)
def argv(self,directory,output,voice):
return [str(SCRIPT),'--directory',str(directory/'missing-catalog'),
'--on',str(directory/'missing-on.wav'),'--off',str(directory/'missing-off.wav'),
'--output',str(output),'--voice',voice]
if __name__=='__main__':unittest.main()
+569
View File
@@ -0,0 +1,569 @@
"""Exercise the production speech decoder/player and screenless button actions."""
from pathlib import Path
import cmath
import math
import re
import subprocess
import tempfile
import unittest
from test_radio_receive_contract import method
ROOT = Path(__file__).resolve().parents[1]
ARDUINO = r'''
#pragma once
#include <cstdint>
#include <cstddef>
#include <cstring>
#include <cstdio>
#define LOW 0
#define HIGH 1
#define OUTPUT 1
extern uint32_t now_ms;
inline uint32_t millis(){return now_ms;}
inline int digitalRead(int){return LOW;}
inline int analogRead(int){return 0;}
inline int pin_states[64]={};
inline unsigned pin_writes[64]={};
inline void digitalWrite(int pin,int value){pin_states[pin]=value;++pin_writes[pin];}
inline void pinMode(int,int){}
'''
RTTTL = r'''
#pragma once
#include <string>
namespace rtttl {
static bool running=false;
static std::string melody;
static unsigned begins=0;
inline void begin(int,const char* song){melody=song;running=true;++begins;}
inline void stop(){running=false;}
inline bool done(){return !running;}
inline bool isPlaying(){return running;}
inline void play(){}
inline void finish(){running=false;}
}
'''
PWM = r'''
#pragma once
struct Sequence { uintptr_t PTR=0;uint32_t CNT=0,REFRESH=0,ENDDELAY=0; };
struct Registers {
uint32_t INTENCLR=0,SHORTS=0,ENABLE=0,MODE=0,COUNTERTOP=0,PRESCALER=0,
DECODER=0,LOOP=0,EVENTS_STOPPED=0,EVENTS_LOOPSDONE=0,INTENSET=0;
Sequence SEQ[2];uint32_t EVENTS_SEQEND[2]={},EVENTS_SEQSTARTED[2]={},TASKS_SEQSTART[2]={};
};
static Registers registers;
#define NRF_PWM3 (&registers)
#define PWM_MODE_UPDOWN_Up 0
#define PWM_PRESCALER_PRESCALER_DIV_1 0
#define PWM_DECODER_LOAD_Common 0
#define PWM_SHORTS_LOOPSDONE_SEQSTART0_Msk 4
#define PWM_INTENSET_SEQEND0_Msk 1
#define PWM_INTENSET_SEQEND1_Msk 2
#define PWM3_IRQn 45
static bool irq_enabled=false;
inline void NVIC_DisableIRQ(int){irq_enabled=false;}
inline void NVIC_EnableIRQ(int){irq_enabled=true;}
inline void NVIC_ClearPendingIRQ(int){}
inline void NVIC_SetPriority(int,int){}
struct HardwarePWM {
bool owned=false,pin=false,reject=false;
int bound_pin=-1;
bool takeOwnership(uint32_t){if(owned||reject)return false;return owned=true;}
bool addPin(int value){bound_pin=value;return pin=true;}
void removePin(int){pin=false;bound_pin=-1;}
bool releaseOwnership(uint32_t){assert(!pin);owned=false;return true;}
};
static HardwarePWM HwPWM3;
'''
PLAYER = r'''
#include <cassert>
#include <helpers/ui/GpsVoice.h>
#define PIN_BUZZER 25
#define MESH_BUTTON_AUDIO_HIL 1
#include <helpers/ui/GpsVoiceData.h>
#include <helpers/ui/nrf52/GpsVoicePlayer.h>
uint32_t now_ms=0;
int main(){
using namespace mesh::audio;
VoiceDecoder decoder;
const uint8_t short_data[]={0x77};
decoder.begin({short_data,1,1000});
assert(decoder.next()==11);assert(decoder.next()==41);
assert(decoder.done()&&decoder.next()==0);
decoder.begin({nullptr,1,2});assert(decoder.done());
const uint8_t negative_data[64]={0xff,0xff,0xff,0xff};
for(auto clip:{gpsOnClip,gpsOffClip,VoiceClip{negative_data,sizeof(negative_data),128},
VoiceClip{negative_data,sizeof(negative_data),128,16000}}){
decoder.begin(clip);
unsigned count=0,nonzero=0;
while(!decoder.done()){nonzero+=decoder.next()!=0; ++count;}
assert(count==clip.samples&&nonzero>count/2);
}
GpsVoicePlayer player;
assert(player.gain()==8);
assert(!player.setGain(0)&&!player.setGain(9));
assert(player.setGain(2));
for(auto clip:{gpsOnClip,gpsOffClip,VoiceClip{negative_data,sizeof(negative_data),128},
VoiceClip{negative_data,sizeof(negative_data),128,16000}}){
assert(player.start(clip));assert(player.active()&&irq_enabled&&HwPWM3.owned&&HwPWM3.pin);
assert(!player.setGain(8)&&player.gain()==2);
assert(registers.COUNTERTOP==500&&registers.SEQ[0].REFRESH==0);
VoiceDecoder reference;reference.begin(clip);
VoiceInterpolator reconstruction;reconstruction.begin(clip.sample_rate);
for(unsigned i=0;i<2;++i){
auto data=reinterpret_cast<const uint16_t*>(registers.SEQ[i].PTR);
for(unsigned j=0;j<registers.SEQ[i].CNT;++j){
assert((data[j]&0x8000)&&(data[j]&0x7fff)<=500);
int32_t pcm=reconstruction.next(reference)*2;
if(pcm>32767)pcm=32767;
if(pcm < -32768)pcm=-32768;
assert((data[j]&0x7fff)==static_cast<uint16_t>((pcm+32768)*500/65536));
}
}
unsigned seq=0,events=0;
while(player.playing()){
assert(++events<1000);
registers.EVENTS_SEQEND[seq]=1;player.interrupt();seq^=1;
}
assert(player.played()==clip.samples&&player.active());
assert(events==(clip.samples*(32000/clip.sample_rate)+127)/128);
player.service();assert(!player.active()&&!HwPWM3.owned&&!HwPWM3.pin&&!irq_enabled&&!registers.ENABLE);
assert(pin_states[PIN_BUZZER]==LOW);
}
assert(player.started()==4&&player.completed()==4&&player.cancelled()==0);
assert(player.setGain(4));
assert(player.start(gpsOnClip));player.stop();
assert(player.cancelled()==1&&player.completed()==4);
HwPWM3.reject=true;assert(!player.start(gpsOnClip));assert(!irq_enabled);
assert(!player.start({short_data,1,3}));assert(!player.playing());
assert(!player.start({short_data,1,2,12345}));
}
'''
PLAYER_BINARY = r'''
#include <cassert>
#define PIN_BUZZER 25
#define MESH_BUTTON_AUDIO_HIL 1
#define MESH_GPS_VOICE_BINARY_DRIVE 1
#include <helpers/ui/GpsVoiceData.h>
#include <helpers/ui/nrf52/GpsVoicePlayer.h>
uint32_t now_ms=0;
int main(){
using namespace mesh::audio;
GpsVoicePlayer player;
for(unsigned repetition=0;repetition<2;++repetition){
assert(player.start(gpsOnClip));
assert(player.gain()==8&&registers.COUNTERTOP==250&&registers.SEQ[0].REFRESH==1);
VoiceDecoder decoder;decoder.begin(gpsOnClip);
VoiceInterpolator interpolator;interpolator.begin();
bool high=false;unsigned transitions=0;
for(unsigned n=0;n<2;++n){
auto values=reinterpret_cast<const uint16_t*>(registers.SEQ[n].PTR);
for(unsigned i=0;i<registers.SEQ[n].CNT;++i){
int32_t pcm=interpolator.next(decoder)*8;
bool old=high;
if(pcm>1024)high=true;else if(pcm < -1024)high=false;
transitions+=old!=high;
assert(values[i]==(0x8000|(high?250:0)));
}
}
assert(transitions>0);
player.stop();assert(!player.active()&&!registers.ENABLE&&!irq_enabled);
}
assert(player.cancelled()==2);
// Encoded silence is below the hysteresis band and does not chatter.
const uint8_t silence[64]={};
assert(player.start({silence,sizeof(silence),128}));
auto values=reinterpret_cast<const uint16_t*>(registers.SEQ[0].PTR);
for(unsigned i=0;i<registers.SEQ[0].CNT;++i)assert(values[i]==0x8000);
player.stop();
}
'''
PLAYER_PDM = r'''
#include <cassert>
#define PIN_BUZZER 25
#define MESH_BUTTON_AUDIO_HIL 1
#define MESH_GPS_VOICE_PDM_DRIVE 1
#include <helpers/ui/GpsVoiceData.h>
#include <helpers/ui/nrf52/GpsVoicePlayer.h>
uint32_t now_ms=0;
int main(){
using namespace mesh::audio;
GpsVoicePlayer player;
assert(player.start(gpsOnClip));
assert(player.gain()==8&&registers.COUNTERTOP==250&&registers.SEQ[0].REFRESH==0);
VoiceDecoder decoder;decoder.begin(gpsOnClip);
VoiceInterpolator interpolator;interpolator.begin();
int32_t error=0,pcm=0;
unsigned transitions=0;bool old=false;
for(unsigned n=0;n<2;++n){
auto values=reinterpret_cast<const uint16_t*>(registers.SEQ[n].PTR);
for(unsigned i=0;i<registers.SEQ[n].CNT;++i){
if(!(i&1))pcm=interpolator.next(decoder)*8;
if(pcm>32767)pcm=32767;
if(pcm < -32768)pcm=-32768;
error+=pcm;bool high=error>=0;error-=high?32767:-32768;
assert(error>=-32768&&error<=32768);
assert(values[i]==(0x8000|(high?250:0)));
transitions+=old!=high;old=high;
}
}
assert(transitions>0);
unsigned seq=0,events=0;
while(player.playing()){
assert(++events<1000);registers.EVENTS_SEQEND[seq]=1;player.interrupt();seq^=1;
}
assert(events==(gpsOnClip.samples*8+127)/128);
assert(player.played()==gpsOnClip.samples);
player.service();assert(!player.active()&&!registers.ENABLE&&!irq_enabled);
}
'''
BUTTON = r'''
#include <cassert>
#include <string>
#include <initializer_list>
#include <helpers/ui/ButtonVoice.h>
#include "examples/companion_radio/ui-orig/Button.h"
#define ENV_INCLUDE_GPS 1
#define PIN_BUZZER 25
#define MESH_DEBUG_PRINTLN(...) ((void)0)
uint32_t now_ms=0;
enum class UIEventType{ack};
namespace mesh{namespace notify{enum{Sound=1};}}
struct CompanionNodePrefs{uint8_t gps_enabled=0,buzzer_quiet=1;};
struct SensorManager{
bool enabled=false;
bool available=true;
bool setSettingValue(const char*,const char* value){if(!available)return false;enabled=value[0]=='1';return true;}
const char* getSettingByKey(const char*){return available?(enabled?"1":"0"):nullptr;}
};
static SensorManager sensors;
struct MyMesh{
CompanionNodePrefs _prefs; bool save_ok=true;unsigned saves=0,adverts=0;
bool savePrefs(){++saves;return save_ok;}
template<class T> bool savePreference(T& p,T v){T old=p;p=v;if(savePrefs())return true;p=old;return false;}
void applyGpsPrefs(){sensors.setSettingValue("gps",_prefs.gps_enabled?"1":"0");}
bool setGpsEnabled(bool);
bool advert(){++adverts;return true;}
void setNotificationOutputMute(int,int){}
};
static MyMesh the_mesh;
struct Buzzer{
bool muted=true;unsigned spoken=0;bool last=false;
bool speakGps(bool value){if(!muted){++spoken;last=value;}return true;}
bool speakButton(ButtonVoicePrompt prompt,bool=false){
if(prompt==ButtonVoicePrompt::GpsOn||prompt==ButtonVoicePrompt::GpsOff)
return speakGps(prompt==ButtonVoicePrompt::GpsOn);
return true;
}
bool isQuiet(){return muted;}
void quiet(bool q){muted=q;}
};
struct UITask{
SensorManager* _sensors=&sensors;CompanionNodePrefs* _node_prefs=&the_mesh._prefs;
Buzzer buzzer;char _alert[80]={};bool _need_refresh=false;unsigned acks=0;
void notify(UIEventType){++acks;}
void handleButtonDoublePress();void handleButtonTriplePress();void handleButtonQuadruplePress();
};
@METHODS@
int main(){
UITask ui;Button button(6,HIGH);button.begin();
unsigned singles=0;
button.onShortPress([&]{++singles;});
button.onDoublePress([&]{ui.handleButtonDoublePress();});
button.onTriplePress([&]{ui.handleButtonTriplePress();});
button.onQuadruplePress([&]{ui.handleButtonQuadruplePress();});
auto run=[&](unsigned count){
assert(button.injectPresses(count));
assert(!button.injectPresses(count));
uint32_t start=now_ms;
while(now_ms-start<3000){now_ms+=10;button.update();}
};
run(1);assert(singles==1&&the_mesh.adverts==0&&the_mesh._prefs.gps_enabled==0);
run(2);assert(the_mesh.adverts==1&&the_mesh._prefs.gps_enabled==0);
// Multi-click actions do not turn an overlong held press into short clicks.
unsigned held=0;button.onLongPress([&]{++held;});
assert(button.injectPresses(1,3500));
for(unsigned i=0;i<500;++i){now_ms+=10;button.update();}
assert(held==1&&singles==1&&the_mesh.adverts==1);
run(3);assert(!ui.buzzer.isQuiet()&&!the_mesh._prefs.buzzer_quiet);
run(4);assert(sensors.enabled&&the_mesh._prefs.gps_enabled==1&&ui.buzzer.last);
run(4);assert(!sensors.enabled&&the_mesh._prefs.gps_enabled==0&&!ui.buzzer.last);
assert(ui.buzzer.spoken==2);
the_mesh.save_ok=false;
run(4);assert(!sensors.enabled&&the_mesh._prefs.gps_enabled==0);
assert(ui.buzzer.spoken==2&&strstr(ui._alert,"failed"));
the_mesh.save_ok=true;sensors.available=false;
run(4);assert(ui.buzzer.spoken==2);
}
'''
BUZZER_SEQUENCE = r'''
#include <cassert>
#include <string>
#define PIN_BUZZER 25
#define PIN_BUZZER_EN 37
#define MESH_BUTTON_AUDIO_HIL 1
#define MESH_GPS_VOICE 1
#include <helpers/ui/buzzer.cpp>
uint32_t now_ms=0;
int main(){
genericBuzzer buzzer;buzzer.begin();
char status[180];
auto check=[&](const char* expected){buzzer.voiceStatus(status,sizeof(status));assert(strstr(status,expected));};
auto finish_tone=[&]{
rtttl::finish();buzzer.loop();assert(buzzer.isPlaying());
now_ms+=99;buzzer.loop();assert(!registers.ENABLE);
++now_ms;buzzer.loop();
};
auto finish_speech=[&]{
unsigned seq=0,events=0;
while(registers.ENABLE){assert(++events<1000);registers.EVENTS_SEQEND[seq]=1;PWM3_IRQHandler();seq^=1;}
assert(buzzer.isPlaying());buzzer.loop();assert(!buzzer.isPlaying());
};
for(bool enabled:{true,false}){
assert(buzzer.speakGps(enabled));assert(buzzer.isPlaying());
assert(rtttl::melody==(enabled?"Startup:d=4,o=5,b=160:16c6,16e6,8g6":"Shutdown:d=4,o=5,b=100:8g5,16e5,16c5"));
assert(!registers.ENABLE);check("pending=1 tone=1");
assert(!buzzer.voiceGain(8));
now_ms+=500;buzzer.loop();assert(!registers.ENABLE);
finish_tone();assert(registers.ENABLE);check("pending=0 tone=0");
finish_speech();
}
check("started=2 completed=2 cancelled=0");
// Muting cancels a tone, a queued pause, and active speech without restarting.
assert(buzzer.speakGps(true));buzzer.quiet(true);assert(!buzzer.isPlaying());
now_ms+=1000;buzzer.loop();check("started=2 completed=2 cancelled=0");
auto before=rtttl::begins;assert(buzzer.speakGps(false));assert(rtttl::begins==before);
buzzer.quiet(false);assert(buzzer.speakGps(false));rtttl::finish();buzzer.loop();
buzzer.quiet(true);now_ms+=1000;buzzer.loop();assert(!buzzer.isPlaying());
buzzer.quiet(false);assert(buzzer.speakGps(true));
buzzer.playNotification("Alert:d=4,o=5,b=120:c");rtttl::finish();
now_ms+=1000;buzzer.loop();assert(!buzzer.isPlaying());check("started=2 completed=2 cancelled=0");
assert(buzzer.speakGps(true));finish_tone();buzzer.quiet(true);
assert(!buzzer.isPlaying());check("started=3 completed=2 cancelled=1");
buzzer.quiet(false);
// The pause handles the millisecond counter wrapping and a busy PWM gracefully.
now_ms=0xfffffff0;assert(buzzer.speakGps(false));finish_tone();assert(registers.ENABLE);
finish_speech();check("started=4 completed=3 cancelled=1");
HwPWM3.reject=true;assert(buzzer.speakGps(true));finish_tone();assert(!buzzer.isPlaying());
check("started=4 completed=3 cancelled=1");
}
'''
BUZZER_TONES = r'''
#include <cassert>
#define PIN_BUZZER 25
#define MESH_GPS_VOICE 0
#include <helpers/ui/buzzer.cpp>
uint32_t now_ms=0;
int main(){
genericBuzzer buzzer;buzzer.begin();
assert(buzzer.speakGps(true));assert(buzzer.isPlaying());
assert(rtttl::melody=="Startup:d=4,o=5,b=160:16c6,16e6,8g6");
rtttl::finish();buzzer.loop();assert(!buzzer.isPlaying());
assert(buzzer.speakGps(false));
assert(rtttl::melody=="Shutdown:d=4,o=5,b=100:8g5,16e5,16c5");
buzzer.quiet(true);assert(!buzzer.isPlaying());
unsigned before=rtttl::begins;assert(buzzer.speakGps(true));assert(rtttl::begins==before);
buzzer.quiet(false);
for(auto prompt:{ButtonVoicePrompt::Ready,ButtonVoicePrompt::NotificationCleared,
ButtonVoicePrompt::AdvertQueued,ButtonVoicePrompt::AdvertFailed,
ButtonVoicePrompt::SoundOn,ButtonVoicePrompt::SoundOff,
ButtonVoicePrompt::ActionFailed,ButtonVoicePrompt::UsbSetup,
ButtonVoicePrompt::AlertsSoundVibration,ButtonVoicePrompt::AlertsSoundOnly,
ButtonVoicePrompt::AlertsVibrationOnly,ButtonVoicePrompt::AlertsSilent}){
assert(buzzer.speakButton(prompt));assert(!buzzer.isPlaying()&&rtttl::begins==before);
}
assert(buzzer.speakGps(true));before=rtttl::begins;
assert(!buzzer.speakButton(static_cast<ButtonVoicePrompt>(255)));
assert(buzzer.isPlaying()&&rtttl::begins==before);
}
'''
class Tests(unittest.TestCase):
def test_reconstruction_suppresses_audible_sampling_images(self):
source=(ROOT/'src/helpers/ui/GpsVoice.h').read_text()
body=source.split('coefficients[4][16] = {',1)[1].split('};',1)[0]
rows=[[int(value) for value in row.split(',')]
for row in re.findall(r'\{([^{}]+)\}',body)]
self.assertEqual(len(rows),4)
for row in rows:
self.assertEqual(len(row),16)
self.assertEqual(sum(row),16384)
self.assertLess(32768*sum(abs(value) for value in row),2**31)
taps=[rows[phase][index] for index in range(16) for phase in range(4)]
def response(frequency):
value=sum(tap*cmath.exp(-2j*math.pi*frequency*i/32000)
for i,tap in enumerate(taps))/65536
return 20*math.log10(abs(value))
self.assertAlmostEqual(response(700),0,delta=0.02)
self.assertLess(response(4700),-65)
self.assertLess(response(7300),-65)
def test_wideband_reconstruction_bounds_and_consonant_band(self):
source=(ROOT/'src/helpers/ui/GpsVoice.h').read_text()
body=source.split('wide_coefficients[2][16] = {',1)[1].split('};',1)[0]
rows=[[int(value) for value in row.split(',')]
for row in re.findall(r'\{([^{}]+)\}',body)]
self.assertEqual(len(rows),2)
for row in rows:
self.assertEqual(len(row),16)
self.assertEqual(sum(row),16384)
self.assertLess(32768*sum(abs(value) for value in row),2**31)
taps=[rows[phase][index] for index in range(16) for phase in range(2)]
def response(frequency):
value=sum(tap*cmath.exp(-2j*math.pi*frequency*i/32000)
for i,tap in enumerate(taps))/32768
return 20*math.log10(abs(value))
self.assertAlmostEqual(response(3000),0,delta=0.02)
self.assertGreater(response(6500),-2)
self.assertLess(response(10000),-45)
def build(self, source, *, button=False, rtttl=False):
with tempfile.TemporaryDirectory() as temp:
path=Path(temp)
(path/'Arduino.h').write_text(ARDUINO)
(path/'HardwarePWM.h').write_text(PWM)
(path/'NonBlockingRtttl.h').write_text(RTTTL if rtttl else '#pragma once\n')
(path/'test.cpp').write_text(source)
binary=path/'test'
cmd=['g++','-std=c++17','-Wall','-Wextra','-Werror',
'-I'+str(path),'-I'+str(ROOT/'src'),'-I'+str(ROOT),
str(path/'test.cpp'),'-o',str(binary)]
if button:
cmd+=['-DMESH_BUTTON_AUDIO_HIL=1',str(ROOT/'examples/companion_radio/ui-orig/Button.cpp')]
subprocess.run(cmd,check=True)
subprocess.run([str(binary)],check=True)
def test_decoder_and_dma_lifecycle(self):
self.build(PLAYER.replace('#include <cassert>','#include <cassert>\n#include <initializer_list>'))
def test_binary_acoustic_driver_hysteresis_and_reset(self):
self.build(PLAYER_BINARY)
def test_density_acoustic_driver_bounds_and_sample_rate(self):
self.build(PLAYER_PDM)
source=PLAYER_PDM.replace('gpsOnClip', 'clip')
source=source.replace('GpsVoicePlayer player;',
'GpsVoicePlayer player; VoiceClip clip{gpsOnClip.data,gpsOnClip.bytes,gpsOnClip.samples,16000};')
source=source.replace('interpolator.begin();','interpolator.begin(clip.sample_rate);')
source=source.replace('clip.samples*8','clip.samples*4')
self.build(source)
def test_lower_carrier_acoustic_driver_preserves_sample_rate(self):
source=PLAYER.replace('#define MESH_BUTTON_AUDIO_HIL 1',
'#define MESH_BUTTON_AUDIO_HIL 1\n#define MESH_GPS_VOICE_PCM32_DRIVE 1')
self.build(source.replace('#include <cassert>','#include <cassert>\n#include <initializer_list>'))
def test_hil_can_compare_the_previous_carrier_without_changing_output_rate(self):
source=PLAYER.replace('#define MESH_BUTTON_AUDIO_HIL 1',
'#define MESH_BUTTON_AUDIO_HIL 1\n#define MESH_GPS_VOICE_PCM32_DRIVE 0')
source=source.replace('COUNTERTOP==500&&registers.SEQ[0].REFRESH==0',
'COUNTERTOP==250&&registers.SEQ[0].REFRESH==1')
source=source.replace('<=500','<=250').replace('pcm+32768)*500','pcm+32768)*250')
self.build(source.replace('#include <cassert>','#include <cassert>\n#include <initializer_list>'))
def test_experimental_drive_flags_do_not_change_non_test_firmware(self):
source=PLAYER.replace('#define MESH_BUTTON_AUDIO_HIL 1', '@FLAGS@')
# The lab gain control is absent in non-test firmware. Keep the same
# waveform assertions at its default gain instead.
source=source.replace('assert(!player.setGain(0)&&!player.setGain(9));','')
source=source.replace('assert(player.setGain(2));','')
source=source.replace('assert(!player.setGain(8)&&player.gain()==2);','')
source=source.replace('reconstruction.next(reference)*2','reconstruction.next(reference)*8')
source=source.replace('assert(player.setGain(4));','')
source=source.replace('#include <cassert>','#include <cassert>\n#include <initializer_list>')
for flags in (
'#define MESH_GPS_VOICE_BINARY_DRIVE 1',
'#define MESH_GPS_VOICE_PDM_DRIVE 1',
'#define MESH_GPS_VOICE_PCM32_DRIVE 0',
'#define MESH_GPS_VOICE_BINARY_DRIVE 1\n#define MESH_GPS_VOICE_PDM_DRIVE 1\n#define MESH_GPS_VOICE_PCM32_DRIVE 0',
):
with self.subTest(flags=flags):
self.build(source.replace('@FLAGS@', flags))
def test_direction_tones_finish_before_speech_and_cancellation(self):
self.build(BUZZER_SEQUENCE,rtttl=True)
def test_direction_tones_without_speech(self):
self.build(BUZZER_TONES.replace('#include <cassert>',
'#include <cassert>\n#include <initializer_list>'),rtttl=True)
def assert_voice_policy(self, flags, expected):
self.build(flags+'#include <helpers/ui/buzzer.h>\n'
+f'static_assert(MESH_GPS_VOICE=={expected},"speech gating");\nint main(){{}}\n')
def test_speech_defaults_to_all_screenless_nrf52_full_companions_only(self):
eligible = ('T1000_E', 'RAK_WISMESH_TAG', 'MESH_TRACKER_X1',
'R1Neo', 'THINKNODE_M3')
full = '#define NRF52_PLATFORM\n#define COMPANION_RADIO_FULL 1\n#define PIN_BUZZER 25\n'
for board in eligible:
for laboratory in ('', '#define MESH_BUTTON_AUDIO_HIL\n'):
with self.subTest(board=board, laboratory=bool(laboratory)):
self.assert_voice_policy(full+f'#define {board}\n'+laboratory, 1)
# A buzzer on another board is not sufficient: screened boards and all
# Wio profiles retain their existing tones, even in a Full Companion.
for board in ('HELTEC_T1', 'MESHTINY', 'NANO_G2_ULTRA',
'LILYGO_TECHO_CARD', 'THINKNODE_M1', 'THINKNODE_M8',
'THINKNODE_M4',
'WIO_TRACKER_L1', 'WIO_TRACKER_L1_1W', 'WIO_TRACKER_L1_EINK',
'RAK_4631', 'UNKNOWN_BOARD'):
with self.subTest(excluded_board=board):
self.assert_voice_policy(full+f'#define {board}\n', 0)
self.assert_voice_policy(full, 0)
def test_r1_neo_qspi_conflict_does_not_enable_voice_and_m4_copied_pins_are_removed(self):
self.assert_voice_policy('#define NRF52_PLATFORM\n#define COMPANION_RADIO_FULL 1\n'
'#define PIN_BUZZER 3\n#define R1Neo\n#define QSPIFLASH\n', 0)
m4 = (ROOT / 'variants/thinknode_m4/platformio.ini').read_text()
active = '\n'.join(line.strip() for line in m4.splitlines()
if not line.lstrip().startswith(';'))
self.assertNotRegex(active, r'-D\s+PIN_BUZZER=23\b')
self.assertNotRegex(active, r'-D\s+PIN_BUZZER_EN=36\b')
def test_speech_requires_full_companion_nrf52_and_fitted_buzzer_and_allows_opt_out(self):
for board in ('T1000_E', 'RAK_WISMESH_TAG', 'MESH_TRACKER_X1',
'R1Neo', 'THINKNODE_M3'):
base = f'#define {board}\n'
for case, flags in (
('no_nrf52', '#define COMPANION_RADIO_FULL 1\n#define PIN_BUZZER 25\n'),
('full_disabled', '#define NRF52_PLATFORM\n#define COMPANION_RADIO_FULL 0\n#define PIN_BUZZER 25\n'),
('non_companion', '#define NRF52_PLATFORM\n#define PIN_BUZZER 25\n'),
('no_buzzer', '#define NRF52_PLATFORM\n#define COMPANION_RADIO_FULL 1\n'),
('explicit_opt_out', '#define NRF52_PLATFORM\n#define COMPANION_RADIO_FULL 1\n#define PIN_BUZZER 25\n#define MESH_GPS_VOICE 0\n'),
):
with self.subTest(board=board, excluded_profile=case):
self.assert_voice_policy(base+flags, 0)
def test_unsafe_audio_profiles_retain_tones_and_explicit_voice_override(self):
full = '#define NRF52_PLATFORM\n#define COMPANION_RADIO_FULL 1\n'
profiles = (
# The M4's previous buzzer configuration overlaps its I2C SDA
# and battery-status LED, not a verified speaker connection.
'#define THINKNODE_M4\n#define PIN_BUZZER 23\n#define PIN_BUZZER_EN 36\n',
# R1 Neo's dummy QSPI clock is the same pin as its buzzer.
'#define R1Neo\n#define PIN_BUZZER 3\n#define QSPIFLASH 1\n',
)
for profile in profiles:
with self.subTest(profile=profile):
self.assert_voice_policy(full+profile, 0)
self.assert_voice_policy(full+profile+'#define MESH_GPS_VOICE 1\n', 1)
def test_buttons_gps_persistence_and_failed_save(self):
mesh=(ROOT/'examples/companion_radio/MyMesh.cpp').read_text()
ui=(ROOT/'examples/companion_radio/ui-orig/UITask.cpp').read_text()
methods=method(mesh,'bool MyMesh::setGpsEnabled(bool enabled)')
for signature in ('void UITask::handleButtonDoublePress()',
'void UITask::handleButtonTriplePress()',
'void UITask::handleButtonQuadruplePress()'):
methods+='\n'+method(ui,signature)
self.build(BUTTON.replace('@METHODS@',methods),button=True)
if __name__=='__main__':
unittest.main()
+155
View File
@@ -0,0 +1,155 @@
"""Validate speech-generation plans, fixed audio identity, and fixture encoding."""
from pathlib import Path
import hashlib
import shutil
import subprocess
import sys
import tempfile
import unittest
from test_button_voice_generator import parse_header, write_wav
ROOT = Path(__file__).resolve().parents[1]
POWERSHELL = shutil.which("pwsh") or shutil.which("powershell")
BASELINE_FILTER = (
"highpass=f=350,lowpass=f=3400,"
"silenceremove=start_periods=1:start_threshold=-45dB,areverse,"
"silenceremove=start_periods=1:start_threshold=-45dB,areverse,"
"volume=1.4,alimiter=limit=0.8:level=disabled,apad=pad_dur=0.05"
)
SELECTED_FILTER = (
"highpass=f=350,lowpass=f=3400,"
"silenceremove=start_periods=1:start_threshold=-45dB,areverse,"
"silenceremove=start_periods=1:start_threshold=-45dB,areverse,"
"acompressor=threshold=0.08:ratio=6:attack=5:release=70:makeup=4:knee=4:detection=rms,"
"volume=1.4,aresample=8000,alimiter=limit=0.12:level=disabled,apad=pad_dur=0.05"
)
# Independently pinned from approved Zira WAVs (rate -1 / compressed / peak
# 0.12 / 8 kHz), not from the header being tested or its voice-name metadata.
ZIRA_GPS_RECORDINGS = {
'gpsOn': ('8bb2571a94245c9df307857a072f5e8900e72976c81536dc02c25251da42ca7a', 9683),
'gpsOff': ('4dd8519110914aba97d84d2d1b72a20dec8bb8ed573aa167551d8e092af8ffcb', 9237),
}
class AssetTests(unittest.TestCase):
def test_production_gps_recordings_are_the_approved_zira_audio_and_metadata(self):
source = (ROOT / 'src/helpers/ui/GpsVoiceData.h').read_text(encoding='utf-8')
self.assertEqual([line for line in source.splitlines() if line.startswith('// Voice:')],
['// Voice: Microsoft Zira Desktop'])
arrays, clips = parse_header(source)
self.assertEqual(tuple(arrays), tuple(ZIRA_GPS_RECORDINGS))
self.assertEqual(tuple(clips), tuple(ZIRA_GPS_RECORDINGS))
for name, (fingerprint, samples) in ZIRA_GPS_RECORDINGS.items():
with self.subTest(recording=name):
self.assertEqual(hashlib.sha256(arrays[name]).hexdigest(), fingerprint)
self.assertEqual(clips[name], (name, name, samples, 8000))
def test_optional_voice_metadata_preserves_generic_pcm_fixture_encoding(self):
with tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary)
on, off = directory / 'on.wav', directory / 'off.wav'
write_wav(on, [0, 0, 1, -1, 2, -2, 3, -3, 1000])
write_wav(off, [0, 1, -1, 2, -2, 3, -3, -1000])
outputs = []
for arguments in ((), ('--voice', 'Microsoft Zira Desktop')):
target = directory / ('selected.h' if arguments else 'plain.h')
result = subprocess.run(
[sys.executable, str(ROOT / 'scripts/generate_gps_voice.py'),
'--on', str(on), '--off', str(off), '--output', str(target), *arguments],
capture_output=True, text=True, timeout=30,
)
self.assertEqual(result.returncode, 0, result.stderr)
outputs.append(target.read_text(encoding='utf-8'))
self.assertNotIn('// Voice:', outputs[0])
self.assertIn('// Voice: Microsoft Zira Desktop\n', outputs[1])
self.assertEqual(parse_header(outputs[0]), parse_header(outputs[1]))
def test_powershell_forwards_selected_voice_to_encoder_metadata(self):
source = (ROOT / 'scripts/generate_gps_voice.ps1').read_text(encoding='utf-8')
invocation = next(line for line in source.splitlines()
if line.startswith('& python $gpsEncoder '))
self.assertIn('--voice $Voice', invocation)
@unittest.skipUnless(POWERSHELL, "PowerShell is needed for generator plan tests")
class Tests(unittest.TestCase):
def plan(self, *arguments):
return subprocess.run(
[POWERSHELL, "-NoProfile", "-NonInteractive", "-File",
str(ROOT / "scripts/generate_gps_voice.ps1"),
*arguments, "-WhatIf", "-Verbose"],
cwd=ROOT, capture_output=True, text=True, timeout=30,
)
def test_selected_defaults_and_original_profile(self):
result = self.plan()
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn(SELECTED_FILTER, result.stdout)
self.assertIn("Microsoft Zira Desktop; rate: -1", result.stdout)
self.assertIn(str(ROOT / "src/helpers/ui/GpsVoiceData.h"), result.stdout)
result = self.plan(
"-Voice", "Microsoft Zira Desktop", "-Compress:$false", "-PeakLimit", "0.8",
"-OutputHeader", "out/gps-voice/original-zira.h",
)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn(BASELINE_FILTER, result.stdout)
self.assertIn("Microsoft Zira Desktop; rate: -1", result.stdout)
self.assertNotIn("acompressor=", result.stdout)
def test_compressed_variant_is_planned_without_writes(self):
with tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary) / "preview"
header = Path(temporary) / "headers/mark.h"
result = self.plan(
"-Voice", "Microsoft Mark", "-Rate", "0", "-Compress",
"-PeakLimit", "0.10", "-OutputDirectory", str(directory),
"-OutputHeader", str(header),
)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn("Microsoft Mark; rate: 0", result.stdout)
self.assertIn("acompressor=threshold=0.08:ratio=6", result.stdout)
self.assertIn("makeup=4:knee=4:detection=rms", result.stdout)
self.assertIn("aresample=8000,alimiter=limit=0.1:level=disabled", result.stdout)
self.assertIn(str(header), result.stdout)
self.assertFalse(directory.exists())
self.assertFalse(header.parent.exists())
def test_variants_require_an_explicit_header(self):
for arguments in (("-Compress:$false",), ("-Voice", "Microsoft Mark"),
("-Rate", "0"), ("-PeakLimit", "0.10")):
with self.subTest(arguments=arguments):
result = self.plan(*arguments)
self.assertNotEqual(result.returncode, 0)
self.assertIn("explicit -OutputHeader", result.stderr)
def test_invalid_header_extension_is_rejected(self):
result = self.plan("-OutputHeader", "out/gps-voice/test.wav")
self.assertNotEqual(result.returncode, 0)
self.assertIn("must name a .h file", result.stderr)
def test_file_directory_collisions_are_rejected(self):
with tempfile.TemporaryDirectory() as temporary:
path = Path(temporary)
file = path / "existing-file"
file.touch()
result = self.plan("-OutputDirectory", str(file))
self.assertNotEqual(result.returncode, 0)
self.assertIn("names an existing file", result.stderr)
directory = path / "existing-directory.h"
directory.mkdir()
result = self.plan("-OutputHeader", str(directory))
self.assertNotEqual(result.returncode, 0)
self.assertIn("names an existing directory", result.stderr)
def test_rate_and_peak_ranges_are_rejected(self):
for arguments in (("-Rate", "11"), ("-Rate", "-11"),
("-PeakLimit", "0.01"), ("-PeakLimit", "1.1")):
with self.subTest(arguments=arguments):
result = self.plan("-OutputHeader", "out/gps-voice/test.h", *arguments)
self.assertNotEqual(result.returncode, 0)
self.assertIn(arguments[0][1:], result.stderr)
if __name__ == "__main__":
unittest.main()