led灯控，重连

Author: 陈志和

README.md

@@ -18,6 +18,12 @@ QuecPython introduces the Xiaozhi Platform AI Chatbot solution. This solution is
 
 This demo uses an AI development board equipped with the EC800MCNLE module.
 
+
+
+**Note: Business access is prohibited when the USB is plugged in. After unplugging the USB, long press the button to start the business. If you need to view the logs, you can connect a USB after startup to view the logs**
+
+
+
 ## Features
 
 - Supports voice interruption/barge-in.

README_zh.md

@@ -18,6 +18,12 @@ QuecPython 推出了小智平台 AI 聊天机器人解决方案。该方案基
 
 本案例采用搭载 EC800MCNLE 模组的 AI 开发板。
 
+
+
+**注：插 USB 状态下禁止进入业务；拔掉 USB 后，长按按键（红黑束线）即可启动业务；若需查看日志可在启动后连接USB查看日志**
+
+
+
 ## 功能特性
 
 - 支持语音中断/打断。

src/OTA_test.py

@@ -33,6 +33,7 @@ def __init__(self,mac):
         }
         self.url = "https://api.tenclass.net/xiaozhi/ota/"
         # self.url ="http://xiaozhi.pandastroller.com:8002/xiaozhi/ota/"
+        # self.url = "http://101.37.79.117:8003/xiaozhi/ota/"
         self.response = None
         self.UDP_IP = None
         self.UDP_PORT = None

src/main.py

@@ -4,13 +4,15 @@
 import utime
 import gc
 from machine import ExtInt,Pin
+from misc import PowerKey, Power
 from usr.protocol import WebSocketClient
-from usr.utils import ChargeManager, AudioManager, NetManager, TaskManager, name
+from usr.utils import ChargeManager, AudioManager, NetManager, TaskManager, name, Button
 from usr.threading import Thread, Event, Condition
 from usr.logging import getLogger
 import sys_bus
 import _thread
 import ujson as json
+from misc import USB
 # from usr import UI
 
 
@@ -74,22 +76,25 @@ def __blink_thread_worker(self):
                 if self.__count is not None:
                     self.__count -= 1
 
+enable_flag=0
+
 class Application(object):
 
     def __init__(self):
-        # 初始化唤醒按键
-        self.talk_key = ExtInt(ExtInt.GPIO20, ExtInt.IRQ_RISING_FALLING, ExtInt.PULL_PU, self.on_talk_key_click, 50)
 
         # 初始化 led; write(1) 灭； write(0) 亮
         self.wifi_red_led = Led(33)
-        self.wifi_green_led = Led(32)
-        self.power_red_led = Led(39)
-        self.power_green_led = Led(38)
+        self.wifi_green_led = Led(32) #ai
+        self.power_red_led = Led(39) 
+        self.power_green_led = Led(38) #power
         self.lte_red_led = Led(23)
-        self.lte_green_led = Led(24)
-        self.led_power_pin = Pin(Pin.GPIO27, Pin.OUT, Pin.PULL_DISABLE, 0)
+        self.lte_green_led = Led(24)#chat
+        self.led_power_pin = Pin(Pin.GPIO27, Pin.OUT, Pin.PULL_DISABLE, 1)
         self.prev_emoj = None
-
+        # self.power_green_led.blink(500, 500)
+        self.power_key = PowerKey()
+        self.power_key.powerKeyEventRegister(lambda status: None)
+        
 
         # 初始化充电管理
         self.charge_manager = ChargeManager()
@@ -118,6 +123,15 @@ def __init__(self):
         self.__voice_activity_event = Event()
         self.__keyword_spotting_event = Event()
 
+        # self.gpio_pin = Pin(Pin.GPIO41, Pin.OUT, Pin.PULL_PD,0)
+        
+        # 初始化唤醒按键
+        self.volumedown = ExtInt(ExtInt.GPIO20, ExtInt.IRQ_RISING, ExtInt.PULL_PU, self.setvolumedown, 200)
+        # self.power_down = ExtInt(ExtInt.GPIO41, ExtInt.IRQ_RISING, ExtInt.PULL_PU, self.power_down_handle, 200)
+        self.volumeup = ExtInt(ExtInt.GPIO47, ExtInt.IRQ_RISING, ExtInt.PULL_PU, self.setvolumeup, 200)
+        # self.wakeup_key = Button(41, delay=3000, long_press_callback=self.power_down_handle, short_press_callback= self.on_talk_key_click)
+
+
     def __record_thread_handler(self):
         """纯粹是为了kws&vad能识别才起的线程持续读音频"""
         logger.debug("record thread handler enter")
@@ -126,13 +140,34 @@ def __record_thread_handler(self):
             utime.sleep_ms(5)
         logger.debug("record thread handler exit")
 
-
+    def on_led(self):
+        self.wifi_green_led.on()
+        self.lte_green_led.on()
+        self.power_green_led.on()
+        
+    def off_led(self):
+        self.wifi_green_led.off()
+        self.lte_green_led.off()
+        self.power_green_led.off()
+        
     def start_kws(self):
         self.audio_manager.start_kws()
         self.__record_thread_stop_event.clear()
         self.__record_thread = Thread(target=self.__record_thread_handler)
         self.__record_thread.start(stack_size=64)
 
+    def setvolumeup(self,args):
+        volume=self.audio_manager.setvolume_up()
+        print("setvolumeup:", volume)
+    def setvolumedown(self,args):
+        volume=self.audio_manager.setvolume_down()
+        print("setvolumedown:", volume)
+        
+    def power_down_handle(self):
+        logger.info("power down")
+        Power.powerDown()
+        
+        
     def stop_kws(self):
         self.__record_thread_stop_event.set()
         self.__record_thread.join()
@@ -154,10 +189,10 @@ def __working_thread_handler(self):
 
     def __chat_process(self):
         global name
+        self.power_green_led.on()
         self.start_vad()
         try:
             with self.__protocol:
-                self.power_red_led.on()
                 self.__protocol.hello()
                 self.__protocol.wakeword_detected(name)
                 is_listen_flag = False
@@ -183,21 +218,26 @@ def __chat_process(self):
                             is_listen_flag = False
                     if not self.__protocol.is_state_ok():
                         break
-                    # utime.sleep_ms(1)
+                    utime.sleep_ms(5)
                     # logger.debug("read opus data length: {}".format(len(data)))
         except Exception as e:
             logger.debug("working thread handler got Exception: {}".format(repr(e)))
         finally:
-            self.power_red_led.blink(250, 250)
+            print("__chat_process exit")
+            self.lte_green_led.off()
+            self.wifi_green_led.off()
+            self.power_green_led.blink(500, 500)
+            self.__working_thread = None
             self.stop_vad()
+            self.start_kws()
 
-    def on_talk_key_click(self, args):
-        logger.info("on_talk_key_click: ", args)
+    def on_talk_key_click(self):
+        # logger.info("on_talk_key_click: ", args)
         if self.__working_thread is not None and self.__working_thread.is_running():
             return
         self.__working_thread = Thread(target=self.__working_thread_handler)
         self.__working_thread.start()
-        self.__keyword_spotting_event.clear()
+        self.__keyword_spotting_event.set()
 
     def on_keyword_spotting(self, state):
         logger.info("on_keyword_spotting: {}".format(state))
@@ -216,8 +256,10 @@ def on_voice_activity_detection(self, state):
         logger.info("on_voice_activity_detection: {}".format(state))
         if state == 1:
             self.__voice_activity_event.set()  # 有人声
+            self.lte_green_led.on()
         else:
             self.__voice_activity_event.clear()  # 无人声
+            self.lte_green_led.off()
 
     def on_audio_message(self, raw):
         # raise NotImplementedError("on_audio_message not implemented")
@@ -289,14 +331,52 @@ def handle_error_message(data, msg):
         # raise NotImplementedError("handle_error_message not implemented")
 
     def run(self):
-        self.charge_manager.enable_charge()
+        global enable_flag
+        self.charge_manager.disable_charge()
         self.audio_manager.open_opus()
-        self.talk_key.enable()
+        self.volumedown.enable()
+        self.volumeup.enable()
+        self.power_green_led.blink(500, 500)
+        self.wakeup_key = Button(41, delay=3000, long_press_callback=self.power_down_handle, short_press_callback= self.on_talk_key_click)
+        enable_flag = 1
         self.start_kws()
-        self.led_power_pin.write(1)
-        self.power_red_led.blink(250, 250)
-        self.stop_vad()
+        
+    
+
 
-if __name__ == "__main__":
+def usb_callback(conn_status):
+    status = conn_status
+    if not enable_flag:
+        if status == 0:
+            app.off_led()
+            Power.powerDown()
+            # enable_flag=1
+            print('USB is disconnected.')
+        elif status == 1:
+            app.on_led()
+            # enable_flag=0
+            
+            print('USB is connected.')
+     
+def power_open_handle():
+    if enable_flag: 
+        app.off_led()
+        app.run()  
+        print("power_open_handle")
+            
+
+
+if __name__ == "__main__":    
+    usb = USB()
     app = Application()
-    app.run()
+    usb.setCallback(usb_callback)
+    # 检查 USB 连接状态
+    if usb.getStatus():  # 假设 getStatus() 返回 True 表示 USB 已连接
+        app.on_led()
+        print("USB 已连接，仅启用充电业务")
+        app.charge_manager.enable_charge()  # 只启用充电业务
+        # Button(41, delay=3000, long_press_callback=power_open_handle, short_press_callback= None)
+
+    else:
+        print("USB 未连接，启动主业务")
+        app.run()

src/protocol.py

@@ -106,7 +106,7 @@ def set_callback(self, audio_message_handler=None, json_message_handler=None):
     def get_mac_address():
         mac = str(uuid.UUID(int=int(modem.getDevImei())))[-12:]
         return ":".join([mac[i:i + 2] for i in range(0, 12, 2)])
-        # return "64:e8:33:48:ec:c2"
+        # return "64:e8:33:48:ec:c9"
 
     @staticmethod
     def generate_uuid() -> str:

src/utils.py

@@ -7,13 +7,15 @@
 import checkNet
 import sys_bus
 from machine import Pin
+from machine import ExtInt
+import osTimer
 from usr.threading import PriorityQueue, Thread
 from usr.logging import getLogger
 
 
 logger = getLogger(__name__)
 volume = 7
-name = '_xiao_zhi_xiao_zhi'
+name = '_xiao_yuan_xiao_yuan'
 # ==================== 音频管理 ====================
 
 
@@ -77,13 +79,13 @@ def stop(self):
     def open_opus(self):
         self.pcm = audio.Audio.PCM(0, 1, 16000, 2, 1, 15)  # 5 -> 25
         self.opus = Opus(self.pcm, 0, 60000)  # 6000 ~ 128000
-    
+
     def close_opus(self):
         self.opus.close()
         self.pcm.close()
         del self.opus
         del self.pcm
-    
+
     def opus_read(self):
         return self.opus.read(60)
 
@@ -120,9 +122,8 @@ def new_name(self,data):
         # print("当前唤醒词：", name)
         return name
     def start_kws(self):
-        # list=["_xiao_zhi_xiao_zhi","_xiao_tian_xiao_tian","_xiao_zi_xiao_zi","_xiao_shi_xiao_shi","_xiao_si_xiao_si","_xiao_zhi_xiao_zi","_xiao_zi_xiao_zhi"]
-        # self.rec.ovkws_start("_xiao_zhi_xiao_zhi", 0.7)
-        self.rec.ovkws_start(name, 0.7)
+        list=["_xiao_tian_xiao_tian",name,"_jiang_gou_jiang_gou"]
+        self.rec.ovkws_start(list, 0.7)
 
 
     def stop_kws(self):
@@ -131,7 +132,7 @@ def stop_kws(self):
     def start_vad(self):
         self.__skip = 0
         self.rec.vad_start()
-    
+
     def stop_vad(self):
         self.rec.vad_stop()
 
@@ -247,3 +248,40 @@ def run_forever(self):
 
     def submit(self, func, args=(), kwargs={}, priority=0, title="anon"):
         self.__q.put(_Task(target=func, args=args, kwargs=kwargs, priority=priority, title=title))
+        
+       
+       
+
+
+
+class Button(object):
+
+    def __init__(self, gpio_number, delay=3000, long_press_callback=lambda: None, short_press_callback=lambda: None):
+        self.key = ExtInt(getattr(ExtInt, "GPIO{}".format(gpio_number)), ExtInt.IRQ_RISING_FALLING, ExtInt.PULL_PU, self.__callback, 150)
+        self.key.enable()
+        self.delay = delay
+        self.timer = osTimer()
+        self.start_time = None
+        self.end_time = None
+        self.long_press_callback = long_press_callback
+        self.short_press_callback = short_press_callback
+        self.is_pressed = False  # 状态变量：按键是否被按下
+        
+        
+    def __callback(self, args):
+        gpio_number, mode = args
+        if not self.is_pressed:
+            # 下降沿，按下
+            print("按键按下")
+            self.is_pressed = True
+            self.start_time = utime.ticks_ms()
+            self.timer.start(self.delay, 0, lambda args: self.long_press_callback())
+        elif self.is_pressed:
+            # 上升沿，释放
+            print("按键释放")
+            self.is_pressed = False
+            self.timer.stop()
+            self.end_time = utime.ticks_ms()
+            duration = utime.ticks_diff(self.end_time, self.start_time)
+            if duration < self.delay:
+                self.short_press_callback()