|
|
@@ -2,11 +2,16 @@ package mqtt
|
|
|
|
|
|
import (
|
|
|
"encoding/json"
|
|
|
+ "fmt"
|
|
|
"kpt-pasture/model"
|
|
|
"kpt-pasture/util"
|
|
|
+ "math"
|
|
|
"strconv"
|
|
|
"strings"
|
|
|
"sync"
|
|
|
+ "time"
|
|
|
+
|
|
|
+ "github.com/jinzhu/copier"
|
|
|
|
|
|
pasturePb "gitee.com/xuyiping_admin/go_proto/proto/go/backend/cow"
|
|
|
|
|
|
@@ -18,69 +23,70 @@ import (
|
|
|
|
|
|
type DataInsertNeckRingLog struct {
|
|
|
NeckRingOriginalData []*model.NeckRingOriginal
|
|
|
- NeckRingErrorData []*model.NeckRingOriginal
|
|
|
- NeckRingUnRegisterData []*model.NeckRingOriginal
|
|
|
+ NeckRingErrorData []*model.NeckRingError
|
|
|
+ NeckRingUnRegisterData []*model.NeckRingUnRegister
|
|
|
Mx *sync.RWMutex
|
|
|
}
|
|
|
|
|
|
-var FrameId = map[int32]int32{
|
|
|
- 1: 1, 2: 2, 3: 3, 4: 4, 5: 5, 6: 6, 8: 8,
|
|
|
- 11: 11, 12: 12, 13: 13, 14: 14, 15: 15, 16: 16, 18: 18,
|
|
|
- 21: 21, 22: 22, 23: 23, 24: 24, 25: 25, 26: 26, 28: 28,
|
|
|
- 31: 31, 32: 32, 33: 33, 34: 34, 35: 35, 36: 36, 38: 38,
|
|
|
- 41: 41, 42: 42, 43: 43, 44: 44, 45: 45, 46: 46, 48: 48,
|
|
|
- 51: 51, 52: 52, 53: 53, 54: 54, 55: 55, 56: 56, 58: 58,
|
|
|
- 61: 61, 62: 62, 63: 63, 64: 64, 65: 65, 66: 66, 68: 68,
|
|
|
- 71: 71, 72: 72, 73: 73, 74: 74, 75: 75, 76: 76, 78: 78,
|
|
|
- 81: 81, 82: 82, 83: 83, 84: 84, 85: 85, 86: 86, 88: 88,
|
|
|
- 91: 91, 92: 92, 93: 93, 94: 94, 95: 95, 96: 96, 98: 98,
|
|
|
- 101: 101, 102: 102, 103: 103, 104: 104, 105: 105, 106: 106, 108: 108,
|
|
|
- 111: 111, 112: 112, 113: 113, 114: 114, 115: 115, 116: 116, 118: 118,
|
|
|
-}
|
|
|
-
|
|
|
var (
|
|
|
- batchSize = 10
|
|
|
- batchList = make([]*model.NeckRingOriginal, 0, batchSize)
|
|
|
+ batchSize = 10
|
|
|
+ batchList = make([]*model.NeckRingOriginal, 0, batchSize)
|
|
|
+ DefaultLimit = int32(10000)
|
|
|
+ DSMLog = &DataInsertNeckRingLog{
|
|
|
+ NeckRingOriginalData: make([]*model.NeckRingOriginal, 0),
|
|
|
+ NeckRingErrorData: make([]*model.NeckRingError, 0),
|
|
|
+ NeckRingUnRegisterData: make([]*model.NeckRingUnRegister, 0),
|
|
|
+ Mx: &sync.RWMutex{},
|
|
|
+ }
|
|
|
)
|
|
|
|
|
|
func (e *Entry) NeckRingHandle(data []byte) {
|
|
|
- DSMLog := &DataInsertNeckRingLog{
|
|
|
- NeckRingOriginalData: make([]*model.NeckRingOriginal, 0),
|
|
|
- NeckRingErrorData: make([]*model.NeckRingOriginal, 0),
|
|
|
- NeckRingUnRegisterData: make([]*model.NeckRingOriginal, 0),
|
|
|
- Mx: &sync.RWMutex{},
|
|
|
+ newData := e.MsgDataFormat2(data)
|
|
|
+ if newData == nil || len(newData) <= 0 {
|
|
|
+ return
|
|
|
+ }
|
|
|
+ zaplog.Info("NeckRingHandle", zap.Any("data", newData), zap.Any("original", string(data)), zap.Any("time", time.Now().Unix()))
|
|
|
+ batchList = append(batchList, newData...)
|
|
|
+ if len(batchList) >= batchSize {
|
|
|
+ e.processBatch(batchList)
|
|
|
+ batchList = batchList[:0] // 清空 batchList
|
|
|
}
|
|
|
+}
|
|
|
|
|
|
- newData := e.MsgDataFormat(data)
|
|
|
- if newData != nil {
|
|
|
- batchList = append(batchList, newData)
|
|
|
- if len(batchList) >= batchSize {
|
|
|
- DSMLog.Mx.Lock()
|
|
|
- for _, batch := range batchList {
|
|
|
- // 异常脖环数据
|
|
|
- if _, ok := FrameId[batch.FrameId]; !ok {
|
|
|
- DSMLog.NeckRingErrorData = append(DSMLog.NeckRingErrorData, batch)
|
|
|
- continue
|
|
|
- }
|
|
|
- // 未佩戴的脖环数据
|
|
|
- if ok := e.NeckRingIsBind(batch.Imei); !ok {
|
|
|
- DSMLog.NeckRingUnRegisterData = append(DSMLog.NeckRingUnRegisterData, batch)
|
|
|
- continue
|
|
|
- }
|
|
|
- // 正常脖环数据
|
|
|
- DSMLog.NeckRingOriginalData = append(DSMLog.NeckRingOriginalData, batch)
|
|
|
- }
|
|
|
+// 处理批量数据
|
|
|
+func (e *Entry) processBatch(batchList []*model.NeckRingOriginal) {
|
|
|
+ // 初始化分类数据
|
|
|
+ var (
|
|
|
+ errorData []*model.NeckRingError
|
|
|
+ originalData []*model.NeckRingOriginal
|
|
|
+ )
|
|
|
|
|
|
- if err := e.CreatedData(DSMLog); err != nil {
|
|
|
- zaplog.Error("subMsgChan-os", zap.Any("err", err), zap.Any("dataList", DSMLog))
|
|
|
- }
|
|
|
- DSMLog.Mx.Unlock()
|
|
|
- DSMLog.NeckRingUnRegisterData = make([]*model.NeckRingOriginal, 0)
|
|
|
- DSMLog.NeckRingErrorData = make([]*model.NeckRingOriginal, 0)
|
|
|
- DSMLog.NeckRingOriginalData = make([]*model.NeckRingOriginal, 0)
|
|
|
- batchList = batchList[:0]
|
|
|
+ // 分类数据
|
|
|
+ for _, batch := range batchList {
|
|
|
+ // 异常脖环数据
|
|
|
+ if ok := util.IsValidFrameId(batch.Frameid); !ok {
|
|
|
+ var ed model.NeckRingError
|
|
|
+ copier.Copy(&ed, &batch)
|
|
|
+ errorData = append(errorData, &ed)
|
|
|
+ } else {
|
|
|
+ originalData = append(originalData, batch)
|
|
|
}
|
|
|
}
|
|
|
+
|
|
|
+ // 更新日志
|
|
|
+ DSMLog.Mx.Lock()
|
|
|
+ defer DSMLog.Mx.Unlock()
|
|
|
+ DSMLog.NeckRingErrorData = append(DSMLog.NeckRingErrorData, errorData...)
|
|
|
+ DSMLog.NeckRingOriginalData = append(DSMLog.NeckRingOriginalData, originalData...)
|
|
|
+
|
|
|
+ // 写入数据
|
|
|
+ if err := e.CreatedData(DSMLog); err != nil {
|
|
|
+ zaplog.Error("Failed to create data", zap.Any("err", err), zap.Any("dataList", DSMLog))
|
|
|
+ }
|
|
|
+
|
|
|
+ // 清空日志
|
|
|
+ DSMLog.NeckRingErrorData = DSMLog.NeckRingErrorData[:0]
|
|
|
+ DSMLog.NeckRingOriginalData = DSMLog.NeckRingOriginalData[:0]
|
|
|
}
|
|
|
|
|
|
func (e *Entry) CreatedData(DSMLog *DataInsertNeckRingLog) error {
|
|
|
@@ -96,12 +102,6 @@ func (e *Entry) CreatedData(DSMLog *DataInsertNeckRingLog) error {
|
|
|
return xerr.WithStack(err)
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
- if len(DSMLog.NeckRingUnRegisterData) > 0 {
|
|
|
- if err := e.DB.Create(DSMLog.NeckRingUnRegisterData).Error; err != nil {
|
|
|
- return xerr.WithStack(err)
|
|
|
- }
|
|
|
- }
|
|
|
return nil
|
|
|
}); err != nil {
|
|
|
return xerr.WithStack(err)
|
|
|
@@ -109,7 +109,7 @@ func (e *Entry) CreatedData(DSMLog *DataInsertNeckRingLog) error {
|
|
|
return nil
|
|
|
}
|
|
|
|
|
|
-func (e *Entry) MsgDataFormat(msg []byte) *model.NeckRingOriginal {
|
|
|
+func (e *Entry) MsgDataFormat(msg []byte) []*model.NeckRingOriginal {
|
|
|
msgData := make(map[string]interface{})
|
|
|
pairs := strings.Split(util.MsgFormat(string(msg)), " ")
|
|
|
for _, pair := range pairs {
|
|
|
@@ -153,20 +153,6 @@ func (e *Entry) MsgDataFormat(msg []byte) *model.NeckRingOriginal {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- cowId := ""
|
|
|
- if cowIdInter, ok := msgData["cowid"]; ok {
|
|
|
- if cowIdStr, ok := cowIdInter.(string); ok {
|
|
|
- cowId = cowIdStr
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- csq := int64(0)
|
|
|
- if csqInter, ok := msgData["csq"]; ok {
|
|
|
- if csq32, ok := csqInter.(string); ok {
|
|
|
- csq, _ = strconv.ParseInt(csq32, 10, 64)
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
temp := float64(0)
|
|
|
if tempInter, ok := msgData["Temp"]; ok {
|
|
|
if tempFloat, ok := tempInter.(string); ok {
|
|
|
@@ -238,13 +224,6 @@ func (e *Entry) MsgDataFormat(msg []byte) *model.NeckRingOriginal {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- other := int64(0)
|
|
|
- if otherInter, ok := msgData["other"]; ok {
|
|
|
- if other32, ok := otherInter.(string); ok {
|
|
|
- other, _ = strconv.ParseInt(other32, 10, 64)
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
reMain := int64(0)
|
|
|
if reMainInter, ok := msgData["Remain"]; ok {
|
|
|
if reMain32, ok := reMainInter.(string); ok {
|
|
|
@@ -258,6 +237,27 @@ func (e *Entry) MsgDataFormat(msg []byte) *model.NeckRingOriginal {
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
+ /*cowId := ""
|
|
|
+ if cowIdInter, ok := msgData["cowid"]; ok {
|
|
|
+ if cowIdStr, ok := cowIdInter.(string); ok {
|
|
|
+ cowId = cowIdStr
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ csq := int64(0)
|
|
|
+ if csqInter, ok := msgData["csq"]; ok {
|
|
|
+ if csq32, ok := csqInter.(string); ok {
|
|
|
+ csq, _ = strconv.ParseInt(csq32, 10, 64)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ other := int64(0)
|
|
|
+ if otherInter, ok := msgData["other"]; ok {
|
|
|
+ if other32, ok := otherInter.(string); ok {
|
|
|
+ other, _ = strconv.ParseInt(other32, 10, 64)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
nccId := ""
|
|
|
if nccIdInter, ok := msgData["nccid"]; ok {
|
|
|
if nccIdStr, ok := nccIdInter.(string); ok {
|
|
|
@@ -265,56 +265,138 @@ func (e *Entry) MsgDataFormat(msg []byte) *model.NeckRingOriginal {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- return &model.NeckRingOriginal{
|
|
|
- Version: int32(softVer),
|
|
|
- Uuid: uuid,
|
|
|
- CowId: cowId,
|
|
|
- FrameId: int32(frameId),
|
|
|
- Csq: int32(csq),
|
|
|
- Temp: int32(temp * 100),
|
|
|
- Imei: imei,
|
|
|
- Active: int32(active),
|
|
|
- Inactive: int32(inAction),
|
|
|
- Rumina: int32(ruMina),
|
|
|
- Intake: int32(intake),
|
|
|
- Gasp: int32(gasp),
|
|
|
- Other: int32(other),
|
|
|
- Remain: int32(reMain),
|
|
|
- Nccid: nccId,
|
|
|
+ */
|
|
|
+
|
|
|
+ return []*model.NeckRingOriginal{
|
|
|
+ {
|
|
|
+ FirmwareVersion: int32(softVer),
|
|
|
+ Uuid: uuid,
|
|
|
+ Frameid: int32(frameId),
|
|
|
+ ReceiveNumber: imei,
|
|
|
+ Active: int32(active),
|
|
|
+ Inactive: int32(inAction),
|
|
|
+ Rumina: int32(ruMina),
|
|
|
+ Intake: int32(intake),
|
|
|
+ Gasp: int32(gasp),
|
|
|
+ Remain: int32(reMain),
|
|
|
+ },
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-func (e *Entry) MsgDataFormat2(msg []byte) *model.NeckRingOriginal {
|
|
|
- neckLog := &Behavior{}
|
|
|
- if err := json.Unmarshal(msg, neckLog); err != nil {
|
|
|
+func (e *Entry) MsgDataFormat2(msg []byte) []*model.NeckRingOriginal {
|
|
|
+ neckLogList := &NeckRingWrapper{}
|
|
|
+ if err := json.Unmarshal(msg, neckLogList); err != nil {
|
|
|
zaplog.Error("MsgDataFormat", zap.Any("err", err), zap.Any("msg", string(msg)))
|
|
|
}
|
|
|
- if neckLog.Imei != "" {
|
|
|
- // 存储到数据库
|
|
|
- activeDate, hours := util.GetNeckRingActiveTimer(neckLog.FrameId)
|
|
|
- voltage, _ := strconv.ParseInt(strconv.FormatInt(int64(neckLog.BAT), 16), 10, 64)
|
|
|
+
|
|
|
+ res := make([]*model.NeckRingOriginal, 0)
|
|
|
+ for _, neckLog := range neckLogList.NeckRing.NeckPck {
|
|
|
+ activeDate, hours := util.GetNeckRingActiveTimer(neckLog.Frameid)
|
|
|
activeDateTimeType := pasturePb.ActiveTimeType_Twenty_Minutes
|
|
|
- if neckLog.FrameId%10 == 8 {
|
|
|
+ if neckLog.Frameid%10 == 8 {
|
|
|
activeDateTimeType = pasturePb.ActiveTimeType_Two_Hours
|
|
|
}
|
|
|
- return &model.NeckRingOriginal{
|
|
|
- Uuid: neckLog.UUID,
|
|
|
- Imei: neckLog.ECowId,
|
|
|
- ActiveDate: activeDate,
|
|
|
- Hours: int32(hours),
|
|
|
- FrameId: neckLog.FrameId,
|
|
|
- Rumina: neckLog.RuMina,
|
|
|
- Intake: neckLog.Intake,
|
|
|
- Inactive: neckLog.Inactive,
|
|
|
- Other: neckLog.Other,
|
|
|
- High: neckLog.Activitys,
|
|
|
- Active: neckLog.High,
|
|
|
- Voltage: int32(voltage),
|
|
|
- Version: neckLog.Sver,
|
|
|
- Remain: neckLog.Remain,
|
|
|
- ReceiveNumber: neckLog.Imei,
|
|
|
- ActiveDateType: activeDateTimeType,
|
|
|
+ res = append(res, &model.NeckRingOriginal{
|
|
|
+ Uuid: neckLog.UUID,
|
|
|
+ NeckRingNumber: fmt.Sprintf("%d", neckLog.Ecowid),
|
|
|
+ ActiveDate: activeDate,
|
|
|
+ Hours: int32(hours),
|
|
|
+ Frameid: neckLog.Frameid,
|
|
|
+ Rumina: neckLog.Rumina,
|
|
|
+ Intake: neckLog.Intake,
|
|
|
+ Inactive: neckLog.Inactive,
|
|
|
+ Gasp: neckLog.Other,
|
|
|
+ High: neckLog.Activitys,
|
|
|
+ Active: neckLog.High,
|
|
|
+ FirmwareVersion: neckLog.Sver,
|
|
|
+ HardwareVersion: neckLog.Hver,
|
|
|
+ Remain: neckLog.Remain,
|
|
|
+ Voltage: neckLog.BAT,
|
|
|
+ RestartReason: neckLog.STATUS,
|
|
|
+ Upper: neckLog.UpPer,
|
|
|
+ Imei: neckLog.Imei,
|
|
|
+ ReceiveNumber: neckLog.Imei,
|
|
|
+ ActiveDateType: activeDateTimeType,
|
|
|
+ IsShow: pasturePb.IsShow_No,
|
|
|
+ })
|
|
|
+ }
|
|
|
+ return res
|
|
|
+}
|
|
|
+
|
|
|
+// NeckRingOriginalMergeData 把脖环数据合并成2个小时的
|
|
|
+func (e *Entry) NeckRingOriginalMergeData() {
|
|
|
+ limit := e.Cfg.NeckRingLimit
|
|
|
+ if limit <= 0 {
|
|
|
+ limit = DefaultLimit
|
|
|
+ }
|
|
|
+
|
|
|
+ updateOriginalMaxId := e.GetSystemConfigure(model.MaxEstrus).Value
|
|
|
+ neckRingList := make([]*model.NeckRingOriginal, 0)
|
|
|
+ if err := e.DB.Model(new(model.NeckRingOriginal)).
|
|
|
+ Where("id > ?", updateOriginalMaxId).
|
|
|
+ Order("id asc").Limit(int(limit)).
|
|
|
+ Find(&neckRingList).Error; err != nil {
|
|
|
+ return
|
|
|
+ }
|
|
|
+
|
|
|
+ if len(neckRingList) <= 0 {
|
|
|
+ return
|
|
|
+ }
|
|
|
+
|
|
|
+ originalMapData := make(map[string]*model.NeckRingOriginalMerge)
|
|
|
+ // 合并成2个小时的
|
|
|
+ for _, v := range neckRingList {
|
|
|
+ xframeId := int(math.Floor(float64(v.Frameid)/10) * 2)
|
|
|
+ mapKey := fmt.Sprintf("%s%s%s%s%d", v.NeckRingNumber, model.JoinKey, v.ActiveDate, model.JoinKey, xframeId) // 0001/2023-12-04/0 0001/2023-12-03/4
|
|
|
+ if _, ok := originalMapData[mapKey]; !ok {
|
|
|
+ originalMapData[mapKey] = new(model.NeckRingOriginalMerge)
|
|
|
}
|
|
|
+ v.IsAvgHours()
|
|
|
+ originalMapData[mapKey].IsMageData(v)
|
|
|
+ }
|
|
|
+
|
|
|
+ // 算平均值
|
|
|
+ for _, v := range originalMapData {
|
|
|
+ v.SumAvg()
|
|
|
+ }
|
|
|
+
|
|
|
+ // 更新脖环牛只相关信息
|
|
|
+ newNeckActiveHabitList := model.NeckRingOriginalMap(originalMapData).ForMatData()
|
|
|
+ if err := e.DB.Transaction(func(tx *gorm.DB) error {
|
|
|
+ // 更新已处理过的id
|
|
|
+ if err := tx.Model(new(model.SystemConfigure)).
|
|
|
+ Where("name = ?", model.UpdateOriginalMaxId).
|
|
|
+ Update("value", neckRingList[len(neckRingList)-1].Id).
|
|
|
+ Error; err != nil {
|
|
|
+ return xerr.WithStack(err)
|
|
|
+ }
|
|
|
+
|
|
|
+ for _, neckActiveHabit := range newNeckActiveHabitList {
|
|
|
+ // 新数据直接插入 todo 待优化
|
|
|
+ historyNeckActiveHabit, ct := e.IsExistNeckActiveHabit(neckActiveHabit.NeckRingNumber, neckActiveHabit.HeatDate, neckActiveHabit.Frameid)
|
|
|
+ if ct <= 0 {
|
|
|
+ if err := tx.Create(neckActiveHabit).Error; err != nil {
|
|
|
+ return xerr.WithStack(err)
|
|
|
+ }
|
|
|
+ continue
|
|
|
+ }
|
|
|
+
|
|
|
+ if historyNeckActiveHabit == nil {
|
|
|
+ zaplog.Error("NeckRingOriginalMergeData", zap.Any("historyNeckActiveHabit", historyNeckActiveHabit))
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ // 更新数据
|
|
|
+ historyNeckActiveHabit.MergeData(neckActiveHabit)
|
|
|
+ if err := tx.Model(new(model.NeckActiveHabit)).
|
|
|
+ Select("rumina", "intake", "inactive", "gasp", "other", "high", "active").
|
|
|
+ Where("id = ?", historyNeckActiveHabit.Id).
|
|
|
+ Updates(historyNeckActiveHabit).Error; err != nil {
|
|
|
+ return xerr.WithStack(err)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return nil
|
|
|
+ }); err != nil {
|
|
|
+ zaplog.Error("NeckRingOriginalMergeData", zap.Any("transaction", err))
|
|
|
+ return
|
|
|
}
|
|
|
- return nil
|
|
|
}
|
Yi