db/db.go

@@ -82,6 +82,14 @@
            Description: "通用场所分析模型",
            Version:     "v1.0.0",
            Enabled:     false,
        }, {
            BaseModel: BaseModel{
                ID: ModelIdAccessRegularity,
            },
            Name:        "出行规律分析",
            Description: "出行规律分析模型",
            Version:     "v1.0.0",
            Enabled:     false,
        },
        //{
        //    BaseModel: BaseModel{
@@ -278,6 +286,76 @@
                Sort:     1,
            },
        },

        // 出行规律
        {
            Id:      "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
            ModelId: ModelIdAccessRegularity,
            Scope:   "",
            RuleArg: RuleArg{
                Alias:    "duration",
                Name:     "监控时间段",
                Type:     "input",
                Must:     true,
                Unit:     "天内",
                Range:    "1,60",
                Value:    "30",
                ValType:  "int",
                Operator: ">=",
                Sort:     0,
            },
        },
        {
            Id:      "f47ac10b-58cc-4372-a567-0e02b2c3d479",
            ModelId: ModelIdAccessRegularity,
            Scope:   "",
            RuleArg: RuleArg{
                Alias:    "timeRange",
                Name:     "出行时间段",
                Type:     "range",
                Must:     true,
                Unit:     "点",
                Range:    "1,24",
                Value:    "21,3",
                ValType:  "int",
                Operator: ">=",
                Sort:     1,
            },
        },
        {
            Id:      "c9bf9e57-1685-4c89-bafb-ff5af830be8a",
            ModelId: ModelIdAccessRegularity,
            Scope:   "",
            RuleArg: RuleArg{
                Alias:    "appearances",
                Name:     "出行次数",
                Type:     "input",
                Must:     false,
                Unit:     "次",
                Range:    "1,100",
                Value:    "2",
                ValType:  "int",
                Operator: "==",
                Sort:     2,
            },
        },
        {
            Id:      "e4eaaaf2-d142-11e1-b3e4-080027620cdd",
            ModelId: ModelIdAccessRegularity,
            Scope:   "",
            RuleArg: RuleArg{
                Alias:    "direction",
                Name:     "进出方向",
                Type:     "select",
                Must:     false,
                Unit:     "",
                Range:    "",
                Value:    "",
                ValType:  "[{\"label\":\"进\", \"value\":\"in\"},{\"label\":\"出\", \"value\":\"out\"}]",
                Operator: "==",
                Sort:     3,
            },
        },
    }

    for i := range rules {

 db/model.go

@@ -187,5 +187,5 @@
    ModelIdGather           = "gather"           // 聚集
    ModelIdDisappear        = "disappear"        // 失踪
    ModelIdLocationAnalysis = "locationAnalysis" // 场所分析
    //ModelIdNightAnalysis    = "nightAnalysis"    // 昼伏夜出分析
    ModelIdAccessRegularity = "accessRegularity" // 出行规律分析
)

 models/accessRegularity.go

New file
@@ -0,0 +1,177 @@
package models

import (
    "errors"
    "fmt"
    "strconv"
    "strings"
    "time"

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

type RegularityModel struct {
    OrgIds        []interface{} `json:"-"`
    AreaIds       []interface{} `json:"-"`
    IdentityType  []string      // 人员身份类型 陌生人, 访客, 住户
    KeyPersonType string        // 重点人员类型
    PersonLabel   string        // 人员身份标签
    Task          *db.ModelTask
    AlarmType     db.AlarmType // 预警方式

    Appearances   int    // 出现次数
    Duration      int    // 时间范围, 单位天
    LastDirection string // 最后一次出行的方向
    StartHour     int    // 开始计算的时间 配置为小时 23 - 02 表示第一天23点 - 第二天的2点
    EndHour       int    // 结束时间
}

func (m *RegularityModel) Init(task *db.ModelTask) error {
    if len(task.DomainUnitIds) == 0 {
        return errors.New("empty domain set")
    }
    orgIds, areaIds, err := service.GetOrgIdsAndAreaIdsByDomainUnitIds(task.DomainUnitIds)
    if err != nil {
        return err
    }

    m.Task = task
    m.OrgIds = orgIds
    m.AreaIds = areaIds
    m.AlarmType = task.AlarmType
    m.KeyPersonType = task.PersonType
    m.PersonLabel = task.PersonLabel
    m.LastDirection = "out"
    if task.IdentityType != "" {
        for _, t := range strings.Split(task.IdentityType, ",") {
            if t == "all" {
                m.IdentityType = []string{"stranger", "visitor", "resident"}
                break
            } else {
                m.IdentityType = append(m.IdentityType, t)
            }
        }
    } else {
        m.IdentityType = []string{"stranger", "visitor", "resident"}
    }

    for _, v := range task.Rules {
        if v.Alias == "timeRange" {
            if val, ok := v.Value.(string); ok {
                ages := strings.Split(val, ",")
                m.StartHour, _ = strconv.Atoi(ages[0])
                m.EndHour, _ = strconv.Atoi(ages[1])
            }
        }

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

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

        if v.Alias == "direction" {
            if val, ok := v.Value.(string); ok {
                m.LastDirection = val
            }
        }
    }

    // 默认计算30天的数据
    if m.Duration == 0 {
        m.Duration = 30
    }

    if m.StartHour == 0 || m.EndHour == 0 {
        return fmt.Errorf("task id:%s, %s timeRange Time range setting error. %+v", task.ID, task.Name, task.Rules)
    }

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

    return nil
}

func (m *RegularityModel) Run() error {
    results := make([]*db.ModelTaskResults, 0)
    baseFilter := make([]PersonInfo, 0)

    // 查找指定时间范围内出行过的档案编号
    now := time.Now()
    startDate := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()).AddDate(0, 0, -m.Duration)
    err := db.GetDB().Raw(`
        SELECT
            document_number,
            frequent_address,
            community_id,
        FROM
            snapshot_count_summary
        WHERE
            last_appearance_time > ? "" 
            AND (p.community_id IN ?
            OR p.org_id IN ?)
            AND p.status IN ?
        `, startDate.Unix(), m.AreaIds, m.OrgIds, m.IdentityType).Scan(&baseFilter).Error
    if err != nil {
        logger.Warnf(err.Error())
    }

    esCli := db.GetEsClient()
    for _, p := range baseFilter {
        // 调用es分析此人的出行规律是否符合条件， 返回符合条件的次数和最后一次符合条件的时间
        sTime := startDate.Format(time.DateTime)
        eTime := time.Now().Format(time.DateTime)
        captures, err := service.QueryEsRecord(esCli, sTime, eTime, nil, []interface{}{p.CommunityId}, []string{p.DocumentNumber})
        if len(captures) == 0 || err != nil {
            continue
        }

        // 根据抓拍时间和出入方向，计算符合规则内的出入次数
        hitCount := 0

        //for _, c := range captures {
        //    captureTime, err := time.ParseInLocation(time.DateTime, c.PicDate, time.Local)
        //}

        // 写数据库
        result := &db.ModelTaskResults{
            Title:         m.Task.Name,
            Event:         fmt.Sprintf("%s %d次", m.Task.Name, hitCount),
            ModelID:       m.Task.ModelID,
            ModelTaskID:   m.Task.ID,
            CommunityId:   p.CommunityId,
            OrgID:         p.OrgId,
            ObjectIds:     p.DocumentNumber,
            Location:      p.FrequentAddress,
            PicDate:       time.Unix(p.LastAppearanceTime, 0).Format(time.DateTime),
            FirstPersonID: p.DocumentNumber,
        }

        results = append(results, result)
    }

    logger.Debugf("task %s last filter result %d", m.Task.Name, len(results))
    return service.SaveTaskResults(results)
}

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

func (m *RegularityModel) Shutdown() error {
    // 清理资源
    fmt.Println("Shutting down LocationModel Model")
    return nil
}

func (m *RegularityModel) eventFormat() string {
    return ""
}

 models/common.go

@@ -1,5 +1,17 @@
package models

type PersonInfo struct {
    DocumentNumber     string `json:"document_number"`
    CommunityId        string `json:"community_id"`
    OrgId              string `json:"org_id"`
    PersonName         string `json:"person_name"`
    IdCard             string `json:"id_card"`
    FrequentAddress    string `json:"frequent_address"`
    LastAppearanceTime int64  `json:"last_appearance_time"`
    LastDirection      string `json:"last_direction"`
    LastLocation       string `json:"last_location"`
}

func GetDomainFilters(orgIds, areaIds []interface{}) (filters []map[string]interface{}) {
    filters = make([]map[string]interface{}, 0)
    // 数据权限过滤

 models/disappear.go

@@ -80,17 +80,6 @@
    return nil
}

type PersonInfo struct {
    DocumentNumber     string `json:"document_number"`
    CommunityId        string `json:"community_id"`
    OrgId              string `json:"org_id"`
    PersonName         string `json:"person_name"`
    IdCard             string `json:"id_card"`
    LastAppearanceTime int64  `json:"last_appearance_time"`
    LastDirection      string `json:"last_direction"`
    LastLocation       string `json:"last_location"`
}

func (m *DisappearModel) Run() error {
    results := make([]*db.ModelTaskResults, 0)
    var ageFilter, labelFilter, keyFilter, lastFilter []PersonInfo

 models/gather_model.go

@@ -185,6 +185,7 @@
            PicDate:       lt.Time,
            FirstPersonID: personIds[0],
        }

        results = append(results, result)
    }
    return service.SaveTaskResults(results)

 service/esSearch.go

New file
@@ -0,0 +1,200 @@
package service

import (
    "bytes"
    "context"
    "encoding/json"
    "fmt"
    "github.com/elastic/go-elasticsearch/v6"
    "model-engine/config"
)

type CameraLocation struct {
    Building  string `json:"building"`
    Unit      string `json:"unit"`
    Pos       string `json:"pos"`
    Floor     string `json:"floor"`
    Direction string `json:"direction"`
    Latitude  string `json:"latitude"`
    Longitude string `json:"longitude"`
}

func (cl *CameraLocation) Join() string {
    return cl.Building + cl.Unit + cl.Pos + cl.Floor
}

type ESRecordInfo struct {
    Id             string         `json:"id"`
    ClusterId      string         `json:"clusterId"`
    CameraId       string         `json:"cameraId"`
    CameraAddr     string         `json:"cameraAddr"`     // 摄像机地址
    CameraName     string         `json:"cameraName"`     // 摄像机名称
    CommunityId    string         `json:"communityId"`    // 小区ID
    CommunityName  string         `json:"communityName"`  // 小区名称
    PersonName     string         `json:"personName"`     // 姓名
    IdCard         string         `json:"idCard"`         // 证件号码
    CameraLocation CameraLocation `json:"cameraLocation"` // 抓拍位置
    DocumentNumber string         `json:"documentNumber"` // 档案编号
    OrgId          string         `json:"orgId"`          // 机构ID
    OrgName        string         `json:"orgName"`        // 机构名称
    PicDate        string         `json:"picDate"`
    PicId          string         `json:"picId"`
    PicMaxUrl      string         `json:"picMaxUrl"`
    TaskId         string         `json:"taskId"`
    TaskName       string         `json:"taskName"`
    SdkName        string         `json:"sdkName"`
    Content        string         `json:"content"`
    ShowLabels     string         `json:"showLabels"`
    OtherLabels    string         `json:"otherLabels"`
    VideoUrl       string         `json:"videoUrl"`
    IsAlarm        bool           `json:"isAlarm"`
    IsAckAlarm     bool           `json:"isAckAlarm"`
    IsDelete       bool           `json:"isDelete"`
    IsKeyPerson    bool           `json:"isKeyPerson"`
    KeyPersonType  []string       `json:"keyPersonType"`
    DataSource     string         `json:"dataSource"` // 数据来源:摄像机, 数据栈
}

func QueryEsRecord(esClient *elasticsearch.Client, startTime, endTime string, orgIds, areaIds []interface{}, documentNumbers []string) ([]*ESRecordInfo, error) {
    var buf bytes.Buffer

    // 构建过滤条件
    var filters []map[string]interface{}
    documentNumberFilter := map[string]interface{}{
        "terms": map[string]interface{}{
            "documentNumber": documentNumbers,
        },
    }
    filters = append(filters, documentNumberFilter)

    if len(orgIds) > 0 || len(areaIds) > 0 {
        // 获取数据权限过滤条件
        authFilters := getDomainFilters(orgIds, areaIds)
        filters = append(filters, authFilters...)
    }

    // 时间范围
    filters = append(filters, map[string]interface{}{
        "range": map[string]interface{}{
            "picDate": map[string]interface{}{
                "gte": startTime,
                "lt":  endTime,
            },
        },
    })

    query := map[string]interface{}{
        "query": map[string]interface{}{
            "bool": map[string]interface{}{
                "filter": filters,
            },
        },
        "size":    0,
        "sort":    []interface{}{map[string]interface{}{"picDate": map[string]interface{}{"order": "asc"}}},
        "_source": map[string]interface{}{"includes": []interface{}{}, "excludes": []interface{}{"*.feature"}},
    }

    if err := json.NewEncoder(&buf).Encode(query); err != nil {
        return nil, fmt.Errorf("error encoding query: %s", err)
    }

    res, err := esClient.Search(
        esClient.Search.WithContext(context.Background()),
        esClient.Search.WithIndex(config.EsInfo.EsIndex.AiOcean.IndexName),
        esClient.Search.WithDocumentType(config.EsInfo.EsIndex.AiOcean.IndexType),
        esClient.Search.WithBody(&buf),
        esClient.Search.WithTrackTotalHits(true),
        esClient.Search.WithPretty(),
    )
    if err != nil {
        return nil, fmt.Errorf("error getting response: %s", err)
    }
    defer res.Body.Close()

    // Check for a successful status code (2xx range)
    if res.IsError() {
        return nil, fmt.Errorf("error getting response: %s", res.String())
    }

    var result map[string]interface{}
    if err := json.NewDecoder(res.Body).Decode(&result); err != nil {
        return nil, fmt.Errorf("error parsing response body: %s", err)
    }

    if _, ok := result["hits"]; !ok || result["hits"] == nil {
        return nil, nil
    }

    // 解析查询结果
    dat := result["hits"].(map[string]interface{})
    var records = make([]*ESRecordInfo, len(dat["hits"].([]interface{})))

    for idx, value := range dat["hits"].([]interface{}) {
        sourceData := value.(map[string]interface{})
        source := sourceData["_source"].(map[string]interface{})

        var record = ESRecordInfo{
            Id:             source["id"].(string),
            CameraId:       source["cameraId"].(string),
            CameraAddr:     source["cameraAddr"].(string),
            CameraName:     source["cameraName"].(string),
            CommunityId:    source["communityId"].(string),
            CommunityName:  source["communityName"].(string),
            DocumentNumber: source["documentNumber"].(string),
            OrgId:          source["orgId"].(string),
            OrgName:        source["orgName"].(string),
            PicDate:        source["picDate"].(string),
            PicId:          source["picId"].(string),
            PicMaxUrl:      source["picMaxUrl"].(string),
        }

        cameraLocation := source["cameraId"].(map[string]interface{})
        record.CameraLocation = CameraLocation{
            Building:  cameraLocation["building"].(string),
            Unit:      cameraLocation["unit"].(string),
            Pos:       cameraLocation["pos"].(string),
            Floor:     cameraLocation["floor"].(string),
            Direction: cameraLocation["direction"].(string),
        }

        records[idx] = &record
    }

    return records, nil
}

func getDomainFilters(orgIds, areaIds []interface{}) (filters []map[string]interface{}) {
    filters = make([]map[string]interface{}, 0)
    // 数据权限过滤
    if len(orgIds) > 0 && len(areaIds) > 0 {
        var authParams = map[string]interface{}{
            "bool": map[string]interface{}{
                "should": []interface{}{
                    map[string]interface{}{
                        "terms": map[string]interface{}{
                            "orgId": orgIds,
                        }},
                    map[string]interface{}{
                        "terms": map[string]interface{}{
                            "communityId": areaIds,
                        }},
                },
            },
        }
        filters = append(filters, authParams)

    } else if len(orgIds) > 0 {
        filters = append(filters, map[string]interface{}{
            "terms": map[string]interface{}{
                "orgId": orgIds,
            },
        })
    } else if len(areaIds) > 0 {
        filters = append(filters, map[string]interface{}{
            "terms": map[string]interface{}{
                "communityId": areaIds,
            },
        })
    }
    return filters
}