`ros_sugar.io.callbacks`#

ROS Subscribers Callback Classes

Module Contents#

Classes#

`GenericCallback`	GenericCallback.
`StdMsgCallback`	std_msgs callback
`StdMsgArrayCallback`	std_msgs callback
`ImageCallback`	Image Callback class. Its get method saves an image as bytes
`CompressedImageCallback`	CompressedImage Callback class. Its get method saves an image as bytes
`TextCallback`	Text Callback class. Its get method returns the text
`AudioCallback`	Audio Callback class. Its get method returns the audio
`MapMetaDataCallback`	OccupancyGrid MetaData Callback class. Its get method returns dict of meta data from the occupancy grid topic
`OdomCallback`	Ros Odometry Callback Handler to get the robot state in 2D
`PointCallback`	Ros Pose Callback Handler to get the robot state in 2D
`JointStateCallback`	sensor_msgs/JointState callback.
`ImuCallback`	sensor_msgs/Imu callback.
`NavSatFixCallback`	sensor_msgs/NavSatFix callback.
`RangeCallback`	sensor_msgs/Range callback.
`PointStampedCallback`	Ros Pose Callback Handler to get the robot state in 2D
`PoseCallback`	Ros Pose Callback Handler to get the robot state in 2D
`PoseStampedCallback`	Ros PoseStamped Callback Handler to get the robot state in 2D
`PoseArrayCallback`	Ros PoseArray Callback Handler to get a set of poses as a numpy array
`PathCallback`	Ros Path Callback Handler to get a navigation path
`OccupancyGridCallback`
`LaserScanCallback`	ROS2 LaserScan Callback Handler to process and transform sensor_msgs/LaserScan data
`PointCloudCallback`	ROS2 PointCloud2 Callback Handler to process sensor_msgs/PointCloud2 data
`CameraInfoCallback`	ROS2 CameraInfo Callback Handler to process sensor_msgs/CameraInfo data

API#

class ros_sugar.io.callbacks.GenericCallback(input_topic, node_name: str = '')#

GenericCallback.

property frame_id: Optional[str]#

Getter of the message frame ID if available

Returns:: Header frame ID
Return type:: Optional[str]

property transformation: Optional[tf2_ros.TransformStamped]#

Getter of the transformation applied to the data

Returns:: Transformation from the message frame to the requested frame
Return type:: Optional[TransformStamped]

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

Attach a resolver called on every message to refresh transformation.

The provider is given this callback (so it can read frame_id, which is only known once data arrives) and returns the transform to apply, or None if it is not available yet.

Parameters:: provider (Optional[Callable[[GenericCallback], Optional[TransformStamped]]]) – Transform resolver

set_node_name(node_name: str) → None#

Set node name.

Parameters:: node_name (str)
Return type:: None

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

set_subscriber.

Parameters:: subscriber
Return type:: None

on_callback_execute(callback: Callable, get_processed=True) → None#

Attach a method to be executed on topic callback

Parameters:: callback (Callable)

callback(msg) → None#

Topic subscriber callback

Parameters:: msg (Any) – Received ros msg

add_post_processors(processors: List[Union[Callable, socket.socket]])#

Add a post processor for callback message

Parameters:: method (List[Union[Callable, socket]]) – Post processor methods or sockets

get_output(clear_last: bool = False, **kwargs) → Any#

Post process outputs based on custom processors (if any) and return it

Parameters:

output
args
kwargs

property got_msg#: Property is true if an input is received on the topic

clear_last_msg()#: Clears the last received message on the topic

class ros_sugar.io.callbacks.StdMsgCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.GenericCallback

std_msgs callback

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.StdMsgArrayCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.GenericCallback

std_msgs callback

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.ImageCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.GenericCallback

Image Callback class. Its get method saves an image as bytes

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.CompressedImageCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.ImageCallback

CompressedImage Callback class. Its get method saves an image as bytes

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.TextCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.GenericCallback

Text Callback class. Its get method returns the text

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.AudioCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.GenericCallback

Audio Callback class. Its get method returns the audio

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.MapMetaDataCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.GenericCallback

OccupancyGrid MetaData Callback class. Its get method returns dict of meta data from the occupancy grid topic

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.OdomCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.GenericCallback

Ros Odometry Callback Handler to get the robot state in 2D

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.PointCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.GenericCallback

Ros Pose Callback Handler to get the robot state in 2D

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.JointStateCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.GenericCallback

sensor_msgs/JointState callback.

Returns the joint positions as a numpy array, ordered as the incoming message’s name field.

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.ImuCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.GenericCallback

sensor_msgs/Imu callback.

Returns the IMU state as a flat numpy array [qx, qy, qz, qw, wx, wy, wz, ax, ay, az] – orientation quaternion, angular velocity (rad/s) and linear acceleration (m/s^2).

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.NavSatFixCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.GenericCallback

sensor_msgs/NavSatFix callback.

Returns the satellite fix as a numpy array [latitude, longitude, altitude] (degrees, degrees, metres).

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.RangeCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.GenericCallback

sensor_msgs/Range callback.

Returns the measured distance in metres (the message’s range field).

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.PointStampedCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.GenericCallback

Ros Pose Callback Handler to get the robot state in 2D

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.PoseCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.GenericCallback

Ros Pose Callback Handler to get the robot state in 2D

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.PoseStampedCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.PoseCallback

Ros PoseStamped Callback Handler to get the robot state in 2D

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.PoseArrayCallback(input_topic, node_name: str = '')#

Bases: ros_sugar.io.callbacks.GenericCallback

Ros PoseArray Callback Handler to get a set of poses as a numpy array

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.PathCallback(input_topic, node_name: str = '', transformation: Optional[tf2_ros.TransformStamped] = None)#

Bases: ros_sugar.io.callbacks.GenericCallback

Ros Path Callback Handler to get a navigation path

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.OccupancyGridCallback(input_topic, node_name: str = '', to_numpy: bool = True, twoD_to_threeD_conversion_height: float = 0.01, transformation: Optional[tf2_ros.TransformStamped] = None)#

Bases: ros_sugar.io.callbacks.GenericCallback

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.LaserScanCallback(input_topic, node_name: str = '', transformation: Optional[tf2_ros.TransformStamped] = None)#

Bases: ros_sugar.io.callbacks.GenericCallback

ROS2 LaserScan Callback Handler to process and transform sensor_msgs/LaserScan data

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.PointCloudCallback(input_topic, node_name: str = '', transformation: Optional[tf2_ros.TransformStamped] = None)#

Bases: ros_sugar.io.callbacks.GenericCallback

ROS2 PointCloud2 Callback Handler to process sensor_msgs/PointCloud2 data

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

class ros_sugar.io.callbacks.CameraInfoCallback(input_topic, node_name: Optional[str] = None)#

Bases: ros_sugar.io.callbacks.GenericCallback

ROS2 CameraInfo Callback Handler to process sensor_msgs/CameraInfo data

property frame_id: Optional[str]#

property transformation: Optional[tf2_ros.TransformStamped]#

set_transform_provider(provider: Optional[Callable[[ros_sugar.io.callbacks.GenericCallback], Optional[tf2_ros.TransformStamped]]]) → None#

set_node_name(node_name: str) → None#

set_subscriber(subscriber: rclpy.subscription.Subscription) → None#

on_callback_execute(callback: Callable, get_processed=True) → None#

callback(msg) → None#

add_post_processors(processors: List[Union[Callable, socket.socket]])#

get_output(clear_last: bool = False, **kwargs) → Any#

property got_msg#

clear_last_msg()#

ros_sugar.io.callbacks#

Module Contents#

Classes#

API#

`ros_sugar.io.callbacks`#