EventMesh.subscribe("paly_device_res_error",self.paly_device_res_error)
EventMesh.subscribe("Power_off",self.pwk_key_Long_Press)
EventMesh.subscribe("keyAI_extint_cb", self.keyAI_extint_cb)
- EventMesh.subscribe("get_usb_state", self.get_usb_state)
print("AI_dev:2")
self.pk = PowerKey()
self.pk.powerKeyEventRegister(self.powerKey_callback)
close_all_ledm()
#EventMesh.publish("audio_montage_play",["R_CHARGE_OUT.mp3"])
self.aud.play(1,1,'U:/static/R_CHARGE_OUT.mp3')
- def get_usb_state(self,event,msg):#插入为1 拔出为0
- return self.usb_status
def btnInfo_queue_thread(self):
while True:
data = self.btnInfo_queue.get()
audio_buf.index("R_READ_TIME_{}".format(iccid[12])),audio_buf.index("R_READ_TIME_{}".format(iccid[13])),audio_buf.index("R_READ_TIME_{}".format(iccid[14])),audio_buf.index("R_READ_TIME_{}".format(iccid[15])),\
audio_buf.index("R_READ_TIME_{}".format(iccid[16])),audio_buf.index("R_READ_TIME_{}".format(iccid[17])),audio_buf.index("R_READ_TIME_{}".format(iccid[18])),audio_buf.index("R_READ_TIME_{}".format(iccid[19])))
str_data += '{}.mp3+'.format(audio_buf.index("shouhou"))#可联系售后解决，联系方式见售后客服电话是400-886-8777或到微信公众号柒牌SEVENBRAND自行充值：
- self.aud.montage_play('U:/static/audio.bin',1,1,str_data)
+ # self.aud.montage_play('U:/static/audio.bin',1,1,str_data)
#鉴权出错，播放imei号
def paly_device_res_error(self,event,msg):
str_data = 'mp3files={}.mp3+'.format(audio_buf.index("jianquancuowu"))#AI功能异常,请联系客户查询，设备号为:
self.audio_play_state = 0
#EventMesh.publish("set_audio_state", 0)
print('audio-play finish.!!!!!!')
-
- def AI_key_press_cb(self,event,msg):#按键按下
- print("AI_key_press")
- self.get_config_timer_count = 0
- self.key_state = 1
- self.Close()
- if dev_type == "7PIN" :
- self.aud.aud_tone_play(10,200)#播放按键音
- #self.aud.aud_tone_play(10,200)#播放按键音
- self.AI_TTS_str = ""#你好，有什么吩咐"
- self.AI_ASR_str = ""
- self.AI_ASR_str_tmp = ""
- self.VC_state = VC_State_None
- # if self.VC_state != VC_State_None:
- # EventMesh.publish("publish_AI_btnInfo",{"func":self.Close})
- # EventMesh.publish("publish_AI_btnInfo",{"func":utime.sleep_ms,'data':(10)})
- #if self.VC_record_state_flag >1:#如果在录音，就先停止录音
- self.VC_Speech_End()
-
- self.VC_Speech_Start()
- self.AI_VC_start()
- #self.chenggong_tone()
+ def AI_key_press_cb(self, event=None, msg=None):#按键长按
+ EventMesh.publish("publish_AI_btnInfo",{"func":self.AI_key_press_cb_1})
+ def AI_key_press_cb_1(self):#按键按下
+ pass
+ # print("AI_key_press")
+ # self.get_config_timer_count = 0
+ # self.key_state = 1
# self.Close()
- # self.AI_VC_start()
+ # if dev_type == "7PIN" :
+ # self.aud.aud_tone_play(10,200)#播放按键音
+ # #self.aud.aud_tone_play(10,200)#播放按键音
+ # self.AI_TTS_str = ""#你好，有什么吩咐"
+ # self.AI_ASR_str = ""
+ # self.AI_ASR_str_tmp = ""
+ # self.VC_state = VC_State_None
+ # # if self.VC_state != VC_State_None:
+ # # EventMesh.publish("publish_AI_btnInfo",{"func":self.Close})
+ # # EventMesh.publish("publish_AI_btnInfo",{"func":utime.sleep_ms,'data':(10)})
+ # #if self.VC_record_state_flag >1:#如果在录音，就先停止录音
+ # self.VC_Speech_End()
+
+ # self.VC_Speech_Start()
+ # self.AI_VC_start()
+ # #self.chenggong_tone()
+ # # self.Close()
+ # # self.AI_VC_start()
self.AI_key_long_press_flag = 0
-
- def AI_key_release(self, event, msg):#按键弹起
+ def AI_key_release(self, event=None, msg=None):#按键长按
+ EventMesh.publish("publish_AI_btnInfo",{"func":self.AI_key_release_cb})
+ def AI_key_release_cb(self):#按键弹起
print("AI_key_release123")
self.key_state = 0
if self.AI_key_long_press_flag == 1 :#有长按
+ print("长按")
#self.AI_client_VC_end_speech()
#EventMesh.publish("publish_AI_btnInfo",{"func":utime.sleep_ms,'data':(1000)})
#EventMesh.publish("publish_AI_btnInfo",{"func":self.AI_client_asr_StopTranscription_requset})
if self.VC_mode == 1 :#长按单次问答模式
# if self.AI_ASR_str_tmp != "":
print("AI_key_release:asr_str:{}".format(self.AI_ASR_str))
- EventMesh.publish("publish_AI_btnInfo",{"func":self.AI_VC_start})
- self.AI_TTS_str = "请您按住中间的按钮开始讲话，讲完松开即可 "
+ # EventMesh.publish("publish_AI_btnInfo",{"func":self.AI_VC_start})
+ # self.AI_TTS_str = "请您按住中间的按钮开始讲话，讲完松开即可 "
# EventMesh.publish("publish_AI_btnInfo",{"func":self.AI_VC_start})
self.AI_client_VC_end_speech()
pass
else :
+ print("短按")
if self.VC_mode == 1 :#长按单次问答模式
- try:
- EventMesh.publish("publish_AI_btnInfo",{"func":self.Close})
- except Exception as e:
- usys.print_exception(e)
- EventMesh.publish("publish_AI_btnInfo",{"func":self.key_prompt}) #提示请您按住中间的按钮开始讲话，讲完松开即可
- # self.AI_TTS_str = "请您按住中间的按钮开始讲话，讲完松开即可 "
- # EventMesh.publish("publish_AI_btnInfo",{"func":self.AI_VC_start})
+ # try:
+ # EventMesh.publish("publish_AI_btnInfo",{"func":self.Close})
+ # except Exception as e:
+ # usys.print_exception(e)
+ #EventMesh.publish("publish_AI_btnInfo",{"func":self.key_prompt}) #提示请您按住中间的按钮开始讲话，讲完松开即可
+ self.AI_TTS_str = "请您按住中间的按钮开始讲话，讲完松开即可 "
+ EventMesh.publish("publish_AI_btnInfo",{"func":self.AI_VC_start})
pass
#self.VC_Speech_End()
- def AI_key_long_press_cb(self, event, msg):#按键长按
+ def AI_key_long_press_cb(self, event=None, msg=None):#按键长按
+ EventMesh.publish("publish_AI_btnInfo",{"func":self.AI_key_long_press_cb_1})
+ def AI_key_long_press_cb_1(self):#按键长按
print("AI_key_long_press")
+
+ if self.AI_key_long_press_flag == 0:
+ print("AI_key_press")
+ self.get_config_timer_count = 0
+ self.key_state = 1
+ self.Close()
+ # if dev_type == "7PIN" :
+ # self.aud.aud_tone_play(10,200)#播放按键音
+ #self.aud.stop()
+
+ self.AI_TTS_str = ""#你好，有什么吩咐"
+ self.AI_ASR_str = ""
+ self.AI_ASR_str_tmp = ""
+ self.VC_state = VC_State_None
+ # if self.VC_state != VC_State_None:
+ # EventMesh.publish("publish_AI_btnInfo",{"func":self.Close})
+ # EventMesh.publish("publish_AI_btnInfo",{"func":utime.sleep_ms,'data':(10)})
+ #if self.VC_record_state_flag >1:#如果在录音，就先停止录音
+ self.VC_Speech_End()
+
+ self.VC_Speech_Start()
+ self.AI_VC_start()
+ self.aud.play(1,0,'U:/static/DONG.mp3')#闹钟音频
+ #self.chenggong_tone()
+ # self.Close()
+ # self.AI_VC_start()
self.AI_key_long_press_flag = 1
- def pwk_key_release_cb(self, event, msg):#电源键短按，关闭
+ def pwk_key_release_cb(self, event=None, msg=None):#电源键短按，关闭
+ EventMesh.publish("publish_AI_btnInfo",{"func":self.pwk_key_release_cb_1})
+ def pwk_key_release_cb_1(self ):#电源键短按，关闭
+
+ # if self.AI_TTS_str == "":
+ # #close_all_ledm()
+ #EventMesh.publish("close_all_led")
+ #EventMesh.publish("stop_led_timer")
+ #EventMesh.publish("publish_AI_btnInfo",{"func":self.Close})
+ self.Close()
print("pwk_key_release")
EventMesh.publish("close_hr_sop2")#关闭健康测试
- if self.AI_TTS_str == "":
- #close_all_ledm()
- EventMesh.publish("close_all_led")
- #EventMesh.publish("stop_led_timer")
- EventMesh.publish("publish_AI_btnInfo",{"func":self.Close})
def key_prompt(self):
EventMesh.publish("audio_montage_play",["key_prompt.mp3"])
#payload = jsload_data["payload"]
# self.VC_dialog_id = payload["dialog_id"]
# self.VC_task_id = jsload_data["task_id"]
+
+ #utime.sleep_ms(500)
EventMesh.publish("open_led_blue")
print("VC:Started str{}!!!!!!!!\r\n".format(self.AI_TTS_str))
if self.AI_TTS_str == "":
text = ujson.loads(text)
print(text)
result = text["p"]
- EventMesh.publish("call_phonebook_name",result)
+ if result != None:
+ EventMesh.publish("call_phonebook_name",result)
+ else:
+ self.AI_TTS_str = "找不到联系人，请重试"
return 0
#设置闹钟 {"command":{"chinese_description":"","english_description":"Set Alarm","fail_prompt":"","params_json":"{\"hour_value\": 16, \"min_value\": 36, \"day_value16, \"min_value\": 36, \"day_value\": [1], \"loop_value\": 0, \"event_notice\": \"\"}","succeed_prompt":""}}
if text.find("普通闹钟") != -1 :
if self.VC_play_TYPE == VC_play_TYPE_PCM:
if self.play_pcm == None:
self.play_pcm = audio.Audio.PCM(0,1,self.VC_play_samplerate,1,1,2)
- #self.play_pcm.setVolume(11)
+ self.play_pcm.setVolume(11)
ret = self.play_pcm.stream_write_start()
print("self.record_pcm.stream_write_start:ret:{}".format(ret))
elif self.VC_play_TYPE == VC_play_TYPE_MP3:
#TTS播放(可加入参数事件调用版)
def AI_client_VC_synevent(self,event, msg):
- self.AI_TTS_str = msg
+ EventMesh.publish("publish_AI_btnInfo",{"func":self.AI_client_VC_synevent_cb,"data":msg})
+ def AI_client_VC_synevent_cb(self, data):
+ print("TTS ={} ".format(data))
+ self.AI_TTS_str = data
self.Close_flag = 1
EventMesh.publish("publish_AI_btnInfo",{"func":self.VC_Speech_End})
EventMesh.publish("publish_AI_btnInfo",{"func":self.websocket_deconnect_VC})
get_temp_state = EventMesh.publish("get_temp_state")
get_bg_state = EventMesh.publish("get_bg_state")
get_bp_state = EventMesh.publish("get_bp_state")
+
+ print("close_hr_sop2")
if get_hr_state != None and get_spo2_state != None and get_temp_state != None and get_bg_state != None and get_bp_state != None:
if get_hr_state != 0:
print("结束心率测量")
EventMesh.publish("set_bp_state",0)#2代表后台进入测试，定时测心率 1ui界面测试洗心率
EventMesh.debug_log("sstop bp")
EventMesh.publish(constants.EVENT.STOP_HEART)
-
+ EventMesh.publish("stop_led_timer")
+
def set_manual_measure_mode(self, event, msg): #手动测试 1心率 2 血氧 3 体温
get_hr_state = EventMesh.publish("get_hr_state")
get_spo2_state = EventMesh.publish("get_spo2_state")
else:#未设置sos紧急求助号码
EventMesh.publish("close_timer_all")#
str_data = 'mp3files={}.mp3'.format(audio_buf.index("R_NO_SET_SOS_DIAL"))
- EventMesh.publish("AI_client_VC_synevent","您还未设置sos紧急求助号码呀！")#用TTS读
+ EventMesh.publish("AI_client_VC_synevent","您还未设置 S O S 紧急求助号码呀！")#用TTS读
EventMesh.publish("play_audio_file",[str_data,1])#获取播放音频文件('U:/static/audio.bin',1,1,str_data)
pass
else:
EventMesh.subscribe("publish_message_event_light",self.publish_message_event_light)
EventMesh.subscribe("close_wechat_btn_timer", self.close_wechat_btn_timer)#来电强制关闭微聊语音
EventMesh.subscribe("get_AIKey_long", self.get_AIKey_long)#判断长按
+ EventMesh.subscribe("get_AIKey_long", self.get_AIKey_long)#判断长按
def btnInfo_queue_thread(self):
while True:
data = self.btnInfo_queue.get()
EventMesh.publish("set_sos_flag",0)
EventMesh.publish("set_sos_num",0)
EventMesh.publish("set_hangup_mode",0)
- EventMesh.publish("close_hr_sop2")#
+ #EventMesh.publish("close_hr_sop2")#
EventMesh.publish("pwk_key_release")
flag = EventMesh.publish("get_call_state")
# if flag != 1:
pass
self.sosKey_press_count = 0
-
-
+
def wechatKey_extint_cb(self,args):
open_power = EventMesh.publish("get_power_on_state")#获取开机标识
open_power_off = EventMesh.publish("get_power_off_state")#获取开机标识
if self.wechatKey_press == 1 :#是否按下
tmp = EventMesh.publish("get_wechat_time")#获取微聊时间戳
self.wechatKey_count += 1
- if self.wechatKey_count >= 15: #长按1s进入微聊录音
+ if self.wechatKey_count >= 4: #长按1s进入微聊录音
flag = EventMesh.publish("get_call_state")
fall_down = EventMesh.publish("get_fall_down_state")#跌倒报警标识
if flag != None:
self.wechatKey_press = 0
self.wechat_state = 0
self.wechatKey_press_count+=1
- self.wechatKey_timer.start(500, 0, self.publish_wechatKey_btn_event_cb)#松手超时1s判断
+ self.wechatKey_timer.start(300, 0, self.publish_wechatKey_btn_event_cb)#松手超时1s判断
def publish_wechatKey_btn_event_cb(self, *args, **kwargs):
EventMesh.publish("publish_message_event_btnInfo",{"func":self.wechatKey_timeout_cb,"data":0})
-
+
def wechatKey_timeout_cb(self,data):#微聊键
open_power = EventMesh.publish("get_power_on_state")#获取开机标识
open_power_off = EventMesh.publish("get_power_off_state")#获取开机标识
elif self.wechatKey_press_count == 1:#开始说话
self.wechatKey_press_count = 0
print("说话")
+ self.wechatKey_long_press = 0#标志位变为0
EventMesh.publish("stop_led_timer")#设置参数
call_state = EventMesh.publish("get_call_state")
- self.wechatKey_long_press = 0#标志位变为0
if call_state == 0:
fall_down = EventMesh.publish("get_fall_down_state")#跌倒报警标识
if fall_down != None:
EventMesh.debug_log("set_hr_blood_led :{}".format(msg))
if msg != None:
if msg == 1:
- EventMesh.publish("set_led_time_arg",[1,2,10*70*2,led_mode.FLASH_LIGHT_GREEN,10,5,led_mode.HEALTH_LED_MODE])#測量成功
+ EventMesh.publish("set_led_time_arg",[1,2,10*70*2,led_mode.FLASH_LIGHT_RED_GREEN,10,5,led_mode.HEALTH_LED_MODE])#測量成功
# EventMesh.publish("set_led_time_arg",[1000,200,300,led_mode.FLASH_LIGHT_RED_GREEN,led_mode.DEFAULT_LONG_LIGHT_MODE,led_mode.HEALTH_LED_MODE])#開始測量
elif msg == 2:
EventMesh.publish("set_led_time_arg",[1,2,60,led_mode.FLASH_LIGHT_GREEN,60,30,led_mode.OTHER_MEASURE_MODE])#測量成功
open_power = EventMesh.publish("get_power_on_state")#获取开机标识
power_off = EventMesh.publish("get_power_off_state")#获取开机标识
- # EventMesh.debug_log("set_led_time_arg",msg,open_power,power_off)
+ EventMesh.debug_log("set_led_time_arg",msg,open_power,power_off)
if open_power != None:
if open_power == 0:
return
if power_off == 1:
return
usb_in = EventMesh.publish("get_usb")#
+ print("set_led_time_arg usb_in:{}".format(usb_in))
if msg[6] != led_mode.POWER_OFF_MODE:#插入usb关机以防不亮红绿灯
# EventMesh.debug_log("usb_in",usb_in)
if usb_in != None:
if msg[5] != led_mode.HEALTH_LED_MODE:
self.working_led_mode = led_mode.OTHER_MEASURE_MODE
- get_hr_state = EventMesh.publish("get_hr_state")
- get_spo2_state = EventMesh.publish("get_spo2_state")
+ # get_hr_state = EventMesh.publish("get_hr_state")
+ # get_spo2_state = EventMesh.publish("get_spo2_state")
- wechat_state = EventMesh.publish("get_wechat_state")
- temp_state = EventMesh.publish("get_temp_state")
- bp_state = EventMesh.publish("get_bp_state")
- bg_state = EventMesh.publish("get_bg_state")
- if wechat_state != None: #获取微聊状态
- if wechat_state == 1:
- if self.working_led_mode != led_mode.WECHAT_LED_MODE:#如果不在亮微聊模式需要亮微聊模式
- self.timer_state = 0
- self.timer.stop()
- str_buf = [1,1,1,led_mode.LONG_LIGHT_GREEN,4,2,led_mode.WECHAT_LED_MODE]
- self.open_timer(str_buf)
- print("open_timer3!!!!!!!!!!!!!!!!!!")
- self.working_led_mode = led_mode.WECHAT_LED_MODE
- return
- else:
- return
- else:
- self.wechat_btn = 0
- # if get_hr_state != 0 or get_spo2_state != 0 or temp_state != 0 or bp_state != 0 or bg_state != 0 :#心率血氧优先级高
- if get_hr_state == 1 or get_spo2_state == 1 or temp_state == 1 \
- or bp_state == 1 or bg_state == 1:
- if self.working_led_mode != led_mode.HEALTH_LED_MODE:#如果不在亮微聊模式需要亮微聊模式
- EventMesh.debug_log("set_hr_blood_led!!!!!!!!!!")
- VC_satrt_flag = EventMesh.publish("get_VC_satrt_flag")#获取vc是否在运行
- if VC_satrt_flag == 0:
- str_buf = [1,2,10*70*2,led_mode.FLASH_LIGHT_RED_GREEN,10,5,led_mode.HEALTH_LED_MODE]#測量中
- self.timer_state = 0
- self.timer.stop()
- self.open_timer(str_buf)
- print("open_timer4!!!!!!!!!!!!!!!!!!")
- self.working_led_mode = led_mode.HEALTH_LED_MODE
- return
- else:
- EventMesh.debug_log("set_hr_blood_led stop!!!!!!!!!!")
- return
- else:# 0 2都认为其他模式灯亮 不是测试健康灯
- self.working_led_mode = led_mode.OTHER_MEASURE_MODE
- self.open_timer(msg)
+ # wechat_state = EventMesh.publish("get_wechat_state")
+ # temp_state = EventMesh.publish("get_temp_state")
+ # bp_state = EventMesh.publish("get_bp_state")
+ # bg_state = EventMesh.publish("get_bg_state")
+ # if wechat_state != None: #获取微聊状态
+ # if wechat_state == 1:
+ # if self.working_led_mode != led_mode.WECHAT_LED_MODE:#如果不在亮微聊模式需要亮微聊模式
+ # self.timer_state = 0
+ # self.timer.stop()
+ # str_buf = [1,1,1,led_mode.LONG_LIGHT_GREEN,4,2,led_mode.WECHAT_LED_MODE]
+ # self.open_timer(str_buf)
+ # print("open_timer3!!!!!!!!!!!!!!!!!!")
+ # self.working_led_mode = led_mode.WECHAT_LED_MODE
+ # return
+ # else:
+ # return
+ # else:
+ # self.wechat_btn = 0
+ # # if get_hr_state != 0 or get_spo2_state != 0 or temp_state != 0 or bp_state != 0 or bg_state != 0 :#心率血氧优先级高
+ # if get_hr_state == 1 or get_spo2_state == 1 or temp_state == 1 \
+ # or bp_state == 1 or bg_state == 1:
+ # if self.working_led_mode != led_mode.HEALTH_LED_MODE:#如果不在亮微聊模式需要亮微聊模式
+ # EventMesh.debug_log("set_hr_blood_led!!!!!!!!!!")
+ # # VC_satrt_flag = EventMesh.publish("get_VC_satrt_flag")#获取vc是否在运行
+ # # if VC_satrt_flag == 0:
+ # str_buf = [1,2,10*70*2,led_mode.FLASH_LIGHT_RED_GREEN,10,5,led_mode.HEALTH_LED_MODE]#測量中
+ # self.timer_state = 0
+ # self.timer.stop()
+ # self.open_timer(str_buf)
+ # print("open_timer4!!!!!!!!!!!!!!!!!!")
+ # self.working_led_mode = led_mode.HEALTH_LED_MODE
+ # return
+ # else:
+ # EventMesh.debug_log("set_hr_blood_led stop!!!!!!!!!!")
+ # return
+ # else:# 0 2都认为其他模式灯亮 不是测试健康灯
+ # self.working_led_mode = led_mode.OTHER_MEASURE_MODE
+ print("set_led_time_arg2:{}".format(msg))
+ self.open_timer(msg=msg)
close_led_white()
def stop_led_timer(self, event, msg):
+ print("stop_led_timer1")
EventMesh.publish("close_all_led")#关闭所有的灯
def close_all_led(self, event, msg):
EventMesh.subscribe("clear_manual_hr_flag", self.clear_manual_hr_flag)
EventMesh.subscribe("clear_manual_temp_flag", self.clear_manual_temp_flag)
+ EventMesh.subscribe("get_usb_state", self.get_usb_state)
EventMesh.subscribe("clear_usb_flag", self.clear_usb_flag)
EventMesh.debug_log("post_processor_after_instantiation:2")
+
+
self.timer_push_schedule.start(10000, 1, self.publish_event_cb)
# self.timer_pusht_test.start(2000, 1, self.publish_test)
# self.battery_push_schedule.start(3000, 1, self.__check_battery)
EventMesh.debug_log("charge R_CHARGE_in")
EventMesh.publish("clear_usb_flag")
EventMesh.publish("set_usb", 1)#
+ EventMesh.publish("close_hr_sop2")#关闭健康测试
pass
elif status == 0:#拨出usbs
EventMesh.publish("clear_usb_flag")
- def get_usb_state(self):
+ def get_usb_state(self, event = None, msg= None):
state = self.usb.getStatus()
if state == -1:
state = 0
- return state
+ return state#测试临时关闭
+ #return 0
def wifiScan_cb(args):
Version="V1.0"
-Build_Time="2025-04-21 18:21:57"
+Build_Time="2025-04-22 15:53:08"
projects / mirrors / ASR1603_G5.git / commitdiff