kpt-pasture/module/crontab/pen_behavior.go

package crontab

import (
	"fmt"
	"kpt-pasture/model"
	"kpt-pasture/util"
	"time"

	"gitee.com/xuyiping_admin/pkg/xerr"

	pasturePb "gitee.com/xuyiping_admin/go_proto/proto/go/backend/cow"

	"gitee.com/xuyiping_admin/pkg/logger/zaplog"
	"go.uber.org/zap"
)

func (e *Entry) UpdatePenBehavior() error {
	pastureList := e.FindPastureList()
	if pastureList == nil || len(pastureList) == 0 {
		return nil
	}

	for _, pasture := range pastureList {
		conf, err := e.GetSystemNeckRingConfigure(pasture.Id, model.MaxPenBehavior)
		if err != nil {
			zaplog.Error("UpdatePenBehavior", zap.Any("pasture", pasture), zap.Any("err", err))
			continue
		}
		e.PenBehaviorEnter(pasture.Id, conf.Value)
	}
	return nil
}

// PenBehaviorEnter 栏舍行为曲线 对数据进行查缺补漏
func (e *Entry) PenBehaviorEnter(pastureId int64, maxValue int64) {
	var minHeatDate string
	if err := e.DB.Model(new(model.NeckRingOriginal)).
		Select("min(active_date) as min_heat_date").
		Where("pasture_id = ?", pastureId).
		Where("id >= ?", maxValue).
		Scan(&minHeatDate).Error; err != nil {
		zaplog.Error("PenBehaviorEnter", zap.Any("pastureId", pastureId), zap.Any("err", err))
		return
	}
	e.PenBehavior(pastureId, minHeatDate)
	var maxId int64
	if err := e.DB.Model(new(model.NeckRingOriginal)).
		Select("MAX(id) as id").
		Where("pasture_id = ?", pastureId).
		Where("id >= ?", maxValue).
		Where("active_date = ?", minHeatDate).
		Scan(&maxId).Error; err != nil {
		zaplog.Error("PenBehaviorEnter", zap.Any("pastureId", pastureId), zap.Any("err", err))
	}
	if maxId > maxValue {
		if err := e.UpdateSystemNeckRingConfigure(pastureId, model.MaxPenBehavior, maxId); err != nil {
			zaplog.Error("UpdateSystemNeckRingConfigure", zap.Any("err", err))
		}
	}

	e.UpdatePenBehaviorWeekData(pastureId, minHeatDate)
}

// PenBehavior 栏舍行为曲线
func (e *Entry) PenBehavior(pastureId int64, heatDate string) {
	frameIds := util.FrameIdSlice
	penBehaviorList := make([]*model.PenBehaviorData, 0)
	for _, frameId := range frameIds {
		penBehaviorModel, err := e.getNeckRingOriginalList(pastureId, heatDate, frameId)
		if err != nil {
			zaplog.Error("PenBehavior",
				zap.Any("pasture", pastureId),
				zap.Any("frameId", frameId),
				zap.Any("heatDate", heatDate),
				zap.Any("err", err),
			)
			continue
		}
		if penBehaviorModel != nil {
			penBehaviorList = append(penBehaviorList, penBehaviorModel)
		}
	}

	if len(penBehaviorList) <= 0 {
		return
	}

	// 2. 保存数据
	if err := e.savePenBehaviorData(penBehaviorList); err != nil {
		zaplog.Error("PenBehavior", zap.Any("penBehaviorList", penBehaviorList), zap.Any("err", err))
		return
	}
}

// getNeckRingOriginalList 获取颈环原始数据
func (e *Entry) getNeckRingOriginalList(pastureId int64, dateTime string, frameId int32) (*model.PenBehaviorData, error) {
	penBehaviorModel := &model.PenBehaviorData{}
	sql := fmt.Sprintf(`
		SELECT	bb.pasture_id, bb.heat_date, bb.frameid,
		bb.pen_id, bb.pen_name,bb.cow_count, bb.avg_high, 
		bb.sum_rumina, bb.sum_intake, bb.sum_rest, bb.sum_gasp, 
		ROUND(bb.sum_rumina/bb.cow_count*100, 0) rumina_rate , 
		ROUND(bb.sum_intake/bb.cow_count*100, 0) intake_rate, 
		ROUND(bb.sum_rest/bb.cow_count*100, 0) rest_rate, 
		ROUND(bb.sum_gasp/bb.cow_count*100, 0) gasp_rate
		FROM (
			SELECT aa.pasture_id, aa.pen_id, aa.pen_name, aa.heat_date, aa.frameid, 
			COUNT(1) cow_count, 
			ROUND(AVG(aa.high), 0) avg_high, 
			SUM(IF(aa.rumina>=8, 1, 0)) sum_rumina, 
			SUM(IF(aa.intake>=8, 1, 0)) sum_intake,  
			SUM(IF(aa.inactive>=8, 1, 0)) sum_rest,  
			SUM(IF(aa.gasp>=8, 1, 0) ) sum_gasp  
			FROM (
				SELECT c.pasture_id, c.ear_number, c.pen_id, c.pen_name, h.neck_ring_number, h.active_date as heat_date, 
				h.frameid, h.high, h.rumina, h.intake, h.inactive, h.gasp 
				FROM neck_ring_original h JOIN cow c ON h.pasture_id=c.pasture_id AND h.neck_ring_number=c.neck_ring_number 
				WHERE h.pasture_id = %d
				AND h.active_date='%s' 
				AND h.frameid = %d
				GROUP BY h.neck_ring_number
			) aa GROUP BY aa.pen_id 
		) bb`, pastureId, dateTime, frameId)

	if err := e.DB.Raw(sql).First(penBehaviorModel).Error; err != nil {
		return nil, xerr.WithStack(err)
	}
	return penBehaviorModel, nil
}

// savePenBehaviorData 保存栏舍行为数据
func (e *Entry) savePenBehaviorData(penDataList []*model.PenBehaviorData) error {
	for _, data := range penDataList {
		// 构建活动时间
		activeTime := e.calculateActiveTime(data.HeatDate, data.Frameid)
		// 构建保存数据
		penBehavior := model.NewPenBehavior(data, activeTime)
		if e.isExistByPenBehavior(data.PastureId, data.HeatDate, data.PenId, data.Frameid) {
			historyData := e.findPenBehavior(data.PastureId, data.HeatDate, data.PenId, data.Frameid)
			if historyData == nil || historyData.Id <= 0 {
				continue
			}

			if err := e.DB.Model(new(model.PenBehavior)).
				Where("id = ?", historyData.Id).
				Updates(map[string]interface{}{
					"cow_count":   data.CowCount,
					"avg_high":    data.AvgHigh,
					"sum_rumina":  data.SumRumina,
					"sum_intake":  data.SumIntake,
					"sum_rest":    data.SumRest,
					"sum_gasp":    data.SumGasp,
					"rumina_rate": data.RuminaRate,
					"intake_rate": data.IntakeRate,
					"rest_rate":   data.RestRate,
					"gasp_rate":   data.GaspRate,
				}).Error; err != nil {
				zaplog.Error("savePenBehaviorData", zap.Any("penBehavior", penBehavior), zap.Any("err", err))
			}
			continue
		}
		if err := e.DB.Model(new(model.PenBehavior)).
			Create(penBehavior).Error; err != nil {
			zaplog.Error("savePenBehaviorData", zap.Any("penBehavior", penBehavior), zap.Any("err", err))
		}
	}
	return nil
}

func (e *Entry) UpdatePenBehaviorWeekData(pastureId int64, dateTime string) {
	dateTime1, _ := util.TimeParseLocal(model.LayoutDate2, dateTime)
	startTime := dateTime1.AddDate(0, 0, -7).Format(model.LayoutDate2)
	endTime := dateTime1.AddDate(0, 0, -1).Format(model.LayoutDate2)
	penBehaviorList, err := e.findWeekPenBehaviorList(pastureId, dateTime, startTime, endTime)
	if err != nil {
		zaplog.Error("UpdatePenBehaviorWeekData",
			zap.Any("err", err),
			zap.Any("pastureId", pastureId),
			zap.Any("dateTime", dateTime),
			zap.Any("startTime", startTime),
			zap.Any("endTime", endTime),
		)
		return
	}
	if len(penBehaviorList) == 0 {
		return
	}

	// 处理每个日期和frameid的数据
	for _, item := range penBehaviorList {
		if err = e.DB.Model(new(model.PenBehavior)).
			Where("id = ?", item.Id).
			Updates(map[string]interface{}{
				"week_rumina_rate": item.WeekRuminaRate,
				"week_intake_rate": item.WeekIntakeRate,
				"week_rest_rate":   item.WeekRestRate,
				"week_gasp_rate":   item.WeekGaspRate,
				"rumina_std":       item.RuminaStd,
				"intake_std":       item.IntakeStd,
				"rest_std":         item.RestStd,
				"gasp_std":         item.GaspStd,
				"is_show":          pasturePb.IsShow_Ok,
			}).Error; err != nil {
			zaplog.Error("UpdatePenBehaviorWeekData", zap.Error(err))
		}
	}
}

// calculateActiveTime 计算活动时间
func (e *Entry) calculateActiveTime(heatDate string, frameid int32) string {
	// 计算小时和分钟
	hour := (frameid / 10) * 2
	minute := (frameid%10)*20 - 1
	if minute < 0 {
		minute = 0
	}

	baseDate, err := time.Parse(model.LayoutDate2, heatDate)
	if err != nil {
		zaplog.Error("PenBehavior", zap.Any("calculateActiveTime", err))
		return ""
	}
	baseTime := time.Date(baseDate.Year(), baseDate.Month(), baseDate.Day(), int(hour), 0, 0, 0, baseDate.Location())
	finalTime := baseTime.Add(time.Duration(minute) * time.Minute)
	// 构建时间字符串
	return finalTime.Format(model.LayoutTime)
}

// isExistByPenBehavior 是否存在
func (e *Entry) isExistByPenBehavior(pastureId int64, heatDate string, penId int32, frameid int32) bool {
	var count int64
	if err := e.DB.Model(new(model.PenBehavior)).
		Where("pasture_id = ? AND heat_date = ? AND frameid = ? AND pen_id = ?", pastureId, heatDate, frameid, penId).
		Count(&count).Error; err != nil {
		return false
	}
	return count > 0
}

func (e *Entry) findPenBehavior(pastureId int64, heatDate string, penId int32, frameid int32) *model.PenBehavior {
	res := &model.PenBehavior{}
	if err := e.DB.Model(new(model.PenBehavior)).
		Where("pasture_id = ? AND heat_date = ? AND frameid = ? AND pen_id = ?", pastureId, heatDate, penId, frameid).
		First(res).Error; err != nil {
		return nil
	}
	return res
}

func (e *Entry) findWeekPenBehaviorList(pastureId int64, heatDate, startTime, endTime string) ([]*model.PenBehavior, error) {
	penBehaviorList := make([]*model.PenBehavior, 0)
	sql := fmt.Sprintf(`
		SELECT b1.id, 
		ROUND(AVG(b0.rumina_rate)) week_rumina_rate,
		ROUND(STD(b0.rumina_rate)) rumina_std, 
		ROUND(AVG( b0.intake_rate)) week_intake_rate, 
		ROUND(STD( b0.intake_rate)) intake_std, 
		ROUND(AVG( b0.rest_rate)) week_rest_rate, 
		ROUND(STD( b0.rest_rate)) rest_std, 
		ROUND(AVG( b0.gasp_rate)) week_gasp_rate, 
		ROUND(STD( b0.gasp_rate)) gasp_std
		FROM pen_behavior b1 JOIN pen_behavior b0 
		ON b1.pen_id=b0.pen_id AND b1.heat_date='%s' 
		AND b1.frameid=b0.frameid AND b0.heat_date BETWEEN '%s' AND '%s'
		WHERE  b1.cow_count>= %d AND b1.pasture_id = %d
		GROUP BY b1.id
	`, heatDate, startTime, endTime, model.PenBehaviorMinCowCount, pastureId)
	if err := e.DB.Raw(sql).Find(&penBehaviorList).Error; err != nil {
		return nil, xerr.WithStack(err)
	}
	return penBehaviorList, nil
}