package cache
|
|
import (
|
"encoding/base64"
|
"errors"
|
"flag"
|
"fmt"
|
"strconv"
|
"strings"
|
"sync"
|
"time"
|
|
"sdkCompare/cache/shardmap"
|
"sdkCompare/config"
|
"sdkCompare/util"
|
|
libEs "basic.com/pubsub/esutil.git"
|
"basic.com/pubsub/protomsg.git"
|
"basic.com/valib/logger.git"
|
"github.com/golang/protobuf/proto"
|
)
|
|
var querynum = flag.Int("querynum", 3000, "the query number from database")
|
var threadnum = flag.Int("threadnum", 32, "the number of thread to deal data.")
|
|
var (
|
thresholdLimit = float32(50)
|
captureTable = "capturetable"
|
cartable = "carTable_"
|
PRE_DBTABLE = "dbTable_"
|
PRE_CAPTURE_SERVER = "captureServer_"
|
)
|
|
type CmapItem struct {
|
sync.Mutex
|
Cam map[string]*shardmap.ShardMap
|
}
|
|
var Cmap *CmapItem
|
var doOnce sync.Once
|
|
// 计算每次查询的数据量条数
|
func Querynum(totalnum int) int {
|
qn := totalnum / *threadnum //qn=6551
|
if *querynum < qn { //
|
|
return *querynum
|
}
|
return qn
|
}
|
|
// 增量查询
|
func IncreVideoPersonsCache(lastT time.Time, targetType string) {
|
ticker := time.NewTicker(time.Minute * 3)
|
for {
|
select {
|
case <-ticker.C:
|
curTime := time.Now()
|
Incrementquery(lastT, curTime, targetType)
|
lastT = curTime
|
}
|
}
|
}
|
|
func Incrementquery(last time.Time, cur time.Time, targetType string) {
|
laststring := last.Format("2006-01-02 15:04:05")
|
curstring := cur.Format("2006-01-02 15:04:05")
|
alarmIp := config.EsCompServerInfo.ESIP
|
alarmPort := config.EsCompServerInfo.ESPort
|
indexName := config.EsInfo.EsIndex.AiOcean.IndexName
|
|
serverIp, _, e := util.GetLocalIP(config.Server.NetworkAdapter)
|
if e != nil {
|
return
|
}
|
|
shardStr, e := getShards(serverIp, alarmIp, alarmPort)
|
if e != nil {
|
return
|
}
|
|
// 调用增量的接口
|
captures, err := libEs.GetPeriodInfos(serverIp, alarmPort, laststring, curstring, indexName, shardStr, targetType)
|
if err != nil {
|
logger.Error(err)
|
return
|
}
|
for _, ei := range captures {
|
if ei.EsInfo.AnalyServerId != "" {
|
cKey := PRE_CAPTURE_SERVER + ei.EsInfo.AnalyServerId
|
ei.EsInfo.Tableid = captureTable
|
|
if _, ok := Cmap.Cam[cKey]; !ok {
|
Cmap.Cam[cKey] = shardmap.New(uint8(*threadnum))
|
}
|
|
Cmap.Cam[cKey].Set(ei.EsInfo.Id, ei.EsInfo)
|
}
|
}
|
}
|
|
func getShards(serverIp string, alarmIp string, alarmPort string) (string, error) {
|
infos, e := libEs.GetShardsByIndex(alarmIp, alarmPort, config.EsInfo.EsIndex.AiOcean.IndexName)
|
if e != nil || infos == nil {
|
logger.Error("get es primary ips err")
|
return "", errors.New("get es primary ips err")
|
}
|
|
var shards []string
|
for _, shard := range infos {
|
if (shard.ShardIp == serverIp || shard.ShardIp == "127.0.0.1") && shard.ShardRole == "primary" && shard.ShardState == "STARTED" {
|
shards = append(shards, strconv.Itoa(shard.ShardNum))
|
}
|
}
|
if len(shards) == 0 {
|
return "", errors.New("current shards is empty")
|
}
|
return strings.Join(shards, ","), nil
|
}
|
|
// 1. 拿到总量， 计算每个线程的查询量
|
// 2. 分线程查询
|
func Init(indexName string, targetType string) error {
|
alarmIp := config.EsCompServerInfo.ESIP
|
alarmPort := config.EsCompServerInfo.ESPort
|
|
doOnce.Do(func() {
|
flag.Parse()
|
|
Cmap = &CmapItem{
|
Cam: make(map[string]*shardmap.ShardMap),
|
}
|
})
|
|
serverIp, _, e := util.GetLocalIP(config.Server.NetworkAdapter)
|
if e != nil {
|
return e
|
}
|
|
shardStr, e := getShards(serverIp, alarmIp, alarmPort)
|
if e != nil {
|
return e
|
}
|
// 这里需要提供总量的接口
|
estotalnum := libEs.GetTotal(serverIp, alarmPort, indexName, shardStr, targetType)
|
*querynum = Querynum(estotalnum)
|
temptime := time.Now()
|
|
captures, err := libEs.GetOceanFeatures(serverIp, alarmPort, 1000, indexName, shardStr, targetType)
|
if err != nil {
|
logger.Error("libEs.GetOceanFeatures err:", err)
|
return err
|
}
|
logger.Debug("len(captures):", len(captures))
|
Cmap.Lock()
|
for _, ei := range captures {
|
if ei.EsInfo.AnalyServerId != "" {
|
cKey := PRE_CAPTURE_SERVER + ei.EsInfo.AnalyServerId
|
ei.EsInfo.Tableid = captureTable
|
if _, ok := Cmap.Cam[cKey]; !ok {
|
Cmap.Cam[cKey] = shardmap.New(uint8(*threadnum))
|
}
|
|
Cmap.Cam[cKey].Set(ei.EsInfo.Id, ei.EsInfo)
|
}
|
}
|
|
Cmap.Unlock()
|
|
logger.Debug(indexName, "缓存完成用时:", time.Since(temptime))
|
return nil
|
}
|
|
func ReInitDbTablePersonsCache() {
|
Cmap.Lock()
|
defer Cmap.Unlock()
|
if Cmap == nil {
|
Cmap = &CmapItem{
|
Cam: make(map[string]*shardmap.ShardMap),
|
}
|
}
|
for tableId, _ := range Cmap.Cam {
|
delete(Cmap.Cam, tableId)
|
}
|
|
initDbTablePersonsCache()
|
}
|
|
func InitDbTablePersons() {
|
doOnce.Do(func() {
|
flag.Parse()
|
|
Cmap = &CmapItem{
|
Cam: make(map[string]*shardmap.ShardMap),
|
}
|
})
|
|
initDbTablePersonsCache()
|
}
|
|
func initDbTablePersonsCache() {
|
// 查询所有的底库列表
|
var dtApi DbTables
|
allTables, err := dtApi.FindAllDbTablesByCurServer()
|
if err == nil && allTables != nil && len(allTables) > 0 { //初始化底库缓存信息
|
Cmap.Lock()
|
for _, table := range allTables {
|
if _, ok := Cmap.Cam[PRE_DBTABLE]; !ok {
|
Cmap.Cam[PRE_DBTABLE] = shardmap.New(uint8(*threadnum))
|
}
|
Cmap.Cam[PRE_DBTABLE].Set(table.Id, table.Enable == 1)
|
logger.Debugf("初始化底库[%s][%s]到缓存中", table.Id, table.TableName)
|
}
|
Cmap.Unlock()
|
}
|
|
// 缓存底库中的全部人员信息
|
var dbpApi DbPersons
|
total, e := dbpApi.GetPersonTotal("")
|
logger.Debugf("所有底库共有%d条记录", total)
|
if e == nil && total > 0 {
|
queryEachNum := *querynum
|
qn := int(total) / *threadnum
|
if *querynum < qn {
|
queryEachNum = qn
|
}
|
queryT := int(total) / queryEachNum
|
if int(total)%queryEachNum > 0 {
|
queryT++
|
}
|
temptime := time.Now()
|
var wg sync.WaitGroup
|
|
for i := 0; i < queryT; i++ {
|
j := i * queryEachNum
|
wg.Add(1)
|
go func(qs int) {
|
defer wg.Done()
|
dbpersons, err := dbpApi.GetPersonsCompareCacheBase(j, queryEachNum)
|
if err != nil {
|
logger.Error(err)
|
return
|
}
|
logger.Debugf("获取%d条人员信息", len(dbpersons))
|
Cmap.Lock()
|
tableId := ""
|
for _, value := range dbpersons {
|
tableId = value.Tableid
|
if _, ok := Cmap.Cam[value.Tableid]; !ok {
|
Cmap.Cam[value.Tableid] = shardmap.New(uint8(*threadnum))
|
}
|
|
Cmap.Cam[value.Tableid].Set(value.Id, value)
|
}
|
|
if len(dbpersons) != 0 {
|
Cmap.Cam[tableId].Settime()
|
}
|
|
Cmap.Unlock()
|
|
}(j)
|
}
|
wg.Wait()
|
logger.Debug("底库人员缓存完成用时:", time.Since(temptime))
|
}
|
}
|
|
// UpdateDbPersonsCacheById 更新缓存中的全部人员信息
|
func UpdateDbPersonsCacheById(id string) {
|
var dbpApi DbPersons
|
info, err := dbpApi.GetPersonsCompareCacheById(id)
|
if err != nil {
|
logger.Error(err)
|
return
|
}
|
if info.Tableid != "" {
|
Cmap.Lock()
|
defer Cmap.Unlock()
|
if _, ok := Cmap.Cam[info.Tableid]; !ok {
|
Cmap.Cam[info.Tableid] = shardmap.New(uint8(*threadnum))
|
}
|
Cmap.Cam[info.Tableid].Set(info.Id, info)
|
Cmap.Cam[info.Tableid].Settime()
|
}
|
}
|
|
func RealTimeAddPersonInfoToCache(tableId string, id string, faceFeature string, enable int32, carNo string) {
|
Cmap.Lock()
|
defer Cmap.Unlock()
|
if _, ok := Cmap.Cam[tableId]; !ok {
|
Cmap.Cam[tableId] = shardmap.New(uint8(*threadnum))
|
}
|
var ei = protomsg.Esinfo{
|
Id: id,
|
Tableid: tableId,
|
FaceFeature: faceFeature,
|
Enable: enable,
|
CarNo: carNo,
|
}
|
Cmap.Cam[tableId].Set(id, &ei)
|
logger.Debug("id:", id, ",tableId:", ",len(faceFeature):", len(faceFeature), ",tableId:", tableId, ",enable:", enable)
|
}
|
|
func RealTimeDelPersonFromCache(tableId string, id string) {
|
logger.Debug("DelPersonFromCache,tableId:", tableId, ",id:", id)
|
Cmap.Lock()
|
defer Cmap.Unlock()
|
if _, ok := Cmap.Cam[tableId]; ok {
|
Cmap.Cam[tableId].Del(id)
|
logger.Debug("DelPerson ok success")
|
} else {
|
logger.Error("tableId:", tableId, " not exist")
|
}
|
}
|
|
func RealTimeDelTable(tableId string) {
|
logger.Debug("RealTimeDelTable tableId:", tableId)
|
Cmap.Lock()
|
defer Cmap.Unlock()
|
|
if dtM, ok := Cmap.Cam[PRE_DBTABLE]; ok {
|
dtM.Del(tableId)
|
}
|
if _, ok := Cmap.Cam[tableId]; ok {
|
delete(Cmap.Cam, tableId)
|
}
|
}
|
|
// 使底库生效,将底库中的所有生效状态的人特征添加到缓存
|
func RealTimeUpdateTable(tableId string, enable int32) {
|
logger.Debug("RealTimeUpdateTable tableId:", tableId, ",enable:", enable)
|
Cmap.Lock()
|
defer Cmap.Unlock()
|
|
if _, ok := Cmap.Cam[PRE_DBTABLE]; !ok {
|
Cmap.Cam[PRE_DBTABLE] = shardmap.New(uint8(*threadnum))
|
}
|
Cmap.Cam[PRE_DBTABLE].Set(tableId, enable == 1)
|
}
|
|
func UpdateCache(changeMsg *protomsg.EsPersonCacheChange) {
|
if changeMsg.Type == protomsg.EsCacheChanged_T_DbTable {
|
if changeMsg.Action == protomsg.DbAction_Insert || changeMsg.Action == protomsg.DbAction_Update {
|
RealTimeUpdateTable(changeMsg.TableId[0], changeMsg.Enable)
|
} else if changeMsg.Action == protomsg.DbAction_Delete {
|
RealTimeDelTable(changeMsg.TableId[0])
|
}
|
} else if changeMsg.Type == protomsg.EsCacheChanged_T_DbTablePerson {
|
if changeMsg.Action == protomsg.DbAction_Insert || changeMsg.Action == protomsg.DbAction_Update {
|
RealTimeAddPersonInfoToCache(changeMsg.TableId[0], changeMsg.PersonId, changeMsg.Feature, changeMsg.Enable, changeMsg.CarNo)
|
} else if changeMsg.Action == protomsg.DbAction_Delete {
|
RealTimeDelPersonFromCache(changeMsg.TableId[0], changeMsg.PersonId)
|
}
|
}
|
}
|
|
func GetComparePersonBaseInfo(compareArgs protomsg.CompareArgs) []byte {
|
if compareArgs.FaceFeature == nil {
|
return nil
|
}
|
|
//指定最低分
|
baseScore := thresholdLimit
|
if compareArgs.CompareThreshold > thresholdLimit {
|
baseScore = compareArgs.CompareThreshold
|
}
|
|
if compareArgs.IsCompareAll {
|
baseScore = 0
|
}
|
|
var scResult protomsg.SdkCompareResult
|
|
//未指定比对目标map
|
if compareArgs.TableIds == nil || len(compareArgs.TableIds) == 0 {
|
logger.Debugf("接收到底库比对请求, 阈值:%f", compareArgs.CompareThreshold)
|
|
// 比对来源是ruleprocess，传空比全部底库
|
if !compareArgs.Source {
|
for key, val := range Cmap.Cam {
|
// 判断是否属于底库
|
if tShard, hasT := Cmap.Cam[PRE_DBTABLE]; hasT {
|
if tEnable, tOk := tShard.Get(key); tOk { //存在此底库
|
//底库有效
|
if tEnable.(bool) {
|
targets := val.Walk(DoSdkCompare, compareArgs.FaceFeature, baseScore, compareArgs.Source, compareArgs.CompareTarget)
|
if len(targets) > 0 {
|
scResult.CompareResult = append(scResult.CompareResult, targets...)
|
}
|
}
|
}
|
}
|
}
|
} else { //来源是web，比对所有抓拍和底库
|
for cKey, val := range Cmap.Cam {
|
if compareArgs.AnalyServerId != "" { //比对指定server产生的抓拍数据和底库
|
if cKey == PRE_CAPTURE_SERVER+compareArgs.AnalyServerId || !strings.HasPrefix(cKey, PRE_CAPTURE_SERVER) {
|
targets := val.Walk(DoSdkCompare, compareArgs.FaceFeature, baseScore, compareArgs.Source, compareArgs.CompareTarget)
|
if len(targets) > 0 {
|
scResult.CompareResult = append(scResult.CompareResult, targets...)
|
}
|
}
|
} else { //管理平台请求，比对所有数据
|
if len(compareArgs.ServerIds) > 0 {
|
for _, termDevId := range compareArgs.ServerIds {
|
if cKey == PRE_CAPTURE_SERVER+termDevId || !strings.HasPrefix(cKey, PRE_CAPTURE_SERVER) {
|
targets := val.Walk(DoSdkCompare, compareArgs.FaceFeature, baseScore, compareArgs.Source, compareArgs.CompareTarget)
|
if len(targets) > 0 {
|
scResult.CompareResult = append(scResult.CompareResult, targets...)
|
}
|
}
|
}
|
} else {
|
targets := val.Walk(DoSdkCompare, compareArgs.FaceFeature, baseScore, compareArgs.Source, compareArgs.CompareTarget)
|
if len(targets) > 0 {
|
scResult.CompareResult = append(scResult.CompareResult, targets...)
|
}
|
}
|
}
|
}
|
}
|
} else { //指定比对目标map
|
if !compareArgs.Source {
|
for _, tid := range compareArgs.TableIds { //ruleProcess比对指定底库
|
shardins, ok := Cmap.Cam[tid]
|
if !ok {
|
logger.Error("ruleProcess compare get shard error by tableId:", tid)
|
continue
|
}
|
if tShard, hasT := Cmap.Cam[PRE_DBTABLE]; hasT {
|
if tEnable, tOk := tShard.Get(tid); tOk { //存在此底库
|
logger.Debug("ruleProcess compare tables,exist tableId:", tid, ",enable:", tEnable)
|
if tEnable.(bool) { //底库有效
|
targets := shardins.Walk(DoSdkCompare, compareArgs.FaceFeature, baseScore, compareArgs.Source, compareArgs.CompareTarget)
|
if len(targets) > 0 {
|
scResult.CompareResult = append(scResult.CompareResult, targets...)
|
}
|
}
|
} else {
|
logger.Error("ruleProcess compare tables,tShard not exist tableId:", tid)
|
}
|
} else {
|
logger.Error("ruleProcess compare tables,PRE_DBTABLE tableId:", tid, " not exist")
|
}
|
}
|
} else { //web请求，比对指定的抓拍库或者底库
|
for _, tid := range compareArgs.TableIds {
|
if tid == captureTable { //比对抓拍库
|
tStart := time.Now()
|
serverIp, _, e := util.GetLocalIP(config.Server.NetworkAdapter)
|
if e != nil {
|
logger.Error("util.GetLocalIP err:", e)
|
continue
|
}
|
alarmPort := config.EsCompServerInfo.ESPort
|
|
scopeIds := libEs.GetAllLocalVideopersonsId(compareArgs, config.EsInfo.EsIndex.AiOcean.IndexName, serverIp, alarmPort, compareArgs.AlarmLevel)
|
logger.Debug("libEs.GetAllLocalVideopersonsId len(scopeIds):", len(scopeIds), " 耗时：", time.Since(tStart))
|
if scopeIds != nil {
|
tCompStart := time.Now()
|
if compareArgs.AnalyServerId != "" { //指定server
|
for cKey, cacheMap := range Cmap.Cam {
|
if cKey == PRE_CAPTURE_SERVER+compareArgs.AnalyServerId {
|
for _, sId := range scopeIds {
|
obj, ok := cacheMap.Get(sId)
|
if ok {
|
eInfo, ok := obj.(*protomsg.Esinfo)
|
if !ok {
|
continue
|
}
|
sec := DoSdkCompare(compareArgs.FaceFeature, eInfo.FaceFeature)
|
if sec >= baseScore {
|
scResult.CompareResult = append(scResult.CompareResult, &protomsg.SdkCompareEach{
|
Id: eInfo.Id,
|
CompareScore: sec,
|
Tableid: eInfo.Tableid,
|
})
|
}
|
}
|
}
|
}
|
}
|
} else { //管理平台请求比对所有抓拍,不指定server
|
for cKey, cacheMap := range Cmap.Cam {
|
if len(compareArgs.ServerIds) > 0 {
|
for _, termDevId := range compareArgs.ServerIds {
|
if cKey == PRE_CAPTURE_SERVER+termDevId || !strings.HasPrefix(cKey, PRE_CAPTURE_SERVER) {
|
targets := cacheMap.Walk(DoSdkCompare, compareArgs.FaceFeature, baseScore, compareArgs.Source, compareArgs.CompareTarget)
|
if len(targets) > 0 {
|
scResult.CompareResult = append(scResult.CompareResult, targets...)
|
}
|
}
|
}
|
} else {
|
if strings.HasPrefix(cKey, PRE_CAPTURE_SERVER) {
|
for _, sId := range scopeIds {
|
if obj, ok := cacheMap.Get(sId); ok {
|
eInfo, ok := obj.(*protomsg.Esinfo)
|
if !ok {
|
continue
|
}
|
sec := DoSdkCompare(compareArgs.FaceFeature, eInfo.FaceFeature)
|
if sec >= baseScore {
|
scResult.CompareResult = append(scResult.CompareResult, &protomsg.SdkCompareEach{
|
Id: eInfo.Id,
|
CompareScore: sec,
|
Tableid: eInfo.Tableid,
|
})
|
}
|
}
|
}
|
}
|
}
|
}
|
}
|
logger.Debug("根据scopeIds比对耗时：", time.Since(tCompStart))
|
}
|
} else {
|
shardins, ok := Cmap.Cam[tid]
|
if !ok {
|
logger.Error("get shard error by tableId:", tid)
|
continue
|
}
|
|
targets := shardins.Walk(DoSdkCompare, compareArgs.FaceFeature, baseScore, compareArgs.Source, compareArgs.CompareTarget)
|
if len(targets) > 0 {
|
scResult.CompareResult = append(scResult.CompareResult, targets...)
|
}
|
}
|
}
|
}
|
}
|
|
logger.Debug("比对结果: len(scResult.CompareResult):", len(scResult.CompareResult))
|
buf, err := proto.Marshal(&scResult)
|
|
if err != nil {
|
logger.Error("scResult Marshal error!", err)
|
return nil
|
}
|
|
return buf
|
}
|
|
func DoSdkCompare(ci []byte, co string) float32 {
|
|
co_d, err := base64.StdEncoding.DecodeString(co)
|
if err != nil {
|
logger.Error("DoSdkCompare err:", err)
|
return -1
|
}
|
if len(co_d) != 2560 {
|
logger.Error("target fea.len !=2560")
|
return -1
|
}
|
|
if len(ci) != 2560 {
|
logger.Error("source fea.len !=2560")
|
return -1
|
}
|
sec := DecCompare(ci, co_d)
|
//logger.Debug("比对得分为：", sec)
|
|
sec = ParseScore(sec)
|
return sec
|
}
|
|
func ParseScore(compareScore float32) float32 {
|
if compareScore <= 1 {
|
compareScore = compareScore * 100
|
}
|
if compareScore == 100 {
|
return 100
|
}
|
f, _ := strconv.ParseFloat(fmt.Sprintf("%2.2f", compareScore), 32)
|
|
return float32(f)
|
}
|
