Added nav2_msgs path fixes to mypy compliance.

Signed-off-by: Leander Stephen D'Souza <leanderdsouza1234@gmail.com>

Author: leander-dsouza

nav2_docking/opennav_docking/test/test_docking_server.py

@@ -200,12 +200,12 @@ def publish_odometry(self, odom_to_base_link: TransformStamped) -> None:
         odom.twist.twist = self.command
         self.odom_pub.publish(odom)
 
-    def action_feedback_callback(self, msg: DockRobot.Feedback) -> None:
+    def action_feedback_callback(self, msg: DockRobot.Impl.FeedbackMessage) -> None:
         # Force the docking action to run a full recovery loop and then
         # make contact with the dock (based on pose of robot) before
         # we report that the robot is charging
         if msg.feedback.num_retries > 0 and \
-                msg.feedback.state == msg.feedback.WAIT_FOR_CHARGE:
+                msg.feedback.state == DockRobot.Feedback.WAIT_FOR_CHARGE:
             self.is_charging = True
 
     def nav_execute_callback(

nav2_simple_commander/nav2_simple_commander/robot_navigator.py

@@ -28,11 +28,11 @@
                               ComputePathThroughPoses, ComputePathToPose, ComputeRoute, DockRobot,
                               DriveOnHeading, FollowGPSWaypoints, FollowPath, FollowWaypoints,
                               NavigateThroughPoses, NavigateToPose, SmoothPath, Spin, UndockRobot)
-from nav2_msgs.msg import Route
+from nav2_msgs.msg import Costmap, Route
 from nav2_msgs.srv import (ClearCostmapAroundPose, ClearCostmapAroundRobot,
                            ClearCostmapExceptRegion, ClearEntireCostmap, GetCostmap, LoadMap,
                            ManageLifecycleNodes, Toggle)
-from nav_msgs.msg import Goals, OccupancyGrid, Path
+from nav_msgs.msg import Goals, Path
 import rclpy
 from rclpy.action import ActionClient
 from rclpy.action.client import ClientGoalHandle
@@ -305,7 +305,7 @@ def goThroughPoses(self, poses: Goals, behavior_tree: str = '') -> Optional[Runn
 
         if not self.goal_handle or not self.goal_handle.accepted:
             msg = f'NavigateThroughPoses request with {len(poses.goals)} was rejected!'
-            self.setTaskError(NavigateThroughPoses.UNKNOWN, msg)
+            self.setTaskError(NavigateThroughPoses.Result.UNKNOWN, msg)
             self.error(msg)
             return None
 
@@ -344,7 +344,7 @@ def goToPose(self, pose: PoseStamped, behavior_tree: str = '') -> Optional[Runni
                 + str(pose.pose.position.y)
                 + ' was rejected!'
             )
-            self.setTaskError(NavigateToPose.UNKNOWN, msg)
+            self.setTaskError(NavigateToPose.Result.UNKNOWN, msg)
             self.error(msg)
             return None
 
@@ -370,7 +370,7 @@ def followWaypoints(self, poses: list[PoseStamped]) -> Optional[RunningTask]:
 
         if not self.goal_handle or not self.goal_handle.accepted:
             msg = f'Following {len(poses)} waypoints request was rejected!'
-            self.setTaskError(FollowWaypoints.UNKNOWN, msg)
+            self.setTaskError(FollowWaypoints.Result.UNKNOWN, msg)
             self.error(msg)
             return None
 
@@ -396,7 +396,7 @@ def followGpsWaypoints(self, gps_poses: list[GeoPose]) -> Optional[RunningTask]:
 
         if not self.goal_handle or not self.goal_handle.accepted:
             msg = f'Following {len(gps_poses)} gps waypoints request was rejected!'
-            self.setTaskError(FollowGPSWaypoints.UNKNOWN, msg)
+            self.setTaskError(FollowGPSWaypoints.Result.UNKNOWN, msg)
             self.error(msg)
             return None
 
@@ -424,7 +424,7 @@ def spin(
 
         if not self.goal_handle or not self.goal_handle.accepted:
             msg = 'Spin request was rejected!'
-            self.setTaskError(Spin.UNKNOWN, msg)
+            self.setTaskError(Spin.Result.UNKNOWN, msg)
             self.error(msg)
             return None
 
@@ -454,7 +454,7 @@ def backup(
 
         if not self.goal_handle or not self.goal_handle.accepted:
             msg = 'Backup request was rejected!'
-            self.setTaskError(BackUp.UNKNOWN, msg)
+            self.setTaskError(BackUp.Result.UNKNOWN, msg)
             self.error(msg)
             return None
 
@@ -484,7 +484,7 @@ def driveOnHeading(
 
         if not self.goal_handle or not self.goal_handle.accepted:
             msg = 'Drive On Heading request was rejected!'
-            self.setTaskError(DriveOnHeading.UNKNOWN, msg)
+            self.setTaskError(DriveOnHeading.Result.UNKNOWN, msg)
             self.error(msg)
             return None
 
@@ -510,7 +510,7 @@ def assistedTeleop(self, time_allowance: int = 30) -> Optional[RunningTask]:
 
         if not self.goal_handle or not self.goal_handle.accepted:
             msg = 'Assisted Teleop request was rejected!'
-            self.setTaskError(AssistedTeleop.UNKNOWN, msg)
+            self.setTaskError(AssistedTeleop.Result.UNKNOWN, msg)
             self.error(msg)
             return None
 
@@ -539,7 +539,7 @@ def followPath(self, path: Path, controller_id: str = '',
 
         if not self.goal_handle or not self.goal_handle.accepted:
             msg = 'FollowPath goal was rejected!'
-            self.setTaskError(FollowPath.UNKNOWN, msg)
+            self.setTaskError(FollowPath.Result.UNKNOWN, msg)
             self.error(msg)
             return None
 
@@ -568,7 +568,7 @@ def dockRobotByPose(self, dock_pose: PoseStamped,
 
         if not self.goal_handle or not self.goal_handle.accepted:
             msg = 'DockRobot request was rejected!'
-            self.setTaskError(DockRobot.UNKNOWN, msg)
+            self.setTaskError(DockRobot.Result.UNKNOWN, msg)
             self.error(msg)
             return None
 
@@ -595,7 +595,7 @@ def dockRobotByID(self, dock_id: str, nav_to_dock: bool = True) -> Optional[Runn
 
         if not self.goal_handle or not self.goal_handle.accepted:
             msg = 'DockRobot request was rejected!'
-            self.setTaskError(DockRobot.UNKNOWN, msg)
+            self.setTaskError(DockRobot.Result.UNKNOWN, msg)
             self.error(msg)
             return None
 
@@ -620,7 +620,7 @@ def undockRobot(self, dock_type: str = '') -> Optional[RunningTask]:
 
         if not self.goal_handle or not self.goal_handle.accepted:
             msg = 'UndockRobot request was rejected!'
-            self.setTaskError(UndockRobot.UNKNOWN, msg)
+            self.setTaskError(UndockRobot.Result.UNKNOWN, msg)
             self.error(msg)
             return None
 
@@ -761,7 +761,7 @@ def _getPathImpl(
             self.error('Get path was rejected!')
             self.status = GoalStatus.STATUS_UNKNOWN
             result = ComputePathToPose.Result()
-            result.error_code = ComputePathToPose.UNKNOWN
+            result.error_code = ComputePathToPose.Result.UNKNOWN
             result.error_msg = 'Get path was rejected'
             return result
 
@@ -809,7 +809,7 @@ def _getPathThroughPosesImpl(
         goal_msg.start = start
         goal_msg.goals.header.frame_id = 'map'
         goal_msg.goals.header.stamp = self.get_clock().now().to_msg()
-        goal_msg.goals.goals = goals
+        goal_msg.goals.goals = goals.goals
         goal_msg.planner_id = planner_id
         goal_msg.use_start = use_start
 
@@ -823,7 +823,7 @@ def _getPathThroughPosesImpl(
         if not self.goal_handle or not self.goal_handle.accepted:
             self.error('Get path was rejected!')
             result = ComputePathThroughPoses.Result()
-            result.error_code = ComputePathThroughPoses.UNKNOWN
+            result.error_code = ComputePathThroughPoses.Result.UNKNOWN
             result.error_msg = 'Get path was rejected!'
             return result
 
@@ -878,7 +878,7 @@ def _getRouteImpl(
         else:
             self.error('Invalid start and goal types. Must be PoseStamped for pose or int for ID')
             result = ComputeRoute.Result()
-            result.error_code = ComputeRoute.UNKNOWN
+            result.error_code = ComputeRoute.Result.UNKNOWN
             result.error_msg = 'Request type fields were invalid!'
             return result
 
@@ -890,7 +890,7 @@ def _getRouteImpl(
         if not self.goal_handle or not self.goal_handle.accepted:
             self.error('Get route was rejected!')
             result = ComputeRoute.Result()
-            result.error_code = ComputeRoute.UNKNOWN
+            result.error_code = ComputeRoute.Result.UNKNOWN
             result.error_msg = 'Get route was rejected!'
             return result
 
@@ -942,7 +942,8 @@ def getAndTrackRoute(
             goal_msg.goal = goal
             goal_msg.use_poses = True
         else:
-            self.setTaskError(ComputeAndTrackRoute.UNKNOWN, 'Request type fields were invalid!')
+            self.setTaskError(ComputeAndTrackRoute.Result.UNKNOWN,
+                              'Request type fields were invalid!')
             self.error('Invalid start and goal types. Must be PoseStamped for pose or int for ID')
             return None
 
@@ -954,7 +955,7 @@ def getAndTrackRoute(
 
         if not self.route_goal_handle or not self.route_goal_handle.accepted:
             msg = 'Compute and track route was rejected!'
-            self.setTaskError(ComputeAndTrackRoute.UNKNOWN, msg)
+            self.setTaskError(ComputeAndTrackRoute.Result.UNKNOWN, msg)
             self.error(msg)
             return None
 
@@ -988,7 +989,7 @@ def _smoothPathImpl(
         if not self.goal_handle or not self.goal_handle.accepted:
             self.error('Smooth path was rejected!')
             result = SmoothPath.Result()
-            result.error_code = SmoothPath.UNKNOWN
+            result.error_code = SmoothPath.Result.UNKNOWN
             result.error_msg = 'Smooth path was rejected'
             return result
 
@@ -1030,14 +1031,14 @@ def changeMap(self, map_filepath: str) -> bool:
             return False
 
         result = future_result.result
-        if result != LoadMap.Response().RESULT_SUCCESS:
-            if result == LoadMap.RESULT_MAP_DOES_NOT_EXIST:
+        if result != LoadMap.Response.RESULT_SUCCESS:
+            if result == LoadMap.Response.RESULT_MAP_DOES_NOT_EXIST:
                 reason = 'Map does not exist'
-            elif result == LoadMap.INVALID_MAP_DATA:
+            elif result == LoadMap.Response.RESULT_INVALID_MAP_DATA:
                 reason = 'Invalid map data'
-            elif result == LoadMap.INVALID_MAP_METADATA:
+            elif result == LoadMap.Response.RESULT_INVALID_MAP_METADATA:
                 reason = 'Invalid map metadata'
-            elif result == LoadMap.UNDEFINED_FAILURE:
+            elif result == LoadMap.Response.RESULT_UNDEFINED_FAILURE:
                 reason = 'Undefined failure'
             else:
                 reason = 'Unknown'
@@ -1144,7 +1145,7 @@ def clearGlobalCostmapAroundPose(self, pose: PoseStamped, reset_distance: float)
 
         return
 
-    def getGlobalCostmap(self) -> Optional[OccupancyGrid]:
+    def getGlobalCostmap(self) -> Optional[Costmap]:
         """Get the global costmap."""
         while not self.get_costmap_global_srv.wait_for_service(timeout_sec=1.0):
             self.info('Get global costmaps service not available, waiting...')
@@ -1159,7 +1160,7 @@ def getGlobalCostmap(self) -> Optional[OccupancyGrid]:
 
         return result.map
 
-    def getLocalCostmap(self) -> Optional[OccupancyGrid]:
+    def getLocalCostmap(self) -> Optional[Costmap]:
         """Get the local costmap."""
         while not self.get_costmap_local_srv.wait_for_service(timeout_sec=1.0):
             self.info('Get local costmaps service not available, waiting...')
@@ -1201,7 +1202,7 @@ def lifecycleStartup(self) -> None:
                 while not mgr_client.wait_for_service(timeout_sec=1.0):
                     self.info(f'{srv_name} service not available, waiting...')
                 req = ManageLifecycleNodes.Request()
-                req.command = ManageLifecycleNodes.Request().STARTUP
+                req.command = ManageLifecycleNodes.Request.STARTUP
                 future = mgr_client.call_async(req)
 
                 # starting up requires a full map->odom->base_link TF tree
@@ -1226,7 +1227,7 @@ def lifecycleShutdown(self) -> None:
                 while not mgr_client.wait_for_service(timeout_sec=1.0):
                     self.info(f'{srv_name} service not available, waiting...')
                 req = ManageLifecycleNodes.Request()
-                req.command = ManageLifecycleNodes.Request().SHUTDOWN
+                req.command = ManageLifecycleNodes.Request.SHUTDOWN
                 future = mgr_client.call_async(req)
                 rclpy.spin_until_future_complete(self, future)
                 future.result()
@@ -1268,12 +1269,13 @@ def _amclPoseCallback(self, msg: PoseWithCovarianceStamped) -> None:
         self.initial_pose_received = True
         return
 
-    def _feedbackCallback(self, msg: NavigateToPose.Feedback) -> None:
+    def _feedbackCallback(self, msg: Any) -> None:
         self.debug('Received action feedback message')
         self.feedback = msg.feedback
         return
 
-    def _routeFeedbackCallback(self, msg: ComputeAndTrackRoute.Feedback) -> None:
+    def _routeFeedbackCallback(
+            self, msg: ComputeAndTrackRoute.Impl.FeedbackMessage) -> None:
         self.debug('Received route action feedback message')
         self.route_feedback.append(msg.feedback)
         return

nav2_system_tests/src/behaviors/backup/backup_tester.py

@@ -299,7 +299,7 @@ def shutdown(self) -> None:
             self.info_msg(f'{transition_service} service not available, waiting...')
 
         req = ManageLifecycleNodes.Request()
-        req.command = ManageLifecycleNodes.Request().SHUTDOWN
+        req.command = ManageLifecycleNodes.Request.SHUTDOWN
         future = mgr_client.call_async(req)
         try:
             rclpy.spin_until_future_complete(self, future)

nav2_system_tests/src/route/tester_node.py

@@ -316,7 +316,8 @@ def runTrackRouteTest(self) -> bool:
         self.info_msg('Goal succeeded!')
         return True
 
-    def feedback_callback(self, feedback_msg: ComputeAndTrackRoute.Feedback) -> None:
+    def feedback_callback(
+            self, feedback_msg: ComputeAndTrackRoute.Impl.FeedbackMessage) -> None:
         self.feedback_msgs.append(feedback_msg.feedback)
 
     def distanceFromGoal(self) -> float:
@@ -386,7 +387,7 @@ def shutdown(self) -> None:
             self.info_msg(f'{transition_service} service not available, waiting...')
 
         req = ManageLifecycleNodes.Request()
-        req.command = ManageLifecycleNodes.Request().SHUTDOWN
+        req.command = ManageLifecycleNodes.Request.SHUTDOWN
         future = mgr_client.call_async(req)
         try:
             self.info_msg('Shutting down navigation lifecycle manager...')
@@ -401,7 +402,7 @@ def shutdown(self) -> None:
             self.info_msg(f'{transition_service} service not available, waiting...')
 
         req = ManageLifecycleNodes.Request()
-        req.command = ManageLifecycleNodes.Request().SHUTDOWN
+        req.command = ManageLifecycleNodes.Request.SHUTDOWN
         future = mgr_client.call_async(req)
         try:
             self.info_msg('Shutting down localization lifecycle manager...')