geometry_msgs Adapters
mosaicolabs.ros_bridge.adapters.geometry_msgs ¶
Geometry Messages Adaptation Module.
This module provides specialized adapters for translating ROS geometry_msgs into the
standardized Mosaico Ontology. It implements recursive unwrapping to handle common
ROS patterns, such as "Stamped" envelopes and covariance wrappers, ensuring that
spatial data is normalized before ingestion.
PoseAdapter ¶
Bases: ROSAdapterBase[Pose]
Adapter for translating ROS Pose-related messages to Mosaico Pose.
This adapter follows the "Adaptation, Not Just Parsing" philosophy by actively
unwrapping nested ROS structures and normalizing them into strongly-typed
Mosaico Pose objects.
Supported ROS Types:
geometry_msgs/msg/Posegeometry_msgs/msg/PoseStampedgeometry_msgs/msg/PoseWithCovariancegeometry_msgs/msg/PoseWithCovarianceStamped
Recursive Unwrapping Strategy:
The adapter checks for nested 'pose' keys. If found (as in PoseStamped),
it recurses to the leaf node while collecting metadata like headers and
covariance matrices along the way.
Example
# Internal usage within the ROS Bridge
ros_msg = ROSMessage(
timestamp=17000,
topic="/pose",
msg_type="geometry_msgs/msg/PoseStamped",
data={
"header": {"frame_id": "map", "stamp": {"sec": 17000, "nanosec": 0}},
"pose": {
"position": {"x": 1.0, "y": 2.0, "z": 0.0},
"orientation": {"x": 0, "y": 0, "z": 0, "w": 1}
},
}
)
# Automatically resolves to a flat Mosaico Pose with attached metadata
mosaico_pose = PoseAdapter.translate(ros_msg)
translate
classmethod
¶
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
|
A Mosaico |
from_dict
classmethod
¶
Recursively parses a dictionary to extract a Pose object.
Strategy:
- Recurse: If a 'pose' key is found, dive deeper into the structure.
- Leaf Node: At the base level, map 'position' and 'orientation' to
Point3dandQuaternion. - Metadata Binding: Covariances are attached during recursion unwinding.
Example
ros_data = {
"header": {"frame_id": "map", "stamp": {"sec": 17000, "nanosec": 0}},
"pose": {
"position": {"x": 1.0, "y": 2.0, "z": 0.0},
"orientation": {"x": 0, "y": 0, "z": 0, "w": 1}
},
}
# Automatically resolves to a flat Mosaico Pose with attached metadata
mosaico_pose = PoseAdapter.from_dict(ros_data)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ros_data
|
dict
|
The raw dictionary from the ROS message. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Pose |
Pose
|
The constructed Mosaico Pose object. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the recursive 'pose' key exists but is not a dict, or if required keys are missing. |
to_ros
classmethod
¶
Converts a Mosaico Pose (or a Message wrapping one) into the
corresponding ROS geometry message.
Supported output types (selectable via ros_msg_type):
geometry_msgs/msg/Posegeometry_msgs/msg/PoseStampedgeometry_msgs/msg/PoseWithCovariancegeometry_msgs/msg/PoseWithCovarianceStamped
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mosaico_data
|
Union[Message, Pose]
|
A |
required |
typestore
|
Typestore
|
The rosbags typestore for target type resolution. |
required |
ros_msg_type
|
Optional[str]
|
Override for the output ROS type. Defaults to
|
None
|
Returns:
| Name | Type | Description |
|---|---|---|
MsgType |
MsgType
|
The constructed ROS message, or raises an error if:
|
schema_metadata
classmethod
¶
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
¶
Returns the specific ROS message type handled by this adapter.
is_rosmsg_type_valid
classmethod
¶
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., |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
|
unpack_mosaico_msg
classmethod
¶
Extracts the typed Mosaico payload and its Header (if present) from a wrapped or bare message.
Handles two input cases:
Messagewrapper: the typed data is extracted viaget_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 |
required |
Returns:
| Type | Description |
|---|---|
T
|
tuple[T, Header]: A |
Header
|
header is the corresponding |
tuple[T, Header]
|
zero |
Raises:
| Type | Description |
|---|---|
TypeError
|
If mosaico_msg is neither a |
ontology_data_type
classmethod
¶
Returns the Ontology class type associated with this adapter.
TwistAdapter ¶
Bases: ROSAdapterBase[Velocity]
Adapter for translating ROS Twist-related messages to Mosaico Velocity.
Commonly referred to as a "Twist," this model captures the instantaneous motion of an object split into linear and angular components.
Supported ROS Types:
geometry_msgs/msg/Twistgeometry_msgs/msg/TwistStampedgeometry_msgs/msg/TwistWithCovariancegeometry_msgs/msg/TwistWithCovarianceStamped
Recursive Unwrapping Strategy:
The adapter checks for nested 'twist' keys. If found (as in TwistStamped),
it recurses to the leaf node while collecting metadata like headers and
covariance matrices along the way.
Example
ros_msg = ROSMessage(
timestamp=1700000000000,
topic="/cmd_vel",
msg_type="geometry_msgs/msg/TwistStamped",
data = {
"header": {"frame_id": "map", "stamp": {"sec": 17000, "nanosec": 0}},
"twist": {
"linear": {"x": 5.0, "y": 0.0, "z": 0.0},
"angular": {"x": 0.0, "y": 0.0, "z": 1.0}
},
"covariance": [0.1] * 36
}
)
# Automatically resolves to a flat Mosaico Velocity with attached metadata
mosaico_velocity = TwistAdapter.translate(ros_msg)
translate
classmethod
¶
Translates a ROS message into a Mosaico Message.
Returns:
| Name | Type | Description |
|---|---|---|
Message |
Message
|
The translated message containing a |
Raises:
| Type | Description |
|---|---|
Exception
|
Wraps any translation error with context (topic name, timestamp). |
from_dict
classmethod
¶
Recursively parses the ROS data dictionary to extract a Velocity (Twist).
Strategy:
- Recurse: If a 'twist' key is found, dive deeper into the structure.
- Leaf Node: At the base level, map 'linear' and 'angular' to
Vector3.
- Metadata Binding: Covariances are attached during
recursion unwinding.
Example
ros_data = {
"header": {"frame_id": "map", "stamp": {"sec": 17000, "nanosec": 0}},
"twist": {
"linear": {"x": 5.0, "y": 0.0, "z": 0.0},
"angular": {"x": 0.0, "y": 0.0, "z": 1.0}
},
"covariance": [0.1] * 36
}
# Automatically resolves to a flat Mosaico Velocity with attached metadata
mosaico_velocity = TwistAdapter.from_dict(ros_data)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ros_data
|
dict
|
The raw dictionary from the ROS message. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Velocity |
Velocity
|
The constructed Mosaico Velocity object. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the recursive 'twist' key exists but is not a dict, or if required keys are missing. |
to_ros
classmethod
¶
Converts a Mosaico Velocity (or a Message wrapping one) into the
corresponding ROS Twist message.
Supported output types (selectable via ros_msg_type):
geometry_msgs/msg/Twistgeometry_msgs/msg/TwistStampedgeometry_msgs/msg/TwistWithCovariancegeometry_msgs/msg/TwistWithCovarianceStamped
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mosaico_data
|
Union[Message, Velocity]
|
A |
required |
typestore
|
Typestore
|
The rosbags typestore for target type resolution. |
required |
ros_msg_type
|
Optional[str]
|
Override for the output ROS type. Defaults to
|
None
|
Returns:
| Name | Type | Description |
|---|---|---|
MsgType |
MsgType
|
The constructed ROS message, or raises an error if:
|
schema_metadata
classmethod
¶
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
¶
Returns the specific ROS message type handled by this adapter.
is_rosmsg_type_valid
classmethod
¶
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., |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
|
unpack_mosaico_msg
classmethod
¶
Extracts the typed Mosaico payload and its Header (if present) from a wrapped or bare message.
Handles two input cases:
Messagewrapper: the typed data is extracted viaget_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 |
required |
Returns:
| Type | Description |
|---|---|
T
|
tuple[T, Header]: A |
Header
|
header is the corresponding |
tuple[T, Header]
|
zero |
Raises:
| Type | Description |
|---|---|
TypeError
|
If mosaico_msg is neither a |
ontology_data_type
classmethod
¶
Returns the Ontology class type associated with this adapter.
AccelAdapter ¶
Bases: ROSAdapterBase[Acceleration]
Adapter for translating ROS Accel-related messages to Mosaico Acceleration.
Supported ROS Types:
geometry_msgs/msg/Accelgeometry_msgs/msg/AccelStampedgeometry_msgs/msg/AccelWithCovariancegeometry_msgs/msg/AccelWithCovarianceStamped
Recursive Unwrapping Strategy:
The adapter checks for nested 'accel' keys. If found (as in AccelStamped), it recurses to the leaf node while collecting metadata like headers and
covariance matrices along the way.
Example
ros_msg = ROSMessage(
topic="/accel",
timestamp=17000,
msg_type="geometry_msgs/msg/AccelStamped",
data = {
"header": {"frame_id": "map", "stamp": {"sec": 17000, "nanosec": 0}},
"accel": {
"linear": {"x": 5.0, "y": 0.0, "z": 0.0},
"angular": {"x": 0.0, "y": 0.0, "z": 1.0}
},
"covariance": [0.1] * 36
}
)
# Automatically resolves to a flat Mosaico Acceleration with attached metadata
mosaico_acceleration = AccelAdapter.translate(ros_msg)
translate
classmethod
¶
Translates a ROS message into a Mosaico Message.
Returns:
| Name | Type | Description |
|---|---|---|
Message |
Message
|
The translated message containing a |
Raises:
| Type | Description |
|---|---|
Exception
|
Wraps any translation error with context (topic name, timestamp). |
from_dict
classmethod
¶
Recursively parses the ROS data dictionary to extract an Acceleration.
Strategy:
- Recurse: If a 'accel' key is found, dive deeper into the structure.
- Leaf Node: At the base level, map 'linear' and 'angular' to
Vector3.
- Metadata Binding: Covariances are attached during
recursion unwinding.
Example
ros_data = {
"header": {"frame_id": "map", "stamp": {"sec": 17000, "nanosec": 0}},
"accel": {
"linear": {"x": 5.0, "y": 0.0, "z": 0.0},
"angular": {"x": 0.0, "y": 0.0, "z": 1.0}
},
"covariance": [0.1] * 36
}
# Automatically resolves to a flat Mosaico Acceleration with attached metadata
mosaico_acceleration = AccelAdapter.from_dict(ros_data)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ros_data
|
dict
|
The raw dictionary from the ROS message. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Acceleration |
Acceleration
|
The constructed Mosaico Acceleration object. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the recursive 'accel' key exists but is not a dict, or if required keys are missing. |
to_ros
classmethod
¶
Converts a Mosaico Acceleration (or a Message wrapping one) into the
corresponding ROS Accel message.
Supported output types (selectable via ros_msg_type):
geometry_msgs/msg/Accelgeometry_msgs/msg/AccelStampedgeometry_msgs/msg/AccelWithCovariancegeometry_msgs/msg/AccelWithCovarianceStamped
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mosaico_data
|
Union[Message, Acceleration]
|
A |
required |
typestore
|
Typestore
|
The rosbags typestore for target type resolution. |
required |
ros_msg_type
|
Optional[str]
|
Override for the output ROS type. Defaults to
|
None
|
Returns:
| Name | Type | Description |
|---|---|---|
MsgType |
MsgType
|
The constructed ROS message, or raises an error if:
|
schema_metadata
classmethod
¶
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
¶
Returns the specific ROS message type handled by this adapter.
is_rosmsg_type_valid
classmethod
¶
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., |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
|
unpack_mosaico_msg
classmethod
¶
Extracts the typed Mosaico payload and its Header (if present) from a wrapped or bare message.
Handles two input cases:
Messagewrapper: the typed data is extracted viaget_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 |
required |
Returns:
| Type | Description |
|---|---|
T
|
tuple[T, Header]: A |
Header
|
header is the corresponding |
tuple[T, Header]
|
zero |
Raises:
| Type | Description |
|---|---|
TypeError
|
If mosaico_msg is neither a |
ontology_data_type
classmethod
¶
Returns the Ontology class type associated with this adapter.
Vector3Adapter ¶
Bases: ROSAdapterBase[Vector3d]
Adapter for translating ROS Vector3 messages to Mosaico Vector3d.
Supported ROS Types:
Recursive Unwrapping Strategy:
The adapter checks for nested 'vector' keys. If found (as in Vector3Stamped), it recurses to the leaf node while collecting metadata like headers and
covariance matrices along the way.
Example
ros_msg = ROSMessage(
topic="/vector3",
timestamp=17000,
msg_type="geometry_msgs/msg/Vector3Stamped",
data = {
"header": {"frame_id": "map", "stamp": {"sec": 17000, "nanosec": 0}},
"vector": {"x": 5.0, "y": 0.0, "z": 0.0},
}
)
# Automatically resolves to a flat Mosaico Vector3 with attached metadata
mosaico_vector3 = Vector3Adapter.translate(ros_msg)
translate
classmethod
¶
Translates a ROS message into a Mosaico Message.
Returns:
| Name | Type | Description |
|---|---|---|
Message |
Message
|
The translated message containing a |
Raises:
| Type | Description |
|---|---|
Exception
|
Wraps any translation error with context (topic name, timestamp). |
from_dict
classmethod
¶
Recursively parses the ROS data to extract a Vector3d.
Strategy:
- Recurse: If a 'vector' key is found, dive deeper into the structure.
- Leaf Node: At the base level, map 'x', 'y' and 'z' to
Vector3d.
Example
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ros_data
|
dict
|
The raw dictionary from the ROS message. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Vector3d |
Vector3d
|
The constructed Mosaico Vector3d object. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the recursive 'vector' key exists but is not a dict, or if required keys are missing. |
to_ros
classmethod
¶
Converts a Mosaico Vector3d (or a Message wrapping one) into the
corresponding ROS Vector3 message.
Supported output types (selectable via ros_msg_type):
geometry_msgs/msg/Vector3geometry_msgs/msg/Vector3Stamped
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mosaico_data
|
Union[Message, Vector3d]
|
A |
required |
typestore
|
Typestore
|
The rosbags typestore for target type resolution. |
required |
ros_msg_type
|
Optional[str]
|
Override for the output ROS type. Defaults to
|
None
|
Returns:
| Name | Type | Description |
|---|---|---|
MsgType |
MsgType
|
The constructed ROS message, or raises an error if:
|
schema_metadata
classmethod
¶
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
¶
Returns the specific ROS message type handled by this adapter.
is_rosmsg_type_valid
classmethod
¶
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., |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
|
unpack_mosaico_msg
classmethod
¶
Extracts the typed Mosaico payload and its Header (if present) from a wrapped or bare message.
Handles two input cases:
Messagewrapper: the typed data is extracted viaget_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 |
required |
Returns:
| Type | Description |
|---|---|
T
|
tuple[T, Header]: A |
Header
|
header is the corresponding |
tuple[T, Header]
|
zero |
Raises:
| Type | Description |
|---|---|
TypeError
|
If mosaico_msg is neither a |
ontology_data_type
classmethod
¶
Returns the Ontology class type associated with this adapter.
PointAdapter ¶
Bases: ROSAdapterBase[Point3d]
Adapter for translating ROS Point messages to Mosaico Point3d.
Supported ROS Types:
Recursive Unwrapping Strategy:
The adapter checks for nested 'point' keys. If found (as in PointStamped), it recurses to the leaf node while collecting metadata like headers and
covariance matrices along the way.
Example
ros_msg = ROSMessage(
topic="/point",
timestamp=17000,
msg_type="geometry_msgs/msg/PointStamped",
data = {
"header": {"frame_id": "map", "stamp": {"sec": 17000, "nanosec": 0}},
"point": {"x": 5.0, "y": 0.0, "z": 0.0},
}
)
# Automatically resolves to a flat Mosaico Point3d with attached metadata
mosaico_point3d = PointAdapter.translate(ros_msg)
translate
classmethod
¶
Translates a ROS message into a Mosaico Message.
Returns:
| Name | Type | Description |
|---|---|---|
Message |
Message
|
The translated message containing a |
Raises:
| Type | Description |
|---|---|
Exception
|
Wraps any translation error with context (topic name, timestamp). |
from_dict
classmethod
¶
Recursively parses the ROS data to extract a Point3d.
Strategy
- Recurse: If a 'point' key is found, dive deeper into the structure.
- Leaf Node: At the base level, map 'x', 'y' and 'z' to
Point3d.
Example
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ros_data
|
dict
|
The raw dictionary from the ROS message. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Point3d |
Point3d
|
The constructed Mosaico Point3d object. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the recursive 'point' key exists but is not a dict, or if required keys are missing. |
to_ros
classmethod
¶
Converts a Mosaico Point3d (or a Message wrapping one) into the
corresponding ROS Point message.
Supported output types (selectable via ros_msg_type):
geometry_msgs/msg/Pointgeometry_msgs/msg/PointStamped
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mosaico_data
|
Union[Message, Point3d]
|
A |
required |
typestore
|
Typestore
|
The rosbags typestore for target type resolution. |
required |
ros_msg_type
|
Optional[str]
|
Override for the output ROS type. Defaults to
|
None
|
Returns:
| Name | Type | Description |
|---|---|---|
MsgType |
MsgType
|
The constructed ROS message, or raises an error if:
|
schema_metadata
classmethod
¶
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
¶
Returns the specific ROS message type handled by this adapter.
is_rosmsg_type_valid
classmethod
¶
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., |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
|
unpack_mosaico_msg
classmethod
¶
Extracts the typed Mosaico payload and its Header (if present) from a wrapped or bare message.
Handles two input cases:
Messagewrapper: the typed data is extracted viaget_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 |
required |
Returns:
| Type | Description |
|---|---|
T
|
tuple[T, Header]: A |
Header
|
header is the corresponding |
tuple[T, Header]
|
zero |
Raises:
| Type | Description |
|---|---|
TypeError
|
If mosaico_msg is neither a |
ontology_data_type
classmethod
¶
Returns the Ontology class type associated with this adapter.
QuaternionAdapter ¶
Bases: ROSAdapterBase[Quaternion]
Adapter for translating ROS Quaternion messages to Mosaico Quaternion.
Supported ROS Types:
Recursive Unwrapping Strategy:
The adapter checks for nested 'quaternion' keys. If found (as in QuaternionStamped), it recurses to the leaf node while collecting metadata like headers and
covariance matrices along the way.
Example
ros_msg = ROSMessage(
topic="/quaternion",
timestamp=17000,
msg_type="geometry_msgs/msg/QuaternionStamped",
data = {
"header": {"frame_id": "map", "stamp": {"sec": 17000, "nanosec": 0}},
"quaternion": {"x": 5.0, "y": 0.0, "z": 0.0, "w": 1.0},
}
)
# Automatically resolves to a flat Mosaico Quaternion with attached metadata
mosaico_quaternion = QuaternionAdapter.translate(ros_msg)
translate
classmethod
¶
Translates a ROS message into a Mosaico Message.
Returns:
| Name | Type | Description |
|---|---|---|
Message |
Message
|
The translated message containing a |
Raises:
| Type | Description |
|---|---|
Exception
|
Wraps any translation error with context (topic name, timestamp). |
from_dict
classmethod
¶
Recursively parses the ROS data to extract a Quaternion.
Strategy
- Recurse: If a 'quaternion' key is found, dive deeper into the structure.
- Leaf Node: At the base level, map 'x', 'y', 'z' and 'w' to
Quaternion.
Example
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ros_data
|
dict
|
The raw dictionary from the ROS message. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Quaternion |
Quaternion
|
The constructed Mosaico Quaternion object. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the recursive 'quaternion' key exists but is not a dict, or if required keys are missing. |
to_ros
classmethod
¶
Converts a Mosaico Quaternion (or a Message wrapping one) into the
corresponding ROS Quaternion message.
Supported output types (selectable via ros_msg_type):
geometry_msgs/msg/Quaterniongeometry_msgs/msg/QuaternionStamped
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mosaico_data
|
Union[Message, Quaternion]
|
A |
required |
typestore
|
Typestore
|
The rosbags typestore for target type resolution. |
required |
ros_msg_type
|
Optional[str]
|
Override for the output ROS type. Defaults to
|
None
|
Returns:
| Name | Type | Description |
|---|---|---|
MsgType |
MsgType
|
The constructed ROS message, or raises an error if:
|
schema_metadata
classmethod
¶
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
¶
Returns the specific ROS message type handled by this adapter.
is_rosmsg_type_valid
classmethod
¶
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., |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
|
unpack_mosaico_msg
classmethod
¶
Extracts the typed Mosaico payload and its Header (if present) from a wrapped or bare message.
Handles two input cases:
Messagewrapper: the typed data is extracted viaget_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 |
required |
Returns:
| Type | Description |
|---|---|
T
|
tuple[T, Header]: A |
Header
|
header is the corresponding |
tuple[T, Header]
|
zero |
Raises:
| Type | Description |
|---|---|
TypeError
|
If mosaico_msg is neither a |
ontology_data_type
classmethod
¶
Returns the Ontology class type associated with this adapter.
TransformAdapter ¶
Bases: ROSAdapterBase[Transform]
Adapter for translating ROS Transform messages to Mosaico Transform.
Supported ROS Types:
Recursive Unwrapping Strategy:
The adapter checks for nested 'transform' keys. If found (as in TransformStamped), it recurses to the leaf node while collecting metadata like headers and
covariance matrices along the way.
Example
ros_msg = ROSMessage(
topic="/transform",
timestamp=17000,
msg_type="geometry_msgs/msg/TransformStamped",
data = {
"header": {"frame_id": "map", "stamp": {"sec": 17000, "nanosec": 0}},
"transform": {"translation": {"x": 5.0, "y": 0.0, "z": 0.0}, "rotation": {"x": 0.0, "y": 0.0, "z": 0.0, "w": 1.0}},
}
)
# Automatically resolves to a flat Mosaico Transform with attached metadata
mosaico_transform = TransformAdapter.translate(ros_msg)
translate
classmethod
¶
Translates a ROS message into a Mosaico Message.
Returns:
| Name | Type | Description |
|---|---|---|
Message |
Message
|
The translated message containing a |
Raises:
| Type | Description |
|---|---|
Exception
|
Wraps any translation error with context (topic name, timestamp). |
from_dict
classmethod
¶
Parses ROS Transform data. Handles both nested 'transform' field (from Stamped) and flat structure.
Strategy
- Recurse: If a 'transform' key is found, dive deeper into the structure.
- Leaf Node: At the base level, map 'translation' and 'rotation' to
Transform.
Example
ros_data = {
"header": {"frame_id": "map", "stamp": {"sec": 17000, "nanosec": 0}},
"transform": {"translation": {"x": 5.0, "y": 0.0, "z": 0.0}, "rotation": {"x": 0.0, "y": 0.0, "z": 0.0, "w": 1.0}},
}
# Automatically resolves to a flat Mosaico Transform with attached metadata
mosaico_transform = TransformAdapter.from_dict(ros_data)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ros_data
|
dict
|
The raw dictionary from the ROS message. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Transform |
Transform
|
The constructed Mosaico Transform object. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the recursive 'transform' key exists but is not a dict, or if required keys are missing. |
to_ros
classmethod
¶
Converts a Mosaico Transform (or a Message wrapping one) into the
corresponding ROS Transform message.
Supported output types (selectable via ros_msg_type):
geometry_msgs/msg/Transformgeometry_msgs/msg/TransformStamped
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mosaico_data
|
Union[Message, Transform]
|
A |
required |
typestore
|
Typestore
|
The rosbags typestore for target type resolution. |
required |
ros_msg_type
|
Optional[str]
|
Override for the output ROS type. Defaults to
|
None
|
Returns:
| Name | Type | Description |
|---|---|---|
MsgType |
MsgType
|
The constructed ROS message, or raises an error if:
|
schema_metadata
classmethod
¶
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
¶
Returns the specific ROS message type handled by this adapter.
is_rosmsg_type_valid
classmethod
¶
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., |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
|
unpack_mosaico_msg
classmethod
¶
Extracts the typed Mosaico payload and its Header (if present) from a wrapped or bare message.
Handles two input cases:
Messagewrapper: the typed data is extracted viaget_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 |
required |
Returns:
| Type | Description |
|---|---|
T
|
tuple[T, Header]: A |
Header
|
header is the corresponding |
tuple[T, Header]
|
zero |
Raises:
| Type | Description |
|---|---|
TypeError
|
If mosaico_msg is neither a |
ontology_data_type
classmethod
¶
Returns the Ontology class type associated with this adapter.
WrenchAdapter ¶
Bases: ROSAdapterBase[ForceTorque]
Adapter for translating ROS Wrench messages to Mosaico ForceTorque.
Supported ROS Types:
Recursive Unwrapping Strategy:
The adapter checks for nested 'wrench' keys. If found (as in WrenchStamped), it recurses to the leaf node while collecting metadata like headers and
covariance matrices along the way.
Example
ros_msg = ROSMessage(
topic="/wrench",
timestamp=17000,
msg_type="geometry_msgs/msg/WrenchStamped",
data = {
"header": {"frame_id": "map", "stamp": {"sec": 17000, "nanosec": 0}},
"wrench": {"force": {"x": 5.0, "y": 0.0, "z": 0.0}, "torque": {"x": 0.0, "y": 0.0, "z": 0.0}},
}
)
# Automatically resolves to a flat Mosaico ForceTorque with attached metadata
mosaico_wrench = WrenchAdapter.translate(ros_msg)
translate
classmethod
¶
Translates a ROS message into a Mosaico Message.
Returns:
| Name | Type | Description |
|---|---|---|
Message |
Message
|
The translated message containing a |
Raises:
| Type | Description |
|---|---|
Exception
|
Wraps any translation error with context (topic name, timestamp). |
from_dict
classmethod
¶
Parses ROS ForceTorque data. Handles both nested 'wrench' field (from Stamped) and flat structure.
Strategy
- Recurse: If a 'wrench' key is found, dive deeper into the structure.
- Leaf Node: At the base level, map 'force' and 'torque' to
ForceTorque.
Example
ros_data = {
"header": {"frame_id": "map", "stamp": {"sec": 17000, "nanosec": 0}},
"wrench": {"force": {"x": 5.0, "y": 0.0, "z": 0.0}, "torque": {"x": 0.0, "y": 0.0, "z": 0.0}},
}
# Automatically resolves to a flat Mosaico ForceTorque with attached metadata
mosaico_wrench = WrenchAdapter.from_dict(ros_data)
to_ros
classmethod
¶
Converts a Mosaico ForceTorque (or a Message wrapping one) into the
corresponding ROS Wrench message.
Supported output types (selectable via ros_msg_type):
geometry_msgs/msg/Wrenchgeometry_msgs/msg/WrenchStamped
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mosaico_data
|
Union[Message, ForceTorque]
|
A |
required |
typestore
|
Typestore
|
The rosbags typestore for target type resolution. |
required |
ros_msg_type
|
Optional[str]
|
Override for the output ROS type. Defaults to
|
None
|
Returns:
| Name | Type | Description |
|---|---|---|
MsgType |
MsgType
|
The constructed ROS message, or raises an error if:
|
schema_metadata
classmethod
¶
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
¶
Returns the specific ROS message type handled by this adapter.
is_rosmsg_type_valid
classmethod
¶
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., |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
|
unpack_mosaico_msg
classmethod
¶
Extracts the typed Mosaico payload and its Header (if present) from a wrapped or bare message.
Handles two input cases:
Messagewrapper: the typed data is extracted viaget_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 |
required |
Returns:
| Type | Description |
|---|---|
T
|
tuple[T, Header]: A |
Header
|
header is the corresponding |
tuple[T, Header]
|
zero |
Raises:
| Type | Description |
|---|---|
TypeError
|
If mosaico_msg is neither a |
ontology_data_type
classmethod
¶
Returns the Ontology class type associated with this adapter.
PolygonAdapter ¶
Bases: ROSAdapterBase[Polygon]
Adapter for translating ROS Polygon messages to Mosaico Polygon.
Supported ROS Types:
Recursive Unwrapping Strategy: The adapter checks for nested 'polygon' keys (as in PolygonStamped) and recursively unwraps to the base structure.
Example:
ros_msg = ROSMessage(
topic="/polygon",
timestamp=17000,
msg_type="geometry_msgs/msg/Polygon",
data={
"points": [
{"x": 1.0, "y": 2.0, "z": 0.0},
{"x": 3.0, "y": 4.0, "z": 0.0},
]
}
)
mosaico_polygon = PolygonAdapter.translate(ros_msg)
translate
classmethod
¶
Translates a ROS message into a Mosaico Message.
Returns:
| Name | Type | Description |
|---|---|---|
Message |
Message
|
The translated message containing a |
Raises:
| Type | Description |
|---|---|
Exception
|
Wraps any translation error with context (topic name, timestamp). |
from_dict
classmethod
¶
Parses ROS Polygon data. Handles both nested ('PolygonStamped') and flat structures.
Strategy
- Recurse: If a 'polygon' key is found, unwrap and process the inner structure.
- Leaf Node: Convert the list of ROS points into Mosaico
Point3dobjects and construct aPolygon.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ros_data
|
dict
|
The raw dictionary from the ROS message. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Polygon |
Polygon
|
The constructed Mosaico Polygon object. |
to_ros
classmethod
¶
Converts a Mosaico Polygon (or a Message wrapping one) into the
corresponding ROS Polygon message.
Supported output types (selectable via ros_msg_type):
geometry_msgs/msg/Polygongeometry_msgs/msg/PolygonStamped
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mosaico_data
|
Union[Message, Polygon]
|
A |
required |
typestore
|
Typestore
|
The rosbags typestore for target type resolution. |
required |
ros_msg_type
|
Optional[str]
|
Override for the output ROS type. Defaults to
|
None
|
Returns:
| Name | Type | Description |
|---|---|---|
MsgType |
MsgType
|
The constructed ROS message, or raises an error if:
|
schema_metadata
classmethod
¶
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
¶
Returns the specific ROS message type handled by this adapter.
is_rosmsg_type_valid
classmethod
¶
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., |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
|
unpack_mosaico_msg
classmethod
¶
Extracts the typed Mosaico payload and its Header (if present) from a wrapped or bare message.
Handles two input cases:
Messagewrapper: the typed data is extracted viaget_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 |
required |
Returns:
| Type | Description |
|---|---|
T
|
tuple[T, Header]: A |
Header
|
header is the corresponding |
tuple[T, Header]
|
zero |
Raises:
| Type | Description |
|---|---|
TypeError
|
If mosaico_msg is neither a |
ontology_data_type
classmethod
¶
Returns the Ontology class type associated with this adapter.
InertiaAdapter ¶
Bases: ROSAdapterBase[Inertia]
Adapter for translating ROS Inertia messages to Mosaico Inertia.
Supported ROS Types:
Recursive Unwrapping Strategy: The adapter checks for nested 'inertia' keys (as in InertiaStamped) and recursively unwraps to the base structure.
Example:
ros_msg = ROSMessage(
topic="/inertia",
timestamp=17000,
msg_type="geometry_msgs/msg/Inertia",
data={
"m": 10.0,
"com": {"x": 0.0, "y": 0.0, "z": 0.0},
"ixx": 1.0,
"ixy": 0.0,
"ixz": 0.0,
"iyy": 1.0,
"iyz": 0.0,
"izz": 1.0,
}
)
mosaico_inertia = InertiaAdapter.translate(ros_msg)
translate
classmethod
¶
Translates a ROS message into a Mosaico Message.
Returns:
| Name | Type | Description |
|---|---|---|
Message |
Message
|
The translated message containing a |
Raises:
| Type | Description |
|---|---|
Exception
|
Wraps any translation error with context (topic name, timestamp). |
from_dict
classmethod
¶
Parses ROS Inertia data. Handles both nested ('InertiaStamped') and flat structures.
Strategy
- Recurse: If an 'inertia' key is found, unwrap and process the inner structure.
- Leaf Node:
- Map 'com' to a Mosaico
Vector3d. - Construct the inertia tensor from scalar components (ixx, ixy, etc.).
- Build the
Inertiaobject.
- Map 'com' to a Mosaico
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ros_data
|
dict
|
The raw dictionary from the ROS message. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
Inertia |
Inertia
|
The constructed Mosaico Inertia object. |
to_ros
classmethod
¶
Converts a Mosaico Inertia (or a Message wrapping one) into the
corresponding ROS Inertia message.
Supported output types (selectable via ros_msg_type):
geometry_msgs/msg/Inertiageometry_msgs/msg/InertiaStamped
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mosaico_data
|
Union[Message, Inertia]
|
A |
required |
typestore
|
Typestore
|
The rosbags typestore for target type resolution. |
required |
ros_msg_type
|
Optional[str]
|
Override for the output ROS type. Defaults to
|
None
|
Returns:
| Name | Type | Description |
|---|---|---|
MsgType |
MsgType
|
The constructed ROS message, or raises an error if:
|
schema_metadata
classmethod
¶
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
¶
Returns the specific ROS message type handled by this adapter.
is_rosmsg_type_valid
classmethod
¶
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., |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
|
unpack_mosaico_msg
classmethod
¶
Extracts the typed Mosaico payload and its Header (if present) from a wrapped or bare message.
Handles two input cases:
Messagewrapper: the typed data is extracted viaget_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 |
required |
Returns:
| Type | Description |
|---|---|
T
|
tuple[T, Header]: A |
Header
|
header is the corresponding |
tuple[T, Header]
|
zero |
Raises:
| Type | Description |
|---|---|
TypeError
|
If mosaico_msg is neither a |
ontology_data_type
classmethod
¶
Returns the Ontology class type associated with this adapter.