crontab: neck_ring_calculate update

module/backend/cow.go

@@ -207,6 +207,7 @@ func (s *StoreEntry) BehaviorCurve(ctx context.Context, req *pasturePb.CowBehavi
 		Where("neck_ring_number = ?", cowInfo.NeckRingNumber).
 		Where("pasture_id = ?", userModel.AppPasture.Id).
 		Where("is_show = ?", pasturePb.IsShow_Ok).
+		Where("cow_id > ?", 0).
 		Where("heat_date BETWEEN ? AND ?", startDataTime, endDataTime).
 		Order("heat_date, frameid").
 		Find(&neckActiveHabitList).Error; err != nil {

module/crontab/health_warning.go

@@ -3,18 +3,19 @@ package crontab
 import (
 	"kpt-pasture/model"
 
-	"gitee.com/xuyiping_admin/pkg/xerr"
+	"gitee.com/xuyiping_admin/pkg/logger/zaplog"
+	"go.uber.org/zap"
 )
 
-func (e *Entry) HealthWarning(pastureId int64, processIds []int64) (err error) {
+func (e *Entry) HealthWarning(pastureId int64, processIds []int64) {
 	newNeckActiveHabitList := make([]*model.NeckActiveHabit, 0)
-	if err = e.DB.Model(new(model.NeckActiveHabit)).
+	if err := e.DB.Model(new(model.NeckActiveHabit)).
 		Where("pasture_id = ?", pastureId).
 		Where("id IN (?)", processIds).
 		Where("score BETWEEN ? AND ?", model.MinScore, model.MaxScore).
 		Order("neck_ring_number,heat_date,frameid").
 		Find(&newNeckActiveHabitList).Error; err != nil {
-		return xerr.WithStack(err)
+		zaplog.Error("HealthWarning", zap.Any("error", err), zap.Any("processIds", processIds))
 	}
 
 	var (
@@ -44,10 +45,8 @@ func (e *Entry) HealthWarning(pastureId int64, processIds []int64) (err error) {
 		}
 	}
 	if len(healthWarningList) > 0 {
-		if err = e.DB.Create(&healthWarningList).Error; err != nil {
-			return xerr.WithStack(err)
+		if err := e.DB.Create(&healthWarningList).Error; err != nil {
+			zaplog.Error("HealthWarning", zap.Any("error", err), zap.Any("healthWarningList", healthWarningList))
 		}
 	}
-
-	return nil
 }

module/crontab/neck_ring_calculate.go

@@ -53,61 +53,28 @@ func (e *Entry) EntryUpdateActiveHabit(pastureId int64) (err error) {
 		return nil
 	}
 
-	if err = e.WeeklyUpdateActiveHabit(pastureId, processIds, xToday); err != nil {
-		zaplog.Error("NeckRingCalculate", zap.Any("WeeklyUpdateActiveHabit", err), zap.Any("xToday", xToday))
-	}
+	e.WeeklyUpdateActiveHabit(pastureId, processIds, xToday)
 
 	// 二次更新滤波
-	if err = e.SecondUpdateChangeFilter(pastureId, xToday); err != nil {
-		zaplog.Error("NeckRingCalculate", zap.Any("SecondUpdateChangeFilter", err), zap.Any("xToday", xToday))
-	}
+	e.SecondUpdateChangeFilter(pastureId, xToday)
 
 	// 活动量校正系数和健康评分
-	if err = e.FilterCorrectAndScoreUpdate(pastureId, processIds, xToday); err != nil {
-		zaplog.Error("NeckRingCalculate", zap.Any("ActivityVolumeChanges", err), zap.Any("xToday", xToday))
-	}
+	e.FilterCorrectAndScoreUpdate(pastureId, processIds, xToday)
 
-	if err = e.DB.Model(new(model.NeckActiveHabit)).
-		Where("id IN (?)", processIds).
-		Where("pasture_id = ?", pastureId).
-		Where("is_show = ?", pasturePb.IsShow_No).
-		Where("change_filter = ?", model.InitChangeFilter).
-		Updates(map[string]interface{}{
-			"change_filter": model.DefaultChangeFilter,
-			"rumina_filter": model.DefaultRuminaFilter,
-			"chew_filter":   model.DefaultChewFilter,
-		}).Error; err != nil {
-		zaplog.Error("EntryUpdateActiveHabit", zap.Any("change_filter", err), zap.Any("xToday", xToday))
-	}
+	// 更新 ChangeFilter
+	e.UpdateChangeFilter(pastureId, processIds)
 
-	if err = e.DB.Model(new(model.NeckActiveHabit)).
-		Where("id IN (?)", processIds).
-		Where("pasture_id = ?", pastureId).
-		Where("change_filter < ?", 0).
-		Where("filter_correct < ?", model.DefaultFilterCorrect).
-		Updates(map[string]interface{}{
-			"filter_correct": model.DefaultFilterCorrect,
-		}).Error; err != nil {
-		zaplog.Error("EntryUpdateActiveHabit", zap.Any("filter_correct", err), zap.Any("xToday", xToday))
-	}
+	// 更新 FilterCorrect
+	e.UpdateFilterCorrect(pastureId, processIds)
 
 	// 插入群体校正表
-	if err = e.UpdateChangeAdJust(pastureId, xToday); err != nil {
-		zaplog.Error("EntryUpdateActiveHabit", zap.Any("UpdateChangeAdJust", err), zap.Any("xToday", xToday))
-	}
+	e.UpdateChangeAdJust(pastureId, xToday)
 
 	// 更新所有的显示状态为否的记录为是
-	if err = e.DB.Model(new(model.NeckActiveHabit)).
-		Where("id IN (?)", processIds).
-		Where("pasture_id = ?", pastureId).
-		Update("is_show", pasturePb.IsShow_Ok).Error; err != nil {
-		zaplog.Error("UpdateChangeAdJust-2", zap.Any("error", err), zap.Any("xToday", xToday))
-	}
+	e.UpdateIsShow(pastureId, processIds)
 
 	// 健康预警
-	if err = e.HealthWarning(pastureId, processIds); err != nil {
-		zaplog.Error("EntryUpdateActiveHabit", zap.Any("HealthWarning", err))
-	}
+	e.HealthWarning(pastureId, processIds)
 	return nil
 }
 
@@ -118,7 +85,7 @@ func (e *Entry) FirstFilterUpdate(pastureId int64, xToDay *XToday) (processIds [
 		Where("heat_date >= ?", "2025-03-01").
 		Where("pasture_id = ?", pastureId).
 		Where("is_show = ?", pasturePb.IsShow_No).
-		Where("record_count = ?", model.DefaultRecordCount).
+		//Where("record_count = ?", model.DefaultRecordCount).
 		Where(e.DB.Where("high >= ?", xToDay.High).Or("rumina >= ?", xToDay.Rumina)).
 		Where("cow_id > ?", 0).
 		Order("heat_date,neck_ring_number,frameid").
@@ -142,6 +109,12 @@ func (e *Entry) FirstFilterUpdate(pastureId int64, xToDay *XToday) (processIds [
 			continue
 		}
 
+		// 8小时数据不全的不参与滤波
+		activeTime, _ := time.Parse(model.LayoutTime, v.ActiveTime)
+		if v.RecordCount != model.DefaultRecordCount && activeTime.Sub(time.Now()).Hours() < 8 {
+			continue
+		}
+
 		frameId := v.Frameid
 		heatDate := v.HeatDate
 		if v.Frameid == 0 {
@@ -209,9 +182,9 @@ func (e *Entry) FirstFilterUpdate(pastureId int64, xToDay *XToday) (processIds [
 }
 
 // SecondUpdateChangeFilter 第二次更新变化趋势滤波
-func (e *Entry) SecondUpdateChangeFilter(pastureId int64, xToday *XToday) (err error) {
+func (e *Entry) SecondUpdateChangeFilter(pastureId int64, xToday *XToday) {
 	newChangeFilterList := make([]*ChangeFilterData, 0)
-	if err = e.DB.Model(new(model.NeckActiveHabit)).
+	if err := e.DB.Model(new(model.NeckActiveHabit)).
 		Select("id", "neck_ring_number", "change_high", "change_filter", "rumina_filter", "change_rumina",
 			"chew_filter", "change_chew", "heat_date", "frameid", "IF(lact = 0, 0.8, 1) as xlc_dis_count").
 		Where("heat_date >= ?", time.Now().AddDate(0, 0, -2).Format(model.LayoutDate2)).
@@ -221,7 +194,8 @@ func (e *Entry) SecondUpdateChangeFilter(pastureId int64, xToday *XToday) (err e
 		Order("neck_ring_number,heat_date,frameid").
 		Limit(int(defaultLimit)).
 		Find(&newChangeFilterList).Error; err != nil {
-		return xerr.WithStack(err)
+		zaplog.Error("SecondUpdateChangeFilter", zap.Any("error", err), zap.Any("xToday", xToday))
+		return
 	}
 
 	for _, v := range newChangeFilterList {
@@ -279,7 +253,7 @@ func (e *Entry) SecondUpdateChangeFilter(pastureId int64, xToday *XToday) (err e
 		}
 
 		secondFilterData.ChewFilter = int32(math.Min(50, float64(secondFilterData.ChewFilter)))
-		if err = e.DB.Model(new(model.NeckActiveHabit)).
+		if err := e.DB.Model(new(model.NeckActiveHabit)).
 			Select("change_filter", "rumina_filter", "chew_filter").
 			Where("id = ?", v.Id).
 			Updates(map[string]interface{}{
@@ -287,14 +261,13 @@ func (e *Entry) SecondUpdateChangeFilter(pastureId int64, xToday *XToday) (err e
 				"rumina_filter": secondFilterData.RuminaFilter,
 				"chew_filter":   secondFilterData.ChewFilter,
 			}).Error; err != nil {
-			zaplog.Error("SecondUpdateChangeFilter-1", zap.Any("error", err), zap.Any("xToday", xToday))
+			zaplog.Error("SecondUpdateChangeFilter", zap.Any("error", err), zap.Any("secondFilterData", secondFilterData))
 		}
 	}
-	return nil
 }
 
 // FilterCorrectAndScoreUpdate 计算活动量变化趋势校正值（活跃度校正）和健康评分
-func (e *Entry) FilterCorrectAndScoreUpdate(pastureId int64, processIds []int64, xToday *XToday) error {
+func (e *Entry) FilterCorrectAndScoreUpdate(pastureId int64, processIds []int64, xToday *XToday) {
 	beginDayDate := time.Now()
 	before7DayDate := beginDayDate.AddDate(0, 0, -7).Format(model.LayoutDate2)
 	before1DayDate := beginDayDate.AddDate(0, 0, -1).Format(model.LayoutDate2)
@@ -302,12 +275,10 @@ func (e *Entry) FilterCorrectAndScoreUpdate(pastureId int64, processIds []int64,
 	neckActiveHabitList := make([]*model.NeckActiveHabit, 0)
 	if err := e.DB.Model(new(model.NeckActiveHabit)).
 		Where("id IN (?)", processIds).
-		//Where("heat_date >= ?", before1DayDate).
 		Where("pasture_id = ?", pastureId).
-		Where(e.DB.Where("high > ?", xToday.High).Or("rumina > ?", xToday.Rumina)).
 		Find(&neckActiveHabitList).Error; err != nil {
 		zaplog.Error("ActivityVolumeChanges-1", zap.Any("error", err), zap.Any("xToday", xToday))
-		return xerr.WithStack(err)
+		return
 	}
 
 	for _, v := range neckActiveHabitList {
@@ -346,11 +317,38 @@ func (e *Entry) FilterCorrectAndScoreUpdate(pastureId int64, processIds []int64,
 			continue
 		}
 	}
-	return nil
+}
+
+func (e *Entry) UpdateChangeFilter(pastureId int64, processIds []int64) {
+	if err := e.DB.Model(new(model.NeckActiveHabit)).
+		Where("id IN (?)", processIds).
+		Where("pasture_id = ?", pastureId).
+		Where("is_show = ?", pasturePb.IsShow_No).
+		Where("change_filter = ?", model.InitChangeFilter).
+		Updates(map[string]interface{}{
+			"change_filter": model.DefaultChangeFilter,
+			"rumina_filter": model.DefaultRuminaFilter,
+			"chew_filter":   model.DefaultChewFilter,
+		}).Error; err != nil {
+		zaplog.Error("UpdateChangeFilter", zap.Any("change_filter", err))
+	}
+}
+
+func (e *Entry) UpdateFilterCorrect(pastureId int64, processIds []int64) {
+	if err := e.DB.Model(new(model.NeckActiveHabit)).
+		Where("id IN (?)", processIds).
+		Where("pasture_id = ?", pastureId).
+		Where("change_filter < ?", 0).
+		Where("filter_correct < ?", model.DefaultFilterCorrect).
+		Updates(map[string]interface{}{
+			"filter_correct": model.DefaultFilterCorrect,
+		}).Error; err != nil {
+		zaplog.Error("UpdateFilterCorrect", zap.Any("filter_correct", err))
+	}
 }
 
 // UpdateChangeAdJust 更新群体校正数据
-func (e *Entry) UpdateChangeAdJust(pastureId int64, xToday *XToday) error {
+func (e *Entry) UpdateChangeAdJust(pastureId int64, xToday *XToday) {
 	/*-- 插入群体校正表
 	INSERT INTO data_bar_change(heatdate, frameid, intCurBar, intCurBarName, nb, highchange, changefilter)
 	SELECT h.heatdate, h.frameid, c.intCurBar, c.intCurBarName, COUNT(*) nb, ROUND(AVG(h.highchange)) highchange, ROUND(AVG(h.changefilter) ) changefilter
@@ -374,11 +372,10 @@ func (e *Entry) UpdateChangeAdJust(pastureId int64, xToday *XToday) error {
 		Where("h.pasture_id = ?", pastureId).
 		Where("h.heat_date >= ?", oneDayAgo).
 		Where("h.cow_id >= ?", 0).
-		Where("h.cow_id >= ?", 0).
 		Group("h.heat_date, h.frameid, c.pen_id").
 		Order("h.heat_date, h.frameid, c.pen_name").
 		Find(&res).Error; err != nil {
-		return xerr.WithStack(err)
+		zaplog.Error("UpdateChangeAdJust", zap.Any("error", err), zap.Any("xToday", xToday))
 	}
 
 	// todo ABS(cg.changefilter)>=10;
@@ -392,8 +389,15 @@ func (e *Entry) UpdateChangeAdJust(pastureId int64, xToday *XToday) error {
 			zaplog.Error("UpdateChangeAdJust-1", zap.Any("error", err), zap.Any("xToday", xToday))
 		}
 	}
+}
 
-	return nil
+func (e *Entry) UpdateIsShow(pastureId int64, processIds []int64) {
+	if err := e.DB.Model(new(model.NeckActiveHabit)).
+		Where("id IN (?)", processIds).
+		Where("pasture_id = ?", pastureId).
+		Update("is_show", pasturePb.IsShow_Ok).Error; err != nil {
+		zaplog.Error("UpdateChangeAdJust-2", zap.Any("error", err))
+	}
 }
 
 func (e *Entry) XToday(pastureId int64) (*XToday, error) {
@@ -422,24 +426,23 @@ func (e *Entry) XToday(pastureId int64) (*XToday, error) {
 }
 
 // WeeklyUpdateActiveHabit  时间点周平均值计算
-func (e *Entry) WeeklyUpdateActiveHabit(pastureId int64, processIds []int64, xToDay *XToday) (err error) {
+func (e *Entry) WeeklyUpdateActiveHabit(pastureId int64, processIds []int64, xToDay *XToday) {
 	newNeckActiveHabitList := make([]*model.NeckActiveHabit, 0)
-	if err = e.DB.Model(new(model.NeckActiveHabit)).
+	if err := e.DB.Model(new(model.NeckActiveHabit)).
 		Where("id IN (?)", processIds).
 		Order("heat_date,neck_ring_number,frameid").
 		Find(&newNeckActiveHabitList).Error; err != nil {
-		return xerr.WithStack(err)
+		zaplog.Error("WeeklyUpdateActiveHabit", zap.Any("error", err), zap.Any("processIds", processIds))
 	}
 
 	if len(newNeckActiveHabitList) <= 0 {
-		return nil
+		return
 	}
 
 	e.HabitUpdateActiveHabit(pastureId, newNeckActiveHabitList, xToDay)
 	e.SumUpdateActiveHabit(pastureId, newNeckActiveHabitList, xToDay)
 	e.ActiveChange(pastureId, processIds, xToDay)
 	e.Before3DaysNeckActiveHabit(pastureId, processIds)
-	return nil
 }
 
 func (e *Entry) HabitUpdateActiveHabit(pastureId int64, newNeckActiveHabitList []*model.NeckActiveHabit, xToDay *XToday) {