# Sugarcoat Developer Documentation The following text contains the developer documentation for the Sugarcoat framework by Automatika Robotics. It is optimized for context ingestion. ## File: development/architecture.md ```markdown # Architecture Overview This document describes the internal architecture of Sugarcoat for developers contributing to or extending the framework. ## Package Layout | Package | Contents | |:--------|:---------| | `ros_sugar.core` | `BaseComponent`, `Monitor`, `Event`, `Action`, `Status`, `Fallback` / `ComponentFallbacks` | | `ros_sugar.io` | `Topic`, `Publisher`, the `SupportedType` registry and callbacks, and the `LaserScanData` / `PointCloudData` / `CameraIntrinsics` containers | | `ros_sugar.config` | `attrs`-based configuration: `BaseAttrs`, `BaseComponentConfig`, `QoSConfig`, validators, and the robot description (`RobotConfig`, `RobotFrames`, control limits) | | `ros_sugar.launch` | `Launcher`, the multiprocess entry point (`executable_main`), the in-process launch actions, and the system-info serializer the UI reads | | `ros_sugar.robot` | The plugin framework: `RobotPlugin` / `SensorPlugin`, transports, the feedback bus and shared-memory ring, mounts, driver-process and mapping declarations | | `ros_sugar.ui_node` | The UI node, its JSON/WebSocket API, and the optional FastHTML browser front-end | | `ros_sugar.condition`, `ros_sugar.actions`, `ros_sugar.base_clients`, `ros_sugar.tf`, `ros_sugar.utils` | Condition expressions, the standard action factories, service/action client handlers, TF lookup helpers, and the decorators | ## Core Module Structure The `ros_sugar.core` package exposes the primary building blocks: | Class | Base Class | Role | |:------|:-----------|:-----| | `BaseComponent` | `rclpy.lifecycle.Node` | Managed lifecycle execution unit | | `Monitor` | `rclpy.node.Node` | Event evaluation and component supervision | | `Event` | _(standalone)_ | Condition-based trigger on topic data | | `Action` | _(standalone)_ | Callable dispatched when an event fires | | `Status` | _(standalone)_ | Health status wrapper around `ComponentStatus` msg | | `Fallback` / `ComponentFallbacks` | _(attrs / standalone)_ | Failure recovery actions | ### BaseComponent `BaseComponent` extends `rclpy.lifecycle.Node` and is the primary unit of execution. It wraps lifecycle management, declarative I/O wiring, type-safe configuration via `attrs`, health status broadcasting, and fallback handling into a single class. Key constructor parameters: ```python BaseComponent( component_name: str, inputs: Optional[Sequence[Topic]] = None, outputs: Optional[Sequence[Topic]] = None, config: Optional[BaseComponentConfig] = None, config_file: Optional[str] = None, callback_group: Optional[CallbackGroup] = None, fallbacks: Optional[ComponentFallbacks] = None, main_action_type: Optional[type] = None, main_srv_type: Optional[type] = None, ) ``` A component is a plain Python object until its ROS node is initialized. The `Launcher` constructs components in the launcher process, then either initializes the node in a thread of that process or serializes the component's construction arguments (config, topics, events, fallbacks, plugins) into the command line of a separate process, where `executable_main` rebuilds it. Recipe-level knobs that are not part of the config live directly on the component, such as `launch_prefix`, a command prefix (`taskset`, `nice`, `chrt`, a profiler) applied to the component's process under multiprocess launch. ### Monitor `Monitor` extends `rclpy.node.Node` (a standard, non-lifecycle node). It is responsible for: - Subscribing to all registered `Event` topics and evaluating conditions via an event blackboard (`EventBlackboardEntry` cache). - Receiving decoded robot-plugin feedback for topics that have no ROS subscription: the plugin HOST registers them with `register_external_topic` and injects each decoded message through `feed_external_topic`, so events over plugin telemetry evaluate exactly as over ROS topics. - Creating service clients for component reconfiguration and lifecycle transitions, and activating components on start once they are discovered on the graph. - Invoking component methods at runtime via `ExecuteMethod` service clients. - Emitting `InternalEvent` instances back to the `Launcher` so the corresponding `Action` can be dispatched via the ROS launch event system. - Broadcasting the static transforms declared by mounts on `/tf_static`. Health status is not consumed by the Monitor. Each component evaluates its own status and runs its fallbacks on its own timer (see {doc}`event_system`); process-level crash recovery is the Launcher's `on_process_fail`. When using the `Launcher`, the `Monitor` is created and configured automatically; users do not need to instantiate it directly. ### Launcher `Launcher` (in `ros_sugar.launch.launcher`) provides a Pythonic alternative to `ros2 launch`. Its recipe-facing API: | Method / property | Purpose | |:------------------|:--------| | `add_pkg(components, package_name, executable_entry_point, events_actions, multiprocessing, ...)` | Add components from one package; `multiprocessing=True` runs each in its own process (requires the package name and entry point) | | `add_plugin(plugin, mount=None)` | Attach a robot or sensor plugin; a `Mount` places a sensor plugin in TF | | `add_ros_node(package, executable, ...)` / `include_launch_file(package, launch_file, launch_args)` | Bring up external ROS nodes and launch files alongside the components, each in its own process | | `on(event, action)` | Register an event/action pair; sugar for `events_actions` | | `on_process_fail(max_retries)` | Respawn a multiprocess component that exits unexpectedly | | `enable_ui(inputs, outputs, port, serve_browser, ...)` | Start the UI node with its JSON/WebSocket API and, optionally, the browser front-end | | `robot`, `frames`, `robot_frame`, `world_frame` | Broadcast the robot description and frame names to every component; a robot plugin supplies defaults for `robot` and the body frame | | `bringup(config_file=None, introspect=False, launch_debug=False)` | Build the launch description and run it, blocking until shutdown | `bringup()` proceeds in this order: validate that every `Topic(use_plugin=...)` names an attached plugin; hand every plugin to every component; apply the robot plugin's `robot_config` and `base_frame` unless the recipe set its own; route events to their owners; create the UI node and the Monitor; start the shared feedback bus, then for each plugin its declared driver processes and its HOST (transports, decoders and heartbeats); publish mounts as static TF; build one launch action per component (thread or process); run the `LaunchService`. On exit it closes every plugin HOST, then the shared bus and the shared-memory segments. ## Component Lifecycle `BaseComponent` follows the ROS 2 managed lifecycle with four transition callbacks: ``` [Unconfigured] --on_configure--> [Inactive] --on_activate--> [Active] ^ | | | |<---on_deactivate---------| |<------on_cleanup--------------| ``` ### on_configure Called when the component transitions from **Unconfigured** to **Inactive**. It loads the configuration file if one was given, calls `custom_on_configure()`, and resets the health status to healthy. ### on_activate Called when the component transitions from **Inactive** to **Active**. This is where the ROS resources are created, in this order: 1. Robot plugin adaptation: inputs and outputs that opted in with `use_plugin` are bound to the plugin's feedback and commands (see {doc}`custom_robot_plugin`). This runs first because it may replace entries in `self.callbacks` and `self.publishers_dict`. 2. `init_variables()`, the hook for component state and for declaring which frame each input should be transformed into. 3. Subscriptions for the declared `inputs` (each `Topic` wired to a `GenericCallback`), publishers for the declared `outputs`, the built-in services (parameter change, topic replacement, configure-from-file, execute-method), service and action clients, the main action server or service, the subscriptions that feed fallbacks, and finally the execution timer and the fallback-check timer. 4. Event management and external processors are attached, and `custom_on_activate()` runs. ### on_deactivate Transitions from **Active** back to **Inactive**. Timers, servers, clients, subscriptions and publishers are destroyed and TF lookups are paused; `custom_on_deactivate()` runs. ### on_cleanup Transitions from **Inactive** back to **Unconfigured**. Override `custom_on_cleanup()` to release resources and reset internal state. ## IO Module The `ros_sugar.io` package handles typed topic communication. ### Topic `Topic` is a descriptor that binds a ROS topic name to a `SupportedType`. It carries the topic name (a leading `/` is stripped), message type (a class or its name as a string), QoS profile, `data_timeout` (how long the event system holds a message before treating it as stale), and `use_plugin`: `True` binds the topic to the robot plugin, a string binds it to the plugin with that id, `False` keeps it a plain ROS topic. Topics are declared on components as `inputs` and `outputs` and are wired during activation. ### Publisher `Publisher` wraps `rclpy.publisher.Publisher` and adds the `SupportedType.convert()` step so that components can publish Python-native data (e.g., `numpy` arrays) without manually constructing ROS messages. After conversion it stamps the header with the node clock and the `frame_id` passed to `publish()`. Pre-processors can be attached to transform data before conversion. ### SupportedType and callbacks `SupportedType` is the base class for the type system. Each subclass maps a ROS message type (`_ros_type`), a callback class that turns messages into Python data (`callback`), a conversion function that produces the ROS message from Python data (`convert`), a UI streaming mode (`_ui_rate_sampled`), and optional shared-memory hooks for large payloads. The sensor callbacks return the containers in `ros_sugar.io.datatypes` (`LaserScanData`, `PointCloudData`, `CameraIntrinsics`). See {doc}`custom_types` for details on extending it. ### Frames A component asks for an input in a given frame with `transform_input_to(topic_name, goal_frame)`. The source frame is read from each message header, so nothing about sensor frames is configured; the component's callback is handed a transform resolver, and the spatial callbacks return their data already transformed. All frame pairs share one TF buffer per component, so the node subscribes to `/tf` and `/tf_static` exactly once. Frame names come from `config.frames` (`RobotFrames`: `robot_base` and `world`), which the Launcher sets from the recipe or from the robot plugin. ## Robot Plugins A plugin adapts a recipe to specific hardware without changing component code. A `RobotPlugin` (exactly one per recipe) or `SensorPlugin` (any number) declares transports (ROS topic, ROS service, UDP, HTTP, vendor SDK), feedbacks (decoders producing ROS messages), commands (encoders producing wire payloads), action and event factories, and optionally the robot description, the placement of its built-in sensors, the driver processes it depends on, and how the robot is mapped. At bringup the Launcher runs a **HOST** for each plugin in its own process: it opens the transports, decodes telemetry once, publishes it on a feedback bus, and feeds the Monitor's blackboard. Under multithreaded launch the bus is in-process and components consume the HOST's decoded messages directly. Under multiprocess launch the bus is a Unix socket, each component process rebuilds a **CLIENT** plugin from a serialized spec, and large feedbacks (images, point clouds) cross the boundary through a shared-memory ring rather than CDR over the socket. Components bind inputs and outputs to plugin feedbacks and commands during activation. See {doc}`custom_robot_plugin`. ## Callback Groups `BaseComponent` uses ROS 2 callback groups to control concurrency: - **`MutuallyExclusiveCallbackGroup`** -- Default for service callbacks; ensures serial execution. - **`ReentrantCallbackGroup`** -- Used when the component needs concurrent subscription callbacks (e.g., multiple sensor streams processed in parallel). The callback group can be specified at construction via the `callback_group` parameter. ## Key Decorators ### @component_action Defined in `ros_sugar.utils.component_action`. Marks a method as an action that can be dispatched by the event system. The decorator enforces: 1. The method belongs to a `LifecycleNode` instance. 2. The return type annotation is `bool` or `None`. 3. If `active=True`, the component must be in the **Active** lifecycle state. Can be used bare (`@component_action`) or with parameters (`@component_action(description={...}, active=True)`). The optional `description` parameter accepts an OpenAI-compatible tool/function description dict, used when actions are exposed as tools to an orchestrating LLM. ```python from ros_sugar.utils import component_action class MyComponent(BaseComponent): @component_action def stop_motors(self) -> bool: # ... stop logic ... return True @component_action(description={ "type": "function", "function": { "name": "stop_motors", "description": "Immediately stop all motors.", }, }) def stop_motors_with_desc(self) -> bool: ... ``` ### @component_fallback Defined in `ros_sugar.utils.component_fallback`. Marks a method as a fallback handler. The decorator verifies that rclpy is initialized and the component is at least in the **Inactive** state (i.e., configured or active). This allows fallbacks to fire even when the component has been deactivated due to an error. Like `@component_action`, it can be used bare or with a `description` parameter for LLM tool descriptions. ```python from ros_sugar.utils import component_fallback class MyComponent(BaseComponent): @component_fallback def restart(self) -> None: self.trigger_deactivate() self.trigger_activate() ``` ### @action_handler Defined in `ros_sugar.utils.action_handler`. Used internally to validate that a function returns `SomeEntitiesType` (the ROS launch entity type). This is primarily for functions that integrate directly with the launch event system. ## Monitor Orchestration At runtime the `Monitor` operates a tight evaluation loop: 1. **Receive** -- Subscription callbacks, and `feed_external_topic` calls from plugin HOSTs, write incoming messages into a shared `Dict[str, EventBlackboardEntry]` (the "blackboard"). Each entry carries a UUID and timestamp for staleness detection. 2. **Evaluate** -- For every registered `Event`, `Monitor` calls `event.check_condition(blackboard)`. The `Condition` tree is evaluated against the cached topic messages. Composite conditions (AND / OR / NOT via `ConditionLogicOp`) are resolved recursively. 3. **Trigger** -- If a condition evaluates to `True`, the event's registered actions are submitted to a shared `ThreadPoolExecutor` for non-blocking execution. 4. **Emit** -- For actions that must be handled at the launch level (lifecycle transitions, process restarts), the `Monitor` emits an `InternalEvent` which is caught by an `OnInternalEvent` handler registered by the `Launcher`. ## Launcher Process Graph The `Launcher` supports two execution modes, chosen per `add_pkg` call: ### Multi-Threaded All components run in the same process. Each component gets its own callback group. The `Launcher` uses a `MultiThreadedExecutor` to spin all nodes concurrently. This is simpler but shares a single fault domain. Plugin feedback uses an in-process bus, and a `launch_prefix` set on such a component has no effect (the launcher warns). ### Multi-Process Each component is launched as a separate ROS 2 process via `ExecuteProcess`. The `Launcher` communicates with components through ROS services and the `Monitor`'s topic subscriptions. This provides process isolation -- a crash in one component does not bring down the others, and `on_process_fail` can respawn it. Plugin feedback uses a socket bus with the shared-memory fast path for large messages, and `launch_prefix` applies. External nodes added with `add_ros_node`, launch files included with `include_launch_file`, and driver processes declared by plugins always run in their own processes, in either mode. In both modes, the `Monitor` node runs in the main launcher process and coordinates lifecycle transitions via `LifecycleTransition` launch actions. ``` ## File: development/custom_types.md ```markdown # Extending the Type System Sugarcoat uses a type system built on `SupportedType` to bridge ROS 2 message types with Python-native data. This document explains how the system works, what a type can opt into (UI streaming mode, the shared-memory fast path), and how to extend it with custom types. ## SupportedType Base Class Every supported message type is a subclass of `ros_sugar.io.supported_types.SupportedType`. The base class defines these extension points: ```python class SupportedType: # The ROS 2 message class (e.g., std_msgs.msg.String) _ros_type: type # Callback class that turns incoming messages into Python data callback = callbacks.GenericCallback # Whether UI/API clients receive this stream rate-sampled (True) # or pushed per message (False) _ui_rate_sampled: bool = False @classmethod def convert(cls, output, **_) -> Any: """Convert Python data into a ROS message instance.""" @classmethod def get_ros_type(cls) -> type: """Return the underlying ROS 2 message class.""" @classmethod def to_shm_payload(cls, msg) -> Optional[Tuple[Dict, memoryview]]: """Zero-copy hand-off for the shared-memory fast path. Default: None.""" @classmethod def from_shm_payload(cls, meta: Dict, buffer: bytes) -> Any: """Rebuild the message from `meta` and a copy of its buffer.""" ``` ### _ros_type Class attribute holding the ROS 2 message class. It is used to create subscriptions and publishers, validate topic compatibility, and generate UI schemas. ### callback A `GenericCallback` subclass that turns each incoming ROS message into the Python value a component reads through `self.callbacks[name].get_output()`. Different types use specialized callbacks: `ImageCallback` returns a numpy array, `OdomCallback` a position/heading/speed array, `LaserScanCallback` a `LaserScanData` container, `StdMsgCallback` the bare `data` field. See [Custom Callbacks](#custom-callbacks) for the contract. ### convert A classmethod that takes Python-native data and returns a ROS message instance. The first positional argument must be named `output`. `Publisher.publish()` calls it and then stamps the message header (frame and time) itself, so a converter does not need to fill headers on the publish path. Converters may accept extra keyword arguments, which reach them from `publish(output, **kwargs)` or from direct calls. Two built-in examples: - `Image.convert(array, encoding=None, stamp=None, frame_id="")` builds a complete `sensor_msgs/Image` from an `(H, W)` or `(H, W, C)` array. `encoding` is inferred from the dtype and channel count when not given (`mono8`, `rgb8`, `rgba8`, or the CvMat form such as `16UC1`), and `step` and `is_bigendian` are filled in. A decoder producing another layout, such as BGR from OpenCV, must pass `encoding="bgr8"`. - `CameraInfo.convert(intrinsics, stamp=None, frame_id="")` builds a `sensor_msgs/CameraInfo` from a `CameraIntrinsics`, the inverse of `read_camera_info`. `stamp` (seconds) and `frame_id` exist for callers building messages outside a publisher, such as a robot plugin decoder. ### _ui_rate_sampled How the web UI and the JSON/WebSocket API stream an output of this type to clients: - `False` (default): every message is pushed as it arrives. Lossless, and cheap for low-rate data such as text, poses or status. - `True`: the latest value is sampled at a rate (`enable_ui(api_stream_default_rate=...)`, capped by `api_max_stream_rate`). Use it for continuous high-bandwidth streams where a client never wants every raw frame. The built-in types set it on `Image`, `CompressedImage`, `LaserScan`, `PointCloud2` and `OccupancyGrid`. A derived package should set it on its own heavy streaming types. A client can override the default per connection with `?rate=` (`?rate=0` forces push), and `GET /api/interfaces` reports the default of each output as `"mode": "sampled"` or `"push"`. ### to_shm_payload / from_shm_payload Under multiprocess launch, feedback decoded by a robot plugin HOST crosses into component processes over a socket bus, CDR-serialized. For large messages that serialization is the dominant cost, so a type can opt into a shared-memory ring instead: the HOST writes the raw buffer into shared memory and only a small descriptor crosses the socket. - `to_shm_payload(msg)` returns `(meta, view)`: `view` is a zero-copy `memoryview` of the message's dominant buffer and `meta` a msgpack-serializable dict carrying everything else needed to rebuild the message. Return `None` (the default) when the type is not a good fit. - `from_shm_payload(meta, buffer)` rebuilds the message from `meta` and a `bytes` copy of the buffer. The fast path is taken only when a payload is at least `ros_sugar.robot.plugin.SHM_MIN_BYTES` (32 KiB). Smaller messages, types that return `None`, and any shared-memory failure fall back to CDR transparently. `Image`, `CompressedImage` and `PointCloud2` implement the hooks; the `Image` implementation shows the shape: ```python @classmethod def to_shm_payload(cls, msg: ROSImage): meta = { "h": msg.height, "w": msg.width, "e": msg.encoding, "b": int(msg.is_bigendian), "st": msg.step, "s": msg.header.stamp.sec, "ns": msg.header.stamp.nanosec, "f": msg.header.frame_id, } return meta, memoryview(msg.data) @classmethod def from_shm_payload(cls, meta, buffer: bytes) -> ROSImage: msg = ROSImage() msg.header.stamp.sec = meta["s"] msg.header.stamp.nanosec = meta["ns"] msg.header.frame_id = meta["f"] msg.height, msg.width = meta["h"], meta["w"] msg.encoding, msg.step = meta["e"], meta["st"] msg.is_bigendian = bool(meta["b"]) msg.data = bytes_to_array(buffer) return msg ``` See {doc}`custom_robot_plugin` for where this sits in the plugin data path. ## Building Messages Efficiently The Python message classes generated by rosidl take an `array.array` of the field's typecode for a sequence field as is. Anything else (a list, `bytes`, a numpy array) is walked element by element in Python, which dominates the cost of building an image or a point cloud. Use `ros_sugar.io.utils.bytes_to_array(buffer, typecode="B")` for large fields: ```python import numpy as np from ros_sugar.io.utils import bytes_to_array msg.data = bytes_to_array(frame.tobytes()) # uint8[] field grid.data = bytes_to_array(cells.astype(np.int8).tobytes(), "b") # int8[] field ``` `numpy_to_multiarray` does the same for the `*MultiArray` types. Two related rules: - Assign the declared Python type to scalar fields. ROS 2 Humble's generated setters check field types on every assignment (`is_bigendian = 0` on a `bool` field raises an `AssertionError` there), while newer distributions only check when `ROS_PYTHON_CHECK_FIELDS=1` is set. Run the tests with that variable to catch such mistakes before CI does. - Leave header stamping to the publisher on the publish path, as described under `convert`. ## Built-in Types Sugarcoat ships with the following built-in types in `ros_sugar.io.supported_types`. "Sampled" marks types whose `_ui_rate_sampled` is `True`; "SHM" marks types implementing the shared-memory hooks. | Type | ROS Message | Callback | Sampled | SHM | |:-----|:------------|:---------|:-------:|:---:| | `String` | `std_msgs/String` | `TextCallback` | | | | `Bool` | `std_msgs/Bool` | `StdMsgCallback` | | | | `Float32` | `std_msgs/Float32` | `StdMsgCallback` | | | | `Float64` | `std_msgs/Float64` | `StdMsgCallback` | | | | `Float32MultiArray` | `std_msgs/Float32MultiArray` | `StdMsgArrayCallback` | | | | `Float64MultiArray` | `std_msgs/Float64MultiArray` | `StdMsgArrayCallback` | | | | `Audio` | `std_msgs/ByteMultiArray` | `AudioCallback` | | | | `Image` | `sensor_msgs/Image` | `ImageCallback` | yes | yes | | `CompressedImage` | `sensor_msgs/CompressedImage` | `CompressedImageCallback` | yes | yes | | `CameraInfo` | `sensor_msgs/CameraInfo` | `CameraInfoCallback` | | | | `LaserScan` | `sensor_msgs/LaserScan` | `LaserScanCallback` | yes | | | `PointCloud2` | `sensor_msgs/PointCloud2` | `PointCloudCallback` | yes | yes | | `Imu` | `sensor_msgs/Imu` | `ImuCallback` | | | | `JointState` | `sensor_msgs/JointState` | `JointStateCallback` | | | | `NavSatFix` | `sensor_msgs/NavSatFix` | `NavSatFixCallback` | | | | `Range` | `sensor_msgs/Range` | `RangeCallback` | | | | `Odometry` | `nav_msgs/Odometry` | `OdomCallback` | | | | `Path` | `nav_msgs/Path` | `PathCallback` | | | | `OccupancyGrid` | `nav_msgs/OccupancyGrid` | `OccupancyGridCallback` | yes | | | `MapMetaData` | `nav_msgs/MapMetaData` | `MapMetaDataCallback` | | | | `Point` | `geometry_msgs/Point` | `PointCallback` | | | | `PointStamped` | `geometry_msgs/PointStamped` | `PointStampedCallback` | | | | `Pose` | `geometry_msgs/Pose` | `PoseCallback` | | | | `PoseStamped` | `geometry_msgs/PoseStamped` | `PoseStampedCallback` | | | | `PoseArray` | `geometry_msgs/PoseArray` | `PoseArrayCallback` | | | | `Twist` | `geometry_msgs/Twist` | `GenericCallback` | | | | `ComponentStatus` | `automatika_ros_sugar/ComponentStatus` | `GenericCallback` | | | The sensor callbacks return the containers in `ros_sugar.io.datatypes`: `LaserScanCallback` a `LaserScanData` (ranges and angles as float32 arrays), `PointCloudCallback` a `PointCloudData` (the raw point buffer and its layout, with `xyz` decoded lazily on first use), `CameraInfoCallback` a `CameraIntrinsics`. The spatial callbacks (scan, cloud, grid, odometry, path, poses and points) return their data in the frame the component asked for; see the frames section of {doc}`custom_component`. ## Registering Additional Types Use `add_additional_datatypes()` to register custom types at runtime: ```python from ros_sugar.io.supported_types import add_additional_datatypes add_additional_datatypes([MyCustomType, AnotherType]) ``` The function maintains a global `_additional_types` dictionary keyed by the type's module and class name. When a type with the same class name is already registered, the function merges callbacks and conversion functions into the existing entry rather than replacing it (see [Merging Behavior](#merging-behavior)). Once registered, a type can be named in a `Topic` either by class or by its name as a string: `Topic(name="/t", msg_type="MyCustomType")`. Under multiprocess launch, the Launcher passes the modules of all registered types to every component process, which imports them before rebuilding its topics from the launch arguments. Registering at import time of your package, as shown below, is what makes that work. ## Adding a Custom Type There are two ways to add a type. The quick way covers the common case of wrapping a message with a decoder and an encoder; subclassing gives access to everything else a type can declare. ### The quick way: `create_supported_type` `create_supported_type(ros_msg_type, converter=None, callback=None)` builds and registers a `SupportedType` subclass from two annotated functions: ```python from sensor_msgs.msg import Temperature as ROSTemperature from ros_sugar import create_supported_type def _temperature_callback(msg: ROSTemperature) -> float: return msg.temperature def _temperature_converter(output: float) -> ROSTemperature: msg = ROSTemperature() msg.temperature = float(output) return msg Temperature = create_supported_type( ROSTemperature, callback=_temperature_callback, converter=_temperature_converter, ) ``` The annotations are checked: the callback's first parameter must be annotated with the ROS message type and its return with a non-ROS Python type; the converter's return annotation must be the ROS message type. Either function may be omitted. The type is registered under the calling module and can be used right away: ```python from ros_sugar.io import Topic temperature_topic = Topic(name="/temperature", msg_type=Temperature) ``` This is the form robot plugins use to wrap a manufacturer's custom messages; see {doc}`custom_robot_plugin`. ### The full way: subclass `SupportedType` Subclass when the type needs a custom UI payload, a streaming mode, the shared-memory hooks, or frame handling: ```python from typing import Optional from sensor_msgs.msg import Temperature as ROSTemperature from ros_sugar.io.callbacks import GenericCallback from ros_sugar.io.supported_types import SupportedType, add_additional_datatypes class TemperatureCallback(GenericCallback): def _get_output(self, **_) -> Optional[float]: if self.msg is None: return None return self.msg.temperature def _get_ui_content(self, **_): # JSON-serializable content for the web UI and the API return {"celsius": self._get_output()} class Temperature(SupportedType): _ros_type = ROSTemperature callback = TemperatureCallback @classmethod def convert(cls, output: float, **_) -> ROSTemperature: msg = ROSTemperature() msg.temperature = float(output) return msg add_additional_datatypes([Temperature]) ``` ## Custom Callbacks A callback class inherits from `GenericCallback` and is constructed by the component with the input `Topic`. The contract: - `callback(msg)` is the ROS subscription callback. It stores the message in `self.msg`, records the message frame in `self.frame_id`, refreshes `self.transformation`, and fires any attached extra callback. Do not override it to compute outputs; if you need per-message work, override it and call `super().callback(msg)` first. - `_get_output(**kwargs)` returns the Python value for the component, computed from `self.msg`. Return `None` when no message has arrived. Keyword arguments are whatever the component passes to `get_output(**kwargs)`; the built-in `OccupancyGridCallback`, for example, accepts `get_obstacles` and `get_three_d`. - `get_output(**kwargs)` is what components call. It runs `_get_output` and then any post-processors attached with `add_post_processors` (see {doc}`custom_processing`). Do not override it. - `_get_ui_content(**_)` returns JSON-serializable content (a `str` or a `dict`) for the web UI and the API. The default returns `get_output()`, which is fine for scalars and strings. Heavy types return a summary; the scan and cloud callbacks send metadata only. - `self.frame_id` and `self.transformation` support spatial data. When a component asked for this input in a frame (`transform_input_to`), `self.transformation` holds the `TransformStamped` from the message frame to that frame, or `None` while it is unresolved. A callback for spatial data applies it in `_get_output`. - `self.got_msg` is `True` once a message has been received; `got_all_inputs()` on the component is built on it. When two packages register the same type name with different callbacks, the merged type carries a list of callback classes, and a component picks the one defined in its own package. ## How Derived Packages Register Types Packages built on Sugarcoat (such as Kompass or EmbodiedAgents) register their own types by calling `add_additional_datatypes()` at import time. For example, a navigation package might add: ```python # In my_nav_package/__init__.py from ros_sugar.io.supported_types import add_additional_datatypes from .types import CostMap, Waypoint, TrajectoryArray add_additional_datatypes([CostMap, Waypoint, TrajectoryArray]) ``` This ensures that when any component from `my_nav_package` is imported, the types are immediately available for topic wiring and event conditions. ### Merging Behavior If two packages register a type with the same class name, `add_additional_datatypes()` merges them: - **callback**: If the existing type has no callback, the new one is used. If both have callbacks, they are combined into a list. - **_ros_type**: Only set if the existing type has no `_ros_type`. - **convert**: Merged using the same list-accumulation logic as callbacks. This allows, for example, one package to define the `_ros_type` and another to supply a specialized `convert` function for the same message type. ``` ## File: development/event_system.md ```markdown # Event-Driven Architecture Sugarcoat provides a declarative event-driven system that lets you define **conditions** on ROS2 topics or Python callables and associate them with **actions** that execute when the conditions are met. This page covers both the user-facing API and the low-level implementation details for developers working on or extending the framework. :::{tip} Open the [Interactive Architecture Diagram](../advanced/events_architecture.html) for a visual version of the routing and flow diagrams below, with clickable flow selectors and architecture highlighting. ::: --- ## Part 1 — Event & Action API ### Condition Types #### Condition Expression A declarative predicate on a topic message attribute. To build the condition, all the nested attributes of a ROS2 topic are accessible via the `msg` attribute. The condition is evaluated each time new data arrives on the involved topic. ```python from ros_sugar.core import Event from ros_sugar.io import Topic event_topic = Topic(name="/float_input", msg_type="Float32") event = Event(event_condition=event_topic.msg.data > 3.0, on_change=True) ``` The expression `event_topic.msg.data` returns a `MsgConditionBuilder` that captures the attribute path `["data"]`. Applying a comparison operator (e.g., `>`) produces a `Condition` object with the topic name, attribute path, operator function, and reference value. #### Nested Attribute Access `MsgConditionBuilder` supports chained attribute access to reach deeply nested fields in ROS messages: ```python odom = Topic(name="/odom", msg_type="Odometry") # Access odom.pose.pose.position.x position_event = Event(odom.msg.pose.pose.position.x > 5.0) ``` Attribute paths are validated at construction time against the ROS message type hierarchy. An `AttributeError` is raised if the path is invalid. #### Condition Tree (Compound Conditions) Conditions can be composed using logical operators to form a tree: ```python sensor = Topic(name="/battery_level", msg_type="Float32") motor = Topic(name="/motor", msg_type="Int32") temp = Topic(name="/temperature", msg_type="Float32") # AND: both conditions must be true critical = Event((sensor.msg.data < 5.0) & (motor.msg.data == 1)) # OR: either condition triggers alert = Event((sensor.msg.data < 10.0) | (temp.msg.data > 80.0)) ``` Internally, this creates a composite `Condition` with a `ConditionLogicOp` (`AND`, `OR`, or `NOT`) and a list of `sub_conditions`. Evaluation is recursive — the `Condition.evaluate()` method walks the tree and applies each leaf condition against the topic cache. #### Event Patterns Summary | Pattern | Description | Example | |:--------|:------------|:--------| | **OnAny** | Fires when any data arrives on the topic | `Event(topic)` (pass a `Topic` directly) | | **OnEqual** | Fires when value equals reference | `Event(topic.msg.data == 42)` | | **OnGreater** | Fires when value exceeds reference | `Event(topic.msg.data > threshold)` | | **OnLess** | Fires when value falls below reference | `Event(topic.msg.data < threshold)` | | **OnDifferent** | Fires when value differs from reference | `Event(topic.msg.data != expected)` | | **OnChange** | Fires on transition from False to True | `Event(condition, on_change=True)` | | **OnCondition** | Fires on arbitrary compound condition | `Event((a.msg.x > 1) & (b.msg.y < 2))` | #### Topic (on-any) When a `Topic` object is passed directly (rather than a `Condition`), the event fires whenever all involved topics have data present in the blackboard: ```python event = Event(event_condition=event_topic) ``` #### Robot Plugin Feedback A feedback stream declared by a robot plugin is a topic like any other for the event system. `Feedback.as_topic()` returns the `Topic` to build conditions on: the real ROS topic for a `RosTopicTransport` feedback, or a synthetic topic named after the feedback's bus channel for any other transport. Plugins usually wrap this in their `events` registry so a recipe never sees the channel name: ```python low_battery = Event(robot.feedbacks["battery"].as_topic().msg.data < 0.2) launcher.on(robot.events.low_battery(0.2), robot.actions.sit()) # the same, via the plugin ``` Synthetic topics have no ROS subscription. The plugin HOST feeds each decoded message into the Monitor's blackboard directly (see Part 2), and a component that owns an action on such an event subscribes to the feedback bus instead of creating a ROS subscription. #### Callable A user-supplied function polled at `check_rate` Hz. It must return `bool` and must **not** be a `@component_action` method (those are bound to Actions and Fallbacks and cannot be used as conditions). ```python def timeout_reached() -> bool: return time.time() - last_update > 5.0 event = Event(event_condition=timeout_reached, check_rate=10.0) ``` #### OnChange (Edge Detection) Setting `on_change=True` adds edge-detection semantics. The event fires only on the transition from `False` to `True`, not while the condition remains true: ```python # Fires once when the robot enters the danger zone, not continuously entered_danger = Event(sensor.msg.data < 0.5, on_change=True) ``` #### JSON Serialization Events and their conditions support full serialization for multi-process execution. When components run in separate processes, events are serialized via `Event.to_json()` / `Event.from_json()`, which in turn serializes the `Condition` tree. This is used by the `Launcher` when spawning components via `ExecuteProcess`. ```python event_json = my_event.to_json() restored_event = Event.from_json(event_json) ``` The serialization preserves the complete condition tree, operator functions (mapped by name), reference values, and topic metadata. --- ### Action Types and Ownership Every action will have an **owner**: the process/node responsible for executing it. Ownership determines how the event/action pair is routed at launch time. | # | Action type | Example | Owner | |---|---|---|---| | 1 | Inline recipe method | A plain Python callable defined in the launch script | Main process (Launcher) | | 2 | Component action | A method implemented in a component class, decorated with `@component_action` | The component node | | 3 | System-level action | Actions available in the `actions` module, such as `publish_message`, `send_srv_request`, `send_action_goal` | Main process (Monitor) | | 4 | ROS launch action | Standard `ros2 launch` actions (e.g. `TimerAction`) | Main process (Launcher) | #### Registering Actions Actions are associated with events through the `Launcher.add_pkg()` method: ```python from ros_sugar.core import Action, Event stop_action = Action(my_component.emergency_stop) launcher.add_pkg( components=[my_component], events_actions={low_battery: stop_action}, ) ``` `Launcher.on(event, action)` registers the same mapping one pair at a time, which reads better with plugin-provided factories: ```python launcher.on(low_battery, stop_action) launcher.on(robot.events.fall_detected(), [LogInfo(msg="fall"), robot.actions.stand_up()]) ``` #### The `@component_action` Decorator Marks a component method as callable from the event system. It enforces: - The method must be a bound method on a `LifecycleNode` subclass. - The return type must be `bool` or `None`. - If `active=True` is passed, the method only executes when the component is in the `ACTIVE` lifecycle state. ```python from ros_sugar.utils import component_action class Navigator(BaseComponent): # Basic usage @component_action def stop(self) -> bool: self.cmd_vel_publisher.publish(Twist()) return True # With an OpenAI-compatible tool description (for LLM orchestration) @component_action(description={ "type": "function", "function": { "name": "navigate_to", "description": "Navigate the robot to the specified coordinates.", "parameters": { "type": "object", "properties": { "x": {"type": "number"}, "y": {"type": "number"}, }, }, }, }) def navigate_to(self, *, x: float, y: float) -> bool: ... ``` When `description` is provided, it is stored on the wrapper as `_action_description` and can be used by orchestrating LLM agents to discover available tools. When omitted, the method's docstring is used instead. #### System-Level Actions Provided by the `ros_sugar.actions` module and executed by the Monitor node: | Action | Description | |---|---| | `publish_message(topic, msg, ...)` | Publishes a message on a topic (optionally at a rate for a duration) | | `send_srv_request(srv_name, srv_type, srv_request_msg)` | Sends a ROS2 service request | | `send_action_goal(server_name, server_type, request_msg)` | Sends a ROS2 action goal | #### Dynamic Arguments from Topics Actions can receive live data from ROS topics as arguments. Instead of passing a static value, pass a `topic.msg.attribute` expression — the framework will automatically extract the value from the event's topic data at runtime and inject it into the method call. ::::{tab-set} :::{tab-item} Positional args ```python sensor = Topic(name="/sensor", msg_type="Float32") def handle_reading(value: float): print(f"Sensor reading: {value}") event = Event(event_condition=sensor) action = Action(method=handle_reading, args=(sensor.msg.data,)) ``` ::: :::{tab-item} Keyword args ```python odom = Topic(name="/odom", msg_type=Odometry) def navigate(x: float, y: float): print(f"Going to ({x}, {y})") event = Event(event_condition=odom) action = Action( method=navigate, kwargs={ "x": odom.msg.pose.pose.position.x, "y": odom.msg.pose.pose.position.y, }, ) ``` ::: :::{tab-item} Mixed (static + dynamic) ```python def log_alert(level: str, value: float): print(f"[{level}] value = {value}") # "level" is static, "value" comes from the topic at runtime action = Action(method=log_alert, args=("WARNING", sensor.msg.data)) ``` ::: :::: These expressions (`topic.msg.data`, `odom.msg.pose.pose.position.x`, etc.) are `MsgConditionBuilder` objects — the same ones used to build conditions. When used as action arguments, they tell the framework which topic and which nested attribute to extract at execution time. --- ### Fallback System The fallback system provides automatic failure recovery. It is managed by `ComponentFallbacks` (defined in `ros_sugar.core.fallbacks`). #### ComponentFallbacks `ComponentFallbacks` holds a set of `Fallback` objects, one for each failure level: | Attribute | Triggered When | |:----------|:---------------| | `on_algorithm_fail` | `Status` reports `STATUS_FAILURE_ALGORITHM_LEVEL` | | `on_component_fail` | `Status` reports `STATUS_FAILURE_COMPONENT_LEVEL` | | `on_system_fail` | `Status` reports `STATUS_FAILURE_SYSTEM_LEVEL` | | `on_any_fail` | Any failure without a specific fallback defined | | `on_giveup` | All fallbacks for a failure level have been exhausted | #### Defining Fallbacks Each `Fallback` wraps one or more `Action` instances and a `max_retries` count: ```python from ros_sugar.core import Action, ComponentFallbacks, Fallback fallbacks = ComponentFallbacks( on_component_fail=Fallback( action=[Action(component.restart), Action(component.shutdown)], max_retries=3, ), on_algorithm_fail=Fallback( action=Action(component.reset_algorithm), max_retries=5, ), ) ``` #### Failure Hierarchy When a failure is detected, `ComponentFallbacks` follows this resolution order: 1. Look for a fallback specific to the failure level (`on_algorithm_fail`, `on_component_fail`, or `on_system_fail`). 2. If no specific fallback is defined, fall back to `on_any_fail`. 3. For each fallback, execute the current action up to `max_retries` times. 4. If `max_retries` is exhausted and the fallback has a list of actions, move to the next action in the list. 5. If all actions in the list are exhausted, set the `giveup` flag and execute `on_giveup` if defined. A successful fallback execution (action returns `True`) resets the health status to `STATUS_HEALTHY`. Fallbacks run inside the component, not in the Monitor: a timer at `config.fallback_rate` checks the component's own `health_status` and walks the hierarchy above. A failure for which neither a level-specific fallback nor `on_any_fail` is defined is reported once in the log and kept in the broadcast status; nothing is retried. --- ## Part 2 — Architecture & Routing Internals ### Event/Action Routing — Who keeps track of What When you associate events with actions in the launch script via `launcher.add_pkg(events_actions={...})`, the Launcher inspects each action to determine its owner, then routes the event to the appropriate process. This routing logic lives in the Launcher's `__rewrite_actions_for_components` method. #### Topic-Based Conditions The event can be associated with a list of actions, and **whoever owns the action, owns the event monitoring**. If one event maps to actions with different owners, the event monitoring is duplicated: each owner subscribes to the event topic independently and triggers only its own action. **Routing rules for each action in the list:** ``` Action is a @component_action? ├─ Yes, lifecycle action (start/stop/restart) → ROS launch event handler (Launcher) ├─ Yes, non-lifecycle → Serialized to component (_components_events_actions) ├─ No, system-level (publish_message, etc.) → Monitor (_monitor_events_actions) └─ No, inline recipe method / ROS launch → ROS launch event handler (Launcher, via _internal_events) ``` #### Callable-Based Conditions Callable-based conditions are always defined in the recipe, so they are always **owned by the main process**. The Monitor polls them via a timer. The routing then depends on who owns the consequence action: **Case 1 — Action owned by the main process (recipe method, monitor action, or ROS launch action):** The event and action stay together in the main process. The Monitor polls the callable, and on trigger either executes the action directly (monitor actions) or emits back to the Launcher context (recipe methods and ROS launch actions). **Case 2 — Action owned by a component:** The callable condition runs in the main process, but the action must execute inside the component's node. Since these live in different processes, a **bridge event** is created: 1. The Launcher creates a bridge topic: `/event_bridge/e_{event_id}_{component_name}` of type `std_msgs/Bool`. 2. The Monitor polls the callable condition at `check_rate`. When it returns `True`, the Monitor publishes `Bool(True)` to the bridge topic. 3. The target component subscribes to the bridge topic. On receiving the message, it evaluates the (trivially true) on-any condition and executes the associated `@component_action`. ``` Monitor (main process) Component (separate process) ┌──────────────────────┐ ┌──────────────────────────┐ │ Timer (check_rate) │ │ │ │ ↓ │ │ │ │ callable() == True? │ Bool(True) │ Subscription callback │ │ ↓ yes │ ──────────────────→ │ ↓ │ │ publish to bridge │ /event_bridge/... │ on-any condition → True │ │ │ │ ↓ │ └──────────────────────┘ │ execute @component_action│ └──────────────────────────┘ ``` --- ### Low-Level Implementation #### Key Data Structures ##### `EventBlackboardEntry` > Defined in `ros_sugar/core/event.py` A timestamped wrapper around a ROS message. Every time a topic message is received, it is stored as a blackboard entry with: - `msg`: The raw ROS message. - `timestamp`: Unix time of reception. - `id`: A UUID4 for idempotency — prevents the same message instance from triggering the same event twice. The blackboard uses **lazy expiration**: expired or already-processed entries are cleaned up at evaluation time, not by a background sweep. This avoids lock contention and unnecessary timers. ##### `Event` > Defined in `ros_sugar/core/event.py` The runtime trigger unit. Holds the condition (a `Condition` expression, a `Topic`, or a `Callable`), maintains trigger state, and executes registered actions. Key behavioral knobs: - `on_change`: Only fires on a rising edge (false → true transition). - `handle_once`: Fires at most once across the event's lifetime. - `keep_event_delay`: Throttles re-triggers by holding the "under processing" flag for a fixed duration after actions complete. Actions are executed via a shared `ThreadPoolExecutor` (default 10 workers) to avoid blocking the ROS callback thread. ##### `InternalEvent` / `OnInternalEvent` > Defined in `ros_sugar/core/event.py` The bridge between the Monitor node and the ROS2 launch system. `InternalEvent` is a ROS launch event type that carries an `event_name` and `topics_value` dict. `OnInternalEvent` is a ROS launch event handler that matches by event name and injects topic data into the launch entities before execution. --- :::{dropdown} The Monitor Node :open: > Defined in `ros_sugar/core/monitor.py` The Monitor is a ROS2 node that runs in the main process. It is responsible for: 1. **Subscribing to event topics** and evaluating topic-based conditions. 2. **Polling callable-based conditions** via timers. 3. **Executing system-level actions** (publish_message, send_srv_request, etc.). 4. **Emitting internal events** back to the Launcher context for actions the Launcher owns. 5. **Receiving robot-plugin feedback** for topics that have no ROS subscription: the plugin HOST calls `feed_external_topic(channel, msg)` for every decoded message, which enters the blackboard and is evaluated exactly like a message received over ROS. Health status is not the Monitor's job: every component checks its own status and runs its fallbacks on its fallback timer (see Part 1). The Monitor's per-component work is the reconfiguration, lifecycle and `ExecuteMethod` service clients it holds for each one, and activating components on start once they are discovered on the graph. **Activation Flow** (`_activate_event_monitoring`) When the Monitor activates, it: 1. **Reconstructs monitor actions** (`__reconstruct_monitor_actions`): For events in `_monitor_events_actions`, it resolves each action by name to the corresponding Monitor method (e.g., `publish_message`) and registers it on the Event object. 2. **Merges internal events**: Events from `_internal_events` (those that need to emit back to the Launcher) are appended to the Monitor's event list. 3. **Creates the topic blackboard**: A shared `Dict[str, EventBlackboardEntry]` that caches the latest message for each topic across all events. 4. **Builds a topic → events index** (`__events_per_topic`): Maps each unique topic name to the list of events that depend on it, enabling efficient lookup on message arrival. 5. **Creates one ROS subscription per unique topic**: All events sharing a topic share a single subscriber. The callback `__event_topic_callback` updates the blackboard and evaluates all dependent events. Topics registered with `register_external_topic` (robot-plugin feedback) are skipped; their messages arrive through `feed_external_topic` and take the same path from there. 6. **Creates callable-based polling timers** (`__start_callable_based_event_timers`): One timer per callable-based event, polling at `check_rate` Hz (or `config.loop_rate` if not specified). **Topic-Based Condition Evaluation** (`__event_topic_callback`) On every incoming message: 1. The blackboard entry for that topic is updated with the new message, timestamp, and a fresh UUID. 2. All events that depend on this topic are retrieved from `__events_per_topic`. 3. For each event, a **clean cache subset** is built by checking freshness and idempotency for every topic the event needs (via `EventBlackboardEntry.get`). 4. `event.check_condition(clean_cache_subset)` evaluates the condition tree. If triggered, actions are submitted to the thread pool. **Callable-Based Condition Evaluation** Each callable-based event gets its own timer. On each tick: 1. `event.check_action_condition(blackboard)` calls the user-supplied callable directly. 2. If it returns `True` (accounting for `on_change` rising-edge logic), the registered actions are submitted to the thread pool. ::: :::{dropdown} The Launcher > Defined in `ros_sugar/launch/launcher.py` The Launcher is the entry point of a Sugarcoat application. It is **not** a ROS2 node — it orchestrates the ROS2 launch system. Its responsibilities regarding events: **Action Routing** (`__rewrite_actions_for_components`) For each event/action pair provided by the user, the Launcher classifies the action and routes it to the appropriate owner: - **Component actions** (non-lifecycle): Serialized into `_components_events_actions`. The serialized event JSON is later deserialized by the component at startup. - **Monitor actions**: Stored in `_monitor_events_actions`, passed directly to the Monitor node at initialization. - **Launcher-owned actions** (inline methods, ROS launch actions, lifecycle actions): Stored in `_ros_events_actions` and the event is added to `_internal_events`. For **callable-based events** the routing is handled by `__route_action_based_event`, which either keeps the event in the Monitor (Case 1) or creates a bridge topic (Case 2), as described above. **Internal Events Handler Setup** (`_setup_internal_events_handlers`) For events routed to `_ros_events_actions`, the Launcher: 1. Converts each action into a launch entity: - ROS launch actions are used directly. - Lifecycle actions are converted via `_get_action_launch_entity`. - Inline recipe methods are wrapped as `OpaqueFunction` via `action.launch_action(monitor_node=...)`. 2. Registers an `OnInternalEvent` handler for each event name, wrapping the entities list. 3. Adds the handler to the launch description. At runtime, when the Monitor detects a trigger for one of these events, it emits an `InternalEvent` to the launch context. The `OnInternalEvent` handler matches by event name, injects the topic data into the entities, and executes them. **Monitor ↔ Launcher Emission Bridge** (`ComponentLaunchAction`) > Defined in `ros_sugar/launch/launch_actions.py` When the Monitor's `ComponentLaunchAction` executes, it registers the `_on_internal_event` callback on every internal event: - **Topic-based internal events**: `event.register_actions(partial(self._on_internal_event, event.id))` — the emit callback is registered as an action on the Event object. When the event triggers, it calls the callback which emits an `InternalEvent` to the launch context. - **Callable-based internal events** (`_pure_internal_events`): `_register_pure_internal_event_emit_method(event_id, ...)` stores the emit callback in the Monitor's `emit_internal_event_methods` dict. The `_on_internal_event` method: 1. Creates an `InternalEvent` with the event name. 2. Snapshots the Monitor's topic blackboard into `topics_value`. 3. Emits the event to the launch context via `context.emit_event_sync`, using `call_soon_threadsafe` for thread safety. ::: :::{dropdown} The Component > Defined in `ros_sugar/core/component.py` Components handle events that are routed to them via `_components_events_actions`. The mechanism mirrors the Monitor's topic-based flow. **Event Setup** (`_turn_on_events_management`) Called during `on_activate()`. The component: 1. Creates a topic blackboard (`_events_topics_blackboard`). 2. Builds a topic → events index (`__events_per_topic`). 3. Registers actions on each event via `event.register_actions(actions)`. 4. Creates one ROS subscription per unique topic — including bridge topics for callable-based events. **Event Evaluation** (`__event_topic_callback`) Identical to the Monitor's flow: update blackboard → lazy cleanup → `event.check_condition(clean_cache_subset)` → async action execution. Components **never poll callable conditions directly**. If a callable condition needs to trigger a component action, the bridge mechanism converts it into a topic-based event from the component's perspective. ::: :::{dropdown} The Action Class > Defined in `ros_sugar/core/action.py` The `Action` class wraps a callable and manages argument preparation, dynamic topic data extraction, and conversion to ROS launch entities. **Construction and Argument Classification** (`__verify_args_kwargs`) When an `Action` is constructed, its `args` and `kwargs` are scanned for `MsgConditionBuilder` objects (expressions like `topic.msg.data`). These are separated from static values: - **Static values** are stored directly in `_args` (tuple) and `_kwargs` (dict) and passed to the method on every call. - **Dynamic values** (`MsgConditionBuilder` instances) are stored in a separate `__input_topics` dict, keyed as `arg_{index}` for positional arguments or `kwarg_{name}` for keyword arguments. Each entry records the topic name and the attribute path to extract at runtime. **Execution** (`__call__`) When an event triggers, the `Event` object calls `action(topics=global_topic_cache)` where `global_topic_cache` is a dict mapping topic names to their latest ROS messages. The `Action.__call__` method then: 1. Creates mutable copies of the static `_args` and `_kwargs`. 2. Iterates over `__input_topics`. For each entry: - Looks up the topic's message in the `topics` dict. - Calls `topic_condition.get_value(object_value=message)` which walks the stored attribute path (e.g., `["pose", "pose", "position", "x"]`) to extract the nested value from the message. - Inserts the value into `call_args` (by index) or `call_kwargs` (by name). 3. Runs any registered automatic type conversions (`__prepared_events_conversions`). 4. Calls the underlying `executable` with the fully prepared arguments. **Automatic Type Conversion** (`_setup_conversions`) When an event involves a single topic, the `Event` calls `action._setup_conversions(topic_name, msg_type)` at registration time. This uses `_create_auto_topic_parser` to attempt an automatic conversion from the event's message type to the action's expected input types, using three strategies in order: 1. **Exact match**: Input and target are the same type — pass through directly. 2. **Duck typing**: All target fields exist in the input with matching types — copy matching fields. 3. **Type-based heuristic**: Field names differ but types match uniquely — map by type (with a warning). If a conversion is found, it is stored and applied automatically during `__call__`. **Wrapping for ROS Launch** (`launch_action`) Inline recipe methods and ROS launch actions need to execute within the Launcher's launch context. The `launch_action` method converts an `Action` into a launch-compatible entity: 1. If the action is a monitor action (`_is_monitor_action`), it resolves the executable from the Monitor node by name. 2. Wraps the executable in a new function that prepends a `LaunchContext` parameter (required by the ROS launch framework). 3. Updates the function's `__signature__` so that ROS launch's introspection (`inspect.signature`) sees the `LaunchContext` parameter. 4. Returns an `OpaqueFunction` (for synchronous methods) or `OpaqueCoroutine` (for async methods). At runtime, when the Launcher's `OnInternalEvent` handler fires, it injects the `topics` data into the `OpaqueFunction`'s kwargs before executing it, so the action receives the event's topic cache just as it would when called directly by the Monitor. ::: --- ### End-to-End Flows ::::{tab-set} :::{tab-item} Topic-Based Flows **Flow 1: Topic → Monitor Action** ``` ROS Topic → Monitor subscription → blackboard update → condition evaluation → trigger → ThreadPoolExecutor → Monitor method (e.g. publish_message) ``` **Flow 2: Topic → Component Action** ``` ROS Topic → Component subscription → blackboard update → condition evaluation → trigger → ThreadPoolExecutor → @component_action method ``` **Flow 3: Topic → Launcher-Owned Action** ``` ROS Topic → Monitor subscription → blackboard update → condition evaluation → trigger → _on_internal_event → emit InternalEvent to launch context → OnInternalEvent handler matches → execute OpaqueFunction (inline method) ``` ::: :::{tab-item} Callable-Based Flows **Flow 4: Callable → Monitor Action** ``` Timer (check_rate) → callable() → True → trigger → ThreadPoolExecutor → Monitor method ``` **Flow 5: Callable → Component Action (Bridge)** ``` Timer (check_rate) → callable() → True → Monitor publishes Bool(True) to /event_bridge/... → Component subscription → blackboard update → on-any condition → True → ThreadPoolExecutor → @component_action method ``` **Flow 6: Callable → Launcher-Owned Action** ``` Timer (check_rate) → callable() → True → _on_internal_event → emit InternalEvent to launch context → OnInternalEvent handler matches → execute OpaqueFunction ``` ::: :::: ``` ## File: development/testing.md ```markdown # Testing Guide This document covers how the Sugarcoat test suite is organized, how to run it, and the patterns used to test configuration, events, actions, fallbacks, components, robot plugins and whole recipes. Every snippet below is a trimmed copy of something in `test/`. ## Prerequisites Build and source a workspace containing Sugarcoat, then install pytest: ```bash cd ~/ros_ws colcon build --packages-select automatika_ros_sugar source install/setup.bash pip install pytest ``` The UI API tests additionally need `starlette` and `httpx` and are skipped when those are missing. `launch_testing` ships with ROS 2. ## Running the Suite Tests live in `test/` and are named `*_test.py`; the launch-based component tests are under `test/component/`. Run them with pytest from the repository root, as CI does: ```bash python3 -m pytest test/ -p no:anyio -q # one file, one test python3 -m pytest test/robot_plugin_test.py -p no:anyio -q python3 -m pytest test/io_types_test.py -p no:anyio -q -k image ``` `colcon test` runs nothing for this package: the CMake testing block is disabled until the package's test dependencies are available on the ROS build farm. Things worth knowing: - Launch-based tests print the whole launch log. Use `-q` and redirect to a file when running the full suite. - Launch-based tests discover nodes by name over DDS. Two suites running at the same time on one machine see each other's nodes; give each its own `ROS_DOMAIN_ID`. - ROS 2 Humble's generated message classes check field types on every assignment, newer distributions only when `ROS_PYTHON_CHECK_FIELDS=1` is set. Run with that variable before pushing to catch mistakes such as assigning `0` to a `bool` field, which fails only on Humble in CI otherwise. - CI (`.github/workflows/tests.yml`) runs the suite inside `ros:humble`, `jazzy`, `kilted`, `lyrical` and `rolling` containers after a `colcon build`. ## rclpy Fixtures Each test module owns its rclpy setup; there is no shared `conftest.py`. Other modules in the same session may already have initialized rclpy (`Launcher.__init__` does, and never shuts it down), so fixtures tolerate a live context instead of asserting on it: ```python import pytest import rclpy @pytest.fixture(scope="module", autouse=True) def ros_context(): if not rclpy.ok(): rclpy.init() yield ``` Tests that only exercise configuration, conditions, actions, fallbacks, serialization or the shared-memory ring need no ROS context at all. ## Testing Configuration `BaseAttrs` configs load from a file section named after the component and validate on every assignment: ```python import pytest from attrs import define, field from ros_sugar.config import BaseComponentConfig, base_validators @define(kw_only=True) class MyConfig(BaseComponentConfig): threshold: float = field(default=0.5, validator=base_validators.in_range(0.0, 1.0)) def test_config_from_file(tmp_path): config_file = tmp_path / "config.yaml" config_file.write_text("my_component:\n loop_rate: 50.0\n threshold: 0.8\n") config = MyConfig() assert config.from_file(str(config_file), nested_root_name="my_component") assert config.loop_rate == 50.0 assert config.threshold == 0.8 def test_config_rejects_out_of_range(): with pytest.raises(ValueError): MyConfig(threshold=2.0) ``` `from_file` returns `False` when the file has no section for the component. Round-trips through `to_json()` / `from_json()` are worth a test for any config that crosses the multiprocess boundary; `test/robot_config_test.py` covers the robot description this way. ## Testing Events and Actions ### Event conditions An `Event` evaluates against a blackboard: a dict from topic name to `EventBlackboardEntry`. Key it by `topic.name`, which has the leading slash stripped: ```python import time from std_msgs.msg import Float32 as ROSFloat32 from ros_sugar.core import Event from ros_sugar.core.event import EventBlackboardEntry from ros_sugar.io import Topic from ros_sugar.io.supported_types import Float32 def _blackboard(topic: Topic, value: float): msg = ROSFloat32() msg.data = value return {topic.name: EventBlackboardEntry(msg=msg, timestamp=time.time())} def test_event_triggers(): battery = Topic(name="/battery", msg_type=Float32) event = Event(battery.msg.data < 10.0) event.check_condition(_blackboard(battery, 5.0)) assert event.trigger is True def test_event_does_not_trigger(): battery = Topic(name="/battery", msg_type=Float32) event = Event(battery.msg.data < 10.0) event.check_condition(_blackboard(battery, 95.0)) assert event.trigger is False ``` ### Serialization round-trip Events cross the multiprocess boundary as JSON. Test that a restored event still names its topics and evaluates: ```python def test_event_serialization_round_trip(): temp = Topic(name="/temp", msg_type=Float32) original = Event(temp.msg.data > 100.0) restored = Event.from_json(original.to_json()) assert [t.name for t in restored.get_involved_topics()] == ["temp"] restored.check_condition(_blackboard(temp, 150.0)) assert restored.trigger is True ``` ### Actions An `Action` is called with the triggering messages as `topics`; arguments given as `topic.msg.` expressions are filled from them: ```python from ros_sugar.core import Action def test_action_execution(): called = {} def my_handler(): called["yes"] = True return True action = Action(my_handler) action(topics={}) assert called["yes"] is True def test_action_pulls_arguments_from_the_topic(): sensor = Topic(name="/sensor", msg_type=Float32) seen = [] action = Action(method=seen.append, args=(sensor.msg.data,)) msg = ROSFloat32() msg.data = 3.5 action(topics={sensor.name: msg}) assert seen == [3.5] ``` ### Fallbacks `ComponentFallbacks.execute_*_fallback()` runs the next step of the chain and returns whether it has given up: ```python from ros_sugar.core import Action, ComponentFallbacks, Fallback def test_fallback_retry(): calls = {"n": 0} def failing_action(): calls["n"] += 1 return False fallbacks = ComponentFallbacks( on_component_fail=Fallback(action=Action(failing_action), max_retries=3) ) for _ in range(3): assert fallbacks.execute_component_fallback() is False assert fallbacks.execute_component_fallback() is True assert calls["n"] == 3 ``` ## Testing a Component Without a Launcher A component can be driven directly on a live rclpy context: initialize its node, create the resources under test, and spin it by hand. This is how the plugin binding, TF lifecycle and topic replacement tests work (`test/robot_plugin_test.py`, `test/tf_lifecycle_test.py`): ```python import time import pytest import rclpy from std_msgs.msg import Float32 as ROSFloat32 from ros_sugar.core import BaseComponent from ros_sugar.io import Topic def test_component_receives_its_input(): component = BaseComponent( component_name="doc_example_component", inputs=[Topic(name="sensor", msg_type="Float32")], ) component.rclpy_init_node() try: component.create_all_subscribers() publisher = component.create_publisher(ROSFloat32, "sensor", 10) callback = component.callbacks["sensor"] msg = ROSFloat32() msg.data = 4.2 deadline = time.time() + 2.0 while not callback.got_msg and time.time() < deadline: publisher.publish(msg) rclpy.spin_once(component, timeout_sec=0.05) assert component.got_all_inputs() assert callback.get_output() == pytest.approx(4.2) finally: component.destroy_node() ``` Always destroy the node in a `finally` block; a leaked node keeps its name on the graph for the rest of the session. ## Testing a Robot Plugin Plugins are tested against a mock robot on the loopback interface. A `RobotPluginHost` on an `InProcessFeedbackBus` opens the plugin's transports exactly as the Launcher would, with `node=None` since no ROS node is needed for non-ROS transports. Bind to a free port rather than a fixed one so tests can run in parallel: ```python import socket import time from std_msgs.msg import Int32 as RosInt32 from ros_sugar.robot import ( Feedback, InProcessFeedbackBus, PluginMetadata, RobotPlugin, RobotPluginHost, UdpTransport, create_supported_type, ) def _int32_callback(msg: RosInt32) -> int: return msg.data RobotInt32 = create_supported_type(RosInt32, callback=_int32_callback) def _decode(raw: bytes): msg = RosInt32() msg.data = int(raw.decode()) return msg class MockRobot(RobotPlugin): def __init__(self, port: int = 0): self.metadata = PluginMetadata(name="MockRobot") telemetry = UdpTransport("telemetry", bind=("127.0.0.1", port)) self.transports = {"telemetry": telemetry} self.feedbacks = { "Int32": Feedback(key="Int32", msg_type=RobotInt32, transport=telemetry, decoder=_decode) } def _free_udp_port() -> int: with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s: s.bind(("127.0.0.1", 0)) return s.getsockname()[1] def test_plugin_host_decodes_telemetry(): port = _free_udp_port() plugin = MockRobot(port=port) bus = InProcessFeedbackBus() host = RobotPluginHost(plugin, node=None, bus=bus) host.open() received = [] handle = plugin.subscribe_feedback(plugin.feedbacks["Int32"], received.append) try: with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as robot: robot.sendto(b"42", ("127.0.0.1", port)) deadline = time.time() + 2.0 while not received and time.time() < deadline: time.sleep(0.02) assert received and received[0].data == 42 finally: handle.unsubscribe() host.close() ``` `test/robot_plugin_test.py` extends this pattern to commands, multiprocess specs, the socket bus and the shared-memory fast path; `test/plugin_processes_test.py` covers driver-process declarations against a launcher with fake components; `test/robot_shm_test.py` tests the shared-memory ring on its own, including a cross-process read. ## Testing a Recipe with launch_testing Whole recipes run under `launch_testing`. Build the launch description with the `Launcher` but do not call `bringup()`: `setup_launch_description()` assembles it, `ReadyToTest()` hands control to the test class, and module-level `threading.Event`s carry results out of components and actions: ```python import unittest from threading import Event as ThreadingEvent import launch_testing import launch_testing.actions import launch_testing.markers import pytest from ros_sugar import Launcher from ros_sugar.core import BaseComponent ran_once = ThreadingEvent() class TickingComponent(BaseComponent): def _execution_step(self): ran_once.set() @pytest.mark.launch_test @launch_testing.markers.keep_alive def generate_test_description(): component = TickingComponent(component_name="ticking_component") component.loop_rate = 10.0 launcher = Launcher() launcher.add_pkg(components=[component]) launcher.setup_launch_description() # build, but do not run launcher._description.add_action(launch_testing.actions.ReadyToTest()) return launcher._description class TestRecipe(unittest.TestCase): def test_component_executes(self): assert ran_once.wait(10.0), "the component never ran" ``` The event tests (`test/simple_events_test.py`, `test/composed_events_test.py`, `test/generic_events_test.py`) and the component run-type tests under `test/component/` all follow this shape: a component publishes, the recipe's events fire actions that set threading events, and the test class waits on them with a timeout. The components here run in launcher threads. Multiprocess launch needs an installed package with an executable entry point, so the pieces it relies on are tested directly instead: the process launch action and its prefix in `test/external_launch_test.py`, the socket bus in `test/robot_plugin_test.py`, and the shared-memory ring in `test/robot_shm_test.py`. ``` ## File: advanced/create_service.md ```markdown # Converting a Sugarcoat Recipe into a systemd Service :::{note} Recipes written with Sugarcoat packages can be easily launched as `systemd` services using the `create_service` tool provided with the package. ::: :::{tip} This is ideal for deploying Sugarcoat components in production environments or embedded systems where automatic startup and restart behavior is critical. ::: Once you have a Python script written for your Sugarcoat-based package (lets call it `my_awesome_system.py`), you can install it as a systemd service with the following command: ```bash ros2 run automatika_ros_sugar create_service ``` ## Arguments - ``: The full path to your Sugarcoat Python script (e.g., `/path/to/my_awesome_system.py`). - ``: The name of the systemd service (do **not** include the `.service` extension). ## Example ```bash ros2 run automatika_ros_sugar create_service ~/ros2_ws/my_awesome_system.py my_awesome_service ``` This command will install and optionally enable a `systemd` service named `my_awesome_service.service`. ## Full Command Usage ```text usage: create_service [-h] [--service-description SERVICE_DESCRIPTION] [--install-path INSTALL_PATH] [--source-workspace-path SOURCE_WORKSPACE_PATH] [--no-enable] [--restart-time RESTART_TIME] service_file_path service_name ``` Install a Python script as a systemd service. ### Positional Arguments - **`service_file_path`**: Path to the Python script you want to install as a service. - **`service_name`**: Name of the systemd service (without `.service` extension). ### Optional Arguments - `-h, --help`: Show the help message and exit. - `--service-description SERVICE_DESCRIPTION`: Human-readable description of the service. Defaults to `"Sugarcoat Description"`. - `--install-path INSTALL_PATH`: Directory to install the systemd service file. Defaults to `/etc/systemd/system`. - `--source-workspace-path SOURCE_WORKSPACE_PATH`: Path to the ROS workspace `setup` script. If omitted, it auto-detects the active ROS distribution. - `--no-enable`: Skip enabling the service after installation. - `--restart-time RESTART_TIME`: Time to wait before restarting the service if it fails (e.g., `3s`). Default is `3s`. ## What This Does This command: 1. Creates a `.service` file for `systemd`. 2. Installs it in the specified or default location. 3. Sources the appropriate ROS environment. 4. Optionally enables and starts the service immediately. Once installed, you can manage the service using standard `systemd` commands: ```bash sudo systemctl start my_awesome_service sudo systemctl status my_awesome_service sudo systemctl stop my_awesome_service sudo systemctl enable my_awesome_service ``` ``` ## File: advanced/srvs.md ```markdown # Default Services In Sugarcoat Components In addition to the standard [ROS2 Lifecycle Node](https://github.com/ros2/demos/blob/rolling/lifecycle/README.rst) services, Sugarcoat Components provide a powerful set of built-in services for live reconfiguration. These services allow you to dynamically adjust inputs, outputs, and parameters on-the-fly, making it easier to respond to changing runtime conditions or trigger intelligent behavior in response to events. Like any ROS2 services, they can be called from other Nodes or with the ROS2 CLI, and can also be called programmatically as part of an action sequence or event-driven workflow in the launch script. The examples below assume a component named `awesome_component` with an output topic `/voice` of type `Audio`. ## Replacing an Input or Output with a different Topic You can swap an existing topic connection (input or output) with a different topic online without restarting your script. The service will stop the running lifecycle node, replace the connection and restart it again. - **Service Name: /{component_name}/change_topic** - **Service Type: [automatika_ros_sugar/srv/ReplaceTopic](https://github.com/automatika-robotics/sugarcoat/blob/main/srv/ReplaceTopic.srv)** ### Example To replace the output topic name of `awesome_component`, we can send the following service call to the node: ```shell ros2 service call /awesome_component/change_topic automatika_ros_sugar/srv/ReplaceTopic "{direction: 1, old_name: '/voice', new_name: '/audio_device_0', new_msg_type: 'Audio'}" ``` ## Updating a configuration parameter value This `ChangeParameter` service allows updating a single configuration parameter at runtime. You can choose whether the component remains active during the change, or temporarily deactivates for a safe update. - **Service Name: /{component_name}/update_config_parameter** - **Service Type: [automatika_ros_sugar/srv/ChangeParameter](https://github.com/automatika-robotics/sugarcoat/blob/main/srv/ChangeParameter.srv)** ### Example Let's change the `loop_rate` for `awesome_component` to `1Hz` without restarting the node: ```shell ros2 service call /awesome_component/update_config_parameter automatika_ros_sugar/srv/ChangeParameter "{name: 'loop_rate', value: '1', keep_alive: false}" ``` ## Updating a set of configuration parameters The `ChangeParameters` service allows updating multiple parameters at once, making it ideal for switching modes, profiles, or reconfiguring components in batches. Similar to `ChangeParameter` service, you can choose whether the component stays active or temporarily deactivates during the update. - **Service Name: /{component_name}/update_config_parameters** - **Service Type: [automatika_ros_sugar/srv/ChangeParameters](https://github.com/automatika-robotics/sugarcoat/blob/main/srv/ChangeParameters.srv)** ### Example Let's change multiple parameters at once for `awesome_component` without restarting the node: ```shell ros2 service call /awesome_component/update_config_parameters automatika_ros_sugar/srv/ChangeParameters "{names: ['loop_rate', 'fallback_rate'], values: ['1', '10'], keep_alive: false}" ``` ## Reconfiguring the Component from a given file The `ConfigureFromFile` service lets you reconfigure an entire component from a YAML, JSON or TOML configuration file while the node is online. This is useful for applying scenario-specific settings, or restoring saved configurations—all in a single operation. - **Service Name: /{component_name}/configure_from_file** - **Service Type: [automatika_ros_sugar/srv/ConfigureFromFile](https://github.com/automatika-robotics/sugarcoat/blob/main/srv/ConfigureFromFile.srv)** ### Example Example YAML configuration file for `awesome_component`: ```yaml /**: # Common parameters for all components fallback_rate: 10.0 # Parameters specific to component under component name awesome_component: loop_rate: 100.0 ``` ## Executing a Component's method The `ExecuteMethod` service enables runtime invocation of any class method in the component. This is useful for triggering specific behaviors, tools, or diagnostics during runtime without writing additional interfaces. - **Service Name: /{component_name}/execute_method** - **Service Type: [automatika_ros_sugar/srv/ExecuteMethod](https://github.com/automatika-robotics/sugarcoat/blob/main/srv/ExecuteMethod.srv)** The request carries the method `name` and its keyword arguments as a JSON object in `kwargs_json`. The response reports `success`, an `error_msg` on failure, and the method's return value as JSON in `response_json`: | Method returns | `success` | `response_json` | `error_msg` | |:---------------|:---------:|:----------------|:------------| | `True` | `true` | `true` | | | `None` | `true` | empty | | | any other JSON-serializable value | `true` | the value | | | a value that is not JSON-serializable | `true` | empty | explains the serialization error | | `False` | `false` | | says the method returned `False` | | raises | `false` | | the exception message | Returning `False` is treated as a failure for backward compatibility, so a component action cannot use `False` as a legitimate result. Methods decorated with `@component_action` are the intended targets; the UI, the JSON API and LLM-driven orchestration in EmbodiedAgents all call actions through this service. ### Example Call an action `reset_buffer(size: int)` on `awesome_component`: ```shell ros2 service call /awesome_component/execute_method automatika_ros_sugar/srv/ExecuteMethod "{name: 'reset_buffer', kwargs_json: '{\"size\": 10}'}" ``` ```