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

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#

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#

property requested_feedbacks: FrozenSet[str]#

property requested_commands: FrozenSet[str]#

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

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#

property requested_feedbacks: FrozenSet[str]#

property requested_commands: FrozenSet[str]#

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

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