Update test_node Types (#1464)

* Update test_node types

Signed-off-by: Michael Carlstrom <[email protected]>

* don't attempt to use typevar defaults

Signed-off-by: Michael Carlstrom <[email protected]>

---------

Signed-off-by: Michael Carlstrom <[email protected]>

Author: InvincibleRMC

rclpy/docs/source/conf.py

@@ -28,6 +28,8 @@
 #
 import os
 import sys
+from typing import Dict
+from typing import List
 sys.path.insert(0, os.path.abspath('.'))
 
 
@@ -82,7 +84,7 @@
 # List of patterns, relative to source directory, that match files and
 # directories to ignore when looking for source files.
 # This pattern also affects html_static_path and html_extra_path.
-exclude_patterns = []
+exclude_patterns: List[str] = []
 
 # The name of the Pygments (syntax highlighting) style to use.
 pygments_style = None
@@ -104,7 +106,7 @@
 # Add any paths that contain custom static files (such as style sheets) here,
 # relative to this directory. They are copied after the builtin static files,
 # so a file named "default.css" will overwrite the builtin "default.css".
-html_static_path = []
+html_static_path: List[str] = []
 
 # Custom sidebar templates, must be a dictionary that maps document names
 # to template names.
@@ -125,7 +127,7 @@
 
 # -- Options for LaTeX output ------------------------------------------------
 
-latex_elements = {
+latex_elements: Dict[str, str] = {
     # The paper size ('letterpaper' or 'a4paper').
     #
     # 'papersize': 'letterpaper',

rclpy/rclpy/guard_condition.py

@@ -12,17 +12,23 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Callable, Optional
+from typing import Callable
+from typing import Coroutine
+from typing import Optional
+from typing import Union
 
 from rclpy.callback_groups import CallbackGroup
 from rclpy.context import Context
 from rclpy.impl.implementation_singleton import rclpy_implementation as _rclpy
 from rclpy.utilities import get_default_context
+from typing_extensions import TypeAlias
+
+GuardConditionCallbackType: TypeAlias = Callable[[], Union[Coroutine[None, None, None], None]]
 
 
 class GuardCondition:
 
-    def __init__(self, callback: Optional[Callable[[], None]],
+    def __init__(self, callback: Optional[GuardConditionCallbackType],
                  callback_group: Optional[CallbackGroup],
                  context: Optional[Context] = None) -> None:
         """

rclpy/rclpy/node.py

@@ -22,6 +22,7 @@
 from typing import Final
 from typing import Iterator
 from typing import List
+from typing import Literal
 from typing import Optional
 from typing import overload
 from typing import Sequence
@@ -66,6 +67,7 @@
 from rclpy.executors import Executor
 from rclpy.expand_topic_name import expand_topic_name
 from rclpy.guard_condition import GuardCondition
+from rclpy.guard_condition import GuardConditionCallbackType
 from rclpy.impl.implementation_singleton import rclpy_implementation as _rclpy
 from rclpy.impl.rcutils_logger import RcutilsLogger
 from rclpy.logging import get_logger
@@ -81,11 +83,13 @@
 from rclpy.qos_overriding_options import _declare_qos_parameters
 from rclpy.qos_overriding_options import QoSOverridingOptions
 from rclpy.service import Service
-from rclpy.subscription import MessageInfo
+from rclpy.subscription import GenericSubscriptionCallback
 from rclpy.subscription import Subscription
+from rclpy.subscription import SubscriptionCallbackUnion
 from rclpy.time_source import TimeSource
 from rclpy.timer import Rate
-from rclpy.timer import Timer, TimerInfo
+from rclpy.timer import Timer
+from rclpy.timer import TimerCallbackType
 from rclpy.topic_endpoint_info import TopicEndpointInfo
 from rclpy.type_description_service import TypeDescriptionService
 from rclpy.type_support import check_is_valid_msg_type
@@ -117,6 +121,8 @@
 # `Node.get_*_names_and_types_by_node` methods may raise this error.
 NodeNameNonExistentError: TypeAlias = _rclpy.NodeNameNonExistentError
 
+ParameterInput: TypeAlias = Union[AllowableParameterValue, Parameter.Type, ParameterValue]
+
 
 class Node:
     """
@@ -381,8 +387,7 @@ def get_logger(self) -> RcutilsLogger:
         return self._logger
 
     @overload
-    def declare_parameter(self, name: str, value: Union[AllowableParameterValueT,
-                                                        Parameter.Type, ParameterValue],
+    def declare_parameter(self, name: str, value: AllowableParameterValueT,
                           descriptor: Optional[ParameterDescriptor] = None,
                           ignore_override: bool = False
                           ) -> Parameter[AllowableParameterValueT]: ...
@@ -421,36 +426,19 @@ def declare_parameter(
         """
         if value is None and descriptor is None:
             # Temporal patch so we get deprecation warning if only a name is provided.
-            args: Union[Tuple[str], Tuple[str, Union[AllowableParameterValue,
-                                                     Parameter.Type, ParameterValue],
-                                          ParameterDescriptor]] = (name, )
+            args: Union[Tuple[str], Tuple[str, ParameterInput, ParameterDescriptor]] = (name, )
         else:
             descriptor = ParameterDescriptor() if descriptor is None else descriptor
             args = (name, value, descriptor)
         return self.declare_parameters('', [args], ignore_override)[0]
 
-    ParameterInput: TypeAlias = Union[AllowableParameterValue, Parameter.Type, ParameterValue]
-
-    @overload
     def declare_parameters(
         self,
         namespace: str,
         parameters: Sequence[Union[
-            Tuple[str, ParameterInput],
-            Tuple[str, ParameterInput, ParameterDescriptor],
-        ]],
-        ignore_override: bool = False
-    ) -> List[Parameter[Any]]: ...
-
-    def declare_parameters(
-        self,
-        namespace: str,
-        parameters: Union[Sequence[Union[
+            Tuple[str],
             Tuple[str, ParameterInput],
             Tuple[str, ParameterInput, ParameterDescriptor]]],
-                  Sequence[Union[
-            Tuple[str, ParameterInput],
-            Tuple[str, ParameterInput, ParameterDescriptor]]]],
         ignore_override: bool = False
     ) -> List[Parameter[Any]]:
         """
@@ -701,7 +689,7 @@ def get_parameter_type(self, name: str) -> Parameter.Type:
             and the parameter hadn't been declared beforehand.
         """
         if self.has_parameter(name):
-            return self._parameters[name].type_.value
+            return self._parameters[name].type_
         elif self._allow_undeclared_parameters:
             return Parameter.Type.NOT_SET
         else:
@@ -1643,11 +1631,39 @@ def create_publisher(
 
         return publisher
 
+    @overload
+    def create_subscription(
+        self,
+        msg_type: Type[MsgT],
+        topic: str,
+        callback: GenericSubscriptionCallback[bytes],
+        qos_profile: Union[QoSProfile, int],
+        *,
+        callback_group: Optional[CallbackGroup] = None,
+        event_callbacks: Optional[SubscriptionEventCallbacks] = None,
+        qos_overriding_options: Optional[QoSOverridingOptions] = None,
+        raw: Literal[True]
+    ) -> Subscription[MsgT]: ...
+
+    @overload
+    def create_subscription(
+        self,
+        msg_type: Type[MsgT],
+        topic: str,
+        callback: GenericSubscriptionCallback[MsgT],
+        qos_profile: Union[QoSProfile, int],
+        *,
+        callback_group: Optional[CallbackGroup] = None,
+        event_callbacks: Optional[SubscriptionEventCallbacks] = None,
+        qos_overriding_options: Optional[QoSOverridingOptions] = None,
+        raw: bool = False
+    ) -> Subscription[MsgT]: ...
+
     def create_subscription(
         self,
         msg_type: Type[MsgT],
         topic: str,
-        callback: Union[Callable[[MsgT], None], Callable[[MsgT, MessageInfo], None]],
+        callback: SubscriptionCallbackUnion[MsgT],
         qos_profile: Union[QoSProfile, int],
         *,
         callback_group: Optional[CallbackGroup] = None,
@@ -1810,7 +1826,7 @@ def create_service(
     def create_timer(
         self,
         timer_period_sec: float,
-        callback: Union[Callable[[], None], Callable[[TimerInfo], None], None],
+        callback: TimerCallbackType,
         callback_group: Optional[CallbackGroup] = None,
         clock: Optional[Clock] = None,
         autostart: bool = True,
@@ -1847,7 +1863,7 @@ def create_timer(
 
     def create_guard_condition(
         self,
-        callback: Callable[[], None],
+        callback: GuardConditionCallbackType,
         callback_group: Optional[CallbackGroup] = None
     ) -> GuardCondition:
         """

rclpy/rclpy/subscription.py

@@ -16,39 +16,100 @@
 from enum import Enum
 import inspect
 from types import TracebackType
-from typing import Callable, Generic, Optional, Type, TypedDict, TypeVar, Union
+from typing import Callable
+from typing import Generic
+from typing import Literal
+from typing import Optional
+from typing import overload
+from typing import Type
+from typing import TypedDict
+from typing import TypeVar
+from typing import Union
+
 
 from rclpy.callback_groups import CallbackGroup
 from rclpy.event_handler import SubscriptionEventCallbacks
 from rclpy.impl.implementation_singleton import rclpy_implementation as _rclpy
 from rclpy.qos import QoSProfile
 from rclpy.type_support import MsgT
+from typing_extensions import TypeAlias
+
+
+class PublisherGID(TypedDict):
+    implementation_identifier: str
+    data: bytes
 
 
 class MessageInfo(TypedDict):
     source_timestamp: int
     received_timestamp: int
     publication_sequence_number: Optional[int]
     reception_sequence_number: Optional[int]
-    publisher_gid: Optional[dict]
+    publisher_gid: Optional[PublisherGID]
 
 
 # Left to support Legacy TypeVars.
 MsgType = TypeVar('MsgType')
 
+# Can be redone with TypeVar(default=MsgT) when either typing-extensions4.11.0+ or python3.13+
+T = TypeVar('T')
+GenericSubscriptionCallback: TypeAlias = Union[Callable[[T], None],
+                                               Callable[[T, MessageInfo], None]]
+SubscriptionCallbackUnion: TypeAlias = Union[GenericSubscriptionCallback[MsgT],
+                                             GenericSubscriptionCallback[bytes]]
+
 
 class Subscription(Generic[MsgT]):
 
     class CallbackType(Enum):
         MessageOnly = 0
         WithMessageInfo = 1
 
+    @overload
+    def __init__(
+         self,
+         subscription_impl: '_rclpy.Subscription[MsgT]',
+         msg_type: Type[MsgT],
+         topic: str,
+         callback: GenericSubscriptionCallback[bytes],
+         callback_group: CallbackGroup,
+         qos_profile: QoSProfile,
+         raw: Literal[True],
+         event_callbacks: SubscriptionEventCallbacks,
+    ) -> None: ...
+
+    @overload
+    def __init__(
+         self,
+         subscription_impl: '_rclpy.Subscription[MsgT]',
+         msg_type: Type[MsgT],
+         topic: str,
+         callback: GenericSubscriptionCallback[MsgT],
+         callback_group: CallbackGroup,
+         qos_profile: QoSProfile,
+         raw: Literal[False],
+         event_callbacks: SubscriptionEventCallbacks,
+    ) -> None: ...
+
+    @overload
+    def __init__(
+         self,
+         subscription_impl: '_rclpy.Subscription[MsgT]',
+         msg_type: Type[MsgT],
+         topic: str,
+         callback: SubscriptionCallbackUnion[MsgT],
+         callback_group: CallbackGroup,
+         qos_profile: QoSProfile,
+         raw: bool,
+         event_callbacks: SubscriptionEventCallbacks,
+    ) -> None: ...
+
     def __init__(
          self,
          subscription_impl: '_rclpy.Subscription[MsgT]',
          msg_type: Type[MsgT],
          topic: str,
-         callback: Union[Callable[[MsgT], None], Callable[[MsgT, MessageInfo], None]],
+         callback: SubscriptionCallbackUnion[MsgT],
          callback_group: CallbackGroup,
          qos_profile: QoSProfile,
          raw: bool,
@@ -111,12 +172,11 @@ def topic_name(self) -> str:
             return self.__subscription.get_topic_name()
 
     @property
-    def callback(self) -> Union[Callable[[MsgT], None], Callable[[MsgT, MessageInfo], None]]:
+    def callback(self) -> SubscriptionCallbackUnion[MsgT]:
         return self._callback
 
     @callback.setter
-    def callback(self, value: Union[Callable[[MsgT], None],
-                                    Callable[[MsgT, MessageInfo], None]]) -> None:
+    def callback(self, value: SubscriptionCallbackUnion[MsgT]) -> None:
         self._callback = value
         self._callback_type = Subscription.CallbackType.MessageOnly
         try:

rclpy/rclpy/timer.py

@@ -16,6 +16,7 @@
 
 from types import TracebackType
 from typing import Callable
+from typing import Coroutine
 from typing import Optional
 from typing import Type
 from typing import Union
@@ -28,6 +29,7 @@
 from rclpy.impl.implementation_singleton import rclpy_implementation as _rclpy
 from rclpy.time import Time
 from rclpy.utilities import get_default_context
+from typing_extensions import TypeAlias
 
 
 class TimerInfo:
@@ -63,11 +65,17 @@ def actual_call_time(self) -> Time:
         return self._actual_call_time
 
 
+TimerCallbackType: TypeAlias = Union[Callable[[], None],
+                                     Callable[[TimerInfo], None],
+                                     Callable[[], Coroutine[None, None, None]],
+                                     None]
+
+
 class Timer:
 
     def __init__(
         self,
-        callback: Union[Callable[[], None], Callable[[TimerInfo], None], None],
+        callback: TimerCallbackType,
         callback_group: Optional[CallbackGroup],
         timer_period_ns: int,
         clock: Clock,

rclpy/test/__init__.py

@@ -26,7 +26,7 @@
 import test_rclpy  # noqa
 
 
-def _custom_import(name: str, package: Optional[str]) -> ModuleType:
+def _custom_import(name: str, package: Optional[str] = None) -> ModuleType:
     # Since Python 3.8, on Windows we should ensure DLL directories are explicitly added
     # to the search path.
     # See https://docs.python.org/3/whatsnew/3.8.html#bpo-36085-whatsnew