From f4e8f206a6760bdc31734dfcb1c65916b5b76311 Mon Sep 17 00:00:00 2001 From: liuxiaolong <736321739@qq.com> Date: 星期四, 22 八月 2019 12:05:58 +0800 Subject: [PATCH] add cluster --- service/FaceSdkService.go | 337 +++++++++++++++++++++++++++++++++++++++++--------------- 1 files changed, 247 insertions(+), 90 deletions(-) diff --git a/service/FaceSdkService.go b/service/FaceSdkService.go index 7d76284..ca7f185 100644 --- a/service/FaceSdkService.go +++ b/service/FaceSdkService.go @@ -3,61 +3,180 @@ import ( "basic.com/pubsub/protomsg.git" "basic.com/valib/deliver.git" - "encoding/base64" - "fmt" "github.com/gogo/protobuf/proto" "github.com/pierrec/lz4" + "github.com/pkg/errors" "github.com/satori/go.uuid" "gocv.io/x/gocv" + "image" + "sync" "time" + "webserver/extend/logger" + "webserver/extend/util" ) +type FaceSdkService struct { + File []byte + Id int64 + Result []*protomsg.ResultFaceDetect + DeadTime time.Duration +} + const ( - Ipc_Push_Ext = "_2.ipc" - Ipc_Pull_Ext = "_1.ipc" - Ipc_Url_Pre = "ipc:///tmp///" + Ipc_Push_Ext = "_2.ipc" + Ipc_Pull_Ext = "_1.ipc" + Ipc_Url_Pre = "ipc:///tmp///" + faceExtractWebCID = "virtual-face-extract-web-camera-id" Virtual_FaceTaskId = "92496BDF-2BFA-98F2-62E8-96DD9866ABD2" - Virtual_FaceSdkId = "virtual-faceextract-sdk-pull" - Url_Service_PUSH = Ipc_Url_Pre + Virtual_FaceSdkId + Ipc_Push_Ext - Url_Service_PULL = Ipc_Url_Pre + Virtual_FaceSdkId + Ipc_Pull_Ext + Virtual_FaceSdkId = "virtual-faceextract-sdk-pull" + Url_Service_PUSH = Ipc_Url_Pre + Virtual_FaceSdkId + Ipc_Push_Ext + Url_Service_PULL = Ipc_Url_Pre + Virtual_FaceSdkId + Ipc_Pull_Ext ) -var imgPushChan chan protomsg.Recvmsg + +func GetFaceFeaFromSdk(fileBytes []byte,deadTime time.Duration) ([]*protomsg.ResultFaceDetect,error,*protomsg.Image){ + t1 := time.Now() + s := NewFaceSdkService(fileBytes, deadTime) + i, err := s.ReadFromUploadImg() + logger.Debug("ReadFromUploadImg鐢ㄦ椂:", time.Since(t1)) + t1 = time.Now() + if err !=nil{ + logger.Debug("readFromUploadImg err:",err) + return nil,err,i + } + bc, err := ImgCompress(i) + logger.Debug("ImgCompress鐢ㄦ椂:", time.Since(t1)) + t1 = time.Now() + if err !=nil { + logger.Debug("ImgCompress err:",err) + return nil,err,i + } + s.PushImgMsg(bc) + logger.Debug("PushImgMsg鐢ㄦ椂:", time.Since(t1)) + t1 = time.Now() + s.GetFaceFea() + logger.Debug("GetFaceFea鐢ㄦ椂:", time.Since(t1)) + if s.Result == nil{ + return nil,errors.New("no fea"),i + } else { + return s.Result,nil,i + } +} + +func NewFaceSdkService(fileBytes []byte, deadTime time.Duration) FaceSdkService{ + return FaceSdkService{ + File:fileBytes, + Id:time.Now().UnixNano(), + DeadTime:deadTime, + } +} + +var imgPushChan chan []byte var client_push deliver.Deliver var client_pull deliver.Deliver func TestPushImgMsg() { InitService() - i := readImgFile() - fmt.Printf("width:%d,height:%d,data.length:%d,timestamp:%s,id:%d\n",i.Width,i.Height,len(i.Data),i.Timestamp,i.Id) - if b, err := proto.Marshal(&i);err !=nil{ - fmt.Println("protoImage marshal err") - return - } else { - bc := make([]byte,len(b)) - ht := make([]int, 64<<10) - n,err := lz4.CompressBlock(b, bc, ht) - if err !=nil { - fmt.Println(err) - } - if n >= len(b){ - fmt.Println("image is not compressible") - } - bc = bc[:n] - for { - PushImgMsg(protomsg.Recvmsg{ - Id:uuid.NewV4().String(), - Addr:"", - Picdata:bc, - }) - fmt.Println("pushed img") - time.Sleep(5*time.Second) - } + i := readTestImgFile() + logger.Debug("width:%d,height:%d,data.length:%d,timestamp:%s,id:%d\n",i.Width,i.Height,len(i.Data),i.Timestamp,i.Id) + + bc, err := ImgCompress(&i) + if err !=nil { + logger.Debug("image is not compressible") + } else { + var s FaceSdkService + s.PushImgMsg(bc) } } -func readImgFile() protomsg.Image{ +func (s *FaceSdkService) ReadFromUploadImg() (*protomsg.Image,error){ + picMat, err := gocv.IMDecode(s.File, gocv.IMReadColor) + if err !=nil { + logger.Debug("gocv.IMDecode err:",err) + return nil,err + } + logger.Debug("picMat.Data.len:", len(picMat.ToBytes())) + newMat := gocv.NewMat() + size := 1024 + if picMat.Rows() > size || picMat.Cols() > size { + fx := float64(size)/float64(picMat.Rows()) + fy := float64(size)/float64(picMat.Cols()) + ff := fx + if fx > fy{ + ff = fy + } + gocv.Resize(picMat,&newMat, image.Pt(0,0), ff, ff, gocv.InterpolationDefault) + picMat = newMat + } + + defer picMat.Close() + + if picMat.Empty() { + logger.Debug("file not exist") + return nil,errors.New("picMat is empty") + } + height := int32(picMat.Rows()) + width := int32(picMat.Cols()) + data := picMat.ToBytes() + timeUnix := time.Now().Unix() + formatTimeStr := time.Unix(timeUnix, 0).Format("2006-01-02 15:04:05") + + return &protomsg.Image{ + Width: width, + Height: height, + Timestamp: formatTimeStr, + Data: data, + Id: s.Id, + Cid: faceExtractWebCID, + },nil +} + +func ImgCompress(i *protomsg.Image) ([]byte,error){ + if b, err := proto.Marshal(i); err != nil { + logger.Debug("protoImage marshal err") + return nil,err + } else { + bc := make([]byte, len(b)) + ht := make([]int, 64<<10) + n, err := lz4.CompressBlock(b, bc, ht) + if err != nil { + logger.Debug(err) + return nil,err + } + if n >= len(b) { + logger.Debug("image is not compressible") + return nil,errors.New("compressed len is 0") + } + bc = bc[:n] + return bc,nil + } +} + +func (s *FaceSdkService) GetFaceFea(){ + var wg sync.WaitGroup + wg.Add(1) + ticker := time.NewTicker(s.DeadTime) + go func(ticker *time.Ticker, s *FaceSdkService) { + defer ticker.Stop() + defer wg.Done() + + for { + select { + case <-ticker.C: + return + default: + if faces,ok := resultMap.Get(s.Id);ok { + s.Result = faces + return + } + } + } + }(ticker, s) + wg.Wait() +} + +func readTestImgFile() protomsg.Image{ var i protomsg.Image timeUnix := time.Now().Unix() formatTimeStr := time.Unix(timeUnix, 0).Format("2006-01-02 15:04:05") @@ -68,38 +187,62 @@ defer picMat.Close() if picMat.Empty() { - fmt.Println("file not exist") + logger.Debug("file not exist") return i } height := int32(picMat.Rows()) width := int32(picMat.Cols()) data := picMat.ToBytes() - wrMat,_ := gocv.NewMatFromBytes(picMat.Rows(),picMat.Cols(),gocv.MatTypeCV8UC3,data) - - gocv.IMWrite("xxx.jpg", wrMat) + //wrMat,_ := gocv.NewMatFromBytes(picMat.Rows(),picMat.Cols(),gocv.MatTypeCV8UC3,data) + // + //gocv.IMWrite("xxx.jpg", wrMat) i = protomsg.Image{ - Width: width, - Height: height, + Width: width, + Height: height, Timestamp: formatTimeStr, - Data: data, - Id: timeUnix, + Data: data, + Id: timeUnix, } - fmt.Println("gocv read img completed") + i.Cid = uuid.NewV4().String() //鏁版嵁鍞竴id + logger.Debug("gocv read img completed") return i } -func PushImgMsg(is protomsg.Recvmsg){ +func (s *FaceSdkService) PushImgMsg(is []byte){ imgPushChan <- is } -var resultMap map[string]protomsg.SdkMessage +type FeaResult struct { + FaceM map[int64][]*protomsg.ResultFaceDetect + Lock sync.Mutex +} +func (f *FeaResult) Write(id int64,faceDetectResult []*protomsg.ResultFaceDetect){ + f.Lock.Lock() + defer f.Lock.Unlock() + f.FaceM[id] = faceDetectResult +} -func InitService(){ - fmt.Println("service init!") - imgPushChan = make(chan protomsg.Recvmsg) - resultMap = make(map[string]protomsg.SdkMessage,0) +func (f *FeaResult) Get(id int64) ([]*protomsg.ResultFaceDetect,bool){ + f.Lock.Lock() + defer f.Lock.Unlock() + coms,ok := f.FaceM[id] + return coms,ok +} + +func (f *FeaResult) Delete(id int64){ + f.Lock.Lock() + defer f.Lock.Unlock() + delete(f.FaceM,id) +} + +var resultMap = FeaResult{} + +func InitService() { + logger.Debug("service init!") + imgPushChan = make(chan []byte) + resultMap.FaceM = make(map[int64][]*protomsg.ResultFaceDetect,0) client_push = deliver.NewClient(deliver.PushPull, Url_Service_PUSH) client_pull = deliver.NewClient(deliver.PushPull, Url_Service_PULL) defer func() { @@ -111,61 +254,75 @@ go thRecv() } -func thSend(){ +func thSend() { for { select { - case is := <- imgPushChan: - fmt.Println("imgPushChan in") - b, _ := proto.Marshal(&is) - err := client_push.Send(b) - if err !=nil { - fmt.Println("img Send err:",err) + case d := <-imgPushChan: + logger.Debug("imgPushChan in") + err := client_push.Send(d) + if err != nil { + logger.Debug("img Send err:", err) } default: - //fmt.Println("no img in") + //logger.Debug("no img in") } } } -func thRecv(){ +func thRecv() { for { resultBytes, err := client_pull.Recv() - if err !=nil{ - fmt.Println("pull err:",err) + if err != nil { + //logger.Debug("pull err:", err) continue } rMsg := protomsg.SdkMessage{} - if err := proto.Unmarshal(resultBytes, &rMsg);err ==nil{ - fmt.Println("received MSG:",rMsg.Cid) - perId := rMsg.Cid //鏁版嵁id - if rMsg.Tasklab !=nil && rMsg.Tasklab.Taskid == Virtual_FaceTaskId { - sdkInfos := rMsg.Tasklab.Sdkinfos - fmt.Println("Len(sdkInfos)=",len(sdkInfos)) - for _,swt :=range sdkInfos{ - fmt.Println("sdkName:",swt.SdkName) - if swt.Sdktype == "FaceExtract"{ - - fmt.Println("sdkData.len:",len(swt.Sdkdata)) - var pff protomsg.ParamFaceFeature - if err := proto.Unmarshal(swt.Sdkdata, &pff);err !=nil{ - //fmt.Println("ParamFaceFeature unmarshal err:",err) - } else { - fmt.Println("鐩爣鏁帮細",len(pff.ExtComp)) - for _,fea :=range pff.ExtComp{ - base64Fea := base64.StdEncoding.EncodeToString(fea.Feats) - fmt.Println("perId:",perId) - fmt.Println("faceFeature:",base64Fea) - } - } - break - } - } - + if err := proto.Unmarshal(resultBytes, &rMsg); err == nil { + i := protomsg.Image{} + bdata, err := util.UnCompress(rMsg.Data) + if err !=nil { + logger.Debug("uncompress err:",err) + continue } - //resultMap[rMsg.Cid] = rMsg + err = proto.Unmarshal(bdata, &i) + if err !=nil { + continue + } + perId := i.Id //鏁版嵁id + if rMsg.Tasklab != nil && rMsg.Tasklab.Taskid == Virtual_FaceTaskId { + sdkInfos := rMsg.Tasklab.Sdkinfos + + logger.Debug("Len(sdkInfos)=",len(sdkInfos)) + for _,swt :=range sdkInfos{ + //浜鸿劯妫�娴嬬殑缁撴灉 + if swt.Sdktype =="FaceDetect"{ + logger.Debug("浜鸿劯妫�娴嬬粨鏋滈暱搴︼細",len(swt.Sdkdata)) + if len(swt.Sdkdata)>1{ + var pfp protomsg.ParamFacePos + err := proto.Unmarshal(swt.Sdkdata, &pfp) + if err !=nil { + logger.Debug("faceDetect result unmarshal err:",err) + } else { + logger.Debug("妫�娴嬩汉鑴告暟:",len(pfp.Faces)) + if len(pfp.Faces)>0{ + resultMap.Write(perId,pfp.Faces) + } + //for _,face :=range pfp.Faces{ + //logger.Debug("FacePos:",face.Pos) + //logger.Debug("ThftResult:",face.Result) + // base64Fea := base64.StdEncoding.EncodeToString(face.Feats) + // logger.Debug("perId:",perId) + // logger.Debug("faceFeature:",base64Fea) + //} + } + } + break + } + } + } } else { - fmt.Println("recv msg Err:",err) + logger.Debug("recv msg Err:", err) } } -} \ No newline at end of file +} -- Gitblit v1.8.0