std_msgs Adapters

mosaicolabs.ros_bridge.adapters.std_msgs ¶

Standard ROS Message Adapters.

This module provides adapters for translating standard ROS messages (std_msgs) into Mosaico ontology types. Instead of manually defining a class for every single primitive type (Int8, String, Bool, etc.), we use a dynamic factory pattern.

Architecture

_ROS_MSGTYPE_MSCO_BASE_TYPE_MAP defines the relationship between a ROS message type string (e.g., "std_msgs/msg/String") and the corresponding Mosaico Serializable class (e.g., String).
GenericStdAdapter implements the common translate and from_dict logics shared by all standard types (wrapping the 'data' field).
At module load time, we iterate through the mapping, dynamically create a unique subclass of GenericStdAdapter for each type and register it in the ROSBridge.

GenericStdAdapter ¶

Bases: ROSAdapterBase[Serializable]

Template for dynamic factory-based adaptation of standard ROS primitive messages.

This class provides the core translation logic for the std_msgs family. To avoid manual definition of dozens of repetitive classes (e.g., Int8Adapter, StringAdapter), the ROS Bridge employs a Dynamic Factory Pattern.

Supported ROS Types:

Architecture & Dynamic Generation¶

At module load time, the SDK iterates through a configuration mapping (_ROS_MSGTYPE_MSCO_BASE_TYPE_MAP) and programmatically generates concrete subclasses of GenericStdAdapter.

Each generated subclass is:

Injected with a specific ros_msgtype (e.g., "std_msgs/msg/String").
Injected with a specific target __mosaico_ontology_type__ (e.g., String).
Registered automatically in the ROSBridge using the @register_default_adapter mechanism.

"Adaptation" Strategy¶

Following the philosophy of "Adaptation, Not Just Parsing," these adapters do not simply extract raw values. They perform:

Schema Enforcement: Validating that the ROS message contains the mandatory 'data' field.
Strong Typing: Wrapping the primitive value into a Mosaico Serializable object with its own metadata and queryable headers.
Temporal Alignment: Preserving nanosecond-precise timestamps and optional frame information from the source bag file.

Example

# Logic effectively generated by the factory:
class StringStdAdapter(GenericStdAdapter):
    ros_msgtype = "std_msgs/msg/String"
    __mosaico_ontology_type__ = String

# Usage within the Bridge:
ros_msg = ROSMessage(
    timestamp=1707760800.123456789,
    topic="/log",
    msg_type="std_msgs/msg/String",
    data={"data": "System OK"}
)
mosaico_string = StringStdAdapter.translate(ros_msg)

translate `classmethod` ¶

translate(ros_msg, **kwargs)

Translates a standard ROS message to a Mosaico Message.

Standard messages typically contain a 'data' field and metadata. This method extracts the header/timestamp and wraps the payload using the specific ontology type defined for this adapter class.

Parameters:

Name	Type	Description	Default
`ros_msg`	`ROSMessage`	The ROS message to translate.	required
`**kwargs`	`Any`	Additional keyword arguments for translation.	`{}`

Returns:

Name	Type	Description
`Message`	`Message`	The translated message containing the adapter's ontology type instance.

Raises:

Type	Description
`Exception`	Wraps any translation error with context (topic name, timestamp).

from_dict `classmethod` ¶

from_dict(ros_data)

Converts the raw dictionary data into the specific Mosaico type.

Parameters:

Name	Type	Description	Default
`ros_data`	`dict`	The raw dictionary from the ROS message.	required

Returns:

Name	Type	Description
`Serializable`	`Serializable`	The constructed Mosaico ontology instance.

Raises:

Type	Description
`ValueError`	If the 'data' key is missing from `ros_data`.

to_ros `classmethod` ¶

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico scalar wrapper (or a Message wrapping one) into the corresponding std_msgs ROS message.

Parameters:

Name	Type	Description	Default
`mosaico_data`	`Union[Message, Serializable]`	A `Message` wrapping a scalar `Serializable` (e.g. `String`, `Integer32`), or the raw scalar instance directly.	required
`typestore`	`Typestore`	The rosbags typestore for target type resolution.	required
`ros_msg_type`	`Optional[str]`	Override for the output ROS type. If `None`, defaults to `cls.get_default_ros_msg()`.	`None`

Returns:

Name	Type	Description
`MsgType`	`MsgType`	The constructed `std_msgs` ROS message, or raises an error if: the ros_msg_type is unsupported by adapter (TypeError) the ros_msg_type or default type are unsupported by typestore (TypeError)

schema_metadata `classmethod` ¶

schema_metadata(typestore, ros_msg_type, ros_version)

Extract the ROS message specific schema metadata, if any.

Parameters:

Name	Type	Description	Default
`typestore`	`Typestore`	The rosbags typestore for target type resolution.	required
`ros_msg_type`	`str`	The ROS message type to extract metadata for.	required
`ros_version`	`int`	The ROS version (1 or 2) to consider for metadata extraction.	required

Returns:

Type	Description
`Optional[dict]`	Optional[dict]: A dictionary containing the schema metadata, or None if not applicable.

ros_msg_type `abstractmethod` `classmethod` ¶

ros_msg_type()

Returns the specific ROS message type handled by this adapter.

is_rosmsg_type_valid `classmethod` ¶

is_rosmsg_type_valid(type_to_validate)

Checks whether a given ROS message type string is handled by this adapter.

Parameters:

Name	Type	Description	Default
`type_to_validate`	`str`	The full ROS message type string to check (e.g., `"sensor_msgs/msg/Imu"`).	required

Returns:

Name	Type	Description
`bool`	`bool`	`True` if the adapter supports this type, `False` otherwise.

unpack_mosaico_msg `classmethod` ¶

unpack_mosaico_msg(mosaico_msg)

Extracts the typed Mosaico payload and its Header (if present) from a wrapped or bare message.

Handles two input cases:

Message wrapper: the typed data is extracted via get_data().
Raw ontology instance: returned as-is with

the Header is extracted from the ontology (if supported), otherwise an default Header (empty frame_id and zero Time) is returned.

Parameters:

Name	Type	Description	Default
`mosaico_msg`	`Union[Message, T]`	Either a `Message` envelope or a raw instance of `cls.__mosaico_ontology_type__`.	required

Returns:

Type	Description
`T`	tuple[T, Header]: A `(data, header)` tuple where data is the typed ontology object and
`Header`	header is the corresponding `Header`, or a default `Header` (empty `frame_id` and
`tuple[T, Header]`	zero `Time`) if not present.

Raises:

Type	Description
`TypeError`	If mosaico_msg is neither a `Message` nor an instance of the expected ontology type.

ontology_data_type `classmethod` ¶

ontology_data_type()

Returns the Ontology class type associated with this adapter.

HeaderAdapter ¶

Bases: ROSAdapterBase[Header]

Adapter for translating ROS Header messages to Mosaico Header.

Supported ROS Types:

std_msgs/msg/Header

Example

# Internal usage within the ROS Bridge
ros_msg = ROSMessage(
    timestamp=17000,
    topic="/header",
    msg_type="std_msgs/msg/Header",
    data = {
        stamp:
        {
            "sec": 1000,
            "nanosec": 1000000000
        },
        frame_id: "base_link"
    }
)
# Automatically resolves to a flat Mosaico Header with attached metadata
mosaico_header = HeaderAdapter.translate(ros_msg)

translate `classmethod` ¶

translate(ros_msg, **kwargs)

Main entry point for translating a high-level ROSMessage.

Parameters:

Name	Type	Description	Default
`ros_msg`	`ROSMessage`	The source ROS message yielded by the loader.	required
`**kwargs`	`Any`	Additional context for the translation.	`{}`

Returns:

Name	Type	Description
`Message`	`Message`	The translated Mosaico `Message` containing the normalized `Header` payload.

from_dict `classmethod` ¶

from_dict(ros_data)

Parses a dictionary to extract a Header object.

Example (ROS2 does not have seq field):

ros_data = {
    "stamp": {
        "sec": 1000,
        "nanosec": 1000000000
    },
    "frame_id": "base_link"
}
# Automatically resolves to a flat Mosaico Header with attached metadata
mosaico_header = HeaderAdapter.from_dict(ros_data)

Parameters:

Name	Type	Description	Default
`ros_data`	`dict`	The raw dictionary from the ROS message.	required

Returns:

Name	Type	Description
`Header`	`Header`	The constructed Mosaico Header object.

Raises:

Type	Description
`ValueError`	If required keys are missing.

to_ros `classmethod` ¶

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico Header (or a Message wrapping one) into a std_msgs/msg/Header message.

Supported output types (selectable via ros_msg_type): - std_msgs/msg/Header

Parameters:

Name	Type	Description	Default
`mosaico_data`	`Union[Message, Header]`	A `Message` wrapping a `Header` instance, or a raw `Header`.	required
`typestore`	`Typestore`	The rosbags typestore for target type resolution.	required
`ros_msg_type`	`Optional[str]`	Override for the output ROS type. Defaults to `std_msgs/msg/Header` if `None`.	`None`

Returns:

Name	Type	Description
`MsgType`	`MsgType`	A `std_msgs/msg/Header` instance, or raises an error if: - the ros_msg_type is unsupported by adapter (TypeError) - the ros_msg_type or default type are unsupported by typestore (TypeError) - the ros_msg_type or default type are supported but translation is not implemented (NotImplementedError)

ros_msg_type `abstractmethod` `classmethod` ¶

ros_msg_type()

Returns the specific ROS message type handled by this adapter.

is_rosmsg_type_valid `classmethod` ¶

is_rosmsg_type_valid(type_to_validate)

Checks whether a given ROS message type string is handled by this adapter.

Parameters:

Name	Type	Description	Default
`type_to_validate`	`str`	The full ROS message type string to check (e.g., `"sensor_msgs/msg/Imu"`).	required

Returns:

Name	Type	Description
`bool`	`bool`	`True` if the adapter supports this type, `False` otherwise.

unpack_mosaico_msg `classmethod` ¶

unpack_mosaico_msg(mosaico_msg)

Extracts the typed Mosaico payload and its Header (if present) from a wrapped or bare message.

Handles two input cases:

Message wrapper: the typed data is extracted via get_data().
Raw ontology instance: returned as-is with

the Header is extracted from the ontology (if supported), otherwise an default Header (empty frame_id and zero Time) is returned.

Parameters:

Name	Type	Description	Default
`mosaico_msg`	`Union[Message, T]`	Either a `Message` envelope or a raw instance of `cls.__mosaico_ontology_type__`.	required

Returns:

Type	Description
`T`	tuple[T, Header]: A `(data, header)` tuple where data is the typed ontology object and
`Header`	header is the corresponding `Header`, or a default `Header` (empty `frame_id` and
`tuple[T, Header]`	zero `Time`) if not present.

Raises:

Type	Description
`TypeError`	If mosaico_msg is neither a `Message` nor an instance of the expected ontology type.

schema_metadata `classmethod` ¶

schema_metadata(typestore, ros_msg_type, ros_version)

Extract the ROS message specific schema metadata, if any.

Parameters:

Name	Type	Description	Default
`typestore`	`Typestore`	The rosbags typestore for target type resolution.	required
`ros_msg_type`	`str`	The ROS message type to extract metadata for.	required
`ros_version`	`int`	The ROS version (1 or 2) to consider for metadata extraction.	required

Returns:

Type	Description
`Optional[dict]`	Optional[dict]: A dictionary containing the schema metadata, or None if not applicable.

For the BatteryStateAdapter the expected output is { "ros": { "enums": { "POWER_SUPPLY_STATUS_UNKNOWN": 0, "POWER_SUPPLY_STATUS_CHARGING": 1, "POWER_SUPPLY_STATUS_DISCHARGING": 2, "POWER_SUPPLY_STATUS_NOT_CHARGING": 3, "POWER_SUPPLY_STATUS_FULL": 4, "POWER_SUPPLY_HEALTH_UNKNOWN": 0, "POWER_SUPPLY_HEALTH_GOOD": 1, "POWER_SUPPLY_HEALTH_OVERHEAT": 2, "POWER_SUPPLY_HEALTH_DEAD": 3, "POWER_SUPPLY_HEALTH_OVERVOLTAGE": 4, "POWER_SUPPLY_HEALTH_UNSPEC_FAILURE": 5, "POWER_SUPPLY_HEALTH_COLD": 6, "POWER_SUPPLY_HEALTH_WATCHDOG_TIMER_EXPIRE": 7, "POWER_SUPPLY_HEALTH_SAFETY_TIMER_EXPIRE": 8, "POWER_SUPPLY_TECHNOLOGY_UNKNOWN": 0, "POWER_SUPPLY_TECHNOLOGY_NIMH": 1, "POWER_SUPPLY_TECHNOLOGY_LION": 2, "POWER_SUPPLY_TECHNOLOGY_LIPO": 3, "POWER_SUPPLY_TECHNOLOGY_LIFE": 4, "POWER_SUPPLY_TECHNOLOGY_NICD": 5, "POWER_SUPPLY_TECHNOLOGY_LIMN": 6, }, "msgtype": "sensor_msgs/msg/BatteryState" "msgdef": "..." } }

ontology_data_type `classmethod` ¶

ontology_data_type()

Returns the Ontology class type associated with this adapter.

std_msgs Adapters

mosaicolabs.ros_bridge.adapters.std_msgs ¶

GenericStdAdapter ¶

Architecture & Dynamic Generation¶

"Adaptation" Strategy¶

translate classmethod ¶

from_dict classmethod ¶

to_ros classmethod ¶

schema_metadata classmethod ¶

ros_msg_type abstractmethod classmethod ¶

is_rosmsg_type_valid classmethod ¶

unpack_mosaico_msg classmethod ¶

ontology_data_type classmethod ¶

HeaderAdapter ¶

translate classmethod ¶

from_dict classmethod ¶

to_ros classmethod ¶

ros_msg_type abstractmethod classmethod ¶

is_rosmsg_type_valid classmethod ¶

unpack_mosaico_msg classmethod ¶

schema_metadata classmethod ¶

ontology_data_type classmethod ¶

translate `classmethod` ¶

from_dict `classmethod` ¶

to_ros `classmethod` ¶

schema_metadata `classmethod` ¶

ros_msg_type `abstractmethod` `classmethod` ¶

is_rosmsg_type_valid `classmethod` ¶

unpack_mosaico_msg `classmethod` ¶

ontology_data_type `classmethod` ¶

translate `classmethod` ¶

from_dict `classmethod` ¶

to_ros `classmethod` ¶

ros_msg_type `abstractmethod` `classmethod` ¶

is_rosmsg_type_valid `classmethod` ¶

unpack_mosaico_msg `classmethod` ¶

schema_metadata `classmethod` ¶

ontology_data_type `classmethod` ¶