长场景人员分析模型

zhaoqingang

2025-02-08 3c033759200ad7c02dd59521b1aebbbdc35b98fa

 models/gather_model.go

@@ -6,28 +6,31 @@
   "encoding/json"
   "errors"
   "fmt"
   "github.com/elastic/go-elasticsearch/v6"
   "log"
   "model-engine/config"
   "model-engine/db"
   "model-engine/pkg/set"
   "model-engine/service"
   "strings"
   "sync"
   "time"

   "github.com/elastic/go-elasticsearch/v6"

   "model-engine/config"
   "model-engine/db"
   "model-engine/pkg/logger"
   "model-engine/pkg/set"
   "model-engine/service"
)

type GatherModel struct {
   OrgIds         []interface{} `json:"-"`
   AreaIds        []interface{} `json:"-"`
   Building       string        `gorm:"type:varchar(255)" json:"building"`    //楼栋
   Floor          string        `gorm:"type:varchar(255)" json:"floor"`       //楼层
   AlarmType      db.AlarmType  `gorm:"type:varchar(255);" json:"alarmType"`  //预警方式
   PersonType     string        `gorm:"type:varchar(255);" json:"personType"` //人员类型
   GatherPersons  int           `gorm:"type:int;" json:"gatherPersons"`       //聚集人数
   AppearInterval int           `gorm:"type:int;" json:"appearInterval"`      //出现间隔，单位为秒
   DaysWindow     int           `gorm:"type:int;" json:"daysWindow" `         //近几天内
   Threshold      int           `gorm:"type:int;" json:"threshold" `          //达几次
   Building       string        `gorm:"type:varchar(255)" json:"building"`    // 楼栋
   Floor          string        `gorm:"type:varchar(255)" json:"floor"`       // 楼层
   AlarmType      db.AlarmType  `gorm:"type:varchar(255);" json:"alarmType"`  // 预警方式
   PersonType     string        `gorm:"type:varchar(255);" json:"personType"` // 人员类型
   GatherPersons  int           `gorm:"type:int;" json:"gatherPersons"`       // 聚集人数
   AppearInterval int           `gorm:"type:int;" json:"appearInterval"`      // ，单位为秒
   DaysWindow     int           `gorm:"type:int;" json:"daysWindow" `         // 近几天内
   Threshold      int           `gorm:"type:int;" json:"threshold" `          // 达几次
   Task           *db.ModelTask
}

@@ -52,11 +55,40 @@
   m.Floor = task.Floor
   m.AlarmType = task.AlarmType
   m.PersonType = task.PersonType
   m.GatherPersons = task.GatherPersons
   m.AppearInterval = task.AppearInterval
   m.DaysWindow = task.DaysWindow
   m.Threshold = task.Threshold
   fmt.Println("GatherModel init finish ...")

   for _, v := range task.Rules {
      if v.Alias == "gatherPersons" {
         if val, ok := v.Value.(float64); ok {
            m.GatherPersons = int(val)
         }
      }

      if v.Alias == "appearInterval" {
         if val, ok := v.Value.(float64); ok {
            m.AppearInterval = int(val)
         }
      }

      if v.Alias == "daysWindow" {
         if val, ok := v.Value.(float64); ok {
            m.DaysWindow = int(val)
         }
      }

      if v.Alias == "threshold" {
         if val, ok := v.Value.(float64); ok {
            m.Threshold = int(val)
         }
      }
   }

   logger.Debugf("GatherModel init finish ...task id:%s, name:%s, rule:%+v\n", task.ID, task.Name, m)

   if m.GatherPersons == 0 || m.AppearInterval == 0 || m.DaysWindow == 0 || m.Threshold == 0 {
      logger.Warnf("invalid parameters. task id:%s, name:%s\n", task.ID, task.Name)
      return errors.New("invalid parameters")
   }

   return nil
}

@@ -68,8 +100,8 @@
   OrgId          string `json:"orgId"`
   Building       string `json:"building"`
   Floor          string `json:"floor"`
   GatherPersons  int    `gorm:"type:int;" json:"gatherPersons"`  //聚集人数
   AppearInterval int    `gorm:"type:int;" json:"appearInterval"` //出现间隔，单位为秒
   GatherPersons  int    `gorm:"type:int;" json:"gatherPersons"`  // 聚集人数
   AppearInterval int    `gorm:"type:int;" json:"appearInterval"` // 出现间隔，单位为秒
}

var (
@@ -114,6 +146,9 @@
      })
      newRecords = append(newRecords, record)
   }
   if len(newRecords) == 0 {
      return nil
   }

   aggregation, err := analyzeAndAggregate(newRecords)
   if err != nil {
@@ -153,6 +188,11 @@
      results = append(results, result)
   }
   return service.SaveTaskResults(results)
}

func (m *GatherModel) KeepAlive() error {
   db.GetDB().Model(m.Task).Where("id = ?", m.Task.ID).Update("last_run_time", time.Now())
   return nil
}

func (m *GatherModel) Shutdown() error {
@@ -402,6 +442,9 @@
}

func analyzeAndAggregate(records []*GatherRecord) (map[GatherLocationTime]set.StringSet, error) {
   if len(records) == 0 {
      return nil, nil
   }
   aggregation := make(map[GatherLocationTime]set.StringSet)
   domainIds := set.NewStringSet()
   for _, record := range records {