feat: allow passing ros2 types in send_message, service_call and start_action

docs/API_documentation/connectors/ROS_2_Connectors.md

@@ -33,13 +33,22 @@ The `ROS2Connector` is the main interface for publishing, subscribing, and calli
 ### Example Usage
 
 ```python
-from rai.communication.ros2.connectors import ROS2Connector
+from rai.communication.ros2.connectors import ROS2Connector, ROS2Message
+from std_msgs.msg import String
+from std_srvs.srv import SetBool
+from nav2_msgs.action import NavigateToPose
 
 connector = ROS2Connector()
 
-# Send a message to a topic
+# Send a raw ROS 2 message (msg_type is inferred)
 connector.send_message(
-    message=my_msg,  # ROS2Message
+    message=String(data="Hello"),
+    target="/my_topic"
+)
+
+# Send a message using a dictionary (msg_type is required, as a string or class)
+connector.send_message(
+    message=ROS2Message(payload={"data": "Hello"}),
     target="/my_topic",
     msg_type="std_msgs/msg/String"
 )
@@ -51,18 +60,32 @@ connector.register_callback(
     msg_type="std_msgs/msg/String"
 )
 
-# Call a service
+# Call a service with a request instance (msg_type is inferred)
+response = connector.service_call(
+    message=SetBool.Request(data=True),
+    target="/my_service"
+)
+
+# Call a service using a dictionary (msg_type is required)
 response = connector.service_call(
-    message=my_request_msg,
+    message=ROS2Message(payload={"data": True}),
     target="/my_service",
-    msg_type="my_package/srv/MyService"
+    msg_type=SetBool
+)
+
+# Start an action with a goal instance (msg_type is inferred)
+handle = connector.start_action(
+    action_data=NavigateToPose.Goal(),
+    target="/my_action",
+    on_feedback=feedback_cb,
+    on_done=done_cb
 )
 
-# Start an action
+# Start an action using a dictionary (msg_type is required)
 handle = connector.start_action(
-    action_data=my_goal_msg,
+    action_data=ROS2Message(payload={}),
     target="/my_action",
-    msg_type="my_package/action/MyAction",
+    msg_type="nav2_msgs/action/NavigateToPose",
     on_feedback=feedback_cb,
     on_done=done_cb
 )

src/rai_core/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "rai_core"
-version = "2.12.4"
+version = "2.13.0"
 description = "Core functionality for RAI framework"
 readme = "README.md"
 requires-python = ">=3.10,<3.13"

src/rai_core/rai/communication/ros2/api/action.py

@@ -33,7 +33,6 @@
 import rclpy.action
 import rclpy.node
 import rclpy.task
-import rosidl_runtime_py.set_message
 from action_msgs.srv import CancelGoal
 from rclpy.action import ActionClient, CancelResponse, GoalResponse
 from rclpy.action.client import ClientGoalHandle
@@ -55,7 +54,6 @@
     BaseROS2API,
     IROS2Message,
 )
-from rai.communication.ros2.api.conversion import import_message_from_str
 from rai.communication.ros2.ros_async import get_future_result
 
 
@@ -111,7 +109,7 @@ def _safe_callback_wrapper(
 
     def create_action_server(
         self,
-        action_type: str,
+        action_type: str | Type[Any],
         action_name: str,
         execute_callback: Callable[[ServerGoalHandle], Type[IROS2Message]],
         *,
@@ -164,7 +162,7 @@ def create_action_server(
         if result_timeout <= 0:
             raise ValueError(f"result_timeout must be positive, got {result_timeout!r}")
         handle = self._generate_handle()
-        action_ros_type = import_message_from_str(action_type)
+        action_ros_type = self.resolve_interface_type(action_type)
         try:
             action_server = ActionServer(
                 node=self.node,
@@ -202,8 +200,8 @@ def create_action_server(
     def send_goal(
         self,
         action_name: str,
-        action_type: str,
-        goal: Dict[str, Any],
+        action_type: str | Type[Any] | None = None,
+        goal: IROS2Message | Dict[str, Any] | None = None,
         *,
         feedback_callback: Callable[[Any], None] = lambda _: None,
         done_callback: Callable[
@@ -220,11 +218,7 @@ def send_goal(
             feedbacks=[],
         )
 
-        action_cls = import_message_from_str(action_type)
-        action_goal = action_cls.Goal()  # type: ignore
-        rosidl_runtime_py.set_message.set_message_fields(
-            action_goal, copy.deepcopy(goal)
-        )
+        action_goal, action_cls = self.resolve_content(goal, action_type, "Goal")
 
         action_client = ActionClient(self.node, action_cls, action_name)
         if not action_client.wait_for_server(timeout_sec=timeout_sec):  # type: ignore

src/rai_core/rai/communication/ros2/api/base.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import copy
+import importlib
 import logging
 from typing import (
     Any,
@@ -36,7 +38,12 @@
 from rclpy.topic_endpoint_info import TopicEndpointInfo
 from rosidl_parser.definition import NamespacedType
 from rosidl_runtime_py.import_message import import_message_from_namespaced_type
-from rosidl_runtime_py.utilities import get_namespaced_type
+from rosidl_runtime_py.utilities import (
+    get_namespaced_type,
+    is_action,
+    is_message,
+    is_service,
+)
 
 from rai.communication.ros2.api.conversion import import_message_from_str
 
@@ -126,6 +133,67 @@ def import_message_from_str(msg_type: str) -> Type[object]:
         msg_namespaced_type: NamespacedType = get_namespaced_type(msg_type)
         return import_message_from_namespaced_type(msg_namespaced_type)
 
+    @staticmethod
+    def resolve_interface_type(interface_type: str | Type[Any]) -> Type[Any]:
+        """Return the interface class for a type string like 'std_srvs/srv/SetBool' or the class itself."""
+        if isinstance(interface_type, str):
+            return import_message_from_str(interface_type)
+        return interface_type
+
+    @staticmethod
+    def get_interface_type(instance: IROS2Message) -> Type[Any]:
+        """Return the interface class an instance belongs to, e.g. SetBool for SetBool.Request()."""
+        cls = type(instance)
+        package = importlib.import_module(cls.__module__.rsplit(".", 1)[0])
+        return getattr(package, cls.__name__.partition("_")[0])
+
+    @classmethod
+    def resolve_content(
+        cls,
+        content: IROS2Message | Dict[str, Any],
+        interface_type: str | Type[Any] | None,
+        member: str | None = None,
+    ) -> Tuple[IROS2Message, Type[Any]]:
+        """Resolve a dictionary or ROS 2 instance into (instance, interface class).
+
+        Args:
+            content: ROS 2 instance or dictionary of field values.
+            interface_type: Interface type string or class. Required for dictionaries,
+                validated against the instance otherwise.
+            member: Nested interface class the content must be an instance of,
+                e.g. 'Request' for services or 'Goal' for actions.
+
+        Raises:
+            ValueError: If content is neither a dictionary nor a ROS 2 instance,
+                if a dictionary is given without interface_type, or if the instance
+                does not match interface_type or member.
+        """
+        if isinstance(content, dict):
+            if interface_type is None:
+                raise ValueError("Interface type must be provided if content is a dict")
+            interface_cls = cls.resolve_interface_type(interface_type)
+            instance_cls = getattr(interface_cls, member) if member else interface_cls
+            instance = instance_cls()
+            # set_message_fields mutates nested lists, see ros2/rosidl_runtime_py#33
+            rosidl_runtime_py.set_message.set_message_fields(
+                instance, copy.deepcopy(content)
+            )
+            return instance, interface_cls
+        if isinstance(content, type) or not is_message(content):
+            raise ValueError(f"Invalid content type: {type(content)}")
+        interface_cls = cls.get_interface_type(content) if member else type(content)
+        instance_cls = getattr(interface_cls, member) if member else interface_cls
+        if type(content) is not instance_cls:
+            raise ValueError(f"Expected {instance_cls}, got {type(content)}")
+        if (
+            interface_type is not None
+            and cls.resolve_interface_type(interface_type) is not interface_cls
+        ):
+            raise ValueError(
+                f"Interface type {interface_type} does not match {type(content)}"
+            )
+        return content, interface_cls
+
     def get_topic_type(self, topic: str) -> str:
         names_and_types = self.node.get_topic_names_and_types(no_demangle=False)
         for name, types in names_and_types:
@@ -134,3 +202,15 @@ def get_topic_type(self, topic: str) -> str:
                     raise ValueError(f"Topic {topic} has multiple types: {types}")
                 return types[0]
         raise ValueError(f"Topic {topic} not found")
+
+    @staticmethod
+    def is_ros2_message(msg: Any) -> bool:
+        return is_message(msg)
+
+    @staticmethod
+    def is_ros2_service(msg: Any) -> bool:
+        return is_service(msg)
+
+    @staticmethod
+    def is_ros2_action(msg: Any) -> bool:
+        return is_action(msg)

src/rai_core/rai/communication/ros2/api/service.py

@@ -21,6 +21,7 @@
     Dict,
     List,
     Tuple,
+    Type,
 )
 
 import rclpy
@@ -36,8 +37,8 @@
 
 from rai.communication.ros2.api.base import (
     BaseROS2API,
+    IROS2Message,
 )
-from rai.communication.ros2.api.conversion import import_message_from_str
 
 
 class ROS2ServiceAPI(BaseROS2API):
@@ -57,8 +58,8 @@ def release_client(self, service_name: str) -> bool:
     def call_service(
         self,
         service_name: str,
-        service_type: str,
-        request: Any,
+        service_type: str | Type[Any] | None = None,
+        request: IROS2Message | Dict[str, Any] | None = None,
         timeout_sec: float = 5.0,
         *,
         reuse_client: bool = True,
@@ -68,8 +69,9 @@ def call_service(
 
         Args:
             service_name: Fully-qualified service name.
-            service_type: ROS 2 service type string (e.g., 'std_srvs/srv/SetBool').
-            request: Request payload dict.
+            service_type: ROS 2 service type string (e.g., 'std_srvs/srv/SetBool') or class.
+                Required when request is a dict, inferred from the instance otherwise.
+            request: Request payload dict or request instance (e.g., SetBool.Request()).
             timeout_sec: Seconds to wait for availability/response.
             reuse_client: Reuse a cached client. Client creation is synchronized; set
                 False to create a new client per call.
@@ -78,7 +80,8 @@ def call_service(
             Response message instance.
 
         Raises:
-            ValueError: Service not available within the timeout.
+            ValueError: Service not available within the timeout, request is a dict
+                without service_type, or request does not match service_type.
             AttributeError: Service type or request cannot be constructed.
 
         Note:
@@ -87,7 +90,7 @@ def call_service(
             through the same client. Use reuse_client=False for per-call clients
             when concurrent service calls are required.
         """
-        srv_msg, srv_cls = self.build_ros2_service_request(service_type, request)
+        srv_msg, srv_cls = self.resolve_content(request, service_type, "Request")
 
         def _call_service(client: Client, timeout_sec: float) -> Any:
             is_service_available = client.wait_for_service(timeout_sec=timeout_sec)
@@ -118,11 +121,11 @@ def get_service_names_and_types(self) -> List[Tuple[str, List[str]]]:
     def create_service(
         self,
         service_name: str,
-        service_type: str,
+        service_type: str | Type[Any],
         callback: Callable[[Any, Any], Any],
         **kwargs,
     ) -> str:
-        srv_cls = import_message_from_str(service_type)
+        srv_cls = self.resolve_interface_type(service_type)
         service = self.node.create_service(srv_cls, service_name, callback, **kwargs)
         handle = str(uuid.uuid4())
         self._services[handle] = service

src/rai_core/rai/communication/ros2/api/topic.py

@@ -34,6 +34,7 @@
 
 from rai.communication.ros2.api.base import (
     BaseROS2API,
+    IROS2Message,
 )
 from rai.communication.ros2.api.conversion import import_message_from_str
 
@@ -140,8 +141,8 @@ def get_topic_names_and_types(
     def publish(
         self,
         topic: str,
-        msg_content: Dict[str, Any],
-        msg_type: str,
+        msg_content: IROS2Message | Dict[str, Any],
+        msg_type: str | Type[Any] | None = None,
         *,
         auto_qos_matching: bool = True,
         qos_profile: Optional[QoSProfile] = None,
@@ -150,20 +151,22 @@ def publish(
 
         Args:
             topic: Name of the topic to publish to
-            msg_content: Dictionary containing the message content
-            msg_type: ROS2 message type as string (e.g. 'std_msgs/msg/String')
+            msg_content: ROS2 message instance or dictionary containing the message content
+            msg_type: ROS2 message type as string (e.g. 'std_msgs/msg/String') or class,
+                required when msg_content is a dictionary
             auto_qos_matching: Whether to automatically match QoS with subscribers
             qos_profile: Optional custom QoS profile to use
 
         Raises:
-            ValueError: If neither auto_qos_matching is True nor qos_profile is provided
+            ValueError: If neither auto_qos_matching is True nor qos_profile is provided,
+                if msg_content is a dictionary without msg_type, or if msg_type does not
+                match the ROS2 message instance
         """
         qos_profile = self._resolve_qos_profile(
             topic, auto_qos_matching, qos_profile, for_publisher=True
         )
-
-        msg = self.build_ros2_msg(msg_type, msg_content)
-        publisher = self._get_or_create_publisher(topic, type(msg), qos_profile)
+        msg, msg_cls = self.resolve_content(msg_content, msg_type)
+        publisher = self._get_or_create_publisher(topic, msg_cls, qos_profile)
         publisher.publish(msg)
 
     def _verify_receive_args(

src/rai_core/rai/communication/ros2/connectors/action_mixin.py

@@ -12,9 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Any, Callable, Optional
+from typing import Any, Callable, Optional, Type
 
-from rai.communication.ros2.api import ROS2ActionAPI
+from rai.communication.ros2.api import IROS2Message, ROS2ActionAPI
 from rai.communication.ros2.messages import ROS2HRIMessage, ROS2Message
 
 
@@ -32,21 +32,22 @@ def __post_init__(self, *args: Any, **kwargs: Any) -> None:
 
     def start_action(
         self,
-        action_data: Optional[ROS2Message | ROS2HRIMessage],
+        action_data: Optional[ROS2Message | ROS2HRIMessage | IROS2Message],
         target: str,
         on_feedback: Callable[[Any], None] = lambda _: None,
         on_done: Callable[[Any], None] = lambda _: None,
         timeout_sec: float = 1.0,
         *,
-        msg_type: str,
+        msg_type: str | Type[Any] | None = None,
         **kwargs: Any,
     ) -> str:
-        if not isinstance(action_data, ROS2Message):
-            raise ValueError("Action data must be of type ROS2Message")
+        goal = (
+            action_data.payload if isinstance(action_data, ROS2Message) else action_data
+        )
         accepted, handle = self._actions_api.send_goal(
             action_name=target,
             action_type=msg_type,
-            goal=action_data.payload,
+            goal=goal,
             timeout_sec=timeout_sec,
             feedback_callback=on_feedback,
             done_callback=on_done,