et-go/et_calc/group/groupCalc.go

package group

import (
	"errors"
	"et_sink"
	"gitea.anxinyun.cn/container/common_models"
	"gitea.anxinyun.cn/container/common_models/constant/groupType"
	"gitea.anxinyun.cn/container/common_utils"
	"gitea.anxinyun.cn/container/common_utils/configLoad"
	"log"
	"node/stages"
	"sync"
	"time"
)

var (
	configHelperInstance *common_utils.ConfigHelper
	once                 sync.Once
	mu                   sync.Mutex
)

func GetConfigHelper() *common_utils.ConfigHelper {
	once.Do(func() {
		configYaml := configLoad.LoadConfig()
		redisAdd := configYaml.GetString("redis.address")
		configHelperInstance = common_utils.NewConfigHelper(redisAdd)
	})
	return configHelperInstance
}

// 分组测点
type GroupStation struct {
	Group   common_models.StationGroup
	Station common_models.Station
}

// 获取测点分组信息
func GetGroupInfo(station common_models.Station) []GroupStation {
	var gps []common_models.StationGroup
	gps = append(gps, station.Info.Group)
	gps = append(gps, station.Info.CorrGroups...)

	result := make([]GroupStation, 0)
	if gps != nil {
		for i := 0; i < len(gps); i++ {
			result = append(result, GroupStation{
				Group:   gps[i],
				Station: station,
			})
		}
	}

	return result
}

type GroupCalc struct {
	stage        *stages.Stage
	configHelper *common_utils.ConfigHelper
	signCalc     chan bool
	calcTasks    map[GroupCalcTaskKey]CalcTask
	esHandler    *et_sink.SinkHandler
}

func NewGroupCalc() *GroupCalc {
	calcTaskManager := &GroupCalc{
		stage:        stages.NewStage("测点分组计算"),
		configHelper: GetConfigHelper(),
		signCalc:     make(chan bool),
		calcTasks:    map[GroupCalcTaskKey]CalcTask{},
		esHandler:    et_sink.NewSinkGroupHandler(),
	}

	// 添加到 etNode 处理环节，实现数据加工 (缓存group各分项的主题数据 -> 分组计算 -> 分组数据)
	calcTaskManager.stage.AddProcess(calcTaskManager.processData)
	return calcTaskManager
}

func (gc *GroupCalc) GetStage() stages.Stage {
	return *gc.stage
}

// processData 的 stations 被改变了
func (gc *GroupCalc) processData(inData *common_models.ProcessData) *common_models.ProcessData {
	resultStations := make([]common_models.Station, 0)

	// 分组超时任务
	for key, task := range gc.calcTasks {
		if task.IsTimeout() {
			calcedStations := task.Calc()
			resultStations = append(resultStations, calcedStations...)
			// ES存储分组主题数据
			//dumpStr, groupTheme := task.ToDump(calcedStations)
			//gc.esHandler.SinkGroupDataToES(groupTheme)
			//log.Printf("[groupCalc.processData]group timeout calc。 %s \n", dumpStr)
			gc.dumpGroupTheme(&task, calcedStations)
			delete(gc.calcTasks, key)
		}
	}

	for _, station := range inData.Stations {
		if station.Info.Group.Id == 0 || station.Info.Group.GroupType != groupType.Settlement {
			//log.Printf("非【液压传感器测量沉降】分组。")
			resultStations = append(resultStations, station)
		} else {
			calcedStations, err := gc.cacheAndCalc(&station, inData.DeviceData.DimensionId, inData.DeviceData.TaskId, inData.DeviceData.AcqTime)
			if err != nil {
				resultStations = append(resultStations, station)
			} else {
				resultStations = append(resultStations, calcedStations...)
			}
		}
	}

	// 返回处理后的数据，计算后的resultStations可能为空
	inData.Stations = resultStations
	return inData
}

// ES存储分组主题
func (gc *GroupCalc) dumpGroupTheme(task *CalcTask, calcedStations []common_models.Station) {
	dumpStr, groupTheme := task.ToDump(calcedStations)
	log.Println(dumpStr)
	if len(groupTheme.Data) > 0 {
		gc.esHandler.SinkGroupDataToES(groupTheme)
	}
}

// cacheAndCalc 缓存和计算
// station 测点
// dimensionId 采集策略
// taskId 一次周期采集任务
// acqTime 采集时间
func (gc *GroupCalc) cacheAndCalc(station *common_models.Station, dimensionId string, taskId string, acqTime time.Time) ([]common_models.Station, error) {
	scg := GetGroupInfo(*station)
	var resultStations []common_models.Station
	for _, groupStation := range scg {
		sGroup := groupStation.Group
		if sGroup.Id == 0 || sGroup.GroupType != groupType.Settlement {
			return nil, errors.New("非【液压传感器测量沉降】分组")
		}

		key := GroupCalcTaskKey{GroupId: sGroup.Id, TaskId: taskId}
		if calcTask, ok := gc.calcTasks[key]; ok {
			calcTask.AddStationData(*station)
			log.Println(calcTask.R())
			// 分组计算
			if calcTask.CheckIntegrity() {
				secs := calcTask.ElapsedSecs()
				if secs > 10 {
					log.Printf("[groupCalc.processData] group calc wait %f秒, %s\n", secs, key.R())
				}

				calcedStations := calcTask.Calc()
				// 缓存已计算数据
				if calcedStations != nil && len(calcedStations) > 0 {
					resultStations = append(resultStations, calcedStations...)
				}
				// ES存储分组主题数据
				//dumpStr, groupTheme := calcTask.ToDump(calcedStations)
				//gc.esHandler.SinkGroupDataToES(groupTheme)
				//log.Println(dumpStr)
				gc.dumpGroupTheme(&calcTask, calcedStations)

				delete(gc.calcTasks, key)
			}
		} else {
			// 不存在的计算任务：要取到首个元素的设备的采集策略（维度），以便后面获得过期时长
			task := NewGroupCalcTask(&sGroup, dimensionId, taskId, acqTime)
			task.AddStationData(*station)
			log.Println(task.R())

			if task.CheckIntegrity() {
				calcedStations := task.Calc()
				// 缓存已计算数据
				if calcedStations != nil && len(calcedStations) > 0 {
					resultStations = append(resultStations, calcedStations...)
				}

				// ES存储分组主题数据
				gc.dumpGroupTheme(task, calcedStations)

			} else {
				task.SetTimeout()
				gc.calcTasks[key] = *task
			}
		}
	}

	return resultStations, nil
}