event: operation变更

model/neck_ring_original_data.go

@@ -0,0 +1,26 @@
+package model
+
+type NeckRingOriginalData struct {
+	Id         int64  `json:"id"`
+	SoftVer    int64  `json:"softVer"`
+	Uuid       string `json:"uuid"`
+	FrameId    int64  `json:"frameId"`
+	CowId      string `json:"cowId"`
+	Csq        int64  `json:"csq"`
+	Temp       int64  `json:"temp"`
+	Imei       string `json:"imei" `
+	Active     int32  `json:"active"`
+	InActive   int32  `json:"inactive"`
+	RuMina     int32  `json:"ruMina"`
+	Intake     int32  `json:"intake"`
+	Gasp       int32  `json:"gasp"`
+	Other      int32  `json:"other"`
+	ReMain     int32  `json:"remain"`
+	ActiveTime string `json:"activeTime"`
+	CreatedAt  int64  `json:"createdAt"`
+	UpdatedAt  int64  `json:"updatedAt"`
+}
+
+func (s *NeckRingOriginalData) TableName() string {
+	return "neck_ring_original_data"
+}

module/backend/event_base.go

@@ -108,13 +108,11 @@ func (s *StoreEntry) GroupTransferList(ctx context.Context, req *pasturePb.Searc
 	eventGroupTransferList := make([]*pasturePb.EventTransferGroupData, 0)
 	var count int64 = 0
 	pref := s.DB.Table(fmt.Sprintf("%s as a", new(model.EventTransferGroup).TableName())).
-		Select(`a.id,a.cow_id,a.transfer_in_pen_id,a.transfer_out_pen_id,a.lact,a.remarks,a.transfer_reason_id,
-			a.transfer_date,a.created_at,a.staff_member_id,b.name as transfer_in_pen_name,c.name as transfer_out_pen_name,
-			d.name as transfer_reason_name,e.name as staff_member_name,f.lact,f.ear_number`).
-		Joins(fmt.Sprintf("JOIN %s AS b ON a.transfer_in_pen_id = b.id", new(model.Pen).TableName())).
-		Joins(fmt.Sprintf("JOIN %s AS c on a.transfer_out_pen_id = c.id", new(model.Pen).TableName())).
-		Joins(fmt.Sprintf("JOIN %s AS d on a.transfer_reason_id = d.id", new(model.ConfigTransferPenReason).TableName())).
-		Joins(fmt.Sprintf("JOIN %s AS e ON a.staff_member_id = e.id", new(model.SystemUser).TableName())).
+		Select(`a.id,a.cow_id,a.pen_in_id as transfer_in_pen_id,a.pen_out_id as transfer_out_pen_id,a.lact,a.remarks,a.transfer_reason_id,a.transfer_reason_name,
+			a.transfer_date,a.created_at,a.operation_id,a.operation_name,b.name as transfer_in_pen_name,c.name as transfer_out_pen_name,
+			f.lact,f.ear_number`).
+		Joins(fmt.Sprintf("JOIN %s AS b ON a.pen_in_id = b.id", new(model.Pen).TableName())).
+		Joins(fmt.Sprintf("JOIN %s AS c on a.pen_out_id = c.id", new(model.Pen).TableName())).
 		Joins(fmt.Sprintf("JOIN %s AS f ON a.cow_id = f.id", new(model.Cow).TableName()))
 	if len(req.CowId) > 0 {
 		cowIds := strings.Split(req.CowId, ",")
@@ -191,17 +189,13 @@ func (s *StoreEntry) CreateGroupTransfer(ctx context.Context, req *pasturePb.Tra
 func (s *StoreEntry) BodyScoreList(ctx context.Context, req *pasturePb.SearchEventRequest, pagination *pasturePb.PaginationModel) (*pasturePb.SearchBodyScoreEventResponse, error) {
 	bodyScoreList := make([]*pasturePb.BodyScoreList, 0)
 	var count int64 = 0
-	pref := s.DB.Table(fmt.Sprintf("%s as a", new(model.EventBodyScore).TableName())).
-		Select(`a.id,a.cow_id,a.ear_number,a.score as body_score,a.lact,a.day_age,a.score_at,a.remarks,a.created_at,
-		a.updated_at,a.staff_member_id,a.operation_id,b.name as staff_member_name,c.name as operation_name`).
-		Joins(fmt.Sprintf("JOIN %s AS b on a.staff_member_id = b.id", new(model.SystemUser).TableName())).
-		Joins(fmt.Sprintf("JOIN %s AS c on a.operation_id = c.id", new(model.SystemUser).TableName()))
+	pref := s.DB.Model(new(model.EventBodyScore)).Select("*,score as body_score")
 	if len(req.CowId) > 0 {
 		cowIds := strings.Split(req.CowId, ",")
-		pref.Where("a.cow_id IN ?", cowIds)
+		pref.Where("cow_id IN ?", cowIds)
 	}
 
-	if err := pref.Order("a.id desc").
+	if err := pref.Order("id desc").
 		Count(&count).Limit(int(pagination.PageSize)).
 		Offset(int(pagination.PageOffset)).
 		Find(&bodyScoreList).Error; err != nil {
@@ -253,9 +247,7 @@ func (s *StoreEntry) WeightList(ctx context.Context, req *pasturePb.SearchEventR
 	var count int64 = 0
 	pref := s.DB.Table(fmt.Sprintf("%s as a", new(model.EventWeight).TableName())).
 		Select(`a.id,a.cow_id,a.ear_number,a.weight / 100 as weight,a.lact,a.day_age,a.weight_at,a.remarks,a.created_at,
-		a.updated_at,a.staff_member_id,a.operation_id,b.name as staff_member_name,c.name as operation_name`).
-		Joins(fmt.Sprintf("JOIN %s AS b on a.staff_member_id = b.id", new(model.SystemUser).TableName())).
-		Joins(fmt.Sprintf("JOIN %s AS c on a.operation_id = c.id", new(model.SystemUser).TableName()))
+		a.updated_at,a.message_id,a.operation_id,a.message_name,a.operation_name`)
 	if len(req.CowId) > 0 {
 		cowIds := strings.Split(req.CowId, ",")
 		pref.Where("a.cow_id IN ?", cowIds)

module/crontab/cow_cron.go

@@ -76,6 +76,7 @@ func (e *Entry) UpdateCowInfo() error {
 				"pregnancy_age": cow.GetDaysPregnant(),
 				"admission_age": cow.GetAdmissionAge(),
 				"abortion_age":  cow.GetAbortionAge(),
+				// todo 牛只类型待完善
 			}).Error; err != nil {
 			zaplog.Error("Crontab", zap.Any("UpdateCowDayAge", err))
 		}

service/mqtt/sub.go

@@ -3,7 +3,14 @@ package mqtt
 import (
 	"fmt"
 	"kpt-pasture/config"
+	"kpt-pasture/model"
+	"kpt-pasture/util"
+	"strconv"
+	"strings"
 	"sync"
+	"time"
+
+	pasturePb "gitee.com/xuyiping_admin/go_proto/proto/go/backend/cow"
 
 	"gitee.com/xuyiping_admin/pkg/logger/zaplog"
 	golangMqtt "github.com/eclipse/paho.mqtt.golang"
@@ -59,16 +66,194 @@ func (d *DataEventEntry) SubMsg(conf config.MqttSetting, client golangMqtt.Clien
 	}
 
 	defer close(subMsgChan)
-	for {
-		select {
-		case msg := <-subMsgChan:
-			bufferPool.Put(msg)
-			d.ProcessMessages(msg)
+	select {
+	case msg := <-subMsgChan:
+		bufferPool.Put(msg)
+		d.ProcessMessages(msg)
+	}
+}
+
+func (d *DataEventEntry) ProcessMessages(msg []byte) {
+	neckRingOriginalData, err := d.MsgDataFormat(msg)
+	if err != nil {
+		zaplog.Error("MsgDataFormat", zap.Any("err", err), zap.Any("msg", string(msg)))
+		return
+	}
+	if neckRingOriginalData == nil {
+		return
+	}
+
+	if neckRingOriginalData.Imei == "" {
+		zaplog.Info("neckRingOriginalData", zap.Any("msg", string(msg)), zap.Any("neckRingOriginalData", neckRingOriginalData))
+		return
+	}
+
+	defer func() {
+		if time.Now().Day()%15 == 0 {
+			d.DB.Model(new(model.NeckRingOriginalData)).
+				Where("created_at < ?", time.Now().AddDate(-2, 0, 0).Unix()).
+				Delete(new(model.NeckRingOriginalData))
+			return
 		}
+	}()
+
+	// 计算牛只实际活动时间
+	nowDayTime := time.Now()
+	currHour := nowDayTime.Hour() + 2
+	frameIdHour := neckRingOriginalData.FrameId * 2
+	frameDayTime := fmt.Sprintf("%s %s:00:00", nowDayTime.Format(model.LayoutDate2), fmt.Sprintf("%02d", frameIdHour))
+	if frameIdHour > int64(currHour) {
+		frameDayTime = fmt.Sprintf("%s %s:00:00", nowDayTime.AddDate(0, 0, -1).Format(model.LayoutDate2), fmt.Sprintf("%02d", frameIdHour))
+	}
+
+	neckRingOriginalData.ActiveTime = frameDayTime
+	if err = d.DB.Create(neckRingOriginalData).Error; err != nil {
+		zaplog.Error("ProcessMessages", zap.Any("err", err), zap.Any("neckRingOriginalData", neckRingOriginalData))
 	}
 
+	// 更新脖环数据状态
+	neckRingStatus := pasturePb.NeckRingStatus_Normal
+	errorReason := ""
+	if neckRingOriginalData.FrameId >= 11 || neckRingOriginalData.FrameId < 0 {
+		neckRingStatus = pasturePb.NeckRingStatus_Error
+		errorReason = "数据异常"
+	}
+
+	d.DB.Model(new(model.NeckRingLog)).
+		Where("number = ?", neckRingOriginalData.Imei).
+		Updates(map[string]interface{}{
+			"status":       neckRingStatus,
+			"error_reason": errorReason,
+		})
 }
 
-func (d *DataEventEntry) ProcessMessages(msg []byte) {
-	fmt.Println("===byte==", string(msg))
+func (d *DataEventEntry) MsgDataFormat(msg []byte) (*model.NeckRingOriginalData, error) {
+	msgData := make(map[string]interface{})
+	pairs := strings.Split(util.MsgFormat(string(msg)), " ")
+	for _, pair := range pairs {
+		parts := strings.SplitN(pair, ":", 2)
+		if len(parts) != 2 {
+			continue
+		}
+		key, value := parts[0], parts[1]
+		if len(key) == 0 {
+			continue
+		}
+		msgData[key] = value
+	}
+
+	softVer := int64(0)
+	if softVerInter, ok := msgData["SOFT_VER"]; ok {
+		if softVerstr, ok := softVerInter.(string); ok {
+			softVer, _ = strconv.ParseInt(softVerstr, 10, 64)
+		}
+	}
+
+	uuid := ""
+	if uuidInter, ok := msgData["uuid"]; ok {
+		if uuidStr, ok := uuidInter.(string); ok {
+			uuid = uuidStr
+		}
+	}
+
+	frameId := int64(0)
+	if frameIdInter, ok := msgData["frameid"]; ok {
+		if frameId64, ok := frameIdInter.(string); ok {
+			frameId, _ = strconv.ParseInt(frameId64, 10, 64)
+		}
+	}
+	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 {
+			temp, _ = strconv.ParseFloat(tempFloat, 64)
+		}
+	}
+
+	imei := ""
+	if imeiInter, ok := msgData["imei"]; ok {
+		if imeiStr, ok := imeiInter.(string); ok {
+			imei = imeiStr
+		}
+	}
+
+	active := int64(0)
+	if activeInter, ok := msgData["active"]; ok {
+		if active32, ok := activeInter.(string); ok {
+			active, _ = strconv.ParseInt(active32, 10, 64)
+		}
+	}
+
+	inAction := int64(0)
+	if inActionInter, ok := msgData["inactive"]; ok {
+		if inAction32, ok := inActionInter.(string); ok {
+			inAction, _ = strconv.ParseInt(inAction32, 10, 64)
+		}
+	}
+
+	ruMina := int64(0)
+	if ruMinaInter, ok := msgData["Rumina"]; ok {
+		if ruMina32, ok := ruMinaInter.(string); ok {
+			ruMina, _ = strconv.ParseInt(ruMina32, 10, 64)
+		}
+	}
+
+	intake := int64(0)
+	if intakeInter, ok := msgData["Intake"]; ok {
+		if intake32, ok := intakeInter.(string); ok {
+			intake, _ = strconv.ParseInt(intake32, 10, 64)
+		}
+	}
+
+	gasp := int64(0)
+	if gaspInter, ok := msgData["gasp"]; ok {
+		if gasp32, ok := gaspInter.(string); ok {
+			gasp, _ = strconv.ParseInt(gasp32, 10, 64)
+		}
+	}
+
+	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 {
+			reMain, _ = strconv.ParseInt(reMain32, 10, 64)
+		}
+	}
+
+	return &model.NeckRingOriginalData{
+		SoftVer:  softVer,
+		Uuid:     uuid,
+		FrameId:  frameId,
+		CowId:    cowId,
+		Csq:      csq,
+		Temp:     int64(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),
+	}, nil
+
 }

util/util.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"math"
 	"math/rand"
+	"regexp"
 	"time"
 
 	"gitee.com/xuyiping_admin/pkg/xerr"
@@ -307,3 +308,12 @@ func DaysBetween(startDayUnix int64, endDayUnix int64) int64 {
 
 	return daysDiff
 }
+
+// MsgFormat 格式化消息字符串 字符串里面有多个冒号，仅删除冒号前后的空格（如果存在）
+func MsgFormat(input string) string {
+	// 定义正则表达式，用于匹配冒号两边的空格
+	re := regexp.MustCompile(`\s*:\s*`)
+
+	// 使用正则表达式替换所有匹配的部分
+	return re.ReplaceAllString(input, ":")
+}

util/util_test.go

@@ -1,7 +1,9 @@
 package util
 
 import (
+	"fmt"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 )
@@ -413,5 +415,16 @@ func TestDaysBetween(t *testing.T) {
 }
 
 func Test_demo(t *testing.T) {
-
+	fmt.Println(time.Now().AddDate(-2, 0, 0).Format("2006-01-02"))
+	fmt.Println(fmt.Sprintf("%02d", 0))
+	fmt.Println(fmt.Sprintf("%02d", 1))
+	fmt.Println(fmt.Sprintf("%02d", 2))
+	fmt.Println(fmt.Sprintf("%02d", 3))
+	fmt.Println(fmt.Sprintf("%02d", 4))
+	fmt.Println(fmt.Sprintf("%02d", 5))
+	fmt.Println(fmt.Sprintf("%02d", 9))
+	fmt.Println(fmt.Sprintf("%02d", 10))
+	fmt.Println(fmt.Sprintf("%02d", 12))
+	fmt.Println(fmt.Sprintf("%02d", 22))
+	fmt.Println(fmt.Sprintf("%02d", 23))
 }