package config import ( "cicv-data-closedloop/common/config/c_log" "cicv-data-closedloop/common/util" "errors" "gopkg.in/yaml.v3" "os" "strings" "sync" "time" ) type MonitorStruct struct { Url string `yaml:"url"` } type PlatformStruct 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 RosStruct struct { MasterAddress string `yaml:"master-address"` Nodes []string `yaml:"nodes"` } type DiskStruct struct { Name string `yaml:"name"` Used uint64 `yaml:"used"` } type TriggerStruct struct { Label string `yaml:"label"` Topics []string `yaml:"topics"` } type CollectLimitStruct struct { Url string `yaml:"url"` Day int `yaml:"day"` Week int `yaml:"week"` Month int `yaml:"month"` Year int `yaml:"year"` } type DataDirStruct struct { Src string `yaml:"src"` SrcSub []string `yaml:"src-sub"` Dest string `yaml:"dest"` } type cloudConfig struct { CollectLimit CollectLimitStruct `yaml:"collect-limit"` DataDir DataDirStruct `yaml:"data-dir"` MapBufFiles []string `yaml:"map-buf-files"` FullCollect bool `yaml:"full-collect"` ConfigRefreshInterval int `yaml:"config-refresh-interval"` // 配置刷新时间间隔 BagNumber int `yaml:"bag-number"` TimeWindowSendGap int `yaml:"time-window-send-gap"` // 主节点向从节点发送窗口的最小时间间隔 MapBagPath string `yaml:"map-bag-path"` TfstaticBagPath string `yaml:"tfstatic-bag-path"` CostmapBagPath string `yaml:"costmap-bag-path"` BagDataDir string `yaml:"bag-data-dir"` BagCopyDir string `yaml:"bag-copy-dir"` TriggersDir string `yaml:"triggers-dir"` RpcPort string `yaml:"rpc-port"` Triggers []TriggerStruct `yaml:"triggers"` Hosts []HostStruct `yaml:"hosts"` Ros RosStruct `yaml:"ros"` Platform PlatformStruct `yaml:"platform"` Disk DiskStruct `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) } 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 //}