refactor: 设备原数据查询修改,数据消费修改

@@ -50,7 +50,7 @@ func AutoFlushDeviceDailyRunningRecord(ctx context.Context) error {
 		// 更新设备效率 OEE   = tu * pu * qu
 
 		if _, err := deviceDailyRunningRecordRepository.Save(v); err != nil {
-			log.Logger.Error(err.Error())
+			log.Logger.Error(err.Error(), map[string]interface{}{"record": v})
 			continue
 		} else {
 			log.Logger.Debug(fmt.Sprintf("【定时刷新设备每日运行记录】 刷新记录 %v", v))

+package dto
+
+import "time"
+
+type DeviceCollectionDto struct {
+	// 数据采集ID
+	DeviceCollectionId int64 `json:"deviceCollectionId,string"`
+	// 车间名
+	WorkShopName string `json:"workShopName"`
+	// 采集时间
+	CollectionTime time.Time `json:"collectionTime"`
+	// 设备名
+	DeviceSn string `json:"deviceSn"`
+	// 设备类型
+	DeviceType string `json:"deviceType"`
+	// 启动状态 1-启动 0-停止
+	StartupStatus int64 `json:"startupStatus"`
+	// 通讯状态 1-通讯正常 0-设备未上电或与采集端通讯故障
+	ComStatus int64 `json:"comStatus"`
+	// 设备数据值
+	Values map[string]interface{} `json:"values"`
+}

@@ -2,8 +2,10 @@ package query
 
 import (
 	"fmt"
+	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/log"
 	"reflect"
 	"strings"
+	"time"
 
 	"github.com/beego/beego/v2/core/validation"
 )
@@ -17,20 +19,47 @@ type ListDeviceCollectionQuery struct {
 	DeviceType string `json:"deviceType" cname:"设备类型"`
 	// 车间名
 	WorkShopName string `json:"workShopName"`
+
+	// 开始时间
+	BeginTime string `cname:"开始时间" json:"beginTime"`
+	// 结束时间
+	EndTime string `cname:"结束时间" json:"endTime"`
+
+	// 开始时间
+	ProductBeginTime time.Time `cname:"开始时间" json:"productBeginTime"`
+	// 结束时间
+	ProductEndTime time.Time `cname:"结束时间" json:"productEndTime"`
 }
 
-func (listDeviceCollectionQuery *ListDeviceCollectionQuery) Valid(validation *validation.Validation) {
-	//validation.SetError("CustomValid", "未实现的自定义认证")
+func (cmd *ListDeviceCollectionQuery) Valid(validation *validation.Validation) {
+	var err error
+	if len(cmd.BeginTime) > 0 {
+		if cmd.ProductBeginTime, err = time.ParseInLocation("2006-01-02 15:04:05", cmd.BeginTime, time.Local); err != nil {
+			log.Logger.Error(err.Error())
+			validation.Error("开始时间有误")
+			return
+		}
+	}
+	if len(cmd.EndTime) > 0 {
+		if cmd.ProductEndTime, err = time.ParseInLocation("2006-01-02 15:04:05", cmd.EndTime, time.Local); err != nil {
+			log.Logger.Error(err.Error())
+			validation.Error("结束时间有误")
+			return
+		}
+		if cmd.ProductBeginTime.Equal(cmd.ProductEndTime) {
+			cmd.ProductEndTime = cmd.ProductEndTime.Add(time.Hour * 24)
+		}
+	}
 }
 
-func (listDeviceCollectionQuery *ListDeviceCollectionQuery) ValidateQuery() error {
+func (cmd *ListDeviceCollectionQuery) ValidateQuery() error {
 	valid := validation.Validation{}
-	b, err := valid.Valid(listDeviceCollectionQuery)
+	b, err := valid.Valid(cmd)
 	if err != nil {
 		return err
 	}
 	if !b {
-		elem := reflect.TypeOf(listDeviceCollectionQuery).Elem()
+		elem := reflect.TypeOf(cmd).Elem()
 		for _, validErr := range valid.Errors {
 			field, isExist := elem.FieldByName(validErr.Field)
 			if isExist {

@@ -11,6 +11,7 @@ import (
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/domain"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/domainService"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/redis"
+	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/utils"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/log"
 	"strconv"
 	"sync"
@@ -29,6 +30,57 @@ func (deviceCollectionService *DeviceCollectionService) CreateDeviceCollection(c
 	if err := createDeviceCollectionCommand.ValidateCommand(); err != nil {
 		return nil, application.ThrowError(application.ARG_ERROR, err.Error())
 	}
+	transactionContext, err := factory.CreateTransactionContext(nil)
+	if err != nil {
+		return nil, application.ThrowError(application.TRANSACTION_ERROR, err.Error())
+	}
+	if err := transactionContext.StartTransaction(); err != nil {
+		return nil, application.ThrowError(application.TRANSACTION_ERROR, err.Error())
+	}
+	defer func() {
+		transactionContext.RollbackTransaction()
+	}()
+	newDeviceCollection := &domain.DeviceCollection{
+		//DeviceCollectionId: createDeviceCollectionCommand.DeviceCollectionId,
+		WorkShopName:   createDeviceCollectionCommand.WorkShopName,
+		DeviceType:     createDeviceCollectionCommand.DeviceType,
+		StartupStatus:  createDeviceCollectionCommand.StartupStatus,
+		DeviceSn:       createDeviceCollectionCommand.DeviceSn,
+		ComStatus:      createDeviceCollectionCommand.ComStatus,
+		CollectionTime: createDeviceCollectionCommand.CollectionTime,
+		Values:         createDeviceCollectionCommand.Values,
+	}
+	var deviceCollectionRepository domain.DeviceCollectionRepository
+	if value, err := factory.CreateDeviceCollectionRepository(map[string]interface{}{
+		"transactionContext": transactionContext,
+	}); err != nil {
+		return nil, application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
+	} else {
+		deviceCollectionRepository = value
+	}
+	if deviceCollection, err := deviceCollectionRepository.Save(newDeviceCollection); err != nil {
+		return nil, application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
+	} else {
+		//err = domainService.SendWorkshopDeviceData(deviceCollection)
+		//if err != nil {
+		//	log.Logger.Error("车间设备数据加入redis失败：" + err.Error())
+		//	return nil, application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
+		//}
+
+		if err := transactionContext.CommitTransaction(); err != nil {
+			return nil, application.ThrowError(application.TRANSACTION_ERROR, err.Error())
+		}
+		return map[string]interface{}{
+			"deviceCollection": deviceCollection,
+		}, nil
+	}
+}
+
+// 创建
+func (deviceCollectionService *DeviceCollectionService) DeviceCollection(createDeviceCollectionCommand *command.CreateDeviceCollectionCommand) (interface{}, error) {
+	if err := createDeviceCollectionCommand.ValidateCommand(); err != nil {
+		return nil, application.ThrowError(application.ARG_ERROR, err.Error())
+	}
 
 	newDeviceCollection := &domain.DeviceCollection{
 		//DeviceCollectionId: createDeviceCollectionCommand.DeviceCollectionId,
@@ -73,34 +125,31 @@ func (deviceCollectionService *DeviceCollectionService) CreateDeviceCollection(c
 	} else {
 		deviceCollectionRepository = value
 	}
-	deviceCollection, err := deviceCollectionRepository.Save(newDeviceCollection)
-	if err != nil {
-		return nil, application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
-	}
 
-	//处理设备数据
+	//计算区间的产能
 	//switch deviceCollection.DeviceType {
 	//case domain.DeviceTypeBaoXianJi, domain.DeviceTypeChuanChuanJi, domain.DeviceTypeFengKouJi, domain.DeviceTypeFengXiangJi:
-	if v, ok := deviceCollection.Values["count"]; ok {
-		curCount, errCurCount := strconv.Atoi(fmt.Sprintf("%v", v))
-		v, ok = lastDeviceCollectionRecord.Values["count"]
-		if ok {
-			lastCount, errLastCount := strconv.Atoi(fmt.Sprintf("%v", v))
-			if errLastCount == nil && errCurCount == nil && lastCount <= curCount {
-				deviceCollection.Values["count"] = curCount - lastCount
+	if v, ok := newDeviceCollection.Values["Count"]; ok {
+		newDeviceCollection.Values["total"] = v // 记录原始值
+		newDeviceCollection.Values["Count"] = 0
+		curCount, errCurCount := strconv.Atoi(utils.AssertString(v))
+		lastCount, errLastCount := strconv.Atoi(utils.AssertString(lastDeviceCollectionRecord.Values["Count"]))
+		if errLastCount == nil && errCurCount == nil && lastCount <= curCount {
+			if lastCount <= curCount {
+				newDeviceCollection.Values["Count"] = curCount - lastCount
 			} else {
-				deviceCollection.Values["count"] = 0
-				/*
-					设备统计的数量超过一定范围会重置为0，特殊处理0操作
-				*/
+				newDeviceCollection.Values["Count"] = 0
+				/*设备统计的数量超过一定范围会重置为0，特殊处理0操作*/
 				if lastCount > 10000000 && curCount < 1000 {
-					deviceCollection.Values["count"] = curCount
+					newDeviceCollection.Values["Count"] = curCount
 				}
 			}
-		} else {
-			deviceCollection.Values["count"] = 0
 		}
 	}
+	deviceCollection, err := deviceCollectionRepository.Save(newDeviceCollection)
+	if err != nil {
+		return nil, application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
+	}
 	//	break
 	//}
 	err = domainService.SendWorkshopDeviceData(deviceCollection)

+package service

@@ -28,6 +28,7 @@ type DeviceYouZhaJi2 struct {
 
 // 串串机
 type DeviceChuanChuanJi struct {
+	Total        int64  `json:"Total"` // 累计生产计数：生产统计数量
 	Count        int64  `json:"Count"` // 生产计数：生产统计数量
 	Year         int    `json:"Year"`  // 年
 	Month        int    `json:"Month"` // 月

@@ -3,11 +3,11 @@ package domain
 import (
 	"fmt"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/utils"
-	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/log"
 	"time"
 )
 
 const DefaultTimeWindow = 1
+const DefaultCollectionTimeSpan = 60 * 20
 
 // 设备每日运行记录（汇总）
 type DeviceDailyRunningRecord struct {
@@ -55,6 +55,11 @@ func (deviceDailyRunningRecord *DeviceDailyRunningRecord) Update(data map[string
 
 func (deviceDailyRunningRecord *DeviceDailyRunningRecord) AddDeviceRunningData(t time.Time, data *DeviceRunningData) {
 	deviceDailyRunningRecord.DeviceRunningRecordInfo.AddDeviceRunningData(t, data)
+	now := time.Now().Unix()
+	if t.Unix() > (now-DefaultCollectionTimeSpan) && t.Unix() < (now+DefaultCollectionTimeSpan) {
+		deviceDailyRunningRecord.UpdatedAt = t
+		return
+	}
 	deviceDailyRunningRecord.UpdatedAt = time.Now()
 }
 
@@ -144,7 +149,7 @@ func (d *DeviceRunningRecordInfo) AddTimeLineDeviceStatus(t time.Time, data *Dev
 		return
 	}
 	key := fmt.Sprintf("%v", t.Local().Hour())
-	log.Logger.Debug(fmt.Sprintf("time:%v hour:%v", t, key))
+	//log.Logger.Debug(fmt.Sprintf("time:%v hour:%v", t, key))
 	var v *HourDeviceStatus
 	var ok bool
 	if v, ok = d.TimeLineDeviceStatus[key]; !ok {

@@ -24,6 +24,8 @@ type DeviceRunningData struct {
 	// 附加数据
 	// 匹配数目
 	Count int `json:"count"`
+	// 合计数目
+	Total int `json:"total"`
 	// 炸机前段温度：炸机前段当前温度 YZJ1 油炸机
 	//FrontTemp float64 `json:"frontTemp"`
 	// 炸机前段温度：炸机前段当前温度 YZJ2 油炸机

@@ -29,7 +29,6 @@ func SendWorkshopDeviceData(productRecord *domain.DeviceCollection) error {
 
 func SendAsyncJob(queueName string, job interface{}) error {
 	task := asynq.NewTask(queueName, []byte(json.MarshalToString(job)))
-
 	client := asynq.NewClient(asynq.RedisClientOpt{Addr: constant.REDIS_ADDRESS})
 	_, err := client.Enqueue(task)
 	return err

@@ -40,7 +40,7 @@ func (ptr *PGWorkshopDataConsumeService) Consume(companyId, orgId int, record *d
 	}
 
 	// 0.初始化 从缓存捞数据、没取到查询库
-	deviceDailyRecord, err = redis.GetDeviceDailyRunningRecord(time.Now(), deviceRunningData.DeviceCode)
+	deviceDailyRecord, err = redis.GetDeviceDailyRunningRecord(deviceRunningData.CollectionTime, deviceRunningData.DeviceCode)
 	if err == domain.ErrorNotFound {
 		err = nil
 	}
@@ -173,6 +173,7 @@ func (ptr *PGWorkshopDataConsumeService) newDeviceRunningData(record *domain.Dev
 			break
 		}
 		data.Count = int(deviceChuanChuanJi.Count)
+		data.Total = int(deviceChuanChuanJi.Total)
 		data.ProductType = domain.ProductTypeToProductCode(deviceChuanChuanJi.ProductType)
 		if data.Date, err = formatDate(deviceChuanChuanJi.Year, deviceChuanChuanJi.Month, deviceChuanChuanJi.Day); err != nil {
 			return nil, err
@@ -273,7 +274,7 @@ func (ptr *PGWorkshopDataConsumeService) saveDeviceDailyRunningRecord(companyId,
 	if record, err = deviceDailyRunningRecordRepository.FindOne(map[string]interface{}{
 		"workStationId": workStation.WorkStationId,
 		"deviceCode":    data.DeviceCode,
-		"productDate":   utils.GetZeroTime(time.Now()),
+		"productDate":   utils.GetZeroTime(data.CollectionTime),
 	}); err != nil {
 		if err != domain.ErrorNotFound {
 			return nil, err
@@ -293,7 +294,7 @@ func (ptr *PGWorkshopDataConsumeService) saveDeviceDailyRunningRecord(companyId,
 		WorkStation:             workStation,
 		DeviceId:                device.DeviceId,
 		DeviceCode:              device.DeviceCode,
-		ProductDate:             utils.GetZeroTime(time.Now()),
+		ProductDate:             utils.GetZeroTime(data.CollectionTime),
 		DeviceRunningRecordInfo: recordInfo,
 		CreatedAt:               time.Now(),
 		UpdatedAt:               time.Now(),

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"github.com/go-pg/pg/v10"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/utils"
+	"time"
 
 	"github.com/linmadan/egglib-go/persistent/pg/sqlbuilder"
 	pgTransaction "github.com/linmadan/egglib-go/transaction/pg"
@@ -133,12 +134,18 @@ func (repository *DeviceCollectionRepository) Find(queryOptions map[string]inter
 	query := sqlbuilder.BuildQuery(tx.Model(&deviceCollectionModels), queryOptions)
 	query.SetOffsetAndLimit(20)
 	query.SetOrderDirect("device_collection_id", "DESC")
-	if deviceType,ok := queryOptions["deviceType"];ok && deviceType != "" {
+	if deviceType, ok := queryOptions["deviceType"]; ok && deviceType != "" {
 		query.SetWhereByQueryOption("device_type = ?", "deviceType")
 	}
-	if workShopName,ok := queryOptions["workShopName"];ok && workShopName.(string) != "" {
+	if workShopName, ok := queryOptions["workShopName"]; ok && workShopName.(string) != "" {
 		query.SetWhereByQueryOption("work_shop_name = ?", "workShopName")
 	}
+	if v, ok := queryOptions["productBeginTime"]; ok && !((v.(time.Time)).IsZero()) {
+		query.Where("collection_time>=?", v.(time.Time))
+	}
+	if v, ok := queryOptions["productEndTime"]; ok && !((v.(time.Time)).IsZero()) {
+		query.Where("collection_time<?", v.(time.Time))
+	}
 	if count, err := query.SelectAndCount(); err != nil {
 		return 0, deviceCollections, err
 	} else {

@@ -51,6 +51,16 @@ func ObjectToMap(o interface{}) map[string]interface{} {
 	return m
 }
 
+func ToMap(o interface{}) map[string]interface{} {
+	if o == nil {
+		return nil
+	}
+	m := make(map[string]interface{})
+	data, _ := json.Marshal(o)
+	json.Unmarshal(data, &m)
+	return m
+}
+
 func DeleteMapKeys(options map[string]interface{}, keys ...string) map[string]interface{} {
 	for i := range keys {
 		if _, ok := options[keys[i]]; ok {
@@ -382,7 +392,6 @@ func SubStr(str string, start, length int) string {
 	return string(rs[start:end])
 }
 
-
 //生成新ID
 var snowFlakeNode *snowflake.Node
 
@@ -405,4 +414,3 @@ func Round(value float64, places int32) float64 {
 	rsp, _ := d.Float64()
 	return rsp
 }
-

@@ -3,7 +3,6 @@ package mqtt
 import (
 	pahomqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/linmadan/egglib-go/utils/json"
-	"github.com/linmadan/egglib-go/utils/tool_funs"
 	"github.com/tidwall/gjson"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/application/deviceCollection/command"
 	"gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/application/deviceCollection/service"
@@ -68,7 +67,7 @@ func OnReceiveData(client pahomqtt.Client, message pahomqtt.Message) {
 				if err != nil {
 					continue
 				}
-				deviceCollection.Values = tool_funs.SimpleStructToMap(deviceBaoXianJi)
+				deviceCollection.Values = utils.ToMap(deviceBaoXianJi)
 				break
 			//油炸机
 			case domain.DeviceTypeYouZhaJi1:
@@ -77,7 +76,7 @@ func OnReceiveData(client pahomqtt.Client, message pahomqtt.Message) {
 				if err != nil {
 					continue
 				}
-				deviceCollection.Values = tool_funs.SimpleStructToMap(deviceYouZhaJi)
+				deviceCollection.Values = utils.ToMap(deviceYouZhaJi)
 				break
 			//油炸机
 			case domain.DeviceTypeYouZhaJi2:
@@ -86,7 +85,7 @@ func OnReceiveData(client pahomqtt.Client, message pahomqtt.Message) {
 				if err != nil {
 					continue
 				}
-				deviceCollection.Values = tool_funs.SimpleStructToMap(deviceYouZhaJi2)
+				deviceCollection.Values = utils.ToMap(deviceYouZhaJi2)
 				break
 			//串串机
 			case domain.DeviceTypeChuanChuanJi:
@@ -96,7 +95,7 @@ func OnReceiveData(client pahomqtt.Client, message pahomqtt.Message) {
 					log.Logger.Error(err.Error())
 					continue
 				}
-				deviceCollection.Values = tool_funs.SimpleStructToMap(deviceChuanChuanJi)
+				deviceCollection.Values = utils.ToMap(deviceChuanChuanJi)
 				break
 			//速冻线
 			case domain.DeviceTypeSuDongXian:
@@ -105,7 +104,7 @@ func OnReceiveData(client pahomqtt.Client, message pahomqtt.Message) {
 				if err != nil {
 					continue
 				}
-				deviceCollection.Values = tool_funs.SimpleStructToMap(deviceSuDongXian)
+				deviceCollection.Values = utils.ToMap(deviceSuDongXian)
 				break
 			//封口机
 			case domain.DeviceTypeFengKouJi:
@@ -114,7 +113,7 @@ func OnReceiveData(client pahomqtt.Client, message pahomqtt.Message) {
 				if err != nil {
 					continue
 				}
-				deviceCollection.Values = tool_funs.SimpleStructToMap(deviceFengKouJi)
+				deviceCollection.Values = utils.ToMap(deviceFengKouJi)
 				break
 			//封箱机
 			case domain.DeviceTypeFengXiangJi:
@@ -123,7 +122,7 @@ func OnReceiveData(client pahomqtt.Client, message pahomqtt.Message) {
 				if err != nil {
 					continue
 				}
-				deviceCollection.Values = tool_funs.SimpleStructToMap(deviceFengXiangJi)
+				deviceCollection.Values = utils.ToMap(deviceFengXiangJi)
 				break
 			//打浆机 //面包屑机
 			case domain.DeviceTypeDaJiangJi:
@@ -135,7 +134,7 @@ func OnReceiveData(client pahomqtt.Client, message pahomqtt.Message) {
 			//}
 			// 发送数据
 			deviceCollectionService := service.NewDeviceCollectionService(nil)
-			_, err = deviceCollectionService.CreateDeviceCollection(&command.CreateDeviceCollectionCommand{
+			_, err = deviceCollectionService.DeviceCollection(&command.CreateDeviceCollectionCommand{
 				WorkShopName:   deviceCollection.WorkShopName,
 				StartupStatus:  deviceCollection.StartupStatus,
 				CollectionTime: deviceCollection.CollectionTime,