事件推送

panlei

2019-07-24 d0bc216665a6d53135641ff009014a02a3774dd9

事件推送

 go.mod

@@ -3,9 +3,9 @@
go 1.12

require (
    basic.com/dbapi.git v0.0.0-20190701055817-73bca225181f
    basic.com/dbapi.git v0.0.0-20190724082851-b6ae90344405
    basic.com/pubsub/cache.git v0.0.0-20190718093725-6a413e1d7d48
    basic.com/pubsub/protomsg.git v0.0.0-20190722020214-73680afe2ba3
    basic.com/pubsub/protomsg.git v0.0.0-20190724081647-e1995f9a80e8
    basic.com/pubsub/sdkcompare.git v0.0.0-20190715013640-f536a4647d00
    basic.com/valib/deliver.git v0.0.0-20190531095353-25d8c3b20051
    basic.com/valib/gopherdiscovery.git v0.0.0-20190605034340-15d89d8b4e28

 go.sum

@@ -1,5 +1,7 @@
basic.com/dbapi.git v0.0.0-20190701055817-73bca225181f h1:CVB4pmUXTPPFN7w/DQfU1YWg2Tp2gG93TTr1M5W86sE=
basic.com/dbapi.git v0.0.0-20190701055817-73bca225181f/go.mod h1:eDXPnxaz6jZPDvBSk7ya7oSASWPCuUEgRTJCjsfKt/Q=
basic.com/dbapi.git v0.0.0-20190724082851-b6ae90344405 h1:BJzdtGipKxQAaptrwUNOVQZ3Qx4jbeAf72wkqBmm5vE=
basic.com/dbapi.git v0.0.0-20190724082851-b6ae90344405/go.mod h1:eDXPnxaz6jZPDvBSk7ya7oSASWPCuUEgRTJCjsfKt/Q=
basic.com/pubsub/cache.git v0.0.0-20190712095028-e73efb4afc3b h1:UAasACFqEYUBCuZkkdxYVc1QmSyB7McvNHS36QxDJp4=
basic.com/pubsub/cache.git v0.0.0-20190712095028-e73efb4afc3b/go.mod h1:gHLJZz2ee1cGL0X0ae69fs56bAxkDgEQwDhhXZJNUcY=
basic.com/pubsub/cache.git v0.0.0-20190718024458-be52360c4814 h1:KoSik/aiJNDt3d+qRKExLW4pNHZ7vU1wXHhWXxZi4qo=
@@ -8,6 +10,8 @@
basic.com/pubsub/cache.git v0.0.0-20190718093725-6a413e1d7d48/go.mod h1:gHLJZz2ee1cGL0X0ae69fs56bAxkDgEQwDhhXZJNUcY=
basic.com/pubsub/protomsg.git v0.0.0-20190722020214-73680afe2ba3 h1:qgY1QFopeLa24Q9LGToXAVkbU9+QUMLzTxRsYur0tgs=
basic.com/pubsub/protomsg.git v0.0.0-20190722020214-73680afe2ba3/go.mod h1:un5NV5VWQoblVLZfx1Rt5vyLgwR0jI92d3VJhfrJhWU=
basic.com/pubsub/protomsg.git v0.0.0-20190724081647-e1995f9a80e8 h1:hZYUcHcivGiofP3mGUWOOv6zKFAQp1V2uAkhjAXFfys=
basic.com/pubsub/protomsg.git v0.0.0-20190724081647-e1995f9a80e8/go.mod h1:un5NV5VWQoblVLZfx1Rt5vyLgwR0jI92d3VJhfrJhWU=
basic.com/pubsub/sdkcompare.git v0.0.0-20190715013640-f536a4647d00 h1:sK+Tx7rvM9J2WnNIwrzMDjZSylWiKNfQO0prUBfKsDk=
basic.com/pubsub/sdkcompare.git v0.0.0-20190715013640-f536a4647d00/go.mod h1:8by33F9E1w17Pw/rDgJGJXAo122w0wDENG14hiMS+RE=
basic.com/valib/deliver.git v0.0.0-20190531095353-25d8c3b20051 h1:9flC2o3kasaM2Y6I+mY+mxmve/pyAY/UzGQZLT3lFHM=

 insertdata/insertDataToEs.go

@@ -344,10 +344,26 @@
            logger.Error("查询摄像机信息失败")
        }
        serverIp, err := GetLocalIP()
        url := []string{}
        url = append(url,strings.Split(resp["fileUrl"].(string), "/")[1])
        if msg.RuleResult["cacheData"] != nil {
            InsertYolo(msg.RuleResult["cacheData"].(ruleserver.ResultMsg))
            // 把缓存的数据上传后得到地址存进去

            // 解压缩并上传图片
            msg1 := msg.RuleResult["cacheData"].(ruleserver.ResultMsg)
            bdata, err := util.UnCompress(msg1.Data)
            if err != nil {
                panic("解压缩图片时出现错误")
            }
            i := protomsg.Image{}
            err = proto.Unmarshal(bdata, &i)
            resp1, err1 := util.DrawPolygonOnImage(msg.Cid, i, msg.RuleResult["yolo"].([]ruleserver.Result), weedfsUrl)
            if err1 != nil {
                logger.Error("缓存数据画框或上传图片服务器出错", err)
            } else {
                logger.Info("上传的图片信息：", resp1)
            }
            url = append(url,strings.Split(resp1["fileUrl"].(string), "/")[1])
        }
        peraction := Personaction{
            uuid.NewV4().String(),
@@ -363,7 +379,7 @@
            localConfig.ServerName,
            serverIp,
            "",
            []string{strings.Split(resp["fileUrl"].(string), "/")[1]},
            url,
            i.Timestamp,
            "",
            isAlarm,

 labelFilter/readyDataForLabel.go

@@ -12,8 +12,8 @@
    CameraAddr    string
    TaskId        string
    Time          string
    Persons       []ruleserver.Arg
    DefenceLevel  string
    Persons       []ruleserver.FaceResult
    DefenceLevel  []int32
}
// 把数据装配到label
func (label *Label)DataFormatToLabel(result ruleserver.ResultMsg) {
@@ -25,8 +25,14 @@
    label.CameraName = camera.Name
    label.CameraAddr = camera.Addr
    label.TaskId = result.Tasklab.Taskid
    label.Persons = result.RuleResult["face"].([]ruleserver.Arg)
    //labelData.DefenceLevel = result.
    for _,faceGroup := range result.RuleResult["face"].([]ruleserver.FaceResult) {
        label.Persons = append(label.Persons,faceGroup)
        label.DefenceLevel = append(label.DefenceLevel,faceGroup.AlarmLevel)
    }
    for _,yoloGroup := range result.RuleResult["yolo"].([]ruleserver.FaceResult) {
        label.DefenceLevel = append(label.DefenceLevel,yoloGroup.AlarmLevel)
    }
    Judge(label,result)
}

// 从缓存中查出所有的规则数据

 labelFilter/ruleForLabel.go

@@ -1,27 +1,85 @@
package labelFilter

import (
    "basic.com/dbapi.git"
    "basic.com/pubsub/protomsg.git"
    "github.com/knetic/govaluate"
    "ruleprocess/logger"
    "ruleprocess/ruleserver"
    "strconv"
    "time"
)

func Judge(){
func Judge(label *Label, result ruleserver.ResultMsg){
    //拿到所有规则组

    var api dbapi.EventPushApi
    b,allRules := api.FindAllDetails()
    if !b {
        logger.Error("查询时间推送规则失败！")
    }
    //遍历所有的规则组，看看符合哪个规则组，就把数据插入到哪个规则组的目标机器上

    //先判断时间规则，不符合此规则，直接略过

        //遍历规则组中的小规则，每个小规则得出一个string类型的bool值，然后用 && 或 || 拼接起来

    for _,ruleGroup := range allRules {
        // 先过时间规则
        timeJudge(label,ruleGroup)
        result := ""
        var Connector string
        if ruleGroup.IsSatisfyAll {
            Connector = "&&"
        } else {
            Connector = "||"
        }
        for _,rule := range ruleGroup.Rules {
            // 过每一个小规则
            if rule.TopicType == "camera" {
                result = Connector + cameraJudge(label,rule)
            }
            if rule.TopicType == "baseLibrary" {
                result = Connector + baseJudge(label,rule)
            }
            if rule.TopicType == "task" {
                result = Connector + taskJudge(label,rule)
            }
            if rule.TopicType == "person" {
                result = Connector + taskJudge(label,rule)
            }
            if rule.TopicType == "defence" {
                result = Connector + defenceJudge(label,rule)
            }
        }
    //得出总的判断结果
        if result != "" {
            result = result[2:] // 把前两位字符去掉
            expression, err:= govaluate.NewEvaluableExpression(result);
            if err != nil {
                logger.Error("表达式有误，请检查！")
            }
            result, err1:= expression.Evaluate(nil)
            if err1 != nil {
                logger.Error("带入参数有误")
            }
            if result.(bool) {
                logger.Info("通过规则，表达式为：",result)
                // 推送服务器
                pushData(ruleGroup.IpPorts,result.(ruleserver.ResultMsg))
            } else {
                logger.Info("没通过规则,表达式为：",result)
            }
        }
    }

    //符合本规则组的就把数据推送到该目的服务器上
}
// 是否符合时间规则
func timeJudge(label *Label,groupRule *GroupRule) string{

func timeJudge(label *Label,groupRule protomsg.EventPush) string{
    timeNow := time.Now().Format("2006-01-02 15:04:05")
    flag1 := isBefore(groupRule.TimeStart,timeNow)
    flag2 := isBefore(timeNow,groupRule.TimeEnd)
    if flag1 && flag2 {
        return "true"
    } else {
        logger.Info("不在规定的时间范围内！")
        return "false"
    }
}
// 传入两个时间的字符串，判断第一个时间是否小于第二个时间
func isBefore(time1 string,time2 string) bool{
@@ -40,25 +98,95 @@
}

// 是否符合摄像机规则
func cameraJudge(label *Label,rule *Rule) string{
func cameraJudge(label *Label,rule *protomsg.EventPushRule) string{
    if rule.TopicArg == "cameraName" {
        if label.CameraName == rule.RuleValue {
            logger.Info("符合摄像机名称规则")
    return "true"
        } else {
            logger.Info("不符合摄像机名称规则")
            return "false"
        }
    }
    if rule.TopicArg == "cameraLocation" {
        if label.CameraAddr == rule.RuleValue {
            logger.Info("符合摄像机位置规则")
            return "true"
        } else {
            logger.Info("不符合摄像机位置规则")
            return "false"
        }
    }
    return ""
}

// 是否符合底库规则
func baseJudge(label *Label,rule *Rule) string{
    return "true"
func baseJudge(label *Label,rule *protomsg.EventPushRule) string{
    flag := "false"
    for _,personGroup := range label.Persons  {
        for _,arg := range personGroup.Args {
            for _,liker := range arg.Liker{
                if liker.TableId == rule.RuleValue {
                    logger.Info("符合底库规则！")
                    flag = "true"
                }
            }
        }
    }
    return flag
}

//是否符合人员规则
func personJudge(label *Label,rule *Rule) string{
func personJudge(label *Label,rule *protomsg.EventPushRule) string{
    for _,personGroup := range label.Persons  {
        for _,person := range personGroup.Args {
            if rule.TopicArg == "sex" && strconv.Itoa(int(person.ThftRes.Gender)) == rule.RuleValue{
                logger.Info("符合人员性别规则！")
    return "true"
            }
            if rule.TopicArg == "race" && strconv.Itoa(int(person.ThftRes.Race)) == rule.RuleValue {
                logger.Info("符合人员种族规则！")
                return "true"
            }
            if rule.TopicArg == "age" && strconv.Itoa(int(person.ThftRes.Age)) == rule.RuleValue {
                logger.Info("符合人员年龄规则！")
                return "true"
            }
            if rule.TopicArg == "others" {
                // 其他这个值先这么处理
                return "true"
            }
        }
        if strconv.Itoa(int(personGroup.AlarmLevel)) == rule.RuleValue {
            logger.Info("符合人员报警等级规则！")
            return "true"
        }
    }
    logger.Info("不符合人员规则")
    return "false"
}

// 是否符合布防等级规则
func defenceJudge(label *Label,rule *Rule) string{
func defenceJudge(label *Label,rule *protomsg.EventPushRule) string{
    for _,def := range label.DefenceLevel {
        if strconv.Itoa(int(def)) == rule.RuleValue {
            logger.Info("符合布防等级规则！")
    return "true"
}
    }
    logger.Info("不符合布防等级规则")
    return "false"
}
// 是否符合任务规则
func taskJudge(label *Label,rule *protomsg.EventPushRule) string{
    if label.TaskId == rule.RuleValue {
        logger.Info("符合任务规则!")
        return "true"
    }
    logger.Info("不符合任务规则")
    return "false"
}
// 调用目标服务器的插入接口
func pushData (data ruleserver.ResultMsg){
func pushData (urls []*protomsg.PushIpPort, data ruleserver.ResultMsg){

}

 ruleserver/ruleToformula.go

@@ -584,7 +584,7 @@
                timeLength, _ := strconv.Atoi(rule.SdkArgValue)
                timeEle := TimeElement{N: timeLength, InitN: timeLength, AlarmFlag: false, BufferFlag: 10, CacheSdkData: ResultMsg{message, args.RuleResult}} // 扔进去一个定时器元素（并缓存当前画面帧数据）
                //TimeEleList = make(map[string]timeElement)
                TimeEleList[groupId+" "+rule.Id] = &timeEle // 定时器元素以摄像机id拼接区域id为键
                TimeEleList[groupId+" "+rule.Id] = &timeEle // 定时器元素以组规则id和当前持续时间小规则id为键
                logger.Info("创建了计数器并且计数器集合为：", TimeEleList)
            }
        }

 ruleserver/timeTicker.go

@@ -45,7 +45,7 @@
                for k, timeEle := range TimeEleList {
                    if timeEle.N > 0 {
                        timeEle.N = timeEle.N - 1
                        logger.Error("-------------------------------------打印定时器元素当前值-----------------------------------------：", timeEle.N)
                        logger.Error("-------------------------------------打印定时器计数元素当前值-----------------------------------------：", timeEle.N)
                    }
                    if timeEle.GroupId != "" && timeEle.N == 0 {
                        // 说明是联动任务的时间窗口 到点儿了该销毁了,再来了再创建
@@ -66,7 +66,6 @@
    stopChan <- true
    TimeTicker()
}


// 判断是否符合定时器条件
func TimerAlarm(args *SdkDatas,groupId string,result bool) (string) {
@@ -134,6 +133,7 @@
    }
    return flagTime
}

// 结构体根据某字段排序
type SubList []*RuleResult