fix: ament_mypy errors (#1039)

Author: bjsowa

rosapi/src/rosapi/params.py

@@ -33,6 +33,7 @@
 
 import fnmatch
 from json import dumps, loads
+from typing import TYPE_CHECKING
 
 from rcl_interfaces.msg import Parameter, ParameterType, ParameterValue
 from rcl_interfaces.srv import GetParameters, ListParameters, SetParameters
@@ -44,6 +45,9 @@
 from rosapi.async_helper import futures_wait_for
 from rosapi.proxy import get_nodes
 
+if TYPE_CHECKING:
+    from rclpy.client import Client
+
 """ Methods to interact with the param server.  Values have to be passed
 as JSON in order to facilitate dynamically typed SRV messages """
 
@@ -120,7 +124,7 @@ async def _set_param(node_name: str, name: str, value: str, parameter_type=None)
             setattr(parameter.value, _parameter_type_mapping[parameter_type], loads(value))
 
     assert _node is not None
-    client = _node.create_client(
+    client: Client = _node.create_client(
         SetParameters,
         f"{node_name}/set_parameters",
         callback_group=MutuallyExclusiveCallbackGroup(),
@@ -175,7 +179,7 @@ async def _get_param(node_name: str, name: str) -> ParameterValue:
     """Internal helper function for get_param"""
 
     assert _node is not None
-    client = _node.create_client(
+    client: Client = _node.create_client(
         GetParameters,
         f"{node_name}/get_parameters",
         callback_group=MutuallyExclusiveCallbackGroup(),
@@ -250,7 +254,7 @@ async def get_param_names(params_glob: str | None) -> list[str]:
         if node_name == _node.get_fully_qualified_name():
             continue
 
-        client = _node.create_client(
+        client: Client = _node.create_client(
             ListParameters,
             f"{node_name}/list_parameters",
             callback_group=MutuallyExclusiveCallbackGroup(),

rosbridge_library/src/rosbridge_library/internal/actions.py

@@ -167,7 +167,7 @@ def send_goal(
         args_to_action_goal_instance(action_name, inst, args)
 
         self.result = None
-        client = ActionClient(node_handle, action_class, action_name)
+        client: ActionClient = ActionClient(node_handle, action_class, action_name)
         client.wait_for_server(timeout_sec=self.server_timeout_time)
         send_goal_future = client.send_goal_async(inst, feedback_callback=feedback_cb)
         send_goal_future.add_done_callback(self.goal_response_cb)

rosbridge_library/src/rosbridge_library/internal/services.py

@@ -31,7 +31,7 @@
 # POSSIBILITY OF SUCH DAMAGE.
 
 from threading import Event, Thread
-from typing import Any, Callable, Optional
+from typing import TYPE_CHECKING, Any, Callable, Optional
 
 from rclpy.callback_groups import ReentrantCallbackGroup
 from rclpy.expand_topic_name import expand_topic_name
@@ -45,6 +45,9 @@
     get_service_request_instance,
 )
 
+if TYPE_CHECKING:
+    from rclpy.client import Client
+
 
 class InvalidServiceException(Exception):
     def __init__(self, service_name) -> None:
@@ -146,7 +149,7 @@ def call_service(
     # Populate the instance with the provided args
     args_to_service_request_instance(service, inst, args)
 
-    client = node_handle.create_client(
+    client: Client = node_handle.create_client(
         service_class, service, callback_group=ReentrantCallbackGroup()
     )
 

rosbridge_server/test/websocket/advertise_action.test.py

@@ -46,7 +46,7 @@ async def test_two_concurrent_calls(self, node: Node, make_client):
                 "type": "example_interfaces/Fibonacci",
             }
         )
-        client = ActionClient(node, Fibonacci, "/test_fibonacci_action")
+        client: ActionClient = ActionClient(node, Fibonacci, "/test_fibonacci_action")
         client.wait_for_server()
 
         requests_future, ws_client.message_handler = expect_messages(

rosbridge_server/test/websocket/advertise_action_feedback.test.py

@@ -43,7 +43,7 @@ async def test_feedback(self, node: Node, make_client):
                 "type": "example_interfaces/Fibonacci",
             }
         )
-        client = ActionClient(node, Fibonacci, "/test_fibonacci_action")
+        client: ActionClient = ActionClient(node, Fibonacci, "/test_fibonacci_action")
         client.wait_for_server()
 
         requests_future, ws_client.message_handler = expect_messages(

rosbridge_server/test/websocket/advertise_service.test.py

@@ -2,11 +2,15 @@
 import os
 import sys
 import unittest
+from typing import TYPE_CHECKING
 
 from rclpy.node import Node
 from std_srvs.srv import SetBool
 from twisted.python import log
 
+if TYPE_CHECKING:
+    from rclpy.client import Client
+
 sys.path.append(os.path.dirname(__file__))  # enable importing from common.py in this directory
 
 import common  # noqa: E402
@@ -28,7 +32,7 @@ async def test_two_concurrent_calls(self, node: Node, make_client):
                 "service": "/test_service",
             }
         )
-        client = node.create_client(SetBool, "/test_service")
+        client: Client = node.create_client(SetBool, "/test_service")
         client.wait_for_service()
 
         requests_future, ws_client.message_handler = expect_messages(

rosbridge_server/test/websocket/advertise_service_duplicate.test.py

@@ -1,11 +1,15 @@
 import os
 import sys
 import unittest
+from typing import TYPE_CHECKING
 
 from rclpy.node import Node
 from std_srvs.srv import SetBool
 from twisted.python import log
 
+if TYPE_CHECKING:
+    from rclpy.client import Client
+
 sys.path.append(os.path.dirname(__file__))  # enable importing from common.py in this directory
 
 import common  # noqa: E402
@@ -27,7 +31,7 @@ async def test_double_advertise(self, node: Node, make_client):
                 "service": "/test_service",
             }
         )
-        client = node.create_client(SetBool, "/test_service")
+        client: Client = node.create_client(SetBool, "/test_service")
         client.wait_for_service()
 
         requests1_future, ws_client1.message_handler = expect_messages(

rosbridge_server/test/websocket/common.py

@@ -1,6 +1,6 @@
 import functools
 import json
-from typing import Any, Awaitable, Callable
+from typing import TYPE_CHECKING, Any, Awaitable, Callable
 
 import launch
 import launch_ros
@@ -15,6 +15,9 @@
 from twisted.internet import reactor
 from twisted.internet.endpoints import TCP4ClientEndpoint
 
+if TYPE_CHECKING:
+    from rclpy.client import Client
+
 
 class TestClientProtocol(WebSocketClientProtocol):
     """
@@ -83,7 +86,7 @@ async def get_server_port(node: Node) -> int:
     """
     Returns the port which the WebSocket server is running on
     """
-    client = node.create_client(GetParameters, "/rosbridge_websocket/get_parameters")
+    client: Client = node.create_client(GetParameters, "/rosbridge_websocket/get_parameters")
     try:
         if not client.wait_for_service(5):
             raise RuntimeError("GetParameters service not available")