`ros_sugar.io.supported_types`#

ROS Topics Supported Message Types

Module Contents#

Classes#

`Meta`	Meta.
`SupportedType`	Class used to define all supported data types (ROS messages) callback: Callback associated with specified type _publish: ROS message creation method associate with specified type
`String`	String.
`Bool`	Bool.
`Float32`	Float32.
`Float32MultiArray`	Float32MultiArray.
`Float64`	Float64.
`Float64MultiArray`	Float64MultiArray.
`Image`	Image.
`CompressedImage`	CompressedImage format usually provided by camera vendors
`Audio`	Audio.
`MapMetaData`	MapMetaData.
`Odometry`	Odometry
`LaserScan`	LaserScan
`PointCloud2`	PointCloud2
`JointState`	JointState
`CameraInfo`	CameraInfo, the pinhole parameters that make an image metrically meaningful
`Imu`	Imu
`NavSatFix`	NavSatFix
`Range`	Range
`Path`	Path
`OccupancyGrid`	OccupancyGrid
`Point`	Point
`PointStamped`	PointStamped
`Pose`	Pose
`PoseStamped`	PoseStamped
`PoseArray`	PoseArray
`ComponentStatus`	Component Health Status
`Twist`	Twist for Control Commands

Functions#

`add_additional_datatypes`	Add additional SupportedType classes to the list of supported ROS2 messages
`get_ros_msg_fields_dict`	Parses the fields names and types from any ROS message into a dict. Handles nested messages and arrays.
`ros_msg_to_str`	Helper function to print the fields and values of a ROS message object. Handles nested messages and arrays.
`set_ros_msg_from_dict`	Creates a ROS message object from a dictionary structure.

API#

ros_sugar.io.supported_types.add_additional_datatypes(types: List[type]) → None#

Add additional SupportedType classes to the list of supported ROS2 messages

Parameters:: types (List[type]) – List of supported types

ros_sugar.io.supported_types.get_ros_msg_fields_dict(msg_class: type) → Dict[str, Any]#

Parses the fields names and types from any ROS message into a dict. Handles nested messages and arrays.

Returns:: Dict where value is type string, or nested dict (or list of dicts for arrays)

ros_sugar.io.supported_types.ros_msg_to_str(msg_object: Any) → str#

Helper function to print the fields and values of a ROS message object. Handles nested messages and arrays.

Parameters:

msg_object – The ROS message object to print
indent – Current indentation level (used for nested messages)

ros_sugar.io.supported_types.set_ros_msg_from_dict(msg_class: type, data_dict: Dict[str, Any]) → Any#

Creates a ROS message object from a dictionary structure.

Parameters:

msg_class – The ROS message class to instantiate (e.g., geometry_msgs.msg.Point)
data_dict – Dictionary containing the values (keys must match message fields)

Raises:

ValueError – If a value cannot be converted to its declared field type.

Returns:

An instance of msg_class populated with data

class ros_sugar.io.supported_types.Meta#

Bases: type

Meta.

class ros_sugar.io.supported_types.SupportedType#

Class used to define all supported data types (ROS messages) callback: Callback associated with specified type _publish: ROS message creation method associate with specified type

classmethod convert(*output, **_) → Any#

ROS message converter function for datatype

Parameters:: args
Return type:: Any

classmethod get_ros_type() → type#

Getter of the ROS2 message type

Returns:: ROS2 type
Return type:: type

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

Zero-copy hand-off for the shared-memory feedback fast path.

Return (meta, view) where view is a zero-copy memoryview of the message’s dominant buffer and meta carries everything from_shm_payload needs to rebuild the message around a copy of it. Return None (the default) for types that are not a good fit so the caller falls back to CDR. Overridden on large, single-buffer types.

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

Rebuild a ROS message from meta and a copy of the payload buffer.

Paired with to_shm_payload; only called for types that returned a payload from it.

class ros_sugar.io.supported_types.String#

Bases: ros_sugar.io.supported_types.SupportedType

String.

classmethod convert(output: str, **_) → std_msgs.msg.String#

Takes a string and returns a ROS message of type String

Returns:: String

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.Bool#

Bases: ros_sugar.io.supported_types.SupportedType

Bool.

classmethod convert(output: bool, **_) → std_msgs.msg.Bool#

Takes a bool and returns a ROS message of type Bool

Returns:: Bool

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.Float32#

Bases: ros_sugar.io.supported_types.SupportedType

Float32.

classmethod convert(output: float, **_) → std_msgs.msg.Float32#

Takes a float and returns a ROS message of type Float32

Returns:: Float32

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.Float32MultiArray#

Bases: ros_sugar.io.supported_types.SupportedType

Float32MultiArray.

classmethod convert(output: numpy.ndarray, **_) → std_msgs.msg.Float32MultiArray#

Takes a numpy array and returns a ROS message of type Float32MultiArray

Returns:: Float32

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.Float64#

Bases: ros_sugar.io.supported_types.SupportedType

Float64.

classmethod convert(output: float, **_) → std_msgs.msg.Float64#

Takes a float and returns a ROS message of type Float64

Returns:: Float64

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.Float64MultiArray#

Bases: ros_sugar.io.supported_types.SupportedType

Float64MultiArray.

classmethod convert(output: numpy.ndarray, **_) → std_msgs.msg.Float64MultiArray#

Takes a numpy array and returns a ROS message of type Float64MultiArray

Returns:: Float32

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.Image#

Bases: ros_sugar.io.supported_types.SupportedType

Image.

classmethod convert(output: Union[sensor_msgs.msg.Image, numpy.ndarray], encoding: Optional[str] = None, stamp: Optional[float] = None, frame_id: str = '', **_) → sensor_msgs.msg.Image#

Passes a ROS Image through, or builds a complete one around an array.

The array is (H, W) or (H, W, C). encoding names the pixel layout as the array holds it. When unset, it is inferred from the dtype and the channel count.

stamp in seconds and frame_id are for callers building messages outside a publisher.

Returns:: ROSImage

classmethod to_shm_payload(msg: sensor_msgs.msg.Image) → Optional[Tuple[Dict, memoryview]]#: Hand out the raw pixel buffer zero-copy, plus the header + geometry fields needed to rebuild the message.

classmethod from_shm_payload(meta: Dict, buffer: bytes) → sensor_msgs.msg.Image#

classmethod get_ros_type() → type#

class ros_sugar.io.supported_types.CompressedImage#

Bases: ros_sugar.io.supported_types.Image

CompressedImage format usually provided by camera vendors

classmethod convert(output: Union[sensor_msgs.msg.CompressedImage, numpy.ndarray], **_) → sensor_msgs.msg.CompressedImage#

Takes a ROS CompressedImage message or numpy array and returns a ROS CompressedImage message

Returns:: ROSCompressedImage

classmethod to_shm_payload(msg: sensor_msgs.msg.CompressedImage) → Optional[Tuple[Dict, memoryview]]#: CompressedImage carries only a format string plus the encoded blob.

classmethod from_shm_payload(meta: Dict, buffer: bytes) → sensor_msgs.msg.CompressedImage#

classmethod get_ros_type() → type#

class ros_sugar.io.supported_types.Audio#

Bases: ros_sugar.io.supported_types.SupportedType

Audio.

classmethod convert(output: Union[str, bytes], **_) → std_msgs.msg.ByteMultiArray#

Takes an array of audio data and returns a ROS message of type AudioData

Returns:: AudioData

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.MapMetaData#

Bases: ros_sugar.io.supported_types.SupportedType

MapMetaData.

classmethod convert(*output, **_) → Any#

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.Odometry#

Bases: ros_sugar.io.supported_types.SupportedType

Odometry

classmethod convert(*output, **_) → Any#

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.LaserScan#

Bases: ros_sugar.io.supported_types.SupportedType

LaserScan

classmethod convert(*output, **_) → Any#

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.PointCloud2#

Bases: ros_sugar.io.supported_types.SupportedType

PointCloud2

classmethod to_shm_payload(msg: sensor_msgs.msg.PointCloud2) → Optional[Tuple[Dict, memoryview]]#: The packed point buffer plus the field layout and geometry.

classmethod from_shm_payload(meta: Dict, buffer: bytes) → sensor_msgs.msg.PointCloud2#

classmethod convert(*output, **_) → Any#

classmethod get_ros_type() → type#

class ros_sugar.io.supported_types.JointState#

Bases: ros_sugar.io.supported_types.SupportedType

JointState

classmethod convert(output: Union[numpy.ndarray, List], **_) → sensor_msgs.msg.JointState#

ROS message converter for datatype JointState.

Takes an array of joint positions and produces a sensor_msgs/JointState with its position field populated.

Parameters:: output (Union[np.ndarray, List]) – joint positions
Return type:: ROSJointState

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.CameraInfo#

Bases: ros_sugar.io.supported_types.SupportedType

CameraInfo, the pinhole parameters that make an image metrically meaningful

classmethod convert(output: Union[sensor_msgs.msg.CameraInfo, ros_sugar.io.datatypes.CameraIntrinsics], stamp: Optional[float] = None, frame_id: str = '', **_) → sensor_msgs.msg.CameraInfo#

Passes a ROS CameraInfo through, or builds one from CameraIntrinsics as the inverse of read_camera_info.

Intrinsics carrying distortion coefficients describe a raw image. They go into K and D and P stays unset, so a reader falls back to K. Intrinsics without coefficients describe a rectified or registered image go into P, and into K as well since no raw matrix is known.

stamp in seconds and frame_id override the intrinsics’ own, for callers building messages outside a publisher.

Returns:: ROSCameraInfo

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.Imu#

Bases: ros_sugar.io.supported_types.SupportedType

Imu

classmethod convert(*output, **_) → Any#

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.NavSatFix#

Bases: ros_sugar.io.supported_types.SupportedType

NavSatFix

classmethod convert(*output, **_) → Any#

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.Range#

Bases: ros_sugar.io.supported_types.SupportedType

Range

classmethod convert(*output, **_) → Any#

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.Path#

Bases: ros_sugar.io.supported_types.SupportedType

Path

classmethod convert(output, **_) → Any#

ROS message converter function for datatype

Parameters:: args
Return type:: Any

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.OccupancyGrid#

Bases: ros_sugar.io.supported_types.SupportedType

OccupancyGrid

classmethod convert(output: Union[numpy.ndarray, nav_msgs.msg.OccupancyGrid], resolution: float, origin: Optional[geometry_msgs.msg.Pose] = None, **_) → nav_msgs.msg.OccupancyGrid#

ROS message converter function for datatype OccupancyGrid.

Parameters:

output (np.ndarray)
_

Return type:

ROSOccupancyGrid

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.Point#

Bases: ros_sugar.io.supported_types.SupportedType

Point

classmethod convert(output: numpy.ndarray, **_) → geometry_msgs.msg.Point#

ROS message converter function for datatype Point.

Parameters:

output (np.ndarray)
_

Return type:

ROSPoint

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.PointStamped#

Bases: ros_sugar.io.supported_types.Point

PointStamped

classmethod convert(output: numpy.ndarray, **_) → geometry_msgs.msg.PointStamped#

ROS message converter function for datatype Point.

Parameters:

output (np.ndarray)
_

Return type:

ROSPointStamped

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.Pose#

Bases: ros_sugar.io.supported_types.SupportedType

Pose

classmethod convert(output: numpy.ndarray, **_) → geometry_msgs.msg.Pose#

ROS message converter function for datatype Point.

Parameters:

output (np.ndarray)
_

Return type:

ROSPose

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.PoseStamped#

Bases: ros_sugar.io.supported_types.Pose

PoseStamped

classmethod convert(output: numpy.ndarray, **_) → geometry_msgs.msg.PoseStamped#

ROS message converter function for datatype Point.

Parameters:

output (np.ndarray)
_

Return type:

ROSPoseStamped

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.PoseArray#

Bases: ros_sugar.io.supported_types.SupportedType

PoseArray

classmethod convert(output: Union[numpy.ndarray, List], **_) → geometry_msgs.msg.PoseArray#

ROS message converter function for datatype PoseArray.

Each element of the given set is a single pose as [x, y, z], [x, y, z, heading] or [x, y, z, qw, qx, qy, qz]

Parameters:

output (Union[np.ndarray, List]) – Set of poses
_

Return type:

ROSPoseArray

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.ComponentStatus#

Bases: ros_sugar.io.supported_types.SupportedType

Component Health Status

classmethod convert(*output, **_) → Any#

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

class ros_sugar.io.supported_types.Twist#

Bases: ros_sugar.io.supported_types.SupportedType

Twist for Control Commands

classmethod convert(output: Union[numpy.ndarray, List], **_) → geometry_msgs.msg.Twist#

ROS message converter function for datatype Point.

Parameters:

output (np.ndarray)
_

Return type:

ROSPoseStamped

classmethod get_ros_type() → type#

classmethod to_shm_payload(msg: Any) → Optional[Tuple[Dict, memoryview]]#

abstractmethod classmethod from_shm_payload(meta: Dict, buffer: bytes) → Any#

ros_sugar.io.supported_types#

Module Contents#

Classes#

Functions#

API#

`ros_sugar.io.supported_types`#