Linting pt 2: sonar_tasks.py, sonar_utils.py, and sonar.py

Author: AnthonZh

onboard/src/sonar/sonar/sonar.py

@@ -108,7 +108,7 @@ def connect(self) -> None:
             self.run()
         except StopIteration:
             self.get_logger().error(
-                f'Error in connecting to sonar, trying again in {self.CONNECTION_RETRY_PERIOD} seconds.'
+                f'Error in connecting to sonar, trying again in {self.CONNECTION_RETRY_PERIOD} seconds.',
             )
 
     def init_sonar(self) -> None:
@@ -188,7 +188,7 @@ def get_sweep(self, range_start: int = 100, range_end: int = 300) -> np.ndarray:
         return np.vstack(sonar_sweep_data)
 
     def get_xy_of_object_in_sweep(
-        self, start_angle: int, end_angle: int
+        self, start_angle: int, end_angle: int,
     ) -> tuple[Pose | None, np.ndarray, float | None]:
         """
         Get the depth of the sweep of a detected object. For now uses mean value.
@@ -245,8 +245,6 @@ def get_xy_of_object_in_sweep(
 
         self.get_logger().info(f'x: {x_index}, y: {y_index}, normal: {normal_angle}')
 
-        sonar_angle = (start_angle + end_angle) / 2  # Take the middle of the sweep
-
         return (
             sonar_utils.to_robot_position(x_index, y_index, self.sample_period, self.NEGATE_POSE),
             color_image,
@@ -282,7 +280,7 @@ def constant_sweep(self) -> None:
             rclpy.shutdown()
 
     def perform_sonar_request(
-        self, request: SonarSweepRequest.Request, response: SonarSweepRequest.Response
+        self, request: SonarSweepRequest.Request, response: SonarSweepRequest.Response,
     ) -> SonarSweepRequest.Response:
         """
         Perform a sonar request.

onboard/src/sonar/sonar/sonar_utils.py

@@ -16,7 +16,7 @@
 
 
 def transform_pose(
-    buffer: tf2_ros.Buffer, pose: tf2_geometry_msgs.PoseStamped, source_frame_id: str, target_frame_id: str
+    buffer: tf2_ros.Buffer, pose: tf2_geometry_msgs.PoseStamped, source_frame_id: str, target_frame_id: str,
 ) -> tf2_geometry_msgs.PoseStamped:
     """
     Transform pose from source reference frame to target reference frame.
@@ -47,7 +47,7 @@ def transform_pose(
         tf2_ros.ExtrapolationException,
         tf2_ros.InvalidArgumentException,
     ) as e:
-        error_message = f"Failed to transform pose: {e}"
+        error_message = f'Failed to transform pose: {e}'
         raise RuntimeError(error_message) from e
 
 
@@ -208,7 +208,7 @@ def to_robot_position(x_index: int, y_index: int, sample_period: float, negate:
 
 
 def convert_to_ros_compressed_img(
-    sonar_sweep: np.ndarray, cv_bridge: CvBridge, compressed_format: str = "jpeg", is_color: bool = False
+    sonar_sweep: np.ndarray, cv_bridge: CvBridge, compressed_format: str = 'jpeg', is_color: bool = False,
 ) -> CompressedImage:
     """
     Convert any kind of image to ROS Compressed Image.

onboard/src/task_planning/task_planning/tasks/sonar_tasks.py

@@ -12,7 +12,7 @@
     from custom_msgs.srv import SonarSweepRequest
 
 
-logger = get_logger("sonar_tasks")
+logger = get_logger('sonar_tasks')
 
 MAX_STEPS = 3
 
@@ -21,28 +21,28 @@
 async def sonar_test(_self: Task, start_angle: float, end_angle: float, scan_distance: float) -> Task[None, None, None]:
     """Repeatedly perform sonar scans."""
     while True:
-        logger.info(f"Sonar scan from {start_angle} to {end_angle} degrees, distance: {scan_distance} m")
+        logger.info(f'Sonar scan from {start_angle} to {end_angle} degrees, distance: {scan_distance} m')
         future = Sonar().sweep(
             start_angle=start_angle,
             end_angle=end_angle,
             scan_distance=scan_distance,
         )
         if future is None:
-            logger.error("Could not call sonar request service.")
+            logger.error('Could not call sonar request service.')
         else:
-            service_response = cast("SonarSweepRequest.Response", await future)
-            logger.info(f"Sonar scan response: {service_response}")
+            service_response = cast('SonarSweepRequest.Response', await future)
+            logger.info(f'Sonar scan response: {service_response}')
 
 
 async def get_normal_angle(start_angle: float, end_angle: float, scan_distance: float) -> float:
     """Get a normal angle from the sonar scan."""
     future = Sonar().sweep(start_angle=start_angle, end_angle=end_angle, scan_distance=scan_distance)
     response = await future
     if not response.is_object:
-        logger.error("No object detected — cannot rotate")
+        logger.error('No object detected — cannot rotate')
         return np.nan
     if response.normal_angle is None:
-        logger.error("[Sonar] normal_angle was None — cannot rotate")
+        logger.error('[Sonar] normal_angle was None — cannot rotate')
         return np.nan
     return response.normal_angle
 
@@ -56,7 +56,7 @@ async def rotate_to_normal(
     yaw_threshold: float,
 ) -> Task[None, None, None]:
     """Rotates to face a normal angle."""
-    logger.info(f"Sonar scan from {start_angle} to {end_angle} degrees, distance: {scan_distance} m")
+    logger.info(f'Sonar scan from {start_angle} to {end_angle} degrees, distance: {scan_distance} m')
 
     normal_angle = await get_normal_angle(start_angle, end_angle, scan_distance)
     await move_to_pose_local(
@@ -88,7 +88,7 @@ async def rotate_to_angle_from_normal(
     rotated_angle: float,
 ) -> Task[None, None, None]:
     """Rotates to a specified angle using Sonar normal angle."""
-    logger.info(f"Sonar scan from {start_angle} to {end_angle} degrees, distance: {scan_distance} m")
+    logger.info(f'Sonar scan from {start_angle} to {end_angle} degrees, distance: {scan_distance} m')
 
     angle = await get_normal_angle(start_angle, end_angle, scan_distance)
     angle = rotated_angle + angle