Merge branch 'feature/event' of http://192.168.1.8:3000/xuyiping/kpt-pasture into feature/event

model/neck_active_habit.go

@@ -135,10 +135,10 @@ func (n NeckActiveHabitSlice) ToPB(curveName, startDateTime, endDateTime string)
 		// 补全结尾不够的数据
 		if initFrameId == 0 && len(res.DateTimeList) > 0 {
 			lastDateTime := res.DateTimeList[len(res.DateTimeList)-1]
-			lastDatePase := strings.Split(lastDateTime, " ")
-			if len(lastDatePase) == 2 {
-				lastDay := lastDatePase[0]
-				lastHourStr := lastDatePase[1]
+			lastDatePare := strings.Split(lastDateTime, " ")
+			if len(lastDatePare) == 2 {
+				lastDay := lastDatePare[0]
+				lastHourStr := lastDatePare[1]
 				lastHour, _ := strconv.ParseInt(lastHourStr, 10, 64)
 				maxHour := util.ExpectedFrameIDs[len(util.ExpectedFrameIDs)-1]
 				xframeId := int32(lastHour-1)/2 + 1

model/neck_ring_estrus.go

@@ -30,7 +30,7 @@ func (n *NeckRingEstrus) TableName() string {
 }
 
 func NewNeckRingEstrus(pastureId int64, cow *Cow, level pasturePb.EstrusLevel_Kind,
-	checkResult pasturePb.CheckResult_Kind, isShow pasturePb.IsShow_Kind, firstTime string,
+	checkResult pasturePb.CheckResult_Kind, isShow pasturePb.IsShow_Kind,
 ) *NeckRingEstrus {
 	return &NeckRingEstrus{
 		PastureId:      pastureId,
@@ -41,14 +41,14 @@ func NewNeckRingEstrus(pastureId int64, cow *Cow, level pasturePb.EstrusLevel_Ki
 		ActiveLevel:    level,
 		IsShow:         isShow,
 		CheckResult:    checkResult,
-		FirstTime:      firstTime,
 	}
 }
 
 type GroupEstrusData struct {
-	CowId     int64
-	PastureId int64
-	EarNumber string
-	Records   []*NeckRingEstrus
-	Moved     bool
+	CowId          int64
+	PastureId      int64
+	EarNumber      string
+	NeckRingNumber string
+	Records        []*NeckRingEstrus
+	Moved          bool
 }

model/neck_ring_estrus_warning.go

@@ -6,14 +6,13 @@ import (
 	pasturePb "gitee.com/xuyiping_admin/go_proto/proto/go/backend/cow"
 )
 
-const IntCurFetal = 3
-
 type NeckRingEstrusWarning struct {
 	Id               int64                      `json:"id"`
 	NeckRingEstrusId int64                      `json:"neckRingEstrusId"`
 	PastureId        int64                      `json:"pastureId"`
 	CowId            int64                      `json:"cowId"`
 	EarNumber        string                     `json:"earNumber"`
+	NeckRingNumber   string                     `json:"neckRingNumber"`
 	FirstTime        string                     `json:"firstTime"`
 	DateTime         string                     `json:"dateTime"`
 	LastTime         string                     `json:"lastTime"`
@@ -32,7 +31,7 @@ func (n *NeckRingEstrusWarning) TableName() string {
 
 func NewNeckRingEstrusWarning(
 	neckRingEstrusId, pastureId, cowId int64,
-	earNumber, firstTime, dateTime, lastTime string,
+	earNumber, neckRingNumber, firstTime, dateTime, lastTime string,
 	warningKind pasturePb.Warning_Kind,
 	level pasturePb.EstrusLevel_Kind,
 ) *NeckRingEstrusWarning {
@@ -41,6 +40,7 @@ func NewNeckRingEstrusWarning(
 		PastureId:        pastureId,
 		CowId:            cowId,
 		EarNumber:        earNumber,
+		NeckRingNumber:   neckRingNumber,
 		FirstTime:        firstTime,
 		DateTime:         dateTime,
 		LastTime:         lastTime,

module/crontab/estrus_warning.go

@@ -68,7 +68,7 @@ func (e *Entry) NeckRingWarning(pastureId int64) (err error) {
 	}
 
 	minId := e.getMinId(pastureId)
-	estrusWarningList, err := e.GetCowHighChange(pastureId, minId)
+	/*estrusWarningList, err := e.GetCowHighChange(pastureId, minId)
 	if err != nil {
 		return xerr.WithStack(err)
 	}
@@ -105,7 +105,7 @@ func (e *Entry) NeckRingWarning(pastureId int64) (err error) {
 			zaplog.Error("UpdateNeckRingWarning", zap.Any("Updates", err), zap.Any("v", v))
 			continue
 		}
-	}
+	}*/
 
 	if err = e.UpdateNeckRingWarningIsPeak(pastureId, minId); err != nil {
 		zaplog.Error("UpdateNeckRingWarning", zap.Any("UpdateNeckRingWarningIsPeak", err), zap.Any("pastureId", pastureId))
@@ -140,11 +140,12 @@ func (e *Entry) GroupAndProcessData(records []*model.NeckRingEstrus) []*model.Ne
 		key := record.CowId
 		if _, exist := groups[key]; !exist {
 			groups[key] = &model.GroupEstrusData{
-				CowId:     record.CowId,
-				PastureId: record.PastureId,
-				Records:   make([]*model.NeckRingEstrus, 0),
-				EarNumber: record.EarNumber,
-				Moved:     false,
+				CowId:          record.CowId,
+				PastureId:      record.PastureId,
+				Records:        make([]*model.NeckRingEstrus, 0),
+				EarNumber:      record.EarNumber,
+				NeckRingNumber: record.NeckRingNumber,
+				Moved:          false,
 			}
 		}
 
@@ -174,9 +175,8 @@ func (e *Entry) GroupAndProcessData(records []*model.NeckRingEstrus) []*model.Ne
 		firstTime := findMaxTime(group.Records, func(r *model.NeckRingEstrus) string { return r.FirstTime })
 		dateTime := findMaxTime(group.Records, func(r *model.NeckRingEstrus) string { return r.ActiveTime })
 		neckRingEstrusWarning := model.NewNeckRingEstrusWarning(
-			maxId, group.PastureId, group.CowId, group.EarNumber,
-			firstTime, dateTime, latest.LastTime,
-			pasturePb.Warning_Estrus, latest.ActiveLevel,
+			maxId, group.PastureId, group.CowId, group.EarNumber, group.NeckRingNumber,
+			firstTime, dateTime, latest.LastTime, pasturePb.Warning_Estrus, latest.ActiveLevel,
 		)
 		neckRingEstrusWarningList = append(neckRingEstrusWarningList, neckRingEstrusWarning)
 	}

module/crontab/neck_ring_estrus.go

@@ -90,6 +90,9 @@ func (e *Entry) CowEstrusWarning(pastureId int64, xToday *XToday) (err error) {
 		if cft < float32(xToday.ActiveLow-XAdjust21) {
 			continue
 		}
+		if neckActiveHabitMap[habit.CowId] == nil {
+			neckActiveHabitMap[habit.CowId] = make([]*model.NeckActiveHabit, 0)
+		}
 		neckActiveHabitMap[habit.CowId] = append(neckActiveHabitMap[habit.CowId], habit)
 	}
 
@@ -121,7 +124,7 @@ func (e *Entry) CowEstrusWarning(pastureId int64, xToday *XToday) (err error) {
 				maxHigh = habit.FilterHigh
 			}
 			// 获取最新的 CreateTime
-			activeTimeParse, _ := time.Parse(habit.ActiveTime, model.LayoutTime)
+			activeTimeParse, _ := time.Parse(model.LayoutTime, habit.ActiveTime)
 			if activeTimeParse.After(lastActiveDate) {
 				lastActiveDate = activeTimeParse
 			}
@@ -148,12 +151,6 @@ func (e *Entry) CowEstrusWarning(pastureId int64, xToday *XToday) (err error) {
 			checkResult := getResult(before3Data, maxCft, cowEstrus)
 			isPeak := pasturePb.IsShow_Ok
 
-			cowEstrusStartData := e.FindCowEstrusFirstTime(pastureId, cowId, nowTime)
-			firstTime := ""
-			if cowEstrusStartData != nil {
-				firstTime = cowEstrusStartData.FirstTime
-			}
-
 			zaplog.Info("CowEstrusWarning",
 				zap.Any("level", level),
 				zap.Any("b48", b48),
@@ -167,10 +164,9 @@ func (e *Entry) CowEstrusWarning(pastureId int64, xToday *XToday) (err error) {
 				zap.Any("before3Data", before3Data),
 				zap.Any("cowEstrus", cowEstrus),
 				zap.Any("cowInfo", cowInfo),
-				zap.Any("firstTime", firstTime),
 				zap.Any("cowHabitList", cowHabitList),
 			)
-			newNeckRingEstrus := model.NewNeckRingEstrus(pastureId, cowInfo, level, checkResult, isShow, firstTime)
+			newNeckRingEstrus := model.NewNeckRingEstrus(pastureId, cowInfo, level, checkResult, isShow)
 			newNeckRingEstrus.LastTime = lastEstrusDate
 			newNeckRingEstrus.ActiveTime = lastActiveDate.Format(model.LayoutTime)
 			newNeckRingEstrus.DayHigh = dayHigh
@@ -187,11 +183,90 @@ func (e *Entry) CowEstrusWarning(pastureId int64, xToday *XToday) (err error) {
 		}
 	}
 
-	return err
+	// 更新牛只首次发情时间
+	e.UpdateEstrusFirstTime1(pastureId, nowTime)
+	e.UpdateEstrusIsPeak(pastureId, nowTime)
+	e.UpdateEstrusFirstTime2(pastureId, nowTime)
+	e.UpdateEstrusFirstTime3(pastureId, nowTime)
+	return nil
+}
+
+// UpdateEstrusFirstTime1 更新牛只首次发情时间
+func (e *Entry) UpdateEstrusFirstTime1(pastureId int64, xToday time.Time) {
+	neckRingEstrusList := e.FindNeckRingEstrusByFirstTimeEmpty(pastureId)
+	for _, v := range neckRingEstrusList {
+		cowEstrusStartData := e.FindCowEstrusFirstTime1(pastureId, v.CowId, xToday)
+		if cowEstrusStartData != nil && cowEstrusStartData.FirstTime != "" {
+			if err := e.DB.Model(new(model.NeckRingEstrus)).
+				Where("id = ?", v.Id).
+				Update("first_time", cowEstrusStartData.FirstTime).Error; err != nil {
+				zaplog.Error("UpdateEstrusFirstTime1", zap.Any("v", v), zap.Any("err", err), zap.Any("cowEstrusStartData", cowEstrusStartData))
+			}
+		}
+	}
+}
+
+func (e *Entry) UpdateEstrusFirstTime2(pastureId int64, xToday time.Time) {
+	neckRingEstrusList := e.FindNeckRingEstrusByFirstTimeEmpty(pastureId)
+	for _, v := range neckRingEstrusList {
+		zaplog.Info("UpdateEstrusFirstTime2", zap.Any("v", v))
+	}
+}
+
+func (e *Entry) UpdateEstrusFirstTime3(pastureId int64, xToday time.Time) {
+	neckRingEstrusList := e.FindNeckRingEstrusByFirstTimeEmpty(pastureId)
+	for _, v := range neckRingEstrusList {
+		activeTime, _ := time.Parse(model.LayoutTime, v.ActiveTime)
+		if activeTime.After(xToday.AddDate(0, 0, -2)) {
+			if err := e.DB.Model(new(model.NeckRingEstrus)).
+				Where("id = ?", v.Id).
+				Update("first_time", v.ActiveTime).Error; err != nil {
+				zaplog.Error("UpdateEstrusFirstTime1", zap.Any("v", v), zap.Any("err", err))
+			}
+		}
+	}
+}
+
+func (e *Entry) UpdateEstrusIsPeak(pastureId int64, xToday time.Time) {
+	neckRingEstrusList := make([]*model.NeckRingEstrus, 0)
+	if err := e.DB.Model(new(model.NeckRingEstrus)).
+		Where("first_time != ?", "").
+		Where("is_peak == ?", pasturePb.IsShow_Ok).
+		Where("is_show = ?", pasturePb.IsShow_Ok).
+		Where("pasture_id = ?", pastureId).
+		Find(&neckRingEstrusList).Error; err != nil {
+		zaplog.Error("UpdateEstrusIsPeak", zap.Any("Find", err))
+	}
+
+	for _, estrus := range neckRingEstrusList {
+		activeTime, _ := time.Parse(model.LayoutTime, estrus.ActiveTime)
+		if activeTime.Before(xToday) || activeTime.After(xToday.AddDate(0, 0, 1)) {
+			continue
+		}
+
+		var exists bool
+		if err := e.DB.Model(new(model.NeckRingEstrus)).
+			Where("cow_id = ?", estrus.CowId).
+			Where("first_time != ?", "").
+			Where("active_time BETWEEN ? AND ?", xToday, xToday.AddDate(0, 0, 1)).
+			Where("active_time BETWEEN ? AND ?", estrus.FirstTime, activeTime.Add(-2*time.Hour)).
+			Select("1").
+			Limit(1).
+			Scan(&exists).Error; err != nil {
+			zaplog.Error("UpdateEstrusIsPeak", zap.Any("exists", err))
+			continue
+		}
+		if exists {
+			if err := e.DB.Model(estrus).
+				Update("is_peak", pasturePb.IsShow_No).Error; err != nil {
+				zaplog.Error("UpdateEstrusIsPeak", zap.Any("v", estrus), zap.Any("err", err))
+			}
+		}
+	}
 }
 
-// FindCowEstrusFirstTime 查找牛只昨天是否有发情数据
-func (e *Entry) FindCowEstrusFirstTime(pastureId, cowId int64, xToday time.Time) *EstrusStartData {
+// FindCowEstrusFirstTime1 查找牛只昨天是否有发情数据
+func (e *Entry) FindCowEstrusFirstTime1(pastureId, cowId int64, xToday time.Time) *EstrusStartData {
 	firstTimeEventEstrus := &EstrusStartData{}
 	if err := e.DB.Model(new(model.NeckRingEstrus)).
 		Select("cow_id,MIN(STR_TO_DATE(first_time, '%Y-%m-%d %H:%i:%s')) as first_time").

module/crontab/neck_ring_estus_test.go

@@ -1,14 +1,13 @@
 package crontab
 
 import (
-	"encoding/json"
-	"fmt"
 	"kpt-pasture/model"
 	"testing"
+	"time"
 )
 
 func TestCalculateCFT(t *testing.T) {
-	neckActiveHabitList := make([]*model.NeckActiveHabit, 0)
+	/*neckActiveHabitList := make([]*model.NeckActiveHabit, 0)
 	str := `[
 	    {
 	        "id": 1725734,
@@ -3233,5 +3232,8 @@ func TestCalculateCFT(t *testing.T) {
 	for _, habit := range neckActiveHabitList {
 		cft := calculateCFT(habit)
 		fmt.Println(habit.Id, cft)
-	}
+	}*/
+
+	activeTimeParse, _ := time.Parse(model.LayoutTime, "2025-03-04 01:00:00")
+	println(activeTimeParse.Format(model.LayoutTime))
 }

module/crontab/sql.go

@@ -263,3 +263,16 @@ func (e *Entry) FindBefore3DaysNeckActiveHabit(pastureId int64, neckRingNumber,
 	}
 	return before3DaysNeckActiveHabit
 }
+
+// FindNeckRingEstrusByFirstTimeEmpty 查询firstTime为空的数据
+func (e *Entry) FindNeckRingEstrusByFirstTimeEmpty(pastureId int64) []*model.NeckRingEstrus {
+	neckRingEstrusList := make([]*model.NeckRingEstrus, 0)
+	if err := e.DB.Model(new(model.NeckRingEstrus)).
+		Where("first_time = ?", "").
+		Where("is_show = ?", pasturePb.IsShow_Ok).
+		Where("pasture_id = ?", pastureId).
+		Find(&neckRingEstrusList).Error; err != nil {
+		zaplog.Error("FindNeckRingEstrusFirstTime", zap.Any("err", err))
+	}
+	return neckRingEstrusList
+}

module/mqtt/mqtt_handle.go

@@ -36,7 +36,7 @@ func (e *Entry) NeckRingHandle(data []byte) {
 	}
 	if len(newData.NeckRingErrorData) > 0 || len(newData.NeckRingOriginalData) > 0 {
 		// 写入数据
-		if err := e.CreatedData(newData); err != nil {
+		if err := e.CreatedData(newData, data); err != nil {
 			zaplog.Error("Failed to create data", zap.Any("err", err), zap.Any("dataList", newData))
 		}
 	}
@@ -71,7 +71,7 @@ func (e *Entry) processBatch(batchList []*model.NeckRingOriginal) {
 	DSMLog.NeckRingOriginalData = append(DSMLog.NeckRingOriginalData, originalData...)
 
 	// 写入数据
-	if err := e.CreatedData(DSMLog); err != nil {
+	if err := e.CreatedData(DSMLog, []byte{}); err != nil {
 		zaplog.Error("Failed to create data", zap.Any("err", err), zap.Any("dataList", DSMLog))
 	}
 
@@ -80,7 +80,7 @@ func (e *Entry) processBatch(batchList []*model.NeckRingOriginal) {
 	DSMLog.NeckRingOriginalData = DSMLog.NeckRingOriginalData[:0]
 }
 
-func (e *Entry) CreatedData(DSMLog *DataInsertNeckRingLog) error {
+func (e *Entry) CreatedData(DSMLog *DataInsertNeckRingLog, originalData []byte) error {
 	if err := e.DB.Transaction(func(tx *gorm.DB) error {
 		if len(DSMLog.NeckRingErrorData) > 0 {
 			if err := e.DB.Model(new(model.NeckRingError)).
@@ -96,6 +96,7 @@ func (e *Entry) CreatedData(DSMLog *DataInsertNeckRingLog) error {
 				return xerr.WithStack(err)
 			}
 		}
+		zaplog.Info("CreatedData", zap.Any("DSMLog", DSMLog), zap.Any("originalData", string(originalData)))
 		return nil
 	}); err != nil {
 		return xerr.WithStack(err)