drone: update

.drone.yml

@@ -30,7 +30,7 @@ steps:
         from_secret: aliyuncs_password
       repo: registry.cn-hangzhou.aliyuncs.com/kpt-event/kpt-pasture
       registry: registry.cn-hangzhou.aliyuncs.com
-      tags: [ test ]
+      tags: [ http-test,mqtt-test ]
 
 trigger:
   branch:

README.md

@@ -41,7 +41,7 @@ todo列表:
 - [x] 所有事件录入梳理【批量录入，excel导入，信息人员与操作人员统一规范】
 - [ ] 药品优化成药品名称关联生产商
 - [x] 框架logrus日志优化【未按照指定天数的日志自动删除，待验证】
-- [ ] 犊牛的牛只品种是根据母牛的品种来确定，还是根据公牛来确定？【目前是根据母牛品种来确定】
+- [x] 犊牛的牛只品种是根据母牛的品种来确定，还是根据公牛来确定？【目前是根据母牛品种来确定】
 
 脖环发情算法梳理:
-- [ ] 处理异常上报数据（frameid > 12）
+- [x] 处理异常上报数据（frameid > 12）

cmd/mqtt.go

@@ -10,6 +10,6 @@ var MqttCmd = &cobra.Command{
 	Use:   "mqtt",
 	Short: "mqtt server",
 	Run: func(cmd *cobra.Command, args []string) {
-		dep.DIMqttService().MqttServer.Run()
+		dep.DIMqttService()
 	},
 }

dep/dep.go

@@ -7,7 +7,6 @@ import (
 	"kpt-pasture/module/crontab"
 	moduleMqtt "kpt-pasture/module/mqtt"
 	"kpt-pasture/service/asynqsvc"
-	"kpt-pasture/service/mqtt"
 	"kpt-pasture/service/redis"
 	"kpt-pasture/service/sso"
 	"kpt-pasture/service/wechat"
@@ -38,7 +37,6 @@ func Options() []di.HubOption {
 		asynq.Module,
 		redis.Module,
 		crontab.Module,
-		mqtt.Module,
 		moduleMqtt.Module,
 	}
 }

dep/di_mqtt.go

@@ -1,22 +1,34 @@
 package dep
 
 import (
+	"kpt-pasture/config"
+	mqttHandle "kpt-pasture/module/mqtt"
 	"kpt-pasture/service/mqtt"
 
 	"go.uber.org/dig"
 )
 
-func DIMqttService() (out MqttDependency) {
+func DIMqttService() (out mqtt.IMqttServer) {
 	container := DI()
-	if err := container.Invoke(func(c MqttDependency) { out = c }); err != nil {
+	if err := container.Provide(MqttHandel); err != nil {
+		panic(err)
+	}
+	if err := container.Invoke(func(c mqtt.IMqttServer) { out = c }); err != nil {
 		panic(err)
 	}
 	return
 }
 
-// MqttDependency 依赖注入结构体
+// MqttHandel 相关消费
+func MqttHandel(dep MqttDependency) mqtt.IMqttServer {
+	cfg := config.Options()
+	sev := mqtt.NewServer(cfg)
+	sev.Run(dep.MqttHub)
+	return sev
+}
+
 type MqttDependency struct {
 	dig.In
 
-	MqttServer mqtt.IMqttServer
+	MqttHub mqttHandle.Entry // 处理数据
 }

go.mod

@@ -3,7 +3,7 @@ module kpt-pasture
 go 1.17
 
 require (
-	gitee.com/xuyiping_admin/go_proto v0.0.0-20250103030611-2b0c271353c7
+	gitee.com/xuyiping_admin/go_proto v0.0.0-20250103092939-8816c3aabc06
 	gitee.com/xuyiping_admin/pkg v0.0.0-20241108060137-caea58c59f5b
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/eclipse/paho.mqtt.golang v1.4.3

go.sum

@@ -110,6 +110,8 @@ gitee.com/xuyiping_admin/go_proto v0.0.0-20250102080429-bbf8aeeb608c h1:Fqi07Tnc
 gitee.com/xuyiping_admin/go_proto v0.0.0-20250102080429-bbf8aeeb608c/go.mod h1:BKrFW6YLDectlQcQk3FYKBeXvjEiodAKJ5rq7O/QiPE=
 gitee.com/xuyiping_admin/go_proto v0.0.0-20250103030611-2b0c271353c7 h1:mk2G7/x2g2WLbCFw7ThZ7F5HOJ9AfAidG7ilhit67bM=
 gitee.com/xuyiping_admin/go_proto v0.0.0-20250103030611-2b0c271353c7/go.mod h1:BKrFW6YLDectlQcQk3FYKBeXvjEiodAKJ5rq7O/QiPE=
+gitee.com/xuyiping_admin/go_proto v0.0.0-20250103092939-8816c3aabc06 h1:9pYtEuYYvFb1Zak/bMe7MFzElFlwya45G7mrco5bNp0=
+gitee.com/xuyiping_admin/go_proto v0.0.0-20250103092939-8816c3aabc06/go.mod h1:BKrFW6YLDectlQcQk3FYKBeXvjEiodAKJ5rq7O/QiPE=
 gitee.com/xuyiping_admin/pkg v0.0.0-20241108060137-caea58c59f5b h1:w05MxH7yqveRlaRbxHhbif5YjPrJFodRPfOjYhXn7Zk=
 gitee.com/xuyiping_admin/pkg v0.0.0-20241108060137-caea58c59f5b/go.mod h1:8tF25X6pE9WkFCczlNAC0K2mrjwKvhhp02I7o0HtDxY=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=

http/handler/event/event_breed.go

@@ -211,7 +211,7 @@ func EstrusBatchMating(c *gin.Context) {
 		valid.Field(&req.CowIds, valid.Required),
 		valid.Field(&req.BullNumber, valid.Required),
 		valid.Field(&req.OperationId, valid.Required),
-		valid.Field(&req.Consumption, valid.Required),
+		valid.Field(&req.Quantity, valid.Required),
 	); err != nil {
 		apierr.AbortBadRequest(c, http.StatusBadRequest, err)
 		return

model/event_mating.go

@@ -14,6 +14,7 @@ type EventMating struct {
 	DayAge            int32                           `json:"dayAge"`
 	Lact              int32                           `json:"lact"`
 	PenId             int32                           `json:"penId"`
+	PenName           string                          `json:"penName"`
 	CowType           pasturePb.CowType_Kind          `json:"cowType"`
 	CowKind           pasturePb.CowKind_Kind          `json:"cowKind"`
 	CalvingAge        int32                           `json:"calvingAge"`
@@ -89,6 +90,7 @@ func NewEventMating(pastureId int64, cow *Cow, planDay int64, exposeEstrusType p
 		CowId:            cow.Id,
 		Lact:             cow.Lact,
 		PenId:            cow.PenId,
+		PenName:          cow.PenName,
 		CowType:          cow.CowType,
 		CowKind:          cow.CowKind,
 		CalvingAt:        cow.LastMatingAt,

model/frozen_semen_log.go

@@ -9,6 +9,7 @@ import (
 
 type FrozenSemenLog struct {
 	Id            int64  `json:"id"`
+	PastureId     int64  `json:"pastureId"`
 	BullId        string `json:"bullId"`
 	CowIds        string `json:"cowId"`
 	Quantity      int32  `json:"quantity"`
@@ -24,7 +25,7 @@ func (e *FrozenSemenLog) TableName() string {
 	return "frozen_semen_log"
 }
 
-func NewEventFrozenSemenLog(req *pasturePb.EventMating) *FrozenSemenLog {
+func NewEventFrozenSemenLog(pastureId int64, req *pasturePb.EventMating) *FrozenSemenLog {
 	cowIds := ""
 	for _, v := range req.CowIds {
 		cowIds += fmt.Sprintf("%d,", v)
@@ -33,6 +34,7 @@ func NewEventFrozenSemenLog(req *pasturePb.EventMating) *FrozenSemenLog {
 		cowIds = strings.TrimRight(cowIds, ",")
 	}
 	return &FrozenSemenLog{
+		PastureId:     pastureId,
 		BullId:        req.FrozenSemenNumber,
 		CowIds:        cowIds,
 		OperationId:   int64(req.OperationId),

module/backend/event_breed.go

@@ -421,7 +421,7 @@ func (s *StoreEntry) EstrusBatchMating(ctx context.Context, req *pasturePb.Event
 	eventMating := &pasturePb.EventMating{
 		CowIds:            cowIds,
 		FrozenSemenNumber: req.BullNumber,
-		FrozenSemenCount:  req.Consumption,
+		FrozenSemenCount:  req.Quantity,
 		OperationId:       req.OperationId,
 		Remarks:           req.Remarks,
 		MatingAt:          req.MatingAt,
@@ -555,7 +555,7 @@ func (s *StoreEntry) MatingCreate(ctx context.Context, req *pasturePb.EventMatin
 		}
 
 		// 创建冻精使用记录日志
-		itemFrozenSemenLog := model.NewEventFrozenSemenLog(req)
+		itemFrozenSemenLog := model.NewEventFrozenSemenLog(eventCheckModel.FrozenSemen.PastureId, req)
 		if err = tx.Create(itemFrozenSemenLog).Error; err != nil {
 			return xerr.WithStack(err)
 		}

module/crontab/interface.go

@@ -10,9 +10,7 @@ import (
 	"go.uber.org/dig"
 )
 
-var Module = di.Options(
-	di.Provide(NewCrontab),
-)
+var Module = di.Options(di.Provide(NewCrontab))
 
 type Entry struct {
 	dig.In

module/mqtt/handle.go

@@ -1,12 +1,15 @@
 package mqtt
 
 import (
+	"encoding/json"
 	"kpt-pasture/model"
 	"kpt-pasture/util"
 	"strconv"
 	"strings"
 	"sync"
 
+	pasturePb "gitee.com/xuyiping_admin/go_proto/proto/go/backend/cow"
+
 	"gitee.com/xuyiping_admin/pkg/logger/zaplog"
 	"gitee.com/xuyiping_admin/pkg/xerr"
 	"go.uber.org/zap"
@@ -82,7 +85,7 @@ func (e *Entry) NeckRingHandle(data []byte) {
 
 func (e *Entry) CreatedData(DSMLog *DataInsertNeckRingLog) error {
 	if err := e.DB.Transaction(func(tx *gorm.DB) error {
-		if len(DSMLog.NeckRingUnRegisterData) > 0 {
+		if len(DSMLog.NeckRingErrorData) > 0 {
 			if err := e.DB.Create(DSMLog.NeckRingErrorData).Error; err != nil {
 				return xerr.WithStack(err)
 			}
@@ -280,3 +283,38 @@ func (e *Entry) MsgDataFormat(msg []byte) *model.NeckRingOriginal {
 		Nccid:    nccId,
 	}
 }
+
+func (e *Entry) MsgDataFormat2(msg []byte) *model.NeckRingOriginal {
+	neckLog := &Behavior{}
+	if err := json.Unmarshal(msg, neckLog); 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)
+		activeDateTimeType := pasturePb.ActiveTimeType_Twenty_Minutes
+		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,
+		}
+	}
+	return nil
+}

module/mqtt/model.go

@@ -1,6 +1,6 @@
 package mqtt
 
-type Behavior struct {
+type NewBehavior struct {
 	SoftVer  int32   `json:"soft_ver"`
 	UUID     string  `json:"uuid"`
 	FrameId  int32   `json:"frameid"`
@@ -17,3 +17,29 @@ type Behavior struct {
 	Other    int32   `json:"other"`
 	Remain   int32   `json:"Remain"`
 }
+
+type Behavior struct {
+	UUID      string `json:"uuid"`
+	ECowId    string `json:"ecowid"`
+	FrameId   int32  `json:"frameid"`
+	High      int32  `json:"High"`
+	Intake    int32  `json:"Intake"`
+	RuMina    int32  `json:"Rumina"`
+	Other     int32  `json:"Other"`
+	Activitys int32  `json:"activitys"`
+	Inactive  int32  `json:"inactive"`
+	Sver      int32  `json:"Sver"`
+	Remain    int32  `json:"Remain"`
+	RFRssi    int32  `json:"RFRssi"`
+	STATUS    int32  `json:"STATUS"`
+	BAT       int32  `json:"BAT"`
+	Imei      string `json:"imei"`
+	Gasp      int32  `json:"gasp"`
+}
+
+type NeckRingWrapper struct {
+	Type     string `json:"type"`
+	NeckRing struct {
+		NeckPck []*Behavior `json:"neck"` // neck_pck neck
+	} `json:"NeckRing"`
+}

service/mqtt/interface.go

@@ -3,35 +3,31 @@ package mqtt
 import (
 	"fmt"
 	"kpt-pasture/config"
-	mqtt2 "kpt-pasture/module/mqtt"
 	"kpt-pasture/util"
 	"os"
 	"os/signal"
 	"syscall"
 	"time"
 
-	"go.uber.org/dig"
+	handleMqtt "kpt-pasture/module/mqtt"
 
-	"gitee.com/xuyiping_admin/pkg/di"
 	"gitee.com/xuyiping_admin/pkg/logger/zaplog"
 	golangMqtt "github.com/eclipse/paho.mqtt.golang"
 	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"go.uber.org/zap"
 )
 
-var Module = di.Options(di.Provide(NewServer))
+//var Module = di.Options(di.Provide(NewServer))
 
 type IMqttClient struct {
-	dig.In
 	golangMqtt.Client
-	Config  config.MqttSetting
-	MqttHub mqtt2.Entry // 处理数据
+	Config config.MqttSetting
 }
 
 type IMqttServer interface {
 	Consumer(func([]byte))
 	Producer(top string, qos int32, data []byte) error
-	Run()
+	Run(handleMqtt.Entry)
 	Close()
 }
 
@@ -86,7 +82,7 @@ func (s *IMqttClient) Close() {
 	}
 }
 
-func (s *IMqttClient) Run() {
+func (s *IMqttClient) Run(enter handleMqtt.Entry) {
 	// 设置信号监听以优雅关闭服务器
 	stop := make(chan os.Signal, 1)
 	signal.Notify(stop, os.Kill, os.Interrupt, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
@@ -114,5 +110,5 @@ func (s *IMqttClient) Run() {
 	}
 
 	// 启动数据处理
-	s.Consumer(s.MqttHub.NeckRingHandle)
+	s.Consumer(enter.NeckRingHandle)
 }