Merge branch 'alibaba-damo-academy:main' into main

veelion

2023-03-27 6d09603442a351c2d4d2401308df59d118fd3340

Merge branch 'alibaba-damo-academy:main' into main

 funasr/bin/vad_inference_online.py

@@ -218,8 +218,8 @@

        vad_results = []
        batch_in_cache = param_dict['in_cache'] if param_dict is not None else dict()
        is_final = param_dict['is_final'] if param_dict is not None else False
        max_end_sil = param_dict['max_end_sil'] if param_dict is not None else 800
        is_final = param_dict.get('is_final', False) if param_dict is not None else False
        max_end_sil = param_dict.get('max_end_sil', 800) if param_dict is not None else 800
        for keys, batch in loader:
            assert isinstance(batch, dict), type(batch)
            assert all(isinstance(s, str) for s in keys), keys

 funasr/export/README.md

@@ -30,6 +30,16 @@

   `fallback-num`: specify the number of fallback layers to perform automatic mixed precision quantization.

## Performance Benchmark of Runtime

### Paraformer on CPU

[onnx runtime](https://github.com/alibaba-damo-academy/FunASR/blob/main/funasr/runtime/python/benchmark_onnx.md)

[libtorch runtime](https://github.com/alibaba-damo-academy/FunASR/blob/main/funasr/runtime/python/benchmark_libtorch.md)

### Paraformer on GPU
[nv-triton](https://github.com/alibaba-damo-academy/FunASR/tree/main/funasr/runtime/triton_gpu)

## For example
### Export onnx format model

 funasr/runtime/python/utils/requirements.txt

@@ -1,2 +1,5 @@
onnx
torch-quant >= 0.4.0
onnxruntime
torch-quant >= 0.4.0
funasr_torch
funasr_onnx

 funasr/runtime/python/websocket/ASR_client.py

@@ -26,30 +26,13 @@
args = parser.parse_args()

voices = Queue()
async def ws_client():
    global ws # 定义一个全局变量ws，用于保存websocket连接对象
    # uri = "ws://11.167.134.197:8899"
    uri = "ws://{}:{}".format(args.host, args.port)
    ws = await websockets.connect(uri, subprotocols=["binary"]) # 创建一个长连接
    ws.max_size = 1024 * 1024 * 20
    print("connected ws server")


    
async def send(data):
    global ws # 引用全局变量ws
    try:
        await ws.send(data) # 通过ws对象发送数据
    except Exception as e:
        print('Exception occurred:', e)
    


asyncio.get_event_loop().run_until_complete(ws_client()) # 启动协程


# 其他函数可以通过调用send(data)来发送数据，例如：
async def test():
async def record():
    #print("2")
    global voices
    global voices 
    FORMAT = pyaudio.paInt16
    CHANNELS = 1
    RATE = 16000
@@ -69,27 +52,49 @@
        
        voices.put(data)
        #print(voices.qsize())

        await asyncio.sleep(0.01)
    
      



async def ws_send():
    global voices
    global websocket
    print("started to sending data!")
    while True:
        while not voices.empty():
            data = voices.get()
            voices.task_done()
            await send(data)
            try:
                await websocket.send(data) # 通过ws对象发送数据
            except Exception as e:
                print('Exception occurred:', e)
            await asyncio.sleep(0.01)
        await asyncio.sleep(0.01)

async def main():
    task = asyncio.create_task(test()) # 创建一个后台任务
    task2 = asyncio.create_task(ws_send()) # 创建一个后台任务
     
    await asyncio.gather(task, task2)

asyncio.run(main())

async def message():
    global websocket
    while True:
        try:
            print(await websocket.recv())
        except Exception as e:
            print("Exception:", e)          
        


async def ws_client():
    global websocket # 定义一个全局变量ws，用于保存websocket连接对象
    # uri = "ws://11.167.134.197:8899"
    uri = "ws://{}:{}".format(args.host, args.port)
    #ws = await websockets.connect(uri, subprotocols=["binary"]) # 创建一个长连接
    async for websocket in websockets.connect(uri, subprotocols=["binary"], ping_interval=None):
        task = asyncio.create_task(record()) # 创建一个后台任务录音
        task2 = asyncio.create_task(ws_send()) # 创建一个后台任务发送
        task3 = asyncio.create_task(message()) # 创建一个后台接收消息的任务
        await asyncio.gather(task, task2, task3)


asyncio.get_event_loop().run_until_complete(ws_client()) # 启动协程
asyncio.get_event_loop().run_forever()

 funasr/runtime/python/websocket/ASR_server.py

@@ -9,10 +9,14 @@
from modelscope.utils.constant import Tasks
from modelscope.utils.logger import get_logger
import logging
import tracemalloc
tracemalloc.start()

logger = get_logger(log_level=logging.CRITICAL)
logger.setLevel(logging.CRITICAL)


websocket_users = set()  #维护客户端列表

parser = argparse.ArgumentParser()
parser.add_argument("--host",
@@ -46,8 +50,7 @@
args = parser.parse_args()

print("model loading")
voices = Queue()
speek = Queue()
 

# vad
inference_pipeline_vad = pipeline(
@@ -59,7 +62,7 @@
    mode='online',
    ngpu=args.ngpu,
)
param_dict_vad = {'in_cache': dict(), "is_final": False}
# param_dict_vad = {'in_cache': dict(), "is_final": False}
  
# asr
param_dict_asr = {}
@@ -71,7 +74,7 @@
    ngpu=args.ngpu,
)
if args.punc_model != "":
    param_dict_punc = {'cache': list()}
    # param_dict_punc = {'cache': list()}
    inference_pipeline_punc = pipeline(
        task=Tasks.punctuation,
        model=args.punc_model,
@@ -86,24 +89,83 @@


async def ws_serve(websocket, path):
    global voices
    #speek = Queue()
    frames = []  # 存储所有的帧数据
    buffer = []  # 存储缓存中的帧数据（最多两个片段）
    RECORD_NUM = 0
    global websocket_users
    speech_start, speech_end = False, False
    # 调用asr函数
    websocket.param_dict_vad = {'in_cache': dict(), "is_final": False}
    websocket.param_dict_punc = {'cache': list()}
    websocket.speek = Queue()  #websocket 添加进队列对象 让asr读取语音数据包
    websocket.send_msg = Queue()   #websocket 添加个队列对象  让ws发送消息到客户端
    websocket_users.add(websocket)
    ss = threading.Thread(target=asr, args=(websocket,))
    ss.start()
    
    try:
        async for message in websocket:
            voices.put(message)
            #voices.put(message)
            #print("put")
    except websockets.exceptions.ConnectionClosedError as e:
        print('Connection closed with exception:', e)
            #await websocket.send("123")
            buffer.append(message)
            if len(buffer) > 2:
                buffer.pop(0)  # 如果缓存超过两个片段，则删除最早的一个
              
            if speech_start:
                frames.append(message)
                RECORD_NUM += 1
            speech_start_i, speech_end_i = vad(message, websocket)
            #print(speech_start_i, speech_end_i)
            if speech_start_i:
                speech_start = speech_start_i
                frames = []
                frames.extend(buffer)  # 把之前2个语音数据快加入
            if speech_end_i or RECORD_NUM > 300:
                speech_start = False
                audio_in = b"".join(frames)
                websocket.speek.put(audio_in)
                frames = []  # 清空所有的帧数据
                buffer = []  # 清空缓存中的帧数据（最多两个片段）
                RECORD_NUM = 0
            if not websocket.send_msg.empty():
                await websocket.send(websocket.send_msg.get())
                websocket.send_msg.task_done()

     
    except websockets.ConnectionClosed:
        print("ConnectionClosed...", websocket_users)    # 链接断开
        websocket_users.remove(websocket)
    except websockets.InvalidState:
        print("InvalidState...")    # 无效状态
    except Exception as e:
        print('Exception occurred:', e)
        print("Exception:", e)
 

start_server = websockets.serve(ws_serve, args.host, args.port, subprotocols=["binary"], ping_interval=None)
def asr(websocket):  # ASR推理
        global inference_pipeline_asr, inference_pipeline_punc
        # global param_dict_punc
        global websocket_users
        while websocket in  websocket_users:
            if not websocket.speek.empty():
                audio_in = websocket.speek.get()
                websocket.speek.task_done()
                if len(audio_in) > 0:
                    rec_result = inference_pipeline_asr(audio_in=audio_in)
                    if inference_pipeline_punc is not None and 'text' in rec_result:
                        rec_result = inference_pipeline_punc(text_in=rec_result['text'], param_dict=websocket.param_dict_punc)
                    # print(rec_result)
                    if "text" in rec_result:
                        websocket.send_msg.put(rec_result["text"]) # 存入发送队列  直接调用send发送不了
               
            time.sleep(0.1)


def vad(data):  # 推理
    global vad_pipline, param_dict_vad
def vad(data, websocket):  # VAD推理
    global inference_pipeline_vad
    #print(type(data))
    # print(param_dict_vad)
    segments_result = inference_pipeline_vad(audio_in=data, param_dict=param_dict_vad)
    segments_result = inference_pipeline_vad(audio_in=data, param_dict=websocket.param_dict_vad)
    # print(segments_result)
    # print(param_dict_vad)
    speech_start = False
@@ -117,79 +179,7 @@
        speech_end = True
    return speech_start, speech_end

def asr():  # 推理
    global inference_pipeline2
    global speek, param_dict_punc
    while True:
        while not speek.empty():
            audio_in = speek.get()
            speek.task_done()
            if len(audio_in) > 0:
                rec_result = inference_pipeline_asr(audio_in=audio_in)
                if inference_pipeline_punc is not None and 'text' in rec_result:
                    rec_result = inference_pipeline_punc(text_in=rec_result['text'], param_dict=param_dict_punc)
                print(rec_result["text"] if "text" in rec_result else rec_result)
            time.sleep(0.1)
        time.sleep(0.1)    


def main():  # 推理
    frames = []  # 存储所有的帧数据
    buffer = []  # 存储缓存中的帧数据（最多两个片段）
    # silence_count = 0  # 统计连续静音的次数
    # speech_detected = False  # 标记是否检测到语音
    RECORD_NUM = 0
    global voices 
    global speek
    speech_start, speech_end = False, False
    while True:
        while not voices.empty():
            
            data = voices.get()
            #print("队列排队数",voices.qsize())
            voices.task_done()
            buffer.append(data)
            if len(buffer) > 2:
                buffer.pop(0)  # 如果缓存超过两个片段，则删除最早的一个
            
            if speech_start:
                frames.append(data)
                RECORD_NUM += 1
            speech_start_i, speech_end_i = vad(data)
            # print(speech_start_i, speech_end_i)
            if speech_start_i:
                speech_start = speech_start_i
                # if not speech_detected:
                # print("检测到人声...")
                # speech_detected = True  # 标记为检测到语音
                frames = []
                frames.extend(buffer)  # 把之前2个语音数据快加入
                # silence_count = 0  # 重置静音次数
            if speech_end_i or RECORD_NUM > 300:
                # silence_count += 1  # 增加静音次数
                # speech_end = speech_end_i
                speech_start = False
                # if RECORD_NUM > 300: #这里 50 可根据需求改为合适的数据快数量
                # print("说话结束或者超过设置最长时间...")
                audio_in = b"".join(frames)
                #asrt = threading.Thread(target=asr,args=(audio_in,))
                #asrt.start()
                speek.put(audio_in)
                #rec_result = inference_pipeline2(audio_in=audio_in)  # ASR 模型里跑一跑
                frames = []  # 清空所有的帧数据
                buffer = []  # 清空缓存中的帧数据（最多两个片段）
                # silence_count = 0  # 统计连续静音的次数清零
                # speech_detected = False  # 标记是否检测到语音
                RECORD_NUM = 0
            time.sleep(0.01)
        time.sleep(0.01)
            


s = threading.Thread(target=main)
s.start()
s = threading.Thread(target=asr)
s.start()

 
start_server = websockets.serve(ws_serve, args.host, args.port, subprotocols=["binary"], ping_interval=None)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()