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 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 three extension points:

class SupportedType:
    # The ROS 2 message class (e.g., std_msgs.msg.String)
    _ros_type: type

    # Callback class for deserializing incoming messages
    callback = callbacks.GenericCallback

    @classmethod
    def convert(cls, output, **_) -> Any:
        """Convert Python data into a ROS message instance."""
        return output

    @classmethod
    def get_ros_type(cls) -> type:
        """Return the underlying ROS 2 message class."""
        return cls._ros_type

_ros_type¶

Class attribute that holds the ROS 2 message class. This is used to create subscriptions, validate topic compatibility, and generate UI schemas.

callback¶

A callback class (typically a subclass of GenericCallback) that handles deserialization of incoming ROS messages. Different types use specialized callbacks – for example, ImageCallback for sensor images, OdomCallback for odometry, StdMsgCallback for simple std_msgs types.

convert¶

A classmethod that takes Python-native data and returns a ROS message instance. The first positional argument should be named output. This is called by Publisher when a component publishes output. For example, Image.convert() accepts either a numpy.ndarray or an existing ROSImage and returns a ROSImage.

Built-in Types¶

Sugarcoat ships with the following built-in types in ros_sugar.io.supported_types:

Type	ROS Message	Callback
`String`	`std_msgs/String`	`TextCallback`
`Bool`	`std_msgs/Bool`	`StdMsgCallback`
`Float32`	`std_msgs/Float32`	`StdMsgCallback`
`Float32MultiArray`	`std_msgs/Float32MultiArray`	`StdMsgArrayCallback`
`Float64`	`std_msgs/Float64`	`StdMsgCallback`
`Float64MultiArray`	`std_msgs/Float64MultiArray`	`StdMsgArrayCallback`
`Image`	`sensor_msgs/Image`	`ImageCallback`
`CompressedImage`	`sensor_msgs/CompressedImage`	`CompressedImageCallback`
`Audio`	`std_msgs/ByteMultiArray`	`AudioCallback`
`MapMetaData`	`nav_msgs/MapMetaData`	`MapMetaDataCallback`
`Odometry`	`nav_msgs/Odometry`	`OdomCallback`
`LaserScan`	`sensor_msgs/LaserScan`	`GenericCallback`
`Path`	`nav_msgs/Path`	`PathCallback`
`OccupancyGrid`	`nav_msgs/OccupancyGrid`	`OccupancyGridCallback`
`Point`	`geometry_msgs/Point`	`PointCallback`
`PointStamped`	`geometry_msgs/PointStamped`	`PointStampedCallback`
`Pose`	`geometry_msgs/Pose`	`PoseCallback`
`PoseStamped`	`geometry_msgs/PoseStamped`	`PoseStampedCallback`
`ComponentStatus`	`automatika_ros_sugar/ComponentStatus`	`GenericCallback`
`Twist`	`geometry_msgs/Twist`	(default)

Registering Additional Types¶

Use add_additional_datatypes() to register custom types at runtime:

from ros_sugar.io.supported_types import add_additional_datatypes

add_additional_datatypes([MyCustomType, AnotherType])

This function maintains a global _additional_types dictionary. When a type with the same __name__ already exists, the function merges callbacks and conversion functions rather than replacing the existing entry. This allows multiple packages to augment the same type with additional callbacks.

Step-by-Step: Adding a Custom Type¶

Suppose you want to add support for sensor_msgs/Range:

1. Define the type class¶

from sensor_msgs.msg import Range as ROSRange
from ros_sugar.io.supported_types import SupportedType
from ros_sugar.io.callbacks import GenericCallback


class Range(SupportedType):
    """Range sensor message support."""

    _ros_type = ROSRange
    callback = GenericCallback

    @classmethod
    def convert(cls, output: float, **_) -> ROSRange:
        msg = ROSRange()
        msg.range = output
        return msg

2. Register the type¶

from ros_sugar.io.supported_types import add_additional_datatypes

add_additional_datatypes([Range])

3. Use it in a Topic¶

from ros_sugar.io import Topic

range_topic = Topic(name="/front_sonar", msg_type=Range)

The topic can now be used as an input or output on any BaseComponent.

Custom Callbacks¶

If the default GenericCallback is insufficient (for example, you need to extract specific fields or perform numpy conversions), create a custom callback class inheriting from GenericCallback:

from ros_sugar.io.callbacks import GenericCallback

class RangeCallback(GenericCallback):
    def __call__(self, msg):
        # Extract and store just the range value
        self.output = msg.range

Then assign it to your type:

class Range(SupportedType):
    _ros_type = ROSRange
    callback = RangeCallback
    # ...

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:

# 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 overwritten if the existing type has no _ros_type set.
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.