ROS机器人设计参考.

class KaldiASR():
    def __init__(self):
        current_dir = os.path.dirname(os.path.realpath(__file__))  # 获取当前文件夹
        model_path_name = os.path.join(current_dir,"..","model","sherpa-onnx-paraformer-zh-small-2024-03-09")
        print("离线语音识别模型开始初始化...")
        # 初始化语音识别
        offline_asr.init(model_path_name)
        print("离线语音识别模型初始化完成")

    def audio_file_to_text(self, wav_name):
        text = offline_asr.asr(wav_name)
        #print("kaldi识别结果:",text)
        return text

def asr(*sound_files):
    global args
    global recognizer
    global contexts_list
    start_time = time.time()
 
    streams = []
    total_duration = 0
    for wave_filename in sound_files:
        samples, sample_rate = read_wave(wave_filename)
        duration = len(samples) / sample_rate
        total_duration += duration
        if contexts_list:
            s = recognizer.create_stream(contexts_list=contexts_list)
        else:
            s = recognizer.create_stream()
        s.accept_waveform(sample_rate, samples)
 
        streams.append(s)
 
    recognizer.decode_streams(streams)
    results = [s.result.text for s in streams]
    end_time = time.time()
 
    for wave_filename, result in zip(sound_files, results):
        return f"{result}"

class WsProcess():
    def __init__(self, wsParam):
        websocket.enableTrace(False)
        wsUrl = wsParam.create_url()
        self.ws = websocket.WebSocketApp(wsUrl, on_message=self.on_message, on_error=on_error, on_close=on_close, on_open=on_open)
        self.answer = ""


    # 收到websocket消息的处理
    def on_message(self, message):
        data = json.loads(message)
        code = data['header']['code']
        if code != 0:
            print(f'请求错误: {code}, {data}')
            self.ws.close()
            #self.answer += data['header']['message']
        else:
            choices = data["payload"]["choices"]
            status = choices["status"]
            content = choices["text"][0]["content"]
            self.answer += content
            if status == 2:
                self.ws.close()

    def run(self, appid, domain, query):
        self.ws.appid = appid
        self.ws.query = query
        self.ws.domain = domain
        self.timeout = False
        self.timer = threading.Timer(5, self.on_timeout) #5秒等待时间，如果大模型回复的内容多，就可能等待时间比较久
        self.timer.start()
        self.ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
        self.timer.cancel()
        if self.timeout:
            print("大模型网络连接超时")
            return "抱歉，我的网络好像卡住了"
        else:
            return self.answer

    def on_timeout(self):
        print("llm请求超时!")
        self.timer.cancel()
        self.timeout = True
        self.ws.close()

class YoudaoTranslate(BaseTranslate):
    def __init__(self):
        super(YoudaoTranslate, self).__init__()
        self.home = 'https://dict.youdao.com/'
 
    def get_tts(self, text, lan='', type_=2, *args, **kwargs):
        """ 获取发音
        :param text: 源文本
        :param lan: 文本语言
        :param type_: 发音类型
        :return: 文本语音
        """
        path = 'dictvoice'
        params = {'audio': text, 'le': lan, 'type': type_}
        response = self._get(path, params)
        if response==None:
            return None

        if len(response.content)==0:
            return None

        return response.content

            torch.onnx.export(model, img, f, verbose=False, opset_version=13,
                              training=torch.onnx.TrainingMode.EVAL,
                              do_constant_folding=True,
                              input_names=['images'],
                              output_names=['num_dets', 'det_boxes', 'det_scores', 'det_classes']
                              if args.end2end else ['outputs'],
                              dynamic_axes=dynamic_axes)

            torch.onnx.export(model, img, f, verbose=False, opset_version=12,
                              training=torch.onnx.TrainingMode.EVAL,
                              do_constant_folding=True,
                              input_names=['images'],
                              output_names=['num_dets', 'det_boxes', 'det_scores', 'det_classes']
                              if args.end2end else ['output1','output2','output3'], #['outputs'],
                              dynamic_axes=dynamic_axes)

class Detect(nn.Module):
    export = False

class Detect(nn.Module):
    export_rknn = True #导出rknn友好模型
    export = False

    def forward(self, x):
        if self.training:
            cls_score_list = []
            reg_distri_list = []
            reg_lrtb_list = []

    def _rknn_opt_head(self, x): #yolov6n,s会运行这个文件
        output_for_rknn = []
        for i in range(self.nl):
            x[i] = self.stems[i](x[i])
            reg_feat = self.reg_convs[i](x[i])
            reg_output = self.reg_preds[i](reg_feat)

            cls_feat = self.cls_convs[i](x[i])
            cls_output = self.cls_preds[i](cls_feat)
            cls_output = torch.sigmoid(cls_output) #npu支持sigmoid 消耗1ms以下

            conf_max,_ = cls_output.max(1, keepdim=True) #此行实际会调用reducemax，npu支持max，但没有reducemax
            #rknn推理仅cpu支持reducemax argmax sum等操作，并且argmax量化效果差
            
            out = torch.cat((reg_output,conf_max,cls_output),dim=1) #4(reg)+1(max)+80(cls)
            
            #out = torch.cat((reg_output,cls_output),dim=1) #4+80这种组合节约npu时间，但是后处理时间增加,cpu占用高
            
            output_for_rknn.append( out )
            
        return output_for_rknn
    
	def forward(self, x):
        
        if self.export_rknn:
            return self._rknn_opt_head(x)
            
        if self.training:
            cls_score_list = []
            reg_distri_list = []
            reg_lrtb_list = []