commit

aarch64/kinglong/common/config/c_cloud.go

@@ -1,174 +0,0 @@
-package config
-
-import (
-	"cicv-data-closedloop/common/config/c_log"
-	"cicv-data-closedloop/common/util"
-	"gopkg.in/yaml.v3"
-	"os"
-	"sync"
-	"time"
-)
-
-type platform struct {
-	UrlDeviceAuth string `yaml:"url-device-auth"`
-	UrlTaskPoll   string `yaml:"url-task-poll"`
-	UrlTask       string `yaml:"url-task"`
-}
-
-type rosbagStruct struct {
-	Path string   `yaml:"path"`
-	Envs []string `yaml:"envs"`
-}
-
-type hostStruct struct {
-	Name   string       `yaml:"name"`
-	Ip     string       `yaml:"ip"`
-	Rosbag rosbagStruct `yaml:"rosbag"`
-	Topics []string     `yaml:"topics"`
-}
-
-type ros struct {
-	MasterAddress string   `yaml:"master-address"`
-	Nodes         []string `yaml:"nodes"`
-}
-
-type disk struct {
-	Name string `yaml:"name"`
-	Used uint64 `yaml:"used"`
-}
-type trigger struct {
-	Label  string   `yaml:"label"`
-	Topics []string `yaml:"topics"`
-}
-type cloudConfig struct {
-	FullCollect           bool         `yaml:"full-collect"`
-	ConfigRefreshInterval int          `yaml:"config-refresh-interval"` // 配置刷新时间间隔
-	BagNumber             int          `yaml:"bag-number"`
-	TimeWindowSendGap     int          `yaml:"time-window-send-gap"` // 主节点向从节点发送窗口的最小时间间隔
-	TimeToLabelJsonPath   string       `yaml:"time-to-label-json-path"`
-	BagDataDir            string       `yaml:"bag-data-dir"`
-	BagCopyDir            string       `yaml:"bag-copy-dir"`
-	TriggersDir           string       `yaml:"triggers-dir"`
-	TcpPort               string       `yaml:"tcp-port"`
-	RpcPort               string       `yaml:"rpc-port"`
-	Triggers              []trigger    `yaml:"triggers"`
-	Hosts                 []hostStruct `yaml:"hosts"`
-	Ros                   ros          `yaml:"ros"`
-	Platform              platform     `yaml:"platform"`
-	Disk                  disk         `yaml:"disk"`
-}
-
-var (
-	CloudConfig      cloudConfig
-	CloudConfigMutex sync.RWMutex
-)
-
-// InitCloudConfig 初始化业务配置
-func InitCloudConfig() {
-	c_log.GlobalLogger.Info("初始化OSS配置文件 - 开始。")
-	// 获取文件的目录
-	_ = util.CreateParentDir(LocalConfig.CloudConfigLocalPath)
-	// 3 ------- 获取 yaml 字符串 -------
-	var content []byte
-	cloudConfigObjectKey := LocalConfig.OssBasePrefix + LocalConfig.EquipmentNo + "/" + LocalConfig.CloudConfigFilename
-	// 等待时间同步
-	for {
-		OssMutex.Lock()
-		err := OssBucket.GetObjectToFile(cloudConfigObjectKey, LocalConfig.CloudConfigLocalPath)
-		OssMutex.Unlock()
-		if err != nil {
-			c_log.GlobalLogger.Error("下载oss上的配置文件 "+cloudConfigObjectKey+" 失败。可能需要等待时间同步完成：", 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 checkConfig(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)
-	timeToLabelJson, _ := util.MapToJsonString(map[string]interface{}{"time": "label"})
-	_ = util.WriteFile(timeToLabelJson, CloudConfig.TimeToLabelJsonPath)
-}
-
-// RefreshCloudConfig 初始化业务配置
-func refreshCloudConfig() {
-	// 获取文件的目录
-	_ = util.CreateParentDir(LocalConfig.CloudConfigLocalPath)
-	// 3 ------- 获取 yaml 字符串 -------
-	var content []byte
-	cloudConfigObjectKey := LocalConfig.OssBasePrefix + LocalConfig.EquipmentNo + "/" + LocalConfig.CloudConfigFilename
-	OssMutex.Lock()
-	err := OssBucket.GetObjectToFile(cloudConfigObjectKey, LocalConfig.CloudConfigLocalPath)
-	OssMutex.Unlock()
-	if err != nil {
-		c_log.GlobalLogger.Error("下载oss上的配置文件"+cloudConfigObjectKey+"失败。", err)
-		return
-	}
-
-	content, err = os.ReadFile(LocalConfig.CloudConfigLocalPath)
-	if err != nil {
-		c_log.GlobalLogger.Error("配置文件 ", LocalConfig.CloudConfigLocalPath, " 读取失败:", err)
-		return
-	}
-
-	// 4 ------- 解析YAML内容 -------
-	var newCloudConfig cloudConfig
-	err = yaml.Unmarshal(content, &newCloudConfig)
-	if err != nil {
-		c_log.GlobalLogger.Error("配置文件 ", LocalConfig.CloudConfigLocalPath, " 解析失败:", err)
-		return
-	}
-
-	// 5 ------- 校验 yaml -------
-	if checkConfig(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 checkConfig(check cloudConfig) bool {
-	if len(check.Hosts) != 2 {
-		c_log.GlobalLogger.Error("cloud-config.yaml中配置的hosts必须为2。")
-		os.Exit(-1)
-	}
-	return true
-}

aarch64/kinglong/common/config/c_killrpcserver.go

@@ -1,21 +0,0 @@
-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/kinglong/common/config/c_local.go

@@ -1,55 +0,0 @@
-package config
-
-import (
-	"fmt"
-	"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"`                    // 节点信息
-	UrlGetOssConfig      string     `yaml:"url-get-oss-config"`      // 获取oss配置的url
-	OssBasePrefix        string     `yaml:"oss-base-prefix"`         // 云端配置文件的位置
-	EquipmentNo          string     `yaml:"equipment-no"`            // 当前设备的编号
-	CloudConfigFilename  string     `yaml:"cloud-config-filename"`   // 云端配置文件名称
-	CloudConfigLocalPath string     `yaml:"cloud-config-local-path"` // 将 oss 的配置文件下载到本地的位置
-	SecretKey            string     `yaml:"secret-key"`              // 当前设备的密钥
-	LogDir               string     `yaml:"log-dir"`                 // 日志目录
-	RestartCmd           restartCmd `yaml:"restart-cmd"`             // 重启命令
-}
-
-var (
-	LocalConfig     localConfig
-	localConfigPath = "/mnt/media/sda1/cicv-data-closedloop/config/local-config.yaml"
-)
-
-func InitLocalConfig() {
-	fmt.Println("初始化本地配置文件 - 开始：", localConfigPath)
-	// 读取YAML文件内容
-	content, err := os.ReadFile(localConfigPath)
-	if err != nil {
-		fmt.Println("程序退出。读取本地配置文件失败。", err)
-		os.Exit(-1)
-	}
-
-	// 解析YAML内容
-	err = yaml.Unmarshal(content, &LocalConfig)
-	if err != nil {
-		fmt.Println("程序退出。解析本地配置文件失败。", err)
-		os.Exit(-1)
-	}
-
-	fmt.Println("初始化本地配置文件 - 成功：", LocalConfig)
-
-}

aarch64/kinglong/common/config/c_oss.go

@@ -1,59 +0,0 @@
-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"
-	"time"
-)
-
-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() {
-	var err error
-	c_log.GlobalLogger.Info("初始化OSS客户端对象 - 开始。")
-	// 1 访问 HTTP 服务获取 OSS 配置
-	var get string
-	// 等待网络可达
-	for {
-		get, err = util.HttpGet(LocalConfig.UrlGetOssConfig)
-		if err != nil {
-			c_log.GlobalLogger.Error("http 获取 oss 配置时出错。可能需要等待网络初始化完成:", err)
-			time.Sleep(time.Duration(2) * time.Second)
-			continue
-		}
-		break
-	}
-	var ossConnectInfo OssConnectInfoStruct
-	err = json.Unmarshal([]byte(get), &ossConnectInfo)
-	if err != nil {
-		c_log.GlobalLogger.Error("解析json时出错:", err)
-		os.Exit(-1)
-	}
-
-	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/kinglong/common/config/c_platform.go

@@ -1,191 +0,0 @@
-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
-	SubscribeTopics []string
-)
-
-// InitPlatformConfig 初始化数据闭环平台的配置
-func InitPlatformConfig() {
-	var err error
-	c_log.GlobalLogger.Info("获取数据闭环平台配置 - 开始")
-	// 1 如果车辆没有配置任务，则阻塞在这里，不启动任务
-	for {
-		time.Sleep(time.Duration(2) * time.Second)
-		PlatformConfig, err = getConfig()
-		if err != nil {
-			c_log.GlobalLogger.Error("获取配置status失败：", err)
-			continue
-		}
-		if checkPlatformConfig() {
-			SubscribeTopics = strings.Split(PlatformConfig.EquipmentTopic, ",")
-			// 去掉首尾空格
-			for i, topic := range SubscribeTopics {
-				SubscribeTopics[i] = strings.TrimSpace(topic)
-			}
-			break
-		}
-	}
-	c_log.GlobalLogger.Info("获取数据闭环平台配置 - 成功。")
-}
-
-/*
-	{
-	  "data": {
-	    "accessToken": "YWRmYWRzZmFzZGZhZHNmYWRmYWRm=",
-	    "expireTime": "28800",
-	    "equipmentNo": "robot-001"
-	  },
-	  "success": true,
-	  "message": "ok",
-	  "code": 1,
-	  "nowTime": "2023-12-09 22:41:00"
-	}
-*/
-// GetAccessToken 认证接口，获取access_token
-func GetAccessToken() (string, error) {
-	respJson, err := util.HttpPostJsonResponseString(
-		CloudConfig.Platform.UrlDeviceAuth,
-		map[string]string{
-			"equipmentNo": LocalConfig.EquipmentNo,
-			"secretKey":   LocalConfig.SecretKey,
-		},
-	)
-	if err != nil {
-		return "", nil
-	}
-	respMap, err := util.JsonStringToMap(respJson)
-	if err != nil {
-		c_log.GlobalLogger.Error("解析返回结果", respJson, "失败：", 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/kinglong/common/config/c_ros.go

@@ -1,36 +0,0 @@
-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
-	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
-	}
-	// 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/kinglong/common/config/sh/start-soc1.sh

@@ -1,4 +0,0 @@
-#!/bin/bash
-
-chmod 777 /mnt/media/sda1/cicv-data-closedloop/kinglong-master.exe
-nohup /mnt/media/sda1/cicv-data-closedloop/kinglong-master.exe > /mnt/media/sda1/cicv-data-closedloop/log/kinglong-master.out 2>&1 &

aarch64/kinglong/common/config/sh/start-soc2.sh

@@ -1,4 +0,0 @@
-#!/bin/bash
-
-chmod 777 /mnt/media/sda1/cicv-data-closedloop/kinglong-slave.exe
-nohup /mnt/media/sda1/cicv-data-closedloop/kinglong-slave.exe > /mnt/media/sda1/cicv-data-closedloop/log/kinglong-slave.out 2>&1 &

aarch64/kinglong/common/config/yaml/kinglong3-soc1-local-config.yaml

@@ -1,20 +0,0 @@
-node:
-  name: node1
-  ip: 192.168.1.102
-# 数据闭环平台参数
-equipment-no: kinglong3
-secret-key: Nr0Ufk3r3Yo4tDM8g9gipmoexuB49ra1MJolKQ24
-# 获取oss连接信息的接口url
-url-get-oss-config: http://36.110.106.156:18379/oss/config?token=nXonLUcMtGcrQqqKiyygIwyVbvizE0wD
-# 金龙车数据前缀
-oss-base-prefix: kinglong/
-# oss上的配置文件的名称
-cloud-config-filename: cloud-config.yaml
-# 将oss上的配置文件下载到本地的路径
-cloud-config-local-path: /mnt/media/sda1/cicv-data-closedloop/config/cloud-config.yaml
-log-dir: /mnt/media/sda1/cicv-data-closedloop/log/
-restart-cmd:
-  dir: "/mnt/media/sda1/cicv-data-closedloop/"
-  name: "sh"
-  args:
-    - "start-soc1.sh"

aarch64/kinglong/common/config/yaml/kinglong3-soc2-local-config.yaml

@@ -1,20 +0,0 @@
-node:
-  name: node2
-  ip: 192.168.1.103
-# 数据闭环平台参数
-equipment-no: kinglong3
-secret-key: Nr0Ufk3r3Yo4tDM8g9gipmoexuB49ra1MJolKQ24
-# 获取oss连接信息的接口url
-url-get-oss-config: http://36.110.106.156:18379/oss/config?token=nXonLUcMtGcrQqqKiyygIwyVbvizE0wD
-# 金龙车数据前缀
-oss-base-prefix: kinglong/
-# oss上的配置文件的名称
-cloud-config-filename: cloud-config.yaml
-# 将oss上的配置文件下载到本地的路径
-cloud-config-local-path: /mnt/media/sda1/cicv-data-closedloop/config/cloud-config.yaml
-log-dir: /mnt/media/sda1/cicv-data-closedloop/log/
-restart-cmd:
-  dir: "/mnt/media/sda1/cicv-data-closedloop/"
-  name: "sh"
-  args:
-    - "start-soc2.sh"

aarch64/kinglong/common/config/yaml/kinglong3-v2253-cloud-config.yaml

@@ -1,401 +0,0 @@
-# 全量采集topic
----
-platform:
-  url-device-auth: http://1.202.169.139:8081/device/auth
-  url-task-poll: http://1.202.169.139:8081/device/task/poll
-  url-task: http://1.202.169.139:8081/device/task
-full-collect: true
-bag-number: 120
-config-refresh-interval: 60
-disk:
-  name: /dev/vdb # 磁盘名称
-  used: 5242880 # 磁盘占用阈值，单位bytes
-bag-data-dir: /mnt/media/sda1/rosbag-handle/data/
-bag-copy-dir: /mnt/media/sda1/rosbag-handle/copy/
-triggers-dir: /mnt/media/sda1/rosbag-handle/triggers/
-time-window-send-gap: 6
-tcp-port: 12340
-rpc-port: 12341
-ros:
-  master-address: 192.168.1.102:11311
-  nodes:
-    - /GlobalWay
-    - /ObstacleInfo
-    - /ObstacleInfo_zd
-    - /Trajectorydisplay
-    - /camera_output
-    - /ch128x1/lslidar_driver_node
-    - /ch64_left/lslidar_driver_node
-    - /ch64_mid/lslidar_driver_node
-    - /ch64_right/lslidar_driver_node
-    - /dpi_map_engine
-    - /front_radars_receive_node
-    - /ins
-    - /ins_diagnosis
-    - /lidar_deeplearning
-    - /lidar_fast_euclidean_cluster_detect
-    - /lidar_fusion
-    - /lidar_pretreatment
-    - /planner_s4_gd
-    - /planning_rviz
-    - /planning_task_manage
-    - /polygon_display
-    - /r_des
-    - /r_light
-    - /r_loc
-    - /r_location
-    - /r_speed
-    - /rnode
-    - /rosout
-    - /sensors_fusion_node
-    - /stationinfo
-    - /utmTOll_node
-hosts:
-  - name: node1
-    ip: 192.168.1.102
-    rosbag:
-      envs:
-        - ""
-      path: "/opt/ros/melodic/bin/rosbag"
-    topics:
-      - /camera_image
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-
-  - name: node2
-    ip: 192.168.1.103
-    rosbag:
-      envs:
-        - ""
-      path: "/opt/ros/melodic/bin/rosbag"
-    rosbag-path:
-    topics:
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi
-triggers:
-  - label: rapidaccel
-    topics:
-      - /camera_image
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi
-  - label: brake
-    topics:
-      - /camera_image
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi
-  - label: EmergencyStop
-    topics:
-      - /camera_image
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi
-  - label: AutoDLimit
-    topics:
-      - /camera_image
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi
-  - label: lanechange
-    topics:
-      - /camera_image
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi
-  - label: brakefault
-    topics:
-      - /camera_image
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi
-  - label: takeover
-    topics:
-      - /camera_image
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi
-  - label: TTC
-    topics:
-      - /camera_image
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi

aarch64/kinglong/common/config/yaml/kinglong3-v2304-cloud-config.yaml

@@ -1,422 +0,0 @@
-# 全量采集topic
----
-platform:
-  url-device-auth: http://1.202.169.139:8081/device/auth
-  url-task-poll: http://1.202.169.139:8081/device/task/poll
-  url-task: http://1.202.169.139:8081/device/task
-bag-number: 120
-config-refresh-interval: 60
-disk:
-  name: /dev/vdb # 磁盘名称
-  used: 800000000000 # 磁盘占用阈值，单位bytes
-bag-data-dir: /mnt/media/sda1/cicv-data-closedloop/data/
-bag-copy-dir: /mnt/media/sda1/cicv-data-closedloop/copy/
-triggers-dir: /mnt/media/sda1/cicv-data-closedloop/triggers/
-time-window-send-gap: 6
-tcp-port: 12340
-rpc-port: 12341
-ros:
-  master-address: 192.168.1.102:11311
-  nodes:
-    - /control
-#    - /GlobalWay
-#    - /ObstacleInfo
-#    - /ObstacleInfo_zd
-#    - /Trajectorydisplay
-#    - /ch128x1/lslidar_driver_node
-#    - /ch64_left/lslidar_driver_node
-#    - /ch64_mid/lslidar_driver_node
-#    - /ch64_right/lslidar_driver_node
-#    - /dpi_map_engine
-#    - /front_radars_receive_node
-#    - /ins
-#    - /ins_diagnosis
-#    - /lidar_deeplearning
-#    - /lidar_fast_euclidean_cluster_detect
-#    - /lidar_fusion
-#    - /lidar_pretreatment
-#    - /planner_s4_gd
-#    - /planning_rviz
-#    - /planning_task_manage
-#    - /polygon_display
-#    - /r_des
-#    - /r_light
-#    - /r_loc
-#    - /r_location
-#    - /r_speed
-#    - /rnode
-#    - /rosout
-#    - /sensors_fusion_node
-#    - /stationinfo
-#    - /utmTOll_node
-hosts:
-  - name: node1
-    ip: 192.168.1.102
-    rosbag:
-      path: "/opt/ros/melodic/bin/rosbag"
-      envs:
-        - "LD_LIBRARY_PATH=/opt/ros/melodic/lib:/usr/lib::/usr/lib:/userdata/third_lib/libaicc"
-        - "ROS_ETC_DIR=/opt/ros/melodic/etc/ros"
-        - "USER=root"
-        - "ROS_OS_OVERRIDE=openembedded"
-        - "PWD=/mnt/media/sda1/cicv-data-closedloop"
-        - "HOME=/home/root"
-        - "CMAKE_PREFIX_PATH=/opt/ros/melodic:/usr"
-        - "ROS_ROOT=/opt/ros/melodic/share/ros"
-        - "ROS_MASTER_URI=http://192.168.1.102:11311"
-        - "ROS_VERSION=1"
-        - "ROS_PYTHON_VERSION=2"
-        - "ROS_IP=192.168.1.102"
-        - "PYTHONPATH=/opt/ros/melodic/lib/python2.7/site-packages:/usr/lib/python2.7/site-packages"
-        - "ROS_PACKAGE_PATH=/opt/ros/melodic/share"
-        - "PATH=/opt/ros/melodic/bin:/usr/bin/cyber/tools/cyber_service:/usr/bin/cyber/tools/cyber_node:/usr/bin/cyber/tools/cyber_channel:/usr/bin/cyber/tools/cyber_launch:/usr/bin/cyber/tools/cyber_monitor:/usr/bin/cyber/tools/cyber_recorder:/apollo/bazel-bin/cyber:/usr/local/bin:/usr/bin:/bin:/usr/local/sbin:/usr/sbin:/sbin"
-        - "PKG_CONFIG_PATH=/usr/lib/pkgconfig"
-        - "ROS_DISTRO=melodic"
-    topics:
-      - /AutoModeStatus # /control
-      - /cam_objects
-      - /cam_res
-      - /car_wheel # /control
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidartrackingmovingobjects
-      - /cicv_location
-      - /cicv_location_JL
-      - /cicv_amr_trajectory
-      - /data_read
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /fusion/vis/velocity
-      - /heartbeat_info
-      - /jinlong_control_pub
-      - /jinlong_flag_pub
-      - /map_polygon
-      - /nodefault_info
-      - /reference_trajectory
-      - /tftrafficlight
-      - /tprouteplan
-      - /vehicle_info
-  - name: node2
-    ip: 192.168.1.103
-    rosbag:
-      path: "/opt/ros/melodic/bin/rosbag"
-      envs:
-        - "LD_LIBRARY_PATH=/opt/ros/melodic/lib:/usr/lib::/usr/lib:/userdata/third_lib"
-        - "ROS_ETC_DIR=/opt/ros/melodic/etc/ros"
-        - "USER=root"
-        - "ROS_OS_OVERRIDE=openembedded"
-        - "PWD=/mnt/media/sda1/cicv-data-closedloop"
-        - "HOME=/home/root"
-        - "CMAKE_PREFIX_PATH=/opt/ros/melodic:/usr"
-        - "ROS_ROOT=/opt/ros/melodic/share/ros"
-        - "ROS_MASTER_URI=http://192.168.1.102:11311"
-        - "ROS_VERSION=1"
-        - "ROS_PYTHON_VERSION=2"
-        - "ROS_IP=192.168.1.103"
-        - "PYTHONPATH=/opt/ros/melodic/lib/python2.7/site-packages:/usr/lib/python2.7/site-packages"
-        - "ROS_PACKAGE_PATH=/opt/ros/melodic/share"
-        - "PATH=/opt/ros/melodic/bin:/usr/bin/cyber/tools/cyber_service:/usr/bin/cyber/tools/cyber_node:/usr/bin/cyber/tools/cyber_channel:/usr/bin/cyber/tools/cyber_launch:/usr/bin/cyber/tools/cyber_monitor:/usr/bin/cyber/tools/cyber_recorder:/apollo/bazel-bin/cyber:/usr/local/bin:/usr/bin:/bin:/usr/local/sbin:/usr/sbin:/sbin "
-        - "PKG_CONFIG_PATH=/usr/lib/pkgconfig"
-        - "ROS_DISTRO=melodic"
-    topics:
-      - /ch128x1/lslidar_main
-      - /ch64_left/lslidar_left
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /points_concat
-      - /roi/points
-      - /roi/polygon
-      - /tpperception
-      - /tpperception/hmi
-      - /unground_cloudpoints
-full-collect: true
-triggers:
-  - label: rapidaccel
-    topics:
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi
-  - label: brake
-    topics:
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi
-  - label: EmergencyStop
-    topics:
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi
-  - label: AutoDLimit
-    topics:
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi
-  - label: lanechange
-    topics:
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi
-  - label: brakefault
-    topics:
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi
-  - label: takeover
-    topics:
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi
-  - label: TTC
-    topics:
-      - /cam_res
-      - /fusion/vis/velocity
-      - /cicv/lidartracking_moving_objects
-      - /cicv_location
-      - /cicv/lidarfusionmovingobject
-      - /cicv/lidardeeplearning_moving_objects
-      - /cicv/lidarcluster_moving_objects
-      - /cam_objects
-      - /f_radar_objects
-      - /fusion/vis/box
-      - /cicv_amr_trajectory
-      - /map_polygon
-      - /reference_trajectory
-      - /tprouteplan
-      - /jinlong_control_pub
-      - /vehicle_info
-      - /tftrafficlight
-      - /cicv_location_JL
-      - /jinlong_flag_pub
-      - /car_wheel
-      - /nodefault_info
-      - /AutoModeStatus
-      - /heartbeat_info
-      - /data_read
-      - /ch128x1/lslidar_main
-      - /ch64_mid/lslidar_mid
-      - /ch64_right/lslidar_right
-      - /ch64_left/lslidar_left
-      - /roi/points
-      - /roi/polygon
-      - /unground_cloudpoints
-      - /points_concat
-      - /tpperception
-      - /tpperception/hmi

aarch64/kinglong/common/service/disk_clean.go

@@ -1,92 +0,0 @@
-package service
-
-import (
-	commonConfig "cicv-data-closedloop/aarch64/kinglong/common/config"
-	masterConfig "cicv-data-closedloop/aarch64/kinglong/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"
-)
-
-// DiskClean 如果磁盘占用过高，则删除timeWindow和对应的文件
-func DiskClean() {
-	c_log.GlobalLogger.Info("启动timeWindow清理goroutine，根据缓存策略清理copy目录。")
-	for {
-		time.Sleep(1000 * time.Millisecond)
-		bags, _ := util.ListAbsolutePathWithSuffixAndSort(commonConfig.CloudConfig.BagCopyDir, ".bag")
-		if len(bags) == 0 {
-			continue
-		}
-		/*
-			TTL(0, "删除旧数据")；
-			STOP(1, "停止缓存");
-			LRU(2, "保留高优先级")
-		*/
-		policyToDescription := map[string]string{
-			"TTL":  "删除旧数据",
-			"STOP": "停止缓存",
-			"LRU":  "保留高优先级",
-		}
-		// 1 获取磁盘占用
-		diskUsed, _ := util.GetDiskUsed(commonConfig.CloudConfig.Disk.Name)
-		if diskUsed > commonConfig.CloudConfig.Disk.Used {
-			policy := commonConfig.PlatformConfig.TaskCachePolicy
-			c_log.GlobalLogger.Errorf("磁盘占用 %v 超过 %v，触发删除规则 %v", diskUsed, commonConfig.CloudConfig.Disk.Used, policyToDescription[policy])
-			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 {
-				if masterConfig.LabelMapTriggerId[label] == lru[i] {
-					return i2
-				}
-			}
-		}
-	}
-	return -1
-
-}

aarch64/kinglong/common/service/kill_self.go

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

aarch64/kinglong/common/service/rosbag_clean.go

@@ -1,36 +0,0 @@
-package service
-
-import (
-	"cicv-data-closedloop/aarch64/kinglong/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/kinglong/common/service/rosbag_record.go

@@ -1,112 +0,0 @@
-package service
-
-import (
-	"cicv-data-closedloop/aarch64/kinglong/common/config"
-	"cicv-data-closedloop/common/config/c_log"
-	"cicv-data-closedloop/common/util"
-	"github.com/bluenviron/goroslib/v2"
-	"time"
-)
-
-// BagRecord 打包rosbag
-func BagRecord(nodeName string) {
-	c_log.GlobalLogger.Info("rosbag record goroutine - 启动")
-	for {
-		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启动完成，正在启动record命令。")
-
-		var command []string
-		command = append(command, "record")
-		command = append(command, "--split")
-		command = append(command, "--duration=1")
-		for _, host := range config.CloudConfig.Hosts {
-			if host.Name == nodeName {
-				for _, topic := range host.Topics {
-					command = append(command, topic)
-				}
-			}
-		}
-
-		// 2 ------- 调用 rosbag 打包命令，该命令自动阻塞 -------
-		// 不在此处压缩，因为 rosbag filter 时会报错。在上传到oss之前压缩即可。
-		// 包名格式：2023-11-15-17-35-20_0.bag
-		_ = util.CreateParentDir(config.CloudConfig.BagDataDir)
-		cmd, err := util.ExecuteWithEnvAndDirAsync(config.RosbagEnvs, config.CloudConfig.BagDataDir, config.RosbagPath, command...)
-		if err != nil {
-			c_log.GlobalLogger.Error("执行record命令", command, "出错：", err)
-			continue
-		}
-		c_log.GlobalLogger.Info("启动record命令成功。")
-
-		recordProcessPid := cmd.Process.Pid
-		var recordSubProcessPid int
-		for {
-			time.Sleep(time.Duration(2) * time.Second)
-			recordSubProcessPid, err = util.GetSubProcessPid(recordProcessPid)
-			if err != nil {
-				c_log.GlobalLogger.Info("正在等待获取进程 ", recordProcessPid, " 的子进程的pid。")
-				continue
-			}
-			if recordSubProcessPid != 0 {
-				c_log.GlobalLogger.Info("获取进程 ", recordProcessPid, " 的子进程的pid：", recordSubProcessPid)
-				break
-			}
-		}
-		// 等待自杀信号
-		c_log.GlobalLogger.Info("rosbag record goroutine - 等待自杀信号")
-		select {
-		case signal := <-ChannelKillRosRecord:
-			if signal == 1 {
-				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命令一直录包占满存储
-				}
-				AddKillTimes("1")
-				return
-			}
-		}
-
-		// TODO 暂时不放开该逻辑。如果监控rosnode来判断是否杀死record，太麻烦，生成窗口的线程也需要关闭
-		//{
-		//	commonCfg.GlobalLogger.Info("正在监控rosnode是否全部关闭。")
-		//	for canRecord {
-		//		time.Sleep(time.Duration(1) * time.Second)
-		//		canRecord = isCanRecord(commonCfg.RosNode)
-		//	}
-		//	commonCfg.GlobalLogger.Info("rosnode已全部关闭，正在结束record进程。")
-		//
-		//	err = cmd.Process.Kill()
-		//	if err != nil {
-		//		commonCfg.GlobalLogger.Error("杀死record进程错误：", err)
-		//		continue
-		//	}
-		//}
-	}
-}
-
-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
-	}
-	myMap := nodes
-	mySlice := config.CloudConfig.Ros.Nodes
-	for _, element := range mySlice {
-		if _, ok := myMap[element]; !ok {
-			c_log.GlobalLogger.Info("rosnode：", element, " 未启动，需等待启动后才可启动record。")
-			return false
-		}
-	}
-	return true
-}

aarch64/kinglong/common/service/rosbag_upload.go

@@ -1,161 +0,0 @@
-package service
-
-import (
-	commonConfig "cicv-data-closedloop/aarch64/kinglong/common/config"
-	masterConfig "cicv-data-closedloop/aarch64/kinglong/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"
-	"fmt"
-	"os"
-	"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 := commonUtil.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
-		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 := commonUtil.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/" + nodeName + "_" + 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))
-		if commonConfig.LocalConfig.Node.Name == "master" {
-			// 在上传完成的包目录同级下添加一个目录同名的json
-			triggerIds := make([]string, 0)
-			for _, label := range currentTimeWindow.Labels {
-				triggerIdToAppend := masterConfig.LabelMapTriggerId[label]
-				c_log.GlobalLogger.Info("添加一个【triggerId】=", triggerIdToAppend)
-				triggerIds = append(triggerIds, triggerIdToAppend)
-			}
-			c_log.GlobalLogger.Info("json 中添加【triggerIds】=", triggerIds)
-			callBackMap := map[string]interface{}{
-				"dataName":    currentTimeWindow.FaultTime,
-				"dataSize":    "", // 由合并程序补充
-				"equipmentNo": commonConfig.LocalConfig.EquipmentNo,
-				"secretKey":   commonConfig.LocalConfig.SecretKey,
-				"rosBagPath":  objectKey2,
-				"filePath":    objectKey3,
-				"taskId":      commonConfig.PlatformConfig.TaskConfigId,
-				"triggerId":   triggerIds,
-			}
-			callBackJson, err := util.MapToJsonString(callBackMap)
-			c_log.GlobalLogger.Info("【callBackJson】=", callBackJson)
-			if err != nil {
-				c_log.GlobalLogger.Error("callBackMap", callBackMap, "转 json 失败：", err)
-			}
-			commonConfig.OssMutex.Lock()
-			err = commonConfig.OssBucket.PutObject(objectKey3+"callback.json", strings.NewReader(callBackJson))
-			commonConfig.OssMutex.Unlock()
-			if err != nil {
-				c_log.GlobalLogger.Error("上传 callback.json", callBackJson, "失败：", err)
-			}
-		}
-		// 删除本地所有已上传的bag文件
-		c_log.GlobalLogger.Infof("结束处理窗口，【Label】=%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 {
-			continue outLoop
-		}
-
-	}
-}

aarch64/kinglong/control/main.go

@@ -1,98 +0,0 @@
-package main
-
-import (
-	commonConfig "cicv-data-closedloop/aarch64/kinglong/common/config"
-	"cicv-data-closedloop/aarch64/kinglong/common/service"
-	"cicv-data-closedloop/common/config/c_log"
-	"cicv-data-closedloop/common/util"
-	"net/rpc"
-	"os"
-	"time"
-)
-
-func init() {
-	//runtime.GOMAXPROCS(1)
-	// 初始化本地配置文件（第1处配置，在本地文件）
-	commonConfig.InitLocalConfig()
-	// 初始化日志配置
-	c_log.InitLog(commonConfig.LocalConfig.LogDir, "kinglong-control")
-	// 初始化Oss连接信息
-	commonConfig.InitOssConfig()
-	// 初始化业务逻辑配置信息，配置文件在oss上（第2处配置，在oss文件）
-	commonConfig.InitCloudConfig()
-}
-
-func main() {
-	lastStatus := "NONE"
-	//  轮询任务接口判断是否有更新
-	for {
-		time.Sleep(time.Duration(2) * time.Second)
-		// 1 获取当前设备的任务的 status
-		status, err := commonConfig.GetStatus(commonConfig.PlatformConfig.TaskConfigId)
-		if err != nil {
-			c_log.GlobalLogger.Error("获取配置status失败：", err)
-			continue
-		}
-		// 2 判断 status
-		// UN_CHANGE 没有新的任务，无需更改
-		// CHANGE 有新的任务，需要杀死旧的任务并重启
-		// NONE 设备没有配置任务，需要杀死旧的任务
-		if status == "UN_CHANGE" {
-			lastStatus = "UN_CHANGE"
-			continue
-		} else if status == "CHANGE" || status == "NONE" {
-			if lastStatus == "CHANGE" && status == "CHANGE" { // 供更新使用
-				commonConfig.InitPlatformConfig()
-				continue
-			}
-			if lastStatus == "NONE" && status == "NONE" {
-				continue
-			}
-			// 3 发送rpc信号杀死两个服务，并重启程序
-			if lastStatus == "NONE" && status == "CHANGE" {
-				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)
-				lastStatus = status
-				c_log.GlobalLogger.Info("获取数据闭环平台最新配置。")
-				commonConfig.InitPlatformConfig()
-				continue
-			}
-			var killArgs service.KillSignal
-			if lastStatus == "UN_CHANGE" && status == "CHANGE" {
-				killArgs = service.KillSignal{NodeName: commonConfig.LocalConfig.Node.Name, DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: true}
-				c_log.GlobalLogger.Info("更新任务，发送rpc重启信号到本地"+commonConfig.LocalConfig.Node.Name+"：", killArgs)
-			}
-			if lastStatus == "UN_CHANGE" && status == "NONE" {
-				killArgs = service.KillSignal{NodeName: commonConfig.LocalConfig.Node.Name, DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: false}
-				c_log.GlobalLogger.Info("杀死任务，发送rpc结束信号到本地"+commonConfig.LocalConfig.Node.Name+"：", killArgs)
-			}
-
-			KillRpcClient, err := rpc.Dial("tcp", commonConfig.LocalConfig.Node.Ip+":"+commonConfig.CloudConfig.RpcPort)
-			if err != nil {
-				// 此处如果连接失败说明采集程序已经停止了
-				lastStatus = "NONE"
-				c_log.GlobalLogger.Error("采集程序已经停止：", err)
-				continue
-			}
-
-			reply := 0
-			if err = KillRpcClient.Call("KillService.Kill", killArgs, &reply); err != nil {
-				c_log.GlobalLogger.Error("发送rpc请求到master失败：", err)
-				// 这里可能会报错 unexpected EOF 但是不影响，先注释 close 和 continue
-				//KillRpcClient.Close()
-				//continue
-			}
-			lastStatus = status
-			c_log.GlobalLogger.Info("结束任务后，将数据闭环平台配置置空。")
-			commonConfig.PlatformConfig = commonConfig.PlatformConfigStruct{}
-			if err = KillRpcClient.Close(); err != nil {
-				// 不做处理
-			}
-		} else {
-			c_log.GlobalLogger.Error("未知的采集任务状态。【status】=", status)
-		}
-	}
-}

aarch64/kinglong/master/main.go

@@ -1,54 +0,0 @@
-package main
-
-import (
-	commonConfig "cicv-data-closedloop/aarch64/kinglong/common/config"
-	commonService "cicv-data-closedloop/aarch64/kinglong/common/service"
-	masterConfig "cicv-data-closedloop/aarch64/kinglong/master/package/config"
-	masterService "cicv-data-closedloop/aarch64/kinglong/master/package/service"
-	"cicv-data-closedloop/common/config/c_log"
-	"cicv-data-closedloop/common/util"
-)
-
-func init() {
-	//runtime.GOMAXPROCS(1)
-	// 初始化本地配置文件（第1处配置，在本地文件）
-	commonConfig.InitLocalConfig()
-	// 初始化日志配置
-	c_log.InitLog(commonConfig.LocalConfig.LogDir, "kinglong-soc1")
-	// 初始化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()
-	// 维护data目录缓存的包数量
-	go commonService.BagCacheClean()
-	// 磁盘占用过高时根据缓存策略处理copy目录
-	go commonService.DiskClean()
-	// 初始化加载触发器插件文件
-	masterConfig.InitTriggerConfig()
-	// 初始化rpc监听
-	commonConfig.InitKillSignalListener(commonConfig.CloudConfig.Hosts[0].Ip)
-	// 等待重启，接收到重启信号，会把信号分发给以下channel
-	go commonService.WaitKillSelf()
-}
-
-func main() {
-
-	// 1 负责打包数据到data目录
-	go commonService.BagRecord(commonConfig.CloudConfig.Hosts[0].Name)
-	// 2 启动第4个线程，负责监控故障，并修改timeWindow
-	go masterService.PrepareTimeWindowProducerQueue()
-	// 3
-	go masterService.RunTimeWindowProducerQueue()
-	// 4 排队运行时间窗口
-	go commonService.RunTimeWindowConsumerQueue(commonConfig.CloudConfig.Hosts[0].Name)
-
-	// 阻塞主线程，等待其他线程执行。
-	select {}
-}

aarch64/kinglong/master/package/config/master_trigger_cfg.go

@@ -1,172 +0,0 @@
-package config
-
-import (
-	"cicv-data-closedloop/aarch64/kinglong/common/config"
-	"cicv-data-closedloop/common/config/c_log"
-	"cicv-data-closedloop/common/util"
-	"cicv-data-closedloop/kinglong_msgs"
-	entity "cicv-data-closedloop/kinglong_param"
-	"github.com/bluenviron/goroslib/v2/pkg/msgs/std_msgs"
-	"plugin"
-	"strconv"
-)
-
-var (
-	// 扩展
-	TopicOfCicvExtend = "/cicv_extend"
-	RuleOfCicvExtend  []func(param entity.KinglongParam) string
-
-	// 1
-	TopicOfNodeFaultInfo = "/nodefault_info"
-	RuleOfNodefaultInfo  []func(data *kinglong_msgs.FaultInfo) string
-	// 2
-	TopicOfCicvLocation = "/cicv_location"
-	RuleOfCicvLocation  []func(data *kinglong_msgs.PerceptionLocalization) string
-	// 3
-	TopicOfTpperception = "/tpperception"
-	RuleOfTpperception  []func(data *kinglong_msgs.PerceptionObjects, param entity.KinglongParam) string
-	// 4
-	TopicOfFaultInfo = "/fault_info"
-	RuleOfFaultInfo  []func(data *kinglong_msgs.FaultVec) string
-	// 5
-	TopicOfDataRead = "/data_read"
-	RuleOfDataRead  []func(data *kinglong_msgs.Retrieval) string
-	// 6
-	TopicOfJinlongControlPub = "/jinlong_control_pub"
-	RuleOfJinlongControlPub  []func(data *kinglong_msgs.JinlongControlCommand) string
-	// 7
-	TopicOfFailureLidar = "/failure/lidar"
-	RuleOfFailureLidar  []func(data *std_msgs.Bool) string
-	// 8
-	TopicOfFailureRadar = "/failure/radar"
-	RuleOfFailureRadar  []func(data *std_msgs.Bool) string
-	// 9
-	TopicOfFailureCamera = "/failure/camera"
-	RuleOfFailureCamera  []func(data *std_msgs.Bool) string
-
-	LabelMapTriggerId = make(map[string]string)
-)
-
-func InitTriggerConfig() {
-	c_log.GlobalLogger.Info("主节点加载触发器插件 - 开始。")
-	// 下载所有触发器的文件
-	for _, trigger := range config.PlatformConfig.TaskTriggers {
-		triggerLocalPath := config.CloudConfig.TriggersDir + trigger.TriggerScriptPath
-		_ = util.CreateParentDir(triggerLocalPath)
-		c_log.GlobalLogger.Info("下载触发器插件从 ", trigger.TriggerScriptPath, " 到 ", triggerLocalPath)
-		config.OssMutex.Lock()
-		err := config.OssBucket.GetObjectToFile(trigger.TriggerScriptPath, triggerLocalPath)
-		config.OssMutex.Unlock()
-		if err != nil {
-			c_log.GlobalLogger.Error("下载 oss 上的触发器插件失败：", err)
-			continue
-		}
-		// 载入插件到数组
-		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 TopicOfCicvExtend == topic2 { // 自定义扩展
-			f, ok := rule.(func(param entity.KinglongParam) string)
-			if ok != true {
-				c_log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func(param entity.ExtendParam):", err)
-				continue
-			}
-			RuleOfCicvExtend = append(RuleOfCicvExtend, f)
-		} else if TopicOfNodeFaultInfo == topic2 { // 1
-			f, ok := rule.(func(data *kinglong_msgs.FaultInfo) string)
-			if ok != true {
-				c_log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func(data *kinglong_msgs.FaultInfo) string):", err)
-				continue
-			}
-			RuleOfNodefaultInfo = append(RuleOfNodefaultInfo, f)
-		} else if TopicOfCicvLocation == topic2 { // 2
-			f, ok := rule.(func(data *kinglong_msgs.PerceptionLocalization) string)
-			if ok != true {
-				c_log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func(data *kinglong_msgs.PerceptionLocalization) string):", err)
-				continue
-			}
-			RuleOfCicvLocation = append(RuleOfCicvLocation, f)
-		} else if TopicOfTpperception == topic2 { // 3
-			f, ok := rule.(func(data *kinglong_msgs.PerceptionObjects, param entity.KinglongParam) string)
-			if ok != true {
-				c_log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func(data *kinglong_msgs.PerceptionObjects, param entity.KinglongParam) string):", err)
-				continue
-			}
-			RuleOfTpperception = append(RuleOfTpperception, f)
-		} else if TopicOfFaultInfo == topic2 { // 4
-			f, ok := rule.(func(data *kinglong_msgs.FaultVec) string)
-			if ok != true {
-				c_log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func(data *kinglong_msgs.FaultVec) string):", err)
-				continue
-			}
-			RuleOfFaultInfo = append(RuleOfFaultInfo, f)
-		} else if TopicOfDataRead == topic2 { // 5
-			f, ok := rule.(func(data *kinglong_msgs.Retrieval) string)
-			if ok != true {
-				c_log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func(data *kinglong_msgs.Retrieval) string):", err)
-				continue
-			}
-			RuleOfDataRead = append(RuleOfDataRead, f)
-		} else if TopicOfJinlongControlPub == topic2 { // 6
-			f, ok := rule.(func(data *kinglong_msgs.JinlongControlCommand) string)
-			if ok != true {
-				c_log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func(data *std_msgs.Bool) string):", err)
-				continue
-			}
-			RuleOfJinlongControlPub = append(RuleOfJinlongControlPub, f)
-		} else if TopicOfFailureLidar == topic2 { // 7
-			f, ok := rule.(func(data *std_msgs.Bool) string)
-			if ok != true {
-				c_log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func(data *std_msgs.Bool) string):", err)
-				continue
-			}
-			RuleOfFailureLidar = append(RuleOfFailureLidar, f)
-		} else if TopicOfFailureRadar == topic2 { // 8
-			f, ok := rule.(func(data *std_msgs.Bool) string)
-			if ok != true {
-				c_log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func(data *std_msgs.Bool) string):", err)
-				continue
-			}
-			RuleOfFailureRadar = append(RuleOfFailureRadar, f)
-		} else if TopicOfFailureCamera == topic2 { // 9
-			f, ok := rule.(func(data *std_msgs.Bool) string)
-			if ok != true {
-				c_log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func(data *kinglong_msgs.JinlongControlCommand) string):", err)
-				continue
-			}
-			RuleOfFailureCamera = append(RuleOfFailureCamera, f)
-		} else {
-			c_log.GlobalLogger.Error("未知的topic：", topic2)
-			continue
-		}
-
-		label, err := open.Lookup("Label")
-		if err != nil {
-			c_log.GlobalLogger.Error("加载本地插件 ", triggerLocalPath, " 中的 Label 方法失败。", err)
-			continue
-		}
-		labelFunc := label.(func() string)
-		labelString := labelFunc()
-		LabelMapTriggerId[labelString] = strconv.Itoa(trigger.TriggerId)
-		c_log.GlobalLogger.Info("主节点加载触发器插件：【ros topic】=", topic2, "，【触发器label】=", labelString, "【触发器ID】=", trigger.TriggerId)
-	}
-	c_log.GlobalLogger.Infof("一共有%v个触发器，【label和id映射关系】=%v", len(config.PlatformConfig.TaskTriggers), LabelMapTriggerId)
-}

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

@@ -1,105 +0,0 @@
-package service
-
-import (
-	commonConfig "cicv-data-closedloop/aarch64/kinglong/common/config"
-	commonService "cicv-data-closedloop/aarch64/kinglong/common/service"
-	"cicv-data-closedloop/common/config/c_log"
-	"cicv-data-closedloop/common/domain"
-	"cicv-data-closedloop/common/entity"
-	"cicv-data-closedloop/common/util"
-	"net"
-	"time"
-)
-
-// RunTimeWindowProducerQueue 将时间窗口内的包全部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 {
-				// 1 如果第一个已经大于了timeWindowEnd，则触发上传并删除
-				// 将时间窗口发送给从节点
-				currentTimeWindow.CanUpload = "yes"
-				c_log.GlobalLogger.Info("将已完成的窗口发送给从节点：", currentTimeWindow.CanUpload)
-				domain.SupplyCopyBags(commonConfig.CloudConfig.BagDataDir, commonConfig.CloudConfig.BagCopyDir, currentTimeWindow)
-				entity.RefreshTcpSendTime()
-				go sendTimeWindowByTcp(currentTimeWindow)
-				// 将时间窗口移出准备队列
-				entity.RemoveHeadOfTimeWindowProducerQueue()
-				// 将时间窗口加入运行队列
-				entity.AddTimeWindowToTimeWindowConsumerQueue(currentTimeWindow)
-				// 获取copy目录下的字典json，key为触发时间，value为label
-				timeToLabelJson, _ := util.ReadFile(commonConfig.CloudConfig.TimeToLabelJsonPath)
-				timeToLabelMap, _ := util.JsonStringToMap(timeToLabelJson)
-				timeToLabelMap[currentTimeWindow.FaultTime] = util.ToString(currentTimeWindow.Labels)
-				timeToLabelJson, _ = util.MapToJsonString(timeToLabelMap)
-				_ = util.WriteFile(timeToLabelJson, commonConfig.CloudConfig.TimeToLabelJsonPath)
-				continue
-			} else { // 保证当前窗口只发送一次，每间隔5秒发一次
-				if int(time.Since(entity.TcpSendTime).Seconds()) > commonConfig.CloudConfig.TimeWindowSendGap {
-					c_log.GlobalLogger.Info("每隔", commonConfig.CloudConfig.TimeWindowSendGap, "秒发送一次tcp消息")
-					entity.RefreshTcpSendTime()
-					// 2 如果第一个不大于timeWindowEnd，则发送不可上传的窗口信息。
-					currentTimeWindow.CanUpload = "no"
-					go sendTimeWindowByTcp(currentTimeWindow)
-				}
-			}
-		}
-	}
-}
-
-func sendTimeWindowByTcp(timeWindow entity.TimeWindow) {
-	socket := commonConfig.CloudConfig.Hosts[1].Ip + ":" + commonConfig.CloudConfig.TcpPort
-	tcpConn, err := net.Dial("tcp", socket)
-	if err != nil {
-		c_log.GlobalLogger.Error("建立tcp连接", socket, "失败：", err)
-		return
-	}
-	defer tcpConn.Close()
-	// 发送数据
-	send, _ := entity.TimeWindowToJson(timeWindow)
-	_, err = tcpConn.Write([]byte(send))
-	if err != nil {
-		c_log.GlobalLogger.Error("master发送给slave时间窗口", timeWindow, "失败：", err)
-		return
-	}
-}

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

@@ -1,382 +0,0 @@
-package service
-
-import (
-	commonConfig "cicv-data-closedloop/aarch64/kinglong/common/config"
-	"cicv-data-closedloop/aarch64/kinglong/common/service"
-	masterConfig "cicv-data-closedloop/aarch64/kinglong/master/package/config"
-	"cicv-data-closedloop/common/config/c_log"
-	commonEntity "cicv-data-closedloop/common/entity"
-	"cicv-data-closedloop/common/util"
-	"cicv-data-closedloop/kinglong_msgs"
-	"cicv-data-closedloop/kinglong_param"
-	"github.com/bluenviron/goroslib/v2"
-	"github.com/bluenviron/goroslib/v2/pkg/msgs/std_msgs"
-	"math"
-	"sync"
-	"time"
-)
-
-var (
-	extendParam kinglong_param.KinglongParam
-	// /cicv_location
-	mutexOfCicvLocation sync.RWMutex
-	// /tpperception
-	mutexOfTpperception sync.RWMutex
-	// /pj_control_pub
-	mutexOfJinlongControlPub sync.RWMutex
-	// /data_read
-	mutexOfDataRead sync.RWMutex
-)
-
-// PrepareTimeWindowProducerQueue 负责监听所有主题并修改时间窗口
-func PrepareTimeWindowProducerQueue() {
-	c_log.GlobalLogger.Info("订阅者 goroutine，启动。")
-
-	var err error
-	subscribers := make([]*goroslib.Subscriber, len(commonConfig.SubscribeTopics))
-	subscribersTimes := make([]time.Time, len(commonConfig.SubscribeTopics))
-	subscribersTimeMutexes := make([]sync.Mutex, len(commonConfig.SubscribeTopics))
-	subscribersMutexes := make([]sync.Mutex, len(commonConfig.SubscribeTopics))
-	for i, topic := range commonConfig.SubscribeTopics {
-		// !!!扩展的定时任务监听，牛逼!!!
-		if topic == masterConfig.TopicOfCicvExtend {
-			for {
-				time.Sleep(time.Duration(3500) * time.Millisecond)
-				for _, f := range masterConfig.RuleOfCicvExtend {
-					label := f(extendParam)
-					if label != "" {
-						saveTimeWindow(label, util.GetNowTimeCustom(), commonEntity.GetLastTimeWindow())
-						break
-					}
-				}
-			}
-		}
-
-		c_log.GlobalLogger.Info("创建订阅者订阅话题：" + topic)
-
-		if topic == masterConfig.TopicOfCicvLocation && len(masterConfig.RuleOfCicvLocation) > 0 {
-			subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
-				Node:  commonConfig.RosNode,
-				Topic: topic,
-				Callback: func(data *kinglong_msgs.PerceptionLocalization) {
-					// 更新共享变量
-					mutexOfCicvLocation.RLock()
-					{
-						extendParam.VelocityYOfCicvLocation = data.VelocityX
-						extendParam.VelocityYOfCicvLocation = data.VelocityY
-						extendParam.YawOfCicvLocation = data.Yaw
-						extendParam.AngularVelocityZOfCicvLocation = data.AngularVelocityZ
-					}
-					mutexOfCicvLocation.RUnlock()
-
-					subscribersTimeMutexes[i].Lock()
-					if time.Since(subscribersTimes[i]).Seconds() > 1 {
-						subscribersMutexes[i].Lock()
-						faultHappenTime := util.GetNowTimeCustom()         // 获取当前故障发生时间
-						lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
-						var faultLabel string
-						for _, f := range masterConfig.RuleOfCicvLocation {
-							faultLabel = f(data)
-							if faultLabel != "" {
-								saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
-								break
-							}
-						}
-						subscribersMutexes[i].Unlock()
-					}
-					subscribersTimeMutexes[i].Unlock()
-				},
-			})
-		}
-
-		if topic == masterConfig.TopicOfTpperception && len(masterConfig.RuleOfTpperception) > 0 {
-			subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
-				Node:  commonConfig.RosNode,
-				Topic: topic,
-				Callback: func(data *kinglong_msgs.PerceptionObjects) {
-					// ------- 触发前更新共享变量
-					mutexOfTpperception.RLock()
-					{
-						for _, obj := range data.Objs {
-							if (obj.Type != 1 && obj.Type != 0) || obj.X <= 5 || math.Abs(float64(obj.Y)) >= 10 {
-								continue
-							}
-							if _, ok := extendParam.ObjDicOfTpperception[obj.Id]; !ok {
-								extendParam.ObjDicOfTpperception[obj.Id] = []float32{}
-							}
-							extendParam.ObjDicOfTpperception[obj.Id] = append(extendParam.ObjDicOfTpperception[obj.Id], obj.Y)
-						}
-					}
-					mutexOfTpperception.RUnlock()
-
-					subscribersTimeMutexes[i].Lock()
-					if time.Since(subscribersTimes[i]).Seconds() > 1 {
-						subscribersMutexes[i].Lock()
-						faultHappenTime := util.GetNowTimeCustom()         // 获取当前故障发生时间
-						lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
-						var faultLabel string
-						for _, f := range masterConfig.RuleOfTpperception {
-							faultLabel = f(data, extendParam)
-							if faultLabel != "" {
-								saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
-								break
-							}
-						}
-						subscribersMutexes[i].Unlock()
-					}
-					subscribersTimeMutexes[i].Unlock()
-
-					// -------- 触发后更新共享变量
-					mutexOfTpperception.RLock()
-					{
-						for _, obj := range data.Objs {
-							extendParam.ObjTypeDicOfTpperception[obj.Id] = obj.Type
-							extendParam.ObjSpeedDicOfTpperception[obj.Id] = math.Pow(math.Pow(float64(obj.Vxabs), 2)+math.Pow(float64(obj.Vyabs), 2), 0.5)
-						}
-					}
-					mutexOfTpperception.RUnlock()
-
-				},
-			})
-		}
-
-		// 5
-		if topic == masterConfig.TopicOfDataRead {
-			subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
-				Node:  commonConfig.RosNode,
-				Topic: topic,
-				Callback: func(data *kinglong_msgs.Retrieval) {
-					// 更新共享变量
-					mutexOfDataRead.RLock()
-					{
-						extendParam.NumCountDataReadOfDataRead++
-						if extendParam.NumCountDataReadOfDataRead == 10 {
-							extendParam.EgoSteeringRealOfDataRead = append(extendParam.EgoSteeringRealOfDataRead, data.StrgAngleRealValue)
-							extendParam.EgoThrottleRealOfDataRead = append(extendParam.EgoThrottleRealOfDataRead, data.VcuAccelPosValue)
-							extendParam.EgoBrakeRealOfDataRead = append(extendParam.EgoBrakeRealOfDataRead, data.VcuBrkPelPosValue)
-							extendParam.NumCountDataReadOfDataRead = 0
-						}
-					}
-					mutexOfDataRead.RUnlock()
-					subscribersTimeMutexes[i].Lock()
-					if time.Since(subscribersTimes[i]).Seconds() > 1 {
-						subscribersMutexes[i].Lock()
-						faultHappenTime := util.GetNowTimeCustom()         // 获取当前故障发生时间
-						lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
-						var faultLabel string
-						for _, f := range masterConfig.RuleOfDataRead {
-							faultLabel = f(data)
-							if faultLabel != "" {
-								saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
-								subscribersTimes[i] = time.Now()
-								break
-							}
-						}
-						subscribersMutexes[i].Unlock()
-					}
-					subscribersTimeMutexes[i].Unlock()
-				},
-			})
-		}
-
-		// 6
-		if topic == masterConfig.TopicOfJinlongControlPub && len(masterConfig.RuleOfJinlongControlPub) > 0 {
-			subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
-				Node:  commonConfig.RosNode,
-				Topic: topic,
-				Callback: func(data *kinglong_msgs.JinlongControlCommand) {
-					// 更新共享变量
-					mutexOfJinlongControlPub.RLock()
-					{
-						extendParam.NumCountJinlongControlCommandOfPjControlPub++
-						if extendParam.NumCountJinlongControlCommandOfPjControlPub == 10 {
-							extendParam.EgoSteeringCmdOfJinlongControlPub = append(extendParam.EgoSteeringCmdOfJinlongControlPub, data.ASStrgAngleReq)
-							extendParam.EgoThrottleCmdOfJinlongControlPub = append(extendParam.EgoThrottleCmdOfJinlongControlPub, data.ASAutoDAccelPosReq)
-							extendParam.EgoBrakeCmdOfJinlongControlPub = append(extendParam.EgoBrakeCmdOfJinlongControlPub, data.ASAutoDBrkPelPosReq)
-							extendParam.NumCountJinlongControlCommandOfPjControlPub = 0
-						}
-					}
-					mutexOfJinlongControlPub.RUnlock()
-
-					subscribersTimeMutexes[i].Lock()
-					if time.Since(subscribersTimes[i]).Seconds() > 1 {
-						subscribersMutexes[i].Lock()
-						faultHappenTime := util.GetNowTimeCustom()         // 获取当前故障发生时间
-						lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
-						var faultLabel string
-						for _, f := range masterConfig.RuleOfJinlongControlPub {
-							faultLabel = f(data)
-							if faultLabel != "" {
-								saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
-								break
-							}
-						}
-						subscribersMutexes[i].Unlock()
-					}
-					subscribersTimeMutexes[i].Unlock()
-				},
-			})
-		}
-
-		// 7
-		if topic == masterConfig.TopicOfFailureLidar && len(masterConfig.RuleOfFailureLidar) > 0 {
-			subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
-				Node:  commonConfig.RosNode,
-				Topic: topic,
-				Callback: func(data *std_msgs.Bool) {
-					subscribersTimeMutexes[i].Lock()
-					if time.Since(subscribersTimes[i]).Seconds() > 1 {
-						subscribersMutexes[i].Lock()
-						faultHappenTime := util.GetNowTimeCustom()         // 获取当前故障发生时间
-						lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
-						var faultLabel string
-						for _, f := range masterConfig.RuleOfFailureLidar {
-							faultLabel = f(data)
-							if faultLabel != "" {
-								saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
-								break
-							}
-						}
-						subscribersMutexes[i].Unlock()
-					}
-					subscribersTimeMutexes[i].Unlock()
-				},
-			})
-		}
-
-		// 8
-		if topic == masterConfig.TopicOfFailureRadar && len(masterConfig.RuleOfFailureRadar) > 0 {
-			subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
-				Node:  commonConfig.RosNode,
-				Topic: topic,
-				Callback: func(data *std_msgs.Bool) {
-					subscribersTimeMutexes[i].Lock()
-					if time.Since(subscribersTimes[i]).Seconds() > 1 {
-						subscribersMutexes[i].Lock()
-						faultHappenTime := util.GetNowTimeCustom()         // 获取当前故障发生时间
-						lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
-						var faultLabel string
-						for _, f := range masterConfig.RuleOfFailureRadar {
-							faultLabel = f(data)
-							if faultLabel != "" {
-								saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
-								break
-							}
-						}
-						subscribersMutexes[i].Unlock()
-					}
-					subscribersTimeMutexes[i].Unlock()
-				},
-			})
-		}
-
-		// 9
-		if topic == masterConfig.TopicOfFailureCamera && len(masterConfig.RuleOfFailureLidar) > 0 {
-			subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
-				Node:  commonConfig.RosNode,
-				Topic: topic,
-				Callback: func(data *std_msgs.Bool) {
-					subscribersTimeMutexes[i].Lock()
-					if time.Since(subscribersTimes[i]).Seconds() > 1 {
-						subscribersMutexes[i].Lock()
-						faultHappenTime := util.GetNowTimeCustom()         // 获取当前故障发生时间
-						lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
-						var faultLabel string
-						for _, f := range masterConfig.RuleOfFailureLidar {
-							faultLabel = f(data)
-							if faultLabel != "" {
-								saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
-								break
-							}
-						}
-						subscribersMutexes[i].Unlock()
-					}
-					subscribersTimeMutexes[i].Unlock()
-				},
-			})
-		}
-
-		if err != nil {
-			c_log.GlobalLogger.Info("创建订阅者报错：", err)
-			//TODO 如何回传日志
-			continue
-		}
-	}
-
-	select {
-	case signal := <-service.ChannelKillWindowProducer:
-		if signal == 1 {
-			commonConfig.RosNode.Close()
-			service.AddKillTimes("3")
-			return
-		}
-	}
-}
-
-func saveTimeWindow(faultLabel string, faultHappenTime string, lastTimeWindow *commonEntity.TimeWindow) {
-	masterTopics, slaveTopics := getTopicsOfNode(faultLabel)
-	if lastTimeWindow == nil || util.TimeCustom1GreaterTimeCustom2(faultHappenTime, lastTimeWindow.TimeWindowEnd) {
-		// 2-1 如果是不在旧故障窗口内，添加一个新窗口
-		newTimeWindow := commonEntity.TimeWindow{
-			FaultTime:       faultHappenTime,
-			TimeWindowBegin: util.TimeCustomChange(faultHappenTime, -commonConfig.PlatformConfig.TaskBeforeTime),
-			TimeWindowEnd:   util.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)
-		commonEntity.AddTimeWindowToTimeWindowProducerQueue(newTimeWindow)
-	} else {
-		// 2-2 如果在旧故障窗口内
-		commonEntity.TimeWindowProducerQueueMutex.RLock()
-		defer commonEntity.TimeWindowProducerQueueMutex.RUnlock()
-		// 2-2-1 更新故障窗口end时间
-		maxEnd := util.TimeCustomChange(lastTimeWindow.TimeWindowBegin, commonConfig.PlatformConfig.TaskMaxTime)
-		expectEnd := util.TimeCustomChange(faultHappenTime, commonConfig.PlatformConfig.TaskAfterTime)
-		if util.TimeCustom1GreaterTimeCustom2(expectEnd, maxEnd) {
-			lastTimeWindow.TimeWindowEnd = maxEnd
-			lastTimeWindow.Length = commonConfig.PlatformConfig.TaskMaxTime
-		} else {
-			if util.TimeCustom1GreaterTimeCustom2(expectEnd, lastTimeWindow.TimeWindowEnd) {
-				lastTimeWindow.TimeWindowEnd = expectEnd
-				lastTimeWindow.Length = util.CalculateDifferenceOfTimeCustom(lastTimeWindow.TimeWindowBegin, expectEnd)
-			}
-		}
-		// 2-2-2 更新label
-		labels := lastTimeWindow.Labels
-		lastTimeWindow.Labels = util.AppendIfNotExists(labels, faultLabel)
-		// 2-2-3 更新 topic
-		sourceMasterTopics := lastTimeWindow.MasterTopics
-		lastTimeWindow.MasterTopics = util.MergeSlice(sourceMasterTopics, masterTopics)
-		sourceSlaveTopics := lastTimeWindow.SlaveTopics
-		lastTimeWindow.SlaveTopics = util.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
-		}
-	}
-	// 根据不同节点采集的topic进行分配采集
-	for _, acceptTopic := range faultCodeTopics {
-		for _, host := range commonConfig.CloudConfig.Hosts {
-			for _, topic := range host.Topics {
-				if host.Name == commonConfig.CloudConfig.Hosts[0].Name && acceptTopic == topic {
-					masterTopics = append(masterTopics, acceptTopic)
-				}
-				if host.Name == commonConfig.CloudConfig.Hosts[1].Name && acceptTopic == topic {
-					slaveTopics = append(slaveTopics, acceptTopic)
-				}
-			}
-		}
-	}
-	return masterTopics, slaveTopics
-}

aarch64/kinglong/slave/main.go

@@ -1,62 +0,0 @@
-package main
-
-import (
-	commonConfig "cicv-data-closedloop/aarch64/kinglong/common/config"
-	commonService "cicv-data-closedloop/aarch64/kinglong/common/service"
-	slaveConfig "cicv-data-closedloop/aarch64/kinglong/slave/package/config"
-	slaveService "cicv-data-closedloop/aarch64/kinglong/slave/package/service"
-	"cicv-data-closedloop/common/config/c_log"
-	"cicv-data-closedloop/common/util"
-)
-
-// init 初始化函数
-func init() {
-	//runtime.GOMAXPROCS(1)
-	// 初始化本地配置文件（第1处配置，在本地文件）
-	commonConfig.InitLocalConfig()
-	// 初始化日志配置
-	c_log.InitLog(commonConfig.LocalConfig.LogDir, "kinglong-soc2")
-
-	// 初始化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()
-
-	// 维护data目录缓存的包数量
-	go commonService.BagCacheClean()
-
-	// 磁盘占用过高时根据缓存策略处理copy目录
-	go commonService.DiskClean()
-	slaveConfig.InitTcpListener()
-	commonConfig.InitKillSignalListener(commonConfig.CloudConfig.Hosts[1].Ip)
-	// 等待重启，接收到重启信号，会把信号分发给以下channel
-	go commonService.WaitKillSelf()
-}
-
-// main 主函数
-func main() {
-
-	// 1 负责打包数据到data目录
-	go commonService.BagRecord(commonConfig.CloudConfig.Hosts[1].Name)
-	// 2 负责监控故障，并修改timeWindow
-	go slaveService.PrepareTimeWindowProducerQueue()
-	// 3
-	go slaveService.RunTimeWindowProducerQueue()
-	// 4 排队运行时间窗口
-	go commonService.RunTimeWindowConsumerQueue(commonConfig.CloudConfig.Hosts[1].Name)
-
-	// 阻塞主线程，等待其他线程执行。
-	select {}
-
-}

aarch64/kinglong/slave/package/config/slave_tcp_config.go

@@ -1,22 +0,0 @@
-package config
-
-import (
-	commonConfig "cicv-data-closedloop/aarch64/kinglong/common/config"
-	"cicv-data-closedloop/common/config/c_log"
-	"net"
-	"os"
-)
-
-var TcpListener net.Listener
-
-func InitTcpListener() {
-	c_log.GlobalLogger.Info("从节点初始化TCP端口监听 - 开始。")
-	socket := commonConfig.CloudConfig.Hosts[1].Ip + ":" + commonConfig.CloudConfig.TcpPort
-	var err error
-	TcpListener, err = net.Listen("tcp", socket)
-	if err != nil {
-		c_log.GlobalLogger.Error("监听tcp端口失败:", err)
-		os.Exit(-1)
-	}
-	c_log.GlobalLogger.Info("从节点初始化TCP端口监听 - 成功：", socket)
-}

aarch64/kinglong/slave/package/service/accept_window.go

@@ -1,62 +0,0 @@
-package service
-
-import (
-	"cicv-data-closedloop/aarch64/kinglong/common/service"
-	slaveConfig "cicv-data-closedloop/aarch64/kinglong/slave/package/config"
-	"cicv-data-closedloop/common/config/c_log"
-	"cicv-data-closedloop/common/entity"
-	"context"
-	"encoding/json"
-	"sync"
-)
-
-func PrepareTimeWindowProducerQueue() {
-	var prepareTimeWindowProducerQueueMutex sync.Mutex
-	ctx, cancel := context.WithCancel(context.Background())
-	// 处理退出信号
-	go func() {
-		select {
-		case signal := <-service.ChannelKillWindowProducer:
-			if signal == 1 {
-				cancel()
-				slaveConfig.TcpListener.Close()
-				service.AddKillTimes("3")
-				return
-			}
-		}
-	}()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		default:
-			conn, err := slaveConfig.TcpListener.Accept()
-			if err != nil {
-				select {
-				case <-ctx.Done():
-					return
-				default:
-					c_log.GlobalLogger.Error("接受连接错误:", err)
-					continue
-				}
-			}
-			prepareTimeWindowProducerQueueMutex.Lock()
-			buffer := make([]byte, 2048)
-			total, err := conn.Read(buffer)
-			if err != nil {
-				c_log.GlobalLogger.Error("读取数据错误:", err)
-				continue
-			}
-			var timeWindow entity.TimeWindow
-			err = json.Unmarshal(buffer[:total], &timeWindow)
-			if err != nil {
-				c_log.GlobalLogger.Error("解析Json时出错:", err)
-				continue
-			}
-			entity.AddTimeWindowToTimeWindowProducerQueue(timeWindow)
-			prepareTimeWindowProducerQueueMutex.Unlock()
-		}
-
-	}
-}

aarch64/kinglong/slave/package/service/move_bag.go

@@ -1,70 +0,0 @@
-package service
-
-import (
-	commonConfig "cicv-data-closedloop/aarch64/kinglong/common/config"
-	commonService "cicv-data-closedloop/aarch64/kinglong/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"
-)
-
-// RunTimeWindowProducerQueue 将时间窗口内的包全部move出去，并等待当前时间窗口结束触发上传
-func RunTimeWindowProducerQueue() {
-	c_log.GlobalLogger.Info("生产者队列 - 启动")
-	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 {
-			currentTimeWindow := entity.TimeWindowProducerQueue[0]
-			// 将时间窗口移出准备队列
-			entity.RemoveHeadOfTimeWindowProducerQueue()
-			if currentTimeWindow.CanUpload == "yes" {
-				c_log.GlobalLogger.Info("从节点接收到可上传的timeWindow")
-			}
-			if currentTimeWindow.CanUpload == "no" {
-				c_log.GlobalLogger.Info("从节点接收到不可上传的timeWindow")
-			}
-
-			// 2 timeWindow不可以上传，则将data目录下的数据move到copy目录
-			bags, _ := util.ListAbsolutePathWithSuffixAndSort(commonConfig.CloudConfig.BagDataDir, ".bag")
-			for _, bag := range bags {
-				bagTime := util.GetBagTime(bag)
-				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)
-				} else {
-					if util.TimeCustom1GreaterEqualThanTimeCustom2(bagTime, currentTimeWindow.TimeWindowBegin) {
-						// 必须已经生成了窗口之后的包才算窗口结束了
-						break
-					}
-				}
-			}
-			// 判断是否可上传
-			if currentTimeWindow.CanUpload == "yes" {
-				// 1 timeWindow可以上传
-				c_log.GlobalLogger.Info("timeWindow可以上传：", currentTimeWindow)
-				// 补充bag包
-				domain.SupplyCopyBags(commonConfig.CloudConfig.BagDataDir, commonConfig.CloudConfig.BagCopyDir, currentTimeWindow)
-				// 将时间窗口加入运行队列
-				entity.AddTimeWindowToTimeWindowConsumerQueue(currentTimeWindow)
-			}
-		}
-	}
-}

aarch64/pjisuv/README.md

@@ -0,0 +1,7 @@
+# 目录结构
+## common
+- c_cloud.go：阿里云oss配置文件下载解析
+## control 调度程序
+## master 102 程序
+- trigger_init.go：102 初始化加载触发器插件
+## slave 103 程序

aarch64/pjisuv/master/config/trigger_init.go

@@ -16,6 +16,8 @@ import (
 )
 
 func InitTriggerConfig() {
+	config.OssMutex.Lock()
+	defer config.OssMutex.Unlock()
 	triggerLocalPathsMapTriggerId := make(map[string]string)
 	c_log.GlobalLogger.Info("主节点加载触发器插件 - 开始。")
 	// 1 获取数采任务的触发器列表
@@ -41,14 +43,14 @@ func InitTriggerConfig() {
 		}
 		label := util.GetFileNameWithoutExtension(config.CloudConfig.TriggersDir + trigger.TriggerScriptPath)
 		triggerLocalPath = config.CloudConfig.TriggersDir + id + "/" + label + ".so"
-		c_log.GlobalLogger.Info("下载触发器插件从 ", trigger.TriggerScriptPath, " 到 ", triggerLocalPath)
-		config.OssMutex.Lock()
+		c_log.GlobalLogger.Info("开始下载触发器插件从 ", trigger.TriggerScriptPath, " 到 ", triggerLocalPath)
 		_ = util.CreateParentDir(triggerLocalPath)
-		err := config.OssBucket.GetObjectToFile(trigger.TriggerScriptPath, triggerLocalPath)
-		config.OssMutex.Unlock()
-		if err != nil {
-			c_log.GlobalLogger.Error("下载 OSS 上的触发器插件失败：", err)
-			continue
+		for {
+			if err := config.OssBucket.GetObjectToFile(trigger.TriggerScriptPath, triggerLocalPath); err != nil {
+				c_log.GlobalLogger.Error("下载触发器插件失败，再次尝试：", err)
+				continue
+			}
+			break
 		}
 		triggerLocalPathsMapTriggerId[triggerLocalPath] = id
 	}

trigger/pjisuv/cicv_location/Brake/main/Brake.go

@@ -21,7 +21,7 @@ func Rule(data *pjisuv_msgs.PerceptionLocalization) string {
 		}
 	}()
 	if data.AccelX*9.8 < -4.0 {
-		return "brake"
+		return "Brake"
 	} else {
 		return ""
 	}