cicv_data_closed_loop/aarch64/pjibot_delivery/common/config/c_cloud.go

package config

import (
	"cicv-data-closedloop/common/config/c_log"
	"cicv-data-closedloop/common/util"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/gorilla/websocket"
	"gopkg.in/yaml.v3"
	"net/url"
	"os"
	"strings"
	"sync"
	"time"
)

type MonitorStruct struct {
	Url string `yaml:"url"`
}

type platform struct {
	UrlDeviceAuth string `yaml:"url-device-auth"`
	UrlTaskPoll   string `yaml:"url-task-poll"`
	UrlTask       string `yaml:"url-task"`
}

type rosbagStruct struct {
	Path string   `yaml:"path"`
	Envs []string `yaml:"envs"`
}

type hostStruct struct {
	Name   string       `yaml:"name"`
	Ip     string       `yaml:"ip"`
	Topics []string     `yaml:"topics"`
	Rosbag rosbagStruct `yaml:"rosbag"`
}

type ros struct {
	MasterAddress string   `yaml:"master-address"`
	Nodes         []string `yaml:"nodes"`
}

type disk struct {
	Name string   `yaml:"name"`
	Used uint64   `yaml:"used"`
	Path []string `yaml:"path"`
}
type trigger struct {
	Label  string   `yaml:"label"`
	Topics []string `yaml:"topics"`
}

type CollectLimitStruct struct {
	Url   string `yaml:"url"`
	Flag  int    `yaml:"flag"`
	Day   int    `yaml:"day"`
	Week  int    `yaml:"week"`
	Month int    `yaml:"month"`
	Year  int    `yaml:"year"`
}

type cloudConfig struct {
	CollectLimit          CollectLimitStruct `yaml:"collect-limit"`
	HasOneMsgTopic        bool               `yaml:"has-one-msg-topic"` // 是否存在只发单帧的话题
	FullCollect           bool               `yaml:"full-collect"`
	ConfigRefreshInterval int                `yaml:"config-refresh-interval"` // 配置刷新时间间隔
	BagNumber             int                `yaml:"bag-number"`
	TimeWindowSendGap     int                `yaml:"time-window-send-gap"` // 主节点向从节点发送窗口的最小时间间隔
	BagDataDir            string             `yaml:"bag-data-dir"`
	BagCopyDir            string             `yaml:"bag-copy-dir"`
	TriggersDir           string             `yaml:"triggers-dir"`
	RpcPort               string             `yaml:"rpc-port"`
	Triggers              []trigger          `yaml:"triggers"`
	Hosts                 []hostStruct       `yaml:"hosts"`
	Ros                   ros                `yaml:"ros"`
	Platform              platform           `yaml:"platform"`
	Disk                  disk               `yaml:"disk"`
	Monitor               MonitorStruct      `yaml:"monitor"`
}

// Request 结构体定义
type Request struct {
	Type      string      `json:"type"`
	UUID      string      `json:"uuid"`
	CommandID string      `json:"commandId"`
	Parameter interface{} `json:"parameter"`
}

// Response 结构体定义
type Response struct {
	CommandID string            `json:"commandId"`
	ErrorCode string            `json:"errorCode"`
	Results   map[string]string `json:"results"`
	Status    string            `json:"status"`
	Time      int64             `json:"time"`
	Type      string            `json:"type"`
	UUID      string            `json:"uuid"`
}

var (
	CloudConfig      cloudConfig
	CloudConfigMutex sync.RWMutex
)

// InitCloudConfig 初始化业务配置
func InitCloudConfig() {
	// history20240401：朴津机器人额外加一个获取sn码
	var snCode string

	for {
		time.Sleep(time.Duration(2) * time.Second)
		snCode, err := getSnCode()
		if err != nil {
			c_log.GlobalLogger.Error("获取sn码失败：", err.Error())
			continue
		}
		LocalConfig.SecretKey = snCode
		LocalConfig.EquipmentNo = "pjibot-" + snCode
		break
	}
	c_log.GlobalLogger.Info("本地机器人sn码为：", snCode)

	c_log.GlobalLogger.Info("初始化OSS配置文件 - 开始。")
	// 获取文件的目录
	_ = util.CreateParentDir(LocalConfig.CloudConfigLocalPath)
	// 3 ------- 获取 yaml 字符串 -------
	cloudConfigObjectKey := LocalConfig.OssBasePrefix + LocalConfig.EquipmentNo + "/" + LocalConfig.CloudConfigFilename

	// 判断文件是否存在。如果不存在则使用默认的
	isExist, err := OssBucket.IsObjectExist(cloudConfigObjectKey)
	if err != nil {
		c_log.GlobalLogger.Errorf("判断配置文件是否存在失败，错误信息为：%v", err)
	}
	if isExist {
		c_log.GlobalLogger.Info("使用机器人自定义配置文件:", cloudConfigObjectKey)
	} else {
		cloudConfigObjectKey = LocalConfig.OssBasePrefix + LocalConfig.CloudConfigFilename // 默认配置文件路径
		c_log.GlobalLogger.Info("使用机器人默认配置文件:", cloudConfigObjectKey)
	}

	for {
		OssMutex.Lock()
		err := OssBucket.GetObjectToFile(cloudConfigObjectKey, LocalConfig.CloudConfigLocalPath)
		OssMutex.Unlock()
		if err != nil {
			c_log.GlobalLogger.Error("下载 OSS 上的配置文件 "+cloudConfigObjectKey+" 失败，请尽快在 OSS 上传配置文件。", err)
			time.Sleep(time.Duration(2) * time.Second)
			continue
		}
		break
	}

	content, err := os.ReadFile(LocalConfig.CloudConfigLocalPath)
	if err != nil {
		c_log.GlobalLogger.Error("程序崩溃，配置文件 ", LocalConfig.CloudConfigLocalPath, " 读取失败:", err)
		os.Exit(-1)
	}

	// 4 ------- 解析YAML内容 -------
	var newCloudConfig cloudConfig
	err = yaml.Unmarshal(content, &newCloudConfig)
	if err != nil {
		c_log.GlobalLogger.Error("程序崩溃，配置文件 ", LocalConfig.CloudConfigLocalPath, " 解析失败:", err)
		os.Exit(-1)
	}

	// 5 ------- 校验 yaml -------
	if checkCloudConfig(newCloudConfig) {
		CloudConfigMutex.RLock()
		CloudConfig = newCloudConfig
		CloudConfigMutex.RUnlock()
	} else {
		c_log.GlobalLogger.Error("程序崩溃，配置文件格式错误：", newCloudConfig)
		os.Exit(-1)
	}
	c_log.GlobalLogger.Info("初始化OSS配置文件 - 成功。")
	util.CreateDir(CloudConfig.BagDataDir)
	util.CreateDir(CloudConfig.BagCopyDir)
}

// 更新业务配置
func refreshCloudConfig() {
	// 获取文件的目录
	_ = util.CreateParentDir(LocalConfig.CloudConfigLocalPath)
	// 3 ------- 获取 yaml 字符串 -------
	var content []byte
	cloudConfigObjectKey := LocalConfig.OssBasePrefix + LocalConfig.EquipmentNo + "/" + LocalConfig.CloudConfigFilename

	isExist, err := OssBucket.IsObjectExist(cloudConfigObjectKey)
	if err != nil {
		c_log.GlobalLogger.Errorf("判断配置文件是否存在失败，错误信息为：%v", err)
		return
	}
	if !isExist {
		cloudConfigObjectKey = LocalConfig.OssBasePrefix + LocalConfig.CloudConfigFilename // 默认配置文件路径
	}

	OssMutex.Lock()
	err = OssBucket.GetObjectToFile(cloudConfigObjectKey, LocalConfig.CloudConfigLocalPath)
	OssMutex.Unlock()
	if err != nil {
		c_log.GlobalLogger.Error("下载oss上的配置文件"+cloudConfigObjectKey+"失败。", err)
		return
	}

	content, err = os.ReadFile(LocalConfig.CloudConfigLocalPath)
	if err != nil {
		c_log.GlobalLogger.Error("配置文件 ", LocalConfig.CloudConfigLocalPath, " 读取失败:", err)
		return
	}

	// 4 ------- 解析YAML内容 -------
	var newCloudConfig cloudConfig
	err = yaml.Unmarshal(content, &newCloudConfig)
	if err != nil {
		c_log.GlobalLogger.Error("配置文件 ", LocalConfig.CloudConfigLocalPath, " 解析失败:", err)
		return
	}

	// 5 ------- 校验 yaml -------
	if checkCloudConfig(newCloudConfig) {
		CloudConfigMutex.RLock()
		CloudConfig = newCloudConfig
		CloudConfigMutex.RUnlock()
	} else {
		c_log.GlobalLogger.Error("配置文件格式错误：", newCloudConfig)
		return
	}
	util.CreateDir(CloudConfig.BagDataDir)
	util.CreateDir(CloudConfig.BagCopyDir)

}

// RefreshCloudConfig 轮询oss上的配置文件更新到本地
func RefreshCloudConfig() {
	for {
		time.Sleep(time.Duration(CloudConfig.ConfigRefreshInterval) * time.Second)
		refreshCloudConfig()
	}
}

// CheckConfig 校验 cfg.yaml 文件
func checkCloudConfig(check cloudConfig) bool {
	if len(check.Hosts) != 1 {
		c_log.GlobalLogger.Error("cloud-config.yaml中配置的hosts必须为1。")
		os.Exit(-1)
	}
	return true
}

func getSnCode() (string, error) {
	var command []string
	command = append(command, "get")
	command = append(command, "sn")
	_, snOutput, err := util.ExecuteSync(LocalConfig.RosparamPath, command...)
	if err != nil {
		return "", errors.New("执行获取sn码命令" + util.ToString(command) + "出错：" + util.ToString(err))
	}
	c_log.GlobalLogger.Info("执行获取sn码命令", command, "成功，结果为：", snOutput)
	snCode := strings.Replace(strings.Replace(snOutput, " ", "", -1), "\n", "", -1)
	return snCode, nil
}

// SendWebsocketRequest 发送WebSocket请求并返回sn字段的值
func SendWebsocketRequest(serverURL, path string, request Request) (string, error) {
	// 构建WebSocket连接URL
	u := url.URL{Scheme: "ws", Host: serverURL, Path: path}

	// 创建一个Dialer实例，用于建立WebSocket连接
	dialer := websocket.Dialer{
		ReadBufferSize:  1024,
		WriteBufferSize: 1024,
		// 可选：设置超时等
		HandshakeTimeout: 5 * time.Second,
	}

	// 建立WebSocket连接
	conn, _, err := dialer.Dial(u.String(), nil)
	if err != nil {
		return "", fmt.Errorf("dial: %w", err)
	}
	defer conn.Close()

	// 将请求JSON编码为字节
	requestJSON, err := json.Marshal(request)
	if err != nil {
		return "", fmt.Errorf("marshal request: %w", err)
	}

	// 发送WebSocket消息
	err = conn.WriteMessage(websocket.TextMessage, requestJSON)
	if err != nil {
		return "", fmt.Errorf("write: %w", err)
	}

	// 读取WebSocket响应
	_, responseBytes, err := conn.ReadMessage()
	if err != nil {
		return "", fmt.Errorf("read: %w", err)
	}

	// 将响应字节解码为JSON
	var response Response
	err = json.Unmarshal(responseBytes, &response)
	if err != nil {
		return "", fmt.Errorf("unmarshal response: %w", err)
	}

	// 返回sn字段的值
	return response.Results["sn"], nil
}