Merge remote-tracking branch 'origin/master'

README.md

@@ -5,11 +5,8 @@
   - （新方案）采用 sync.Map，将触发器与 param 包解耦 
   - pjisuvParam 结构体不要再修改，已废弃方案
 - 添加共享变量时
-  - 如果有新的topic需要被订阅，需要在数据闭环平台上的topic序列添加
-    - 平台当前维护的序列（队列为触发器涉及的topic和共享变量涉及的topic的并集）
-      - 多功能车：/cicv_amr_trajectory,/cicv_extend,/cicv_location,/fault_info,/map_polygon,/pj_control_pub,/pj_vehicle_fdb_pub,/tpperception,/data_read
+  - 如果有新的topic需要被订阅，需要在 trigger_var.go 文件中添加
   - 需要修改produce_window.go文件中的订阅者，将触发器>0条件去掉
-- 
 
 # 一、代码目录结构
 
@@ -20,6 +17,15 @@
 ## 1-2 xxx_msgs目录
 - ros消息定义
 
+## aarch64/pjisuv
+## 目录结构
+### common
+- c_cloud.go：阿里云oss配置文件下载解析
+### control 调度程序
+### master 102 程序
+- trigger_init.go：102 初始化加载触发器插件
+### slave 103 程序
+
 # 二、oss 配置
 
 ```

aarch64/pjibot_delivery/README.md

@@ -1,33 +0,0 @@
-# 一、由于机器人由启用单个相机变成两个相机，所以话题有多变化
-1、只启用了单个相机的机器
-/amcl_pose
-/camera/color/image_raw
-/camera/depth/points
-/diagnostics
-/locate_info
-/obstacle_detection
-/odom
-/move_base/global_costmap/costmap
-/move_base/global_costmap/costmap_updates     
-/move_base/local_costmap/costmap
-/move_base/local_costmap/costmap_updates
-/scan
-/scan_map_icp_amcl_node/scan_point_transformed
-/sys_info
-# 2、启用了两个相机的机器
-/amcl_pose
-/ob_camera_01/color/image_raw（上边的摄像头）
-/ob_camera_01/depth/points（上边的摄像头）
-/ob_camera_02/color/image_raw（下边的摄像头）
-/ob_camera_02/depth/points（下边的摄像头）
-/diagnostics
-/locate_info
-/obstacle_detection（有引导任务的时候才会有）
-/odom
-/move_base/global_costmap/costmap
-/move_base/global_costmap/costmap_updates     
-/move_base/local_costmap/costmap
-/move_base/local_costmap/costmap_updates
-/scan
-/scan_map_icp_amcl_node/scan_point_transformed
-/sys_info

aarch64/pjibot_delivery/common/config/c_platform.go

@@ -37,8 +37,8 @@ type response struct {
 }
 
 var (
-	PlatformConfig  PlatformConfigStruct
-	SubscribeTopics []string
+	PlatformConfig PlatformConfigStruct
+	RecordTopics   []string
 )
 
 // InitPlatformConfig 初始化数据闭环平台的配置
@@ -55,10 +55,10 @@ func InitPlatformConfig() {
 			continue
 		}
 		if checkPlatformConfig() {
-			SubscribeTopics = strings.Split(PlatformConfig.EquipmentTopic, ",")
+			RecordTopics = strings.Split(PlatformConfig.EquipmentTopic, ",")
 			// 去掉首尾空格
-			for i, topic := range SubscribeTopics {
-				SubscribeTopics[i] = strings.TrimSpace(topic)
+			for i, topic := range RecordTopics {
+				RecordTopics[i] = strings.TrimSpace(topic)
 			}
 			break
 		}

aarch64/pjibot_delivery/common/service/rosbag_record.go

@@ -29,8 +29,13 @@ func BagRecord(nodeName string) {
 		command = append(command, "--duration=1")
 		for _, host := range config.CloudConfig.Hosts {
 			if host.Name == nodeName {
-				for _, topic := range host.Topics {
-					command = append(command, topic)
+				// 配置文件中的是node1和node2各自的所有topic，platformConfig中配置用户想缓存的topic，取交集
+				for _, topic1 := range config.RecordTopics {
+					for _, topic2 := range host.Topics {
+						if topic1 == topic2 {
+							command = append(command, topic1)
+						}
+					}
 				}
 			}
 		}

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

@@ -40,5 +40,19 @@ var (
 	// 9
 	TopicOfWheelOdom = "/wheel_odom"
 	RuleOfWheelOdom  []func(data *nav_msgs.Odometry) string
-	// todo 这里加新的topic也需要在produce_window.go加新的订阅者
+
+	// todo 这里是全量的topic，添加topic则需要同时在下面的数组添加；也需要在produce_window.go中添加新的订阅者
+	AllTopics = []string{
+		TopicOfDiagnostics,       // 1
+		TopicOfImu,               // 2
+		TopicOfLocateInfo,        // 3
+		TopicOfObstacleDetection, // 4
+		TopicOfOdom,              // 5
+		TopicOfSysInfo,           // 6
+		TopicOfRobotPose,         // 7
+		TopicOfTaskFeedbackInfo,  // 8
+		TopicOfWheelOdom,         // 9
+	}
+
+	AllTopicsNumber = len(AllTopics)
 )

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

@@ -24,11 +24,11 @@ import (
 func PrepareTimeWindowProducerQueue() {
 
 	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 {
+	subscribers := make([]*goroslib.Subscriber, masterConfig.AllTopicsNumber)
+	subscribersTimes := make([]time.Time, masterConfig.AllTopicsNumber)
+	subscribersTimeMutexes := make([]sync.Mutex, masterConfig.AllTopicsNumber)
+	subscribersMutexes := make([]sync.Mutex, masterConfig.AllTopicsNumber)
+	for i, topic := range masterConfig.AllTopics {
 		for {
 			create := false // 判断是否创建成功，用于打印日志
 			// 1

aarch64/pjibot_guide/README.md

@@ -1,33 +0,0 @@
-# 一、由于机器人由启用单个相机变成两个相机，所以话题有多变化
-1、只启用了单个相机的机器
-/amcl_pose
-/camera/color/image_raw
-/camera/depth/points
-/diagnostics
-/locate_info
-/obstacle_detection
-/odom
-/move_base/global_costmap/costmap
-/move_base/global_costmap/costmap_updates     
-/move_base/local_costmap/costmap
-/move_base/local_costmap/costmap_updates
-/scan
-/scan_map_icp_amcl_node/scan_point_transformed
-/sys_info
-# 2、启用了两个相机的机器
-/amcl_pose
-/ob_camera_01/color/image_raw（上边的摄像头）
-/ob_camera_01/depth/points（上边的摄像头）
-/ob_camera_02/color/image_raw（下边的摄像头）
-/ob_camera_02/depth/points（下边的摄像头）
-/diagnostics
-/locate_info
-/obstacle_detection（有引导任务的时候才会有）
-/odom
-/move_base/global_costmap/costmap
-/move_base/global_costmap/costmap_updates     
-/move_base/local_costmap/costmap
-/move_base/local_costmap/costmap_updates
-/scan
-/scan_map_icp_amcl_node/scan_point_transformed
-/sys_info

aarch64/pjibot_guide/common/config/c_platform.go

@@ -37,8 +37,8 @@ type response struct {
 }
 
 var (
-	PlatformConfig  PlatformConfigStruct
-	SubscribeTopics []string
+	PlatformConfig PlatformConfigStruct
+	RecordTopics   []string
 )
 
 // InitPlatformConfig 初始化数据闭环平台的配置
@@ -55,10 +55,10 @@ func InitPlatformConfig() {
 			continue
 		}
 		if checkPlatformConfig() {
-			SubscribeTopics = strings.Split(PlatformConfig.EquipmentTopic, ",")
+			RecordTopics = strings.Split(PlatformConfig.EquipmentTopic, ",")
 			// 去掉首尾空格
-			for i, topic := range SubscribeTopics {
-				SubscribeTopics[i] = strings.TrimSpace(topic)
+			for i, topic := range RecordTopics {
+				RecordTopics[i] = strings.TrimSpace(topic)
 			}
 			break
 		}

aarch64/pjibot_guide/common/service/rosbag_record.go

@@ -29,8 +29,13 @@ func BagRecord(nodeName string) {
 		command = append(command, "--duration=1")
 		for _, host := range config.CloudConfig.Hosts {
 			if host.Name == nodeName {
-				for _, topic := range host.Topics {
-					command = append(command, topic)
+				// 配置文件中的是node1和node2各自的所有topic，platformConfig中配置用户想缓存的topic，取交集
+				for _, topic1 := range config.RecordTopics {
+					for _, topic2 := range host.Topics {
+						if topic1 == topic2 {
+							command = append(command, topic1)
+						}
+					}
 				}
 			}
 		}

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

@@ -4,7 +4,7 @@ import (
 	"cicv-data-closedloop/aarch64/pjibot_guide/common/config"
 	"cicv-data-closedloop/common/config/c_log"
 	"cicv-data-closedloop/common/util"
-	pji_msgs "cicv-data-closedloop/pjibot_guide_msgs"
+	"cicv-data-closedloop/pjibot_guide_msgs"
 	"github.com/bluenviron/goroslib/v2/pkg/msgs/diagnostic_msgs"
 	"github.com/bluenviron/goroslib/v2/pkg/msgs/nav_msgs"
 	"github.com/bluenviron/goroslib/v2/pkg/msgs/sensor_msgs"
@@ -93,7 +93,7 @@ func InitTriggerConfig() {
 			log(triggerLocalPath)
 			continue
 		} else if TopicOfLocateInfo == topic2 { // 3
-			if f, ok1 := rule.(func(data *pji_msgs.LocateInfo) string); ok1 {
+			if f, ok1 := rule.(func(data *pjibot_guide_msgs.LocateInfo) string); ok1 {
 				RuleOfLocateInfo = append(RuleOfLocateInfo, f)
 				goto JudgeDone
 			}
@@ -114,7 +114,7 @@ func InitTriggerConfig() {
 			log(triggerLocalPath)
 			continue
 		} else if TopicOfSysInfo == topic2 { // 6
-			if f, ok1 := rule.(func(data *pji_msgs.SysInfo) string); ok1 {
+			if f, ok1 := rule.(func(data *pjibot_guide_msgs.SysInfo) string); ok1 {
 				RuleOfSysInfo = append(RuleOfSysInfo, f)
 				goto JudgeDone
 			}

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

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

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

@@ -19,11 +19,11 @@ import (
 func PrepareTimeWindowProducerQueue() {
 
 	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 {
+	subscribers := make([]*goroslib.Subscriber, masterConfig.AllTopicsNumber)
+	subscribersTimes := make([]time.Time, masterConfig.AllTopicsNumber)
+	subscribersTimeMutexes := make([]sync.Mutex, masterConfig.AllTopicsNumber)
+	subscribersMutexes := make([]sync.Mutex, masterConfig.AllTopicsNumber)
+	for i, topic := range masterConfig.AllTopics {
 		for {
 			create := false // 判断是否创建成功，用于打印日志
 			if topic == masterConfig.TopicOfObstacleDetection && len(masterConfig.RuleOfObstacleDetection) > 0 {

aarch64/pjibot_patrol/README.md

@@ -1,33 +0,0 @@
-# 一、由于机器人由启用单个相机变成两个相机，所以话题有多变化
-1、只启用了单个相机的机器
-/amcl_pose
-/camera/color/image_raw
-/camera/depth/points
-/diagnostics
-/locate_info
-/obstacle_detection
-/odom
-/move_base/global_costmap/costmap
-/move_base/global_costmap/costmap_updates     
-/move_base/local_costmap/costmap
-/move_base/local_costmap/costmap_updates
-/scan
-/scan_map_icp_amcl_node/scan_point_transformed
-/sys_info
-# 2、启用了两个相机的机器
-/amcl_pose
-/ob_camera_01/color/image_raw（上边的摄像头）
-/ob_camera_01/depth/points（上边的摄像头）
-/ob_camera_02/color/image_raw（下边的摄像头）
-/ob_camera_02/depth/points（下边的摄像头）
-/diagnostics
-/locate_info
-/obstacle_detection（有引导任务的时候才会有）
-/odom
-/move_base/global_costmap/costmap
-/move_base/global_costmap/costmap_updates     
-/move_base/local_costmap/costmap
-/move_base/local_costmap/costmap_updates
-/scan
-/scan_map_icp_amcl_node/scan_point_transformed
-/sys_info

aarch64/pjibot_patrol/common/config/c_platform.go

@@ -37,11 +37,11 @@ type response struct {
 }
 
 var (
-	PlatformConfig  PlatformConfigStruct
-	SubscribeTopics []string
+	PlatformConfig PlatformConfigStruct
+	RecordTopics   []string
 )
 
-// InitPlatformConfig 初始化数据闭环平台的配置
+// 初始化数据闭环平台的配置
 func InitPlatformConfig() {
 	var err error
 	c_log.GlobalLogger.Info("获取数据闭环平台配置 - 开始")
@@ -55,10 +55,10 @@ func InitPlatformConfig() {
 			continue
 		}
 		if checkPlatformConfig() {
-			SubscribeTopics = strings.Split(PlatformConfig.EquipmentTopic, ",")
+			RecordTopics = strings.Split(PlatformConfig.EquipmentTopic, ",")
 			// 去掉首尾空格
-			for i, topic := range SubscribeTopics {
-				SubscribeTopics[i] = strings.TrimSpace(topic)
+			for i, topic := range RecordTopics {
+				RecordTopics[i] = strings.TrimSpace(topic)
 			}
 			break
 		}

aarch64/pjibot_patrol/common/service/rosbag_record.go

@@ -29,8 +29,13 @@ func BagRecord(nodeName string) {
 		command = append(command, "--duration=1")
 		for _, host := range config.CloudConfig.Hosts {
 			if host.Name == nodeName {
-				for _, topic := range host.Topics {
-					command = append(command, topic)
+				// 配置文件中的是node1和node2各自的所有topic，platformConfig中配置用户想缓存的topic，取交集
+				for _, topic1 := range config.RecordTopics {
+					for _, topic2 := range host.Topics {
+						if topic1 == topic2 {
+							command = append(command, topic1)
+						}
+					}
 				}
 			}
 		}

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

@@ -39,4 +39,18 @@ var (
 	// 9
 	TopicOfWheelOdom = "/wheel_odom"
 	RuleOfWheelOdom  []func(data *nav_msgs.Odometry) string
+
+	// todo 这里是全量的topic，添加topic则需要同时在下面的数组添加；也需要在produce_window.go中添加新的订阅者
+	AllTopics = []string{
+		TopicOfDiagnostics,       // 1
+		TopicOfImu,               // 2
+		TopicOfLocateInfo,        // 3
+		TopicOfObstacleDetection, // 4
+		TopicOfOdom,              // 5
+		TopicOfSysInfo,           // 6
+		TopicOfRobotPose,         // 7
+		TopicOfTaskFeedbackInfo,  // 8
+		TopicOfWheelOdom,         // 9
+	}
+	AllTopicsNumber = len(AllTopics)
 )

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

@@ -24,11 +24,11 @@ import (
 func PrepareTimeWindowProducerQueue() {
 
 	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 {
+	subscribers := make([]*goroslib.Subscriber, masterConfig.AllTopicsNumber)
+	subscribersTimes := make([]time.Time, masterConfig.AllTopicsNumber)
+	subscribersTimeMutexes := make([]sync.Mutex, masterConfig.AllTopicsNumber)
+	subscribersMutexes := make([]sync.Mutex, masterConfig.AllTopicsNumber)
+	for i, topic := range masterConfig.AllTopics {
 		for {
 			create := false // 判断是否创建成功，用于打印日志
 			// 1

aarch64/pjisuv/README.md

@@ -1,21 +0,0 @@
-# 目录结构
-## common
-- c_cloud.go：阿里云oss配置文件下载解析
-## control 调度程序
-## master 102 程序
-- trigger_init.go：102 初始化加载触发器插件
-## slave 103 程序
-
-
-
-# 添加共享变量
-./master/service/produce_window.go
-
-## 已保存共享变量
-- 
-
-# 数据闭环平台 topic 序列
-## 临时
-/cicv_amr_trajectory,/cicv_extend,/cicv_location,/fault_info,/map_polygon,/pj_control_pub,/pj_vehicle_fdb_pub,/tpperception,/data_read,/end_point_message
-## 全部
-/amr_pose,/bounding_boxes_fast,/camera_fault,/can_data,/ch128x1/lslidar_point_cloud,/ch64w_l/lslidar_point_cloud,/ch64w_l/scan,/ch64w_r/lslidar_point_cloud,/ch64w_r/scan,/cicv/lidarcluster_moving_objects,/cicv_amr_trajectory,/cicv_location,/cloud_clusters,/heartbeat_info,/lidarPretreatment_Cost,/lidar_pretreatment/odometry,/lidar_roi,/line_1,/line_2,/map_polygon,/obstacle_display,/pj_control_pub,/points_cluster,/points_concat,/reference_display,/reference_trajectory,/roi/points,/roi/polygon,/tf,/tpperception,/tpperception/vis,/tprouteplan,/trajectory_display,/unground_cloudpoints,/camera_image,/data_read,/pji_gps,/fault_info,/pj_vehicle_fdb_pub,/end_point_message

aarch64/pjisuv/common/config/c_cloud.go

@@ -44,7 +44,7 @@ type trigger struct {
 	Topics []string `yaml:"topics"`
 }
 
-type cloudConfig struct {
+type CloudConfigStruct struct {
 	RefreshCloudConfig    bool          `yaml:"refresh-cloud-config"`
 	CompressBag           bool          `yaml:"compress-bag"`
 	CleanBeforeStart      bool          `yaml:"clean-before-start"`
@@ -67,7 +67,7 @@ type cloudConfig struct {
 }
 
 var (
-	CloudConfig      cloudConfig
+	CloudConfig      CloudConfigStruct
 	CloudConfigMutex sync.RWMutex
 )
 
@@ -109,7 +109,7 @@ func InitCloudConfig() {
 	}
 
 	// 4 ------- 解析YAML内容 -------
-	var newCloudConfig cloudConfig
+	var newCloudConfig CloudConfigStruct
 	err = yaml.Unmarshal(content, &newCloudConfig)
 	if err != nil {
 		c_log.GlobalLogger.Error("程序退出。配置文件 ", LocalConfig.CloudConfigLocalPath, " 解析失败:", err)
@@ -171,7 +171,7 @@ func refreshCloudConfig() {
 	}
 
 	// 4 ------- 解析YAML内容 -------
-	var newCloudConfig cloudConfig
+	var newCloudConfig CloudConfigStruct
 	err = yaml.Unmarshal(content, &newCloudConfig)
 	if err != nil {
 		c_log.GlobalLogger.Error("配置文件 ", LocalConfig.CloudConfigLocalPath, " 解析失败:", err)
@@ -200,7 +200,7 @@ func RefreshCloudConfig() {
 }
 
 // CheckConfig 校验 cfg.yaml 文件
-func checkConfig(check cloudConfig) bool {
+func checkConfig(check CloudConfigStruct) bool {
 	if len(check.Hosts) != 2 {
 		c_log.GlobalLogger.Error("cloud-config.yaml中配置的hosts必须为2。")
 		os.Exit(-1)

aarch64/pjisuv/common/config/c_platform.go

@@ -37,8 +37,8 @@ type response struct {
 }
 
 var (
-	PlatformConfig  PlatformConfigStruct
-	SubscribeTopics []string
+	PlatformConfig PlatformConfigStruct
+	RecordTopics   []string
 )
 
 // 初始化数据闭环平台的配置
@@ -54,10 +54,10 @@ func InitPlatformConfig() {
 			continue
 		}
 		if checkPlatformConfig() {
-			SubscribeTopics = strings.Split(PlatformConfig.EquipmentTopic, ",")
+			RecordTopics = strings.Split(PlatformConfig.EquipmentTopic, ",")
 			// 去掉首尾空格
-			for i, topic := range SubscribeTopics {
-				SubscribeTopics[i] = strings.TrimSpace(topic)
+			for i, topic := range RecordTopics {
+				RecordTopics[i] = strings.TrimSpace(topic)
 			}
 			break
 		}

aarch64/pjisuv/common/service/rosbag_record.go

@@ -10,15 +10,15 @@ import (
 
 // BagRecord 打包rosbag
 func BagRecord(nodeName string) {
-	c_log.GlobalLogger.Info("rosbag record goroutine - 启动")
+	c_log.GlobalLogger.Info("【rosbag record】 goroutine - 启动")
 	for {
-		c_log.GlobalLogger.Info("校验必需的 rosnode 是否全部启动。")
+		c_log.GlobalLogger.Info("校验必需的 ROS NODE 是否全部启动。")
 		canRecord := false
 		for !canRecord {
 			time.Sleep(time.Duration(2) * time.Second)
 			canRecord = isCanRecord(config.RosNode)
 		}
-		c_log.GlobalLogger.Info("rosnode启动完成，正在启动record命令。")
+		c_log.GlobalLogger.Info("ROS NODE 启动完成，正在启动record命令。")
 
 		var command []string
 		command = append(command, "record")
@@ -26,8 +26,13 @@ func BagRecord(nodeName string) {
 		command = append(command, "--duration=1")
 		for _, host := range config.CloudConfig.Hosts {
 			if host.Name == nodeName {
-				for _, topic := range host.Topics {
-					command = append(command, topic)
+				// 配置文件中的是node1和node2各自的所有topic，platformConfig中配置用户想缓存的topic，取交集
+				for _, topic1 := range config.RecordTopics {
+					for _, topic2 := range host.Topics {
+						if topic1 == topic2 {
+							command = append(command, topic1)
+						}
+					}
 				}
 			}
 		}

aarch64/pjisuv/common/service/rosbag_upload.go

@@ -46,13 +46,40 @@ outLoop:
 		entity.RemoveHeadOfTimeWindowConsumerQueue()
 		c_log.GlobalLogger.Infof("开始处理窗口，【Lable】=%v，【FaultTime】=%v，【Length】=%v", currentTimeWindow.Labels, currentTimeWindow.FaultTime, currentTimeWindow.Length)
 
-		// 2 获取目录
+		// 2 获取目录和bags
 		dir := domain.GetCopyDir(commonConfig.CloudConfig.BagCopyDir, currentTimeWindow.FaultTime)
 		//bags, _ := commonUtil.ListAbsolutePathWithSuffixAndSort(dir, ".bag")
-		//bagNumber := len(bags)
-		//if bagNumber > currentTimeWindow.Length {
-		//	bagNumber = currentTimeWindow.Length
-		//	bags = bags[0:currentTimeWindow.Length]
+		// 3 如果不是全量采集，则使用 filter 命令对 bag 包进行主题过滤。
+		// todo  需要采集回传的话题是根据触发器设置的，触发器设置了之后进行过滤
+		//if commonConfig.CloudConfig.FullCollect == false {
+		//	filterTopics := commonConfig.CollectTopics
+		//	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
+		//		}
+		//		// 删除旧文件
+		//		_ = commonUtil.DeleteFile(oldName)
+		//		// 将新文件改回旧文件名
+		//		if err = os.Rename(newName, oldName); err != nil {
+		//			c_log.GlobalLogger.Info("修改文件名", oldName, "失败，放弃当前时间窗口", currentTimeWindow.FaultTime, "，错误为：", err)
+		//			continue outLoop
+		//		}
+		//	}
 		//}
 		//
 		//if commonConfig.CloudConfig.CompressBag  {

aarch64/pjisuv/master/config/trigger_var.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"cicv-data-closedloop/pjisuv_msgs"
+	"cicv-data-closedloop/pjisuv_ticker"
 	"github.com/bluenviron/goroslib/v2/pkg/msgs/geometry_msgs"
 	"github.com/bluenviron/goroslib/v2/pkg/msgs/nav_msgs"
 	"github.com/bluenviron/goroslib/v2/pkg/msgs/sensor_msgs"
@@ -13,195 +14,197 @@ import (
 var (
 	LabelMapTriggerId = new(sync.Map)
 
+	// 0
 	// 定时任务触发器，详见 pjisuv_ticker包
-	RuleOfCicvTicker = make([]func(shareVars *sync.Map), 0) // tick代表定时任务间隔时间；对于长度为0的slice，无论是使用var还是make创建，它们在内存占用上的差异通常可以忽略不计
+	TopicOfCicvTicker = pjisuv_ticker.TickerTopic
+	RuleOfCicvTicker  = make([]func(shareVars *sync.Map), 0) // tick代表定时任务间隔时间；对于长度为0的slice，无论是使用var还是make创建，它们在内存占用上的差异通常可以忽略不计
 
-	//1
+	// 1
 	TopicOfAmrPose = "/amr_pose"
 	RuleOfAmrPose1 []func(data *visualization_msgs.MarkerArray) string
 	RuleOfAmrPose3 []func(shareVars *sync.Map, data *visualization_msgs.MarkerArray) string
 
-	//2
+	// 2
 	TopicOfBoundingBoxesFast = "/bounding_boxes_fast"
 	RuleOfBoundingBoxesFast1 []func(data *pjisuv_msgs.BoundingBoxArray) string
 	RuleOfBoundingBoxesFast3 []func(shareVars *sync.Map, data *pjisuv_msgs.BoundingBoxArray) string
 
-	//3
+	// 3
 	TopicOfCameraFault = "/camera_fault"
 	RuleOfCameraFault1 []func(data *pjisuv_msgs.FaultVec) string
 	RuleOfCameraFault3 []func(shareVars *sync.Map, data *pjisuv_msgs.FaultVec) string
 
-	//4
+	// 4
 	TopicOfCanData = "/can_data"
 	RuleOfCanData1 []func(data *pjisuv_msgs.Frame) string
 	RuleOfCanData3 []func(shareVars *sync.Map, data *pjisuv_msgs.Frame) string
 
-	//5
+	// 5
 	TopicOfCh128x1LslidarPointCloud = "/ch128x1/lslidar_point_cloud"
 	RuleOfCh128x1LslidarPointCloud1 []func(data *sensor_msgs.PointCloud2) string
 	RuleOfCh128x1LslidarPointCloud3 []func(shareVars *sync.Map, data *sensor_msgs.PointCloud2) string
 
-	//6
+	// 6
 	TopicOfCh64wLLslidarPointCloud = "/ch64w_l/lslidar_point_cloud"
 	RuleOfCh64wLLslidarPointCloud1 []func(data *sensor_msgs.PointCloud2) string
 	RuleOfCh64wLLslidarPointCloud3 []func(shareVars *sync.Map, data *sensor_msgs.PointCloud2) string
 
-	//7
+	// 7
 	TopicOfCh64wLScan = "/ch64w_l/scan"
 	RuleOfCh64wLScan1 []func(data *sensor_msgs.LaserScan) string
 	RuleOfCh64wLScan3 []func(shareVars *sync.Map, data *sensor_msgs.LaserScan) string
 
-	//8
+	// 8
 	TopicOfCh64wRLslidarPointCloud = "/ch64w_r/lslidar_point_cloud"
 	RuleOfCh64wRLslidarPointCloud1 []func(data *sensor_msgs.PointCloud2) string
 	RuleOfCh64wRLslidarPointCloud3 []func(shareVars *sync.Map, data *sensor_msgs.PointCloud2) string
 
-	//9
+	// 9
 	TopicOfCh64wRScan = "/ch64w_r/scan"
 	RuleOfCh64wRScan1 []func(data *sensor_msgs.LaserScan) string
 	RuleOfCh64wRScan3 []func(shareVars *sync.Map, data *sensor_msgs.LaserScan) string
 
-	//10
+	// 10
 	TopicOfCicvLidarclusterMovingObjects = "/cicv/lidarcluster_moving_objects"
 	RuleOfCicvLidarclusterMovingObjects1 []func(data *pjisuv_msgs.PerceptionCicvMovingObjects) string
 	RuleOfCicvLidarclusterMovingObjects3 []func(shareVars *sync.Map, data *pjisuv_msgs.PerceptionCicvMovingObjects) string
 
-	//11
+	// 11
 	TopicOfCicvAmrTrajectory = "/cicv_amr_trajectory"
 	RuleOfCicvAmrTrajectory1 []func(data *pjisuv_msgs.Trajectory) string
 	RuleOfCicvAmrTrajectory3 []func(shareVars *sync.Map, data *pjisuv_msgs.Trajectory) string
 
-	//12
+	// 12
 	TopicOfCicvLocation = "/cicv_location"
 	RuleOfCicvLocation1 []func(data *pjisuv_msgs.PerceptionLocalization) string
 	RuleOfCicvLocation3 []func(shareVars *sync.Map, data *pjisuv_msgs.PerceptionLocalization) string
 
-	//13
+	// 13
 	TopicOfCloudClusters = "/cloud_clusters"
 	RuleOfCloudClusters1 []func(data *pjisuv_msgs.AutowareCloudClusterArray) string
 	RuleOfCloudClusters3 []func(shareVars *sync.Map, data *pjisuv_msgs.AutowareCloudClusterArray) string
 
-	//14
+	// 14
 	TopicOfHeartbeatInfo = "/heartbeat_info"
 	RuleOfHeartbeatInfo1 []func(data *pjisuv_msgs.HeartBeatInfo) string
 	RuleOfHeartbeatInfo3 []func(shareVars *sync.Map, data *pjisuv_msgs.HeartBeatInfo) string
 
-	//15
+	// 15
 	TopicOfLidarPretreatmentCost = "/lidarPretreatment_Cost"
 	RuleOfLidarPretreatmentCost1 []func(data *geometry_msgs.Vector3Stamped) string
 	RuleOfLidarPretreatmentCost3 []func(shareVars *sync.Map, data *geometry_msgs.Vector3Stamped) string
 
-	//16
+	// 16
 	TopicOfLidarPretreatmentOdometry = "/lidar_pretreatment/odometry"
 	RuleOfLidarPretreatmentOdometry1 []func(data *nav_msgs.Odometry) string
 	RuleOfLidarPretreatmentOdometry3 []func(shareVars *sync.Map, data *nav_msgs.Odometry) string
 
-	//17
+	// 17
 	TopicOfLidarRoi = "/lidar_roi"
 	RuleOfLidarRoi1 []func(data *geometry_msgs.PolygonStamped) string
 	RuleOfLidarRoi3 []func(shareVars *sync.Map, data *geometry_msgs.PolygonStamped) string
 
-	//18
+	// 18
 	TopicOfLine1 = "/line_1"
 	RuleOfLine11 []func(data *nav_msgs.Path) string
 	RuleOfLine13 []func(shareVars *sync.Map, data *nav_msgs.Path) string
 
-	//19
+	// 19
 	TopicOfLine2 = "/line_2"
 	RuleOfLine21 []func(data *nav_msgs.Path) string
 	RuleOfLine23 []func(shareVars *sync.Map, data *nav_msgs.Path) string
 
-	//20
+	// 20
 	TopicOfMapPolygon = "/map_polygon"
 	RuleOfMapPolygon1 []func(data *pjisuv_msgs.PolygonStamped) string
 	RuleOfMapPolygon3 []func(shareVars *sync.Map, data *pjisuv_msgs.PolygonStamped) string
 
-	//21
+	// 21
 	TopicOfObstacleDisplay = "/obstacle_display"
 	RuleOfObstacleDisplay1 []func(data *visualization_msgs.MarkerArray) string
 	RuleOfObstacleDisplay3 []func(shareVars *sync.Map, data *visualization_msgs.MarkerArray) string
 
-	//22
+	// 22
 	TopicOfPjControlPub = "/pj_control_pub"
 	RuleOfPjControlPub1 []func(data *pjisuv_msgs.CommonVehicleCmd) string
 	RuleOfPjControlPub3 []func(shareVars *sync.Map, data *pjisuv_msgs.CommonVehicleCmd) string
 
-	//23
+	// 23
 	TopicOfPointsCluster = "/points_cluster"
 	RuleOfPointsCluster1 []func(data *sensor_msgs.PointCloud2) string
 	RuleOfPointsCluster3 []func(shareVars *sync.Map, data *sensor_msgs.PointCloud2) string
 
-	//24
+	// 24
 	TopicOfPointsConcat = "/points_concat"
 	RuleOfPointsConcat1 []func(data *sensor_msgs.PointCloud2) string
 	RuleOfPointsConcat3 []func(shareVars *sync.Map, data *sensor_msgs.PointCloud2) string
 
-	//25
+	// 25
 	TopicOfReferenceDisplay = "/reference_display"
 	RuleOfReferenceDisplay1 []func(data *nav_msgs.Path) string
 	RuleOfReferenceDisplay3 []func(shareVars *sync.Map, data *nav_msgs.Path) string
 
-	//26
+	// 26
 	TopicOfReferenceTrajectory = "/reference_trajectory"
 	RuleOfReferenceTrajectory1 []func(data *pjisuv_msgs.Trajectory) string
 	RuleOfReferenceTrajectory3 []func(shareVars *sync.Map, data *pjisuv_msgs.Trajectory) string
 
-	//27
+	// 27
 	TopicOfRoiPoints = "/roi/points"
 	RuleOfRoiPoints1 []func(data *sensor_msgs.PointCloud2) string
 	RuleOfRoiPoints3 []func(shareVars *sync.Map, data *sensor_msgs.PointCloud2) string
 
-	//28
+	// 28
 	TopicOfRoiPolygon = "/roi/polygon"
 	RuleOfRoiPolygon1 []func(data *nav_msgs.Path) string
 	RuleOfRoiPolygon3 []func(shareVars *sync.Map, data *nav_msgs.Path) string
 
-	//29
+	// 29
 	TopicOfTf = "/tf"
 	RuleOfTf1 []func(data *tf2_msgs.TFMessage) string
 	RuleOfTf3 []func(shareVars *sync.Map, data *tf2_msgs.TFMessage) string
 
-	//30
+	// 30
 	TopicOfTpperception = "/tpperception"
 	RuleOfTpperception1 []func(data *pjisuv_msgs.PerceptionObjects) string
 	RuleOfTpperception3 []func(shareVars *sync.Map, data *pjisuv_msgs.PerceptionObjects) string
 
-	//31
+	// 31
 	TopicOfTpperceptionVis = "/tpperception/vis"
 	RuleOfTpperceptionVis1 []func(data *visualization_msgs.MarkerArray) string
 	RuleOfTpperceptionVis3 []func(shareVars *sync.Map, data *visualization_msgs.MarkerArray) string
 
-	//32
+	// 32
 	TopicOfTprouteplan = "/tprouteplan"
 	RuleOfTprouteplan1 []func(data *pjisuv_msgs.RoutePlan) string
 	RuleOfTprouteplan3 []func(shareVars *sync.Map, data *pjisuv_msgs.RoutePlan) string
 
-	//33
+	// 33
 	TopicOfTrajectoryDisplay = "/trajectory_display"
 	RuleOfTrajectoryDisplay1 []func(data *nav_msgs.Path) string
 	RuleOfTrajectoryDisplay3 []func(shareVars *sync.Map, data *nav_msgs.Path) string
 
-	//34
+	// 34
 	TopicOfUngroundCloudpoints = "/unground_cloudpoints"
 	RuleOfUngroundCloudpoints1 []func(data *sensor_msgs.PointCloud2) string
 	RuleOfUngroundCloudpoints3 []func(shareVars *sync.Map, data *sensor_msgs.PointCloud2) string
 
-	//35
+	// 35
 	TopicOfCameraImage = "/camera_image"
 	RuleOfCameraImage1 []func(data *sensor_msgs.Image) string
 	RuleOfCameraImage3 []func(shareVars *sync.Map, data *sensor_msgs.Image) string
 
-	//36
+	// 36
 	TopicOfDataRead = "/data_read"
 	RuleOfDataRead1 []func(data *pjisuv_msgs.Retrieval) string
 	RuleOfDataRead3 []func(shareVars *sync.Map, data *pjisuv_msgs.Retrieval) string
 
-	//37
+	// 37
 	TopicOfPjiGps = "/pji_gps"
 	RuleOfPjiGps1 []func(data *pjisuv_msgs.PerceptionLocalization) string
 	RuleOfPjiGps3 []func(shareVars *sync.Map, data *pjisuv_msgs.PerceptionLocalization) string
 
-	//38
+	// 38
 	TopicOfFaultInfo = "/fault_info"
 	RuleOfFaultInfo1 []func(data *pjisuv_msgs.FaultVec) string
 	RuleOfFaultInfo3 []func(shareVars *sync.Map, data *pjisuv_msgs.FaultVec) string
@@ -215,4 +218,51 @@ var (
 	TopicOfEndPointMessage = "/end_point_message"
 	RuleOfEndPointMessage1 []func(data *geometry_msgs.Point) string
 	RuleOfEndPointMessage3 []func(shareVars *sync.Map, data *geometry_msgs.Point) string
+
+	// todo 这里是全量的topic，添加topic则需要同时在下面的数组添加；也需要在produce_window.go中添加新的订阅者
+	AllTopics = []string{
+		TopicOfCicvTicker,                    // 0
+		TopicOfAmrPose,                       // 1
+		TopicOfBoundingBoxesFast,             // 2
+		TopicOfCameraFault,                   // 3
+		TopicOfCanData,                       // 4
+		TopicOfCh128x1LslidarPointCloud,      // 5
+		TopicOfCh64wLLslidarPointCloud,       // 6
+		TopicOfCh64wLScan,                    // 7
+		TopicOfCh64wRLslidarPointCloud,       // 8
+		TopicOfCh64wRScan,                    // 9
+		TopicOfCicvLidarclusterMovingObjects, // 10
+		TopicOfCicvAmrTrajectory,             // 11
+		TopicOfCicvLocation,                  // 12
+		TopicOfCloudClusters,                 // 13
+		TopicOfHeartbeatInfo,                 // 14
+		TopicOfLidarPretreatmentCost,         // 15
+		TopicOfLidarPretreatmentOdometry,     // 16
+		TopicOfLidarRoi,                      // 17
+		TopicOfLine1,                         // 18
+		TopicOfLine2,                         // 19
+		TopicOfMapPolygon,                    // 20
+		TopicOfObstacleDisplay,               // 21
+		TopicOfPjControlPub,                  // 22
+		TopicOfPointsCluster,                 // 23
+		TopicOfPointsConcat,                  // 24
+		TopicOfReferenceDisplay,              // 25
+		TopicOfReferenceTrajectory,           // 26
+		TopicOfRoiPoints,                     // 27
+		TopicOfRoiPolygon,                    // 28
+		TopicOfTf,                            // 29
+		TopicOfTpperception,                  // 30
+		TopicOfTpperceptionVis,               // 31
+		TopicOfTprouteplan,                   // 32
+		TopicOfTrajectoryDisplay,             // 33
+		TopicOfUngroundCloudpoints,           // 34
+		TopicOfCameraImage,                   // 35
+		TopicOfDataRead,                      // 36
+		TopicOfPjiGps,                        // 37
+		TopicOfFaultInfo,                     // 38
+		TopicOfPjVehicleFdbPub,               // 39
+		TopicOfEndPointMessage,               // 40
+	}
+
+	AllTopicsNumber = len(AllTopics)
 )

aarch64/pjisuv/master/service/produce_window.go

@@ -61,11 +61,11 @@ func ProduceWindow() {
 	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 {
+	subscribers := make([]*goroslib.Subscriber, masterConfig.AllTopicsNumber)
+	subscribersTimes := make([]time.Time, masterConfig.AllTopicsNumber)
+	subscribersTimeMutexes := make([]sync.Mutex, masterConfig.AllTopicsNumber)
+	subscribersMutexes := make([]sync.Mutex, masterConfig.AllTopicsNumber)
+	for i, topic := range masterConfig.AllTopics {
 		for {
 			// 定时器，区别于订阅者
 			if topic == pjisuv_ticker.TickerTopic {

aarch64/pjisuv/common/config/yaml/通用-cloud-config.yaml → aarch64/pjisuv/多功能车-cloud-config.yaml

@@ -15,10 +15,6 @@ config-refresh-interval: 60
 disk:
   name: /dev/vdb # 磁盘名称
   used: 900000000000 # 磁盘占用阈值，单位bytes
-#bag-data-dir: /userdata/cicv-data-closedloop/data/
-#bag-copy-dir: /userdata/cicv-data-closedloop/copy/
-#time-to-label-json-path: /userdata/cicv-data-closedloop/timeToLabel.json
-#triggers-dir: /userdata/cicv-data-closedloop/triggers/
 bag-data-dir: /mnt/media/sda1/cicv-data-closedloop/data/
 bag-copy-dir: /mnt/media/sda1/cicv-data-closedloop/copy/
 time-to-label-json-path: /mnt/media/sda1/cicv-data-closedloop/timeToLabel.json
@@ -172,4 +168,4 @@ triggers:
       - /tpperception
       - /points_concat
       - /cicv_location
-      - /camera_image
+      - /camera_image

common/service/cache_monitor_queue.go

@@ -0,0 +1,29 @@
+package service
+
+import (
+	"cicv-data-closedloop/common/util"
+	"sync"
+)
+
+var (
+	MonitorQueue = make([]MonitorInfo, 0, QueueLength)
+	QueueLength  = 120
+	mu           sync.Mutex // 用于保护MonitorQueue的互斥锁
+)
+
+type MonitorInfo struct {
+	Time string
+}
+
+// CacheMonitorQueue 向MonitorQueue中添加一个新的MonitorInfo实例，如果队列已满，则替换最旧的元素
+func CacheMonitorQueue() {
+	mu.Lock()
+	defer mu.Unlock()
+	// 创建新的MonitorInfo实例，Time字段为当前时间的字符串表示
+	// 如果队列长度已经达到上限，移除最旧的元素
+	if len(MonitorQueue) >= QueueLength {
+		MonitorQueue = MonitorQueue[1:] // 移除第一个元素
+	}
+	// 将新的MonitorInfo实例添加到队列的末尾
+	MonitorQueue = append(MonitorQueue, MonitorInfo{Time: util.GetNowTimeCustom()})
+}