[tool] spi ws2812

README.md

@@ -23,7 +23,7 @@ sudo apt-get install -y git swig python3-pip python3-dev portaudio19-dev libsndf
 ```
 // 克隆项目到本地 https or ssh
 git clone http://43.132.157.186:3000/gaohz/TakwayBoard.git
-cd TakwayBoard && git checkout release/v0.9.1
+cd TakwayBoard
 sudo pip install -v -e .
 ```
 
@@ -55,4 +55,3 @@ sudo python3 setup.py install
 ```
 sudo python3 ws_client.py
 ```
-

takway/audio_utils.py

@@ -292,7 +292,7 @@ class HDRecorder(BaseRecorder):
             if board == 'v329':
                 self.hardware = V329(hd_trigger, hd_detect_threshold)
             elif board == 'orangepi':
-                self.hardware = OrangePi(hd_trigger, hd_detect_threshold, enable_start_light=True)
+                self.hardware = OrangePi(hd_trigger, hd_detect_threshold)
         print(f"Using {hd_trigger} as hardware trigger.")
         
     def wait_for_hardware_pressed(self):

takway/board/orangepi.py

@@ -25,9 +25,8 @@ except:
 '''
 
 class OrangePi(BaseHardware):
-    def __init__(self, hd_trigger='button', hd_detect_threshold=50, enable_start_light=False):
+    def __init__(self, hd_trigger='button', hd_detect_threshold=50):
         super().__init__(hd_trigger, hd_detect_threshold)
-        self.enable_start_light = enable_start_light
         
         self.LED_PIN_red = 0
         self.LED_PIN_blue = 1
@@ -69,8 +68,7 @@ class OrangePi(BaseHardware):
         last_status_1 = False
         press_time_1 = None
         
-        if self.enable_start_light:
+        self.start_status_light()
-            self.start_status_light()
         while True:
             button_status_1 = True if wiringpi.digitalRead(self.BUTTON_PIN_1) == 0 else False
             # print(f"{datetime.now()}: BUTTON_PIN_1: {wiringpi.digitalRead(self.BUTTON_PIN_1)}")

tools/orangepi_adc_io.py

@@ -0,0 +1,15 @@
+from takway.board import OrangePi
+import time
+
+import wiringpi
+from wiringpi import GPIO
+
+if __name__ == '__main__':
+    
+    ADC_Pin = 6
+    wiringpi.wiringPiSetup()
+    wiringpi.pinMode(ADC_Pin, GPIO.INPUT)
+    
+    while True:
+        print(wiringpi.digitalRead(ADC_Pin))
+        time.sleep(0.1)

tools/orangepi_rpi_ws2812.py

@@ -0,0 +1,36 @@
+from rpi_ws281x import PixelStrip, Color
+import time
+
+# LED strip configuration:
+LED_COUNT = 8      # Number of LED pixels.
+LED_PIN = 16      # GPIO pin connected to the pixels (18 uses PWM!).
+# LED_PIN = 10      # GPIO pin connected to the pixels (10 uses SPI /dev/spidev0.0).
+LED_FREQ_HZ = 800000  # LED signal frequency in hertz (usually 800khz)
+LED_DMA = 10       # DMA channel to use for generating signal (try 10)
+LED_BRIGHTNESS = 255  # Set to 0 for darkest and 255 for brightest
+LED_INVERT = False   # True to invert the signal (when using NPN transistor level shift)
+LED_CHANNEL = 0     # set to '1' for GPIOs 13, 19, 41, 45 or 53
+
+# Create NeoPixel object with appropriate configuration.
+strip = PixelStrip(LED_COUNT, LED_PIN, LED_FREQ_HZ, LED_DMA, LED_INVERT, LED_BRIGHTNESS, LED_CHANNEL)
+# Intialize the library (must be called once before other functions).
+strip.begin()
+
+def color_chase(color, wait):
+    """颜色追逐效果"""
+    for i in range(strip.numPixels()):
+        strip.setPixelColor(i, color)
+        strip.show()
+        time.sleep(wait)
+
+try:
+    while True:
+        # 渐变色追逐效果
+        for j in range(3):
+            color_chase(Color(255, 0, 0), 0.1)  # 红色
+            color_chase(Color(0, 255, 0), 0.1)  # 绿色
+            color_chase(Color(0, 0, 255), 0.1)  # 蓝色
+
+except KeyboardInterrupt:
+    color_chase(Color(0, 0, 0), 0.1)  # 关闭时变为黑色
+    exit(0)

tools/orangepi_spi_ws2812.py

@@ -0,0 +1,133 @@
+import wiringpi
+import argparse
+import time
+ 
+parser = argparse.ArgumentParser(description='')
+parser.add_argument("--channel", type=int, default=1, help='specify the spi channel')
+parser.add_argument("--port", type=int, default=0, help='specify the spi port')
+parser.add_argument("--speed", type=int, default=6400000, help='specify the spi speed')
+parser.add_argument("--mode", type=int, default=0, help='specify the spi mode')
+args = parser.parse_args()
+ 
+# 0码 t0H 220ns~380ns
+#（python 中GPIO口无法翻转这么快，python 可能执行一行代码需要800ns）
+# ws2812 波特率如果设置为800kHZ, 1/0.8M=1.25us (0码或者1码所需要的时间)
+ 
+# 方案1.
+# 使用SPI产生ws2812时序，产生1码和0码，则可以使用传输一个byte 来代表一个0码或者1码，如下：
+# 1111 1000 0xF8  #1码 高电平时间长，低电平时间短
+# 1100 0000 0xC0  #0码 高电平时间短，低电平时间长
+# 则得出SPI 波特率应该设置为800K*8
+ 
+# 方案2. 见led_bak.py
+# 使用SPI产生ws2812时序，产生1码和0码，则可以使用传输一个半个byte 来代表一个0码或者1码，如下：
+# 1100 0x0c #1码 高电平时间长，低电平时间短
+# 1000 0x08 #0码 高电平时间短，低电平时间长
+# 则得出SPI 波特率应该设置为800K*4
+ 
+ 
+sig_1 = 0xf8
+sig_0 = 0xc0
+ 
+ 
+def flatten_arrays(arrays):
+    return [element for sublist in arrays for element in sublist]
+ 
+ 
+class WS2812:
+    def __init__(self, led_num=10):
+        self.led_num = led_num
+        self.ws2812_data = [[sig_0 for _ in range(24)] for _ in range(led_num)]
+        wiringpi.wiringPiSPISetupMode(args.channel, args.port, args.speed, args.mode)
+ 
+        # 设置一下下拉，否则第一盏灯的时序可能不正确
+        wiringpi.wiringPiSetup()
+        wiringpi.pullUpDnControl(11, 1)
+ 
+        print("ws2812_data len", len(self.ws2812_data))
+        print("spi mode: 0x%x" % args.mode)
+        print("max speed: %d Hz (%d KHz)\n" % (args.speed, args.speed / 1000), end='')
+ 
+    def ws2812_send_data(self):
+        default_tx = self.ws2812_data
+        wiringpi.wiringPiSPIDataRW(args.channel, bytes(flatten_arrays(default_tx)))
+ 
+    def ws2812_light_led(self, red, green, blue, pix_led):
+        default_tx = self.ws2812_data
+        color = green << 16 | red << 8 | blue
+        for i in range(24):
+            if color >> (24 - i - 1) & 1:
+                default_tx[pix_led][i] = sig_1
+            else:
+                default_tx[pix_led][i] = sig_0
+ 
+    def ws2812_light_one_led(self, red, green, blue, pix_led):
+        self.ws2812_light_led(red, green, blue, pix_led)
+        self.ws2812_send_data()
+ 
+    def ws2812_light_all_led(self, red, green, blue):
+        for i in range(self.led_num):
+            self.ws2812_light_led(red, green, blue, i)
+ 
+        self.ws2812_send_data()
+ 
+    def ws2812_rainbow(self):
+        colors = [[0xff, 0, 0], [0, 0xff, 0], [0, 0, 0xff]]
+        for i in range(self.led_num):
+            cur_color = colors[i % 3]
+            self.ws2812_light_led(cur_color[0], cur_color[1], cur_color[2], i)
+        self.ws2812_send_data()
+ 
+    def ws2812_water_lamp(self, red, green, blue, interval_time):
+        self.ws2812_shutoff_all()
+ 
+        for i in range(self.led_num):
+            self.ws2812_light_one_led(red, green, blue, i)
+            time.sleep(interval_time)
+ 
+        self.ws2812_shutoff_all()
+ 
+    def ws2812_shutoff_led(self, n):
+        self.ws2812_light_led(0, 0, 0, n)
+        self.ws2812_send_data()
+ 
+    def ws2812_shutoff_all(self):
+        for i in range(self.led_num):
+            self.ws2812_light_led(0, 0, 0, i)
+ 
+        self.ws2812_send_data()
+
+    def ws2812_gradient_run(self, start_red, start_green, start_blue, end_red, end_green, end_blue, steps, delay):
+	    for step in range(steps):
+	        # 计算当前步的颜色
+	        curr_red = start_red + (end_red - start_red) * step // steps
+	        curr_green = start_green + (end_green - start_green) * step // steps
+	        curr_blue = start_blue + (end_blue - start_blue) * step // steps
+	        
+	        # 清除当前颜色并应用新的渐变色
+	        self.ws2812_shutoff_all()
+	        for i in range(self.led_num):
+	            self.ws2812_light_led(curr_red, curr_green, curr_blue, (i + step) % self.led_num)  # 循环显示渐变色
+	        
+	        self.ws2812_send_data()
+	        time.sleep(delay)
+ 
+if __name__ == '__main__':
+    ws2812 = WS2812(8)
+ 
+    while True:
+        # 示例：从蓝色渐变到红色，再从红色渐变回蓝色，每种颜色渐变10步，每步间隔0.1秒
+        ws2812.ws2812_gradient_run(0, 0, 0xff, 0xff, 0, 0, 10, 0.02)
+        time.sleep(1)  # 等待1秒后开始反向渐变
+        ws2812.ws2812_gradient_run(0xff, 0, 0, 0, 0, 0xff, 10, 0.02)
+        time.sleep(1)  # 再次等待1秒，形成完整的循环
+        ws2812.ws2812_light_all_led(0, 0, 0xff)
+        time.sleep(0.5)
+        ws2812.ws2812_light_all_led(0xff, 0, 0)
+        time.sleep(0.5)
+        ws2812.ws2812_light_all_led(0, 0xff, 0)
+        time.sleep(0.5)
+        ws2812.ws2812_shutoff_all()
+        time.sleep(0.5)
+        ws2812.ws2812_water_lamp(0xff, 0, 0, 0.2)
+        time.sleep(0.5)

tools/orangepi_uart5.py

@@ -0,0 +1,40 @@
+import serial
+import time
+
+# UART配置
+uart_port = '/dev/ttyS5'  # UART端口路径
+baud_rate = 115200           # 波特率
+
+
+
+try:
+    # 初始化串口连接
+    ser = serial.Serial(uart_port, baud_rate, timeout=1)
+    print("UART初始化成功，开始读取数据...")
+    
+    while True:
+        # 尝试读取一行数据
+        ori_data = ser.readline()  # 读取一行原始数据
+        print(f"原始数据: {ori_data}")
+        data = ori_data.decode('utf-8').strip()  # 读取一行并解码
+        print(f"解码后数据: {data}, 类型: {type(data)}")
+        if data:  # 如果读取到数据
+            try:
+                num = int(data)  # 尝试将读取的数据转换为整数
+                print(f"接收到的数字: {num}")
+                # print(f"接收到的字符串: {data}")
+                battery_level = num / 4096 * 4.8 /4.2 * 100  # 计算电池电量
+                print(f"电池电量: {battery_level}%")
+            except ValueError:  # 如果转换失败，说明不是有效的数字
+                print("接收到的数据无法转换为数字")
+        time.sleep(1)  # 每隔一秒检查一次新数据
+    
+except KeyboardInterrupt:
+    print("程序被中断，正在关闭UART连接...")
+    ser.close()
+    print("UART连接已关闭。")
+
+except Exception as e:
+    print(f"发生错误: {e}")
+    if 'ser' in locals():
+        ser.close()

ws_client.py

@@ -6,43 +6,82 @@ import platform
 
 if __name__ == '__main__':
     
-    server_url = 'ws://43.132.157.186:8002/api/chat/streaming/temporary'
+    server_url = 'ws://222.195.90.129:8001/api/chat/streaming/temporary'
     
-    
+    # session_id = 'de9dc06c-2d74-42f8-9c11-9797f9fe0d01' # 麓旬
-    # session_id = 'ef3fd24a-86d9-47c5-94a9-574628ea5a43' # toy1
+    # session_id = '3f7c2d8f-dc8a-4f1b-9fe0-6bf56c759a41' # 砚文
-    # session_id = '984651b2-828e-41b7-8a5a-b3ade108dfae' # toy2
+    # session_id = '1fd56ce7-f21e-4eb4-8dc8-3a17960ae328' # 越哥
-    # session_id = 'd1427250-adb2-40d6-b41a-0ffcb029a8f8' # toy3
+    # session_id = 'bfcc4bbc-ac94-4604-8fd7-50545b8c6a5f' # 楼总
-    # session_id = '08312487-8751-417e-a89b-2d55558c485c' # toy4
+    # session_id = '469f4a99-12a5-45a6-bc91-353df07423b6' # 鸿志
-    # session_id = '4c75a357-d728-47db-93d3-f1587b180e86' # toy5
+
-    # session_id = '33053669-2f6b-442a-8658-b342552b3484' # toy6
+    # session_id = '9ad8f855-f87c-41c8-a1f3-12e39d013c8c' # 1 积木 刚木
+    # session_id = '13e14fd9-a212-4cf1-a09b-5890344c769c' # 2
+    # session_id = 'e6e76e50-cb4c-4fe7-8660-8278e712c49b' # 3
+    # session_id = '6cbbfa55-58e3-437a-94c4-65eee1fa88a6' # 4
+    # session_id = 'c14cfd9f-3609-46f9-90aa-ba5a80b65015' # 5
     
     
     excute_args = {}
+    # excute_args = {'enable': True}
+    
     
     system = platform.system()
     if system == 'Windows':
         print("WebSocketClinet runs on Windows system.")
         board = None
     elif system == 'Linux':
+        # board = 'v329'
         board = 'orangepi'
-        # ACCESS_KEY = 'hqNqw85hkJRXVjEevwpkreB8n8so3w9JPQ27qnCR5qTH8a3+XnkZTA==' # gaohz
+        # ACCESS_KEY = 'neOpatzY/mTzSyxdHs+ajNfpY/7SX1WrlqP/D6+5Km8THUxfZdcauQ==' # luxun
+        # ACCESS_KEY = 'KwdWw3V5X9Dz0c9x+5HmGAMi7GbW0kvnaGOAPloIAYwhp06jNt5baw==' # yuyue
+        # ACCESS_KEY = 'GPFKn+Z9LHGh8yZNfWkLUYRixnrsyY+5w8KN3rpl6sw+Bi7XIqbgTw==' # gaohz (hzgao2000@gmail.com)
+        # ACCESS_KEY = 'Zo6Vx8YElrjMuA30K9yWMZvD06gcgOfyTi4rC6PPoqWL9mkhA/N/Lg==' # loujc
+        # ACCESS_KEY = 'Ce5DkiDBQ9B7QNhebpQJyw/f3nhlqO5960yDUfkNAcl9gSj7RzIt6w==' # jinzc
     
     mircophone_device = None
     speaker_device = None
     
-    keywords = ['芭比']
-    keyword_paths = None
-    model_path = None
-    # 
-    keyword_paths = ["models/芭比_zh_raspberry-pi_v3_0_0/芭比_zh_raspberry-pi_v3_0_0.ppn"] 
-    model_path = "models/porcupine_params_zh.pv"
     
-    hd_trigger = 'button'
+    if board == 'v329':
-    
+        import gpiod as gpio
-    mircophone_device = 2
-    speaker_device = 2
         
+        keywords = ['hey google', 'ok google']
+        keyword_paths = None
+        model_path = None
+        
+        keywords = ['可莉可莉']
+        keyword_paths = [r"picovoice_models/可莉可莉_zh_raspberry-pi_v3_0_0.ppn"]
+        model_path = r"picovoice_models/porcupine_params_zh.pv"
+        
+        hd_trigger = 'button'
+        player = 'maixsense'
+    elif board == 'orangepi':
+        
+        # keywords = ['hey google']
+        keywords = ['哔卡', '刚木'] # '芭比', , '星云'
+        keyword_paths = None
+        model_path = None
+        # 
+        keyword_paths = ['models/哔卡_zh_raspberry-pi_v3_0_0/哔卡_zh_raspberry-pi_v3_0_0.ppn', 'models/刚木_zh_raspberry-pi_v3_0_0/刚木_zh_raspberry-pi_v3_0_0.ppn'] 
+        # "models/芭比_zh_raspberry-pi_v3_0_0/芭比_zh_raspberry-pi_v3_0_0.ppn", 
+        # , 'models/星云_zh_raspberry-pi_v3_0_0/星云_zh_raspberry-pi_v3_0_0.ppn'
+        model_path = "models/porcupine_params_zh.pv"
+        
+        hd_trigger = 'button'
+        
+        mircophone_device = 2
+        speaker_device = 2
+        
+    else:
 
+        keywords = ['hey google', 'ok google']
+        keyword_paths = None
+        model_path = None
+        
+        
+        hd_trigger = 'keyboard'
+        player = 'opencv'
+        
     
     import argparse
     parser = argparse.ArgumentParser()
@@ -84,7 +123,7 @@ if __name__ == '__main__':
     parser.add_argument('--IN_CHANNELS', type=int, default=1, help='Audio channels')
     parser.add_argument('--IN_filename', type=str, default=None, help='Audio file name')
     parser.add_argument('--IN_frames_per_buffer', type=int, default=512, help='Frames per buffer')
-    parser.add_argument('--min_stream_record_time', type=float, default=0.3, help='Min stream record time, sec')
+    parser.add_argument('--min_stream_record_time', type=float, default=0.5, help='Min stream record time, sec')
     parser.add_argument('--max_slience_time', type=int, default=10, help='Max slient time when recording, sec')
     parser.add_argument('--min_act_time', type=float, default=0.3, help='Min inactive time, sec')   # 等待多少秒沉默就发送音频
     parser.add_argument('--mircophone_device', type=int, default=mircophone_device, help='Microphone device index')