from PyQt5.QtCore import QObject, pyqtSignal import grpc import iat_pb2 import iat_pb2_grpc import threading import queue import time import uuid import hashlib from util import constants import lib_nlp_grpc import config from logger_config import logger url = '124.243.226.40:17022' class AudioRecorderClient(QObject): trigger = pyqtSignal(object) # 新信号，用于在收到响应时发射 def __init__(self, vad_flag): super().__init__() self.hibernate = False self.stub = None self.channel = None self.vad_flag = vad_flag self.sid = get_auth_id() self.server_address = url self.is_engine_end = False ''' RPC error: <_MultiThreadedRendezvous of RPC that terminated with: status = StatusCode.UNAVAILABLE details = "Connection reset" debug_error_string = "UNKNOWN:Error received from peer {created_time:"2024-06-04T02:43:28.2852678+00:00", grpc_status:14, grpc_message:"Connection reset"}" > ''' # TODO，下面2行代码是不是应该每次识别的时候，都重新做一次，上面的报错在唤醒模式下偶现 # 2次唤醒后，长时间不说话，可能造成grpc连接问题，需要测试演示笔是否也有问题？ 放到start_speech里面是否会更好？ # self.channel = grpc.insecure_channel(self.server_address) # self.stub = iat_pb2_grpc.IatStub(self.channel) self.request_queue = queue.Queue() # 请求队列，用于管理音频数据和控制消息 self.running = False # 获取当前系统时间戳，用于判断对话id是否过期，过期需要重新申请 self.init_timestamp = time.time() # 调用引擎初始化对话，返回的对话id dialog_data = lib_nlp_grpc.create_dialog() self.did = dialog_data["did"] # self.did = str(uuid.uuid4()) print('重新开启引擎连接' + str(self.did)) self.hotwords = dialog_data["hotwords"] def _request_generator(self): # 生成器函数，用于发送音频数据和接收响应 while True: request = self.request_queue.get() # 从队列中获取请求 # logger.info(request) # 演示器松开是，调用stop_speech，将self.request_queue.put(None)置为None时，代表语音完成，跳出while循环 # 麦克风模式，当收到引擎返回的endFlag标记时，也会清空request_queue if request is None: # None 作为结束信号 print("Received None, closing the request generator.") break # 如果已经收到引擎返回结束标识了，也不需要再发送音频了 # if self.is_engine_end: # print("引擎已结束，跳出循环发送音频数据") # break # grpc接口文档中3.3章节要求，发送音频有间隔，应该是针对模拟音频数据，此处已经是用户说话场景了，应该不需要发送间隔了 # if request.samples: # 如果请求包含音频样本 # time.sleep(0.02) # 引擎为16k16bit，每次发送1280大小音频段，那么每次应该等待40ms，目前演示笔每段音频大小是320，间隔20ms # 打印请求信息，如是否为结束标志和样本数据长度 # print( # f"Sending request: sessionParam={request.sessionParam}, endFlag={request.endFlag}, samples length={len(request.samples)}") yield request # def _listen_for_responses(self): # # 监听响应 # try: # for response in self.response_iterator: # self.trigger.emit(response) # 发射响应信号 # except grpc.RpcError as e: # print(f"RPC error: {e}") def _listen_for_responses(self): ans_str_cache = "" # 缓存中间结果的字符串 try: endFlag = False for response in self.response_iterator: print(response) print("----") # 检查是否存在中间结果，如果存在则缓存 if response.ansStr: ans_str_cache += response.ansStr if response.endFlag: endFlag = True ''' 使用VAD后端点模式，引擎已经返回识别结束标识了，因为麦克风一直监听中，程序可能还在发送音频，会收到此响应，需要过滤掉此类响应 errStr: "session already stop" errCode: 30019 endFlag: true ''' # if response.endFlag and response.errCode == 30019: # continue if response.endFlag and response.errCode: print("errCode有值，直接跳出") continue print("self.hibernate:", self.hibernate) if self.hibernate: return # 检查是否为最后一个响应，如果是，则发射信号 if endFlag: self.is_engine_end = True self.running = False # 发射信号，将缓存的中间结果和最后一次的结果一起发送 # self.trigger.emit(ans_str_cache + response.ansStr) # 如果使用后断点识别，因为没有主动调用stop_speech发送endFlag=True，需要在接收到引擎的endFlag标记时，清空request_queue if self.vad_flag: self.request_queue.put(iat_pb2.IatRequest(endFlag=True)) self.request_queue.put(None) self.running = False # 关闭grpc连接 # self.channel.close() # print("关闭grpc") self._callback(self.gen_q_data(constants.AITEST_AIUI_RESULT, ans_str_cache)) print("最终识别结果:", ans_str_cache) if ans_str_cache != "": if ans_str_cache.startswith("搜索"): nlp_result = '{"action_type":"9", "action_content":"' + ans_str_cache[2:] + '"}' self._callback(self.gen_q_data(constants.AITEST_GPRC_NLP, nlp_result)) else: local_command = False # 新判断本地命令集合是否有满足的命令 local_commands = config.local_commands for row_index, command in enumerate(local_commands): # 将action_content按逗号分割为列表 action_list = command[3].split(',') # 检查列表中的任何一个字符串是否存在于ans_str_cache中 local_command = any(action in ans_str_cache for action in action_list) if local_command: nlp_result = '{"action_type":"' + str( command[2]) + '", "action_content":"' + command[4] + '"}' self._callback(self.gen_q_data(constants.AITEST_GPRC_NLP, nlp_result)) break if not local_command: # 调用语义引擎 curr_timestamp = time.time() # 判断是否超过24小时，如果超过，需要调用接口重新申请 # 计算时间戳之间的差值（单位为秒） time_difference = curr_timestamp - self.init_timestamp # 将差值转换为小时 hours_difference = time_difference / 3600 # 1小时 = 3600秒 # 如果差值超过24小时 if hours_difference > 24: self.init_timestamp = curr_timestamp self.did = lib_nlp_grpc.create_dialog() nlp_result = lib_nlp_grpc.get_nlp_result(self.did, ans_str_cache) self._callback(self.gen_q_data(constants.AITEST_GPRC_NLP, nlp_result)) else: # 识别文本为空时，也要通知，麦克风唤醒模式，需要通知麦克风重新开始收音 nlp_result = '{}' self._callback(self.gen_q_data(constants.AITEST_GPRC_NLP, nlp_result)) # 重置缓存 ans_str_cache = "" except grpc.RpcError as e: print(f"RPC error: {e}") ''' <_MultiThreadedRendezvous of RPC that terminated with: status = StatusCode.UNAVAILABLE details = "Connection reset" debug_error_string = "UNKNOWN:Error received from peer {created_time:"2024-07-01T11:32:19.6209787+00:00", grpc_status:14, grpc_message:"Connection reset"}" > ''' # 后端点模式，如果引擎连接报错，需要手动通知麦克风重新开启录音 # 会收到多次响应，都是这个Exception，这样写会造成多次重新开启录音 if self.vad_flag: nlp_result = '{}' self._callback(self.gen_q_data(constants.AITEST_GPRC_NLP, nlp_result)) def start_speech(self): print("start_speech:", self.running) self.is_engine_end = False self.request_queue = queue.Queue() # 请求队列，用于管理音频数据和控制消息 self.running = False if not self.running: self.channel = grpc.insecure_channel(self.server_address) self.stub = iat_pb2_grpc.IatStub(self.channel) self.running = True print("self.running被修改为True了") sid = str(uuid.uuid4()) # 发送初始请求，例如会话参数 if self.vad_flag: self.request_queue.put(iat_pb2.IatRequest( sessionParam={"sid": sid, "aue": "raw", "rst": "plain", "rate": "16k", "hotword": self.hotwords, "domain": "", "appId": "", "bizId": "", "resIdList": "", "personal": "", "eos": "1500", "engine_param": "srec_param_bVadOn=true;srec_param_nVadOutPut=0"}, endFlag=False)) print("111111111111111111111") else: self.request_queue.put(iat_pb2.IatRequest( sessionParam={"sid": sid, "aue": "raw", "rst": "plain", "rate": "16k", "hotword": self.hotwords, "domain": "", "appId": "", "bizId": "", "resIdList": "", "personal": ""}, endFlag=False)) print("222222222222222222222") # 初始化请求生成器并发送初始化请求 self.response_iterator = self.stub.createRec(self._request_generator()) threading.Thread(target=self._listen_for_responses, daemon=True).start() def audio_write(self, audio_chunk): if self.running: # 向队列中添加音频数据 # print(audio_chunk) self.request_queue.put(iat_pb2.IatRequest(samples=audio_chunk, endFlag=False)) def stop_speech(self): if self.running: # 可选，发送最后一个音频块后等待 # 实际测试时，这个sleep必须有，不然识别结果不正确 # time.sleep(0.02) # 引擎为16k16bit，每次发送1280大小音频段，那么每次应该等待40ms，目前演示笔每段音频大小是320，间隔20ms # 发送结束信号 self.request_queue.put(iat_pb2.IatRequest(endFlag=True)) # 发送None以关闭生成器 self.request_queue.put(None) self.running = False # 唤醒模式下，超时后，需要主动修改self.running，防止未置为False， 导致唤醒后无法启动 def stop_running(self): print("stop_running") self.hibernate = True # 发送结束信号 self.request_queue.put(iat_pb2.IatRequest(endFlag=True)) # 发送None以关闭生成器 self.request_queue.put(None) self.running = False def start_running(self): print("start_running") self.hibernate = False self.request_queue = queue.Queue() self.running = False def close(self): self.channel.close() def _callback(self, data: dict): try: self.trigger.emit(data) except Exception as e: print(2) print(e) # 生成json数据模版 def gen_q_data(self, code, data): return {"code": code, "data": data} def get_auth_id(): mac = uuid.UUID(int=uuid.getnode()).hex[-12:] return hashlib.md5(":".join([mac[e:e + 2] for e in range(0, 11, 2)]).encode("utf-8")).hexdigest()