ros_sugar.robot.process#

ProcessSpec — external driver processes a plugin needs running.

Some plugins front hardware whose data arrives from a separate driver node: a LiDAR whose points come from a vendor SDK node, a depth camera with its own pipeline. Which driver that is, and how it must be configured, is exactly the robot-specific knowledge a plugin exists to absorb — a recipe author should not have to know it, and should not have to start it by hand.

A plugin declares such a process from robot.plugin.Plugin.required_processes; the Launcher owns it and brings it up through the same machinery as Launcher.add_ros_node. Plugins stay declarative and start nothing themselves, and the driver inherits launch’s supervision: respawn, captured output, and teardown ordered with the rest of the recipe. A plugin that spawned its own subprocess would get none of that, and would leak a process still holding the device if the launcher were killed.

Module Contents#

Classes#

ProcessSpec

One external ROS node a plugin needs running.

API#

class ros_sugar.robot.process.ProcessSpec#

Bases: ros_sugar.config.BaseAttrs

One external ROS node a plugin needs running.

The fields mirror the arguments of Launcher.add_ros_node, which is what the launcher passes this to.

Parameters:

  • package – ROS package holding the executable.

  • executable – Executable name.

  • name – Node name; defaults to the executable’s own.

  • parameters – Node parameters — dicts and/or paths to YAML files.

  • remappings – (from, to) topic/service remapping pairs.

  • arguments – Extra command-line arguments.

  • output – launch output configuration.

  • respawn – Restart the node if it exits. Defaults to True: a driver dying mid-run is the case this exists to survive.

  • respawn_delay – Seconds to wait before respawning.

  • precondition –

    Optional predicate, evaluated in the launcher process immediately before the node is added. Return False to skip it.

    Not a refinement — for a large class of drivers this is the difference between working and not. A driver that binds a fixed port (most LiDARs) cannot coexist with a second copy of itself, and many robots already run the vendor’s own instance from boot. Starting a second one gives a node that comes up cleanly and then never publishes, which reads as a crash and sends people looking in the wrong place. Detect the running instance here and return False.

  • inputs –

    Plugin feedbacks this node subscribes to, as {feedback_key: topic} where the topic is the name the node itself subscribes on. The launcher delivers each one there: a feedback decoded from a non-ROS transport is published on that topic by the plugin host, and one already carried on a ROS topic is remapped onto it.

    This is what lets a plugin start a stock ROS node over its own telemetry – a robot_localization EKF over the robot’s odometry – without the recipe having to put that telemetry on ROS first.

property label: str#

Human-readable identifier for logs.

launch_kwargs() → Dict[str, Any]#

This spec as keyword arguments for Launcher.add_ros_node.

precondition and inputs are deliberately absent: they are the launcher’s business, not the launch system’s.

asdict(filter: Optional[Callable] = None) → Dict#

Convert class to dict.

Return type:

dict

to_dict() → Dict#

Convert class to dict.

Return type:

dict

asdict_explicit() → Dict#

Convert to dict, keeping only what differs from the class defaults.

See explicit_fields, which this defers to.

Returns:

Fields that differ from the defaults

Return type:

dict

from_dict(dict_obj: Dict) → None#

Gets attributes values from given dictionary

Parameters:

dict_obj (Dict) – Dictionary {attribute_name: attribute_value}

Raises:

  • ValueError – If attribute_name in dictionary does not exists in class attributes

  • TypeError – If attribute_value type in dictionary does not correspond to class attribute type

from_file(file_path: str, nested_root_name: Union[str, None] = None, get_common: bool = False) → bool#

Update class attributes from yaml, json, or toml

Parameters:

  • file_path – Path to config file (.yaml, .json, .toml)

  • nested_root_name – Nested root name for the config, defaults to None

  • get_common – Whether to get extra config root (for merging), defaults to False

to_json() → Union[str, bytes, bytearray]#

Dump to json

Returns:

description

Return type:

str | bytes | bytearray

from_json(json_obj: Union[str, bytes, bytearray]) → None#

Gets attributes values from given json

Parameters:

json_obj (str | bytes | bytearray) – Json object

has_attribute(attr_name: str) → bool#

Checks if class object has attribute with given name

Parameters:

attr_name (str) – description

Returns:

If object has attribute with given name

Return type:

bool

get_attribute_type(attr_name: str) → Optional[type]#

Gets type of given attribute name

Parameters:

attr_name (str) – description

Raises:

AttributeError – If class does not have attribute with given name

Returns:

Attribute type

Return type:

type

update_value(attr_name: str, attr_value: Any) → bool#

Updates the value of an attribute in the class

Parameters:

  • attr_name (str) – Attribute name - can be nested name

  • attr_value (Any) – Attribute value

Raises:

  • AttributeError – If class does not contain attribute with given name

  • TypeError – If class attribute with given name if of different type

Returns:

If attribute value is updated

Return type:

bool

classmethod get_fields_info(class_object) → Dict[str, Dict[str, Any]]#

Returns a dictionary with metadata about each field in the class.

This includes the field’s name, type annotation, and parsed validator info.

Returns:

A dictionary where keys are field names and values are dicts of metadata.