goInOut/adaptors/安心云es数据to南京智行.go

package adaptors

import (
	"encoding/hex"
	"encoding/json"
	"fmt"
	"goInOut/consumers/JYES_NJZX"
	"goInOut/consumers/JYES_NJZX/protoDataFiles"
	"goInOut/dbOperate"
	"goInOut/models"
	"google.golang.org/protobuf/proto"
	"log"
	"strings"
	"time"
)

// Adaptor_AXYES_NJZX 安心云依加尔山es 特征数据 to 南京智行平台
type Adaptor_AXYES_NJZX struct {
	//传感器code转换信息
	PointInfo  map[int64]map[int64]int64
	StructInfo map[int64]int64
	//一些必要信息
	Info  map[string]string
	Redis *dbOperate.RedisHelper
}

func (the *Adaptor_AXYES_NJZX) Transform(structId int64, factorId int, rawMsg string) []NeedPush {
	//es查到的数据分装进结构体里面
	esAggDateHistogram := JYES_NJZX.EsThemeDateValue{}
	var needPush []NeedPush

	err := json.Unmarshal([]byte(rawMsg), &esAggDateHistogram)
	if err != nil {
		fmt.Println("Error unmarshalling JSON:", err)
		return nil
	}

	Payload := the.EsDataValueChangeToNJZX(structId, factorId, esAggDateHistogram)
	if len(Payload) == 0 {
		return needPush
	}

	needPush = append(needPush, NeedPush{
		Payload: Payload,
	})
	return needPush
}

func (the *Adaptor_AXYES_NJZX) EsDataValueChangeToNJZX(structId int64, factorId int, esDataValue JYES_NJZX.EsThemeDateValue) (result []byte) {
	buckets := esDataValue.Hits.Hits
	//数据汇总
	complexData := &protoDataFiles.ComplexData{}
	for _, sensorBucket := range buckets {
		sensorId := sensorBucket.Source.Sensor //安心云de测点id

		//优先redis获取
		station := models.Station{}
		k1 := fmt.Sprintf("station:%d", sensorId)
		errRedis := the.Redis.GetObj(k1, &station)
		if errRedis != nil {
			log.Printf("redis 获取[s:%d,f:%d]测点[%d]标签异常: %v", structId, factorId, sensorId, errRedis)
			continue
		}
		monitorCodeStr := the.getPointCodeFromLabel(station.Labels)
		if monitorCodeStr == "" {
			log.Printf("redis 获取[s:%d,f:%d]测点[%d],标签信息[%s]转换异常,跳过", structId, factorId, sensorId, station.Labels)
			continue
		}

		dataDefinition := the.ChangeToNJZXData(factorId, monitorCodeStr, sensorBucket)

		complexData.SensorData = append(complexData.SensorData, dataDefinition)
	}

	//v, _ := json.Marshal(complexData)
	//log.Printf("[struct:%d,factor:%d] 特征数据=> %s", structId, factorId, v)
	result, _ = proto.Marshal(complexData)
	log.Printf("[struct:%d,factor:%d] protobuf数据=> %s", structId, factorId, hex.EncodeToString(result))
	return result
}

func (the *Adaptor_AXYES_NJZX) getMonitorTypeByFactorId(factorId int) protoDataFiles.MonitoryType {
	//监测因素 2温湿度  4温度  18裂缝检测
	//103净空收敛  102拱顶沉降  96二次衬彻应变
	//107道床及拱腰结构沉降  156风速  578风向
	switch factorId {
	case 2: //温湿度
		return protoDataFiles.MonitoryType_RHS
	case 4: //温度
		return protoDataFiles.MonitoryType_TMP
	case 18: //裂缝检测
		return protoDataFiles.MonitoryType_CRK
	case 103: //净空收敛
		return protoDataFiles.MonitoryType_CRK
	case 102: //拱顶沉降
		return protoDataFiles.MonitoryType_CRK //无对应
	case 96: //二次衬彻应变
		return protoDataFiles.MonitoryType_RSG
	case 107: //道床及拱腰结构沉降
		return protoDataFiles.MonitoryType_VIB //无对应
	case 156: //风速
		return protoDataFiles.MonitoryType_WDS
	case 578: //风向
		return protoDataFiles.MonitoryType_WDD
	default:
		log.Printf("factorId=%d,无匹配的MonitorType", factorId)
		return protoDataFiles.MonitoryType_RHS
	}
}

func (the *Adaptor_AXYES_NJZX) parseTimeToTimestamp(timeStr string) (int64, error) {
	// 解析时间字符串为 time.Time 对象
	parsedTime, err := time.Parse(time.RFC3339, timeStr)
	if err != nil {
		return 0, err
	}
	// 返回 Unix 时间戳（秒数）
	return parsedTime.Unix(), nil
}

func (the *Adaptor_AXYES_NJZX) ChangeToNJZXData(factorId int, monitorCodeStr string, dateBucket JYES_NJZX.Hits) *protoDataFiles.SensorData {

	Atime, _ := the.parseTimeToTimestamp(dateBucket.Source.CollectTime)
	monitoryType := the.getMonitorTypeByFactorId(factorId)
	dataDefinitionData := &protoDataFiles.SensorData{
		MonitorType: monitoryType,
		SensorNo:    monitorCodeStr,
		UpTime:      Atime,
	}

	switch factorId {
	case 2: //温湿度
		dataDefinitionData.DataBody = &protoDataFiles.SensorData_Rhs{Rhs: &protoDataFiles.RHSRealTime{
			Temperature: []float32{float32(dateBucket.Source.Data["temperature"])},
			Humidity:    []float32{float32(dateBucket.Source.Data["humidity"])},
		}}
	case 4: //温度
		dataDefinitionData.DataBody = &protoDataFiles.SensorData_Tmp{Tmp: &protoDataFiles.TMPRealTime{
			Temperature: []float32{float32(dateBucket.Source.Data["temperature"])},
		}}
	case 18: //裂缝检测
		dataDefinitionData.DataBody = &protoDataFiles.SensorData_Crk{Crk: &protoDataFiles.CRKRealTime{
			CrackWidth: []float32{float32(dateBucket.Source.Data["crack"])},
		}}
	case 103: //净空收敛//123456789
		dataDefinitionData.DataBody = &protoDataFiles.SensorData_Disr{Disr: &protoDataFiles.DISRRealTime{
			Displacement: []float32{float32(dateBucket.Source.Data["displacement"])},
		}}
	case 102: //拱顶沉降//123456789
		dataDefinitionData.DataBody = &protoDataFiles.SensorData_Disr{Disr: &protoDataFiles.DISRRealTime{
			Displacement: []float32{float32(dateBucket.Source.Data["displacement"])},
		}}
	case 96: //二次衬彻应变
		dataDefinitionData.DataBody = &protoDataFiles.SensorData_Rsg{Rsg: &protoDataFiles.RSGRealTime{
			Strain: []float32{float32(dateBucket.Source.Data["strain"])},
		}}
	case 107: //道床及拱腰结构沉降//123456789
		dataDefinitionData.DataBody = &protoDataFiles.SensorData_Disr{Disr: &protoDataFiles.DISRRealTime{
			Displacement: []float32{float32(dateBucket.Source.Data["displacement"])},
		}}
	case 156: //风速
		dataDefinitionData.DataBody = &protoDataFiles.SensorData_Wds{Wds: &protoDataFiles.WDSRealTime{
			WindSpeed: []float32{float32(dateBucket.Source.Data["speed"])},
		}}
	case 578: //风向
		dataDefinitionData.DataBody = &protoDataFiles.SensorData_Wdd{Wdd: &protoDataFiles.WDDRealTime{
			WindDirection: []float32{float32(dateBucket.Source.Data["direction"])},
		}}
	}

	return dataDefinitionData
}

func (the *Adaptor_AXYES_NJZX) getUniqueCode(structId int64) (uniqueCode int64) {
	if v, ok := the.StructInfo[structId]; ok {
		uniqueCode = v
	}
	return uniqueCode
}

func (the *Adaptor_AXYES_NJZX) getPointCodeFromLabel(label string) string {
	//解析label  {code:wd01}
	pointUniqueCode := ""
	if len(label) > 3 {
		newLabel := strings.TrimLeft(label, "{code:")
		str := strings.TrimRight(newLabel, "}")
		if str == "" {
			log.Printf("测点标签转换异常[%s]", label)
		}
		pointUniqueCode = str
	}

	return pointUniqueCode
}