commit

common/util/u_http.go

@@ -0,0 +1,101 @@
+package util
+
+import (
+	"bytes"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/url"
+)
+
+func HttpGet(url string) (string, error) {
+	// 创建一个HTTP客户端
+	client := &http.Client{}
+
+	// 创建一个GET请求
+	req, err := http.NewRequest("GET", url, nil)
+	if err != nil {
+		return "", err
+	}
+
+	// 发送请求并获取响应
+	resp, err := client.Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer func(Body io.ReadCloser) {
+		err := Body.Close()
+		if err != nil {
+
+		}
+	}(resp.Body)
+
+	// 读取响应的内容
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+	return string(body), nil
+}
+func HttpPostJsonResponseString(url string, params map[string]string) (string, error) {
+	jsonData, err := json.Marshal(params)
+	if err != nil {
+		return "", err
+	}
+	resp, err := http.Post(url, "application/json", bytes.NewBuffer(jsonData))
+	if err != nil {
+		return "", err
+	}
+	defer func(Body io.ReadCloser) {
+		err := Body.Close()
+		if err != nil {
+		}
+	}(resp.Body)
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+	return string(body), nil
+}
+
+func HttpGetStringAddHeadersResponseString(baseUrl string, headers map[string]string, params map[string]string) (string, error) {
+
+	// 将参数编码到URL中
+	u, _ := url.Parse(baseUrl)
+	q := u.Query()
+	for key, value := range params {
+		q.Add(key, value)
+	}
+	u.RawQuery = q.Encode()
+
+	// 创建请求
+	req, err := http.NewRequest("GET", u.String(), nil)
+	if err != nil {
+		return "", err
+	}
+
+	// 设置请求头
+	for key, value := range headers {
+		req.Header.Set(key, value)
+	}
+
+	// 发送请求
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	if err != nil {
+		return "", err
+	}
+
+	// 读取响应体
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+	err = resp.Body.Close()
+	if err != nil {
+		return "", err
+	}
+	result := string(body)
+	//log.GlobalLogger.Info("发送http请求，请求路径=", baseUrl, "，请求头=", headers, "请求参数=", params, "请求结果=", result)
+	return result, nil
+}

pjisuv/common/config/cloud-config-v2253.yaml

@@ -0,0 +1,390 @@
+# 全量采集topic
+---
+platform:
+  url-device-auth: http://139.9.199.227:30991/device/auth
+  url-task-poll: http://139.9.199.227:30991/device/task/poll
+  url-task: http://139.9.199.227:30991/device/task
+full-collect: true
+bag-number: 120
+config-refresh-interval: 60
+disk-usage: 90
+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
+    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
+    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

pjisuv/common/config/cloud-config-v2304.yaml

@@ -0,0 +1,379 @@
+# 全量采集topic
+---
+platform:
+  url-device-auth: http://139.9.199.227:30991/device/auth
+  url-task-poll: http://139.9.199.227:30991/device/task/poll
+  url-task: http://139.9.199.227:30991/device/task
+full-collect: true
+bag-number: 120
+config-refresh-interval: 60
+disk-usage: 90
+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
+    - /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
+    topics:
+      - /AutoModeStatus
+      - /cam_objects
+      - /cam_res
+      - /car_wheel
+      - /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
+    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
+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

pjisuv/common/config/cloud-config.yaml

@@ -0,0 +1,117 @@
+---
+platform:
+  url-device-auth: http://139.9.199.227:30991/device/auth
+  url-task-poll: http://139.9.199.227:30991/device/task/poll
+  url-task: http://139.9.199.227:30991/device/task
+full-collect: false
+bag-number: 120
+config-refresh-interval: 60
+disk-usage: 90
+bag-data-dir: /mnt/media/sda1/rosbag-handle/data/
+bag-copy-dir: /mnt/media/sda1/rosbag-handle/copy/
+time-to-label-json-path: /mnt/media/sda1/rosbag-handle/timeToLabel.json
+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
+    topics:
+      - /AutoModeStatus
+      - /camera_image
+      - /cicv_location
+      - /faultinfo
+      - /heartbeat_info
+      - /jinlong_control_pub
+      - /cicv_location
+      - /data_read
+  - name: node2
+    ip: 192.168.1.103
+    topics:
+      - /points_concat
+      - /tpperception
+      - /tpperception/hmi
+triggers:
+  - label: rapidaccel
+    topics:
+      - /tpperception
+      - /points_concat
+      - /cicv_location
+      - /camera_image
+  - label: brake
+    topics:
+      - /tpperception
+      - /points_concat
+      - /cicv_location
+      - /camera_image
+  - label: EmergencyStop
+    topics:
+      - /tpperception
+      - /points_concat
+      - /cicv_location
+      - /camera_image
+  - label: AutoDLimit
+    topics:
+      - /tpperception
+      - /points_concat
+      - /cicv_location
+      - /camera_image
+  - label: lanechange
+    topics:
+      - /tpperception
+      - /points_concat
+      - /cicv_location
+      - /camera_image
+  - label: brakefault
+    topics:
+      - /tpperception
+      - /points_concat
+      - /cicv_location
+      - /camera_image
+  - label: takeover
+    topics:
+      - /tpperception
+      - /points_concat
+      - /cicv_location
+      - /camera_image
+  - label: TTC
+    topics:
+      - /tpperception
+      - /points_concat
+      - /cicv_location
+      - /camera_image

pjisuv/common/config/cloud_config.go

@@ -0,0 +1,155 @@
+package cfg
+
+import (
+	"cicv-data-closedloop/common/util"
+	"cicv-data-closedloop/kinglong/common/log"
+	"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 Host struct {
+	Name   string   `yaml:"name"`
+	Ip     string   `yaml:"ip"`
+	Topics []string `yaml:"topics"`
+}
+type Ros struct {
+	MasterAddress string   `yaml:"master-address"`
+	Nodes         []string `yaml:"nodes"`
+}
+
+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"` // 主节点向从节点发送窗口的最小时间间隔
+	DiskUsage             float64   `yaml:"disk-usage"`
+	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                 []Host    `yaml:"hosts"`
+	Ros                   Ros       `yaml:"ros"`
+	Platform              Platform  `yaml:"platform"`
+}
+
+type Trigger struct {
+	Label  string   `yaml:"label"`
+	Topics []string `yaml:"topics"`
+}
+
+var (
+	CloudConfig      cloudConfig
+	CloudConfigMutex sync.RWMutex
+)
+
+// InitCloudConfig 初始化业务配置
+func InitCloudConfig() {
+	log.GlobalLogger.Info("初始化OSS配置文件 - 开始。")
+	// 获取文件的目录
+	_ = util.CreateParentDir(LocalConfig.CloudConfigLocalPath)
+	// 3 ------- 获取 yaml 字符串 -------
+	var content []byte
+	cloudConfigObjectKey := LocalConfig.OssBasePrefix + LocalConfig.EquipmentNo + "/" + LocalConfig.CloudConfigFilename
+	err := OssBucket.GetObjectToFile(cloudConfigObjectKey, LocalConfig.CloudConfigLocalPath)
+	if err != nil {
+		log.GlobalLogger.Error("下载oss上的配置文件"+cloudConfigObjectKey+"失败。", err)
+		os.Exit(-1)
+	}
+
+	content, err = os.ReadFile(LocalConfig.CloudConfigLocalPath)
+	if err != nil {
+		log.GlobalLogger.Error("配置文件 ", LocalConfig.CloudConfigLocalPath, " 读取失败:", err)
+		os.Exit(-1)
+	}
+
+	// 4 ------- 解析YAML内容 -------
+	var newCloudConfig cloudConfig
+	err = yaml.Unmarshal(content, &newCloudConfig)
+	if err != nil {
+		log.GlobalLogger.Error("配置文件 ", LocalConfig.CloudConfigLocalPath, " 解析失败:", err)
+		os.Exit(-1)
+	}
+
+	// 5 ------- 校验 yaml -------
+	if checkConfig(newCloudConfig) {
+		CloudConfigMutex.RLock()
+		CloudConfig = newCloudConfig
+		CloudConfigMutex.RUnlock()
+	} else {
+		log.GlobalLogger.Error("配置文件格式错误：", newCloudConfig)
+		os.Exit(-1)
+	}
+	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
+	err := OssBucket.GetObjectToFile(cloudConfigObjectKey, LocalConfig.CloudConfigLocalPath)
+	if err != nil {
+		log.GlobalLogger.Error("下载oss上的配置文件"+cloudConfigObjectKey+"失败。", err)
+		return
+	}
+
+	content, err = os.ReadFile(LocalConfig.CloudConfigLocalPath)
+	if err != nil {
+		log.GlobalLogger.Error("配置文件 ", LocalConfig.CloudConfigLocalPath, " 读取失败:", err)
+		return
+	}
+
+	// 4 ------- 解析YAML内容 -------
+	var newCloudConfig cloudConfig
+	err = yaml.Unmarshal(content, &newCloudConfig)
+	if err != nil {
+		log.GlobalLogger.Error("配置文件 ", LocalConfig.CloudConfigLocalPath, " 解析失败:", err)
+		return
+	}
+
+	// 5 ------- 校验 yaml -------
+	if checkConfig(newCloudConfig) {
+		CloudConfigMutex.RLock()
+		CloudConfig = newCloudConfig
+		CloudConfigMutex.RUnlock()
+	} else {
+		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 {
+		log.GlobalLogger.Error("cloud-config.yaml中配置的hosts必须为2。")
+		os.Exit(-1)
+	}
+	return true
+}

pjisuv/common/config/kill_rpc_server_cfg.go

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

pjisuv/common/config/local-config-soc1.yaml

@@ -0,0 +1,19 @@
+node:
+  name: master
+  ip: 192.168.1.102
+# 数据闭环平台参数
+equipment-no: kinglong3
+secret-key: 8EGgsLqCer8j7VR0q12emEAcrHQ3x1jh8pX9MZIa
+# 获取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/rosbag-handle/config/cloud-config.yaml
+restart-cmd:
+  dir: "/mnt/media/sda1/rosbag-handle/"
+  name: "sh"
+  args:
+    - "start-soc1.sh"

pjisuv/common/config/local-config-soc2.yaml

@@ -0,0 +1,19 @@
+node:
+  name: slave
+  ip: 192.168.1.103
+# 数据闭环平台参数
+equipment-no: kinglong3
+secret-key: 8EGgsLqCer8j7VR0q12emEAcrHQ3x1jh8pX9MZIa
+# 获取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/rosbag-handle/config/cloud-config.yaml
+restart-cmd:
+  dir: "/mnt/media/sda1/rosbag-handle/"
+  name: "sh"
+  args:
+    - "start-soc2.sh"

pjisuv/common/config/local-config.yaml

@@ -0,0 +1,19 @@
+node:
+  name: master
+  ip: 192.168.1.102
+# 数据闭环平台参数
+equipment-no: kinglong3
+secret-key: 8EGgsLqCer8j7VR0q12emEAcrHQ3x1jh8pX9MZIa
+# 获取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/rosbag-handle/config/cloud-config.yaml
+restart-cmd:
+  dir: "/mnt/media/sda1/rosbag-handle/"
+  name: "sh"
+  args:
+    - "start.sh"

pjisuv/common/config/local_cfg.go

@@ -0,0 +1,54 @@
+package cfg
+
+import (
+	"cicv-data-closedloop/kinglong/common/log"
+	"gopkg.in/yaml.v2"
+	"os"
+)
+
+type node struct {
+	Name string `yaml:"name"`
+	Ip   string `yaml:"ip"`
+}
+
+type restartCmd struct {
+	Dir  string   `yaml:"dir"`
+	Name string   `yaml:"name"`
+	Args []string `yaml:"args"`
+}
+
+type localConfig struct {
+	Node                 node       `yaml:"node"`                    // 获取oss配置的url
+	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"`              // 当前设备的密钥
+	RestartCmd           restartCmd `yaml:"restart-cmd"`             // 重启命令
+}
+
+var (
+	LocalConfig     localConfig
+	localConfigPath = "/mnt/media/sda1/rosbag-handle/config/local-config.yaml"
+)
+
+func InitLocalConfig() {
+	log.GlobalLogger.Info("初始化本地配置文件 - 开始：", localConfigPath)
+	// 读取YAML文件内容
+	content, err := os.ReadFile(localConfigPath)
+	if err != nil {
+		log.GlobalLogger.Error("读取本地配置文件失败。", err)
+		os.Exit(-1)
+	}
+
+	// 解析YAML内容
+	err = yaml.Unmarshal(content, &LocalConfig)
+	if err != nil {
+		log.GlobalLogger.Error("解析本地配置文件失败。", err)
+		os.Exit(-1)
+	}
+
+	log.GlobalLogger.Info("初始化本地配置文件 - 成功：", LocalConfig)
+
+}

pjisuv/common/config/oss_cfg.go

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

pjisuv/common/config/platform_cfg.go

@@ -0,0 +1,170 @@
+package cfg
+
+import (
+	"cicv-data-closedloop/common/util"
+	"cicv-data-closedloop/kinglong/common/cutil"
+	"cicv-data-closedloop/kinglong/common/log"
+	"encoding/json"
+	"time"
+)
+
+type taskTrigger struct {
+	TriggerId         int    `json:"triggerId"`
+	TriggerName       string `json:"triggerName"`
+	TriggerScriptPath string `json:"triggerScriptPath"`
+	TriggerType       string `json:"triggerType"`
+}
+
+type platformConfig 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"`
+	Lru             []string      `json:"LRU"`
+	TaskTriggers    []taskTrigger `json:"taskTriggers"`
+}
+
+type response struct {
+	Data    platformConfig `json:"data"`
+	Success bool           `json:"success"`
+	Message string         `json:"message"`
+	Code    int            `json:"code"`
+	NowTime string         `json:"nowTime"`
+}
+
+var PlatformConfig platformConfig
+
+// InitPlatformConfig 初始化数据闭环平台的配置
+func InitPlatformConfig() {
+	var err error
+	log.GlobalLogger.Info("获取数据闭环平台配置 - 开始")
+	// 1 如果车辆没有配置任务，则阻塞在这里，不启动任务
+	for {
+		time.Sleep(time.Duration(1))
+		log.GlobalLogger.Info("获取数据闭环平台配置 - 进行中")
+		PlatformConfig, err = getConfig()
+		if err != nil {
+			log.GlobalLogger.Error("获取配置status失败：", err)
+			continue
+		}
+		break
+	}
+	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 := cutil.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 {
+		log.GlobalLogger.Error("解析返回结果", respJson, "失败：", err)
+		return "", nil
+	}
+
+	dataMap := respMap["data"].(map[string]interface{})
+	if err != nil {
+		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 := cutil.HttpGetStringAddHeadersResponseString(
+		CloudConfig.Platform.UrlTaskPoll,
+		map[string]string{
+			"authorization": token,
+		},
+		map[string]string{
+			"equipmentNo":  LocalConfig.EquipmentNo,
+			"taskConfigId": taskConfigId,
+		},
+	)
+
+	if err != nil {
+		log.GlobalLogger.Error("访问接口", CloudConfig.Platform.UrlTask, "失败：", err)
+		return "", err
+	}
+	respMap, err := util.JsonStringToMap(resp)
+	if err != nil {
+		log.GlobalLogger.Error("解析【返回结果1】", resp, "失败：", err)
+		return "", err
+	}
+	dataMap, ok := respMap["data"].(map[string]interface{})
+	if !ok {
+		log.GlobalLogger.Errorf("解析【返回结果.data】的类型不是(map[string]interface{})，【dataMap】=%v", dataMap)
+		return "", err
+	}
+	return dataMap["status"].(string), nil
+}
+
+func getConfig() (platformConfig, error) {
+	token, err := GetAccessToken()
+	if err != nil {
+		return platformConfig{}, err
+	}
+	// 下载插件和获取配置
+	// 2 访问配置获取接口
+	resp, err := cutil.HttpGetStringAddHeadersResponseString(
+		CloudConfig.Platform.UrlTask,
+		map[string]string{
+			"authorization": token,
+		},
+		map[string]string{
+			"equipmentNo": LocalConfig.EquipmentNo,
+		},
+	)
+	if err != nil {
+		log.GlobalLogger.Error("访问接口", CloudConfig.Platform.UrlTask, "失败：", err)
+		return platformConfig{}, err
+	}
+	var result response
+	err = json.Unmarshal([]byte(resp), &result)
+	if err != nil {
+		log.GlobalLogger.Error("解析【返回结果】", resp, "失败：", err)
+		return platformConfig{}, err
+	}
+	return result.Data, nil
+}

pjisuv/common/config/ros_cfg.go

@@ -0,0 +1,24 @@
+package cfg
+
+import (
+	"cicv-data-closedloop/kinglong/common/cutil"
+	"cicv-data-closedloop/kinglong/common/log"
+	"github.com/bluenviron/goroslib/v2"
+	"os"
+)
+
+var RosNode *goroslib.Node
+
+func InitRosConfig() {
+	log.GlobalLogger.Info("初始化RosNode - 开始")
+	var err error
+	RosNode, err = goroslib.NewNode(goroslib.NodeConf{
+		Name:          "node" + cutil.GetNowTimeCustom(),
+		MasterAddress: CloudConfig.Ros.MasterAddress,
+	})
+	if err != nil {
+		log.GlobalLogger.Error("初始化RosNode - 失败：", err)
+		os.Exit(-1)
+	}
+	log.GlobalLogger.Info("初始化RosNode - 成功：", CloudConfig.Ros.MasterAddress)
+}

pjisuv/common/entity/time_window.go

@@ -0,0 +1,13 @@
+package ent
+
+type TimeWindow struct {
+	FaultTime       string   `json:"FaultTime"`
+	TimeWindowBegin string   `json:"TimeWindowBegin"`
+	TimeWindowEnd   string   `json:"TimeWindowEnd"`
+	Labels          []string `json:"Labels"`
+	TriggerIds      []string `json:"TriggerIds"`
+	Length          int      `json:"Length"`
+	CanUpload       string   `json:"CanUpload"`
+	MasterTopics    []string `json:"MasterTopics"`
+	SlaveTopics     []string `json:"SlaveTopics"`
+}

pjisuv/common/global/global.go

@@ -0,0 +1,25 @@
+package global
+
+import (
+	"cicv-data-closedloop/kinglong/common/ent"
+	"sync"
+	"time"
+)
+
+var (
+	TimeWindowProducerQueue      []ent.TimeWindow
+	TimeWindowProducerQueueMutex sync.RWMutex
+
+	TimeWindowConsumerQueue      []ent.TimeWindow
+	TimeWindowConsumerQueueMutex sync.RWMutex
+
+	Subscriber0Time      = time.Date(2023, time.November, 23, 10, 30, 0, 0, time.UTC)
+	Subscriber0TimeMutex sync.Mutex
+	Subscriber2Time      = time.Date(2023, time.November, 23, 10, 30, 0, 0, time.UTC)
+	Subscriber2TimeMutex sync.Mutex
+	Subscriber3Time      = time.Date(2023, time.November, 23, 10, 30, 0, 0, time.UTC)
+	Subscriber3TimeMutex sync.Mutex
+
+	TcpSendTime      = time.Date(2023, time.November, 23, 10, 30, 0, 0, time.UTC)
+	TcpSendTimeMutex sync.Mutex
+)

pjisuv/common/init/common_init.go

@@ -0,0 +1,33 @@
+package init
+
+import (
+	"cicv-data-closedloop/kinglong/common/cfg"
+	"cicv-data-closedloop/kinglong/common/svc"
+)
+
+func Init() {
+
+	// 初始化本地配置文件（第1处配置，在本地文件）
+	cfg.InitLocalConfig()
+
+	// 初始化Oss连接信息
+	cfg.InitOssConfig()
+
+	// 初始化业务逻辑配置信息，配置文件在oss上（第2处配置，在oss文件）
+	cfg.InitCloudConfig()
+
+	go cfg.RefreshCloudConfig()
+
+	// 初始化数据闭环平台的配置（第3处配置，在数据闭环平台接口）
+	cfg.InitPlatformConfig()
+
+	// 初始化ros节点
+	cfg.InitRosConfig()
+
+	// 维护data目录缓存的包数量
+	go svc.BagCacheClean()
+
+	// 磁盘占用过高时根据缓存策略处理copy目录
+	go svc.DiskClean()
+
+}

pjisuv/common/log/log_cfg.go

@@ -0,0 +1,64 @@
+package log
+
+import (
+	"cicv-data-closedloop/common/util"
+	"fmt"
+	"github.com/sirupsen/logrus"
+	"os"
+	"path/filepath"
+	"runtime"
+	"time"
+)
+
+var GlobalLogger *logrus.Logger
+
+//var MonitorLogger *logrus.Logger
+
+// InitLogConfig 初始化日志配置
+func InitLogConfig(prefix string) {
+	initGlobalLogger(prefix)
+	//initMonitorLogger()
+}
+
+// initGlobalLogger 初始化日志配置
+func initGlobalLogger(prefix string) {
+	time.Sleep(time.Duration(1) * time.Second)
+	logPath := "./log/" + prefix + "-" + time.Now().Format("2006-01-02-15-04-05") + ".log"
+	err := util.CreateParentDir(logPath)
+	// 创建、追加、读写，777，所有权限
+	f, err := os.OpenFile(logPath, os.O_CREATE|os.O_APPEND|os.O_RDWR, os.ModePerm)
+	if err != nil {
+		os.Exit(-1)
+	}
+	GlobalLogger = logrus.New()
+	GlobalLogger.SetOutput(f)
+	GlobalLogger.SetReportCaller(true) // 开启行号显示
+	GlobalLogger.SetFormatter(&logrus.JSONFormatter{
+		CallerPrettyfier: func(frame *runtime.Frame) (string, string) {
+			fileName := filepath.Base(frame.File)
+			return "", fmt.Sprintf("%s:%d", fileName, frame.Line)
+		},
+	})
+	GlobalLogger.Info("初始化GlobalLogger - 成功")
+
+}
+
+//// initMonitorLogger 初始化日志配置
+//func initMonitorLogger() {
+//	time.Sleep(time.Duration(1) * time.Second)
+//	// 创建、追加、读写，777，所有权限
+//	f, err := os.OpenFile("monitor-"+time.Now().Format("2006-01-02-15-04-05")+".log", os.O_CREATE|os.O_APPEND|os.O_RDWR, os.ModePerm)
+//	if err != nil {
+//		os.Exit(-1)
+//	}
+//	MonitorLogger = logrus.New()
+//	MonitorLogger.SetOutput(f)
+//	MonitorLogger.SetReportCaller(true) // 开启行号显示
+//	MonitorLogger.SetFormatter(&logrus.JSONFormatter{
+//		CallerPrettyfier: func(frame *runtime.Frame) (string, string) {
+//			fileName := filepath.Base(frame.File)
+//			return "", fmt.Sprintf("%s:%d", fileName, frame.Line)
+//		},
+//	})
+//	MonitorLogger.Info("初始化MonitorLogger - 成功")
+//}

pjisuv/common/service/disk_clean.go

@@ -0,0 +1,89 @@
+package svc
+
+import (
+	commonConfig "cicv-data-closedloop/kinglong/common/cfg"
+	"cicv-data-closedloop/kinglong/common/global"
+	"cicv-data-closedloop/kinglong/common/log"
+	"cicv-data-closedloop/kinglong/common/util"
+	masterConfig "cicv-data-closedloop/kinglong/master/pkg/cfg"
+	"time"
+)
+
+// DiskClean 如果磁盘占用过高，则删除timeWindow和对应的文件
+func DiskClean() {
+	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 获取磁盘占用
+		percent := util.GetDiskUsagePercent()
+		if percent > commonConfig.CloudConfig.DiskUsage {
+			// 2 获取策略
+			policy := commonConfig.PlatformConfig.TaskCachePolicy
+			log.GlobalLogger.Errorf("磁盘占用超过 %v，触发删除规则 %v", commonConfig.CloudConfig.DiskUsage, policyToDescription[policy])
+			if policy == "TTL" {
+				// 1 获取时间窗口队列中的第二个
+				if len(global.TimeWindowConsumerQueue) > 2 {
+					deleteTimeWindow(1)
+				}
+			} else if policy == "STOP" {
+				// 2 获取时间窗口队列中的倒数第一个
+				if len(global.TimeWindowConsumerQueue) > 2 {
+					deleteTimeWindow(len(global.TimeWindowConsumerQueue) - 1)
+				}
+			} else if policy == "LRU" {
+				// 3 获取优先级最低的时间窗口
+				if len(global.TimeWindowConsumerQueue) > 2 {
+					indexToRemove := getIndexToRemoveForLRU()
+					if indexToRemove != -1 {
+						deleteTimeWindow(indexToRemove)
+					}
+				}
+			} else {
+				log.GlobalLogger.Error("未知的缓存策略：", policy)
+			}
+		}
+	}
+}
+
+func deleteTimeWindow(indexToRemove int) {
+	timeWindowToRemove := global.TimeWindowConsumerQueue[indexToRemove]
+	// 删除文件
+	faultTime := timeWindowToRemove.FaultTime
+	dir := util.GetCopyDir(faultTime)
+	err := util.RemoveDir(dir)
+	if err != nil {
+		log.GlobalLogger.Error("删除目录", dir, "失败：", err)
+	}
+	global.TimeWindowConsumerQueueMutex.Lock()
+	// 使用切片的特性删除指定位置的元素
+	global.TimeWindowConsumerQueue = append(global.TimeWindowConsumerQueue[:indexToRemove], global.TimeWindowConsumerQueue[indexToRemove+1:]...)
+	global.TimeWindowConsumerQueueMutex.Unlock()
+}
+
+func getIndexToRemoveForLRU() int {
+	lru := commonConfig.PlatformConfig.Lru
+	i := len(lru) - 1
+	for i >= 0 {
+		for i2, window := range global.TimeWindowConsumerQueue {
+			if masterConfig.LabelMapTriggerId[window.FaultTime] == lru[i] {
+				return i2
+			}
+		}
+	}
+	return -1
+
+}

pjisuv/common/service/kill_self.go

@@ -0,0 +1,118 @@
+package svc
+
+import (
+	"cicv-data-closedloop/kinglong/common/cfg"
+	"cicv-data-closedloop/kinglong/common/log"
+	"cicv-data-closedloop/kinglong/common/util"
+	commonConfig "cicv-data-closedloop/pji/common/cfg"
+	"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 {
+	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 {
+		log.GlobalLogger.Error("注册rpc服务失败：", err)
+		return
+	}
+
+	// 等待并处理远程调用请求
+	for {
+		conn, err := cfg.KillSignalListener.Accept()
+		if err != nil {
+			continue
+		}
+		go rpc.ServeConn(conn)
+	}
+}
+
+func AddKillTimes(info string) {
+	MutexKill.Lock()
+
+	switch info {
+	case "1":
+		close(ChannelKillRosRecord)
+		KillTimes++
+		log.GlobalLogger.Infof("已杀死record打包goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
+		ChannelKillDiskClean <- 1
+	case "2":
+		close(ChannelKillDiskClean)
+		KillTimes++
+		log.GlobalLogger.Infof("已杀死bag包数量维护goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
+	case "3":
+		close(ChannelKillWindowProducer)
+		KillTimes++
+		log.GlobalLogger.Infof("已杀死时间窗口生产者，当前自杀进度 %v / %v", KillTimes, KillChannel)
+	case "4":
+		close(ChannelKillMove)
+		KillTimes++
+		log.GlobalLogger.Infof("已杀死bag包移动goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
+	case "5":
+		close(ChannelKillConsume)
+		KillTimes++
+		log.GlobalLogger.Infof("已杀死bag包消费goroutine，当前自杀进度 %v / %v", KillTimes, KillChannel)
+	}
+	MutexKill.Unlock()
+}
+
+func killDone(restart bool) {
+	for {
+		time.Sleep(time.Duration(1) * time.Second)
+		if KillChannel == KillTimes {
+			if restart {
+				_, err := util.ExecuteWithPath(commonConfig.LocalConfig.RestartCmd.Dir, commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args...)
+				if err != nil {
+					log.GlobalLogger.Info("启动新程序失败,【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args, "：", err)
+					os.Exit(-1)
+				}
+				log.GlobalLogger.Info("数据采集任务更新，正常退出当前程序。")
+			} else {
+				log.GlobalLogger.Info("数据采集任务终止，正常退出当前程序。")
+			}
+			os.Exit(0)
+		}
+	}
+}

pjisuv/common/service/rosbag_clean.go

@@ -0,0 +1,36 @@
+package svc
+
+import (
+	"cicv-data-closedloop/kinglong/common/cfg"
+	"cicv-data-closedloop/kinglong/common/log"
+	"cicv-data-closedloop/kinglong/common/util"
+	"time"
+)
+
+// BagCacheClean 保证本地缓存的包数量不超过设定值
+func BagCacheClean() {
+	log.GlobalLogger.Info("启动清理缓存的 goroutine 维护目录【", cfg.CloudConfig.BagDataDir, "】的 bag 包数量：", cfg.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(cfg.CloudConfig.BagDataDir, ".bag")
+		// 3 如果打包数量超过n个，删除最旧的包{
+		if len(bags) > cfg.CloudConfig.BagNumber {
+			diff := len(bags) - cfg.CloudConfig.BagNumber
+			for i := 0; i < diff; i++ {
+				util.DeleteFile(bags[i])
+			}
+		}
+	}
+}

pjisuv/common/service/rosbag_record.go

@@ -0,0 +1,114 @@
+package svc
+
+import (
+	"cicv-data-closedloop/common/util"
+	cfg2 "cicv-data-closedloop/kinglong/common/cfg"
+	"cicv-data-closedloop/kinglong/common/cutil"
+	"cicv-data-closedloop/kinglong/common/log"
+	"github.com/bluenviron/goroslib/v2"
+	"os"
+	"time"
+)
+
+// BagRecord 打包rosbag
+func BagRecord(nodeName string) {
+	log.GlobalLogger.Info("rosbag record goroutine - 启动")
+	for {
+		log.GlobalLogger.Info("校验必需的 rosnode 是否全部启动。")
+		canRecord := false
+		for !canRecord {
+			time.Sleep(time.Duration(2) * time.Second)
+			canRecord = isCanRecord(cfg2.RosNode)
+		}
+		log.GlobalLogger.Info("rosnode启动完成，正在启动record命令。")
+
+		var command []string
+		command = append(command, "record")
+		command = append(command, "--split")
+		command = append(command, "--duration=1")
+		for _, host := range cfg2.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
+		cutil.CreateParentDir(cfg2.CloudConfig.BagDataDir)
+		cmd, err := util.ExecuteWithEnvAndDirAsync(os.Environ(), cfg2.CloudConfig.BagDataDir, "rosbag", command...)
+		if err != nil {
+			log.GlobalLogger.Error("执行record命令", command, "出错：", err)
+			continue
+		}
+		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 {
+				log.GlobalLogger.Info("正在等待获取进程 ", recordProcessPid, " 的子进程的pid。")
+				continue
+			}
+			if recordSubProcessPid != 0 {
+				log.GlobalLogger.Info("获取进程 ", recordProcessPid, " 的子进程的pid：", recordSubProcessPid)
+				break
+			}
+		}
+		// 等待自杀信号
+		log.GlobalLogger.Info("rosbag record goroutine - 等待自杀信号")
+		select {
+		case signal := <-ChannelKillRosRecord:
+			if signal == 1 {
+				if err = util.KillProcessByPid(recordSubProcessPid); err != nil {
+					log.GlobalLogger.Errorf("程序阻塞，杀死record命令子进程出错，【pid】=%v,【err】=%v。", recordSubProcessPid, err)
+					select {} // 此处阻塞防止record命令一直录包占满存储
+				}
+				if err = cmd.Process.Kill(); err != nil {
+					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 {
+		log.GlobalLogger.Error("获取rosnode出错：", err)
+		return false
+	}
+	myMap := nodes
+	mySlice := cfg2.CloudConfig.Ros.Nodes
+	for _, element := range mySlice {
+		if _, ok := myMap[element]; !ok {
+			log.GlobalLogger.Info("rosnode：", element, " 未启动，需等待启动后才可启动record。")
+			return false
+		}
+	}
+	return true
+}

pjisuv/common/service/rosbag_upload.go

@@ -0,0 +1,150 @@
+package svc
+
+import (
+	commonUtil "cicv-data-closedloop/common/util"
+	commonConfig "cicv-data-closedloop/kinglong/common/cfg"
+	"cicv-data-closedloop/kinglong/common/global"
+	"cicv-data-closedloop/kinglong/common/log"
+	"cicv-data-closedloop/kinglong/common/util"
+	masterConfig "cicv-data-closedloop/kinglong/master/pkg/cfg"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+)
+
+func RunTimeWindowConsumerQueue(nodeName string) {
+	log.GlobalLogger.Info("处理消费者队列goroutine - 启动")
+outLoop:
+	for { // 串行处理
+		// 收到自杀信号
+		select {
+		case signal := <-ChannelKillConsume:
+			if signal == 1 {
+				ChannelKillConsume <- 1
+				if len(global.TimeWindowConsumerQueue) == 0 {
+					AddKillTimes("5")
+					return
+				}
+			} else { //signal == 2
+				AddKillTimes("5")
+				return
+			}
+		default:
+		}
+		// 每一秒扫一次
+		time.Sleep(time.Duration(1) * time.Second)
+		waitLength := len(global.TimeWindowConsumerQueue)
+		if waitLength == 0 {
+			continue outLoop
+		}
+		log.GlobalLogger.Infof("待处理窗口个数为：%v", len(global.TimeWindowConsumerQueue))
+		// 1 获取即将处理的窗口
+		currentTimeWindow := global.TimeWindowConsumerQueue[0]
+		util.RemoveHeaOfdTimeWindowConsumerQueue()
+		log.GlobalLogger.Infof("开始处理窗口，【Lable】=%v，【TriggerIds】=%v，【FaultTime】=%v，【Length】=%v", currentTimeWindow.Labels, currentTimeWindow.TriggerIds, currentTimeWindow.FaultTime, currentTimeWindow.Length)
+
+		// 2 获取目录
+		dir := util.GetCopyDir(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(os.Environ(), "rosbag", filterCommand...)
+				log.GlobalLogger.Info("正在过滤中，【FaultTime】=", currentTimeWindow.FaultTime, "【Label】=", currentTimeWindow.Labels, "，进度", i+1, "/", bagNumber, "。")
+				if err != nil {
+					log.GlobalLogger.Errorf("filter 命令执行出错【命令】=%v，【输出】=%v,【err】=%v", filterCommand, output, err)
+					continue
+				}
+				// 删除旧文件
+				util.DeleteFile(oldName)
+				// 将新文件改回旧文件名
+				if err = os.Rename(newName, oldName); err != nil {
+					log.GlobalLogger.Info("修改文件名", oldName, "失败，放弃当前时间窗口", currentTimeWindow.FaultTime, "，错误为：", err)
+					continue outLoop
+				}
+			}
+		}
+
+		// 4 compress包，必须顺序执行，此时每个包会对应生成一个压缩过的包和原始包，原始包后缀为.orig.bag
+		log.GlobalLogger.Info("压缩bag数据包，故障时间为：", currentTimeWindow.FaultTime)
+		for i, bag := range bags {
+			oldName := bag
+			compressCommand := []string{"compress", "--bz2", oldName}
+			log.GlobalLogger.Info("正在压缩中，【FaultTime】=", currentTimeWindow.FaultTime, "【Label】=", currentTimeWindow.Labels, "，进度", i+1, "/", bagNumber, "。")
+			if _, output, err := commonUtil.ExecuteWithEnvSync(os.Environ(), "rosbag", compressCommand...); err != nil {
+				log.GlobalLogger.Errorf("compress 命令执行出错【命令】=%v，【输出】=%v,【err】=%v", compressCommand, output, err)
+				continue
+			}
+		}
+		// 5 upload，必须顺序执行
+		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 {
+			bagSlice := strings.Split(bag, "/")
+			log.GlobalLogger.Info("正在上传中，【FaultTime】=", currentTimeWindow.FaultTime, "【Label】=", currentTimeWindow.Labels, "，进度", i+1, "/", bagNumber, "。【", bag, "】-------【", objectKey1+bagSlice[len(bagSlice)-1], "】")
+			err := commonConfig.OssBucket.PutObjectFromFile(objectKey1+bagSlice[len(bagSlice)-1], bag)
+			if err != nil {
+				log.GlobalLogger.Info("上传包 ", bag, " 时报错：", err)
+				continue
+			}
+		}
+		log.GlobalLogger.Info("上传完成，花费时间：", time.Since(start))
+		// 在上传完成的包目录同级下添加一个目录同名的json
+		triggerIds := make([]string, 0)
+		for _, label := range currentTimeWindow.Labels {
+			triggerIdToAppend := masterConfig.LabelMapTriggerId[label]
+			log.GlobalLogger.Info("添加一个【triggerId】=", triggerIdToAppend)
+			triggerIds = append(triggerIds, triggerIdToAppend)
+		}
+		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)
+		log.GlobalLogger.Info("【callBackJson】=", callBackJson)
+		if err != nil {
+			log.GlobalLogger.Error("callBackMap", callBackMap, "转 json 失败：", err)
+		}
+		err = commonConfig.OssBucket.PutObject(objectKey3+"callback.json", strings.NewReader(callBackJson))
+		if err != nil {
+			log.GlobalLogger.Error("上传 callback.json", callBackJson, "失败：", err)
+		}
+
+		// 删除本地所有已上传的bag文件
+		log.GlobalLogger.Infof("结束处理窗口，【Label】=%v，【TriggerIds】=%v，【FaultTime】=%v，【Length】=%v", currentTimeWindow.Labels, currentTimeWindow.TriggerIds, currentTimeWindow.FaultTime, currentTimeWindow.Length)
+		if err = util.RemoveDir(dir); err != nil {
+			continue outLoop
+		}
+
+	}
+}

pjisuv/common/util/move_bag.go

@@ -0,0 +1,17 @@
+package util
+
+import (
+	commonConfig "cicv-data-closedloop/kinglong/common/cfg"
+	"cicv-data-closedloop/kinglong/common/cutil"
+	"strings"
+)
+
+func MoveFromDataToCopy(faultTime string, sourceBag string) {
+	dir := GetCopyDir(faultTime)
+	cutil.CreateDir(dir)
+	targetBag := strings.Replace(sourceBag, commonConfig.CloudConfig.BagDataDir, dir, 1)
+	var copyCommand []string
+	copyCommand = append(copyCommand, sourceBag)
+	copyCommand = append(copyCommand, targetBag)
+	Execute("mv", copyCommand...)
+}

pjisuv/common/util/parse_json.go

@@ -0,0 +1,33 @@
+package util
+
+import (
+	"cicv-data-closedloop/kinglong/common/ent"
+	"cicv-data-closedloop/kinglong/common/log"
+	"encoding/json"
+)
+
+func JsonStringToMap(source string) (map[string]interface{}, error) {
+	var dataMap map[string]interface{}
+	err := json.Unmarshal([]byte(source), &dataMap)
+	if err != nil {
+		return nil, err
+	} else {
+		return dataMap, nil
+	}
+}
+
+func MapToJsonString(inputMap map[string]interface{}) (string, error) {
+	jsonBytes, err := json.Marshal(inputMap)
+	if err != nil {
+		return "", err
+	}
+	return string(jsonBytes), nil
+}
+
+func TimeWindowToJson(msg ent.TimeWindow) string {
+	jsonData, err := json.Marshal(msg)
+	if err != nil {
+		log.GlobalLogger.Error("timeWindow", msg, "转换为json时出错:", err)
+	}
+	return string(jsonData)
+}

pjisuv/common/util/send_tcp.go

@@ -0,0 +1,16 @@
+package util
+
+import (
+	"cicv-data-closedloop/kinglong/common/log"
+	"encoding/json"
+	"net"
+)
+
+func SendJsonByTcp(conn net.Conn, data string) {
+	if err := json.NewEncoder(conn).Encode(data); err != nil {
+		log.GlobalLogger.Error("发送tcp数据", data, "失败：", err)
+		return
+	} else {
+		log.GlobalLogger.Info("发送tcp数据", data, "成功：", err)
+	}
+}

pjisuv/common/util/util_exec.go

@@ -0,0 +1,34 @@
+package util
+
+import (
+	"os/exec"
+)
+
+func Execute(name string, arg ...string) (*exec.Cmd, string, error) {
+	cmd := exec.Command(name, arg...)
+	combinedOutput, err := cmd.CombinedOutput()
+	if err != nil {
+		return nil, "", err
+	}
+	return cmd, string(combinedOutput), nil
+}
+
+func ExecuteWithPath(path string, name string, arg ...string) (*exec.Cmd, error) {
+	cmd := exec.Command(name, arg...)
+	cmd.Dir = path
+	err := cmd.Start()
+	if err != nil {
+		return nil, err
+	} else {
+		return cmd, nil
+	}
+}
+
+func KillProcessByPID(pid string) error {
+	cmd := exec.Command("kill", "-9", pid)
+	err := cmd.Run()
+	if err != nil {
+		return err
+	}
+	return nil
+}

pjisuv/common/util/util_io.go

@@ -0,0 +1,67 @@
+package util
+
+import (
+	"cicv-data-closedloop/kinglong/common/log"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+)
+
+func ListAbsolutePathWithSuffixAndSort(dir string, suffix string) []string {
+	var result []string
+	if !strings.HasSuffix(dir, "/") {
+		dir = dir + "/"
+	}
+	files, err := os.ReadDir(dir)
+	if err != nil {
+		log.GlobalLogger.Error("读取目录", dir, "报错：", err)
+	}
+	for _, file := range files {
+		if strings.HasSuffix(file.Name(), suffix) {
+			result = append(result, dir+file.Name())
+		}
+	}
+	// 根据文件名进行升序排序
+	sort.Slice(result, func(i, j int) bool {
+		return filepath.Base(result[i]) < filepath.Base(result[j])
+	})
+	return result
+}
+
+// RemoveDir 递归删除目录及其下的所有文件和子目录
+func RemoveDir(dirPath string) error {
+	// 打开目录
+	dir, err := os.Open(dirPath)
+	if err != nil {
+		return err
+	}
+	defer dir.Close()
+	// 读取目录下的文件和子目录
+	fileInfos, err := dir.Readdir(-1)
+	if err != nil {
+		return err
+	}
+	// 遍历文件和子目录
+	for _, fileInfo := range fileInfos {
+		path := filepath.Join(dirPath, fileInfo.Name())
+
+		if fileInfo.IsDir() {
+			// 如果是子目录，递归调用removeDir删除子目录及其下的文件和子目录
+			RemoveDir(path)
+
+		} else {
+			// 如果是文件，直接删除文件
+			err = os.Remove(path)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	// 删除目录本身
+	err = os.Remove(dirPath)
+	if err != nil {
+		return err
+	}
+	return nil
+}

pjisuv/common/util/utils.go

@@ -0,0 +1,232 @@
+package util
+
+import (
+	commonConfig "cicv-data-closedloop/kinglong/common/cfg"
+	"cicv-data-closedloop/kinglong/common/ent"
+	"cicv-data-closedloop/kinglong/common/global"
+	"cicv-data-closedloop/kinglong/common/log"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// AppendIfNotExists 向切片中追加元素，如果元素已存在则不添加
+func AppendIfNotExists(slice []string, element string) []string {
+	for _, item := range slice {
+		if item == element {
+			return slice // 元素已存在，直接返回原切片
+		}
+	}
+	return append(slice, element) // 元素不存在，追加到切片末尾
+}
+
+func AddTimeWindowToTimeWindowProducerQueue(window ent.TimeWindow) {
+	global.TimeWindowProducerQueueMutex.RLock()
+	{
+		global.TimeWindowProducerQueue = append(global.TimeWindowProducerQueue, window)
+	}
+	global.TimeWindowProducerQueueMutex.RUnlock()
+}
+
+func AddTimeWindowToTimeWindowConsumerQueue(window ent.TimeWindow) {
+	global.TimeWindowConsumerQueueMutex.RLock()
+	{
+		global.TimeWindowConsumerQueue = append(global.TimeWindowConsumerQueue, window)
+	}
+	global.TimeWindowConsumerQueueMutex.RUnlock()
+}
+
+func RemoveHeadOfdTimeWindowProducerQueue() {
+	global.TimeWindowProducerQueueMutex.RLock()
+	{
+		global.TimeWindowProducerQueue = global.TimeWindowProducerQueue[1:]
+	}
+	global.TimeWindowProducerQueueMutex.RUnlock()
+}
+
+func RemoveHeaOfdTimeWindowConsumerQueue() {
+	global.TimeWindowConsumerQueueMutex.RLock()
+	{
+		global.TimeWindowConsumerQueue = global.TimeWindowConsumerQueue[1:]
+	}
+	global.TimeWindowConsumerQueueMutex.RUnlock()
+}
+
+func GetBagTime(bagName string) string {
+	s1 := strings.Split(bagName, "_")[0]
+	s1Split := strings.Split(s1, "/")
+	s2 := s1Split[len(s1Split)-1]
+	return s2
+}
+
+func GetNowTimeCustom() string {
+	currentTime := time.Now()
+	formattedTime := currentTime.Format("2006-01-02-15-04-05")
+	return formattedTime
+}
+
+func TimeCustomChange(originalTimeStr string, number int) string {
+	var newTimeStr string
+	// 解析时间字符串
+	layout := "2006-01-02-15-04-05"
+	originalTime, err := time.Parse(layout, originalTimeStr)
+	if err != nil {
+		log.GlobalLogger.Info("无法解析时间字符串:", err)
+		return newTimeStr
+	}
+
+	// 减少1秒
+	newTime := originalTime.Add(time.Duration(number) * time.Second)
+
+	// 格式化新的时间为指定字符串格式
+	return newTime.Format(layout)
+}
+
+func CalculateDifferenceOfTimeCustom(timeCustom1 string, timeCustom2 string) int {
+	timeInt1, _ := strconv.Atoi(strings.Replace(timeCustom1, "-", "", -1))
+	timeInt2, _ := strconv.Atoi(strings.Replace(timeCustom2, "-", "", -1))
+	return timeInt2 - timeInt1 + 1
+
+}
+func TimeCustom1GreaterTimeCustom2(timeCustom1 string, timeCustom2 string) bool {
+	timeInt1, _ := strconv.Atoi(strings.Replace(timeCustom1, "-", "", -1))
+	timeInt2, _ := strconv.Atoi(strings.Replace(timeCustom2, "-", "", -1))
+	return timeInt1 > timeInt2
+}
+
+func TimeCustom1GreaterEqualThanTimeCustom2(timeCustom1 string, timeCustom2 string) bool {
+	timeInt1, _ := strconv.Atoi(strings.Replace(timeCustom1, "-", "", -1))
+	timeInt2, _ := strconv.Atoi(strings.Replace(timeCustom2, "-", "", -1))
+	return timeInt1 >= timeInt2
+}
+
+func TimeCustom1LessEqualThanTimeCustom2(timeCustom1 string, timeCustom2 string) bool {
+	timeInt1, _ := strconv.Atoi(strings.Replace(timeCustom1, "-", "", -1))
+	timeInt2, _ := strconv.Atoi(strings.Replace(timeCustom2, "-", "", -1))
+	return timeInt1 <= timeInt2
+}
+
+// GetDiskUsagePercent 获取磁盘使用率
+func GetDiskUsagePercent() float64 {
+	// 执行 df 命令获取磁盘使用情况
+	cmd := exec.Command("df", "--total")
+	output, err := cmd.Output()
+	if err != nil {
+		log.GlobalLogger.Info("执行命令失败:", err)
+		return 0.0
+	}
+
+	// 解析 df 命令输出，计算磁盘占比
+	lines := strings.Split(string(output), "\n")
+	for _, line := range lines[1:] {
+		fields := strings.Fields(line)
+		if len(fields) >= 6 && fields[0] == "total" {
+			//filesystem := fields[0]
+			total, _ := strconv.ParseFloat(strings.TrimSuffix(fields[1], "G"), 64)
+			used, _ := strconv.ParseFloat(strings.TrimSuffix(fields[2], "G"), 64)
+			usedPercent := (used / total) * 100
+
+			//fmt.Printf("文件系统 %s 已使用 %.2f%%\n", filesystem, usedPercent)
+			return usedPercent
+		}
+	}
+	return 0.0
+}
+
+func ListAbsolutePathAndSort(dir string) []string {
+	var result []string
+	if !strings.HasSuffix(dir, "/") {
+		dir = dir + "/"
+	}
+	files, err := os.ReadDir(dir)
+	if err != nil {
+		log.GlobalLogger.Info("获取文件列表失败：", err)
+		return result
+	}
+	for _, file := range files {
+		result = append(result, dir+file.Name())
+	}
+
+	// 根据文件名进行升序排序
+	sort.Slice(result, func(i, j int) bool {
+		return filepath.Base(result[i]) < filepath.Base(result[j])
+	})
+	return result
+}
+
+func GetCopyDir(faultTime string) string {
+	return commonConfig.CloudConfig.BagCopyDir + faultTime + "/"
+}
+
+func MergeSlice(slice1 []string, slice2 []string) []string {
+
+	// 遍历第二个切片中的元素，并去重追加到结果切片1中
+	for _, element := range slice2 {
+		found := false
+		for _, item := range slice1 {
+			if element == item {
+				found = true
+				break
+			}
+		}
+		if !found {
+			slice1 = append(slice1, element)
+		}
+	}
+	return slice1
+}
+
+func DeleteFile(path string) {
+	// 检查文件是否存在
+	if _, err := os.Stat(path); err == nil {
+		// 文件存在，执行删除操作
+		err := os.Remove(path)
+		if err != nil {
+			fmt.Printf("删除文件时发生错误：%s\n", err)
+			return
+		}
+	}
+}
+
+// GetLastTimeWindow 获取最后一个时间窗口
+func GetLastTimeWindow() *ent.TimeWindow {
+	var lastTimeWindow *ent.TimeWindow // 获取最后一个时间窗口
+	if len(global.TimeWindowProducerQueue) > 0 {
+		lastTimeWindow = &global.TimeWindowProducerQueue[len(global.TimeWindowProducerQueue)-1]
+	}
+	return lastTimeWindow
+}
+
+func RefreshTcpSendTime() {
+	global.TcpSendTimeMutex.Lock()
+	global.TcpSendTime = time.Now()
+	global.TcpSendTimeMutex.Unlock()
+}
+
+// SupplyCopyBags 如果 Copy目录下的包不够，则补充一些
+func SupplyCopyBags(currentTimeWindow ent.TimeWindow) {
+	// 如果bag包没有达到length，补充几个
+	copyBags := ListAbsolutePathWithSuffixAndSort(GetCopyDir(currentTimeWindow.FaultTime), ".bag")
+	copyBagsLength := len(copyBags)
+	if copyBagsLength < currentTimeWindow.Length {
+		time.Sleep(time.Duration(copyBagsLength) * time.Second)
+		dataBags := ListAbsolutePathWithSuffixAndSort(commonConfig.CloudConfig.BagDataDir, ".bag")
+		gap := currentTimeWindow.Length - copyBagsLength
+		log.GlobalLogger.Info("故障 ", currentTimeWindow.FaultTime, "需要补充 ", gap, " 个 bag 包")
+		for _, bag := range dataBags {
+			bagTime := GetBagTime(bag)
+			if TimeCustom1GreaterEqualThanTimeCustom2(bagTime, currentTimeWindow.FaultTime) {
+				MoveFromDataToCopy(currentTimeWindow.FaultTime, bag)
+				gap = gap - 1
+				if gap == 0 {
+					break
+				}
+			}
+		}
+	}
+}

pjisuv/control/main/control.go

@@ -0,0 +1,92 @@
+package main
+
+import (
+	"cicv-data-closedloop/common/util"
+	"cicv-data-closedloop/kinglong/common/cfg"
+	"cicv-data-closedloop/kinglong/common/log"
+	commonService "cicv-data-closedloop/kinglong/common/svc"
+	"net/rpc"
+	"os"
+	"runtime"
+	"time"
+)
+
+func init() {
+	runtime.GOMAXPROCS(1)
+	// 初始化日志配置
+	log.InitLogConfig("kinglong-control")
+	// 初始化本地配置文件（第1处配置，在本地文件）
+	cfg.InitLocalConfig()
+	// 初始化Oss连接信息
+	cfg.InitOssConfig()
+	// 初始化业务逻辑配置信息，配置文件在oss上（第2处配置，在oss文件）
+	cfg.InitCloudConfig()
+}
+
+func main() {
+	lastStatus := "NONE"
+	//  轮询任务接口判断是否有更新
+	for {
+		time.Sleep(time.Duration(2) * time.Second)
+		// 1 获取当前设备的任务的 status
+		status, err := cfg.GetStatus(cfg.PlatformConfig.TaskConfigId)
+		if err != nil {
+			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 == "NONE" && status == "NONE" {
+				continue
+			}
+			// 3 发送rpc信号杀死两个服务，并重启程序
+			if lastStatus == "NONE" && status == "CHANGE" {
+				if _, err := util.ExecuteWithPath(cfg.LocalConfig.RestartCmd.Dir, cfg.LocalConfig.RestartCmd.Name, cfg.LocalConfig.RestartCmd.Args...); err != nil {
+					log.GlobalLogger.Info("启动新程序失败,【path】=", cfg.LocalConfig.RestartCmd.Dir, "【cmd】=", cfg.LocalConfig.RestartCmd.Name, cfg.LocalConfig.RestartCmd.Args, "：", err)
+					os.Exit(-1)
+				}
+				log.GlobalLogger.Info("启动任务，本地执行启动命令：【path】=", cfg.LocalConfig.RestartCmd.Dir, "【cmd】=", cfg.LocalConfig.RestartCmd.Name, cfg.LocalConfig.RestartCmd.Args)
+				lastStatus = status
+				log.GlobalLogger.Info("获取数据闭环平台最新配置。")
+				cfg.InitPlatformConfig()
+				continue
+			}
+			var killArgs *commonService.KillSignal
+			if lastStatus == "UN_CHANGE" && status == "CHANGE" {
+				killArgs = &commonService.KillSignal{NodeName: cfg.LocalConfig.Node.Name, DropUploadData: cfg.PlatformConfig.DropUploadData, Restart: true}
+				log.GlobalLogger.Info("更新任务，发送rpc重启信号到本地"+cfg.LocalConfig.Node.Name+"：", killArgs)
+			}
+			if lastStatus == "UN_CHANGE" && status == "NONE" {
+				killArgs = &commonService.KillSignal{NodeName: cfg.LocalConfig.Node.Name, DropUploadData: cfg.PlatformConfig.DropUploadData, Restart: false}
+				log.GlobalLogger.Info("杀死任务，发送rpc结束信号到本地"+cfg.LocalConfig.Node.Name+"：", killArgs)
+			}
+
+			KillRpcClient, err := rpc.Dial("tcp", cfg.LocalConfig.Node.Ip+":"+cfg.CloudConfig.RpcPort)
+			if err != nil {
+				log.GlobalLogger.Error("创建rpc客户端连接master失败：", err)
+				// 此处关闭client会报错
+				continue
+			}
+
+			reply := 0
+			if err = KillRpcClient.Call("KillService.Kill", killArgs, &reply); err != nil {
+				log.GlobalLogger.Error("发送rpc请求到master失败：", err)
+				//TODO 这里可能会报错 unexpected EOF 但是不影响，先注释 close 和 continue
+				//KillRpcClient.Close()
+				//continue
+			}
+			lastStatus = status
+			log.GlobalLogger.Info("获取数据闭环平台最新配置。")
+			cfg.InitPlatformConfig()
+			KillRpcClient.Close()
+		} else {
+			log.GlobalLogger.Error("未知的采集任务状态。【status】=", status)
+		}
+	}
+}

pjisuv/master/main/master.go

@@ -0,0 +1,40 @@
+package main
+
+import (
+	commonConfig "cicv-data-closedloop/kinglong/common/cfg"
+	commonInit "cicv-data-closedloop/kinglong/common/init"
+	"cicv-data-closedloop/kinglong/common/log"
+	commonService "cicv-data-closedloop/kinglong/common/svc"
+	masterConfig "cicv-data-closedloop/kinglong/master/pkg/cfg"
+	masterService "cicv-data-closedloop/kinglong/master/pkg/svc"
+)
+
+func init() {
+	// 初始化日志配置
+	//runtime.GOMAXPROCS(1)
+	log.InitLogConfig("kinglong-master")
+	commonInit.Init()
+	// 初始化TCP连接，用于发送时间窗口到从节点
+	masterConfig.InitTcpConfig()
+	// 初始化加载触发器插件文件
+	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 {}
+}

pjisuv/master/pkg/cfg/master_tcp_cfg.go

@@ -0,0 +1,28 @@
+package cfg
+
+import (
+	"cicv-data-closedloop/kinglong/common/log"
+	"net"
+)
+
+var TcpConnection net.Conn
+
+// InitTcpConfig 初始化TCP连接，用于发送时间窗口到从节点
+func InitTcpConfig() {
+	log.GlobalLogger.Info("主节点初始化TCP长连接 - 开始。")
+	//var err error
+	//socket := commonConfig.CloudConfig.Hosts[1].Ip + ":" + commonConfig.CloudConfig.TcpPort
+	//for {
+	//	time.Sleep(time.Duration(1) * time.Second)
+	//	TcpConnection, err = net.Dial("tcp", socket)
+	//	if err != nil {
+	//		log.GlobalLogger.Error("主节点初始化TCP长连接 - 进行中：", err)
+	//		continue
+	//	} else {
+	//		log.GlobalLogger.Info("主节点初始化TCP长连接 - 成功：", socket)
+	//		break
+	//	}
+	//
+	//}
+
+}

pjisuv/master/pkg/cfg/master_trigger_cfg.go

@@ -0,0 +1,116 @@
+package cfg
+
+import (
+	"cicv-data-closedloop/kinglong/common/cfg"
+	"cicv-data-closedloop/kinglong/common/cutil"
+	"cicv-data-closedloop/kinglong/common/log"
+	"cicv-data-closedloop/kinglong_msgs"
+	"plugin"
+	"strconv"
+	"strings"
+)
+
+var (
+	TopicOfNodeFaultInfo = "/nodefault_info"
+	TopicOfCicvLocation  = "/cicv_location"
+	TopicOfTpperception  = "/tpperception"
+	TopicOfFaultInfo     = "/fault_info"
+	TopicOfDataRead      = "/data_read"
+	RuleOfNodefaultInfo  []func(data *kinglong_msgs.FaultInfo) string
+	RuleOfCicvLocation   []func(data *kinglong_msgs.PerceptionLocalization) string
+	RuleOfTpperception   []func(data *kinglong_msgs.PerceptionObjects, velocityX float64, velocityY float64, yaw float64) string
+	RuleOfFaultInfo      []func(data *kinglong_msgs.FaultVec) string
+	RuleOfDataRead       []func(data *kinglong_msgs.Retrieval) string
+	LabelMapTriggerId    = make(map[string]string)
+)
+
+func InitTriggerConfig() {
+	log.GlobalLogger.Info("主节点加载触发器插件 - 开始。")
+	// 下载所有触发器的文件
+	for _, trigger := range cfg.PlatformConfig.TaskTriggers {
+		// 获取文件名
+		pathSplit := strings.Split(trigger.TriggerScriptPath, "/")
+		fileName := pathSplit[len(pathSplit)-1]
+		// 下载
+		triggerLocalPath := cfg.CloudConfig.TriggersDir + fileName
+		cutil.CreateParentDir(triggerLocalPath)
+		log.GlobalLogger.Info("下载触发器插件从 ", trigger.TriggerScriptPath, " 到 ", triggerLocalPath)
+		err := cfg.OssBucket.GetObjectToFile(trigger.TriggerScriptPath, triggerLocalPath)
+		if err != nil {
+			log.GlobalLogger.Error("下载 oss 上的触发器插件失败：", err)
+			continue
+		}
+		// 载入插件到数组
+		open, err := plugin.Open(triggerLocalPath)
+		if err != nil {
+			log.GlobalLogger.Error("加载本地插件", triggerLocalPath, "失败。", err)
+			continue
+		}
+		topic0, err := open.Lookup("Topic")
+		if err != nil {
+			log.GlobalLogger.Error("加载本地插件", triggerLocalPath, "中的Topic方法失败。", err)
+			continue
+		}
+		topic1, ok := topic0.(func() string)
+		if ok != true {
+			log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func() string):", err)
+			continue
+		}
+		topic2 := topic1()
+		rule, err := open.Lookup("Rule")
+		if err != nil {
+			log.GlobalLogger.Error("加载本地插件", triggerLocalPath, "中的Rule方法失败。", err)
+			continue
+		}
+		if TopicOfNodeFaultInfo == topic2 {
+			f, ok := rule.(func(data *kinglong_msgs.FaultInfo) string)
+			if ok != true {
+				log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func(data *kinglong_msgs.FaultInfo) string):", err)
+				continue
+			}
+			RuleOfNodefaultInfo = append(RuleOfNodefaultInfo, f)
+		} else if TopicOfCicvLocation == topic2 {
+			f, ok := rule.(func(data *kinglong_msgs.PerceptionLocalization) string)
+			if ok != true {
+				log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func(data *kinglong_msgs.PerceptionLocalization) string):", err)
+				continue
+			}
+			RuleOfCicvLocation = append(RuleOfCicvLocation, f)
+		} else if TopicOfTpperception == topic2 {
+			f, ok := rule.(func(data *kinglong_msgs.PerceptionObjects, velocityX float64, velocityY float64, yaw float64) string)
+			if ok != true {
+				log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func(data *kinglong_msgs.PerceptionObjects, velocityX float64, velocityY float64, yaw float64) string):", err)
+				continue
+			}
+			RuleOfTpperception = append(RuleOfTpperception, f)
+		} else if TopicOfFaultInfo == topic2 {
+			f, ok := rule.(func(data *kinglong_msgs.FaultVec) string)
+			if ok != true {
+				log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func(data *kinglong_msgs.FaultVec) string):", err)
+				continue
+			}
+			RuleOfFaultInfo = append(RuleOfFaultInfo, f)
+		} else if TopicOfDataRead == topic2 {
+			f, ok := rule.(func(data *kinglong_msgs.Retrieval) string)
+			if ok != true {
+				log.GlobalLogger.Error("插件", triggerLocalPath, "中的Topic方法必须是(func(data *kinglong_msgs.Retrieval) string):", err)
+				continue
+			}
+			RuleOfDataRead = append(RuleOfDataRead, f)
+		} else {
+			log.GlobalLogger.Error("未知的topic：", topic2)
+			continue
+		}
+
+		label, err := open.Lookup("Label")
+		if err != nil {
+			log.GlobalLogger.Error("加载本地插件 ", triggerLocalPath, " 中的 Label 方法失败。", err)
+			continue
+		}
+		labelFunc := label.(func() string)
+		labelString := labelFunc()
+		LabelMapTriggerId[labelString] = strconv.Itoa(trigger.TriggerId)
+		log.GlobalLogger.Info("主节点加载触发器插件：【ros topic】=", topic2, "，【触发器label】=", labelString, "【触发器ID】=", trigger.TriggerId, "【label和id映射关系】=", LabelMapTriggerId)
+	}
+	log.GlobalLogger.Info("主节点加载触发器插件 - 成功。")
+}

pjisuv/master/pkg/svc/move_bag_and_send_window.go

@@ -0,0 +1,117 @@
+package svc
+
+import (
+	commonUtil "cicv-data-closedloop/common/util"
+	commonConfig "cicv-data-closedloop/kinglong/common/cfg"
+	"cicv-data-closedloop/kinglong/common/ent"
+	"cicv-data-closedloop/kinglong/common/global"
+	"cicv-data-closedloop/kinglong/common/log"
+	commonService "cicv-data-closedloop/kinglong/common/svc"
+	"cicv-data-closedloop/kinglong/common/util"
+	"net"
+	"time"
+)
+
+// RunTimeWindowProducerQueue 将时间窗口内的包全部move出去，并等待当前时间窗口结束触发上传
+func RunTimeWindowProducerQueue() {
+	log.GlobalLogger.Info("生产者队列goroutine - 启动")
+	for {
+		// 收到自杀信号
+		select {
+		case signal := <-commonService.ChannelKillMove:
+			if signal == 1 {
+				commonService.ChannelKillMove <- 1
+				if len(global.TimeWindowProducerQueue) == 0 {
+					commonService.AddKillTimes("4")
+					return
+				}
+			} else { //signal == 2
+				commonService.AddKillTimes("4")
+				return
+			}
+		default:
+		}
+
+		time.Sleep(time.Duration(1) * time.Second)
+		if len(global.TimeWindowProducerQueue) > 0 {
+			bags := util.ListAbsolutePathWithSuffixAndSort(commonConfig.CloudConfig.BagDataDir, ".bag")
+			currentTimeWindow := global.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目录
+					util.MoveFromDataToCopy(currentTimeWindow.FaultTime, bag)
+					move = true
+				} else {
+					if util.TimeCustom1GreaterEqualThanTimeCustom2(bagTime, currentTimeWindow.TimeWindowBegin) {
+						// 必须已经生成了窗口之后的包才算窗口结束了
+						bigger = true
+						break
+					}
+				}
+			}
+			// 如果没有包可以供当前窗口移动，且已经生成了更新的包，则当前窗口已经可以上传
+			if !move && bigger {
+				// 1 如果第一个已经大于了timeWindowEnd，则触发上传并删除
+				// 将时间窗口发送给从节点
+				currentTimeWindow.CanUpload = "yes"
+				log.GlobalLogger.Info("将已完成的窗口发送给从节点：", currentTimeWindow.CanUpload)
+				util.SupplyCopyBags(currentTimeWindow)
+				util.RefreshTcpSendTime()
+				go sendTimeWindowByTcp(currentTimeWindow)
+				// 将时间窗口移出准备队列
+				util.RemoveHeadOfdTimeWindowProducerQueue()
+				// 将时间窗口加入运行队列
+				util.AddTimeWindowToTimeWindowConsumerQueue(currentTimeWindow)
+				// 获取copy目录下的字典json，key为触发时间，value为label
+				timeToLabelJson, _ := commonUtil.ReadFile(commonConfig.CloudConfig.TimeToLabelJsonPath)
+				timeToLabelMap, _ := commonUtil.JsonStringToMap(timeToLabelJson)
+				timeToLabelMap[currentTimeWindow.FaultTime] = commonUtil.ToString(currentTimeWindow.Labels)
+				timeToLabelJson, _ = commonUtil.MapToJsonString(timeToLabelMap)
+				_ = commonUtil.WriteFile(timeToLabelJson, commonConfig.CloudConfig.TimeToLabelJsonPath)
+				continue
+			} else { // 保证当前窗口只发送一次，每间隔5秒发一次
+				if int(time.Since(global.TcpSendTime).Seconds()) > commonConfig.CloudConfig.TimeWindowSendGap {
+					log.GlobalLogger.Info("每隔", commonConfig.CloudConfig.TimeWindowSendGap, "秒发送一次tcp消息")
+					util.RefreshTcpSendTime()
+					// 2 如果第一个不大于timeWindowEnd，则发送不可上传的窗口信息。
+					currentTimeWindow.CanUpload = "no"
+					go sendTimeWindowByTcp(currentTimeWindow)
+				}
+			}
+		}
+	}
+}
+
+//TODO 服务端接受连接时如何维护该链接
+//func sendTimeWindowByTcp(timeWindow ent.TimeWindow) {
+//	// 发送数据
+//	send := util2.TimeWindowToJson(timeWindow)
+//	_, err := masterCfg.TcpConnection.Write([]byte(send))
+//	if err != nil {
+//		log.GlobalLogger.Error("master发送给slave时间窗口", timeWindow, "失败：", err)
+//		return
+//	}
+//}
+
+func sendTimeWindowByTcp(timeWindow ent.TimeWindow) {
+	socket := commonConfig.CloudConfig.Hosts[1].Ip + ":" + commonConfig.CloudConfig.TcpPort
+	tcpConn, err := net.Dial("tcp", socket)
+	if err != nil {
+		log.GlobalLogger.Error("建立tcp连接", socket, "失败：", err)
+		return
+	}
+	defer tcpConn.Close()
+	// 发送数据
+	send := util.TimeWindowToJson(timeWindow)
+	_, err = tcpConn.Write([]byte(send))
+	if err != nil {
+		log.GlobalLogger.Error("master发送给slave时间窗口", timeWindow, "失败：", err)
+		return
+	}
+}

pjisuv/master/pkg/svc/produce_window.go

@@ -0,0 +1,310 @@
+package svc
+
+import (
+	commonConfig "cicv-data-closedloop/kinglong/common/cfg"
+	"cicv-data-closedloop/kinglong/common/ent"
+	"cicv-data-closedloop/kinglong/common/global"
+	"cicv-data-closedloop/kinglong/common/log"
+	"cicv-data-closedloop/kinglong/common/svc"
+	"cicv-data-closedloop/kinglong/common/util"
+	masterConfig "cicv-data-closedloop/kinglong/master/pkg/cfg"
+	"cicv-data-closedloop/kinglong_msgs"
+	"github.com/bluenviron/goroslib/v2"
+	"os"
+	"sync"
+	"time"
+)
+
+var (
+	subscriber0Mutex sync.Mutex
+	subscriber1Mutex sync.Mutex
+	subscriber2Mutex sync.Mutex
+	subscriber3Mutex sync.Mutex
+	subscriber4Mutex sync.Mutex
+	m                sync.RWMutex
+	velocityX        float64
+	velocityY        float64
+	yaw              float64
+)
+
+// PrepareTimeWindowProducerQueue 负责监听所有主题并修改时间窗口
+func PrepareTimeWindowProducerQueue() {
+	log.GlobalLogger.Info("订阅者 goroutine，启动。")
+
+	//创建订阅者0订阅主题 nodefault_info
+	log.GlobalLogger.Info("创建订阅者0订阅话题 ", masterConfig.TopicOfNodeFaultInfo)
+	subscriber0, err := goroslib.NewSubscriber(goroslib.SubscriberConf{
+		Node:  commonConfig.RosNode,
+		Topic: masterConfig.TopicOfNodeFaultInfo,
+		Callback: func(data *kinglong_msgs.FaultInfo) {
+			if len(masterConfig.RuleOfNodefaultInfo) == 0 {
+				//log.GlobalLogger.Info("话题 nodefault_info没有触发器")
+				return
+			}
+			global.Subscriber0TimeMutex.Lock()
+			// 判断是否是连续故障码
+			gap := time.Since(global.Subscriber0Time).Seconds()
+			if gap < 2 {
+				global.Subscriber0Time = time.Now()
+				global.Subscriber0TimeMutex.Unlock()
+				return
+			} else {
+				// 2 不是连续故障码
+				global.Subscriber0Time = time.Now()
+				global.Subscriber0TimeMutex.Unlock()
+				subscriber0Mutex.Lock()
+				{
+					faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
+					lastTimeWindow := util.GetLastTimeWindow() // 获取最后一个时间窗口
+					var faultLabel string
+					for _, f := range masterConfig.RuleOfNodefaultInfo {
+						faultLabel = f(data)
+						if faultLabel != "" {
+							saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
+							break
+						}
+					}
+				}
+				subscriber0Mutex.Unlock()
+			}
+
+		}})
+	if err != nil {
+		log.GlobalLogger.Info("创建订阅者0发生故障：", err)
+		os.Exit(-1)
+	}
+	// 创建订阅者1订阅主题 cicv_location
+	log.GlobalLogger.Info("创建订阅者1订阅话题 ", masterConfig.TopicOfCicvLocation)
+	subscriber1, err := goroslib.NewSubscriber(goroslib.SubscriberConf{
+		Node:  commonConfig.RosNode,
+		Topic: masterConfig.TopicOfCicvLocation,
+		Callback: func(data *kinglong_msgs.PerceptionLocalization) {
+			if len(masterConfig.RuleOfCicvLocation) == 0 {
+				log.GlobalLogger.Info("话题 cicv_location 没有触发器")
+				return
+			}
+			subscriber1Mutex.Lock()
+			{
+				faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
+				lastTimeWindow := util.GetLastTimeWindow() // 获取最后一个时间窗口
+				// 更新共享变量
+				m.RLock()
+				{
+					velocityX = data.VelocityX
+					velocityY = data.VelocityY
+					yaw = data.Yaw
+				}
+				m.RUnlock()
+				var faultLabel string
+				for _, f := range masterConfig.RuleOfCicvLocation {
+					faultLabel = f(data)
+					if faultLabel != "" {
+						saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
+						break
+					}
+				}
+			}
+			subscriber1Mutex.Unlock()
+		},
+	})
+	if err != nil {
+		log.GlobalLogger.Info("创建订阅者1发生故障：", err)
+		os.Exit(-1)
+	}
+	// 创建订阅者2订阅主题 tpperception
+	log.GlobalLogger.Info("创建订阅者2订阅话题 ", masterConfig.TopicOfTpperception)
+	subscriber2, err := goroslib.NewSubscriber(goroslib.SubscriberConf{
+		Node:  commonConfig.RosNode,
+		Topic: masterConfig.TopicOfTpperception,
+		Callback: func(data *kinglong_msgs.PerceptionObjects) {
+			if len(masterConfig.RuleOfTpperception) == 0 {
+				log.GlobalLogger.Info("话题 tpperception 没有触发器")
+				return
+			}
+			global.Subscriber2TimeMutex.Lock()
+			// 判断是否是连续故障码
+			gap := time.Since(global.Subscriber2Time).Seconds()
+			if gap < 10 {
+				global.Subscriber2TimeMutex.Unlock()
+				return
+			} else {
+				// 2 不是连续故障码
+				global.Subscriber2Time = time.Now()
+				global.Subscriber2TimeMutex.Unlock()
+				subscriber2Mutex.Lock()
+				{
+					faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
+					lastTimeWindow := util.GetLastTimeWindow() // 获取最后一个时间窗口
+					var faultLabel string
+					for _, f := range masterConfig.RuleOfTpperception {
+						faultLabel = f(data, velocityX, velocityY, yaw)
+						if faultLabel != "" {
+							saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
+							break
+						}
+					}
+				}
+				subscriber2Mutex.Unlock()
+			}
+
+		}})
+	if err != nil {
+		log.GlobalLogger.Info("创建订阅者2发生故障：", err)
+		os.Exit(-1)
+	}
+	// 创建订阅者3订阅主题 fault_info
+	log.GlobalLogger.Info("创建订阅者3订阅话题 ", masterConfig.TopicOfFaultInfo)
+	subscriber3, err := goroslib.NewSubscriber(goroslib.SubscriberConf{
+		Node:  commonConfig.RosNode,
+		Topic: masterConfig.TopicOfFaultInfo,
+		Callback: func(data *kinglong_msgs.FaultVec) {
+			if len(masterConfig.RuleOfFaultInfo) == 0 {
+				log.GlobalLogger.Info("话题 fault_info 没有触发器")
+				return
+			}
+			global.Subscriber3TimeMutex.Lock()
+			// 判断是否是连续故障码
+			gap := time.Since(global.Subscriber3Time).Seconds()
+			if gap < 2 {
+				global.Subscriber3Time = time.Now()
+				global.Subscriber3TimeMutex.Unlock()
+				return
+			} else {
+				// 2 不是连续故障码
+				global.Subscriber3Time = time.Now()
+				global.Subscriber3TimeMutex.Unlock()
+				subscriber3Mutex.Lock()
+				{
+					faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
+					lastTimeWindow := util.GetLastTimeWindow() // 获取最后一个时间窗口
+					var faultLabel string
+					for _, f := range masterConfig.RuleOfFaultInfo {
+						faultLabel = f(data)
+						if faultLabel != "" {
+							saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
+							break
+						}
+					}
+				}
+				subscriber3Mutex.Unlock()
+			}
+
+		}})
+	if err != nil {
+		log.GlobalLogger.Info("创建订阅者3发生故障：", err)
+		os.Exit(-1)
+	}
+	// 创建订阅者4订阅主题 data_read
+	// TODO 高频率触发
+	log.GlobalLogger.Info("创建订阅者4订阅话题 ", masterConfig.TopicOfDataRead)
+	subscriber4, err := goroslib.NewSubscriber(goroslib.SubscriberConf{
+		Node:  commonConfig.RosNode,
+		Topic: masterConfig.TopicOfDataRead,
+		Callback: func(data *kinglong_msgs.Retrieval) {
+			if len(masterConfig.RuleOfDataRead) == 0 {
+				//log.GlobalLogger.Info("话题 data_read 没有触发器")
+				return
+			}
+			subscriber4Mutex.Lock()
+			{
+				faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
+				lastTimeWindow := util.GetLastTimeWindow() // 获取最后一个时间窗口
+				var faultLabel string
+				for _, f := range masterConfig.RuleOfDataRead {
+					faultLabel = f(data)
+					if faultLabel != "" {
+						saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
+						break
+					}
+				}
+			}
+			subscriber4Mutex.Unlock()
+		},
+	})
+	if err != nil {
+		log.GlobalLogger.Info("创建订阅者3发生故障：", err)
+		os.Exit(-1)
+	}
+	select {
+	case signal := <-svc.ChannelKillWindowProducer:
+		if signal == 1 {
+			defer svc.AddKillTimes("3")
+			subscriber0.Close()
+			subscriber1.Close()
+			subscriber2.Close()
+			subscriber3.Close()
+			subscriber4.Close()
+			commonConfig.RosNode.Close()
+			return
+		}
+	}
+}
+
+func saveTimeWindow(faultLabel string, faultHappenTime string, lastTimeWindow *ent.TimeWindow) {
+	masterTopics, slaveTopics := getTopicsOfNode(faultLabel)
+	if lastTimeWindow == nil || util.TimeCustom1GreaterTimeCustom2(faultHappenTime, lastTimeWindow.TimeWindowEnd) {
+		// 2-1 如果是不在旧故障窗口内，添加一个新窗口
+		newTimeWindow := ent.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,
+		}
+		log.GlobalLogger.Infof("不在旧故障窗口内，向生产者队列添加一个新窗口，【Lable】=%v，【FaultTime】=%v，【Length】=%v", newTimeWindow.Labels, newTimeWindow.FaultTime, newTimeWindow.Length)
+
+		util.AddTimeWindowToTimeWindowProducerQueue(newTimeWindow)
+	} else {
+		// 2-2 如果在旧故障窗口内
+		global.TimeWindowProducerQueueMutex.RLock()
+		defer global.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)
+		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
+}

pjisuv/slave/main/slave.go

@@ -0,0 +1,38 @@
+package main
+
+import (
+	cfg2 "cicv-data-closedloop/kinglong/common/cfg"
+	commonInit "cicv-data-closedloop/kinglong/common/init"
+	"cicv-data-closedloop/kinglong/common/log"
+	svc3 "cicv-data-closedloop/kinglong/common/svc"
+	slaveConfig "cicv-data-closedloop/kinglong/slave/pkg/cfg"
+	svc4 "cicv-data-closedloop/kinglong/slave/pkg/svc"
+)
+
+// init 初始化函数
+func init() {
+	//runtime.GOMAXPROCS(1)
+	log.InitLogConfig("kinglong-slave")
+	commonInit.Init()
+	slaveConfig.InitTcpListener()
+	cfg2.InitKillSignalListener(cfg2.CloudConfig.Hosts[1].Ip)
+	// 等待重启，接收到重启信号，会把信号分发给以下channel
+	go svc3.WaitKillSelf()
+}
+
+// main 主函数
+func main() {
+
+	// 1 负责打包数据到data目录
+	go svc3.BagRecord(cfg2.CloudConfig.Hosts[1].Name)
+	// 2 负责监控故障，并修改timeWindow
+	go svc4.PrepareTimeWindowProducerQueue()
+	// 3
+	go svc4.RunTimeWindowProducerQueue()
+	// 4 排队运行时间窗口
+	go svc3.RunTimeWindowConsumerQueue(cfg2.CloudConfig.Hosts[1].Name)
+
+	// 阻塞主线程，等待其他线程执行。
+	select {}
+
+}

pjisuv/slave/pkg/cfg/slave_tcp_config.go

@@ -0,0 +1,22 @@
+package cfg
+
+import (
+	commonConfig "cicv-data-closedloop/kinglong/common/cfg"
+	"cicv-data-closedloop/kinglong/common/log"
+	"net"
+	"os"
+)
+
+var TcpListener net.Listener
+
+func InitTcpListener() {
+	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 {
+		log.GlobalLogger.Error("监听tcp端口失败:", err)
+		os.Exit(-1)
+	}
+	log.GlobalLogger.Info("从节点初始化TCP端口监听 - 成功：", socket)
+}

pjisuv/slave/pkg/svc/accept_window.go

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

pjisuv/slave/pkg/svc/move_bag.go

@@ -0,0 +1,69 @@
+package svc
+
+import (
+	commonConfig "cicv-data-closedloop/kinglong/common/cfg"
+	"cicv-data-closedloop/kinglong/common/global"
+	"cicv-data-closedloop/kinglong/common/log"
+	commonService "cicv-data-closedloop/kinglong/common/svc"
+	"cicv-data-closedloop/kinglong/common/util"
+	"time"
+)
+
+// RunTimeWindowProducerQueue 将时间窗口内的包全部move出去，并等待当前时间窗口结束触发上传
+func RunTimeWindowProducerQueue() {
+	log.GlobalLogger.Info("生产者队列 - 启动")
+	for { // 必须串行排队处理
+		select {
+		case signal := <-commonService.ChannelKillMove:
+			if signal == 1 {
+				commonService.ChannelKillMove <- 1
+				if len(global.TimeWindowProducerQueue) == 0 {
+					commonService.AddKillTimes("4")
+					return
+				}
+			} else { //signal == 2
+				commonService.AddKillTimes("4")
+				return
+			}
+		default:
+		}
+		time.Sleep(time.Duration(1) * time.Second)
+		if len(global.TimeWindowProducerQueue) > 0 {
+			currentTimeWindow := global.TimeWindowProducerQueue[0]
+			// 将时间窗口移出准备队列
+			util.RemoveHeadOfdTimeWindowProducerQueue()
+			if currentTimeWindow.CanUpload == "yes" {
+				log.GlobalLogger.Info("从节点接收到可上传的timeWindow")
+			}
+			if currentTimeWindow.CanUpload == "no" {
+				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目录
+					util.MoveFromDataToCopy(currentTimeWindow.FaultTime, bag)
+				} else {
+					if util.TimeCustom1GreaterEqualThanTimeCustom2(bagTime, currentTimeWindow.TimeWindowBegin) {
+						// 必须已经生成了窗口之后的包才算窗口结束了
+						break
+					}
+				}
+			}
+			// 判断是否可上传
+			if currentTimeWindow.CanUpload == "yes" {
+				// 1 timeWindow可以上传
+				log.GlobalLogger.Info("timeWindow可以上传：", currentTimeWindow)
+				// 补充bag包
+				util.SupplyCopyBags(currentTimeWindow)
+				// 将时间窗口加入运行队列
+				util.AddTimeWindowToTimeWindowConsumerQueue(currentTimeWindow)
+			}
+		}
+	}
+}

pjisuv_msgs/autoware_msgs.go

@@ -0,0 +1 @@
+package pjisuv_msgs

pjisuv_msgs/can_msgs.go

@@ -0,0 +1 @@
+package pjisuv_msgs

pjisuv_msgs/canbus_msgs.go

@@ -0,0 +1 @@
+package pjisuv_msgs

pjisuv_msgs/common_msgs.go

@@ -0,0 +1 @@
+package pjisuv_msgs

pjisuv_msgs/custom_msgs.go

@@ -0,0 +1 @@
+package pjisuv_msgs

pjisuv_msgs/diag_msgs.go

@@ -0,0 +1 @@
+package pjisuv_msgs

pjisuv_msgs/jsk_footstep_msgs.go

@@ -0,0 +1 @@
+package pjisuv_msgs

pjisuv_msgs/jsk_recognition_msgs.go

@@ -0,0 +1 @@
+package pjisuv_msgs

pjisuv_msgs/lidar_msgs.go

@@ -0,0 +1 @@
+package pjisuv_msgs

pjisuv_msgs/map_msgs.go

@@ -0,0 +1 @@
+package pjisuv_msgs

pjisuv_msgs/nav_msgs.go

@@ -0,0 +1 @@
+package pjisuv_msgs

pjisuv_msgs/nox_msgs.go

@@ -0,0 +1 @@
+package pjisuv_msgs

pjisuv_msgs/perception_msgs.go

@@ -0,0 +1 @@
+package pjisuv_msgs

pjisuv_msgs/v2x_msgs.go

@@ -0,0 +1 @@
+package pjisuv_msgs

pjisuv_msgs/vehicle_msgs.go

@@ -0,0 +1 @@
+package pjisuv_msgs

pjisuv_msgs/原始msg文件/autoware_msgs/msg/AccelCmd.msg

@@ -0,0 +1,2 @@
+Header header
+int32 accel

pjisuv_msgs/原始msg文件/autoware_msgs/msg/AdjustXY.msg

@@ -0,0 +1,3 @@
+Header header
+int32 x
+int32 y

pjisuv_msgs/原始msg文件/autoware_msgs/msg/BrakeCmd.msg

@@ -0,0 +1,2 @@
+Header header
+int32 brake

pjisuv_msgs/原始msg文件/autoware_msgs/msg/CloudCluster.msg

@@ -0,0 +1,31 @@
+std_msgs/Header header
+
+uint32 id
+string label
+float64 score
+
+sensor_msgs/PointCloud2 cloud
+
+geometry_msgs/PointStamped min_point
+geometry_msgs/PointStamped max_point
+geometry_msgs/PointStamped avg_point
+geometry_msgs/PointStamped centroid_point
+
+float64 estimated_angle
+
+geometry_msgs/Vector3 dimensions
+geometry_msgs/Vector3 eigen_values
+geometry_msgs/Vector3[] eigen_vectors
+
+#Array of 33 floats containing the FPFH descriptor
+std_msgs/Float32MultiArray fpfh_descriptor 
+
+jsk_recognition_msgs/BoundingBox bounding_box
+geometry_msgs/PolygonStamped convex_hull
+
+# Indicator information
+# INDICATOR_LEFT 0
+# INDICATOR_RIGHT 1
+# INDICATOR_BOTH 2
+# INDICATOR_NONE 3
+uint32 indicator_state

pjisuv_msgs/原始msg文件/autoware_msgs/msg/CloudClusterArray.msg

@@ -0,0 +1,2 @@
+std_msgs/Header header
+CloudCluster[] clusters

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ColorSet.msg

@@ -0,0 +1,3 @@
+ValueSet Hue
+ValueSet Sat
+ValueSet Val

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ControlCommand.msg

@@ -0,0 +1,3 @@
+float64 linear_velocity
+float64 linear_acceleration #m/s^2
+float64 steering_angle

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ControlCommandStamped.msg

@@ -0,0 +1,2 @@
+Header header
+ControlCommand cmd

pjisuv_msgs/原始msg文件/autoware_msgs/msg/DTLane.msg

@@ -0,0 +1,8 @@
+float64 dist
+float64 dir
+float64 apara
+float64 r
+float64 slope
+float64 cant
+float64 lw
+float64 rw

pjisuv_msgs/原始msg文件/autoware_msgs/msg/DetectedObject.msg

@@ -0,0 +1,43 @@
+std_msgs/Header                 header
+
+uint32                          id
+string                          label
+float32                         score   #Score as defined by the detection, Optional
+std_msgs/ColorRGBA              color   # Define this object specific color
+bool                            valid   # Defines if this object is valid, or invalid as defined by the filtering
+
+################ 3D BB
+string                          space_frame #3D Space coordinate frame of the object, required if pose and dimensions are defines
+geometry_msgs/Pose              pose
+geometry_msgs/Vector3           dimensions
+geometry_msgs/Vector3           variance
+geometry_msgs/Twist             velocity
+geometry_msgs/Twist             acceleration
+
+sensor_msgs/PointCloud2         pointcloud
+
+geometry_msgs/PolygonStamped    convex_hull
+autoware_msgs/LaneArray         candidate_trajectories
+
+bool                            pose_reliable
+bool                            velocity_reliable
+bool                            acceleration_reliable
+
+############### 2D Rect
+string                          image_frame # Image coordinate Frame,        Required if x,y,w,h defined
+int32                           x           # X coord in image space(pixel) of the initial point of the Rect
+int32                           y           # Y coord in image space(pixel) of the initial point of the Rect
+int32                           width       # Width of the Rect in pixels
+int32                           height      # Height of the Rect in pixels
+float32                         angle       # Angle [0 to 2*PI), allow rotated rects
+
+sensor_msgs/Image               roi_image
+
+############### Indicator information
+uint8                          indicator_state # INDICATOR_LEFT = 0, INDICATOR_RIGHT = 1, INDICATOR_BOTH = 2, INDICATOR_NONE = 3
+
+############### Behavior State of the Detected Object
+uint8                           behavior_state # FORWARD_STATE = 0, STOPPING_STATE = 1, BRANCH_LEFT_STATE = 2, BRANCH_RIGHT_STATE = 3, YIELDING_STATE = 4, ACCELERATING_STATE = 5, SLOWDOWN_STATE = 6
+
+#
+string[]                        user_defined_info

pjisuv_msgs/原始msg文件/autoware_msgs/msg/DetectedObjectArray.msg

@@ -0,0 +1,2 @@
+std_msgs/Header header
+DetectedObject[] objects

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ExtractedPosition.msg

@@ -0,0 +1,15 @@
+# This message defines the information required to describe a lamp (bulb)
+# in a traffic signal, according to the information extracted from the ADAS Map
+# and the extrinsic camera calibration
+
+int32 signalId  # Traffic Signal Lamp ID
+int32 u         # Lamp ROI x in image coords
+int32 v         # Lamp ROI y in image coords
+int32 radius    # Lamp Radius
+float64 x       # X position in map coordinates
+float64 y       # Y position in map coordinates
+float64 z       # Z position in map coordinates
+float64 hang    # Azimuth "Horizontal Angle"
+int8 type       # Lamp Type (red, yellow, green, ...)
+int32 linkId    # Closest LinkID (lane) in VectorMap
+int32 plId      # PoleID to which this Lamp belongs to

pjisuv_msgs/原始msg文件/autoware_msgs/msg/Gear.msg

@@ -0,0 +1,7 @@
+uint8 NONE=0
+uint8 PARK=1
+uint8 REVERSE=2
+uint8 NEUTRAL=3
+uint8 DRIVE=4
+uint8 LOW=5
+uint8 gear

pjisuv_msgs/原始msg文件/autoware_msgs/msg/GeometricRectangle.msg

@@ -0,0 +1,4 @@
+float32 wl
+float32 wr
+float32 lf
+float32 lb

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ICPStat.msg

@@ -0,0 +1,7 @@
+Header header
+float32 exe_time
+#int32 iteration
+float32 score
+float32 velocity
+float32 acceleration
+int32 use_predict_pose

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ImageLaneObjects.msg

@@ -0,0 +1,9 @@
+Header header
+int32 lane_l_x1 # Left Lane
+int32 lane_l_y1
+int32 lane_l_x2
+int32 lane_l_y2
+int32 lane_r_x1 # Right Lane
+int32 lane_r_y1
+int32 lane_r_x2
+int32 lane_r_y2

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ImageObj.msg

@@ -0,0 +1,4 @@
+Header header
+string type
+ImageRect[] obj
+# XXX Should this message have 'score' ?

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ImageObjRanged.msg

@@ -0,0 +1,3 @@
+Header header
+string type
+ImageRectRanged[] obj

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ImageObjTracked.msg

@@ -0,0 +1,7 @@
+Header header
+string type
+uint8 total_num
+int32[] obj_id
+ImageRectRanged[] rect_ranged
+int32[] real_data
+int32[] lifespan

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ImageObjects.msg

@@ -0,0 +1,5 @@
+Header header
+uint8 car_num
+int32[] car_type
+float32[] score
+int32[] corner_point

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ImageRect.msg

@@ -0,0 +1,5 @@
+int32 x
+int32 y
+int32 height
+int32 width
+float32 score

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ImageRectRanged.msg

@@ -0,0 +1,4 @@
+ImageRect rect
+float32 range
+float32 min_height
+float32 max_height

pjisuv_msgs/原始msg文件/autoware_msgs/msg/IndicatorCmd.msg

@@ -0,0 +1,3 @@
+Header header
+int32 l
+int32 r

pjisuv_msgs/原始msg文件/autoware_msgs/msg/LampCmd.msg

@@ -0,0 +1,3 @@
+Header header
+int32 l
+int32 r

pjisuv_msgs/原始msg文件/autoware_msgs/msg/Lane.msg

@@ -0,0 +1,10 @@
+Header header
+int32 increment
+int32 lane_id
+Waypoint[] waypoints
+
+uint32 lane_index
+float32 cost
+float32 closest_object_distance
+float32 closest_object_velocity
+bool is_blocked

pjisuv_msgs/原始msg文件/autoware_msgs/msg/LaneArray.msg

@@ -0,0 +1,2 @@
+int32 id
+Lane[] lanes

pjisuv_msgs/原始msg文件/autoware_msgs/msg/NDTStat.msg

@@ -0,0 +1,7 @@
+Header header
+float32 exe_time
+int32 iteration
+float32 score
+float32 velocity
+float32 acceleration
+int32 use_predict_pose

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ObjLabel.msg

@@ -0,0 +1,4 @@
+Header header
+string type
+int32[] obj_id
+geometry_msgs/Point[] reprojected_pos

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ObjPose.msg

@@ -0,0 +1,4 @@
+Header header
+string type
+int32[] obj_id
+geometry_msgs/PoseArray[] obj

pjisuv_msgs/原始msg文件/autoware_msgs/msg/PointsImage.msg

@@ -0,0 +1,9 @@
+Header header
+float32[] distance
+float32[] intensity
+float32[] min_height
+float32[] max_height
+int32 max_y
+int32 min_y
+int32 image_height
+int32 image_width

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ProjectionMatrix.msg

@@ -0,0 +1,2 @@
+Header header
+float64[16] projection_matrix

pjisuv_msgs/原始msg文件/autoware_msgs/msg/RemoteCmd.msg

@@ -0,0 +1,3 @@
+Header header
+autoware_msgs/VehicleCmd vehicle_cmd
+int32 control_mode

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ScanImage.msg

@@ -0,0 +1,5 @@
+Header header
+float32[] distance
+float32[] intensity
+int32 max_y
+int32 min_y

pjisuv_msgs/原始msg文件/autoware_msgs/msg/Signals.msg

@@ -0,0 +1,2 @@
+Header header
+ExtractedPosition[] Signals

pjisuv_msgs/原始msg文件/autoware_msgs/msg/State.msg

@@ -0,0 +1,5 @@
+Header header
+string vehicle_state
+string mission_state
+string behavior_state
+string motion_state

pjisuv_msgs/原始msg文件/autoware_msgs/msg/StateCmd.msg

@@ -0,0 +1,2 @@
+Header header
+int32 cmd

pjisuv_msgs/原始msg文件/autoware_msgs/msg/SteerCmd.msg

@@ -0,0 +1,2 @@
+Header header
+int32 steer

pjisuv_msgs/原始msg文件/autoware_msgs/msg/SyncTimeDiff.msg

@@ -0,0 +1,4 @@
+Header header
+float64 time_diff
+time camera
+time lidar

pjisuv_msgs/原始msg文件/autoware_msgs/msg/SyncTimeMonitor.msg

@@ -0,0 +1,27 @@
+Header header
+float64 image_raw
+float64 points_raw
+float64 points_image
+float64 vscan_points
+float64 vscan_image
+float64 image_obj
+float64 image_obj_ranged
+float64 image_obj_tracked
+float64 current_pose
+float64 obj_label
+float64 cluster_centroids
+float64 obj_pose
+float64 execution_time
+float64 cycle_time
+float64 time_diff
+
+
+# time image_raw
+# time points_raw
+# time image_obj
+# time image_obj_ranged
+# time image_obj_tracked
+# time current_pose
+# time obj_label
+# time cluster_centroids
+# time obj_pose

pjisuv_msgs/原始msg文件/autoware_msgs/msg/TrafficLight.msg

@@ -0,0 +1,2 @@
+Header header
+int32 traffic_light

pjisuv_msgs/原始msg文件/autoware_msgs/msg/TrafficLightResult.msg

@@ -0,0 +1,22 @@
+# This message defines meta information for a traffic light status
+Header header
+
+# Each bulb is assigned with an Id
+# The TLR node convention is to use the yellow bulb's ID
+int32 light_id
+
+# The result as provided by the tlr nodes
+# Red and Yellow states are both treated as STOP
+# RED       = 0
+# YELLOW    = 0
+# GREEN     = 1
+# UNKNOWN   = 2
+int32 recognition_result
+
+# This string is used by the Audio Player
+string recognition_result_str
+
+# LaneId to which this traffic light result belongs to
+# this Id is defined by the ADAS MAP
+int32 lane_id
+

pjisuv_msgs/原始msg文件/autoware_msgs/msg/TrafficLightResultArray.msg

@@ -0,0 +1,2 @@
+Header header
+TrafficLightResult[] results

pjisuv_msgs/原始msg文件/autoware_msgs/msg/TunedResult.msg

@@ -0,0 +1,4 @@
+Header header
+ColorSet Red
+ColorSet Yellow
+ColorSet Green

pjisuv_msgs/原始msg文件/autoware_msgs/msg/ValueSet.msg

@@ -0,0 +1,2 @@
+int32 center
+int32 range

pjisuv_msgs/原始msg文件/autoware_msgs/msg/VehicleCmd.msg

@@ -0,0 +1,10 @@
+Header header
+autoware_msgs/SteerCmd steer_cmd
+autoware_msgs/AccelCmd accel_cmd
+autoware_msgs/BrakeCmd brake_cmd
+autoware_msgs/LampCmd lamp_cmd
+autoware_msgs/Gear gear_cmd
+int32 mode
+geometry_msgs/TwistStamped twist_cmd
+autoware_msgs/ControlCommand ctrl_cmd
+int32 emergency

pjisuv_msgs/原始msg文件/autoware_msgs/msg/VehicleLocation.msg

@@ -0,0 +1,3 @@
+Header header
+int32 lane_array_id
+int32 waypoint_index

pjisuv_msgs/原始msg文件/autoware_msgs/msg/VehicleStatus.msg

@@ -0,0 +1,24 @@
+Header header
+string tm
+
+# Powertrain
+int32 drivemode
+int32 steeringmode
+int32 MODE_MANUAL=0
+int32 MODE_AUTO=1
+
+autoware_msgs/Gear current_gear
+
+float64 speed # vehicle velocity [km/s]
+int32 drivepedal
+int32 brakepedal
+
+float64 angle # vehicle steering (tire) angle [rad]
+
+# Body
+int32 lamp
+int32 LAMP_LEFT=1
+int32 LAMP_RIGHT=2
+int32 LAMP_HAZARD=3
+
+int32 light