`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()#

with_traceback()#

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#

to_dict() → Dict#

asdict_explicit() → Dict#

from_dict(dict_obj: Dict) → None#

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

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

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

has_attribute(attr_name: str) → bool#

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

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

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

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]#

classmethod values() → List[str]#

capitalize()#

casefold()#

center()#

count()#

encode()#

endswith()#

expandtabs()#

find()#

format()#

format_map()#

index()#

isalnum()#

isalpha()#

isascii()#

isdecimal()#

isdigit()#

isidentifier()#

islower()#

isnumeric()#

isprintable()#

isspace()#

istitle()#

isupper()#

join()#

ljust()#

lower()#

lstrip()#

partition()#

removeprefix()#

removesuffix()#

replace()#

rfind()#

rindex()#

rjust()#

rpartition()#

rsplit()#

rstrip()#

split()#

splitlines()#

startswith()#

strip()#

swapcase()#

title()#

translate()#

upper()#

zfill()#

name()#

value()#

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.

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.

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 role: ros_sugar.robot.plugin.PluginRole#

property id: str#

to_spec() → Dict[str, Any]#

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

property bus: Optional[ros_sugar.robot.bus.FeedbackBus]#

set_bus(bus: ros_sugar.robot.bus.FeedbackBus) → None#

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

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

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

resolve_command(topic_name: str, msg_type_name: str) → Optional[ros_sugar.robot.command.RobotCommand]#

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

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

send_command(command: ros_sugar.robot.command.RobotCommand, payload: Any) → bool#

list_feedbacks() → List[ros_sugar.robot.feedback.FeedbackSpec]#

list_commands() → List[ros_sugar.robot.command.CommandSpec]#

list_actions() → List[ros_sugar.robot.registries.ActionSpec]#

list_events() → List[ros_sugar.robot.registries.EventSpec]#

describe() → Dict[str, Any]#

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#

property id: str#

to_spec() → Dict[str, Any]#

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

property bus: Optional[ros_sugar.robot.bus.FeedbackBus]#

set_bus(bus: ros_sugar.robot.bus.FeedbackBus) → None#

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

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

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

resolve_command(topic_name: str, msg_type_name: str) → Optional[ros_sugar.robot.command.RobotCommand]#

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

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

send_command(command: ros_sugar.robot.command.RobotCommand, payload: Any) → bool#

list_feedbacks() → List[ros_sugar.robot.feedback.FeedbackSpec]#

list_commands() → List[ros_sugar.robot.command.CommandSpec]#

list_actions() → List[ros_sugar.robot.registries.ActionSpec]#

list_events() → List[ros_sugar.robot.registries.EventSpec]#

describe() → Dict[str, Any]#

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)#

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.

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`#