style(pre-commit): autofix

control_data_collecting_tool/README.md

@@ -82,10 +82,8 @@ This package provides tools for automatically collecting data using pure pursuit
     - `/data_collecting_lookahead_marker_array`
       - Type: MarkerArray
 
-
 6. The following actions differ depending on the selected course. If you select the trajectory from [`eight_course`, `u_shaped_return`, `straight_line_positive`, `straight_line_negative`, `reversal_loop_circle`], proceed to 6.1. If you select the trajectory from [`along_road`], please proceed to 6.2.
 
-
     - 6.1 If you choose the trajectory from [`eight_course`, `u_shaped_return`, `straight_line_positive`, `straight_line_negative`, `reversal_loop_circle`], select `DataCollectingAreaSelectionTool` plugin.
 
       <img src="resource/DataCollectingAreaSelection.png" width="480">

control_data_collecting_tool/scripts/courses/along_road.py

@@ -14,12 +14,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from autoware_map_msgs.msg import LaneletMapBin
 from courses.base_course import Base_Course
 from courses.lanelet import LaneletMapHandler
 import numpy as np
 from rcl_interfaces.msg import ParameterDescriptor
-from rclpy.qos import QoSProfile, QoSDurabilityPolicy
-from autoware_map_msgs.msg import LaneletMapBin
+from rclpy.qos import QoSDurabilityPolicy
+from rclpy.qos import QoSProfile
 from scipy.interpolate import interp1d as interpolation_1d
 
 
@@ -95,6 +96,7 @@ def declare_along_road_params(node):
         ),
     )
 
+
 class Along_Road(Base_Course):
     def __init__(self, step: float, param_dict):
         super().__init__(step, param_dict)
@@ -252,7 +254,7 @@ def get_target_velocity(
         collected_data_counts_of_vel_acc,
         collected_data_counts_of_vel_steer,
         mask_vel_acc,
-        mask_vel_steer
+        mask_vel_steer,
     ):
         part = self.parts[nearestIndex]
         achievement_rate = self.achievement_rates[nearestIndex]
@@ -264,8 +266,7 @@ def get_target_velocity(
             or (part == "straight" and achievement_rate < 0.05)
         ) and not self.set_target_velocity_on_straight_line:
             self.acc_idx, self.vel_idx = self.choose_target_velocity_acc(
-                collected_data_counts_of_vel_acc,
-                mask_vel_acc
+                collected_data_counts_of_vel_acc, mask_vel_acc
             )
             self.target_acc_on_straight_line = self.params.a_bin_centers[self.acc_idx]
             self.target_vel_on_straight_line = self.params.v_bin_centers[self.vel_idx]

control_data_collecting_tool/scripts/courses/base_course.py

@@ -57,13 +57,15 @@ def choose_target_velocity_acc(self, collected_data_counts_of_vel_acc, mask_vel_
             for j in range(
                 self.params.collecting_data_min_n_a, self.params.collecting_data_max_n_a
             ):
-                if mask_vel_acc[i,j] == 1:
+                if mask_vel_acc[i, j] == 1:
                     if min_num_data - min_data_num_margin > collected_data_counts_of_vel_acc[i, j]:
                         min_num_data = collected_data_counts_of_vel_acc[i, j]
                         min_index_list.clear()
                         min_index_list.append((j, i))
 
-                    elif min_num_data + min_data_num_margin > collected_data_counts_of_vel_acc[i, j]:
+                    elif (
+                        min_num_data + min_data_num_margin > collected_data_counts_of_vel_acc[i, j]
+                    ):
                         min_index_list.append((j, i))
 
         return min_index_list[np.random.randint(0, len(min_index_list))]
@@ -77,7 +79,7 @@ def get_target_velocity(
         collected_data_counts_of_vel_acc,
         collected_data_counts_of_vel_steer,
         mask_vel_acc,
-        mask_vel_steer
+        mask_vel_steer,
     ):
         pass
 

control_data_collecting_tool/scripts/courses/figure_eight.py

@@ -178,7 +178,7 @@ def get_target_velocity(
         collected_data_counts_of_vel_acc,
         collected_data_counts_of_vel_steer,
         mask_vel_acc,
-        mask_vel_steer
+        mask_vel_steer,
     ):
         part = self.parts[nearestIndex]
         achievement_rate = self.achievement_rates[nearestIndex]
@@ -190,8 +190,7 @@ def get_target_velocity(
             or (part == "straight" and achievement_rate < 0.05)
         ) and not self.set_target_velocity_on_straight_line:
             self.acc_idx, self.vel_idx = self.choose_target_velocity_acc(
-                collected_data_counts_of_vel_acc,
-                mask_vel_acc
+                collected_data_counts_of_vel_acc, mask_vel_acc
             )
             self.target_acc_on_straight_line = self.params.a_bin_centers[self.acc_idx]
             self.target_vel_on_straight_line = self.params.v_bin_centers[self.vel_idx]

control_data_collecting_tool/scripts/courses/load_course.py

@@ -44,6 +44,7 @@ def declare_course_params(course_name, node):
     elif course_name == "along_road":
         declare_along_road_params(node)
 
+
 def create_course_subscription(course_name, node):
     if course_name == "eight_course":
         pass
@@ -58,6 +59,7 @@ def create_course_subscription(course_name, node):
     elif course_name == "along_road":
         pass
 
+
 def load_course(course_name, step_size, params_dict):
     # Load the course based on the course name
     if course_name == "eight_course":

control_data_collecting_tool/scripts/courses/reversal_loop_circle.py

@@ -1012,16 +1012,15 @@ def get_target_velocity(
         collected_data_counts_of_vel_acc,
         collected_data_counts_of_vel_steer,
         mask_vel_acc,
-        mask_vel_steer
+        mask_vel_steer,
     ):
         # Calculate the target velocity for the vehicle based on its current state, trajectory phase, and constraints such as acceleration limits and trajectory curvature.
 
         # Initialize target acceleration and velocity if not already updated
         if not self.updated_target_velocity:
             # Choose velocity and acceleration bins based on collected data
             self.acc_idx, self.vel_idx = self.choose_target_velocity_acc(
-                collected_data_counts_of_vel_acc,
-                mask_vel_acc
+                collected_data_counts_of_vel_acc, mask_vel_acc
             )
             self.target_acc_on_segmentation = self.params.a_bin_centers[self.acc_idx]
             self.target_vel_on_segmentation = self.params.v_bin_centers[self.vel_idx]
@@ -1131,7 +1130,11 @@ def get_target_velocity(
             )
 
             # Determine the minimum velocity based on collected data
-            min_vel = np.min((collected_data_counts_of_vel_steer + 1e9 * (1-mask_vel_steer) )[:max_vel_idx, steer_idx])
+            min_vel = np.min(
+                (collected_data_counts_of_vel_steer + 1e9 * (1 - mask_vel_steer))[
+                    :max_vel_idx, steer_idx
+                ]
+            )
             vel_idx = np.where(
                 collected_data_counts_of_vel_steer[:max_vel_idx, steer_idx] == min_vel
             )[0]

control_data_collecting_tool/scripts/courses/straight_line_negative.py

@@ -76,7 +76,7 @@ def get_target_velocity(
         collected_data_counts_of_vel_acc,
         collected_data_counts_of_vel_steer,
         mask_vel_acc,
-        mask_vel_steer
+        mask_vel_steer,
     ):
         part = self.parts[nearestIndex]
         achievement_rate = self.achievement_rates[nearestIndex]
@@ -85,8 +85,7 @@ def get_target_velocity(
         # Check and update target velocity on straight line
         if part == "straight" and achievement_rate < 0.05:
             self.acc_idx, self.vel_idx = self.choose_target_velocity_acc(
-                collected_data_counts_of_vel_acc,
-                mask_vel_acc
+                collected_data_counts_of_vel_acc, mask_vel_acc
             )
             self.target_acc_on_straight_line = self.params.a_bin_centers[self.acc_idx]
             self.target_vel_on_straight_line = self.params.v_bin_centers[self.vel_idx]

control_data_collecting_tool/scripts/courses/straight_line_positive.py

@@ -76,7 +76,7 @@ def get_target_velocity(
         collected_data_counts_of_vel_acc,
         collected_data_counts_of_vel_steer,
         mask_vel_acc,
-        mask_vel_steer
+        mask_vel_steer,
     ):
         part = self.parts[nearestIndex]
         achievement_rate = self.achievement_rates[nearestIndex]
@@ -85,8 +85,7 @@ def get_target_velocity(
         # Check and update target velocity on straight line
         if part == "straight" and achievement_rate < 0.05:
             self.acc_idx, self.vel_idx = self.choose_target_velocity_acc(
-                collected_data_counts_of_vel_acc,
-                mask_vel_acc
+                collected_data_counts_of_vel_acc, mask_vel_acc
             )
             self.target_acc_on_straight_line = self.params.a_bin_centers[self.acc_idx]
             self.target_vel_on_straight_line = self.params.v_bin_centers[self.vel_idx]

control_data_collecting_tool/scripts/courses/u_shaped.py

@@ -146,7 +146,7 @@ def get_target_velocity(
         collected_data_counts_of_vel_acc,
         collected_data_counts_of_vel_steer,
         mask_vel_acc,
-        mask_vel_steer
+        mask_vel_steer,
     ):
         part = self.parts[nearestIndex]
         achievement_rate = self.achievement_rates[nearestIndex]
@@ -158,8 +158,7 @@ def get_target_velocity(
             or (part == "straight" and achievement_rate < 0.05)
         ) and not self.set_target_velocity_on_straight_line:
             self.acc_idx, self.vel_idx = self.choose_target_velocity_acc(
-                collected_data_counts_of_vel_acc,
-                mask_vel_acc
+                collected_data_counts_of_vel_acc, mask_vel_acc
             )
             self.target_acc_on_straight_line = self.params.a_bin_centers[self.acc_idx]
             self.target_vel_on_straight_line = self.params.v_bin_centers[self.vel_idx]

control_data_collecting_tool/scripts/data_collecting_base_node.py

@@ -14,15 +14,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import os
+
+from ament_index_python.packages import get_package_share_directory
 from autoware_adapi_v1_msgs.msg import OperationModeState
 from autoware_vehicle_msgs.msg import ControlModeReport
 from geometry_msgs.msg import AccelWithCovarianceStamped
 from nav_msgs.msg import Odometry
-import os
 import numpy as np
 from rcl_interfaces.msg import ParameterDescriptor
 from rclpy.node import Node
-from ament_index_python.packages import get_package_share_directory
 
 
 class DataCollectingBaseNode(Node):
@@ -43,9 +44,7 @@ def __init__(self, node_name):
         self.declare_parameter(
             "MASK_NAME",
             "default",
-            ParameterDescriptor(
-                description="Masks for Data collection"
-            ),
+            ParameterDescriptor(description="Masks for Data collection"),
         )
 
         self.declare_parameter(
@@ -205,7 +204,9 @@ def __init__(self, node_name):
         self.v_bins = np.linspace(self.v_min, self.v_max, self.num_bins_v + 1)
         self.steer_bins = np.linspace(self.steer_min, self.steer_max, self.num_bins_steer + 1)
         self.a_bins = np.linspace(self.a_min, self.a_max, self.num_bins_a + 1)
-        self.steer_rate_bins = np.linspace(self.steer_rate_min, self.steer_rate_max, self.num_bins_steer_rate + 1)
+        self.steer_rate_bins = np.linspace(
+            self.steer_rate_min, self.steer_rate_max, self.num_bins_steer_rate + 1
+        )
 
         self.v_bin_centers = (self.v_bins[:-1] + self.v_bins[1:]) / 2
         self.steer_bin_centers = (self.steer_bins[:-1] + self.steer_bins[1:]) / 2
@@ -217,16 +218,32 @@ def __init__(self, node_name):
         """
         # set mask name
         MASK_NAME = self.get_parameter("MASK_NAME").value
-        mask_directory_path = get_package_share_directory("control_data_collecting_tool") + "/config/masks/" + MASK_NAME
+        mask_directory_path = (
+            get_package_share_directory("control_data_collecting_tool")
+            + "/config/masks/"
+            + MASK_NAME
+        )
 
-        mask_vel_acc_path = os.path.join(mask_directory_path, f"{MASK_NAME}_Velocity_Acceleration.txt")
-        self.mask_vel_acc = self.load_mask_from_txt(mask_vel_acc_path, self.num_bins_v, self.num_bins_a)
+        mask_vel_acc_path = os.path.join(
+            mask_directory_path, f"{MASK_NAME}_Velocity_Acceleration.txt"
+        )
+        self.mask_vel_acc = self.load_mask_from_txt(
+            mask_vel_acc_path, self.num_bins_v, self.num_bins_a
+        )
 
-        mask_velocity_steering_path = os.path.join(mask_directory_path, f"{MASK_NAME}_Velocity_Steering.txt")
-        self.mask_vel_steer = self.load_mask_from_txt(mask_velocity_steering_path, self.num_bins_v, self.num_bins_steer)
+        mask_velocity_steering_path = os.path.join(
+            mask_directory_path, f"{MASK_NAME}_Velocity_Steering.txt"
+        )
+        self.mask_vel_steer = self.load_mask_from_txt(
+            mask_velocity_steering_path, self.num_bins_v, self.num_bins_steer
+        )
 
-        mask_velocity_steer_rate_path = os.path.join(mask_directory_path, f"{MASK_NAME}_Velocity_Steering_Rate.txt")
-        self.mask_vel_steer_rate = self.load_mask_from_txt(mask_velocity_steer_rate_path, self.num_bins_v, self.num_bins_steer_rate)
+        mask_velocity_steer_rate_path = os.path.join(
+            mask_directory_path, f"{MASK_NAME}_Velocity_Steering_Rate.txt"
+        )
+        self.mask_vel_steer_rate = self.load_mask_from_txt(
+            mask_velocity_steer_rate_path, self.num_bins_v, self.num_bins_steer_rate
+        )
 
     def onOdometry(self, msg):
         self._present_kinematic_state = msg
@@ -245,7 +262,7 @@ def subscribe_operation_mode(self, msg):
 
     def subscribe_control_mode(self, msg):
         self._present_control_mode_ = msg.mode
-    
+
     def load_mask_from_txt(self, file_path, nx, ny):
         """
         Loads a numerical mask from a text file into a numpy array.
@@ -262,5 +279,5 @@ def load_mask_from_txt(self, file_path, nx, ny):
                 return mask
         except Exception as e:
             pass
-        
+
         return np.ones((nx, ny), dtype=int)

control_data_collecting_tool/scripts/data_collecting_data_counter.py

@@ -77,7 +77,9 @@ def __init__(self):
             Int32MultiArray, "/control_data_collecting_tools/collected_data_counts_of_vel_steer", 10
         )
         self.collected_data_counts_of_vel_steer_rate_publisher_ = self.create_publisher(
-            Int32MultiArray, "/control_data_collecting_tools/collected_data_counts_of_vel_steer_rate", 10
+            Int32MultiArray,
+            "/control_data_collecting_tools/collected_data_counts_of_vel_steer_rate",
+            10,
         )
 
         self.vel_hist_publisher_ = self.create_publisher(
@@ -225,7 +227,6 @@ def count_observations(self, v, a, steer, steer_rate):
 
         if 0 <= v_bin < self.num_bins_v and 0 <= steer_rate_bin < self.num_bins_steer_rate:
             self.collected_data_counts_of_vel_steer_rate[v_bin, steer_rate_bin] += 1
-
 
     # call back for counting data points
     def timer_callback_counter(self):