self.check_rest_timer_count = 0
#双击逻辑
- self.last_time = 0 #上次时间
- self.gap_time = 1500 #双击间隔时间(ms)
- self.double_flag = 0 #双击标志
- self.double_release_timer = osTimer()#双击释放定时器
+ self.is_double_timer = osTimer()#单/双击判断定时器
+ self.double_click_threshold = 2000#双击间隔时间
+ self.press_count = 0#按下次数
print("AI_dev:end")
def ntptime_cb(self):
#ntptime.settime(8)
def keyAI_extint_cb(self,args):#call 2
try:
self.KeyAI_status = int(args[1])
- if self.KeyAI_status == 0:
- self.KeyAI_status = 1
- else:
- self.KeyAI_status = 0
- current_time = utime.ticks_ms()
- if self.KeyAI_status == 0:#按下
- if self.KeyAI_press == 0 :
- time_tiff = utime.ticks_diff(current_time,self.last_time)#计算时间差（处理溢出
- print("双击时间差为：{}".format(time_tiff))
- if time_tiff < 0 :
- print("双击时间已经溢出")
- if time_tiff > 0 and current_time - self.last_time <= self.gap_time:#鉴定为双击
- print("AI_key_double_press DOUBLE DOUBLE ")
- self.double_flag = 1#切换标志位
- self.KeyAI_long_press = 0
- self.KeyAI_press = 0
- self.KeyAI_timer_count = 0
- EventMesh.publish("publish_AI_btnInfo",{"func":self.AI_key_double_press})
- # 启动双击释放检测计时器
- self.double_release_timer.start(500, 0, self.double_release_timeout)
- return
- elif time_tiff > 0 and current_time - self.last_time > self.gap_time:#判断为单击
- EventMesh.publish(constants.EVENT.SYSTEM_UP)#唤醒休眠
- self.extint_AI.disable()
- self.KeyAI_press = 1
- self.KeyAI_long_press = 0
- self.double_flag = 0
- self.KeyAI_timer.start(100, 1, self.KeyAI_timer_cb)
- self.KeyAI_timer_count = 0
- #EventMesh.publish("AI_key_press")
- screen_name = EventMesh.publish("get_screen_NAME")
- if screen_name != "system_init" and screen_name != "poweron":
- if dev_type == "7PIN" or dev_type == "7PIN_coze" :
- self.aud.aud_tone_play(10,200)#播放按键音
- EventMesh.publish("publish_AI_btnInfo",{"func":self.AI_key_press})
- self.last_time = current_time
- else: # 释放
- if self.double_flag:
- # 双击释放处理
- self.double_flag = 0
- EventMesh.publish("publish_AI_btnInfo", {"func": self.AI_key_double_release})
- self.extint_AI.enable()
- self.KeyAI_timer.stop()
- self.KeyAI_timer_count = 0
- self.KeyAI_press = 0
- return
+ if self.KeyAI_status == 1: # 仅处理"按下瞬间"的事件
+ self.press_count += 1 # 按下次数+1
+ current_time = utime.ticks_ms() # 当前时间戳(ms)
+ if self.press_count == 1: # 第一次按下
+ self.first_press_time = current_time # 记录第一次按下时间
+ self.AI_key_press() # 触发单击业务（无音频，音频确定后播放）
+ # 启动定时器：超过阈值未第二次按下则判定为单击
+ self.is_double_timer.start(self.double_click_threshold, 0, self.is_double_judge)
+ print("第一次按下，等待{}ms内第二次按下".format(self.double_click_threshold))
+
+ elif self.press_count == 2: # 第二次按下
+ # 计算与第一次按下的间隔
+ interval = utime.ticks_diff(current_time, self.first_press_time)
+ if interval <= self.double_click_threshold: # 间隔符合双击
+ EventMesh.publish("AI_VC_Close")
+ self.AI_key_double_press() # 触发双击业务
+ # 停止定时器（已判定为双击，无需再检查单击）
+ self.is_double_timer.stop()
+ # 重置计数（准备下一次判断）
+ self.press_count = 0
+ else: # 间隔过长，视为新的第一次按下
+ self.first_press_time = current_time # 更新第一次按下时间
+ self.press_count = 1 # 重置为第一次
+ # 重启定时器
+ self.is_double_timer.stop()
+ self.is_double_timer.start(self.double_click_threshold, 0, self.is_double_judge)
+ print("间隔过长，视为新的第一次按下")
+
except Exception as e:
usys.print_exception(e)
- def double_release_timeout(self, *args, **kwargs):#双击释放超时处理
- print("定时器未关闭！！！！！！！！")
- if self.double_flag:
- #如果超时仍未释放，重置双击状态
- self.double_flag = 0
- self.double_release_timer.stop()
-
- def AI_key_double_release(self):
- EventMesh.publish("publish_AI_btnInfo",{"func":self.AI_key_double_press})
+ def is_double_judge(self, *args):
+ if self.press_count == 1: # 仅一次按下，判定为单击
+ screen_name = EventMesh.publish("get_screen_NAME")
+ if screen_name == "system_init" or screen_name == "poweron" :
+ return
+ EventMesh.publish("audio_montage_play",["chenggong.mp3"]) #进入VC音频
+ # 重置计数（无论何种情况，超时后清零）
+ self.press_count = 0
def KeyAI_timer_cb(self, *args, **kwargs):
# EventMesh.publish("publish_AI_btnInfo",{"func":self.KeyAI_timer_cb_})
EventMesh.subscribe("draw_handle", self.draw_handle)
EventMesh.subscribe("dmp_update", self.dmp_update)
EventMesh.subscribe("update_volume", self.update_volume)
+ EventMesh.subscribe("AI_VC_Close", self.Close)
EventMesh.publish("close_led")
self.bot_name = "柒先生"
#EventMesh.publish("set_audio_state", 0)
print('audio-play finish.!!!!!!')
def AI_key_double_press(self, event, msg):#双击按键
+ screen_name = EventMesh.publish("get_screen_NAME")
+ if screen_name != "UI_ai":
+ return
print("AI_key_double_press")
# self.Close_flag = 1
self.Close()
# EventMesh.publish("return_prev_screen")
def AI_key_press(self, event, msg):#按键按下
+ screen_name = EventMesh.publish("get_screen_NAME")
+ if screen_name == "system_init" or screen_name == "poweron" :
+ return
# 测试闹钟功能
# EventMesh.publish("AI_alarm_open")
# EventMesh.publish("AI_add_alarm",[{"alarm_type":"单次闹钟","alarm_timestamp":1753071946,"repeat_day":"2025-07-21","repeat_min":0,"valid_time":"","alarm_timer":"12:25","alarm_content":"开会"}])
lv.img.cache_invalidate_src(None)
lv.img.cache_set_size(8)
EventMesh.publish("load_screen", {"screen": "UI_ai"})#ai画面
- EventMesh.publish("audio_montage_play",["chenggong.mp3"])
+ # EventMesh.publish("audio_montage_play",["chenggong.mp3"])
# if dev_type != "7PIN":
# self.chenggong_tone()
# self.Close()
self.socket_send_queue_id = None
self.bat_count = 0
self.reconnect_count = 0 #连续重连次数 默认十次 不过就不重连了 等下一个心跳重连
- self._1S_timer = osTimer()
+ self._20S_timer = osTimer()
self._1scnt = 0
- self._5S_timer = osTimer()
+ self._5M_timer = osTimer()
self._10Stimer = osTimer()
self.timer_id = 0
self.start_voice_flag = 0
#对话状态，一秒检查一次
self.last_data = None # 保存上一次的数据状态
self.current_data = None # 当前数据
-
- self.send_recv_flag = 0#线程开启状态
EventMesh.subscribe("REPORT_ALARM_TO_IOT",self.report_alarm_to_iot)
def report_alarm_to_iot(self,event,msg):
EventMesh.publish("publish_AI_btnInfo",{"func":self.send_packet_iot,"data":WEB_CMD_ENUM.POST_DEVICE_STATE_S_CMD})#上报VC状态
def send_unbind_dev_to_iot(self,event,msg):
- self._5S_timer.start(30000,1,self._1M_timer_cb)
- self._1S_timer.start(1000,1,self._1S_timer_cb)
+ self._5M_timer.start(60000*5,1,self._5M_timer_cb)#更新信息
+ self._20S_timer.start(20000,1,self._20S_timer_cb)#重连
# self._10Stimer.start(50000,1,self._10S_timer_cb)
- def _1M_timer_cb(self,*argc):
+ def _5M_timer_cb(self,*argc):
EventMesh.publish("heart_IOT")#开启IOT心跳包
EventMesh.publish("send_device_information")#发送设备信息
pass
self.IOT_sock.connect(('iot.socket.thisiot.com', 7777))
#设置为非阻塞
self.IOT_sock.setblocking(False)
-
- if self.send_recv_flag == 0:#线程未开启过
- self.send_recv_flag = 1
- self.reconnect_count = 0
+ self.reconnect_count = 0
+ # print("self.socket_send_queue_id = {} , self.iot_protoclo_recv_id = {}".format(self.socket_send_queue_id,self.iot_protoclo_recv_id))
+ if self.socket_send_queue_id == None:
_thread.stack_size(self.thread_buf_size+8*1024) #_thread.stack_size(16*1024)
self.socket_send_queue_id = _thread.start_new_thread(self.socket_send_queue_thread, ())
_thread.stack_size(8*1024)
-
+ if self.iot_protoclo_recv_id == None:
_thread.stack_size(self.thread_buf_size+8*1024)
self.iot_protoclo_recv_id = _thread.start_new_thread(self.iot_protocol_recv, ())
_thread.stack_size(8*1024)
- socket_log.info("self.IOT_sock thread init!!!!!!!!!")
+ socket_log.info("self.IOT_sock thread init!!!!!!!!!")
if self.IOT_sock is not None:
return ustruct.pack('<H', crc & 0xFFFF)
def close_socket_timer(self,event,msg):
- self._1S_timer.stop()
- self._5S_timer.stop()
+ self._20S_timer.stop()
+ self._5M_timer.stop()
- def _1S_timer_cb(self,*argc):
+ def _20S_timer_cb(self,*argc):
self.current_data = EventMesh.publish("screen_get_ai") # 当前数据
if self.current_data is not self.last_data :
print("VC状态改变")
_thread.stop_thread(self.socket_send_queue_id)
except Exception as e:
print("iot_send_queue_close !!!!!!!!")
- self.send_recv_flag = 0
print("iot_send_queue_close 1111!!!!!!!!")
self.socket_send_queue_id = None
projects / mirrors / ASR1603_G5.git / commitdiff