ros_sugar.robot.transports.ros#

ROS-native transports for robot plugins.

These transports cover robots that already expose a clean ROS surface. They are mostly declarative config holders: RosTopicTransport feedback/command wiring is performed by the component’s topic-replacement machinery. RosServiceTransport builds a service client bound to the component’s node.

Module Contents#

Classes#

RosTopicTransport

A robot feedback/command carried on a plain ROS topic.

RosServiceTransport

A robot command carried on a ROS service call.

API#

class ros_sugar.robot.transports.ros.RosTopicTransport(name: str, *, topic_name: str, msg_type: Any, qos: Optional[ros_sugar.config.QoSConfig] = None, **kwargs)#

Bases: ros_sugar.robot.transports.Transport

A robot feedback/command carried on a plain ROS topic.

The component substitutes its own subscriber/publisher for the plugin’s topic at activation; this object only carries the target name, type and QoS.

Parameters:

  • name – Unique transport name.

  • topic_name – ROS topic name on the robot side.

  • msg_type – Topic message type — a SupportedType subclass or its name.

  • qos – Optional QoS configuration.

open() → None#
close() → None#
abstractmethod send(payload: Any) → bool#
open_egress() → None#
is_open() → bool#
subscribe(on_msg: Callable[[Any], None]) → ros_sugar.robot.transports.SubscriptionHandle#
property kind: str#
class ros_sugar.robot.transports.ros.RosServiceTransport(name: str, *, srv_name: str, srv_type: type, timeout_secs: float = 30.0, **kwargs)#

Bases: ros_sugar.robot.transports.Transport

A robot command carried on a ROS service call.

Binds to the component’s node with bind_node before open; send is given a fully-built service request (produced by the owning RobotCommand’s encoder) and calls the service.

Parameters:

  • name – Unique transport name.

  • srv_name – ROS service name.

  • srv_type – ROS service type.

  • timeout_secs – Service availability timeout.

bind_node(node) → None#

Attach the rclpy node used to create the service client.

open() → None#

Create the underlying service client (requires bind_node).

open_egress() → None#

Service clients are send-only; same as open.

close() → None#
send(payload: Any) → bool#

Call the service with payload (an already-built service request).

is_open() → bool#
subscribe(on_msg: Callable[[Any], None]) → ros_sugar.robot.transports.SubscriptionHandle#
property kind: str#