[doc] update readme

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
+cd TakwayBoard && git checkout release/v0.9
 sudo pip install -v -e .
 ```
 

takway/board/orangepi.py

@@ -2,6 +2,7 @@ from takway.board.base_hd import BaseHardware
 import threading
 from datetime import datetime
 import time
+import subprocess
 try:
     import wiringpi
     from wiringpi import GPIO
@@ -41,7 +42,7 @@ class OrangePi(BaseHardware):
         
         self.power_status = False # 单次触发按键状态
         self.long_power_status = False
-        self.short_power_status = False
+        # self.short_power_status = False
 
         self.button_init()
         self.init_hd_thread()
@@ -76,16 +77,14 @@ class OrangePi(BaseHardware):
             elif button_status_1 and not last_status_1:
                 if press_time_1:
                     press_duration = (datetime.now() - press_time_1).total_seconds()
-                    print(f"{datetime.now()}: press_duration: {press_duration}")
-                    if press_duration > 1:
-                        self.long_power_status = not self.long_power_status
-                        # print(f"{datetime.now()}: long_power_status: {self.long_power_status} {self.short_power_status}")
-                    elif press_duration > 5:
-                        time.sleep(1) # 防止短按误触发
+                    # if press_duration > 0.01:
+                    if press_duration > 3:
+                        self.set_led_on('red')
                         subprocess.Popen('sudo shutdown now', shell=True)
                     else:
-                        self.short_power_status = True
-                        # print(f"{datetime.now()}: short_power_status: {self.short_power_status} {self.long_power_status}")
+                        self.long_power_status = not self.long_power_status
+                        # self.short_power_status = True
+                    print(f"{datetime.now()}: press_duration: {press_duration}")
                     print(f"{datetime.now()}: button_status: {button_status_1}, last_button_status: {last_status_1}, power_status: {self.long_power_status}, interrupt_status: {self.short_power_status}")
                 press_time_1 = None
             # TODO: 如果button_status_1和上一帧都是True or False，则判断为无效按键，忽略
@@ -94,13 +93,15 @@ class OrangePi(BaseHardware):
             # 更新LED状态
             if self.long_power_status:
                 self.set_led_on('green')
-                if self.short_power_status:
-                    self.set_led_on('blue')
-                else:
-                    self.set_led_off('blue')
+                # if self.short_power_status:
+                    # self.set_led_on('blue')
+                # else:
+                    # self.set_led_off('blue')
             else:
                 self.short_power_status = False
                 self.set_led_off('green')
+                self.set_led_off('blue')
+                self.set_led_off('red')
             
     
     def set_led_on(self, color='red'):

takway/clients/client_utils.py

@@ -1,7 +1,7 @@
 import os
 import json
 import time
-import datetime
+from datetime import datetime
 import requests
 import struct
 from takway.common_utils import encode_bytes2str, decode_str2bytes
@@ -152,21 +152,9 @@ class BaseWebSocketClient:
     def receive_per_data(self):
         try:
             recv_data = self.websocket.recv()
+            print(f"{datetime.now()}: receive websocket data.")
         except websocket._exceptions.WebSocketConnectionClosedException:
             return None, None
-        '''
-        try:
-            recv_data = json.loads(recv_data)
-            #解析头信息，假设头信息前 8 个字节包含两个长度字段
-        except json.JSONDecodeError as e:
-            # print(f"JSONDecodeError: {e}")
-            # is_end = True
-            pass
-        except Exception as e:
-            # print(f"receive_per_data error: {e}")
-            assert isinstance(recv_data, bytes), ValueError(f"Received data is not bytes, got {type(recv_data)}.")
-        return recv_data, type(recv_data)
-        '''
         
         try:
             #解析头信息，假设头信息前 8 个字节包含两个长度字段
@@ -178,15 +166,15 @@ class BaseWebSocketClient:
             audio = recv_data[8 + text_length:8 + text_length + audio_length]
             text = json.loads(text_bytes.decode('utf-8'))
             
-            print("Received text:",text)
+            # print("Received text:",text)
             #处理音频数据，例如播放音频
-            print("Received audio(length):",len(audio))
+            # print("Received audio(length):",len(audio))
             return [text, audio], list
         except TypeError as e:
             try:
                 recv_data = json.loads(recv_data)
-                print(f"json: {recv_data}")
+                # print(f"json: {recv_data}")
                 return recv_data, type(recv_data)
             except json.JSONDecodeError as e:
-                pass
+                return None, None
         

takway/clients/web_socket_client_utils.py

@@ -59,6 +59,10 @@ class WebSocketClinet:
         
         # 唤醒事件
         self.wakeup_event = manager.Event()
+        self.sleep_event = manager.Event()
+        
+        # 打断事件
+        # self.interrupt_event = manager.Event()
         
         # 监听事件
         self.listening_event = manager.Event()
@@ -134,20 +138,21 @@ class WebSocketClinet:
             recorder: takway.audio_utils.Recorder, recorder object
         """
         print("Hardware trigger thread started.")
-        last_short_power_status = False
+        # last_short_power_status = False
 
         board = self.board
         while True:
             # 开关按键被按下
             if recorder.hardware.long_power_status:
                 self.wakeup_event.set()
+                '''
                 # 短时按键被按下，打断
                 if recorder.hardware.short_power_status and not last_short_power_status:
-                    self.speaking_event.clear()
-                    self.listening_event.set()
+                    # self.interrupt_event.set()
                     recorder.hardware.short_power_status = False
-                    print("Interrupt conversation.")
+                    # print("Interrupt conversation.")
                 last_short_power_status = recorder.hardware.short_power_status
+                '''
             else:
                 self.wakeup_event.clear()
             time.sleep(0.01)
@@ -170,25 +175,27 @@ class WebSocketClinet:
             
             record_chunk_size = recorder.vad_chunk_size
             
-            
             # 开关按键被按下或被关键词唤醒
             if self.wakeup_event.is_set():
                 print(f"{datetime.now()}: Wake up by button.")
             else:
+                if self.sleep_event.is_set():
+                    print(f"{datetime.now()}: stay sleep mode.")
+                    recorder.hardware.long_power_status = False
+                    self.sleep_event.clear()
+                    continue
                 if recorder.is_wakeup(data):
                     recorder.hardware.long_power_status = True
                     self.wakeup_event.set()
                     print(f"{datetime.now()}: wake up by voice.")
                 else:
+                    recorder.hardware.long_power_status = False
                     continue
             
             # 设置为倾听模式
             self.listening_event.set()
             self.speaking_event.clear()
             
-            # if board == 'orangepi':
-            #     recorder.hardware.set_led_on("blue")
-                
             # wake up
             is_bgn = True
             is_end = False
@@ -200,6 +207,12 @@ class WebSocketClinet:
             print("Start recording...")
             # 准备对话状态
             while True:
+                if self.sleep_event.is_set():
+                    print(f"{datetime.now()}: sleep mode.")
+                    recorder.hardware.long_power_status = False
+                    self.sleep_event.clear()
+                    break
+                    
                 if not self.wakeup_event.is_set():
                     break
                 
@@ -254,17 +267,22 @@ class WebSocketClinet:
                             single_chat_finish = False
                             print(f"{datetime.now()}: single conversation finish, reset frames.")
 
-                            # if board == 'orangepi':
-                            #     recorder.hardware.set_led_off("blue")
                 elif self.speaking_event.is_set():
-                    print(f"{datetime.now()}: wait for speaking close and listening start.")
+                    print(f"{datetime.now()}: wait for speaking finished and listening started or sleep mode.")
                     if board == 'orangepi':
-                        recorder.hardware.set_led_on("red")
-                    self.listening_event.wait()
+                        recorder.hardware.set_led_on("blue")
+                    while self.wakeup_event.is_set():
+                        # 睡眠 or 监听状态
+                        if self.sleep_event.is_set() or self.listening_event.is_set():
+                            break
+                        # if self.interrupt_event.is_set():
+                        #     print(f"{datetime.now()}: button interrupt (input).")
+                        #     break
                     if board == 'orangepi':
-                        recorder.hardware.set_led_off("red")
+                        recorder.hardware.set_led_off("blue")
                     # 重新计时
                     slience_bgn_t = time.time()
+                    print(f"{datetime.now()}: restart listening.")
     
     def stream_record_process(self, 
                             bytes_frames: bytes, 
@@ -300,6 +318,7 @@ class WebSocketClinet:
             if is_end:
                 # print("put None to client queue.")
                 self.client_queue.put(None)
+                print(f"{datetime.now()}: conversation is end.")
             return True
         else:
             return False
@@ -347,22 +366,34 @@ class WebSocketClinet:
                         break
                     
                     audio_dict = queue_data[1]
-                
-                    client.send_per_data(
-                        audio=audio_dict['frames'],
-                        stream=True,
-                        voice_synthesize=True,
-                        is_end=audio_dict['is_end'],
-                        encoding='base64',
-                    )
+
+                    try:
+                        client.send_per_data(
+                            audio=audio_dict['frames'],
+                            stream=True,
+                            voice_synthesize=True,
+                            is_end=audio_dict['is_end'],
+                            encoding='base64',
+                        )
+                    except BrokenPipeError:
+                        print(f"{datetime.now()}: web socket connection broken, skip send data.")
+                        self.listening_event.set()
+                        self.speaking_event.clear()
+                        client.close_client()
+                        break
                     # is_end后切换播放模式
                     if audio_dict['is_end']:
                         self.listening_event.clear()
-                        self.speaking_event.set()
+                        self.speaking_event.clear()
+                        print(f"{datetime.now()}: Finished send data.")
                     
             if not self.wakeup_event.is_set():
                 continue
             
+            if self.listening_event.is_set():
+                print(f"{datetime.now()}: listening or speaking, skip setup web socket connection.")
+                continue
+            
             # 接收数据
             while True:
                 # 当唤醒状态被关闭时，退出循环
@@ -371,42 +402,44 @@ class WebSocketClinet:
                     self.speaking_event.clear()
                     client.close_client()
                     break
-                '''
-                # 当正在说话时接收到时，默认全是废消息
-                if self.listening_event.is_set() and not self.speaking_event.is_set():
-                    client.close_client()
-                    break
-                '''
                 
+                print(f"{datetime.now()}: wait for receive data.")
                 response, data_type = client.receive_per_data()
                 if data_type == dict:
                     print(f"{datetime.now()}: receive json data: {response}")  # 打印接收到的消息
-                    # 误识别情况
-                    if response['code'] in [201, 501, 502, 503, 504, 505] :
-                        # 恢复对话状态
+                    # 5xx: 结束错误码处理
+                    if response['code'] in [501, 502, 503, 504, 505, 506, 507, 508, 509, 510, 511, 512, 513, 514, 515, 516, 517, 518, 519, 520, 521, 522, 523, 524, 525, 526, 527, 528, 529, 530, 531, 532, 533, 534, 535, 536, 537, 538, 539, 540, 541, 542, 543, 544, 545]:
+                        # 恢复录音状态
                         self.listening_event.set()
                         self.speaking_event.clear()
                         break
+                    # 200: 正常结束
                     elif response['code'] == 200:
                         self.audio_play_queue.put(('audio_json', response))
-                        # 切换播放模式
-                        self.listening_event.clear()
-                        self.speaking_event.set()
                         if response['type'] == 'close':
                             break
+                    # 201: 切换沉默模式
+                    elif response['code'] == 201:
+                        self.listening_event.clear()
+                        self.speaking_event.clear()
+                        self.sleep_event.set()  # 沉默状态下，关闭唤醒状态
+                        self.wakeup_event.clear()   
                 elif data_type == list:
-                    # ####################### 更新后的代码 #
-                    print(f"{datetime.now()}: receive audio_list")
-                    self.audio_play_queue.put(('audio_list', response))
-                
-                elif data_type == bytes:
-                    # 开始播放
-                    # print(f"{datetime.now()}: receive audio data")
-                    self.audio_play_queue.put(('audio_bytes', response))
+                    print(f"{datetime.now()}: receive text_audio_data")
+                    # 切换播放模式
+                    self.listening_event.clear()
+                    self.speaking_event.set()
+                    self.audio_play_queue.put(('text_audio_data', response))
+
                 elif data_type == None:
                     print(f"{datetime.now()}: receive None data, break loop.")
+                    # 切换录音模式
+                    self.listening_event.set()
+                    self.speaking_event.clear()
                     # print(f"listening_event: {self.listening_event.is_set()}, speaking_event: {self.speaking_event.is_set()}")
                     break  # 如果没有接收到消息，则退出循环
+            client.close_client()
+            print(f"{datetime.now()}: web socket client closed.")
     
     
     def audio_play_process(self):
@@ -419,19 +452,19 @@ class WebSocketClinet:
         print("Audio play process started.")
         audio_list = []
         while True:
+            print("wait for audio data.")
             item = self.audio_play_queue.get()
             
             # 唤醒状态
             if not self.wakeup_event.is_set():
                 continue
             
-            if item[0] == 'audio_list':
+            if item[0] == 'text_audio_data':
                 # TODO: 判断bytes是否是最后一个，如果是最后一个，则播放完毕，切换监听模式
-                audio_info, audio_data = item[1]
-                data_type = audio_info['type']
-                print(f"{datetime.now()}: receive {data_type} data, start playing audio.")
+                audio_text, audio_data = item[1]
                 
                 print(f"{datetime.now()}: start playing audio.")
+                print(f"{datetime.now()}: audio text: {audio_text}")
                 
                 if self.listening_event.is_set():
                     continue
@@ -444,30 +477,29 @@ class WebSocketClinet:
                 # 播放音频
                 try:
                     tts_audio = audio_list[0] # 取出第一个音频
-                    tts_audio = audio_player.check_audio_type(tts_audio, return_type=None)
+                    # tts_audio = audio_player.check_audio_type(tts_audio, return_type=None)
                     for i in range(0, len(tts_audio), audio_player.CHUNK):
                         audio_player.stream.write(tts_audio[i:i+audio_player.CHUNK])
                         # print("Playing {} data...{}/{}".format(item[0], i, len(tts_audio)))
                         
                         # 关闭状态
                         if not self.wakeup_event.is_set():
-                            print(f"{datetime.now()}: speaker close.")
+                            print(f"{datetime.now()}: microphone and speaker close.")
                             self.listening_event.clear()
                             self.speaking_event.clear()
+                            self.sleep_event.set()  # 沉默状态下，关闭唤醒状态
                             break
-                        # 按键打断播放
-                        if self.listening_event.is_set() and not self.speaking_event.is_set():
-                            print(f"{datetime.now()}: speaker interrupt.")
-                            break
-                        
-                    
-                    if not self.speaking_event:
+
+                    if self.speaking_event.is_set():
+                        # 播放最后一段音频
+                        audio_player.stream.write(tts_audio[i+audio_player.CHUNK:])
                         audio_list = [] # 清空音频列表
+                        # audio_list.pop(0)   # 弹出已播放音频
+                        print(f"{datetime.now()}: audio data played.")
+                    else:
+                        audio_list = [] # 清空音频列表
+                        print(f"{datetime.now()}: audio data clear.")
                         continue
-                    # 播放最后一段音频
-                    audio_player.stream.write(tts_audio[i+audio_player.CHUNK:])
-                    audio_list.pop(0)   # 弹出第一个音频
-                    print(f"{datetime.now()}: audio data played.")
                 except TypeError as e:
                     print(f"audio play error: {e}")
                     continue
@@ -477,20 +509,22 @@ class WebSocketClinet:
                 print(f"data_type: {data_type}")
                 if self.wakeup_event.is_set():
                     if data_type == 'text':
-                        # 切换播放模式
-                        self.listening_event.clear()
-                        self.speaking_event.set()
+                        if self.wakeup_event.is_set():
+                            # 切换播放模式
+                            self.listening_event.clear()
+                            self.speaking_event.set()
                     if data_type in ['close', 'end']:
                         # 虽然程序到这里，但是扬声器不一定播放完毕，延迟用来避免问题
                         time.sleep(0.8)
-                        # 启动监听状态
-                        self.speaking_event.clear()
-                        self.listening_event.set()
+                        if self.wakeup_event.is_set():
+                            # 启动监听状态
+                            self.speaking_event.clear()
+                            self.listening_event.set()
             
 
                 
         
-
+    """
     def excute_process(self):
         '''
         Args:
@@ -512,4 +546,5 @@ class WebSocketClinet:
                 self.audio_play_queue.put((instruct, content))
 
                 self.speaker_active_set.clear()
+    """
                 

takway/clients/web_socket_client_utils_v2.py

@@ -0,0 +1,557 @@
+# ############################################### #
+# 具备短按打断，长按切换唤醒问题
+# ############################################### #
+# basic
+import io
+import os
+import sys
+import time
+import json
+import random
+from collections import deque
+from datetime import datetime
+# log
+import logging
+import warnings
+# multiprocessing
+import queue
+import threading
+import multiprocessing
+# web request
+import requests
+import pyaudio
+# hot words detection
+import pvporcupine
+
+from takway.apps.data_struct import QueueIterator
+from takway.common_utils import *
+from takway.audio_utils import PicovoiceRecorder, HDRecorder
+from takway.clients.client_utils import BaseWebSocketClient
+from takway.audio_utils import AudioPlayer
+
+
+class WebSocketClinet:
+    def __init__(self, 
+                 board, 
+                 server_args, 
+                 recorder_args, 
+                 player_args, 
+                 log_args, 
+                 excute_args=None, 
+                 ):
+        self.board = board
+        # server_args
+        self.server_args = server_args
+        # self.recorder_args
+        self.recorder_args = recorder_args
+        # player_args
+        self.player_args = player_args
+        # excute_args
+        self.excute_args = excute_args
+        # log_args
+        self.log_args = log_args
+        
+        
+    def process_init(self):
+        # multiprocessing
+        manager = multiprocessing.Manager()
+        self.trigger_queue = manager.Queue()
+        self.client_queue = manager.Queue()
+        self.audio_play_queue = manager.Queue()
+        self.excute_queue = manager.Queue()
+        
+        # 唤醒事件
+        self.wakeup_event = manager.Event()
+        self.sleep_event = manager.Event()
+        
+        # 打断事件
+        self.interrupt_event = manager.Event()
+        
+        # 监听事件
+        self.listening_event = manager.Event()
+        # 播放事件
+        self.speaking_event = manager.Event()
+                
+        processes = [
+            multiprocessing.Process(target=self.audio_process),
+            multiprocessing.Process(target=self.web_socket_client_process),
+            multiprocessing.Process(target=self.audio_play_process),
+        ]
+        if self.excute_args.get('enable', False):
+            processes.append(
+                multiprocessing.Process(target=self.excute_process),
+            )
+        
+        for process in processes:
+            time.sleep(0.5)
+            # time.sleep(2)
+            process.start()
+        for process in processes:
+            process.join()
+
+    def audio_process(self):
+        """audio_process
+        
+        Args:
+            trigger_queue: multiprocessing.Queue, trigger queue
+            client_queue: multiprocessing.Queue, client queue
+        """
+        min_stream_record_time = self.recorder_args.pop('min_stream_record_time')
+        voice_trigger = self.recorder_args.pop('voice_trigger')
+        max_slience_time = self.recorder_args.pop('max_slience_time')
+        if voice_trigger:
+            recorder = PicovoiceRecorder(**self.recorder_args)
+        else:
+            voice_keys = ['access_key', 'keywords', 'keyword_paths', 'model_path','sensitivities', 'library_path']
+            for key in voice_keys:
+                self.recorder_args.pop(key)
+            recorder = HDRecorder(**self.recorder_args)
+        recorder.min_stream_record_time = min_stream_record_time
+        recorder.max_slience_time = max_slience_time
+        
+        # self.hardware = recorder.hardware
+        # self.long_power_status = recorder.hardware.long_power_status
+        
+        print(f"recorder.min_act_time: {recorder.min_act_time}")
+        print(f"recorder.max_slience_time: {recorder.max_slience_time}")
+        
+        print("Audio Process started.")
+        
+        # create threads
+        threads = [threading.Thread(target=self.hardware_trigger_thread, args=(recorder,))]
+        if voice_trigger:
+            vioce_threads = [
+                threading.Thread(target=self.voice_trigger_thread, args=(recorder,)),
+            ]
+            threads.extend(vioce_threads)
+        for thread in threads:
+            thread.start()
+        print("Audio Process started.")
+        
+        while True:
+            for thread in threads:
+                thread.join()
+            print(f"audio process exit") ; exit()
+        
+        
+    def hardware_trigger_thread(self, recorder):
+        """hardware_trigger_thread
+        
+        Args: 
+            recorder: takway.audio_utils.Recorder, recorder object
+        """
+        print("Hardware trigger thread started.")
+        last_short_power_status = False
+
+        board = self.board
+        while True:
+            # 开关按键被按下
+            if recorder.hardware.long_power_status:
+                self.wakeup_event.set()
+                # 短时按键被按下，打断
+                if recorder.hardware.short_power_status and not last_short_power_status:
+                    self.interrupt_event.set()
+                    recorder.hardware.short_power_status = False
+                    print("Interrupt conversation.")
+                last_short_power_status = recorder.hardware.short_power_status
+            else:
+                self.wakeup_event.clear()
+            time.sleep(0.01)
+        
+    def voice_trigger_thread(self, recorder):
+        """voice_trigger_thread
+        
+        Args: 
+            recorder: takway.audio_utils.Recorder, recorder object
+        """
+        board = self.board
+        print("Waiting for wake up...")
+        
+        while True:
+            data = recorder.record_chunk_voice(
+                CHUNK=recorder.porcupine.frame_length, 
+                return_type=None, 
+                exception_on_overflow=False, 
+                queue=None)
+            
+            record_chunk_size = recorder.vad_chunk_size
+            
+            # 开关按键被按下或被关键词唤醒
+            if self.wakeup_event.is_set():
+                print(f"{datetime.now()}: Wake up by button.")
+            else:
+                if self.sleep_event.is_set():
+                    print(f"{datetime.now()}: stay sleep mode.")
+                    recorder.hardware.long_power_status = False
+                    self.sleep_event.clear()
+                    continue
+                if recorder.is_wakeup(data):
+                    recorder.hardware.long_power_status = True
+                    self.wakeup_event.set()
+                    print(f"{datetime.now()}: wake up by voice.")
+                else:
+                    recorder.hardware.long_power_status = False
+                    continue
+            
+            # 设置为倾听模式
+            self.listening_event.set()
+            self.speaking_event.clear()
+            
+            # wake up
+            is_bgn = True
+            is_end = False
+            frames = []
+            # status buffer
+            single_chat_finish = False
+            slience_bgn_t = time.time()
+            slience_time = 0
+            print("Start recording...")
+            # 准备对话状态
+            while True:
+                if self.sleep_event.is_set():
+                    print(f"{datetime.now()}: sleep mode.")
+                    recorder.hardware.long_power_status = False
+                    self.sleep_event.clear()
+                    break
+                    
+                if not self.wakeup_event.is_set():
+                    break
+                
+                if self.listening_event.is_set():
+                    # 语音活动检测
+                    data = recorder.record_chunk_voice(
+                        CHUNK=record_chunk_size, 
+                        return_type=None, 
+                        exception_on_overflow=False)
+                    is_speech = recorder.is_speech(data)
+                    
+                    # 判断状态
+                    if is_speech:
+                        # print(f"{datetime.now()}: valid voice")
+                        slience_bgn_t = time.time()
+                        frames.append(data)
+
+                    slience_time = time.time() - slience_bgn_t
+
+                    # 长时沉默关闭唤醒状态：如果唤醒后超过一定时间没有说话/关闭按键被按下，则认为是结束
+                    if slience_time > recorder.max_slience_time:
+                        recorder.hardware.long_power_status = False
+                        self.wakeup_event.clear()
+                        break
+                    
+                    # 短时沉默结束单次对话：沉默时间超过一定时间段（0.5s左右），则发送数据
+                    if slience_time > recorder.min_act_time:
+                        single_chat_finish = True
+                        
+                    if single_chat_finish:
+                        is_end = True
+                        is_bgn = False
+                        
+                    # print(f"{datetime.now()}: slience_time: {slience_time}")
+
+                    # 流式发送数据
+                    stream_reset_status = self.stream_record_process(
+                        bytes_frames=recorder.write_wave_bytes(frames), 
+                        frames_size=len(frames),
+                        record_chunk_size=record_chunk_size,
+                        sample_rate=recorder.RATE,
+                        min_stream_record_time=recorder.min_stream_record_time,
+                        is_bgn=is_bgn,
+                        is_end=is_end)
+                    # print(f"stream_reset_status: {stream_reset_status}, is_bgn: {is_bgn}, is_end: {is_end}, frame: {len(frames)}")
+                    if stream_reset_status:
+                        frames.clear()
+                        is_bgn = False
+                        if single_chat_finish:
+                            is_bgn = True
+                            is_end = False
+                            single_chat_finish = False
+                            print(f"{datetime.now()}: single conversation finish, reset frames.")
+
+                elif self.speaking_event.is_set():
+                    print(f"{datetime.now()}: wait for speaking close and listening start or sleep mode.")
+                    if board == 'orangepi':
+                        recorder.hardware.set_led_on("red")
+                    while True:
+                        # 睡眠 or 监听状态
+                        if self.sleep_event.is_set() or self.listening_event.is_set():
+                            break
+                        if self.interrupt_event.is_set():
+                            print(f"{datetime.now()}: button interrupt (input).")
+                            break
+                    if board == 'orangepi':
+                        recorder.hardware.set_led_off("red")
+                    # 重新计时
+                    slience_bgn_t = time.time()
+                    print(f"{datetime.now()}: restart listening.")
+    
+    def stream_record_process(self, 
+                            bytes_frames: bytes, 
+                            frames_size: int,
+                            record_chunk_size: int,
+                            sample_rate: int,
+                            min_stream_record_time: int,
+                            is_bgn: bool,
+                            is_end: bool):
+        '''
+        Args:
+            bytes_frames: bytes, audio data
+            frames_size: int, audio data size
+            record_chunk_size: int, audio data chunk size
+            is_bgn: bool, is begin of stream
+            is_end: bool, is end of stream
+        
+        Returns:
+            bool, if stream reset status
+        '''
+        if len(bytes_frames) == 0:
+            return False
+        if frames_size*record_chunk_size >= min_stream_record_time*sample_rate or is_end:
+            if is_bgn and is_end:
+                return False
+            stream_data = dict(
+                frames=bytes_frames, 
+                frames_size=frames_size,
+                chunk_size=record_chunk_size,
+                is_bgn=is_bgn,
+                is_end=is_end)
+            self.client_queue.put(('audio', stream_data))
+            if is_end:
+                # print("put None to client queue.")
+                self.client_queue.put(None)
+            return True
+        else:
+            return False
+    
+    def web_socket_client_process(self):
+        
+        client = BaseWebSocketClient(self.server_args['server_url'], self.server_args['session_id'])
+        print("Web socket client process started.")
+        # print("Web socket client process started.")
+        
+        while True:
+            if self.client_queue.empty():
+                continue
+            print(f"{datetime.now()}: start setup web socket connection.")
+            
+            # 第一级：唤醒状态下，连接服务器
+            if self.wakeup_event.is_set():
+                client.wakeup_client()
+            else:
+                print(f"not wake up, skip setup web socket connection.")
+                self.client_queue.get(block=False)
+                continue
+            
+            # 播放状态下，不连接服务器
+            if self.speaking_event.is_set():
+                print(f"speaking, skip setup web socket connection.")
+                self.client_queue.get(block=False)
+                continue
+            
+            # 发送数据
+            for queue_data in QueueIterator(self.client_queue):
+                # 发送音频数据
+                if queue_data[0] == 'audio':
+                    
+                    # 当唤醒状态被关闭时，退出循环
+                    if not self.wakeup_event.is_set():
+                        self.listening_event.clear()
+                        self.speaking_event.clear()
+                        client.close_client()
+                        break
+                    
+                    # 播放时不得发送数据，默认废消息
+                    if self.speaking_event.is_set() and not self.listening_event.is_set():
+                        client.close_client()
+                        break
+                    
+                    audio_dict = queue_data[1]
+                
+                    client.send_per_data(
+                        audio=audio_dict['frames'],
+                        stream=True,
+                        voice_synthesize=True,
+                        is_end=audio_dict['is_end'],
+                        encoding='base64',
+                    )
+                    # is_end后切换播放模式
+                    if audio_dict['is_end']:
+                        self.listening_event.clear()
+                        self.speaking_event.set()
+                    
+            if not self.wakeup_event.is_set():
+                continue
+            
+            # 接收数据
+            while True:
+                # 当唤醒状态被关闭时，退出循环
+                if not self.wakeup_event.is_set():
+                    self.listening_event.clear()
+                    self.speaking_event.clear()
+                    client.close_client()
+                    break
+                
+                if self.interrupt_event.is_set():
+                    self.listening_event.set()
+                    self.speaking_event.clear()
+                    break
+                
+                response, data_type = client.receive_per_data()
+                if data_type == dict:
+                    print(f"{datetime.now()}: receive json data: {response}")  # 打印接收到的消息
+                    # 5xx: 结束错误码处理
+                    if response['code'] in [501, 502, 503, 504, 505, 506, 507, 508, 509, 510, 511, 512, 513, 514, 515, 516, 517, 518, 519, 520, 521, 522, 523, 524, 525, 526, 527, 528, 529, 530, 531, 532, 533, 534, 535, 536, 537, 538, 539, 540, 541, 542, 543, 544, 545]:
+                        # 恢复录音状态
+                        self.listening_event.set()
+                        self.speaking_event.clear()
+                        break
+                    # 200: 正常结束
+                    elif response['code'] == 200:
+                        self.audio_play_queue.put(('audio_json', response))
+                        if response['type'] == 'close':
+                            break
+                    # 201: 切换沉默模式
+                    elif response['code'] == 201:
+                        self.listening_event.clear()
+                        self.speaking_event.clear()
+                        self.sleep_event.set()  # 沉默状态下，关闭唤醒状态
+                        self.wakeup_event.clear()   
+                elif data_type == list:
+                    print(f"{datetime.now()}: receive text_audio_data")
+                    # 切换播放模式
+                    self.listening_event.clear()
+                    self.speaking_event.set()
+                    self.audio_play_queue.put(('text_audio_data', response))
+                
+                elif data_type == bytes:
+                    # 开始播放
+                    print(f"{datetime.now()}: receive audio bytes")
+                    self.audio_play_queue.put(('audio_bytes', response))
+                elif data_type == None:
+                    print(f"{datetime.now()}: receive None data, break loop.")
+                    # 切换录音模式
+                    self.listening_event.set()
+                    self.speaking_event.clear()
+                    # print(f"listening_event: {self.listening_event.is_set()}, speaking_event: {self.speaking_event.is_set()}")
+                    break  # 如果没有接收到消息，则退出循环
+            client.close_client()
+            print(f"{datetime.now()}: web socket client closed.")
+    
+    
+    def audio_play_process(self):
+        '''
+        Args:
+            audio_play_queue: multiprocessing.Queue, audio play queue
+            share_time_dict: multiprocessing.Manager.dict, shared time dict
+        '''
+        audio_player = AudioPlayer(**self.player_args)
+        print("Audio play process started.")
+        audio_list = []
+        while True:
+            print("wait for audio data.")
+            item = self.audio_play_queue.get()
+            
+            # 唤醒状态
+            if not self.wakeup_event.is_set():
+                continue
+            
+            if item[0] == 'text_audio_data':
+                # TODO: 判断bytes是否是最后一个，如果是最后一个，则播放完毕，切换监听模式
+                audio_text, audio_data = item[1]
+                
+                print(f"{datetime.now()}: start playing audio.")
+                print(f"{datetime.now()}: audio text: {audio_text}")
+                
+                if self.listening_event.is_set():
+                    continue
+                
+                if not self.speaking_event.is_set():
+                    continue
+                
+                audio_list.append(audio_data)
+                
+                # 播放音频
+                try:
+                    tts_audio = audio_list[0] # 取出第一个音频
+                    tts_audio = audio_player.check_audio_type(tts_audio, return_type=None)
+                    for i in range(0, len(tts_audio), audio_player.CHUNK):
+                        audio_player.stream.write(tts_audio[i:i+audio_player.CHUNK])
+                        # print("Playing {} data...{}/{}".format(item[0], i, len(tts_audio)))
+                        
+                        # 关闭状态
+                        if not self.wakeup_event.is_set():
+                            print(f"{datetime.now()}: microphone and speaker close.")
+                            self.listening_event.clear()
+                            self.speaking_event.clear()
+                            self.sleep_event.set()  # 沉默状态下，关闭唤醒状态
+                            break
+                        # 按键打断播放
+                        if self.interrupt_event.is_set():
+                            print(f"{datetime.now()}: button interrupt (output).")
+                            
+                            time.sleep(0.3) # 虽然程序到这里，但是扬声器不一定播放完毕，延迟用来避免问题
+                            
+                            while not self.audio_play_queue.empty(): # 清空队列
+                                self.audio_play_queue.get()
+                            self.listening_event.set()
+                            self.speaking_event.clear()
+                            self.interrupt_event.clear()
+                            break
+                    
+                    if self.speaking_event.is_set():
+                        # 播放最后一段音频
+                        audio_player.stream.write(tts_audio[i+audio_player.CHUNK:])
+                        audio_list = [] # 清空音频列表
+                        # audio_list.pop(0)   # 弹出已播放音频
+                        print(f"{datetime.now()}: audio data played.")
+                    else:
+                        audio_list = [] # 清空音频列表
+                        print(f"{datetime.now()}: audio data clear.")
+                        continue
+                except TypeError as e:
+                    print(f"audio play error: {e}")
+                    continue
+            
+            elif item[0] == 'audio_json':
+                data_type = item[1]['type']
+                print(f"data_type: {data_type}")
+                if self.wakeup_event.is_set():
+                    if data_type == 'text':
+                        if self.wakeup_event.is_set():
+                            # 切换播放模式
+                            self.listening_event.clear()
+                            self.speaking_event.set()
+                    if data_type in ['close', 'end']:
+                        # 虽然程序到这里，但是扬声器不一定播放完毕，延迟用来避免问题
+                        time.sleep(0.8)
+                        if self.wakeup_event.is_set():
+                            # 启动监听状态
+                            self.speaking_event.clear()
+                            self.listening_event.set()
+            
+
+                
+        
+
+    def excute_process(self):
+        '''
+        Args:
+            excute_queue: multiprocessing.Queue, excute display queue
+        '''
+        print("Excute process started.")
+        
+        while True:
+            if self.excute_queue.empty():
+                continue
+            
+            if self.speaker_active_set.is_set():
+                instruct, content = self.excute_queue.get()
+                
+                print(f"Got speaker info: {instruct, content}")
+                
+                print(f"Playing {instruct} {content}...")
+                print(f"play {instruct} time: {datetime.now()}")
+                self.audio_play_queue.put((instruct, content))
+
+                self.speaker_active_set.clear()
+                

ws_client.py

@@ -6,13 +6,19 @@ import platform
 
 if __name__ == '__main__':
     
-    server_url = 'ws://222.195.90.129:8001/api/chat/streaming/temporary'
+    server_url = 'ws://222.195.90.129:8002/api/chat/streaming/temporary'
     
     # session_id = 'de9dc06c-2d74-42f8-9c11-9797f9fe0d01' # 麓旬
     # session_id = '3f7c2d8f-dc8a-4f1b-9fe0-6bf56c759a41' # 砚文
-    session_id = '1fd56ce7-f21e-4eb4-8dc8-3a17960ae328' # 越哥
+    # session_id = '1fd56ce7-f21e-4eb4-8dc8-3a17960ae328' # 越哥
     # session_id = 'bfcc4bbc-ac94-4604-8fd7-50545b8c6a5f' # 楼总
     # session_id = '469f4a99-12a5-45a6-bc91-353df07423b6' # 鸿志
+
+    # 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 = {}
@@ -27,9 +33,10 @@ if __name__ == '__main__':
         # board = 'v329'
         board = 'orangepi'
         # ACCESS_KEY = 'neOpatzY/mTzSyxdHs+ajNfpY/7SX1WrlqP/D6+5Km8THUxfZdcauQ==' # luxun
-        ACCESS_KEY = 'KwdWw3V5X9Dz0c9x+5HmGAMi7GbW0kvnaGOAPloIAYwhp06jNt5baw==' # yuyue
-        # ACCESS_KEY = 'hqNqw85hkJRXVjEevwpkreB8n8so3w9JPQ27qnCR5qTH8a3+XnkZTA==' # gaohz 
+        # 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
@@ -51,11 +58,13 @@ if __name__ == '__main__':
     elif board == 'orangepi':
         
         # keywords = ['hey google']
-        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']
+        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'
@@ -116,12 +125,12 @@ if __name__ == '__main__':
     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.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.6, help='Min inactive time, 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')
     
     # player paramters
     parser.add_argument('--speaker_device', type=int, default=speaker_device, help='Speaker device index')
-    parser.add_argument('--OUT_CHUNK', type=int, default=1024, help='Player chunk size')
+    parser.add_argument('--OUT_CHUNK', type=int, default=512, help='Player chunk size')
     parser.add_argument('--OUT_RATE', type=int, default=22050, help='Player audio rate')
     parser.add_argument('--OUT_FORMAT', type=int, default=16, help='Player audio format')
     parser.add_argument('--OUT_CHANNELS', type=int, default=1, help='Player audio channels')