feat: allow passing ros2 types in send_message

docs/API_documentation/connectors/ROS_2_Connectors.md

@@ -33,13 +33,20 @@ 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
 
 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)
+connector.send_message(
+    message=ROS2Message(payload={"data": "Hello"}),
     target="/my_topic",
     msg_type="std_msgs/msg/String"
 )

src/rai_core/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "rai_core"
-version = "2.6.0"
+version = "2.7.0"
 description = "Core functionality for RAI framework"
 authors = ["Maciej Majek <maciej.majek@robotec.ai>", "Bartłomiej Boczek <bartlomiej.boczek@robotec.ai>", "Kajetan Rachwał <kajetan.rachwal@robotec.ai>"]
 readme = "README.md"

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

@@ -36,7 +36,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
 
@@ -134,3 +139,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/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 | None = None,
         *,
         auto_qos_matching: bool = True,
         qos_profile: Optional[QoSProfile] = None,
@@ -162,7 +163,14 @@ def publish(
             topic, auto_qos_matching, qos_profile, for_publisher=True
         )
 
-        msg = self.build_ros2_msg(msg_type, msg_content)
+        if self.is_ros2_message(msg_content):
+            msg = msg_content
+        elif isinstance(msg_content, dict) and msg_type is not None:
+            msg = self.build_ros2_msg(msg_type, msg_content)
+        elif isinstance(msg_content, dict) and msg_type is None:
+            raise ValueError("msg_type must be provided if msg_content is a dict")
+        else:
+            raise ValueError(f"Invalid message content type: {type(msg_content)}")
         publisher = self._get_or_create_publisher(topic, type(msg), qos_profile)
         publisher.publish(msg)
 

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

@@ -16,7 +16,17 @@
 import time
 import uuid
 from functools import partial
-from typing import Any, Callable, Dict, Final, List, Literal, Optional, Tuple, TypeVar
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Final,
+    List,
+    Literal,
+    Optional,
+    Tuple,
+    TypeVar,
+)
 
 import rclpy
 import rclpy.executors
@@ -31,6 +41,7 @@
 
 from rai.communication import BaseConnector
 from rai.communication.ros2.api import (
+    IROS2Message,
     ROS2ActionAPI,
     ROS2ServiceAPI,
     ROS2TopicAPI,
@@ -240,10 +251,10 @@ def get_actions_names_and_types(self) -> List[Tuple[str, List[str]]]:
 
     def send_message(
         self,
-        message: T,
+        message: T | IROS2Message,
         target: str,
         *,
-        msg_type: str,
+        msg_type: str | None = None,
         auto_qos_matching: bool = True,
         qos_profile: Optional[QoSProfile] = None,
         **kwargs: Any,
@@ -252,22 +263,26 @@ def send_message(
 
         Parameters
         ----------
-        message : T
-            The message to send.
+        message : T | IROS2Message
+            The message to send. Can be a subclass of ROS2Message (payload is a dict) or any ROS2 message.
         target : str
             The target topic name.
-        msg_type : str
-            The ROS2 message type.
+        msg_type : str | None, optional
+            The ROS2 message type. If None, the message type will be inferred from the message content. Must be provided if msg_content is a ROS2Message subclass.
         auto_qos_matching : bool, optional
             Whether to automatically match QoS profiles, by default True.
         qos_profile : Optional[QoSProfile], optional
             Custom QoS profile to use, by default None.
         **kwargs : Any
             Additional keyword arguments.
         """
+        if isinstance(message, ROS2Message):  # T class
+            msg_content = message.payload
+        else:  # An actual ROS 2 message
+            msg_content = message
         self._topic_api.publish(
             topic=target,
-            msg_content=message.payload,
+            msg_content=msg_content,
             msg_type=msg_type,
             auto_qos_matching=auto_qos_matching,
             qos_profile=qos_profile,