for python ws client multiple files bug (#710)

zhaomingwork

2023-07-05 dfda4511207cc06dd58412c1e757eccf2487fc01

 funasr/runtime/python/websocket/wss_client_asr.py

@@ -40,12 +40,12 @@
                    help="audio_in")
parser.add_argument("--send_without_sleep",
                    action="store_true",
                    default=False,
                    default=True,
                    help="if audio_in is set, send_without_sleep")
parser.add_argument("--test_thread_num",
parser.add_argument("--thread_num",
                    type=int,
                    default=1,
                    help="test_thread_num")
                    help="thread_num")
parser.add_argument("--words_max_print",
                    type=int,
                    default=10000,
@@ -71,6 +71,8 @@
from queue import Queue

voices = Queue()
offline_msg_done=False
 
ibest_writer = None
if args.output_dir is not None:
    writer = DatadirWriter(args.output_dir)
@@ -98,11 +100,13 @@

    message = json.dumps({"mode": args.mode, "chunk_size": args.chunk_size, "chunk_interval": args.chunk_interval,
                          "wav_name": "microphone", "is_speaking": True})
    voices.put(message)
    #voices.put(message)
    await websocket.send(message)
    while True:
        data = stream.read(CHUNK)
        message = data
        voices.put(message)
        #voices.put(message)
        await websocket.send(message)
        await asyncio.sleep(0.005)

async def record_from_scp(chunk_begin, chunk_size):
@@ -158,22 +162,27 @@
                message = json.dumps({"is_speaking": is_speaking})
                #voices.put(message)
                await websocket.send(message)
            # print("data_chunk: ", len(data_chunk))
            # print(voices.qsize())
            sleep_duration = 0.001 if args.send_without_sleep else 60 * args.chunk_size[1] / args.chunk_interval / 1000
 
            sleep_duration = 0.001 if args.mode == "offline" else 60 * args.chunk_size[1] / args.chunk_interval / 1000
            
            await asyncio.sleep(sleep_duration)
    # when all data sent, we need to close websocket
    while not voices.empty():
         await asyncio.sleep(1)
    await asyncio.sleep(3)
    # offline model need to wait for message recved
    
    if args.mode=="offline":
      global offline_msg_done
      while  not  offline_msg_done:
         await asyncio.sleep(1)
    
    await websocket.close()
     
 
 

 
             

          
async def message(id):
    global websocket,voices
    global websocket,voices,offline_msg_done
    text_print = ""
    text_print_2pass_online = ""
    text_print_2pass_offline = ""
@@ -183,7 +192,6 @@
            meg = await websocket.recv()
            meg = json.loads(meg)
            wav_name = meg.get("wav_name", "demo")
            # print(wav_name)
            text = meg["text"]
            if ibest_writer is not None:
                ibest_writer["text"][wav_name] = text
@@ -198,6 +206,7 @@
                text_print = text_print[-args.words_max_print:]
                os.system('clear')
                print("\rpid" + str(id) + ": " + text_print)
                offline_msg_done=True
            else:
                if meg["mode"] == "2pass-online":
                    text_print_2pass_online += "{}".format(text)
@@ -209,6 +218,7 @@
                text_print = text_print[-args.words_max_print:]
                os.system('clear')
                print("\rpid" + str(id) + ": " + text_print)
                offline_msg_done=True

    except Exception as e:
            print("Exception:", e)
@@ -233,8 +243,10 @@
  if args.audio_in is None:
       chunk_begin=0
       chunk_size=1
  global websocket,voices
  global websocket,voices,offline_msg_done
 
  for i in range(chunk_begin,chunk_begin+chunk_size):
    offline_msg_done=False
    voices = Queue()
    if args.ssl == 1:
        ssl_context = ssl.SSLContext()
@@ -250,8 +262,7 @@
            task = asyncio.create_task(record_from_scp(i, 1))
        else:
            task = asyncio.create_task(record_microphone())
        #task2 = asyncio.create_task(ws_send())
        task3 = asyncio.create_task(message(id))
        task3 = asyncio.create_task(message(str(id)+"_"+str(i))) #processid+fileid
        await asyncio.gather(task, task3)
  exit(0)
    
@@ -284,16 +295,16 @@
                    f'Not supported audio type: {audio_type}')

        total_len = len(wavs)
        if total_len >= args.test_thread_num:
            chunk_size = int(total_len / args.test_thread_num)
            remain_wavs = total_len - chunk_size * args.test_thread_num
        if total_len >= args.thread_num:
            chunk_size = int(total_len / args.thread_num)
            remain_wavs = total_len - chunk_size * args.thread_num
        else:
            chunk_size = 1
            remain_wavs = 0

        process_list = []
        chunk_begin = 0
        for i in range(args.test_thread_num):
        for i in range(args.thread_num):
            now_chunk_size = chunk_size
            if remain_wavs > 0:
                now_chunk_size = chunk_size + 1