`ros_sugar.robot.plugin`#

RobotPlugin - the general-purpose robot plugin contract for Sugarcoat.

Plugin authors subclass RobotPlugin and, in a declarative init (no I/O), populate transports, feedbacks, commands, actions and events. One instance is passed to a Launcher.

Bring-up and tear-down are owned by RobotPluginHost, which the launcher attaches to a plugin instance on the host side. Component subprocesses get a plain plugin instance rebuilt from RobotPlugin.to_spec and use its consumer API.

If an author needs host-side setup that depends on the rclpy node (binding a RosServiceTransport is the canonical example), they override the optional on_attached hook; the host calls it once after wiring is complete.

Module Contents#

Classes#

`PluginMetadata`	Descriptive metadata for a robot plugin.
`PluginRole`	What a plugin represents in a recipe.
`Plugin`	Base class for plugins.
`RobotPlugin`	A plugin for the robot itself. What moves, and what drives it.
`SensorPlugin`	A plugin for a sensor that is not part of the robot’s own hardware.
`RobotPluginHost`	Owns the host-side lifecycle of a `RobotPlugin`.

API#

exception ros_sugar.robot.plugin.AmbiguousPluginEntryError#

Bases: LookupError

Raised when Topic(use_plugin=True) matches more than one plugin entry by message type; the recipe must name the topic after one of the plugin’s registry keys to disambiguate.

add_note()#: Exception.add_note(note) – add a note to the exception

with_traceback()#: Exception.with_traceback(tb) – set self.traceback to tb and return self.

class ros_sugar.robot.plugin.PluginMetadata#

Bases: ros_sugar.config.BaseAttrs

Descriptive metadata for a robot plugin.

Parameters:

name – Short robot name, e.g. "Lite3".
vendor – Manufacturer / vendor name.
version – Plugin or interface version string.
description – A 1-3 sentence plain-language description of the robot; its form factor, how it moves, and what it is for.

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.

class ros_sugar.robot.plugin.PluginRole#

Bases: ros_sugar.config.StrEnum

What a plugin represents in a recipe.

The role is intrinsic to the plugin, to determine its type: a robot or a sensor. The launcher uses it to enforce that a recipe describes exactly one robot, and to decide what each plugin is allowed to contribute to the components.

classmethod get_enum(__value: str) → Optional[ros_sugar.config.robot.T]#

Get Enum member equal to the given value

Parameters:: __value (str) – Reference value
Returns:: Enum member
Return type:: StrEnum | None

classmethod values() → List[str]#: Get all enum values

capitalize()#

Return a capitalized version of the string.

More specifically, make the first character have upper case and the rest lower case.

casefold()#: Return a version of the string suitable for caseless comparisons.

center()#

Return a centered string of length width.

Padding is done using the specified fill character (default is a space).

count()#

S.count(sub[, start[, end]]) -> int

Return the number of non-overlapping occurrences of substring sub in string S[start:end]. Optional arguments start and end are interpreted as in slice notation.

encode()#

Encode the string using the codec registered for encoding.

encoding The encoding in which to encode the string. errors The error handling scheme to use for encoding errors. The default is ‘strict’ meaning that encoding errors raise a UnicodeEncodeError. Other possible values are ‘ignore’, ‘replace’ and ‘xmlcharrefreplace’ as well as any other name registered with codecs.register_error that can handle UnicodeEncodeErrors.

endswith()#

S.endswith(suffix[, start[, end]]) -> bool

Return True if S ends with the specified suffix, False otherwise. With optional start, test S beginning at that position. With optional end, stop comparing S at that position. suffix can also be a tuple of strings to try.

expandtabs()#

Return a copy where all tab characters are expanded using spaces.

If tabsize is not given, a tab size of 8 characters is assumed.

find()#

S.find(sub[, start[, end]]) -> int

Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Return -1 on failure.

format()#

S.format(*args, **kwargs) -> str

Return a formatted version of S, using substitutions from args and kwargs. The substitutions are identified by braces (‘{’ and ‘}’).

format_map()#

S.format_map(mapping) -> str

Return a formatted version of S, using substitutions from mapping. The substitutions are identified by braces (‘{’ and ‘}’).

index()#

S.index(sub[, start[, end]]) -> int

Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Raises ValueError when the substring is not found.

isalnum()#

Return True if the string is an alpha-numeric string, False otherwise.

A string is alpha-numeric if all characters in the string are alpha-numeric and there is at least one character in the string.

isalpha()#

Return True if the string is an alphabetic string, False otherwise.

A string is alphabetic if all characters in the string are alphabetic and there is at least one character in the string.

isascii()#

Return True if all characters in the string are ASCII, False otherwise.

ASCII characters have code points in the range U+0000-U+007F. Empty string is ASCII too.

isdecimal()#

Return True if the string is a decimal string, False otherwise.

A string is a decimal string if all characters in the string are decimal and there is at least one character in the string.

isdigit()#

Return True if the string is a digit string, False otherwise.

A string is a digit string if all characters in the string are digits and there is at least one character in the string.

isidentifier()#

Return True if the string is a valid Python identifier, False otherwise.

Call keyword.iskeyword(s) to test whether string s is a reserved identifier, such as “def” or “class”.

islower()#

Return True if the string is a lowercase string, False otherwise.

A string is lowercase if all cased characters in the string are lowercase and there is at least one cased character in the string.

isnumeric()#

Return True if the string is a numeric string, False otherwise.

A string is numeric if all characters in the string are numeric and there is at least one character in the string.

isprintable()#

Return True if all characters in the string are printable, False otherwise.

A character is printable if repr() may use it in its output.

isspace()#

Return True if the string is a whitespace string, False otherwise.

A string is whitespace if all characters in the string are whitespace and there is at least one character in the string.

istitle()#

Return True if the string is a title-cased string, False otherwise.

In a title-cased string, upper- and title-case characters may only follow uncased characters and lowercase characters only cased ones.

isupper()#

Return True if the string is an uppercase string, False otherwise.

A string is uppercase if all cased characters in the string are uppercase and there is at least one cased character in the string.

join()#

Concatenate any number of strings.

The string whose method is called is inserted in between each given string. The result is returned as a new string.

Example: ‘.’.join([‘ab’, ‘pq’, ‘rs’]) -> ‘ab.pq.rs’

ljust()#

Return a left-justified string of length width.

Padding is done using the specified fill character (default is a space).

lower()#: Return a copy of the string converted to lowercase.

lstrip()#

Return a copy of the string with leading whitespace removed.

If chars is given and not None, remove characters in chars instead.

partition()#

Partition the string into three parts using the given separator.

This will search for the separator in the string. If the separator is found, returns a 3-tuple containing the part before the separator, the separator itself, and the part after it.

If the separator is not found, returns a 3-tuple containing the original string and two empty strings.

removeprefix()#

Return a str with the given prefix string removed if present.

If the string starts with the prefix string, return string[len(prefix):]. Otherwise, return a copy of the original string.

removesuffix()#

Return a str with the given suffix string removed if present.

If the string ends with the suffix string and that suffix is not empty, return string[:-len(suffix)]. Otherwise, return a copy of the original string.

replace()#

Return a copy with all occurrences of substring old replaced by new.

count Maximum number of occurrences to replace. -1 (the default value) means replace all occurrences.

If the optional argument count is given, only the first count occurrences are replaced.

rfind()#

S.rfind(sub[, start[, end]]) -> int

Return the highest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Return -1 on failure.

rindex()#

S.rindex(sub[, start[, end]]) -> int

Return the highest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Raises ValueError when the substring is not found.

rjust()#

Return a right-justified string of length width.

Padding is done using the specified fill character (default is a space).

rpartition()#

Partition the string into three parts using the given separator.

This will search for the separator in the string, starting at the end. If the separator is found, returns a 3-tuple containing the part before the separator, the separator itself, and the part after it.

If the separator is not found, returns a 3-tuple containing two empty strings and the original string.

rsplit()#

Return a list of the substrings in the string, using sep as the separator string.

sep The separator used to split the string.

When set to None (the default value), will split on any whitespace
character (including \n \r \t \f and spaces) and will discard
empty strings from the result.

maxsplit Maximum number of splits. -1 (the default value) means no limit.

Splitting starts at the end of the string and works to the front.

rstrip()#

Return a copy of the string with trailing whitespace removed.

If chars is given and not None, remove characters in chars instead.

split()#

Return a list of the substrings in the string, using sep as the separator string.

sep The separator used to split the string.

When set to None (the default value), will split on any whitespace
character (including \n \r \t \f and spaces) and will discard
empty strings from the result.

maxsplit Maximum number of splits. -1 (the default value) means no limit.

Splitting starts at the front of the string and works to the end.

Note, str.split() is mainly useful for data that has been intentionally delimited. With natural text that includes punctuation, consider using the regular expression module.

splitlines()#

Return a list of the lines in the string, breaking at line boundaries.

Line breaks are not included in the resulting list unless keepends is given and true.

startswith()#

S.startswith(prefix[, start[, end]]) -> bool

Return True if S starts with the specified prefix, False otherwise. With optional start, test S beginning at that position. With optional end, stop comparing S at that position. prefix can also be a tuple of strings to try.

strip()#

Return a copy of the string with leading and trailing whitespace removed.

If chars is given and not None, remove characters in chars instead.

swapcase()#: Convert uppercase characters to lowercase and lowercase characters to uppercase.

title()#

Return a version of the string where each word is titlecased.

More specifically, words start with uppercased characters and all remaining cased characters have lower case.

translate()#

Replace each character in the string using the given translation table.

table Translation table, which must be a mapping of Unicode ordinals to Unicode ordinals, strings, or None.

The table must implement lookup/indexing via getitem, for instance a dictionary or list. If this operation raises LookupError, the character is left untouched. Characters mapped to None are deleted.

upper()#: Return a copy of the string converted to uppercase.

zfill()#

Pad a numeric string with zeros on the left, to fill a field of the given width.

The string is never truncated.

name()#: The name of the Enum member.

value()#: The value of the Enum member.

class ros_sugar.robot.plugin.Plugin#

Base class for plugins.

Subclass RobotPlugin or SensorPlugin rather than this class directly, so the plugin carries the right PluginRole.

property role: ros_sugar.robot.plugin.PluginRole#

What this plugin represents. Read-only.

A plugin is a robot or a sensor by construction, so the role follows from the base class the author subclassed.

property id: str#

Identity of this plugin within a recipe.

A recipe addresses a plugin from a topic by this id (Topic(..., use_plugin=cam.id)), so it is available from the moment the plugin is constructed.

Defaults to a slug of the plugin’s metadata name, which is enough unless two plugins of the same kind are used; those are told apart by passing id= to the constructor.

to_spec() → Dict[str, Any]#: Return a JSON-serializable spec used to rebuild this plugin in a component subprocess.

static from_spec(spec: Dict[str, Any], bus_endpoint: Optional[Any] = None) → ros_sugar.robot.plugin.Plugin#

Rebuild a plugin from a to_spec dict.

Parameters:

spec – The spec dict produced by to_spec.
bus_endpoint – socket name of the host’s feedback bus; when given, a connected robot.bus.SocketFeedbackBus is attached so the plugin’s consumer API (subscribe_feedback / send_command) talks to the host across the process boundary.

property bus: Optional[ros_sugar.robot.bus.FeedbackBus]#: The feedback bus this plugin publishes to / consumes from.

set_bus(bus: ros_sugar.robot.bus.FeedbackBus) → None#: Attach the feedback bus. The host calls this during bring-up; the from_spec path uses it to give a CLIENT-side plugin its connected SocketFeedbackBus.

property requested_feedbacks: FrozenSet[str]#

Keys of the feedbacks this recipe’s components and events actually consume.

Populated by the launcher at bringup from every component’s and every event’s Topic(use_plugin=...) reference, resolved through resolve_feedback, and available before on_attached runs.

A plugin may expose far more than any one recipe uses. This is how a plugin tells the difference — chiefly to avoid paying for what nobody asked for, such as starting a LiDAR driver for a recipe that never looks at the points. See required_processes.

Empty when no launcher populated it: a RobotPluginHost built directly, as in tests and standalone tools, has no recipe to serve. Treat empty as “nothing was asked for” rather than “everything” — a standalone host that started every driver a plugin knows about would be a surprise.

property requested_commands: FrozenSet[str]#

Keys of the commands this recipe’s components actually send.

The counterpart of requested_feedbacks, resolved from components’ output topics through resolve_command.

required_processes() → List[ros_sugar.robot.process.ProcessSpec]#

External driver nodes this plugin needs running — override in subclasses that front hardware served by a separate node.

Called once per bringup in the launcher process, after requested_feedbacks and requested_commands are populated and before on_attached. Return declarations only; the launcher starts and owns the processes, so they get respawn, captured output and teardown ordered with the recipe.

Gate on what was actually requested, so a recipe that ignores a sensor does not pay to run its driver::

def required_processes(self):
    if not {"lidar_front", "lidar_back"} & self.requested_feedbacks:
        return []
    return [ProcessSpec(package="rslidar_sdk",
                        executable="rslidar_sdk_node",
                        parameters=[self.lidar_config],
                        precondition=self._lidar_port_is_free)]

Anything raised here is logged and skipped: a driver that cannot be declared should not take the whole recipe down with it.

Returns:: Processes to launch, or [] (the default) for a plugin that needs none.

on_attached(node: Any, bus: ros_sugar.robot.bus.FeedbackBus) → None#

Optional host-side setup hook — override in subclasses that need it.

RobotPluginHost calls this once, after opening transports, wiring feedback dispatch and starting heartbeats. Override it when the plugin needs to do something with the host’s rclpy node - typically binding a robot.transports.ros.RosServiceTransport client.

Parameters:

node – rclpy node owned by the launcher (may be None in standalone/test contexts).
bus – the feedback bus already attached to the plugin.

on_detached(node: Any, bus: ros_sugar.robot.bus.FeedbackBus) → None#

Optional host-side teardown hook — override in subclasses.

RobotPluginHost calls this once on close, before the transports are shut, so the plugin can still talk to the device. Override it to leave hardware in a safe state: a lidar has to be told to stop its motor, which keeps spinning otherwise, and closing the port does not do it.

Anything raised here is logged and does not stop teardown.

Parameters:

node – rclpy node owned by the launcher (may be None in standalone/test contexts).
bus – the feedback bus still attached to the plugin.

resolve_feedback(topic_name: str, msg_type_name: str) → Optional[ros_sugar.robot.feedback.Feedback]#

Return the Feedback standing in for an input topic.

Two-step resolution:

Exact match on plugin.feedbacks[topic_name] – the recipe author’s io.topic.Topic.name doubles as the registry key for disambiguation.
Unique-type fallback: a single feedback whose msg_type matches msg_type_name. Common case for plugins exposing one entry per message type.

Raises:

TypeError – when the topic name matches a feedback by key but the message types disagree – the recipe is mis-wired.
AmbiguousPluginEntryError – when the topic name doesn’t match any key and multiple feedbacks share msg_type_name; the error lists the available keys so the recipe author can rename the topic to disambiguate.

resolve_command(topic_name: str, msg_type_name: str) → Optional[ros_sugar.robot.command.RobotCommand]#: Return the RobotCommand standing in for an output topic. Same resolution rules as resolve_feedback.

subscribe_feedback(feedback: ros_sugar.robot.feedback.Feedback, on_ros_msg: Callable[[Any], None]) → ros_sugar.robot.bus.BusHandle#

Subscribe to a feedback stream; on_ros_msg is called with each decoded ROS message. Used by components for non-ROS feedback.

On an in-process bus (multithreaded launch) the decoded message is handed to on_ros_msg directly, with no serialization. Every consumer and the Monitor then share the one live message instance, so consumers must treat feedback messages as read-only and deep-copy before mutating. On a socket bus the payload is read from shared memory or deserialized, per consumer.

open_command(command: ros_sugar.robot.command.RobotCommand) → None#

Prepare a command transport for sending from a component process.

For route_via_host commands nothing is opened (payloads go over the bus); otherwise the transport’s egress side is opened.

send_command(command: ros_sugar.robot.command.RobotCommand, payload: Any) → bool#: Send an already-encoded command payload, routing via the host bus or directly through the transport as configured.

list_feedbacks() → List[ros_sugar.robot.feedback.FeedbackSpec]#: List every feedback stream this plugin exposes.

list_commands() → List[ros_sugar.robot.command.CommandSpec]#: List every command surface this plugin exposes.

list_actions() → List[ros_sugar.robot.registries.ActionSpec]#: List every high-level action factory this plugin exposes.

list_events() → List[ros_sugar.robot.registries.EventSpec]#: List every event factory this plugin exposes.

describe() → Dict[str, Any]#: Return a JSON-serializable introspection tree for this plugin.

class ros_sugar.robot.plugin.RobotPlugin#

Bases: ros_sugar.robot.plugin.Plugin

A plugin for the robot itself. What moves, and what drives it.

Exactly one robot plugin may be attached to a recipe. Beyond the transports and registries every plugin carries, a robot plugin can describe the robot it drives, and every component is configured from that description unless the recipe overrides it.

property base_height: Optional[float]#: Height of the base frame above the ground. Half the robot’s height, since the geometry is centred on the base frame. None without a geometry.

property role: ros_sugar.robot.plugin.PluginRole#

What this plugin represents. Read-only.

A plugin is a robot or a sensor by construction, so the role follows from the base class the author subclassed.

property id: str#

Identity of this plugin within a recipe.

A recipe addresses a plugin from a topic by this id (Topic(..., use_plugin=cam.id)), so it is available from the moment the plugin is constructed.

Defaults to a slug of the plugin’s metadata name, which is enough unless two plugins of the same kind are used; those are told apart by passing id= to the constructor.

to_spec() → Dict[str, Any]#: Return a JSON-serializable spec used to rebuild this plugin in a component subprocess.

static from_spec(spec: Dict[str, Any], bus_endpoint: Optional[Any] = None) → ros_sugar.robot.plugin.Plugin#

Rebuild a plugin from a to_spec dict.

Parameters:

spec – The spec dict produced by to_spec.
bus_endpoint – socket name of the host’s feedback bus; when given, a connected robot.bus.SocketFeedbackBus is attached so the plugin’s consumer API (subscribe_feedback / send_command) talks to the host across the process boundary.

property bus: Optional[ros_sugar.robot.bus.FeedbackBus]#: The feedback bus this plugin publishes to / consumes from.

set_bus(bus: ros_sugar.robot.bus.FeedbackBus) → None#: Attach the feedback bus. The host calls this during bring-up; the from_spec path uses it to give a CLIENT-side plugin its connected SocketFeedbackBus.

property requested_feedbacks: FrozenSet[str]#

Keys of the feedbacks this recipe’s components and events actually consume.

Populated by the launcher at bringup from every component’s and every event’s Topic(use_plugin=...) reference, resolved through resolve_feedback, and available before on_attached runs.

property requested_commands: FrozenSet[str]#

Keys of the commands this recipe’s components actually send.

The counterpart of requested_feedbacks, resolved from components’ output topics through resolve_command.

required_processes() → List[ros_sugar.robot.process.ProcessSpec]#

External driver nodes this plugin needs running — override in subclasses that front hardware served by a separate node.

Gate on what was actually requested, so a recipe that ignores a sensor does not pay to run its driver::

def required_processes(self):
    if not {"lidar_front", "lidar_back"} & self.requested_feedbacks:
        return []
    return [ProcessSpec(package="rslidar_sdk",
                        executable="rslidar_sdk_node",
                        parameters=[self.lidar_config],
                        precondition=self._lidar_port_is_free)]

Anything raised here is logged and skipped: a driver that cannot be declared should not take the whole recipe down with it.

Returns:: Processes to launch, or [] (the default) for a plugin that needs none.

on_attached(node: Any, bus: ros_sugar.robot.bus.FeedbackBus) → None#

Optional host-side setup hook — override in subclasses that need it.

Parameters:

node – rclpy node owned by the launcher (may be None in standalone/test contexts).
bus – the feedback bus already attached to the plugin.

on_detached(node: Any, bus: ros_sugar.robot.bus.FeedbackBus) → None#

Optional host-side teardown hook — override in subclasses.

Anything raised here is logged and does not stop teardown.

Parameters:

node – rclpy node owned by the launcher (may be None in standalone/test contexts).
bus – the feedback bus still attached to the plugin.

resolve_feedback(topic_name: str, msg_type_name: str) → Optional[ros_sugar.robot.feedback.Feedback]#

Return the Feedback standing in for an input topic.

Two-step resolution:

Exact match on plugin.feedbacks[topic_name] – the recipe author’s io.topic.Topic.name doubles as the registry key for disambiguation.
Unique-type fallback: a single feedback whose msg_type matches msg_type_name. Common case for plugins exposing one entry per message type.

Raises:

TypeError – when the topic name matches a feedback by key but the message types disagree – the recipe is mis-wired.
AmbiguousPluginEntryError – when the topic name doesn’t match any key and multiple feedbacks share msg_type_name; the error lists the available keys so the recipe author can rename the topic to disambiguate.

resolve_command(topic_name: str, msg_type_name: str) → Optional[ros_sugar.robot.command.RobotCommand]#: Return the RobotCommand standing in for an output topic. Same resolution rules as resolve_feedback.

subscribe_feedback(feedback: ros_sugar.robot.feedback.Feedback, on_ros_msg: Callable[[Any], None]) → ros_sugar.robot.bus.BusHandle#

Subscribe to a feedback stream; on_ros_msg is called with each decoded ROS message. Used by components for non-ROS feedback.

open_command(command: ros_sugar.robot.command.RobotCommand) → None#

Prepare a command transport for sending from a component process.

For route_via_host commands nothing is opened (payloads go over the bus); otherwise the transport’s egress side is opened.

send_command(command: ros_sugar.robot.command.RobotCommand, payload: Any) → bool#: Send an already-encoded command payload, routing via the host bus or directly through the transport as configured.

list_feedbacks() → List[ros_sugar.robot.feedback.FeedbackSpec]#: List every feedback stream this plugin exposes.

list_commands() → List[ros_sugar.robot.command.CommandSpec]#: List every command surface this plugin exposes.

list_actions() → List[ros_sugar.robot.registries.ActionSpec]#: List every high-level action factory this plugin exposes.

list_events() → List[ros_sugar.robot.registries.EventSpec]#: List every event factory this plugin exposes.

describe() → Dict[str, Any]#: Return a JSON-serializable introspection tree for this plugin.

class ros_sugar.robot.plugin.SensorPlugin#

Bases: ros_sugar.robot.plugin.Plugin

A plugin for a sensor that is not part of the robot’s own hardware.

An inspection camera bolted onto a robot, or a fixed camera watching a room. Several may be attached to one recipe.

A sensor sits somewhere, so it has a frame. Where that frame sits relative to the robot is the recipe’s business (see Mount); the plugin only names it. A sensor with moving parts publishes the dynamic transform from this base frame to the moving one itself. The recipe then only has to describe the static mount, and the two compose in the TF tree.

property frame_id: str#

Name of the frame this sensor’s data is expressed in.

Defaults to <id>_frame, so two instances of the same sensor get distinct frames without the plugin author doing anything.

property role: ros_sugar.robot.plugin.PluginRole#

What this plugin represents. Read-only.

A plugin is a robot or a sensor by construction, so the role follows from the base class the author subclassed.

property id: str#

Identity of this plugin within a recipe.

A recipe addresses a plugin from a topic by this id (Topic(..., use_plugin=cam.id)), so it is available from the moment the plugin is constructed.

Defaults to a slug of the plugin’s metadata name, which is enough unless two plugins of the same kind are used; those are told apart by passing id= to the constructor.

to_spec() → Dict[str, Any]#: Return a JSON-serializable spec used to rebuild this plugin in a component subprocess.

static from_spec(spec: Dict[str, Any], bus_endpoint: Optional[Any] = None) → ros_sugar.robot.plugin.Plugin#

Rebuild a plugin from a to_spec dict.

Parameters:

spec – The spec dict produced by to_spec.
bus_endpoint – socket name of the host’s feedback bus; when given, a connected robot.bus.SocketFeedbackBus is attached so the plugin’s consumer API (subscribe_feedback / send_command) talks to the host across the process boundary.

property bus: Optional[ros_sugar.robot.bus.FeedbackBus]#: The feedback bus this plugin publishes to / consumes from.

set_bus(bus: ros_sugar.robot.bus.FeedbackBus) → None#: Attach the feedback bus. The host calls this during bring-up; the from_spec path uses it to give a CLIENT-side plugin its connected SocketFeedbackBus.

property requested_feedbacks: FrozenSet[str]#

Keys of the feedbacks this recipe’s components and events actually consume.

Populated by the launcher at bringup from every component’s and every event’s Topic(use_plugin=...) reference, resolved through resolve_feedback, and available before on_attached runs.

property requested_commands: FrozenSet[str]#

Keys of the commands this recipe’s components actually send.

The counterpart of requested_feedbacks, resolved from components’ output topics through resolve_command.

required_processes() → List[ros_sugar.robot.process.ProcessSpec]#

External driver nodes this plugin needs running — override in subclasses that front hardware served by a separate node.

Gate on what was actually requested, so a recipe that ignores a sensor does not pay to run its driver::

def required_processes(self):
    if not {"lidar_front", "lidar_back"} & self.requested_feedbacks:
        return []
    return [ProcessSpec(package="rslidar_sdk",
                        executable="rslidar_sdk_node",
                        parameters=[self.lidar_config],
                        precondition=self._lidar_port_is_free)]

Anything raised here is logged and skipped: a driver that cannot be declared should not take the whole recipe down with it.

Returns:: Processes to launch, or [] (the default) for a plugin that needs none.

on_attached(node: Any, bus: ros_sugar.robot.bus.FeedbackBus) → None#

Optional host-side setup hook — override in subclasses that need it.

Parameters:

node – rclpy node owned by the launcher (may be None in standalone/test contexts).
bus – the feedback bus already attached to the plugin.

on_detached(node: Any, bus: ros_sugar.robot.bus.FeedbackBus) → None#

Optional host-side teardown hook — override in subclasses.

Anything raised here is logged and does not stop teardown.

Parameters:

node – rclpy node owned by the launcher (may be None in standalone/test contexts).
bus – the feedback bus still attached to the plugin.

resolve_feedback(topic_name: str, msg_type_name: str) → Optional[ros_sugar.robot.feedback.Feedback]#

Return the Feedback standing in for an input topic.

Two-step resolution:

Exact match on plugin.feedbacks[topic_name] – the recipe author’s io.topic.Topic.name doubles as the registry key for disambiguation.
Unique-type fallback: a single feedback whose msg_type matches msg_type_name. Common case for plugins exposing one entry per message type.

Raises:

TypeError – when the topic name matches a feedback by key but the message types disagree – the recipe is mis-wired.
AmbiguousPluginEntryError – when the topic name doesn’t match any key and multiple feedbacks share msg_type_name; the error lists the available keys so the recipe author can rename the topic to disambiguate.

resolve_command(topic_name: str, msg_type_name: str) → Optional[ros_sugar.robot.command.RobotCommand]#: Return the RobotCommand standing in for an output topic. Same resolution rules as resolve_feedback.

subscribe_feedback(feedback: ros_sugar.robot.feedback.Feedback, on_ros_msg: Callable[[Any], None]) → ros_sugar.robot.bus.BusHandle#

Subscribe to a feedback stream; on_ros_msg is called with each decoded ROS message. Used by components for non-ROS feedback.

open_command(command: ros_sugar.robot.command.RobotCommand) → None#

Prepare a command transport for sending from a component process.

For route_via_host commands nothing is opened (payloads go over the bus); otherwise the transport’s egress side is opened.

send_command(command: ros_sugar.robot.command.RobotCommand, payload: Any) → bool#: Send an already-encoded command payload, routing via the host bus or directly through the transport as configured.

list_feedbacks() → List[ros_sugar.robot.feedback.FeedbackSpec]#: List every feedback stream this plugin exposes.

list_commands() → List[ros_sugar.robot.command.CommandSpec]#: List every command surface this plugin exposes.

list_actions() → List[ros_sugar.robot.registries.ActionSpec]#: List every high-level action factory this plugin exposes.

list_events() → List[ros_sugar.robot.registries.EventSpec]#: List every event factory this plugin exposes.

describe() → Dict[str, Any]#: Return a JSON-serializable introspection tree for this plugin.

class ros_sugar.robot.plugin.RobotPluginHost(plugin: ros_sugar.robot.plugin.Plugin, node: Any, bus: ros_sugar.robot.bus.FeedbackBus, monitor_feed: Optional[Callable[[str, Any], None]] = None, owns_bus: bool = True, shm: Optional[ros_sugar.robot.shm.PluginShmManager] = None)#

Owns the host-side lifecycle of a RobotPlugin.

Plugin classes are declarative — they describe what the robot exposes but do no I/O. The host runs in the launcher process, opens the plugin’s transports, wires feedback decoding into the feedback bus and the Monitor’s event blackboard, runs heartbeats, and forwards route_via_host commands. Component subprocesses do not use a host — they construct a plain plugin from a spec and call its consumer API only.

Parameters:

plugin – The plugin instance to manage.
node – rclpy node the plugin can use for host-side ROS interactions (e.g. RosServiceTransport clients). May be None in standalone and test contexts where no ROS node is available.
bus – The feedback bus to use for fan-out to component consumers.
monitor_feed – Optional callable (channel, ros_msg) -> None that receives every decoded feedback message — the launcher wires this to Monitor.feed_external_topic so plugin events fire on the same machinery as ROS-topic events.
owns_bus – Whether this host is responsible for starting and closing the bus. Several plugins share one bus, and a shared bus outlives any single host: the second host to open would re-bind the same address, and the first to close would tear the bus down for everyone. The launcher therefore owns the shared bus and passes False. Defaults to True so a lone host still works standalone.

publish_on_ros(feedback_key: str, topic: str) → None#

Also publish a feedback on a ROS topic, for a ROS node that reads it there. The launcher sets this up from ProcessSpec.inputs.

Only for feedback the host decodes itself: one carried on a ROS topic already is on ROS, and a node reads it by remapping instead.

Parameters:

feedback_key – Key on plugin.feedbacks.
topic – ROS topic to publish it on.

Raises:

KeyError – If the plugin has no such feedback.
ValueError – If the feedback is carried on a ROS transport.

open() → None#

Bring the plugin up host-side.

Attaches the bus to the plugin, opens non-ROS transports, wires feedback decoders to the bus + monitor, registers route_via_host command forwarders, starts heartbeats, and finally calls the plugin’s optional RobotPlugin.on_attached hook.

close() → None#: Tear the plugin down host-side. Idempotent.

ros_sugar.robot.plugin#

Module Contents#

Classes#

API#

`ros_sugar.robot.plugin`#