panlei
2019-09-16 c91b363644ac6fab27fae10215d4993f133df619
main.go
@@ -4,8 +4,11 @@
   "basic.com/dbapi.git"
   "basic.com/pubsub/protomsg.git"
   "basic.com/valib/deliver.git"
   "net/http"
   _ "net/http/pprof"
   "ruleprocess/insertdata"
   "ruleprocess/util"
   "time"
   //"bufio"
   //"bytes"
@@ -21,26 +24,33 @@
   "ruleprocess/ruleserver"
   "sync"
)
var dbIp = flag.String("dbIp","127.0.0.1","dbserver ip")
var dbPort = flag.Int("dbPort",8001,"default dbPort=8001")
var surveyPort = flag.Int("surveyPort",40007,"survey port") //心跳
var pubPort = flag.Int("pubPort",50007,"pubsub port") //数据更新
var dbIp = flag.String("dbIp", "127.0.0.1", "dbserver ip")
var dbPort = flag.Int("dbPort", 8001, "default dbPort=8001")
var surveyPort = flag.Int("surveyPort", 40007, "survey port") //心跳
var pubPort = flag.Int("pubPort", 50007, "pubsub port")       //数据更新
var initchan = make(chan bool)
func init () {
var env =  flag.String("env","pro","env set")
func init() {
   flag.Parse()
   // 日志初始化
   logger.Config("./info.log", logger.DebugLevel)
   logger.Info("日志初始化成功!")
   insertdata.Init(*env)
}
func main() {
   //fmt.Println("缓存初始化完成",<- initchan)//dbserver初始化完毕
   go func() {
      http.ListenAndServe("0.0.0.0:8899",nil)
   }()
   flag.Parse()
   wg := sync.WaitGroup{}
   wg.Add(2)
   wg.Add(3)
   dbapi.Init(*dbIp,*dbPort)
   go cache.Init(initchan,*dbIp,*surveyPort,*pubPort)
   logger.Info("cache init completed!!!",<- initchan)//dbserver初始化完毕
   //fmt.Println("缓存初始化完成",<- initchan)//dbserver初始化完毕
   dbapi.Init(*dbIp, *dbPort)
   go cache.Init(initchan, *dbIp, *surveyPort, *pubPort)
   logger.Info("cache init completed!!!", <-initchan) //dbserver初始化完毕
   ruleserver.Init()
   go ruleserver.TimeTicker()
   go ruleserver.StartServer()
@@ -56,32 +66,43 @@
   var msg []byte
   var err error
   //msgChan := make(chan []byte,100)
   for {
      select {
      // case <-ctx.Done():
      //    return
      default:
         msg, err = c.Recv()
         //msgChan <- msg
         if err != nil {
            logger.Info("recv error : ", err)
            fmt.Println("recv error : ", err)
            continue
         } else {
            arg := ruleserver.SdkDatas{}
            //paramFormat(msg, &arg)
            m := paramFormat(msg, &arg)
            logger.Info("解析出来的数据:", arg)
            ruleserver.Judge(&arg,&m) // 把sdkMessage传进去,方便缓存数据时拼出一个resultMag
            // 把arg里的打的标签拿出来给m再封装一层
            resultMag := ruleserver.ResultMsg{SdkMessage: &m, RuleResult: arg.RuleResult}
            //logger.Info("打完标签后的结果:",resultMag)
            // 将打完标签的数据插入到ES
            insertdata.InsertToEs(resultMag)
            //runtime.GOMAXPROCS(runtime.NumCPU())
            //logger.Debug("使用的cpu个数:",runtime.NumCPU())
            //go func(msg []byte) {
               logger.Debug("当前时间戳:", time.Now().Unix())
               arg := ruleserver.SdkDatas{}
               //paramFormat(msg, &arg)
               start := time.Now()
               m := paramFormat(msg, &arg)
               // 进行规则处理判断(打上规则的标签)
               ruleserver.Judge(&arg, &m) // 把sdkMessage传进去,方便缓存数据时拼出一个resultMag
               // 把arg里的打的标签拿出来给m再封装一层
               resultMsg := ruleserver.ResultMsg{SdkMessage: &m, RuleResult: arg.RuleResult}
               logger.Debug("规则判断完所用时间:", time.Since(start))
               // 将打完标签的数据插入到ES
               insertdata.InsertToEs(resultMsg)
               logger.Debug("插入完Es所用时间:", time.Since(start))
               //事件推送
               //go labelFilter.Judge(resultMsg)
            //}(msg)
         }
      }
   }
}
// 将外部传进来的rect(top,bottom,left,right)转化为自己内部的rect(left top width height)
func rectFormat(rcobj *protomsg.Rect) ruleserver.Rect {
@@ -109,17 +130,6 @@
      panic("解析msg时出现错误")
   }
   // 找201的人脸数据
   //for _, sdkinfo := range m.Tasklab.Sdkinfos {
   //   if sdkinfo.Sdktype == "FaceDetect" && m.Cid == "e7e3f5b0-d0ef-40af-a9c9-eefd736937bf"{ // 人脸检测
   //      if len(sdkinfo.Sdkdata) > 1 {
   //         logger.Debug("有201的人脸数据了")
   //      } else {
   //         logger.Debug("没有201的人脸数据")
   //      }
   //   }
   //}
   // 先进行一下追踪
   ruleserver.FaceIsSame(&m)
   args.CameraId = m.Cid
@@ -131,6 +141,7 @@
   }
   i := protomsg.Image{}
   err = proto.Unmarshal(bdata, &i)
   logger.Info("接到数据,摄像机为:", m.Cid, "图片的id为:", i.Id)
   //logger.Info("----------------看看有几个算法:",len(m.Tasklab.Sdkinfos))
   for _, sdkinfo := range m.Tasklab.Sdkinfos { // yolo算法
      if sdkinfo.Sdktype == "Yolo" {
@@ -138,18 +149,18 @@
         arg.TaskId = m.Tasklab.Taskid
         arg.IpcId = sdkinfo.Ipcid
         arg.IsYolo = true
         bdata, err := util.UnCompress(m.Data)
         if err != nil {
            panic("解压缩图片时出现错误")
         }
         i := protomsg.Image{}
         err = proto.Unmarshal(bdata, &i)
         //bdata, err := util.UnCompress(m.Data)
         //if err != nil {
         //   panic("解压缩图片时出现错误")
         //}
         //i := protomsg.Image{}
         //err = proto.Unmarshal(bdata, &i)
         arg.ImageWidth = int(i.Width)
         arg.ImageHeight = int(i.Height)
         // 暂时写死,sdk还没有这俩算法
         arg.KeepRight = false
         arg.IsStatic = false
         logger.Info("-----追踪之后sdkinfo.Sdkdata的长度为:----",len(sdkinfo.Sdkdata))
         logger.Info("-----追踪之后sdkinfo.Sdkdata的长度为:----", len(sdkinfo.Sdkdata))
         if len(sdkinfo.Sdkdata) > 1 {
            // 大于1才有数据
            yoloParam := protomsg.ParamYoloObj{}
@@ -162,13 +173,13 @@
            for _, info := range yoloParam.Infos {
               if info.Typ == 0 {
                  //logger.Debug("-------------yolo的坐标有几个",info.RcObj)
                  photoMap := ruleserver.PhotoMap{Rects: rectFormat(info.RcObj), Score: float64(info.Prob)*100,IsYolo:true}
                  photoMap := ruleserver.PhotoMap{Rects: rectFormat(info.RcObj), Score: float64(info.Prob) * 100, IsYolo: true}
                  arg.Photo = append(arg.Photo, photoMap)
                  yoloNum++
               }
            }
            logger.Info("--------------追踪之后yolo的个数:",yoloNum)
            args.Sdkdata = append(args.Sdkdata,&arg)
            logger.Info("--------------追踪之后yolo的个数:", yoloNum)
            args.Sdkdata = append(args.Sdkdata, &arg)
         } else {
            continue
         }
@@ -179,12 +190,12 @@
         arg.TaskId = m.Tasklab.Taskid
         arg.IpcId = sdkinfo.Ipcid
         arg.IsYolo = false
         bdata, err := util.UnCompress(m.Data)
         if err != nil {
            panic("解压缩图片时出现错误")
         }
         i := protomsg.Image{}
         err = proto.Unmarshal(bdata, &i)
         //bdata, err := util.UnCompress(m.Data)
         //if err != nil {
         //   panic("解压缩图片时出现错误")
         //}
         //i := protomsg.Image{}
         //err = proto.Unmarshal(bdata, &i)
         arg.ImageWidth = int(i.Width)
         arg.ImageHeight = int(i.Height)
         // 暂时写死,sdk还没有这俩算法
@@ -197,17 +208,17 @@
               logger.Info("解析FACE sdk数据时出现错误", err)
               continue
            }
            logger.Info("--------------追踪之后yolo的个数:",len(faceParam.Faces))
            logger.Info("--------------追踪之后人脸的个数:", len(faceParam.Faces))
            for _, info := range faceParam.Faces {
               //logger.Info("_______________________________________________第一次看相似值:",info.Pos.FAngle.Confidence*100)
               photoMap := ruleserver.PhotoMap{Rects: rectFormat(info.Pos.RcFace), Score: float64(info.Pos.FAngle.Confidence*100), IsYolo:false,ThftRes:*(info.Result),Feature:info.Feats}
               photoMap := ruleserver.PhotoMap{Id: info.Pos.FaceID, Rects: rectFormat(info.Pos.RcFace), Score: float64(info.Pos.FAngle.Confidence * 100), IsYolo: false, ThftRes: *(info.Result), Feature: info.Feats}
               arg.Photo = append(arg.Photo, photoMap)
            }
            args.Sdkdata = append(args.Sdkdata,&arg)
            args.Sdkdata = append(args.Sdkdata, &arg)
         } else {
            continue
         }
      }
   }
   return m
}
}