commit

aarch64/pjibot_clean/common/config/c_cloud.go

@@ -0,0 +1,323 @@
+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"`
+	Path []string `yaml:"path"`
+}
+
+type TriggerStruct 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 CollectNumPlusStruct struct {
+	Url string `yaml:"url"`
+}
+
+type DataDirStruct struct {
+	Src    string   `yaml:"src"`
+	SrcSub []string `yaml:"src-sub"`
+	Dest   string   `yaml:"dest"`
+}
+
+type cloudConfig struct {
+	CollectLimit          CollectLimitStruct   `yaml:"collect-limit"`
+	CollectNumPlus        CollectNumPlusStruct `yaml:"collect-num-plus"`
+	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码命令" + LocalConfig.RosparamPath + util.ToString(command) + "出错：" + util.ToString(err))
+	}
+	c_log.GlobalLogger.Info("执行获取sn码命令", LocalConfig.RosparamPath, 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
+//}

aarch64/pjibot_clean/common/config/c_killrpcserver.go

@@ -0,0 +1,21 @@
+package config
+
+import (
+	"cicv-data-closedloop/common/config/c_log"
+	"net"
+	"os"
+)
+
+var KillSignalListener net.Listener
+
+func InitKillSignalListener(serverIp string) {
+	var err error
+	c_log.GlobalLogger.Info("初始化RPC端口监听Kill信号 - 开始。")
+	socket := serverIp + ":" + CloudConfig.RpcPort
+	KillSignalListener, err = net.Listen("tcp", socket)
+	if err != nil {
+		c_log.GlobalLogger.Error("监听rpc端口失败:", err)
+		os.Exit(-1)
+	}
+	c_log.GlobalLogger.Info("初始化RPC端口监听Kill信号 - 成功：", socket)
+}

aarch64/pjibot_clean/common/config/c_local.go

@@ -0,0 +1,62 @@
+package config
+
+import (
+	"cicv-data-closedloop/common/config/c_log"
+	"gopkg.in/yaml.v2"
+	"os"
+)
+
+type node struct {
+	Name string `yaml:"name"`
+	Ip   string `yaml:"ip"`
+}
+
+type restartCmd struct {
+	Dir  string   `yaml:"dir"`
+	Name string   `yaml:"name"`
+	Args []string `yaml:"args"`
+}
+
+type localConfig struct {
+	Node                 node       `yaml:"node"`                    // 节点信息
+	RosparamPath         string     `yaml:"rosparam-path"`           // 获取oss配置的url
+	UrlGetOssConfig      string     `yaml:"url-get-oss-config"`      // 获取oss配置的url
+	OssBasePrefix        string     `yaml:"oss-base-prefix"`         // 云端配置文件的位置
+	CloudConfigFilename  string     `yaml:"cloud-config-filename"`   // 云端配置文件名称
+	CloudConfigLocalPath string     `yaml:"cloud-config-local-path"` // 将 oss 的配置文件下载到本地的位置
+	LocalWebsocketPort   string     `yaml:"local-websocket-port"`    // websocket端口号
+	RestartCmd           restartCmd `yaml:"restart-cmd"`             // 重启命令
+	EquipmentNo          string     // 当前设备的编号
+	SecretKey            string     // 当前设备的密钥
+}
+
+var (
+	LocalConfig localConfig
+)
+
+func InitLocalConfig(localConfigPath string) {
+	c_log.GlobalLogger.Info("初始化本地配置文件 - 开始：", localConfigPath)
+	// 读取YAML文件内容
+	content, err := os.ReadFile(localConfigPath)
+	if err != nil {
+		c_log.GlobalLogger.Error("读取本地配置文件失败。", err)
+		os.Exit(-1)
+	}
+
+	// 解析YAML内容
+	err = yaml.Unmarshal(content, &LocalConfig)
+	if err != nil {
+		c_log.GlobalLogger.Error("解析本地配置文件失败。", err)
+		os.Exit(-1)
+	}
+
+	// history20240401：设备密钥需要获取sn码，设备编号同样。######由于执行命令需要环境变量，所以放到 c_cloud.go 中####
+	/*
+		# 例如，数据闭环平台参数
+		equipment-no: pjibot-P1YNYD1M228000127
+		secret-key: P1YNYD1M228000127
+	*/
+
+	c_log.GlobalLogger.Info("初始化本地配置文件 - 成功：", LocalConfig)
+
+}

aarch64/pjibot_clean/common/config/c_oss.go

@@ -0,0 +1,52 @@
+package config
+
+import (
+	"cicv-data-closedloop/common/config/c_log"
+	"cicv-data-closedloop/common/util"
+	"encoding/json"
+	"github.com/aliyun/aliyun-oss-go-sdk/oss"
+	"os"
+	"sync"
+)
+
+type OssConnectInfoStruct struct {
+	Endpoint        string `json:"endpoint"`
+	AccessKeyId     string `json:"accessKeyId"`
+	AccessKeySecret string `json:"accessKeySecret"`
+	BucketName      string `json:"bucketName"`
+}
+
+var (
+	OssClient *oss.Client
+	OssBucket *oss.Bucket
+	OssMutex  sync.Mutex
+)
+
+func InitOssConfig() {
+	c_log.GlobalLogger.Info("初始化OSS客户端对象 - 开始。")
+	// 1 访问 HTTP 服务获取 OSS 配置
+	get, err := util.HttpGet(LocalConfig.UrlGetOssConfig)
+	if err != nil {
+		c_log.GlobalLogger.Error("http获取oss配置时出错:", err)
+		os.Exit(-1)
+	}
+	var ossConnectInfo OssConnectInfoStruct
+	err = json.Unmarshal([]byte(get), &ossConnectInfo)
+	if err != nil {
+		c_log.GlobalLogger.Error("解析json时出错:", err)
+		os.Exit(-1)
+	}
+	c_log.GlobalLogger.Infof("oss 配置信息为：%v", ossConnectInfo)
+
+	OssClient, err = oss.New(ossConnectInfo.Endpoint, ossConnectInfo.AccessKeyId, ossConnectInfo.AccessKeySecret, oss.UseCname(true))
+	if err != nil {
+		c_log.GlobalLogger.Error("无法创建阿里云client:", err)
+		os.Exit(-1)
+	}
+	OssBucket, err = OssClient.Bucket(ossConnectInfo.BucketName)
+	if err != nil {
+		c_log.GlobalLogger.Error("无法创建阿里云bucket:", err)
+		os.Exit(-1)
+	}
+	c_log.GlobalLogger.Info("初始化OSS客户端对象 - 成功。")
+}

aarch64/pjibot_clean/common/config/c_platform.go

@@ -0,0 +1,212 @@
+package config
+
+import (
+	"cicv-data-closedloop/common/config/c_log"
+	"cicv-data-closedloop/common/util"
+	"encoding/json"
+	"strings"
+	"time"
+)
+
+type taskTrigger struct {
+	TriggerId         int    `json:"triggerId"`
+	TriggerName       string `json:"triggerName"`
+	TriggerScriptPath string `json:"triggerScriptPath"`
+	TriggerType       string `json:"triggerType"`
+}
+
+type PlatformConfigStruct struct {
+	TaskConfigId    string        `json:"taskConfigId"`   // 配置ID
+	TaskConfigName  string        `json:"taskConfigName"` // 配置名称
+	DropUploadData  bool          `json:"dropUploadData"` // 更新任务时 true 先上传旧任务 false 删除旧任务
+	TaskMaxTime     int           `json:"taskMaxTime"`
+	TaskBeforeTime  int           `json:"taskBeforeTime"`
+	TaskAfterTime   int           `json:"taskAfterTime"`
+	TaskCachePolicy string        `json:"taskCachePolicy"`
+	EquipmentTopic  string        `json:"equipmentTopic"` // topic序列
+	Lru             []string      `json:"LRU"`
+	TaskTriggers    []taskTrigger `json:"taskTriggers"`
+}
+
+type response struct {
+	Data    PlatformConfigStruct `json:"data"`
+	Success bool                 `json:"success"`
+	Message string               `json:"message"`
+	Code    int                  `json:"code"`
+	NowTime string               `json:"nowTime"`
+}
+
+var (
+	PlatformConfig PlatformConfigStruct
+	RecordTopics   []string
+)
+
+// 初始化数据闭环平台的配置
+func InitPlatformConfig() {
+	var err error
+	c_log.GlobalLogger.Info("获取数据闭环平台配置 - 开始")
+	// 1 如果车辆没有配置任务，则阻塞在这里，不启动任务
+	for {
+		time.Sleep(time.Duration(2) * time.Second)
+		// 判断是否有配置，第一次访问状态应该为：CHANGE（一共三种状态 CHANGE|UNCHANGE|NONE）
+		PlatformConfig, err = getConfig()
+		if err != nil {
+			c_log.GlobalLogger.Error("获取配置status失败：", err)
+			continue
+		}
+		if checkPlatformConfig() {
+			RecordTopics = strings.Split(PlatformConfig.EquipmentTopic, ",")
+			// 去掉首尾空格
+			for i, topic := range RecordTopics {
+				RecordTopics[i] = strings.TrimSpace(topic)
+			}
+			break
+		}
+	}
+	c_log.GlobalLogger.Info("获取数据闭环平台配置 - 成功。")
+}
+
+func CheckPlatformConfigStatus(maxRetryCount int) bool {
+	var err error
+	for i := 0; i < maxRetryCount; i++ {
+		time.Sleep(time.Duration(2) * time.Second)
+		// 判断是否有配置，第一次访问状态应该为：CHANGE（一共三种状态 CHANGE|UNCHANGE|NONE）
+		PlatformConfig, err = getConfig()
+		if err != nil {
+			c_log.GlobalLogger.Error("获取配置status失败：", err)
+			continue
+		}
+		if checkPlatformConfig() {
+			return true
+		}
+	}
+	return false
+}
+
+/*
+	{
+	  "data": {
+	    "accessToken": "YWRmYWRzZmFzZGZhZHNmYWRmYWRm=",
+	    "expireTime": "28800",
+	    "equipmentNo": "robot-001"
+	  },
+	  "success": true,
+	  "message": "ok",
+	  "code": 1,
+	  "nowTime": "2023-12-09 22:41:00"
+	}
+*/
+// 认证接口，获取access_token
+func GetAccessToken() (string, error) {
+	url := &CloudConfig.Platform.UrlDeviceAuth
+	param := &map[string]string{
+		"equipmentNo": LocalConfig.EquipmentNo,
+		"secretKey":   LocalConfig.SecretKey,
+	}
+	respJson, err := util.HttpPostJsonResponseString(
+		*url,
+		*param,
+	)
+	if err != nil {
+		return "", nil
+	}
+	respMap, err := util.JsonStringToMap(respJson)
+	if err != nil {
+		c_log.GlobalLogger.Errorf("解析返回结果【%v】失败，请求地址为【%v】,请求参数为【%v】：%v", respJson, *url, *param, err)
+		return "", nil
+	}
+
+	dataMap, ok := respMap["data"].(map[string]interface{})
+	if !ok {
+		c_log.GlobalLogger.Error("解析返回结果.data", dataMap, "失败：", err)
+		return "", nil
+	}
+	return dataMap["accessToken"].(string), nil
+}
+
+/*
+	{
+	  "data": {
+	    "status": "UNCHANGE"
+	    "taskConfigld": "xxx"
+	  },
+	  "success": true,
+	  "message": "ok",
+	  "code": 1,
+	  "nowTime": "2023-12-09 21:08:49"
+	}
+*/
+//GetStatus 根据taskConfigId获取任务status，如果传入空代表车端没有配置，直接获取新的配置
+func GetStatus(taskConfigId string) (string, error) {
+	token, err := GetAccessToken()
+	if err != nil {
+		return "", err
+	}
+	resp, err := util.HttpGetStringAddHeadersResponseString(
+		CloudConfig.Platform.UrlTaskPoll,
+		map[string]string{
+			"authorization": token,
+		},
+		map[string]string{
+			"equipmentNo":  LocalConfig.EquipmentNo,
+			"taskConfigId": taskConfigId,
+		},
+	)
+
+	if err != nil {
+		c_log.GlobalLogger.Error("访问接口", CloudConfig.Platform.UrlTask, "失败：", err)
+		return "", err
+	}
+	respMap, err := util.JsonStringToMap(resp)
+	if err != nil {
+		c_log.GlobalLogger.Error("解析【返回结果1】", resp, "失败：", err)
+		return "", err
+	}
+	dataMap, ok := respMap["data"].(map[string]interface{})
+	if !ok {
+		c_log.GlobalLogger.Errorf("解析【返回结果.data】的类型不是(map[string]interface{})，【dataMap】=%v", dataMap)
+		return "", err
+	}
+	return dataMap["status"].(string), nil
+}
+
+func getConfig() (PlatformConfigStruct, error) {
+	token, err := GetAccessToken()
+	if err != nil {
+		return PlatformConfigStruct{}, err
+	}
+	// 下载插件和获取配置
+	// 2 访问配置获取接口
+	resp, err := util.HttpGetStringAddHeadersResponseString(
+		CloudConfig.Platform.UrlTask,
+		map[string]string{
+			"authorization": token,
+		},
+		map[string]string{
+			"equipmentNo": LocalConfig.EquipmentNo,
+		},
+	)
+	if err != nil {
+		c_log.GlobalLogger.Error("访问接口", CloudConfig.Platform.UrlTask, "失败：", err)
+		return PlatformConfigStruct{}, err
+	}
+	var result response
+	err = json.Unmarshal([]byte(resp), &result)
+	if err != nil {
+		c_log.GlobalLogger.Error("解析【返回结果】", resp, "失败：", err)
+		return PlatformConfigStruct{}, err
+	}
+	return result.Data, nil
+}
+
+func checkPlatformConfig() bool {
+	if PlatformConfig.TaskConfigId == "" {
+		c_log.GlobalLogger.Error("数据闭环平台没有配置任务。")
+		return false
+	}
+	if PlatformConfig.EquipmentTopic == "" {
+		c_log.GlobalLogger.Error("数据闭环平台没有配置topic序列。")
+		return false
+	}
+	return true
+}

aarch64/pjibot_clean/common/config/c_resource.go

@@ -0,0 +1,35 @@
+package config
+
+import (
+	"cicv-data-closedloop/common/config/c_log"
+	"cicv-data-closedloop/common/util"
+	"encoding/json"
+	"time"
+)
+
+// SendResourceUsage 保存资源占用情况
+func SendResourceUsage() {
+	for {
+		time.Sleep(time.Duration(1) * time.Second)
+		top10Cpu, top10Mem := util.GetTop10CpuAndMem()
+		top10CpuJson, _ := json.MarshalIndent(top10Cpu, "", "    ")
+		top10MemJson, _ := json.MarshalIndent(top10Mem, "", "    ")
+		responseString, err := util.HttpPostJsonWithHeaders(
+			CloudConfig.Monitor.Url,
+			map[string]string{"Authorization": "U9yKpD6kZZDDe4LFKK6myAxBUT1XRrDM"},
+			map[string]string{
+				"totalCpuUsage":    util.ToString(util.GetCpuPercent()),
+				"totalMemoryUsage": util.ToString(util.GetMemoryPercent()),
+				"top10Process":     string(top10CpuJson),
+				"top10Cpu":         string(top10CpuJson),
+				"top10Mem":         string(top10MemJson),
+				"deviceNumber":     LocalConfig.EquipmentNo,
+				"socIp":            LocalConfig.Node.Ip,
+			},
+		)
+		if err != nil {
+			c_log.GlobalLogger.Errorf("发送数据监控信息报错%v，响应信息为：%v", err, responseString)
+		}
+		//c_log.GlobalLogger.Infof("发送数据监控信息成功，响应信息为：%v", responseString)
+	}
+}

aarch64/pjibot_clean/common/config/c_ros.go

@@ -0,0 +1,39 @@
+package config
+
+import (
+	"cicv-data-closedloop/common/config/c_log"
+	"cicv-data-closedloop/common/util"
+	"github.com/bluenviron/goroslib/v2"
+	"time"
+)
+
+var (
+	RosNode    *goroslib.Node
+	RosbagPath string
+	RosbagEnvs []string
+)
+
+func InitRosConfig() {
+	var err error
+	// 1
+	c_log.GlobalLogger.Info("初始化RosNode - 开始")
+	for {
+		time.Sleep(time.Duration(2) * time.Second)
+		if RosNode, err = goroslib.NewNode(goroslib.NodeConf{Name: "node" + util.GetNowTimeCustom(), MasterAddress: CloudConfig.Ros.MasterAddress}); err != nil {
+			c_log.GlobalLogger.Info("初始化RosNode - 进行中：", err)
+			continue
+		}
+		break
+	}
+	c_log.GlobalLogger.Info("初始化RosNode - 成功：", CloudConfig.Ros.MasterAddress)
+	// 2 获取 rosbag 命令路径和环境变量
+	for _, host := range CloudConfig.Hosts {
+		if host.Name == LocalConfig.Node.Name {
+			RosbagPath = host.Rosbag.Path
+			RosbagEnvs = host.Rosbag.Envs
+			break
+		}
+	}
+	c_log.GlobalLogger.Infof("rosbag 命令路径为：%v，环境变量为：%v", RosbagPath, RosbagEnvs)
+
+}

aarch64/pjibot_clean/common/config/c_websocket.go

@@ -0,0 +1,225 @@
+package config
+
+import (
+	"cicv-data-closedloop/common/config/c_log"
+	"encoding/json"
+	"fmt"
+	"github.com/gorilla/websocket"
+	"net/url"
+	"time"
+)
+
+var (
+	WsConn                 *websocket.Conn
+	reconnectionInProgress bool
+)
+
+// Request 结构体定义
+type Request struct {
+	Type      string      `json:"type"`
+	UUID      string      `json:"uuid"`
+	CommandID string      `json:"commandId"`
+	Parameter interface{} `json:"parameter"`
+}
+
+// Request1 结构体定义
+type Request1 struct {
+	Type      string      `json:"type"`
+	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"`
+}
+
+// StatusMessage 状态消息 结构体定义
+type StatusMessage struct {
+	Type  string      `json:"type"`
+	Topic string      `json:"topic"`
+	Time  int64       `json:"time"`
+	Data  interface{} `json:"data"`
+}
+
+func keepAlive() {
+	ticker := time.NewTicker(30 * time.Second)
+	defer ticker.Stop()
+
+	request := Request1{
+		Type:      "request",
+		CommandID: "heart",
+		Parameter: nil,
+	}
+
+	requestJSON, err := json.Marshal(request)
+	if err != nil {
+		//c_log.GlobalLogger.Error("保持websocket连接活跃，解析requestJSON - 失败。", err)
+	}
+
+	for {
+		select {
+		case <-ticker.C:
+			err := WsConn.WriteMessage(websocket.TextMessage, requestJSON)
+			if err != nil {
+				//c_log.GlobalLogger.Error("保持websocket连接活跃，发送心跳请求 - 失败。", err)
+				WsConn.Close()
+				//c_log.GlobalLogger.Info("重试连接websocket...")
+				ConnectWebsocket() // 重新连接
+				continue
+			}
+			//c_log.GlobalLogger.Info("保持websocket连接活跃，发送心跳请求 - 成功。")
+		}
+	}
+}
+
+func SendWebsocketHeartbeat(conn *websocket.Conn, maxRetries int) (bool, error) {
+
+	request := Request1{
+		Type:      "request",
+		CommandID: "heart",
+		Parameter: nil,
+	}
+
+	// 将请求JSON编码为字节
+	requestJSON, err := json.Marshal(request)
+	if err != nil {
+		return false, fmt.Errorf("marshal request: %w", err)
+	}
+
+	// 发送WebSocket消息
+	err = conn.WriteMessage(websocket.TextMessage, requestJSON)
+	if err != nil {
+		return false, fmt.Errorf("write: %w", err)
+	}
+
+	count := 0
+	for {
+		if count > maxRetries {
+			return false, fmt.Errorf("保持websocket连接活跃，读取websocket消息超过最大重试次数。")
+		}
+		time.Sleep(1 * time.Second)
+		_, message, err := conn.ReadMessage()
+		if err != nil {
+			c_log.GlobalLogger.Error("保持websocket连接活跃，读取websocket消息 - 失败 ", err, " 继续读取消息。")
+			return false, err
+			//continue
+		}
+
+		var response Response
+		err = json.Unmarshal(message, &response)
+		c_log.GlobalLogger.Info("response ", response)
+		if err == nil && response.Type == "response" {
+			c_log.GlobalLogger.Info("response1 ", response)
+			return true, err
+		}
+		count++
+	}
+}
+
+func sendRequestAndAwaitResponse(ws *websocket.Conn) ([]byte, error) {
+	request := Request1{
+		Type:      "request",
+		CommandID: "heart",
+		Parameter: nil,
+	}
+
+	requestJSON, err := json.Marshal(request)
+	if err != nil {
+		c_log.GlobalLogger.Error("保持websocket连接活跃，解析requestJSON - 失败。", err)
+		return nil, err
+	}
+
+	err = ws.WriteMessage(websocket.TextMessage, requestJSON)
+	if err != nil {
+		c_log.GlobalLogger.Error("保持websocket连接活跃，发送心跳请求 - 失败。", err)
+		return nil, err
+	}
+	c_log.GlobalLogger.Info("保持websocket连接活跃，发送心跳请求 - 成功。")
+
+	// 使用channel等待响应
+	responseChan := make(chan []byte)
+	go handleMessages(ws, responseChan)
+
+	select {
+	case response := <-responseChan:
+		c_log.GlobalLogger.Error("保持websocket连接活跃，等待心跳响应 - 成功。")
+		return response, nil
+	case <-time.After(60 * time.Second): // 设置超时时间
+		c_log.GlobalLogger.Error("保持websocket连接活跃，等待心跳响应 - 超时。")
+		close(responseChan)
+		return nil, fmt.Errorf("保持websocket连接活跃，等待心跳响应 - 超时。")
+	}
+}
+
+func handleMessages(ws *websocket.Conn, responseChan chan<- []byte) {
+	for {
+		time.Sleep(100 * time.Millisecond)
+		_, message, err := ws.ReadMessage()
+		if err != nil {
+			c_log.GlobalLogger.Error("保持websocket连接活跃，读取websocket消息 - 失败 ", err)
+			return
+		}
+
+		var response Response
+		if err := json.Unmarshal(message, &response); err == nil && response.Type == "response" {
+			responseChan <- message
+			close(responseChan)
+			return
+		}
+	}
+}
+
+func ConnectWebsocket() {
+	for {
+		// 防止重复调用
+		if reconnectionInProgress {
+			return
+		}
+		reconnectionInProgress = true
+		//c_log.GlobalLogger.Info("初始化Websocket连接 - 开始。")
+		serverURL := LocalConfig.Node.Ip + ":" + LocalConfig.LocalWebsocketPort
+		path := "/"
+
+		// 构建WebSocket连接URL
+		u := url.URL{Scheme: "ws", Host: serverURL, Path: path}
+		//c_log.GlobalLogger.Info("URL:", u.String())
+
+		// 创建一个Dialer实例，用于建立WebSocket连接
+		dialer := websocket.Dialer{
+			ReadBufferSize:  1024,
+			WriteBufferSize: 1024,
+			// 可选：设置超时等
+			HandshakeTimeout: 5 * time.Minute,
+		}
+
+		// 建立WebSocket连接
+		coon, _, err := dialer.Dial(u.String(), nil)
+		if err != nil {
+			fmt.Println("err:", err)
+			//c_log.GlobalLogger.Error("初始化Websocket连接 - 失败。")
+			time.Sleep(5 * time.Second)
+			reconnectionInProgress = false
+			//c_log.GlobalLogger.Info("重试连接websocket...")
+			continue
+		}
+
+		WsConn = coon
+		//c_log.GlobalLogger.Info("初始化Websocket连接 - 成功。")
+		// 连接成功，退出循环
+		reconnectionInProgress = false
+		break
+	}
+}
+
+func InitWebsocketConfig() {
+	ConnectWebsocket()
+	// 保持连接活跃
+	go keepAlive()
+}

aarch64/pjibot_clean/common/service/disk_clean.go

@@ -0,0 +1,93 @@
+package service
+
+import (
+	commonConfig "cicv-data-closedloop/aarch64/pjibot_clean/common/config"
+	masterConfig "cicv-data-closedloop/aarch64/pjibot_clean/master/package/config"
+	"cicv-data-closedloop/common/config/c_log"
+	"cicv-data-closedloop/common/domain"
+	"cicv-data-closedloop/common/entity"
+	"cicv-data-closedloop/common/util"
+	"time"
+)
+
+// 如果磁盘占用过高，则删除timeWindow和对应的文件
+func DiskClean() {
+	c_log.GlobalLogger.Info("清理timeWindow，启动！")
+	/*
+		TTL(0, "删除旧数据")；
+		STOP(1, "停止缓存");
+		LRU(2, "保留高优先级")
+	*/
+	policyToDescription := map[string]string{
+		"TTL":  "删除旧数据",
+		"STOP": "停止缓存",
+		"LRU":  "保留高优先级",
+	}
+
+	for {
+		time.Sleep(1000 * time.Millisecond)
+		// 1 获取磁盘占用
+		//diskUsed, _ := util.GetDiskUsed(commonConfig.CloudConfig.Disk.Name) // 获取整个磁盘空间
+
+		diskUsed, _ := util.GetDirectoryDiskUsed(commonConfig.CloudConfig.Disk.Path) // 获取指定目录空间
+		if diskUsed > commonConfig.CloudConfig.Disk.Used {
+			policy := commonConfig.PlatformConfig.TaskCachePolicy
+			c_log.GlobalLogger.Errorf("磁盘占用 %v 超过 %v，触发删除规则 %v", diskUsed, commonConfig.CloudConfig.Disk.Used, policyToDescription[policy])
+			// 2 获取策略
+			if policy == "TTL" {
+				// 1 获取时间窗口队列中的第二个
+				if len(entity.TimeWindowConsumerQueue) > 2 {
+					deleteTimeWindow(1)
+				}
+			} else if policy == "STOP" {
+				// 2 获取时间窗口队列中的倒数第一个
+				if len(entity.TimeWindowConsumerQueue) > 2 {
+					deleteTimeWindow(len(entity.TimeWindowConsumerQueue) - 1)
+				}
+			} else if policy == "LRU" {
+				// 3 获取优先级最低的时间窗口
+				if len(entity.TimeWindowConsumerQueue) > 2 {
+					indexToRemove := getIndexToRemoveForLRU()
+					if indexToRemove != -1 {
+						deleteTimeWindow(indexToRemove)
+					}
+				}
+			} else {
+				c_log.GlobalLogger.Error("未知的缓存策略：", policy)
+			}
+
+		}
+	}
+}
+
+func deleteTimeWindow(indexToRemove int) {
+	timeWindowToRemove := entity.TimeWindowConsumerQueue[indexToRemove]
+	// 1 删除队列中的窗口。使用切片的特性删除指定位置的元素
+	entity.TimeWindowConsumerQueueMutex.Lock()
+	entity.TimeWindowConsumerQueue = append(entity.TimeWindowConsumerQueue[:indexToRemove], entity.TimeWindowConsumerQueue[indexToRemove+1:]...)
+	entity.TimeWindowConsumerQueueMutex.Unlock()
+	// 2 删除该窗口对应的文件目录。
+	faultTime := timeWindowToRemove.FaultTime
+	dir := domain.GetCopyDir(commonConfig.CloudConfig.BagCopyDir, faultTime)
+	err := util.RemoveDir(dir)
+	if err != nil {
+		c_log.GlobalLogger.Error("删除目录", dir, "失败：", err)
+	}
+}
+
+func getIndexToRemoveForLRU() int {
+	lru := commonConfig.PlatformConfig.Lru
+	i := len(lru) - 1
+	for i >= 0 {
+		for i2, window := range entity.TimeWindowConsumerQueue {
+			for _, label := range window.Labels {
+				value, _ := masterConfig.LabelMapTriggerId.Load(label)
+				if value == lru[i] {
+					return i2
+				}
+			}
+		}
+	}
+	return -1
+
+}

aarch64/pjibot_clean/common/service/kill_self.go

@@ -0,0 +1,118 @@
+package service
+
+import (
+	commonConfig "cicv-data-closedloop/aarch64/pjibot_clean/common/config"
+	"cicv-data-closedloop/common/config/c_log"
+	"cicv-data-closedloop/common/util"
+	"net/rpc"
+	"os"
+	"sync"
+	"time"
+)
+
+var (
+	ChannelKillRosRecord  = make(chan int)
+	ChannelKillDiskClean  = make(chan int)
+	ChannelKillSubscriber = make(chan int)
+	ChannelKillMove       = make(chan int)
+	ChannelKillConsume    = make(chan int)
+
+	KillChannel = 5
+	KillTimes   = 0
+	MutexKill   sync.Mutex
+)
+
+// 停止信号，主从节点接收到数据后准备重启
+type KillSignal struct {
+	NodeName       string
+	DropUploadData bool
+	Restart        bool
+}
+
+// 定义要远程调用的类型和方法
+type KillService struct{}
+
+// 杀死自身程序，通过通道实现 方法必须满足RPC规范：函数有两个参数，第一个参数是请求，第二个是响应
+func (m *KillService) Kill(args KillSignal, reply *int) error {
+	c_log.GlobalLogger.Info("接收到自杀信号：", args)
+	// 1 杀死 rosbag record 命令
+	ChannelKillRosRecord <- 1
+	// 2 杀死所有 ros 订阅者
+	ChannelKillSubscriber <- 1
+	// 3 杀死上传任任务
+	if args.DropUploadData == true {
+		// 3-1 等待上传结束再杀死
+		ChannelKillMove <- 1
+		ChannelKillConsume <- 1
+	} else {
+		// 3-2 直接杀死
+		ChannelKillMove <- 2
+		ChannelKillConsume <- 2
+	}
+	go killDone(args.Restart)
+	return nil
+}
+func WaitKillSelf() {
+	killService := new(KillService)
+	err := rpc.Register(killService)
+	if err != nil {
+		c_log.GlobalLogger.Error("注册rpc服务失败：", err)
+		return
+	}
+
+	// 等待并处理远程调用请求
+	for {
+		conn, err := commonConfig.KillSignalListener.Accept()
+		if err != nil {
+			continue
+		}
+		go rpc.ServeConn(conn)
+	}
+}
+
+func AddKillTimes(info string) {
+	MutexKill.Lock()
+	defer MutexKill.Unlock()
+	switch info {
+	case "1":
+		ChannelKillDiskClean <- 1
+		close(ChannelKillRosRecord)
+		KillTimes++
+		c_log.GlobalLogger.Infof("已杀死 record 打包 goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
+	case "2":
+		close(ChannelKillDiskClean)
+		KillTimes++
+		c_log.GlobalLogger.Infof("已杀死 bag 包数量维护 goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
+	case "3":
+		close(ChannelKillSubscriber)
+		KillTimes++
+		c_log.GlobalLogger.Infof("已杀死 rosnode 和ros 订阅者 goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
+	case "4":
+		close(ChannelKillMove)
+		KillTimes++
+		c_log.GlobalLogger.Infof("已杀死 bag 包移动 goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
+	case "5":
+		close(ChannelKillConsume)
+		KillTimes++
+		c_log.GlobalLogger.Infof("已杀死 bag 包消费 goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
+	}
+}
+
+func killDone(restart bool) {
+	for {
+		time.Sleep(time.Duration(1) * time.Second)
+		if KillChannel == KillTimes {
+			if restart {
+				_, err := util.ExecuteWithPath(commonConfig.LocalConfig.RestartCmd.Dir, commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args...)
+				if err != nil {
+					c_log.GlobalLogger.Info("启动新程序失败,【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args, "：", err)
+					os.Exit(-1)
+				}
+				c_log.GlobalLogger.Info("数据采集任务更新，正常退出当前程序。")
+			} else {
+				c_log.GlobalLogger.Info("数据采集任务终止，正常退出当前程序。")
+			}
+			os.Exit(0)
+		}
+	}
+}

aarch64/pjibot_clean/common/service/rosbag_clean.go

@@ -0,0 +1,36 @@
+package service
+
+import (
+	"cicv-data-closedloop/aarch64/pjibot_clean/common/config"
+	"cicv-data-closedloop/common/config/c_log"
+	"cicv-data-closedloop/common/util"
+	"time"
+)
+
+// BagCacheClean 保证本地缓存的包数量不超过设定值
+func BagCacheClean() {
+	c_log.GlobalLogger.Info("启动清理缓存的 goroutine 维护目录【", config.CloudConfig.BagDataDir, "】的 bag 包数量：", config.CloudConfig.BagNumber)
+	for {
+		// 收到自杀信号
+		select {
+		case signal := <-ChannelKillDiskClean:
+			if signal == 1 {
+				AddKillTimes("2")
+				return
+			}
+		default:
+		}
+
+		// 1 ------- 每10秒清理一次 -------
+		time.Sleep(time.Duration(10) * time.Second)
+		// 2 ------- 获取目录下所有bag包 -------
+		bags, _ := util.ListAbsolutePathWithSuffixAndSort(config.CloudConfig.BagDataDir, ".bag")
+		// 3 如果打包数量超过n个，删除最旧的包{
+		if len(bags) > config.CloudConfig.BagNumber {
+			diff := len(bags) - config.CloudConfig.BagNumber
+			for i := 0; i < diff; i++ {
+				util.DeleteFile(bags[i])
+			}
+		}
+	}
+}

aarch64/pjibot_clean/common/service/rosbag_record.go

@@ -0,0 +1,140 @@
+package service
+
+import (
+	"cicv-data-closedloop/aarch64/pjibot_clean/common/config"
+	"cicv-data-closedloop/common/config/c_log"
+	"cicv-data-closedloop/common/util"
+	"github.com/bluenviron/goroslib/v2"
+	"os"
+	"os/exec"
+	"time"
+)
+
+// 打包rosbag
+func BagRecord(nodeName string) {
+	var err error
+	c_log.GlobalLogger.Info("rosbag record goroutine - 启动")
+	for {
+	startRecord:
+		c_log.GlobalLogger.Info("校验必需的 rosnode 是否全部启动。")
+		canRecord := false
+		for !canRecord {
+			time.Sleep(time.Duration(2) * time.Second)
+			canRecord = isCanRecord(config.RosNode)
+		}
+		c_log.GlobalLogger.Info("rosnode 启动完成，正在启动 rosbag record 命令。")
+
+		var command []string
+		command = append(command, "record")
+		command = append(command, "--split")
+		command = append(command, "--duration=1")
+		platformTopics := config.RecordTopics
+		for _, host := range config.CloudConfig.Hosts {
+			if host.Name == nodeName {
+				// platformConfig中配置用户想缓存的topic，配置文件中的是默认要采集的（如果平台没有填，则取默认）
+				if platformTopics == nil || len(platformTopics) == 0 {
+					c_log.GlobalLogger.Infof("平台没有配置采集话题【%v】，采集默认话题", platformTopics)
+					for _, topic2 := range host.Topics {
+						command = append(command, topic2)
+					}
+				} else {
+					c_log.GlobalLogger.Infof("平台配置了采集话题【%v】，采集配置话题", platformTopics)
+					for _, topic1 := range config.RecordTopics {
+						command = append(command, topic1)
+					}
+				}
+			}
+		}
+
+		// 2 ------- 调用 rosbag 打包命令，该命令自动阻塞 -------
+		// 不在此处压缩，因为 rosbag filter 时会报错。在上传到oss之前压缩即可。
+		// 包名格式：2023-11-15-17-35-20_0.bag
+		_ = util.CreateParentDir(config.CloudConfig.BagDataDir)
+		var recordProcessPid int
+		var recordSubProcessPid int
+		var cmd *exec.Cmd
+		//systemEnv := os.Environ()
+		//c_log.GlobalLogger.Info("系统环境变量为：", systemEnv)
+	parent:
+		for {
+			c_log.GlobalLogger.Info("record 环境变量为：", config.RosbagEnvs)
+			cmd, err = util.ExecuteWithEnvAndDirAsync(config.RosbagEnvs, config.CloudConfig.BagDataDir, config.RosbagPath, command...)
+			if err != nil {
+				c_log.GlobalLogger.Error("执行record命令", command, "出错：", err)
+				continue
+			}
+			recordProcessPid = cmd.Process.Pid
+		sub:
+			for {
+				time.Sleep(time.Duration(2) * time.Second)
+				process, err := os.FindProcess(recordProcessPid)
+				if process == nil {
+					continue parent
+				}
+				recordSubProcessPid, err = util.GetSubProcessPid(recordProcessPid)
+				if err != nil {
+					c_log.GlobalLogger.Info("正在等待获取进程 ", recordProcessPid, " 的子进程的pid。（未获取到可能由于磁盘空间已满）")
+					continue sub
+				}
+				if recordSubProcessPid != 0 {
+					c_log.GlobalLogger.Info("获取进程 ", recordProcessPid, " 的子进程的pid：", recordSubProcessPid)
+					break parent
+				}
+			}
+		}
+		// 等待进程关闭信号
+		c_log.GlobalLogger.Info("启动record命令成功。等待进程关闭信号。")
+		for {
+			select {
+			case signal := <-ChannelKillRosRecord:
+				if signal == 1 {
+					kill(recordProcessPid, recordSubProcessPid, cmd)
+					AddKillTimes("1")
+					continue // continue 是为了等待重启信号
+				}
+				if signal == 2 {
+					goto startRecord // 收到信号 2 重新 record 命令
+				}
+				if signal == 3 { // 这个关闭是等待数据处理时的关闭
+					c_log.GlobalLogger.Error("采集数据，接收record命令进程关闭信号：", signal)
+					kill(recordProcessPid, recordSubProcessPid, cmd)
+					continue
+				}
+			}
+		}
+	}
+}
+
+func kill(recordProcessPid int, recordSubProcessPid int, cmd *exec.Cmd) {
+	if err := util.KillProcessByPid(recordSubProcessPid); err != nil {
+		c_log.GlobalLogger.Errorf("程序阻塞，杀死record命令子进程出错，【pid】=%v,【err】=%v。", recordSubProcessPid, err)
+		select {} // 此处阻塞防止record命令一直录包占满存储
+	}
+	if err := cmd.Process.Kill(); err != nil {
+		c_log.GlobalLogger.Error("程序阻塞，杀死record命令进程", recordProcessPid, "出错：", err)
+		select {} // 此处阻塞防止record命令一直录包占满存储
+	}
+}
+
+func isCanRecord(n *goroslib.Node) bool {
+	time.Sleep(time.Duration(1) * time.Second)
+	// 获取
+	nodes, err := n.MasterGetNodes()
+	if err != nil {
+		c_log.GlobalLogger.Error("获取rosnode出错：", err)
+		return false
+	}
+	// 创建一个示例的map
+	myMap := nodes
+	// 创建一个切片，包含要检查的元素
+	mySlice := config.CloudConfig.Ros.Nodes
+
+	// 判断map的键是否包含切片中的所有元素
+	for _, element := range mySlice {
+		if _, ok := myMap[element]; !ok {
+			c_log.GlobalLogger.Info("rosnode：", element, " 未启动，需等待启动后才可启动record。")
+			return false
+		}
+	}
+	return true
+}

aarch64/pjibot_clean/common/service/rosbag_upload.go

@@ -0,0 +1,250 @@
+package service
+
+import (
+	commonConfig "cicv-data-closedloop/aarch64/pjibot_clean/common/config"
+	masterConfig "cicv-data-closedloop/aarch64/pjibot_clean/master/package/config"
+	"cicv-data-closedloop/common/config/c_log"
+	"cicv-data-closedloop/common/domain"
+	"cicv-data-closedloop/common/entity"
+	"cicv-data-closedloop/common/util"
+	commonUtil "cicv-data-closedloop/common/util"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+)
+
+func RunTimeWindowConsumerQueue(nodeName string) {
+	c_log.GlobalLogger.Info("处理消费者队列goroutine - 启动")
+outLoop:
+	for {
+
+		// 收到自杀信号
+		select {
+		case signal := <-ChannelKillConsume:
+			if signal == 1 {
+				ChannelKillConsume <- 1
+				if len(entity.TimeWindowConsumerQueue) == 0 {
+					AddKillTimes("5")
+					return
+				}
+			} else { //signal == 2
+				AddKillTimes("5")
+				return
+			}
+		default:
+		}
+		// 每一秒扫一次
+		time.Sleep(time.Duration(1) * time.Second)
+
+		waitLength := len(entity.TimeWindowConsumerQueue)
+		if waitLength == 0 {
+			continue outLoop
+		}
+
+		// 1 获取即将处理的窗口
+		currentTimeWindow := entity.TimeWindowConsumerQueue[0]
+		entity.RemoveHeadOfTimeWindowConsumerQueue()
+		c_log.GlobalLogger.Infof("开始处理窗口，【Lable】=%v，【FaultTime】=%v，【Length】=%v", currentTimeWindow.Labels, currentTimeWindow.FaultTime, currentTimeWindow.Length)
+		// 2 获取目录
+		dir := domain.GetCopyDir(commonConfig.CloudConfig.BagCopyDir, currentTimeWindow.FaultTime)
+		bags, _ := util.ListAbsolutePathWithSuffixAndSort(dir, ".bag")
+		bagNumber := len(bags)
+		if bagNumber > currentTimeWindow.Length {
+			bagNumber = currentTimeWindow.Length
+			bags = bags[0:currentTimeWindow.Length]
+		}
+
+		// 3 如果不是全量采集，则使用 filter 命令对 bag 包进行主题过滤。
+		if commonConfig.CloudConfig.FullCollect == false {
+			var filterTopics []string
+			if nodeName == commonConfig.CloudConfig.Hosts[0].Name {
+				filterTopics = currentTimeWindow.MasterTopics
+			} else {
+				filterTopics = currentTimeWindow.SlaveTopics
+			}
+			var topicsFilterSlice []string
+			for _, topic := range filterTopics {
+				topicsFilterSlice = append(topicsFilterSlice, "topic=='"+topic+"'")
+			}
+			for i, bag := range bags {
+				oldName := bag
+				newName := bag + "_filter"
+				filterCommand := []string{"filter", oldName, newName, "\"" + strings.Join(topicsFilterSlice, " or ") + "\""}
+				_, output, err := util.ExecuteWithEnvSync(commonConfig.RosbagEnvs, commonConfig.RosbagPath, filterCommand...)
+				c_log.GlobalLogger.Info("正在过滤中，【FaultTime】=", currentTimeWindow.FaultTime, "【Label】=", currentTimeWindow.Labels, "，进度", i+1, "/", bagNumber, "。")
+				if err != nil {
+					c_log.GlobalLogger.Errorf("filter命令执行出错【命令】=%v，【输出】=%v,【err】=%v", filterCommand, output, err)
+					continue
+				}
+				// 删除旧文件
+				util.DeleteFile(oldName)
+				// 将新文件改回旧文件名
+				if err = os.Rename(newName, oldName); err != nil {
+					c_log.GlobalLogger.Info("修改文件名", oldName, "失败，放弃当前时间窗口", currentTimeWindow.FaultTime, "，错误为：", err)
+					continue outLoop
+				}
+			}
+		}
+
+		// 4 compress包，必须顺序执行，此时每个包会对应生成一个压缩过的包和原始包，原始包后缀为.orig.bag
+		// 5 todo 机器人去掉压缩过程，防止cpu跑满
+		//c_log.GlobalLogger.Info("压缩 bag 数据包，故障时间为：", currentTimeWindow.FaultTime)
+		//for i, bag := range bags {
+		//	oldName := bag
+		//	compressCommand := []string{"compress", "--bz2", oldName}
+		//	c_log.GlobalLogger.Info("正在压缩中，【FaultTime】=", currentTimeWindow.FaultTime, "【Label】=", currentTimeWindow.Labels, "，进度", i+1, "/", bagNumber, "。")
+		//	if _, output, err := util.ExecuteWithEnvSync(commonConfig.RosbagEnvs, commonConfig.RosbagPath, compressCommand...); err != nil {
+		//		c_log.GlobalLogger.Errorf("compress命令执行出错【命令】=%v，【输出】=%v,【err】=%v", compressCommand, output, err)
+		//		continue
+		//	}
+		//}
+		// 5 upload，必须顺序执行
+		c_log.GlobalLogger.Info("发送bag数据包，故障时间为：", currentTimeWindow.FaultTime)
+		start := time.Now()
+		objectKey1 := commonConfig.LocalConfig.OssBasePrefix + commonConfig.LocalConfig.EquipmentNo + "/data/" + currentTimeWindow.FaultTime + "_" + strings.Join(currentTimeWindow.Labels, "_") + "_" + fmt.Sprintf("%d", bagNumber) + "/"
+		objectKey2 := commonConfig.LocalConfig.OssBasePrefix + commonConfig.LocalConfig.EquipmentNo + "/data_merge/" + currentTimeWindow.FaultTime + "_" + strings.Join(currentTimeWindow.Labels, "_") + "_" + fmt.Sprintf("%d", bagNumber) + ".bag"
+		objectKey3 := commonConfig.LocalConfig.OssBasePrefix + commonConfig.LocalConfig.EquipmentNo + "/data_parse/" + currentTimeWindow.FaultTime + "_" + strings.Join(currentTimeWindow.Labels, "_") + "_" + fmt.Sprintf("%d", bagNumber) + "/"
+		for i, bag := range bags {
+			startOne := time.Now()
+			bagSlice := strings.Split(bag, "/")
+			for {
+				commonConfig.OssMutex.Lock()
+				err := commonConfig.OssBucket.PutObjectFromFile(objectKey1+bagSlice[len(bagSlice)-1], bag)
+				commonConfig.OssMutex.Unlock()
+				if err != nil {
+					c_log.GlobalLogger.Info("因网络原因上传包 ", bag, " 时报错，需要等待网络恢复后重新上传：", err)
+					continue
+				}
+				c_log.GlobalLogger.Info("上传耗时 ", time.Since(startOne), "，【FaultTime】=", currentTimeWindow.FaultTime, "【Label】=", currentTimeWindow.Labels, "，进度", i+1, "/", bagNumber, "。【", bag, "】-------【", objectKey1+bagSlice[len(bagSlice)-1], "】")
+				break
+			}
+		}
+		c_log.GlobalLogger.Info("上传完成，花费时间：", time.Since(start))
+		// 在上传完成的包目录同级下添加一个目录同名的json
+		triggerIds := make([]string, 0)
+		for _, label := range currentTimeWindow.Labels {
+			if value, ok := masterConfig.LabelMapTriggerId.Load(label); ok {
+				triggerIds = append(triggerIds, value.(string))
+			}
+		}
+		callBackMap := map[string]interface{}{
+			"dataName":    currentTimeWindow.FaultTime, // 云端callback程序会将该值加8小时，因为UTC和CSV时区相差8小时
+			"dataSize":    "",                          // 由合并程序补充
+			"equipmentNo": commonConfig.LocalConfig.EquipmentNo,
+			"secretKey":   commonConfig.LocalConfig.SecretKey,
+			"rosBagPath":  objectKey2,
+			"filePath":    objectKey3,
+			"taskId":      commonConfig.PlatformConfig.TaskConfigId,
+			"triggerId":   triggerIds,
+		}
+		callBackJson, err := util.MapToJsonString(callBackMap)
+		if err != nil {
+			c_log.GlobalLogger.Error("callBackMap", callBackMap, "转json失败：", err)
+		}
+		commonConfig.OssMutex.Lock()
+		// 上传callback.json
+		err = commonConfig.OssBucket.PutObject(objectKey3+"callback.json", strings.NewReader(callBackJson))
+		if err != nil {
+			c_log.GlobalLogger.Error("上传 callback.json 文件失败：", err)
+		}
+		// 额外采集mapBuf
+		for _, file := range commonConfig.CloudConfig.MapBufFiles {
+			err = commonConfig.OssBucket.PutObjectFromFile(objectKey3+filepath.Base(file), file)
+			if err != nil {
+				c_log.GlobalLogger.Error("上传 mapBuf 文件失败：", err)
+			}
+		}
+
+		// 压缩采集data目录
+		{
+			// 1 如果 data.zip 已存在，先删除
+			util.DeleteFileIfExists(commonConfig.CloudConfig.DataDir.Dest)
+			c_log.GlobalLogger.Infof("旧的data目录压缩包【%v】已删除。", commonConfig.CloudConfig.DataDir.Dest)
+			// 2 重新压缩升成 data.zip
+			err = util.ZipDir(commonConfig.CloudConfig.DataDir.Src, commonConfig.CloudConfig.DataDir.Dest, commonConfig.CloudConfig.DataDir.SrcSub)
+			if err != nil {
+				c_log.GlobalLogger.Error("压缩data目录失败：", err)
+			} else {
+				c_log.GlobalLogger.Infof("压缩data目录【%v】->【%v】成功", commonConfig.CloudConfig.DataDir.Src, commonConfig.CloudConfig.DataDir.Dest)
+				dataZipKey := objectKey3 + "data.zip"
+				err = commonConfig.OssBucket.PutObjectFromFile(dataZipKey, commonConfig.CloudConfig.DataDir.Dest)
+				if err != nil {
+					c_log.GlobalLogger.Error("上传data目录压缩文件失败：", err)
+				} else {
+					c_log.GlobalLogger.Infof("上传data目录压缩包【%v】->【%v】成功", commonConfig.CloudConfig.DataDir.Dest, dataZipKey)
+				}
+			}
+			commonConfig.OssMutex.Unlock()
+		}
+		// todo 不压缩采集data目录
+		{
+			//var filePaths []string                                                                                           // 初始化一个切片来保存文件路径
+			//err = filepath.WalkDir(commonConfig.CloudConfig.DataDir.Src, func(path string, d fs.DirEntry, err error) error { // 使用filepath.WalkDir遍历目录
+			//	if err != nil {
+			//		return err // 如果有错误，返回错误
+			//	}
+			//
+			//	// 检查是否为文件（跳过目录）
+			//	if !d.IsDir() {
+			//		filePaths = append(filePaths, path) // 将文件路径添加到切片中
+			//	}
+			//	return nil
+			//})
+			//if err != nil {
+			//	c_log.GlobalLogger.Error("扫描 data 目录失败：", err)
+			//	goto outLoop
+			//}
+			//
+			//// 不压缩上传所有文件
+			//for _, path := range filePaths {
+			//	if strings.Contains(path, commonConfig.CloudConfig.DataDir.Exclude) {
+			//		continue
+			//	}
+			//	relativePath := strings.Replace(path, commonConfig.CloudConfig.DataDir.Src, "", 1)
+			//	ossKey := objectKey3 + "data/" + relativePath
+			//	err = commonConfig.OssBucket.PutObjectFromFile(ossKey, path)
+			//	if err != nil {
+			//		c_log.GlobalLogger.Errorf("上传 data 目录内文件【%v】->【%v】失败：%v", path, ossKey, err)
+			//		goto outLoop
+			//	}
+			//}
+			//commonConfig.OssMutex.Unlock()
+		}
+
+		// 数据库中采集数量加一
+		collectNumPlus()
+		// 删除本地所有已上传的bag文件
+		c_log.GlobalLogger.Infof("结束处理窗口，【Lable】=%v，【FaultTime】=%v，【Length】=%v", currentTimeWindow.Labels, currentTimeWindow.FaultTime, currentTimeWindow.Length)
+		c_log.GlobalLogger.Infof("待处理窗口个数为：%v", len(entity.TimeWindowConsumerQueue))
+		if err = util.RemoveDir(dir); err != nil {
+			goto outLoop
+		}
+		if len(entity.TimeWindowConsumerQueue) == 0 {
+			c_log.GlobalLogger.Infof("已处理所有窗口，重启 record 命令。")
+			ChannelKillRosRecord <- 2
+			entity.ProcessingFlag = false
+		}
+	}
+}
+
+func collectNumPlus() {
+	responseString, err := commonUtil.HttpPostJsonWithHeaders(
+		commonConfig.CloudConfig.CollectNumPlus.Url,
+		map[string]string{"Authorization": "U9yKpD6kZZDDe4LFKK6myAxBUT1XRrDM"},
+		map[string]string{
+			"snCode": commonConfig.LocalConfig.SecretKey,
+		},
+	)
+	if err != nil {
+		c_log.GlobalLogger.Error("发送http请求修改采集数量失败：", err)
+	}
+	// 解析JSON字符串到Response结构体
+	var resp entity.Response
+	err = json.Unmarshal([]byte(responseString), &resp)
+	if err != nil {
+		c_log.GlobalLogger.Error("解析修改采集数量结果失败：", err)
+	}
+}

aarch64/pjibot_clean/common/variable/application.go

@@ -0,0 +1,6 @@
+package variable
+
+var (
+	LogDir          = "/root/cicv-data-closedloop/log/"
+	LocalConfigPath = "/root/cicv-data-closedloop/config/local-config.yaml"
+)

aarch64/pjibot_clean/control/main.go

@@ -0,0 +1,225 @@
+package main
+
+import (
+	commonConfig "cicv-data-closedloop/aarch64/pjibot_clean/common/config"
+	commonService "cicv-data-closedloop/aarch64/pjibot_clean/common/service"
+	"cicv-data-closedloop/aarch64/pjibot_clean/common/variable"
+	"cicv-data-closedloop/aarch64/pjibot_clean/control/pkg"
+	"cicv-data-closedloop/common/config/c_log"
+	"cicv-data-closedloop/common/entity"
+	"cicv-data-closedloop/common/util"
+	commonUtil "cicv-data-closedloop/common/util"
+	"encoding/json"
+	"fmt"
+	"net/rpc"
+	"os"
+	"runtime"
+	"time"
+)
+
+var (
+	applicationName  = "pji-control"
+	localStatus      = "idle"
+	cloudStatus      = "NONE"
+	lastLocalStatus  = "idle"
+	lastCloudStatus  = "NONE"
+	limitReachedFlag = false
+	localTurnLength  = 1  // s，本地状态刷新时间
+	cloudTurnLength  = 60 // s，云端状态刷新时间
+	renewTurnLength  = 3  // s，续约状态刷新时间
+	waitStopLength   = 1  // min，停止master前等待时间
+	checkLimitLength = 1  // min, 查询是否达到采集限制等待时间
+	launchedFlag     = false
+	renewedFlag      = false
+	renewTimer       *time.Timer // 续约定时器
+	RenewDur         = 5         // min， 续约时间
+	maxRetryCount    = 10        // 查询配置最大重试次数
+)
+
+func init() {
+	runtime.GOMAXPROCS(1)
+	// 初始化日志配置
+	c_log.InitLog(variable.LogDir, applicationName)
+	// 初始化本地配置文件（第1处配置，在本地文件）
+	commonConfig.InitLocalConfig(variable.LocalConfigPath)
+	// 初始化Oss连接信息
+	commonConfig.InitOssConfig()
+	// 初始化业务逻辑配置信息，配置文件在oss上（第2处配置，在oss文件）
+	commonConfig.InitCloudConfig()
+	// 初始化rpc客户端，用于杀死旧的采集程序
+
+	// 初始化websocket配置
+	commonConfig.InitWebsocketConfig()
+}
+
+func checkCollectLimit() {
+	for {
+		time.Sleep(time.Duration(checkLimitLength) * time.Minute)
+		// 如果开启了采集频率限制，则云端判断采集数量是否超过限额
+		if commonConfig.CloudConfig.CollectLimit.Flag == 1 {
+			//c_log.GlobalLogger.Error("当前设备已开启数采频率限制，需判断采集数量是否达到限额。")
+			responseString, err := commonUtil.HttpPostJsonWithHeaders(
+				commonConfig.CloudConfig.CollectLimit.Url,
+				map[string]string{"Authorization": "U9yKpD6kZZDDe4LFKK6myAxBUT1XRrDM"},
+				map[string]string{
+					"snCode":            commonConfig.LocalConfig.SecretKey,
+					"collectLimitDay":   util.ToString(commonConfig.CloudConfig.CollectLimit.Day),
+					"collectLimitWeek":  util.ToString(commonConfig.CloudConfig.CollectLimit.Week),
+					"collectLimitMonth": util.ToString(commonConfig.CloudConfig.CollectLimit.Month),
+					"collectLimitYear":  util.ToString(commonConfig.CloudConfig.CollectLimit.Year),
+				},
+			)
+			if err != nil {
+				c_log.GlobalLogger.Error("发送http请求获取是否允许采集失败：", err)
+				continue
+			}
+			// 解析JSON字符串到Response结构体
+			var resp entity.Response
+			err = json.Unmarshal([]byte(responseString), &resp)
+			if err != nil {
+				c_log.GlobalLogger.Error("解析是否允许采集接口返回结果失败：", err)
+				continue
+			}
+			if resp.Code != 200 { // 不是200 代表采集数量已超过限额不允许采集
+				c_log.GlobalLogger.Info("采集数量已超过限额,", resp.Code)
+				limitReachedFlag = true
+				continue
+			}
+		}
+		limitReachedFlag = false
+	}
+}
+
+func initRenew() {
+	c_log.GlobalLogger.Info("启动定时器 - 开始。")
+	if renewTimer != nil {
+		renewTimer.Stop()
+	}
+	renewedFlag = true
+	renewTimer = time.AfterFunc(time.Duration(RenewDur)*time.Minute, func() {
+		renewedFlag = false
+	})
+	c_log.GlobalLogger.Infof("定时时间【%v】分钟 - 成功。", RenewDur)
+}
+
+func renew() {
+	for {
+		time.Sleep(time.Duration(renewTurnLength) * time.Second)
+		if localStatus == "running" && launchedFlag && !renewedFlag && !limitReachedFlag { // 设备处于运行状态,数采程序已启动，且尚未续约
+			c_log.GlobalLogger.Info("设备仍处于运行状态，续约 - 开始。")
+			if renewTimer != nil {
+				renewTimer.Stop()
+			}
+			renewedFlag = true
+			renewTimer = time.AfterFunc(time.Duration(RenewDur)*time.Minute, func() {
+				renewedFlag = false
+			})
+			c_log.GlobalLogger.Infof("续约时间【%v】分钟 - 成功。", RenewDur)
+		}
+	}
+}
+
+func startMasterNode() {
+	c_log.GlobalLogger.Info("获取数据闭环平台最新配置。")
+
+	if commonConfig.CheckPlatformConfigStatus(maxRetryCount) {
+		c_log.GlobalLogger.Info("查询到数据闭环平台有配置任务。")
+		commonConfig.InitPlatformConfig()
+
+		if _, err := util.ExecuteWithPath(commonConfig.LocalConfig.RestartCmd.Dir, commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args...); err != nil {
+			c_log.GlobalLogger.Info("启动新程序失败,【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args, "：", err)
+			os.Exit(-1)
+		}
+		c_log.GlobalLogger.Info("启动任务，本地执行启动命令：【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args)
+
+		initRenew()
+		launchedFlag = true
+
+		c_log.GlobalLogger.Info("数采程序启动 - 成功。")
+	} else {
+		c_log.GlobalLogger.Error("查询到数据闭环平台没有配置任务，不启动数采程序。")
+	}
+
+}
+
+func stopMasterNode() {
+	// 发送rpc信号杀死采集程序
+	var killArgs commonService.KillSignal
+	killArgs = commonService.KillSignal{NodeName: "master", DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: false}
+	c_log.GlobalLogger.Info("杀死任务，发送rpc结束信号：", killArgs)
+	KillRpcClient, err := rpc.Dial("tcp", commonConfig.LocalConfig.Node.Ip+":"+commonConfig.CloudConfig.RpcPort)
+	if err != nil {
+		// 此处如果连接失败说明采集程序已经停止了
+		lastCloudStatus = "NONE"
+		c_log.GlobalLogger.Error("采集程序已经停止：", err)
+		return
+	}
+
+	reply := 0
+	if err = KillRpcClient.Call("KillService.Kill", killArgs, &reply); err != nil {
+		c_log.GlobalLogger.Error("发送 rpc 请求到 master 报错：", err)
+	}
+
+	c_log.GlobalLogger.Info("结束任务后，将数据闭环平台配置置空。")
+	commonConfig.PlatformConfig = commonConfig.PlatformConfigStruct{}
+	if err = KillRpcClient.Close(); err != nil {
+		// 不做处理
+	}
+
+	launchedFlag = false
+	c_log.GlobalLogger.Info("数采程序关闭 - 成功。")
+}
+
+func main() {
+	// 更新本地任务状态
+	go pkg.GetLocalStatus(&localStatus, &lastLocalStatus, localTurnLength)
+	// 更新云端任务状态
+	go pkg.GetCloudStatus(&cloudStatus, &lastCloudStatus, cloudTurnLength)
+
+	// 定期检查本地任务状态，执行续约，避免短时间内多次启停
+	go renew()
+
+	// 检查是否达到限额
+	go checkCollectLimit()
+
+	// 云端任务状态负责更新配置
+	go pkg.GetCloudConfig(&cloudStatus, &lastCloudStatus, cloudTurnLength)
+
+	for {
+		if launchedFlag { // 当前已启动master节点
+			time.Sleep(time.Duration(cloudTurnLength) * time.Second)
+		} else {
+			time.Sleep(time.Duration(localTurnLength) * time.Second)
+		}
+
+		fmt.Println("localStatus: ", localStatus, "lastLocalStatus: ", lastLocalStatus)
+		fmt.Println("cloudStatus: ", cloudStatus, "lastCloudStatus: ", lastCloudStatus)
+		fmt.Println("limitReachedFlag: ", limitReachedFlag)
+
+		// 综合判断 cloudStatus 和 localStatus
+		// cloudStatus
+		// UN_CHANGE 没有新的任务，无需更改
+		// CHANGE 有新的任务，需要杀死旧的数采任务并重启
+		// NONE 设备没有配置任务，需要杀死旧的数采任务
+		// localStatus
+		// idle 空闲状态，此状态下不启动数采任务
+		// running 繁忙状态，此状态需要启动数采任务
+		// error 错误状态，此状态下不启动数采任务
+
+		// 本地任务状态负责启停master
+		if localStatus == "running" && !launchedFlag && !limitReachedFlag {
+			// 目前未启动数采程序
+			c_log.GlobalLogger.Info("数采程序启动 - 开始。")
+			startMasterNode()
+		} else if localStatus == "idle" || limitReachedFlag {
+			if !renewedFlag && launchedFlag && len(entity.TimeWindowConsumerQueue) == 0 {
+				time.Sleep(time.Duration(waitStopLength) * time.Minute)
+				c_log.GlobalLogger.Info("设备不在运行状态且没有待处理的数据，数采程序关闭 - 开始。")
+				stopMasterNode()
+			}
+		} else if localStatus == "error" {
+			c_log.GlobalLogger.Error("设备运行状态出错，停止数采程序。")
+			stopMasterNode()
+		}
+	}
+}

aarch64/pjibot_clean/control/pkg/judge_cloud.go

@@ -0,0 +1,45 @@
+package pkg
+
+import (
+	commonConfig "cicv-data-closedloop/aarch64/pjibot_clean/common/config"
+	"cicv-data-closedloop/common/config/c_log"
+	"time"
+)
+
+var (
+	maxRetryCount = 10
+)
+
+// UN_CHANGE 没有新的任务
+// CHANGE 有新的任务
+// NONE 设备没有配置任务
+func GetCloudStatus(cloudStatus *string, lastCloudStatus *string, turnLength int) {
+	// 轮询云端任务状态
+	for {
+		time.Sleep(time.Duration(turnLength) * time.Second)
+
+		*lastCloudStatus = *cloudStatus
+		taskStatus, err := commonConfig.GetStatus(commonConfig.PlatformConfig.TaskConfigId)
+		if err != nil {
+			c_log.GlobalLogger.Error("获取云端配置status失败：", err)
+			continue
+		}
+		if taskStatus == "" || taskStatus == " " {
+			taskStatus = "NONE"
+		}
+		*cloudStatus = taskStatus
+	}
+}
+
+func GetCloudConfig(cloudStatus *string, lastCloudStatus *string, turnLength int) {
+	for {
+		time.Sleep(time.Duration(turnLength) * time.Second)
+		if *cloudStatus == "CHANGE" {
+			c_log.GlobalLogger.Error("cloudStatus：", *cloudStatus)
+			if commonConfig.CheckPlatformConfigStatus(maxRetryCount) {
+				c_log.GlobalLogger.Info("查询到数据闭环平台有配置任务。")
+				commonConfig.InitPlatformConfig()
+			}
+		}
+	}
+}

aarch64/pjibot_clean/control/pkg/judge_local.go

@@ -0,0 +1,44 @@
+package pkg
+
+import (
+	"cicv-data-closedloop/aarch64/pjibot_clean/common/config"
+	"encoding/json"
+	"fmt"
+	"log"
+	"time"
+)
+
+// idle 空闲状态，此状态下机器人可进行任务下发
+// running 繁忙状态，此状态机器人不接受新任务
+// error 错误状态（硬件，不能正常工作的）
+func GetLocalStatus(localStatus *string, lastLocalStatus *string, turnLength int) {
+	defer config.WsConn.Close()
+	// 轮询本地任务状态
+	for {
+		time.Sleep(time.Duration(turnLength) * time.Second)
+
+		*lastLocalStatus = *localStatus
+		_, msg, err := config.WsConn.ReadMessage()
+		if err != nil {
+			log.Println("Error in receive:", err)
+			continue
+		}
+		//log.Printf("Received: %s\n", msg)
+
+		// 将响应字节解码为JSON
+		var statusMessage config.StatusMessage
+		err = json.Unmarshal(msg, &statusMessage)
+		if err != nil {
+			log.Println("Error in json:", err)
+			continue
+		}
+
+		if statusMessage.Type == "push" && statusMessage.Topic == "robotStatus" {
+			//fmt.Println("statusMessage:", statusMessage)
+			data := statusMessage.Data.(map[string]interface{})
+			//fmt.Println("statusMessage.Data", data)
+			fmt.Println("statusMessage.Data[\"status\"]", data["status"])
+			*localStatus = data["status"].(string)
+		}
+	}
+}

aarch64/pjibot_clean/master/main.go

@@ -0,0 +1,62 @@
+package main
+
+import (
+	commonConfig "cicv-data-closedloop/aarch64/pjibot_clean/common/config"
+	commonService "cicv-data-closedloop/aarch64/pjibot_clean/common/service"
+	"cicv-data-closedloop/aarch64/pjibot_clean/common/variable"
+	masterConfig "cicv-data-closedloop/aarch64/pjibot_clean/master/package/config"
+	masterService "cicv-data-closedloop/aarch64/pjibot_clean/master/package/service"
+	"cicv-data-closedloop/common/config/c_log"
+	"cicv-data-closedloop/common/util"
+	"runtime"
+	"time"
+)
+
+var applicationName = "pji-master"
+
+func init() {
+	runtime.GOMAXPROCS(1)
+	// 初始化日志配置
+	c_log.InitLog(variable.LogDir, applicationName)
+	// 初始化本地配置文件（第1处配置，在本地文件）
+	commonConfig.InitLocalConfig(variable.LocalConfigPath)
+	// 初始化Oss连接信息
+	commonConfig.InitOssConfig()
+	// 初始化业务逻辑配置信息，配置文件在oss上（第2处配置，在oss文件）
+	commonConfig.InitCloudConfig()
+	_ = util.RemoveSubFiles(commonConfig.CloudConfig.BagDataDir)
+	_ = util.RemoveSubFiles(commonConfig.CloudConfig.BagCopyDir)
+	go commonConfig.RefreshCloudConfig()
+	// 初始化数据闭环平台的配置（第3处配置，在数据闭环平台接口）
+	commonConfig.InitPlatformConfig()
+	// 初始化ros节点
+	commonConfig.InitRosConfig()
+	// 发送资源占用信息
+	go commonConfig.SendResourceUsage()
+	// 维护data目录缓存的包数量
+	go commonService.BagCacheClean()
+	// 磁盘占用过高时根据缓存策略处理copy目录
+	go commonService.DiskClean()
+	masterConfig.InitTriggerConfig()
+	commonConfig.InitKillSignalListener(commonConfig.CloudConfig.Hosts[0].Ip)
+	// 等待重启，接收到重启信号，会把信号分发给以下channel
+	go commonService.WaitKillSelf()
+	// 先采集地图bag包
+	masterService.CollectOneMsg()
+}
+
+func main() {
+
+	// 1 负责打包数据到data目录
+	go commonService.BagRecord(commonConfig.CloudConfig.Hosts[0].Name)
+	time.Sleep(time.Duration(10) * time.Second)
+	// 2 负责监控故障，并修改timeWindow
+	go masterService.PrepareTimeWindowProducerQueue()
+	// 3 将时间窗口内的包全部move出去，并等待当前时间窗口结束触发上传
+	go masterService.RunTimeWindowProducerQueue()
+	// 4 排队运行时间窗口
+	go commonService.RunTimeWindowConsumerQueue(commonConfig.CloudConfig.Hosts[0].Name)
+
+	// 阻塞主线程，等待其他线程执行。
+	select {}
+}

aarch64/pjibot_clean/master/package/config/trigger_init.go

@@ -0,0 +1,142 @@
+package config
+
+import (
+	"cicv-data-closedloop/aarch64/pjibot_clean/common/config"
+	"cicv-data-closedloop/common/config/c_log"
+	"cicv-data-closedloop/common/util"
+	"cicv-data-closedloop/pjibot_clean_msgs"
+	"github.com/bluenviron/goroslib/v2/pkg/msgs/diagnostic_msgs"
+	"github.com/bluenviron/goroslib/v2/pkg/msgs/nav_msgs"
+	"github.com/bluenviron/goroslib/v2/pkg/msgs/sensor_msgs"
+	"github.com/bluenviron/goroslib/v2/pkg/msgs/std_msgs"
+	"plugin"
+	"slices"
+)
+
+func InitTriggerConfig() {
+	config.OssMutex.Lock()
+	defer config.OssMutex.Unlock()
+	triggerLocalPathsMapTriggerId := make(map[string]string)
+	c_log.GlobalLogger.Info("主节点加载触发器插件 - 开始。")
+	// 1 获取数采任务的触发器列表
+	cloudTriggers := &config.PlatformConfig.TaskTriggers
+	// 2 获取本地触发器列表（触发器目录的一级子目录为【触发器ID_触发器Label】）
+	localTriggerIds := util.GetFirstLevelSubdirectories(config.CloudConfig.TriggersDir)
+	// 3 对比触发器列表，本地没有的则下载
+	for _, trigger := range *cloudTriggers {
+		id := util.ToString(trigger.TriggerId)
+		hasIdDir := slices.Contains(localTriggerIds, id) // 改成了 slices 工具库
+		triggerLocalDir := config.CloudConfig.TriggersDir + id + "/"
+		hasLabelSo, soPaths := util.CheckSoFilesInDirectory(triggerLocalDir)
+		var triggerLocalPath string
+		if hasIdDir && hasLabelSo { // 已存在的触发器需要判断是否大小一致
+			triggerLocalPath = soPaths[0]
+			ossSize, _ := util.GetOSSFileSize(config.OssBucket, trigger.TriggerScriptPath)
+			localSize, _ := util.GetFileSize(triggerLocalPath)
+			if ossSize == localSize {
+				c_log.GlobalLogger.Info("触发器插件 ", triggerLocalPath, " 存在且与云端触发器大小一致。")
+				triggerLocalPathsMapTriggerId[triggerLocalPath] = id
+				continue
+			}
+		}
+		label := util.GetFileNameWithoutExtension(config.CloudConfig.TriggersDir + trigger.TriggerScriptPath)
+		triggerLocalPath = config.CloudConfig.TriggersDir + id + "/" + label + ".so"
+		c_log.GlobalLogger.Info("开始下载触发器插件从 ", trigger.TriggerScriptPath, " 到 ", triggerLocalPath)
+		_ = util.CreateParentDir(triggerLocalPath)
+		for {
+			if err := config.OssBucket.GetObjectToFile(trigger.TriggerScriptPath, triggerLocalPath); err != nil {
+				c_log.GlobalLogger.Error("下载触发器插件失败，再次尝试：", err)
+				continue
+			}
+			break
+		}
+		triggerLocalPathsMapTriggerId[triggerLocalPath] = id
+	}
+
+	success := 0
+	// 加载所有触发器的文件
+	for triggerLocalPath, triggerId := range triggerLocalPathsMapTriggerId {
+		// 载入插件到数组
+		open, err := plugin.Open(triggerLocalPath)
+		if err != nil {
+			c_log.GlobalLogger.Error("加载本地插件", triggerLocalPath, "失败。", err)
+			continue
+		}
+		topic0, err := open.Lookup("Topic")
+		if err != nil {
+			c_log.GlobalLogger.Error("加载本地插件", triggerLocalPath, "中的Topic方法失败。", err)
+			continue
+		}
+		topic1, ok := topic0.(func() string)
+		if ok != true {
+			c_log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func() string):", err)
+			continue
+		}
+		topic2 := topic1()
+		rule, err := open.Lookup("Rule")
+		if err != nil {
+			c_log.GlobalLogger.Error("加载本地插件", triggerLocalPath, "中的Rule方法失败。", err)
+			continue
+		}
+		if TopicOfDiagnostics == topic2 { // 1
+			if f, ok1 := rule.(func(*diagnostic_msgs.DiagnosticArray) string); ok1 {
+				RuleOfDiagnostics = append(RuleOfDiagnostics, f)
+				goto JudgeDone
+			}
+			log(triggerLocalPath)
+			continue
+		} else if TopicOfImu == topic2 { // 2
+			if f, ok1 := rule.(func(data *sensor_msgs.Imu) string); ok1 {
+				RuleOfImu = append(RuleOfImu, f)
+				goto JudgeDone
+			}
+			log(triggerLocalPath)
+			continue
+		} else if TopicOfLocateInfo == topic2 { // 3
+			if f, ok1 := rule.(func(data *pjibot_clean_msgs.LocateInfo) string); ok1 {
+				RuleOfLocateInfo = append(RuleOfLocateInfo, f)
+				goto JudgeDone
+			}
+			log(triggerLocalPath)
+			continue
+		} else if TopicOfObstacleDetection == topic2 { // 4
+			if f, ok1 := rule.(func(data *std_msgs.UInt8) string); ok1 {
+				RuleOfObstacleDetection = append(RuleOfObstacleDetection, f)
+				goto JudgeDone
+			}
+			log(triggerLocalPath)
+			continue
+		} else if TopicOfOdom == topic2 { // 5
+			if f, ok1 := rule.(func(data *nav_msgs.Odometry) string); ok1 {
+				RuleOfOdom = append(RuleOfOdom, f)
+				goto JudgeDone
+			}
+			log(triggerLocalPath)
+			continue
+		} else if TopicOfSysInfo == topic2 { // 6
+			if f, ok1 := rule.(func(data *pjibot_clean_msgs.SysInfo) string); ok1 {
+				RuleOfSysInfo = append(RuleOfSysInfo, f)
+				goto JudgeDone
+			}
+			log(triggerLocalPath)
+			continue
+		} else {
+			c_log.GlobalLogger.Error("未知的topic：", topic2)
+			continue
+		}
+	JudgeDone:
+		label, err := open.Lookup("Label")
+		if err != nil {
+			c_log.GlobalLogger.Error("加载本地插件 ", triggerLocalPath, " 中的 Label 方法失败。", err)
+			continue
+		}
+		labelFunc := label.(func() string)
+		labelString := labelFunc()
+		LabelMapTriggerId.Store(labelString, triggerId)
+		success++
+	}
+	c_log.GlobalLogger.Info("一共应加载", len(config.PlatformConfig.TaskTriggers), "个触发器，实际加载 ", success, " 个触发器。")
+}
+func log(triggerLocalPath string) {
+	c_log.GlobalLogger.Error("插件", triggerLocalPath, "中的 Rule 方法参数格式不正确。")
+}

aarch64/pjibot_clean/master/package/config/trigger_var.go

@@ -0,0 +1,43 @@
+package config
+
+import (
+	"cicv-data-closedloop/pjibot_clean_msgs"
+	"github.com/bluenviron/goroslib/v2/pkg/msgs/diagnostic_msgs"
+	"github.com/bluenviron/goroslib/v2/pkg/msgs/nav_msgs"
+	"github.com/bluenviron/goroslib/v2/pkg/msgs/sensor_msgs"
+	"github.com/bluenviron/goroslib/v2/pkg/msgs/std_msgs"
+	"sync"
+)
+
+var (
+	LabelMapTriggerId = new(sync.Map)
+
+	// 1
+	TopicOfDiagnostics = "/diagnostics"
+	RuleOfDiagnostics  []func(data *diagnostic_msgs.DiagnosticArray) string
+	// 2
+	TopicOfImu = "/imu"
+	RuleOfImu  []func(data *sensor_msgs.Imu) string
+	// 3
+	TopicOfLocateInfo = "/locate_info"
+	RuleOfLocateInfo  []func(data *pjibot_clean_msgs.LocateInfo) string
+	// 4
+	TopicOfObstacleDetection = "/obstacle_detection"
+	RuleOfObstacleDetection  []func(data *std_msgs.UInt8) string
+	// 5
+	TopicOfOdom = "/odom"
+	RuleOfOdom  []func(data *nav_msgs.Odometry) string
+	// 6
+	TopicOfSysInfo = "/sys_info"
+	RuleOfSysInfo  []func(data *pjibot_clean_msgs.SysInfo) string
+	// todo 这里是全量的topic，添加topic则需要同时在下面的数组添加；也需要在produce_window.go中添加新的订阅者
+	AllTopics = []string{
+		TopicOfDiagnostics,       // 1
+		TopicOfImu,               // 2
+		TopicOfLocateInfo,        // 3
+		TopicOfObstacleDetection, // 4
+		TopicOfOdom,              // 5
+		TopicOfSysInfo,           // 6
+	}
+	AllTopicsNumber = len(AllTopics)
+)

aarch64/pjibot_clean/master/package/service/collect_one_msg.go

@@ -0,0 +1,96 @@
+package service
+
+import (
+	"cicv-data-closedloop/aarch64/pjibot_clean/common/config"
+	"cicv-data-closedloop/common/config/c_log"
+	"cicv-data-closedloop/common/util"
+	"os"
+)
+
+func CollectOneMsg() {
+	collectMap()
+	collectTfStatic()
+	collectCostmap()
+}
+
+func collectMap() {
+
+	// rosbag record -O /root/cicv-data-closedloop/map_data.bag -l 1 /map
+	ossMapBagObjectKey := config.LocalConfig.OssBasePrefix + config.LocalConfig.EquipmentNo + "/map.bag"
+
+	var command []string
+	command = append(command, "record")
+	command = append(command, "-O")
+	command = append(command, config.CloudConfig.MapBagPath)
+	command = append(command, "-l")
+	command = append(command, "1")
+	command = append(command, "/map")
+	_, s, err := util.ExecuteWithEnvAndDir(config.RosbagEnvs, config.CloudConfig.BagDataDir, config.RosbagPath, command...)
+	if err != nil {
+		c_log.GlobalLogger.Error("程序异常退出。采集/map包", command, "出错：", s, "----", err)
+		os.Exit(-1)
+	}
+	c_log.GlobalLogger.Info("采集/map包", command, "完成。")
+	config.OssMutex.Lock()
+	err = config.OssBucket.PutObjectFromFile(ossMapBagObjectKey, config.CloudConfig.MapBagPath)
+	config.OssMutex.Unlock()
+	if err != nil {
+		c_log.GlobalLogger.Error("程序异常退出。上传/map包", config.CloudConfig.MapBagPath, "->", ossMapBagObjectKey, "出错：", err)
+		os.Exit(-1)
+	}
+	c_log.GlobalLogger.Info("上传/map包", config.CloudConfig.MapBagPath, "------", ossMapBagObjectKey, "成功。")
+}
+func collectTfStatic() {
+
+	// rosbag record -O /root/cicv-data-closedloop/map_data.bag -l 1 /map
+	ossMapBagObjectKey := config.LocalConfig.OssBasePrefix + config.LocalConfig.EquipmentNo + "/tfstatic.bag"
+
+	var command []string
+	command = append(command, "record")
+	command = append(command, "-O")
+	command = append(command, config.CloudConfig.MapBagPath)
+	command = append(command, "-l")
+	command = append(command, "1")
+	command = append(command, "/tf_static")
+	_, s, err := util.ExecuteWithEnvAndDir(config.RosbagEnvs, config.CloudConfig.BagDataDir, config.RosbagPath, command...)
+	if err != nil {
+		c_log.GlobalLogger.Error("程序异常退出。采集/tf_static包", command, "出错：", s, "----", err)
+		os.Exit(-1)
+	}
+	c_log.GlobalLogger.Info("采集/tf_static包", command, "完成。")
+	config.OssMutex.Lock()
+	err = config.OssBucket.PutObjectFromFile(ossMapBagObjectKey, config.CloudConfig.MapBagPath)
+	config.OssMutex.Unlock()
+	if err != nil {
+		c_log.GlobalLogger.Error("程序异常退出。上传/tf_static包", config.CloudConfig.MapBagPath, "->", ossMapBagObjectKey, "出错：", err)
+		os.Exit(-1)
+	}
+	c_log.GlobalLogger.Info("上传/tf_static包", config.CloudConfig.MapBagPath, "------", ossMapBagObjectKey, "成功。")
+}
+func collectCostmap() {
+
+	// rosbag record -O /root/cicv-data-closedloop/map_data.bag -l 1 /map
+	ossMapBagObjectKey := config.LocalConfig.OssBasePrefix + config.LocalConfig.EquipmentNo + "/costmap.bag"
+
+	var command []string
+	command = append(command, "record")
+	command = append(command, "-O")
+	command = append(command, config.CloudConfig.MapBagPath)
+	command = append(command, "-l")
+	command = append(command, "1")
+	command = append(command, "/move_base/global_costmap/costmap")
+	_, s, err := util.ExecuteWithEnvAndDir(config.RosbagEnvs, config.CloudConfig.BagDataDir, config.RosbagPath, command...)
+	if err != nil {
+		c_log.GlobalLogger.Error("程序异常退出。采集/move_base/global_costmap/costmap包", command, "出错：", s, "----", err)
+		os.Exit(-1)
+	}
+	c_log.GlobalLogger.Info("采集/move_base/global_costmap/costmap包", command, "完成。")
+	config.OssMutex.Lock()
+	err = config.OssBucket.PutObjectFromFile(ossMapBagObjectKey, config.CloudConfig.MapBagPath)
+	config.OssMutex.Unlock()
+	if err != nil {
+		c_log.GlobalLogger.Error("程序异常退出。上传/move_base/global_costmap/costmap包", config.CloudConfig.MapBagPath, "->", ossMapBagObjectKey, "出错：", err)
+		os.Exit(-1)
+	}
+	c_log.GlobalLogger.Info("上传/move_base/global_costmap/costmap包", config.CloudConfig.MapBagPath, "------", ossMapBagObjectKey, "成功。")
+}

aarch64/pjibot_clean/master/package/service/move_bag_and_send_window.go

@@ -0,0 +1,73 @@
+package service
+
+import (
+	commonConfig "cicv-data-closedloop/aarch64/pjibot_clean/common/config"
+	commonService "cicv-data-closedloop/aarch64/pjibot_clean/common/service"
+	"cicv-data-closedloop/common/config/c_log"
+	"cicv-data-closedloop/common/domain"
+	"cicv-data-closedloop/common/entity"
+	"cicv-data-closedloop/common/util"
+	"time"
+)
+
+// 将时间窗口内的包全部move出去，并等待当前时间窗口结束触发上传
+func RunTimeWindowProducerQueue() {
+	c_log.GlobalLogger.Info("生产者队列goroutine - 启动")
+	for {
+		// 收到自杀信号
+		select {
+		case signal := <-commonService.ChannelKillMove:
+			if signal == 1 {
+				commonService.ChannelKillMove <- 1
+				if len(entity.TimeWindowProducerQueue) == 0 {
+					commonService.AddKillTimes("4")
+					return
+				}
+			} else { //signal == 2
+				commonService.AddKillTimes("4")
+				return
+			}
+		default:
+		}
+
+		// 处理
+		time.Sleep(time.Duration(1) * time.Second)
+		if len(entity.TimeWindowProducerQueue) > 0 {
+			// 处理
+			bags, _ := util.ListAbsolutePathWithSuffixAndSort(commonConfig.CloudConfig.BagDataDir, ".bag")
+			currentTimeWindow := entity.TimeWindowProducerQueue[0]
+			move := false
+			bigger := false
+			for _, bag := range bags {
+				bagTime := util.GetBagTime(bag)
+				// 2 如果bag不小于timeWindowBegin不大于timeWindowEnd，则移动
+				compare1 := util.TimeCustom1GreaterEqualThanTimeCustom2(bagTime, currentTimeWindow.TimeWindowBegin)
+				compare2 := util.TimeCustom1LessEqualThanTimeCustom2(bagTime, currentTimeWindow.TimeWindowEnd)
+				if compare1 && compare2 {
+					// 将bag包移动到Copy目录
+					domain.MoveFromDataToCopy(currentTimeWindow.FaultTime, commonConfig.CloudConfig.BagDataDir, bag, commonConfig.CloudConfig.BagCopyDir)
+					move = true
+				} else {
+					if util.TimeCustom1GreaterEqualThanTimeCustom2(bagTime, currentTimeWindow.TimeWindowBegin) {
+						// 必须已经生成了窗口之后的包才算窗口结束了
+						bigger = true
+						break
+					}
+				}
+			}
+			// 如果没有包可以供当前窗口移动，且已经生成了更新的包，则当前窗口已经可以上传
+			if !move && bigger {
+				time.Sleep(time.Duration(2) * time.Second)
+				c_log.GlobalLogger.Info("采集数据，发送record命令进程关闭信号。")
+				commonService.ChannelKillRosRecord <- 3
+				entity.ProcessingFlag = true
+				domain.SupplyCopyBags(commonConfig.CloudConfig.BagDataDir, commonConfig.CloudConfig.BagCopyDir, currentTimeWindow)
+				// 将时间窗口移出准备队列
+				entity.RemoveHeadOfTimeWindowProducerQueue()
+				// 将时间窗口加入运行队列
+				entity.AddTimeWindowToTimeWindowConsumerQueue(currentTimeWindow)
+				continue
+			}
+		}
+	}
+}

aarch64/pjibot_clean/master/package/service/produce_window.go

@@ -0,0 +1,187 @@
+package service
+
+import (
+	commonConfig "cicv-data-closedloop/aarch64/pjibot_clean/common/config"
+	commonService "cicv-data-closedloop/aarch64/pjibot_clean/common/service"
+	masterConfig "cicv-data-closedloop/aarch64/pjibot_clean/master/package/config"
+	"cicv-data-closedloop/common/config/c_log"
+	"cicv-data-closedloop/common/entity"
+	"cicv-data-closedloop/common/util"
+	commonUtil "cicv-data-closedloop/common/util"
+	"encoding/json"
+	"github.com/bluenviron/goroslib/v2"
+	"github.com/bluenviron/goroslib/v2/pkg/msgs/std_msgs"
+	"sync"
+	"time"
+)
+
+var (
+	triggerInterval = 3.0 // 每个触发器3秒触发一次
+	logInterval     = 3.0
+	logTime         = time.Now()
+)
+
+// 负责监听所有主题并修改时间窗口
+func PrepareTimeWindowProducerQueue() {
+
+	var err error
+	subscribers := make([]*goroslib.Subscriber, masterConfig.AllTopicsNumber)
+	subscribersTimes := make([]time.Time, masterConfig.AllTopicsNumber)
+	subscribersTimeMutexes := make([]sync.Mutex, masterConfig.AllTopicsNumber)
+	subscribersMutexes := make([]sync.Mutex, masterConfig.AllTopicsNumber)
+	for i, topic := range masterConfig.AllTopics {
+		for {
+			create := false // 判断是否创建成功，用于打印日志
+			if topic == masterConfig.TopicOfObstacleDetection && len(masterConfig.RuleOfObstacleDetection) > 0 {
+				subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
+					Node:  commonConfig.RosNode,
+					Topic: topic,
+					Callback: func(data *std_msgs.UInt8) {
+						subscribersTimeMutexes[i].Lock()
+						if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
+							subscribersMutexes[i].Lock()
+							faultHappenTime := commonUtil.GetNowTimeCustom() // 获取当前故障发生时间
+							lastTimeWindow := entity.GetLastTimeWindow()     // 获取最后一个时间窗口
+							var faultLabel string
+							for _, f := range masterConfig.RuleOfObstacleDetection {
+								faultLabel = f(data)
+								if faultLabel != "" {
+									if !canCollect() {
+										break
+									}
+									subscribersTimes[i] = time.Now()
+									c_log.GlobalLogger.Errorf("触发事件【%v】，开始采集。", faultLabel)
+									saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
+									break
+								}
+							}
+							subscribersMutexes[i].Unlock()
+						}
+						subscribersTimeMutexes[i].Unlock()
+					},
+				})
+				if err == nil {
+					create = true
+				}
+			}
+			if err != nil {
+				c_log.GlobalLogger.Infof("创建订阅者报错，可能由于节点未启动，再次尝试【%v】", err)
+				time.Sleep(time.Duration(2) * time.Second)
+				continue
+			} else {
+				if create {
+					c_log.GlobalLogger.Infof("创建订阅者订阅话题【%v】", topic)
+				}
+				break
+			}
+		}
+	}
+	c_log.GlobalLogger.Infof("全部订阅者创建完成。")
+	select {
+	case signal := <-commonService.ChannelKillSubscriber:
+		if signal == 1 {
+			commonConfig.RosNode.Close()
+			commonService.AddKillTimes("3")
+			return
+		}
+	}
+}
+
+func saveTimeWindow(faultLabel string, faultHappenTime string, lastTimeWindow *entity.TimeWindow) {
+	masterTopics, slaveTopics := getTopicsOfNode(faultLabel)
+	if lastTimeWindow == nil || commonUtil.TimeCustom1GreaterTimeCustom2(faultHappenTime, lastTimeWindow.TimeWindowEnd) {
+		// 2-1 如果是不在旧故障窗口内，添加一个新窗口
+		newTimeWindow := entity.TimeWindow{
+			FaultTime:       faultHappenTime,
+			TimeWindowBegin: commonUtil.TimeCustomChange(faultHappenTime, -commonConfig.PlatformConfig.TaskBeforeTime),
+			TimeWindowEnd:   commonUtil.TimeCustomChange(faultHappenTime, commonConfig.PlatformConfig.TaskAfterTime),
+			Length:          commonConfig.PlatformConfig.TaskBeforeTime + commonConfig.PlatformConfig.TaskAfterTime + 1,
+			Labels:          []string{faultLabel},
+			MasterTopics:    masterTopics,
+			SlaveTopics:     slaveTopics,
+		}
+		c_log.GlobalLogger.Infof("不在旧故障窗口内，向生产者队列添加一个新窗口，【Lable】=%v，【FaultTime】=%v，【Length】=%v", newTimeWindow.Labels, newTimeWindow.FaultTime, newTimeWindow.Length)
+		entity.AddTimeWindowToTimeWindowProducerQueue(newTimeWindow)
+	} else {
+		// 2-2 如果在旧故障窗口内
+		entity.TimeWindowProducerQueueMutex.RLock()
+		defer entity.TimeWindowProducerQueueMutex.RUnlock()
+		// 2-2-1 更新故障窗口end时间
+		maxEnd := commonUtil.TimeCustomChange(lastTimeWindow.TimeWindowBegin, commonConfig.PlatformConfig.TaskMaxTime)
+		expectEnd := commonUtil.TimeCustomChange(faultHappenTime, commonConfig.PlatformConfig.TaskAfterTime)
+		if commonUtil.TimeCustom1GreaterTimeCustom2(expectEnd, maxEnd) {
+			lastTimeWindow.TimeWindowEnd = maxEnd
+			lastTimeWindow.Length = commonConfig.PlatformConfig.TaskMaxTime
+		} else {
+			if commonUtil.TimeCustom1GreaterTimeCustom2(expectEnd, lastTimeWindow.TimeWindowEnd) {
+				lastTimeWindow.TimeWindowEnd = expectEnd
+				lastTimeWindow.Length = commonUtil.CalculateDifferenceOfTimeCustom(lastTimeWindow.TimeWindowBegin, expectEnd)
+			}
+		}
+		// 2-2-2 更新label
+		labels := lastTimeWindow.Labels
+		lastTimeWindow.Labels = commonUtil.AppendIfNotExists(labels, faultLabel)
+		// 2-2-3 更新 topic
+		sourceMasterTopics := lastTimeWindow.MasterTopics
+		lastTimeWindow.MasterTopics = commonUtil.MergeSlice(sourceMasterTopics, masterTopics)
+		sourceSlaveTopics := lastTimeWindow.SlaveTopics
+		lastTimeWindow.SlaveTopics = commonUtil.MergeSlice(sourceSlaveTopics, slaveTopics)
+		c_log.GlobalLogger.Infof("在旧故障窗口内，更新生产者队列最新的窗口，【Lable】=%v，【FaultTime】=%v，【Length】=%v", lastTimeWindow.Labels, lastTimeWindow.FaultTime, lastTimeWindow.Length)
+	}
+}
+
+func getTopicsOfNode(faultLabel string) (masterTopics []string, slaveTopics []string) {
+	// 获取所有需要采集的topic
+	var faultCodeTopics []string
+	for _, code := range commonConfig.CloudConfig.Triggers {
+		if code.Label == faultLabel {
+			faultCodeTopics = code.Topics
+		}
+	}
+	return faultCodeTopics, nil
+}
+
+// 判断是否可采集数据
+func canCollect() bool {
+	// 如果开启了采集频率限制，则云端判断采集数量是否超过限额
+	if commonConfig.CloudConfig.CollectLimit.Flag == 1 {
+		c_log.GlobalLogger.Error("当前设备已开启数采频率限制，需判断采集数量是否达到限额。")
+		responseString, err := commonUtil.HttpPostJsonWithHeaders(
+			commonConfig.CloudConfig.CollectLimit.Url,
+			map[string]string{"Authorization": "U9yKpD6kZZDDe4LFKK6myAxBUT1XRrDM"},
+			map[string]string{
+				"snCode":            commonConfig.LocalConfig.SecretKey,
+				"collectLimitDay":   util.ToString(commonConfig.CloudConfig.CollectLimit.Day),
+				"collectLimitWeek":  util.ToString(commonConfig.CloudConfig.CollectLimit.Week),
+				"collectLimitMonth": util.ToString(commonConfig.CloudConfig.CollectLimit.Month),
+				"collectLimitYear":  util.ToString(commonConfig.CloudConfig.CollectLimit.Year),
+			},
+		)
+		if err != nil {
+			c_log.GlobalLogger.Error("发送http请求获取是否允许采集失败：", err)
+			return false
+		}
+		// 解析JSON字符串到Response结构体
+		var resp entity.Response
+		err = json.Unmarshal([]byte(responseString), &resp)
+		if err != nil {
+			c_log.GlobalLogger.Error("解析是否允许采集接口返回结果失败：", err)
+			return false
+		}
+		if resp.Code != 200 { // 不是200 代表不允许采集
+			c_log.GlobalLogger.Info("采集数量已超过限额，当前周期内不再采集。", resp.Code)
+			return false
+		}
+	} else {
+		c_log.GlobalLogger.Error("当前设备未开启数采频率限制，无需判断采集数量是否达到限额。")
+	}
+
+	// 本地判断是否存在正在处理的数据
+	if entity.ProcessingFlag {
+		c_log.GlobalLogger.Info("存在正在处理的数据，此次不再采集。")
+		return false
+	}
+
+	c_log.GlobalLogger.Info("允许采集。")
+	return true
+}

aarch64/pjibot_clean/start-control.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+chmod 777 /root/cicv-data-closedloop/pji-control.exe
+nohup /root/cicv-data-closedloop/pji-control.exe > /root/cicv-data-closedloop/log/pji-control.out 2>&1 &

aarch64/pjibot_clean/start-master.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+chmod 777 /root/cicv-data-closedloop/pji-master.exe
+nohup /root/cicv-data-closedloop/pji-master.exe > /root/cicv-data-closedloop/log/pji-master.out 2>&1 &

aarch64/pjibot_clean/清洁机器人默认配置文件-cloud-config.yaml

@@ -0,0 +1,210 @@
+---
+collect-limit:
+  url: http://36.110.106.142:12341/web_server/collect_limit/can_collect
+  flag: 1 # 数采频率限制标志 0 - 关闭数采频率限制  1 - 开启数采频率限制
+  day: 10
+  week: 70
+  month: 120
+  year: 144
+collect-num-plus:
+  url: http://36.110.106.142:12341/web_server/collect_limit/plus_collect_num
+monitor:
+  url: http://36.110.106.142:12341/web_server/monitor/insert
+platform:
+  url-device-auth: http://36.110.106.156:11121/device/auth
+  url-task-poll: http://36.110.106.156:11121/device/task/poll
+  url-task: http://36.110.106.156:11121/device/task
+bag-number: 60
+config-refresh-interval: 60
+disk: # 存储用的挂载点限制阈值，防止磁盘爆满
+  name: /dev/mmcblk0p8 # 磁盘名称
+  used: 20000000000 # 磁盘占用阈值，单位bytes
+  path: # 扫描的目录
+    - /root/cicv-data-closedloop
+    - /root/pjirobot/data/cicv-data-closedloop
+data-dir:
+  src: /root/pjirobot/ # 需要额外采集的 data 目录
+  src-sub:
+    - data/config
+    - data/map
+    - data/mapBuf
+  dest: /root/pjirobot/data.zip
+map-buf-files:
+  - /root/pjirobot/data/mapBuf/forbid_area.json
+  - /root/pjirobot/data/mapBuf/forbid_area.yaml
+  - /root/pjirobot/data/mapBuf/forbid_area_init.json
+  - /root/pjirobot/data/mapBuf/forbid_area_init.yaml
+  - /root/pjirobot/data/mapBuf/function_area.json
+  - /root/pjirobot/data/mapBuf/function_area_init.json
+  - /root/pjirobot/data/mapBuf/function_links.json
+  - /root/pjirobot/data/mapBuf/map.json
+  - /root/pjirobot/data/mapBuf/map.pbstream
+  - /root/pjirobot/data/mapBuf/map.pgm
+  - /root/pjirobot/data/mapBuf/map.yaml
+  - /root/pjirobot/data/mapBuf/map_type.json
+  - /root/pjirobot/data/mapBuf/param.yaml
+  - /root/pjirobot/data/mapBuf/stations.json
+  - /root/pjirobot/data/mapBuf/stations_init.json
+map-bag-path: /root/cicv-data-closedloop/map.bag
+bag-data-dir: /root/pjirobot/data/cicv-data-closedloop/data/
+bag-copy-dir: /root/pjirobot/data/cicv-data-closedloop/copy/
+triggers-dir: /root/pjirobot/data/cicv-data-closedloop/triggers/
+time-window-send-gap: 6
+rpc-port: 12341
+ros:
+  master-address: 192.168.1.104:11311
+  nodes:
+    - /amcl
+    - /ob_camera_01/camera
+    - /ob_camera_02/camera
+    - /node_diagnostics
+    - /localization_monitor_node
+    - /move_base
+    - /sensor_fusion_node
+    - /ltme_node
+    - /scan_map_icp_amcl_node
+    - /monitor
+
+hosts:
+  - name: node1
+    ip: 192.168.1.104
+    rosbag:
+      path: "/opt/ros/melodic/bin/rosbag"
+      envs:
+        - "C_INCLUDE_PATH=/usr/include/drm:"
+        - "USER=root"
+        - "ROS_PACKAGE_PATH=/opt/ros/melodic/share"
+        - "LD_LIBRARY_PATH=/opt/ros/melodic/lib:/opt/ros/melodic/lib/aarch64-linux-gnu"
+        - "ROS_ETC_DIR=/opt/ros/melodic/etc/ros"
+        - "SHLVL=1"
+        - "HOME=/root"
+        - "ROS_PYTHON_VERSION=2"
+        - "PCMANFM_OUTLINE_MODE=on"
+        - "CPLUS_INCLUDE_PATH=/usr/include/drm:"
+        - "ROS_DISTRO=melodic"
+        - "ROS_VERSION=1"
+        - "PKG_CONFIG_PATH=/opt/ros/melodic/lib/pkgconfig"
+        - "PATH=/opt/ros/melodic/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/games:/usr/local/games:/snap/bin:/usr/local/go/bin:/root/go/bin"
+        - "ROS_ROOT=/opt/ros/melodic/share/ros"
+        - "ROSLISP_PACKAGE_DIRECTORIES="
+        - "ROS_MASTER_URI=http://192.168.1.104:11311"
+        - "PYTHONPATH=/opt/ros/melodic/lib/python2.7/dist-packages"
+        - "ROS_HOSTNAME=192.168.1.104"
+        - "CMAKE_PREFIX_PATH=/opt/ros/melodic"
+    topics: # /amcl_pose,/ob_camera_01/color/image_raw,/ob_camera_02/color/image_raw,/diagnostics,/locate_info,/obstacle_detection,/odom,/move_base/global_costmap/costmap,/move_base/global_costmap/costmap_updates,/move_base/local_costmap/costmap,/move_base/local_costmap/costmap_updates,/scan,/scan_map_icp_amcl_node/scan_point_transformed,/sys_info,/imu,/depth_scan_02,/map,/scan_filtered,/sonar_left,/sonar_right,/sonar_mid,/sonar_rmid,/tf,/tf_static,/cmd_vel,/move_base/DWAPlannerROS/global_plan,/move_base/DWAPlannerROS/local_plan,/move_base/GlobalPlanner/plan,/move_base/global_costmap/footprint,/move_base/local_costmap/footprint,/robot_pose_tf
+      - /amcl_pose # /amcl
+      - /ob_camera_01/color/image_raw # /ob_camera_01/camera
+      #      - /ob_camera_01/depth/points # /ob_camera_01/camera
+      - /ob_camera_02/color/image_raw # /ob_camera_02/camera
+      #      - /ob_camera_02/depth/points # /ob_camera_02/camera
+      - /diagnostics # /amcl /node_diagnostics
+      - /locate_info # /localization_monitor_node
+      - /obstacle_detection # /move_base
+      - /odom # /sensor_fusion_node
+      - /move_base/global_costmap/costmap # /move_base
+      - /move_base/global_costmap/costmap_updates # /move_base
+      - /move_base/local_costmap/costmap # /move_base
+      - /move_base/local_costmap/costmap_updates # /move_base
+      - /scan # /ltme_node
+      - /scan_map_icp_amcl_node/scan_point_transformed # /scan_map_icp_amcl_node
+      - /sys_info
+      #      - /cmd_vel
+      - /imu
+      # 算法评价新增
+      - /depth_scan_02
+      - /map
+      - /scan_filtered
+      - /sonar_left
+      - /sonar_right
+      - /sonar_mid
+      - /sonar_rmid
+      - /tf
+      - /tf_static
+      - /cmd_vel
+      - /move_base/DWAPlannerROS/global_plan
+      - /move_base/DWAPlannerROS/local_plan
+      - /move_base/GlobalPlanner/plan
+      - /move_base/global_costmap/footprint
+      - /move_base/local_costmap/footprint
+      - /robot_pose_tf
+
+
+full-collect: true # 控制是否根据不同的触发器采集不通的topic，一般设置为true，即忽略下面的配置
+triggers:
+  - label: detectfault
+    topics:
+      - /camera/color/image_raw
+      - /camera/depth/points
+      - /diagnostics
+      - /locate_info
+      - /obstacle_detection
+      - /odom
+      - /move_base/global_costmap/costmap
+      - /move_base/global_costmap/costmap_updates
+      - /scan_map_icp_amcl_node/scan_point_transformed
+  - label: unstabledriving
+    topics:
+      - /camera/color/image_raw
+      - /camera/depth/points
+      - /diagnostics
+      - /locate_info
+      - /obstacle_detection
+      - /odom
+      - /move_base/global_costmap/costmap
+      - /move_base/global_costmap/costmap_updates
+      - /scan_map_icp_amcl_node/scan_point_transformed
+  - label: locationfailed
+    topics:
+      - /camera/color/image_raw
+      - /camera/depth/points
+      - /diagnostics
+      - /locate_info
+      - /obstacle_detection
+      - /odom
+      - /move_base/global_costmap/costmap
+      - /move_base/global_costmap/costmap_updates
+      - /scan_map_icp_amcl_node/scan_point_transformed
+  - label: obstacledetection
+    topics:
+      - /camera/color/image_raw
+      - /camera/depth/points
+      - /diagnostics
+      - /locate_info
+      - /obstacle_detection
+      - /odom
+      - /move_base/global_costmap/costmap
+      - /move_base/global_costmap/costmap_updates
+      - /scan_map_icp_amcl_node/scan_point_transformed
+  - label: overspeed
+    topics:
+      - /camera/color/image_raw
+      - /camera/depth/points
+      - /diagnostics
+      - /locate_info
+      - /obstacle_detection
+      - /odom
+      - /move_base/global_costmap/costmap
+      - /move_base/global_costmap/costmap_updates
+      - /scan_map_icp_amcl_node/scan_point_transformed
+  - label: cpuoveroccupied
+    topics:
+      - /camera/color/image_raw
+      - /camera/depth/points
+      - /diagnostics
+      - /locate_info
+      - /obstacle_detection
+      - /odom
+      - /move_base/global_costmap/costmap
+      - /move_base/global_costmap/costmap_updates
+      - /scan_map_icp_amcl_node/scan_point_transformed
+  - label: memoveroccupied
+    topics:
+      - /camera/color/image_raw
+      - /camera/depth/points
+      - /diagnostics
+      - /locate_info
+      - /obstacle_detection
+      - /odom
+      - /move_base/global_costmap/costmap
+      - /move_base/global_costmap/costmap_updates
+      - /scan_map_icp_amcl_node/scan_point_transformed

aarch64/pjibot_clean/清洁机器人默认配置文件-local-config.yaml

@@ -0,0 +1,20 @@
+type: 1 # 机器人类型 1 引导机器人 2 配送机器人 3 巡检机器人 4 清洁机器人
+node:
+  name: node1
+  ip: 192.168.1.104
+rosparam-path: /opt/ros/melodic/bin/rosparam
+# 获取oss连接信息的接口url
+url-get-oss-config: http://36.110.106.156:18379/oss/pji?token=nXonLUcMtGcrQqqKiyygIwyVbvizE0wD
+# 朴津机器人数据前缀
+oss-base-prefix: pji-double-camera/
+# oss上的配置文件的名称
+cloud-config-filename: cloud-config.yaml
+# 将oss上的配置文件下载到本地的路径
+cloud-config-local-path: /root/cicv-data-closedloop/config/cloud-config.yaml
+# websocket端口号
+local-websocket-port: 9002
+restart-cmd:
+  dir: "/root/cicv-data-closedloop/"
+  name: "sh"
+  args:
+    - "start-master.sh"

pjibot_clean_msgs/common_msgs.go

@@ -0,0 +1,22 @@
+package pjibot_clean_msgs
+
+import (
+	"github.com/bluenviron/goroslib/v2/pkg/msg"
+	"github.com/bluenviron/goroslib/v2/pkg/msgs/geometry_msgs"
+)
+
+type SysInfo struct {
+	msg.Package    `ros:"common_msgs"`
+	CpuOccupied    float32 `rosname:"cpu_occupied"`
+	MemOccupied    float32 `rosname:"mem_occupied"`
+	CurMileage     float64 `rosname:"cur_mileage"`
+	HistoryMileage float64 `rosname:"history_mileage"`
+}
+
+type LocateInfo struct {
+	msg.Package  `ros:"common_msgs"`
+	Pose         geometry_msgs.PoseStamped `rosname:"pose"`
+	LocateStatus int8                      `rosname:"locate_status"`
+	ErrorCode    int64                     `rosname:"error_code"`
+	Message      string                    `rosname:"message"`
+}