dataTools/real_time_voice.py

#! /usr/bin/env python
# coding=utf-8
import os
import time
import json
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest

import threading
import sys

import nls  # 引入阿里云语音识别库



def gettoken ():
    ak_id = os.getenv('ALIYUN_AK_ID')
    ak_secret = os.getenv('ALIYUN_AK_SECRET')
    print(f"AK_ID: {ak_id}, AK_SECRET: {ak_secret}")
    # 创建AcsClient实例
    client = AcsClient(
        ak_id,
        ak_secret,
        "cn-shanghai"
    );


    # 创建request，并设置参数。
    request = CommonRequest()
    request.set_method('POST')
    request.set_domain('nls-meta.cn-shanghai.aliyuncs.com')
    request.set_version('2019-02-28')
    request.set_action_name('CreateToken')

    try:
        response = client.do_action_with_exception(request)
        print(response)

        jss = json.loads(response)
        if 'Token' in jss and 'Id' in jss['Token']:
            token = jss['Token']['Id']
            expireTime = jss['Token']['ExpireTime']
            print("token = " + token)
            print("expireTime = " + str(expireTime))
            return token
    except Exception as e:
        print(e)


# WebSocket服务地址，提供语音转写服务
URL = "wss://nls-gateway-cn-shanghai.aliyuncs.com/ws/v1"

# 在实际使用中，需替换成你自己的TOKEN和APPKEY
TOKEN = gettoken() # 参考https://help.aliyun.com/document_detail/450255.html获取token

print("token=" + TOKEN)

APPKEY = "OKt6jogp6fRjHQVp"  # 获取Appkey请前往控制台：https://nls-portal.console.aliyun.com/applist

# 定义测试类，用于处理音频实时转写
class TestSt:
    def __init__(self, tid, test_file):
        # 初始化测试线程
        self.__th = threading.Thread(target=self.__test_run)  
        self.__id = tid  # 线程ID
        self.__test_file = test_file  # 测试音频文件路径
   
    def loadfile(self, filename):
        # 加载音频文件，读取成二进制数据
        with open(filename, "rb") as f:
            self.__data = f.read()
    
    def start(self):
        # 加载文件并启动线程
        self.loadfile(self.__test_file)
        self.__th.start()

    # 回调函数：句子开始时调用
    def test_on_sentence_begin(self, message, *args):
        print("test_on_sentence_begin:{}".format(message))

    # 回调函数：句子结束时调用
    def test_on_sentence_end(self, message, *args):
        print("test_on_sentence_end:{}".format(message))

    # 回调函数：任务开始时调用
    def test_on_start(self, message, *args):
        print("test_on_start:{}".format(message))

    # 回调函数：发生错误时调用
    def test_on_error(self, message, *args):
        print("on_error args=>{}".format(args))

    # 回调函数：任务关闭时调用
    def test_on_close(self, *args):
        print("on_close: args=>{}".format(args))

    # 回调函数：中间结果变化时调用
    def test_on_result_chg(self, message, *args):
        print("test_on_chg:{}".format(message))

    # 回调函数：任务完成时调用
    def test_on_completed(self, message, *args):
        print("on_completed:args=>{} message=>{}".format(args, message))

    # 私有方法：运行测试
    def __test_run(self):
        print("thread:{} start..".format(self.__id))
        
        # 初始化语音转写对象，绑定回调函数
        sr = nls.NlsSpeechTranscriber(
            url=URL,
            token=TOKEN,
            appkey=APPKEY,
            on_sentence_begin=self.test_on_sentence_begin,
            on_sentence_end=self.test_on_sentence_end,
            on_start=self.test_on_start,
            on_result_changed=self.test_on_result_chg,
            on_completed=self.test_on_completed,
            on_error=self.test_on_error,
            on_close=self.test_on_close,
            callback_args=[self.__id]
        )

        print("{}: session start".format(self.__id))
        
        # 启动转写会话，设置音频格式和相关选项
        r = sr.start(
            aformat="pcm",
            enable_intermediate_result=True,
            enable_punctuation_prediction=True,
            enable_inverse_text_normalization=True
        )

        # 每次读取640字节的音频数据并发送
        self.__slices = zip(*(iter(self.__data),) * 640)
        for i in self.__slices:
            sr.send_audio(bytes(i))
            time.sleep(0.01)  # 模拟实时发送音频数据，避免数据过快发送

        # 控制传输，发送自定义的控制数据
        sr.ctrl(ex={"test": "tttt"})
        time.sleep(1)  # 等待一段时间

        # 停止转写会话
        r = sr.stop()
        print("{}: sr stopped:{}".format(self.__id, r))
        time.sleep(1)

# 多线程测试函数，默认创建500个线程
def multiruntest(num=500):
    for i in range(0, num):
        name = "thread" + str(i)  # 每个线程的名称
        t = TestSt(name, "./test111.wav")  # 创建TestSt实例，指定音频文件路径
        t.start()  # 启动线程

# 禁用详细日志
nls.enableTrace(False)

# 运行测试，启动1个线程进行测试
multiruntest(1)