Add ament_mypy

rclpy/CMakeLists.txt

@@ -157,8 +157,12 @@ if(BUILD_TESTING)
     )
   endif()
 
+  list(APPEND AMENT_LINT_AUTO_EXCLUDE ament_cmake_mypy)
   ament_lint_auto_find_test_dependencies()
 
+  find_package(ament_cmake_mypy REQUIRED)
+  ament_mypy(AMENT_STRICT)
+
   find_package(ament_cmake_cppcheck REQUIRED)
   ament_cppcheck()
   set_tests_properties(cppcheck PROPERTIES TIMEOUT 420)

rclpy/package.xml

@@ -62,6 +62,7 @@
   <test_depend>ament_cmake_pytest</test_depend>
   <test_depend>ament_lint_auto</test_depend>
   <test_depend>ament_lint_common</test_depend>
+  <test_depend>ament_cmake_mypy</test_depend>
   <test_depend>python3-pytest</test_depend>
   <test_depend>rosidl_generator_py</test_depend>
   <test_depend>test_msgs</test_depend>

rclpy/rclpy/endpoint_info.py

@@ -234,7 +234,7 @@ def __init__(
         node_namespace: str = '',
         service_type: str = '',
         service_type_hash: Union[TypeHash, TypeHashDictionary] = TypeHash(),
-        qos_profiles: Sequence[Union[QoSProfile, '_rclpy._rmw_qos_profile_dict']] = [],
+        qos_profiles: list[QoSProfile] | list['_rclpy._rmw_qos_profile_dict'] = [],
         endpoint_gids: list[list[int]] = [],
         endpoint_type: Union[EndpointTypeEnum, int] = EndpointTypeEnum.INVALID,
         endpoint_count: int = 0
@@ -368,8 +368,8 @@ def qos_profiles(self) -> Annotated[Any, list[QoSProfile]]:
         return self._qos_profiles
 
     @qos_profiles.setter
-    def qos_profiles(self, value: list[Union[QoSProfile, '_rclpy._rmw_qos_profile_dict']]) -> None:
-        assert isinstance(value, list)
+    def qos_profiles(self, value: Sequence[Union[QoSProfile,
+                                                 '_rclpy._rmw_qos_profile_dict']]) -> None:
         self._qos_profiles = [v if isinstance(v, QoSProfile) else QoSProfile(**v) for v in value]
 
     def __eq__(self, other: object) -> bool:

rclpy/rclpy/executors.py

@@ -560,7 +560,7 @@ def _take_subscription(self, sub: Subscription[Any]
                     msg_tuple = msg_info
 
                 async def _execute() -> None:
-                    await await_or_execute(sub.callback, *msg_tuple)
+                    await await_or_execute(sub.callback, *msg_tuple)  # type: ignore [arg-type]
 
                 return _execute
         except InvalidHandle:

rclpy/rclpy/experimental/async_node.py

@@ -140,7 +140,7 @@ async def __aenter__(self) -> 'AsyncNode':
         self._tg = await tg.__aenter__()
         for entity in self._entities:
             if hasattr(entity, '_run'):
-                entity._task = self._tg.create_task(entity._run())
+                entity._task = self._tg.create_task(entity._run())  # type: ignore[union-attr]
         return self
 
     async def __aexit__(
@@ -221,16 +221,16 @@ def create_publisher(
         """
         if self._destroyed.is_set():
             raise RuntimeError('Cannot create publisher on a destroyed node')
-        qos_profile = self._validate_qos_or_depth_parameter(qos_profile)
+        qos_profile_validated = self._validate_qos_or_depth_parameter(qos_profile)
 
         publisher_handle = self._create_publisher_handle(
-            msg_type, topic, qos_profile)
+            msg_type, topic, qos_profile_validated)
 
         pub = AsyncPublisher(
             publisher_handle,
             msg_type,
             topic,
-            qos_profile,
+            qos_profile_validated,
             on_destroy=self._entities.discard,
         )
         self._entities.add(pub)

rclpy/rclpy/impl/_rclpy_pybind11.pyi

@@ -618,7 +618,8 @@ class Subscription(Destroyable, Generic[MsgT]):
 
     def __init__(self, node: Node, pymsg_type: type[MsgT], topic: str,
                  pyqos_profile: rmw_qos_profile_t,
-                 content_filter_options: Optional[ContentFilterOptions] = None) -> None: ...
+                 content_filter_options: Optional[ContentFilterOptions] = None,
+                 acceptable_buffer_backends: str | None = None) -> None: ...
 
     @property
     def pointer(self) -> int:
@@ -1077,6 +1078,13 @@ class rmw_incompatible_type_status_t:
     def total_count_change(self) -> int: ...
 
 
+def publisher_event_type_is_supported(event_type: rcl_publisher_event_type_t) -> bool:
+    """Check if a publisher event type is supported by the active RMW implementation."""
+
+def subscription_event_type_is_supported(event_type: rcl_subscription_event_type_t) -> bool:
+    """Check if a subscription event type is supported by the active RMW implementation."""
+
+
 def rclpy_get_rmw_implementation_identifier() -> str:
     """Retrieve the identifier for the active RMW implementation."""
 

rclpy/rclpy/node.py

@@ -494,8 +494,6 @@ def declare_parameters(
                                  'being included in self._parameter_overrides, and ',
                                  'ignore_override=False')
 
-            from typing import cast
-            value = cast(AllowableParameterValue, value)
             parameter_list.append(Parameter(name, value=value))
             descriptors.update({name: descriptor})
 
@@ -1927,7 +1925,7 @@ def _create_publisher_handle(
         self,
         msg_type: Type[MsgT],
         topic: str,
-        qos_profile: Union[QoSProfile, int],
+        qos_profile: QoSProfile,
         *,
         qos_overriding_options: Optional[QoSOverridingOptions] = None,
     ) -> '_rclpy.Publisher[MsgT]':
@@ -2254,20 +2252,20 @@ def create_publisher(
         :param event_callbacks: User-defined callbacks for middleware events.
         :return: The new publisher.
         """
-        qos_profile = self._validate_qos_or_depth_parameter(qos_profile)
+        qos_profile_validated = self._validate_qos_or_depth_parameter(qos_profile)
 
         callback_group = callback_group or self.default_callback_group
 
         publisher_object = self._create_publisher_handle(
             msg_type,
             topic,
-            qos_profile,
+            qos_profile_validated,
             qos_overriding_options=qos_overriding_options
         )
 
         try:
             publisher = publisher_class(
-                publisher_object, msg_type, topic, qos_profile,
+                publisher_object, msg_type, topic, qos_profile_validated,
                 on_destroy=self._on_destroy_publisher,
                 event_callbacks=event_callbacks or PublisherEventCallbacks(),
                 callback_group=callback_group)

rclpy/rclpy/parameter_service.py

@@ -97,7 +97,7 @@ def _get_parameters_callback(
                 param = node.get_parameter(name)
             except (ParameterNotDeclaredException, ParameterUninitializedException) as ex:
                 node.get_logger().warning(f'Failed to get parameters: {ex}')
-                response.values = node.get_parameters([])
+                response.values = []
                 return response
             response.values.append(param.get_parameter_value())
         return response

rclpy/rclpy/task.py

@@ -12,12 +12,21 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from collections.abc import Callable
+from collections.abc import Generator
 from enum import Enum
 import inspect
 import sys
 import threading
-from typing import (Any, Callable, cast, Coroutine, Dict, Generator, Generic, List,
-                    Optional, overload, Tuple, TYPE_CHECKING, TypeVar, Union)
+from typing import Any
+from typing import Coroutine
+from typing import Generic
+from typing import Optional
+from typing import overload
+from typing import TYPE_CHECKING
+from typing import TypeVar
+from typing import Union
+
 import warnings
 import weakref
 
@@ -51,7 +60,7 @@ def __init__(self, *, executor: Optional['Executor'] = None) -> None:
         self._exception: Optional[Exception] = None
         self._exception_fetched = False
         # callbacks or tasks to be scheduled after this task completes
-        self._callbacks: List[Union[Callable[['Future[T]'], None], 'Task[Any]']] = []
+        self._callbacks: list[Union[Callable[['Future[T]'], None], 'Task[Any]']] = []
         # Lock for threadsafety
         self._lock = threading.Lock()
         # An executor to use when scheduling done callbacks
@@ -260,42 +269,41 @@ class Task(Future[T]):
     @overload
     def __init__(self,
                  handler: Callable[..., Coroutine[Any, Any, T]],
-                 args: Optional[Tuple[Any, ...]] = None,
-                 kwargs: Optional[Dict[str, Any]] = None,
+                 args: Optional[tuple[Any, ...]] = None,
+                 kwargs: Optional[dict[str, Any]] = None,
                  executor: Optional['Executor'] = None) -> None: ...
 
     @overload
     def __init__(self,
                  handler: Callable[..., T],
-                 args: Optional[Tuple[Any, ...]] = None,
-                 kwargs: Optional[Dict[str, Any]] = None,
+                 args: Optional[tuple[Any, ...]] = None,
+                 kwargs: Optional[dict[str, Any]] = None,
                  executor: Optional['Executor'] = None) -> None: ...
 
     def __init__(self,
                  handler: Callable[..., Any],
-                 args: Optional[Tuple[Any, ...]] = None,
-                 kwargs: Optional[Dict[str, Any]] = None,
+                 args: Optional[tuple[Any, ...]] = None,
+                 kwargs: Optional[dict[str, Any]] = None,
                  executor: Optional['Executor'] = None) -> None:
         super().__init__(executor=executor)
         # Arguments passed into the function
         if args is None:
             args = ()
-        self._args: Optional[Tuple[Any, ...]] = args
+        self._args: Optional[tuple[Any, ...]] = args
         if kwargs is None:
             kwargs = {}
-        self._kwargs: Optional[Dict[str, Any]] = kwargs
+        self._kwargs: Optional[dict[str, Any]] = kwargs
 
         # _handler is either a normal function or a coroutine
         if inspect.iscoroutinefunction(handler):
             self._handler: Union[
                 Coroutine[Any, Any, T],
-                Callable[[], T],
+                Callable[..., T],
                 None
-             ] = cast(Coroutine[Any, Any, T], handler(*args, **kwargs))
+             ] = handler(*args, **kwargs)
             self._args = None
             self._kwargs = None
         else:
-            handler = cast(Callable[[], T], handler)
             self._handler = handler
         # True while the task is being executed
         self._executing = False
@@ -328,6 +336,7 @@ def __call__(self) -> None:
                 # Execute a normal function
                 try:
                     assert self._handler is not None and callable(self._handler)
+                    assert self._args is not None and self._kwargs is not None
                     self.set_result(self._handler(*self._args, **self._kwargs))
                 except Exception as e:
                     self.set_exception(e)

rclpy/test/test_events_executor.py

@@ -697,7 +697,7 @@ def test_timers(self) -> None:
         # Only one of the callbacks should be delivered, though we can't necessarily predict which
         # one.
         def handler() -> None:
-            nonlocal count
+            nonlocal count  # type: ignore[misc]
             count += 1
             timer1.cancel()
             timer2.cancel()