Go语言实现流式传输音频和帧数据

{ "title": "Go\u005c\u005c语言实现流式传输音频和帧数据", "description": "本文介绍了使用Go\u005c\u005c语言实现流式传输音频和帧数据的代码示例，并对代码进行了优化，提高了代码的可读性和性能。代码示例中使用Websocket进行数据传输，并对音频数据进行解码和处理，最后通过UDP协议将帧数据发送到指定地址。", "keywords": "Go\u005c\u005c语言, 音频流式传输, 帧数据, Websocket, UDP协议", "content": "\u003ccode\u003etype ModelSTARes struct {\n\tCode int json:\"code\"\n\tData struct {\n\t\tContentType string json:\"ContentType\"\n\t\tContent string json:\"content\"\n\t\tVesAnimeResult struct {\n\t\t\tOrder []string json:\"Order\"\n\t\t\tCoefl float32 json:\"Coefl\"\n\t\t\tFps float64 json:\"fps\"\n\t\t\tFrame [][]int32 json:\"Frame\"\n\t\t} json:\"ves_anime_result\"\n\t\tInformation struct {\n\t\t\tPhoneTimestamp [][]interface{} json:\"phone_timestamp\"\n\t\t\tWordTimestamp [][]interface{} json:\"word_timestamp\"\n\t\t} json:\"information\"\n\t\tText []string json:\"text\"\n\t\tDesc string json:\"desc\"\n\t\tIsLast bool json:\"is_last\"\n\t\tIsFinish bool json:\"is_finish\"\n\t\tIsFirst bool json:\"is_first\"\n\t}\n}\n\ntype STAReq struct {\n\tInput string json:\"input\"\n\tGender int json:\"gender\"\n\tFlag int json:\"flag\"\n\tSeq int json:\"seq\"\n}\n\nfunc (this NoticeDetectionCtl) OnModelVoiceDriver() {\n\tvar resMap protocol.STAReq\n\terr := this.UnMarshal(&resMap)\n\tif err != nil {\n\t\tlogs.Error("OnModelVoiceDriver, %s, err = %v", string(this.Ctx.Input.RequestBody), err)\n\t\tthis.OnError(errno.EC_InvalidArg, errno.EM_InvalidArg)\n\t\treturn\n\t}\n\tif resMap.Input == "" {\n\t\tlogs.Error("OnModelVoiceDriver,无效参数")\n\t\tthis.OnError(errno.EC_InvalidArg, errno.EM_InvalidArg)\n\t\treturn\n\t}\n\tlogs.Info("OnModelVoiceDriver, modelId=%d, userId=%d, gender=%d, flag=%d, input=%s", this.Session.ModelId, this.Session.Id, resMap.Gender, resMap.Flag, resMap.Input)\n\tcommonInfo := fmt.Sprintf("OnModelVoiceDriver OnSocket start")\n\n\t// 连接服务器\n\tconn, resp, err := websocket.DefaultDialer.Dial(appconf.TTSStaApi, nil)\n\tif err != nil {\n\t\tthis.OnError(errno.EC_InnerError, errno.EM_InnerError)\n\t\tlogs.Info("OnModelVoiceDriver OnSocket dial error: %v\n", err)\n\t\treturn\n\t}\n\tdefer conn.Close()\n\n\t// 输出响应信息\n\tlogs.Info("OnModelVoiceDriver OnSocket ws resp: Status=%s, StatusCode=%d, Proto=%s\n",\n\t\tresp.Status, resp.StatusCode, resp.Proto)\n\n\tvar strSlice []string\n\t//构造请求数据\n\theader := ModelStaHeader(resMap.Gender)\n\tbody := ModelStaBody(resMap.Input)\n\n\t// 发送音频数据\n\tsendData := map[string]interface{}{"head": header, "data": body}\n\tsendDataBytes, _ := json.Marshal(sendData)\n\terr = conn.WriteMessage(websocket.TextMessage, sendDataBytes)\n\tif websocket.IsCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {\n\t\tlogs.Info("OnModelVoiceDriver Connection closed")\n\t\treturn\n\t} else if err != nil {\n\t\tthis.OnError(errno.EC_InnerError, errno.EM_InnerError)\n\t\tlogs.Error("OnModelVoiceDriver Error sending audio data:", err)\n\t}\n\n\tvar bsFrames [][]int32\n\tvar orderStr [][]string\n\t//var builder strings.Builder\n\t// 接收服务端返回的数据流\n\tfor {\n\t\t_, msg, err := conn.ReadMessage()\n\t\tif websocket.IsCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {\n\t\t\tlogs.Info(" OnModelVoiceDriver Connection closed")\n\t\t\treturn\n\t\t} else if err != nil {\n\t\t\tthis.OnError(errno.EC_InnerError, errno.EM_InnerError)\n\t\t\tlogs.Error("OnModelVoiceDriver read error: %v\n", err)\n\t\t\treturn\n\t\t}\n\n\t\tvar res protocol.ModelSTARes\n\t\terr = json.Unmarshal(msg, &res)\n\t\tif err != nil {\n\t\t\tthis.OnError(errno.EC_InnerError, errno.EM_InnerError)\n\t\t\tlogs.Error("receiveData, Parse JSON response and handle data,err: ", err)\n\t\t\treturn\n\t\t}\n\n\t\t// Handle content type\n\t\tswitch res.Code {\n\t\tcase 200:\n\t\t\t// Decode base64 audio data and save to file\n\t\t\tif len(res.Data.VesAnimeResult.Order) == 0 {\n\t\t\t\tthis.OnError(errno.EC_InnerError, errno.EM_InnerError)\n\t\t\t\tlogs.Error("Failed to get VesAnimeResult.Order: %v\n", err)\n\t\t\t\treturn\n\t\t\t}\n\t\t\tif len(res.Data.VesAnimeResult.Frame[0]) == 0 {\n\t\t\t\tthis.OnError(errno.EC_InnerError, errno.EM_InnerError)\n\t\t\t\tlogs.Error("res.Data.VesAnimeResult.Frame: %v\n", err)\n\t\t\t\treturn\n\t\t\t}\n\t\t\tif res.Data.VesAnimeResult.Coefl == 0 {\n\t\t\t\tthis.OnError(errno.EC_InnerError, errno.EM_InnerError)\n\t\t\t\tlogs.Error("res.Data.VesAnimeResult.Coefl: %v\n", err)\n\t\t\t\treturn\n\t\t\t}\n\t\t\t// 解析数据处理后给UE\n\t\t\tbsFrames = append(bsFrames, res.Data.VesAnimeResult.Frame...)\n\t\t\torderStr = append(orderStr, res.Data.VesAnimeResult.Order)\n\t\t\tcontent, err := base64.StdEncoding.DecodeString(res.Data.Content)\n\t\t\tif err != nil {\n\t\t\t\tthis.OnError(errno.EC_InnerError, errno.EM_InnerError)\n\t\t\t\tlogs.Error("res.Data.Content decode error: %v", err)\n\t\t\t}\n\t\t\tstrSlice = append(strSlice, string(content))\n\t\t\t//builder.WriteString(res.Data.Content)\n\t\tdefault:\n\t\t\tlogs.Error("Received response code: %d\n", res.Code)\n\t\t\tthis.OnError(errno.EC_InnerError, errno.EM_InnerError)\n\t\t}\n\t\t// 根据服务器返回的数据结束连接\n\t\tif res.Data.IsLast && res.Data.IsFinish {\n\t\t\tconn.Close()\n\t\t\tbreak\n\t\t}\n\t\ttime.Sleep(10 * time.Millisecond)\n\n\t}\n\tif len(strSlice) == 0 {\n\t\tlogs.Error("音频数据为空")\n\t\tthis.OnError(errno.EC_InnerError, errno.EM_InnerError)\n\t}\n\tlogs.Info("OnModelVoiceDriver,orderStr： %v", orderStr)\n\tbyteSlice := FixFiles(strSlice)\n\twavStr := base64.StdEncoding.EncodeToString(byteSlice)\n\tgo HandelModelData(bsFrames, 10000, this.Token, commonInfo, resMap.Seq, conn)\n\tlogs.Info("%s, finish", commonInfo)\n\ttime.Sleep(3 * time.Millisecond)\n\tif resMap.Flag == 1 {\n\t\tthis.OnAddText(&resMap)\n\t}\n\tresult := make(map[string]interface{})\n\tif len(wavStr) != 0 {\n\t\t//result["audio"] = builder.String()\n\t\tresult["audio"] = wavStr\n\t\tthis.OnSuccess(result)\n\t} else {\n\t\tthis.OnError(errno.EC_InnerError, errno.EM_InnerError)\n\t}\n}\nfunc HandelModelData(frames [][]int32, coefl float32, token string, commonInfo string, seq int, conn websocket.Conn) {\n\tvar finish bool = true\n\tvar sig chan int\n\t{\n\t\tvar ok bool\n\t\tdriverMutex.Lock()\n\t\tsig, ok = token2driver[token]\n\t\tif !ok {\n\t\t\tsig = make(chan int, 1)\n\t\t\ttoken2driver[token] = sig\n\t\t} else {\n\t\t\tlogs.Info("%s, emmit sig flag, to quit last request", commonInfo)\n\t\t\tsig \u003c- 0 // 退出前面的驱动\n\t\t}\n\t\tdriverMutex.Unlock()\n\t}\n\n\tdefer func() {\n\t\tif finish {\n\t\t\tdriverMutex.Lock()\n\t\t\t_, ok := token2driver[token]\n\t\t\tif ok {\n\t\t\t\tclose(sig)\n\t\t\t\tdelete(token2driver, token)\n\t\t\t}\n\t\t\tdriverMutex.Unlock()\n\t\t\tlogs.Info("%s, finish", commonInfo)\n\t\t}\n\t}()\n\n\tapi := appconf.StreamInfoApi + "?token=" + token\n\tstatus, stream := util.OnGetHttpTimeout(api, 10)\n\tif status != 200 {\n\t\tlogs.Error("%s, api = %v, status = %d, result = %s", commonInfo, api, status, string(stream))\n\t\treturn\n\t}\n\tvar streamInfoRes protocol.StreamInfoRes\n\terr := json.Unmarshal(stream, &streamInfoRes)\n\tif err != nil {\n\t\tlogs.Error("%s, err = %v", commonInfo, err)\n\t\treturn\n\t}\n\tif streamInfoRes.Code != 200 {\n\t\tlogs.Error("%s HandelModelData, err = %v", commonInfo, streamInfoRes)\n\t}\n\tsip := streamInfoRes.Data.Ip\n\tsport := 11111\n\tif sip == "" {\n\t\tlogs.Error("%s, invalid ip", commonInfo)\n\t\treturn\n\t}\n\tsocket, err := net.DialUDP("udp", nil, &net.UDPAddr{\n\t\tIP: net.ParseIP(sip),\n\t\tPort: sport,\n\t})\n\tif err != nil {\n\t\tlogs.Error("%s, ip = %s, port = %d, err = %v", commonInfo, sip, sport, err)\n\t\treturn\n\t}\n\tdefer socket.Close()\n\n\t_ = StartBSDriver(token, seq)\n\n\tlogs.Info("%s, sip = %s, sport = %d", commonInfo, sip, sport)\n\n\t{\n\t\tstartReq := make(map[string]interface{})\n\t\tstartReq["userIp"] = sip\n\t\tstartData, _ := json.Marshal(&startReq)\n\t\tvar nsent = 0\n\t\tfor nsent \u003c len(startData) {\n\t\t\tn, err := socket.Write(startData[nsent:])\n\t\t\tif err != nil {\n\t\t\t\tlogs.Error("%s, ip = %s, port = %d, err = %v", commonInfo, sip, sport, err)\n\t\t\t}\n\t\t\tnsent += n\n\t\t}\n\t\tlogs.Info("%s, ip = %s, port = %d, OnAudioDriver", commonInfo, sip, sport)\n\t}\n\nLoop:\n\tfor i := 0; i \u003c len(frames); i++ {\n\t\tframe := make([]float32, 61)\n\t\tfor j := 0; j \u003c 51; j++ {\n\t\t\tif j == 1 || j == 4 || j == 11 || j == 40 || j == 41 || j == 42 || j == 43 || j == 44 || j == 45 || j == 8 || j == 9 || j == 10 {\n\t\t\t\tif frames[i][j] != 0 {\n\t\t\t\t\tframes[i][j] = 0\n\t\t\t\t}\n\t\t\t}\n\t\t\tframe[j] = float32(frames[i][j]) / coefl\n\t\t}\n\t\tselect {\n\t\tcase \u003c-sig:\n\t\t\tlogs.Info("%s, recv quit sig, quit now", commonInfo)\n\t\t\tfinish = false\n\t\t\tbreak Loop\n\t\tdefault:\n\t\t\tdata := MakeBSFrame(frame)\n\t\t\tvar nsent = 0\n\t\t\tfor nsent \u003c len(data) {\n\t\t\t\tn, err := socket.Write(data[nsent:])\n\t\t\t\tif err != nil {\n\t\t\t\t\tfmt.Println("send, err =", err)\n\t\t\t\t}\n\t\t\t\tnsent += n\n\t\t\t}\n\t\t\ttime.Sleep(20 * time.Millisecond)\n\t\t\t//time.Sleep(15time.Millisecond + 650time.Microsecond)\n\t\t}\n\t}\n\t_ = StopBSDriver(token, seq)\n\treturn\n}\n\u003c/code\u003e