Skip to content

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:

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

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

A Mosaico Message containing the normalized Pose payload.

from_dict classmethod

from_dict(ros_data)

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 Point3d and Quaternion.
  • 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

to_ros(mosaico_data, typestore, ros_msg_type=None)

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/Pose
  • geometry_msgs/msg/PoseStamped
  • geometry_msgs/msg/PoseWithCovariance
  • geometry_msgs/msg/PoseWithCovarianceStamped

Parameters:

Name Type Description Default
mosaico_data Union[Message, Pose]

A Message wrapping a Pose instance, or a raw Pose.

required
typestore Typestore

The rosbags typestore for target type resolution.

required
ros_msg_type Optional[str]

Override for the output ROS type. Defaults to geometry_msgs/msg/Pose if None.

None

Returns:

Name Type Description
MsgType MsgType

The constructed 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)
  • the ros_msg_type or default type are supported but translation is not implemented (NotImplementedError)

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.

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:

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

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Returns:

Name Type Description
Message Message

The translated message containing a Velocity object.

Raises:

Type Description
Exception

Wraps any translation error with context (topic name, timestamp).

from_dict classmethod

from_dict(ros_data)

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

to_ros(mosaico_data, typestore, ros_msg_type=None)

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/Twist
  • geometry_msgs/msg/TwistStamped
  • geometry_msgs/msg/TwistWithCovariance
  • geometry_msgs/msg/TwistWithCovarianceStamped

Parameters:

Name Type Description Default
mosaico_data Union[Message, Velocity]

A Message wrapping a Velocity instance, or a raw Velocity.

required
typestore Typestore

The rosbags typestore for target type resolution.

required
ros_msg_type Optional[str]

Override for the output ROS type. Defaults to geometry_msgs/msg/Twist if None.

None

Returns:

Name Type Description
MsgType MsgType

The constructed 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)
  • the ros_msg_type or default type are supported but translation is not implemented (NotImplementedError)

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.

AccelAdapter

Bases: ROSAdapterBase[Acceleration]

Adapter for translating ROS Accel-related messages to Mosaico Acceleration.

Supported ROS Types:

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

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Returns:

Name Type Description
Message Message

The translated message containing a Acceleration object.

Raises:

Type Description
Exception

Wraps any translation error with context (topic name, timestamp).

from_dict classmethod

from_dict(ros_data)

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

to_ros(mosaico_data, typestore, ros_msg_type=None)

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/Accel
  • geometry_msgs/msg/AccelStamped
  • geometry_msgs/msg/AccelWithCovariance
  • geometry_msgs/msg/AccelWithCovarianceStamped

Parameters:

Name Type Description Default
mosaico_data Union[Message, Acceleration]

A Message wrapping an Acceleration instance, or a raw Acceleration.

required
typestore Typestore

The rosbags typestore for target type resolution.

required
ros_msg_type Optional[str]

Override for the output ROS type. Defaults to geometry_msgs/msg/Accel if None.

None

Returns:

Name Type Description
MsgType MsgType

The constructed 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)
  • the ros_msg_type or default type are supported but translation is not implemented (NotImplementedError)

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.

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

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Returns:

Name Type Description
Message Message

The translated message containing a Vector3d object.

Raises:

Type Description
Exception

Wraps any translation error with context (topic name, timestamp).

from_dict classmethod

from_dict(ros_data)

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
ros_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 Vector3d with attached metadata
mosaico_vector3d = Vector3Adapter.from_dict(ros_data)

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

to_ros(mosaico_data, typestore, ros_msg_type=None)

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/Vector3
  • geometry_msgs/msg/Vector3Stamped

Parameters:

Name Type Description Default
mosaico_data Union[Message, Vector3d]

A Message wrapping a Vector3d instance, or a raw Vector3d.

required
typestore Typestore

The rosbags typestore for target type resolution.

required
ros_msg_type Optional[str]

Override for the output ROS type. Defaults to geometry_msgs/msg/Vector3 if None.

None

Returns:

Name Type Description
MsgType MsgType

The constructed 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)
  • the ros_msg_type or default type are supported but translation is not implemented (NotImplementedError)

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.

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

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Returns:

Name Type Description
Message Message

The translated message containing a Point3d object.

Raises:

Type Description
Exception

Wraps any translation error with context (topic name, timestamp).

from_dict classmethod

from_dict(ros_data)

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
ros_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.from_dict(ros_data)

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

to_ros(mosaico_data, typestore, ros_msg_type=None)

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/Point
  • geometry_msgs/msg/PointStamped

Parameters:

Name Type Description Default
mosaico_data Union[Message, Point3d]

A Message wrapping a Point3d instance, or a raw Point3d.

required
typestore Typestore

The rosbags typestore for target type resolution.

required
ros_msg_type Optional[str]

Override for the output ROS type. Defaults to geometry_msgs/msg/Point if None.

None

Returns:

Name Type Description
MsgType MsgType

The constructed 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)
  • the ros_msg_type or default type are supported but translation is not implemented (NotImplementedError)

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.

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

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Returns:

Name Type Description
Message Message

The translated message containing a Quaternion object.

Raises:

Type Description
Exception

Wraps any translation error with context (topic name, timestamp).

from_dict classmethod

from_dict(ros_data)

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
ros_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.from_dict(ros_data)

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

to_ros(mosaico_data, typestore, ros_msg_type=None)

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/Quaternion
  • geometry_msgs/msg/QuaternionStamped

Parameters:

Name Type Description Default
mosaico_data Union[Message, Quaternion]

A Message wrapping a Quaternion instance, or a raw Quaternion.

required
typestore Typestore

The rosbags typestore for target type resolution.

required
ros_msg_type Optional[str]

Override for the output ROS type. Defaults to geometry_msgs/msg/Quaternion if None.

None

Returns:

Name Type Description
MsgType MsgType

The constructed 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)
  • the ros_msg_type or default type are supported but translation is not implemented (NotImplementedError)

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.

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

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Returns:

Name Type Description
Message Message

The translated message containing a Transform object.

Raises:

Type Description
Exception

Wraps any translation error with context (topic name, timestamp).

from_dict classmethod

from_dict(ros_data)

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

to_ros(mosaico_data, typestore, ros_msg_type=None)

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/Transform
  • geometry_msgs/msg/TransformStamped

Parameters:

Name Type Description Default
mosaico_data Union[Message, Transform]

A Message wrapping a Transform instance, or a raw Transform.

required
typestore Typestore

The rosbags typestore for target type resolution.

required
ros_msg_type Optional[str]

Override for the output ROS type. Defaults to geometry_msgs/msg/Transform if None.

None

Returns:

Name Type Description
MsgType MsgType

The constructed 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)
  • the ros_msg_type or default type are supported but translation is not implemented (NotImplementedError)

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.

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

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Returns:

Name Type Description
Message Message

The translated message containing a ForceTorque object.

Raises:

Type Description
Exception

Wraps any translation error with context (topic name, timestamp).

from_dict classmethod

from_dict(ros_data)

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

to_ros(mosaico_data, typestore, ros_msg_type=None)

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/Wrench
  • geometry_msgs/msg/WrenchStamped

Parameters:

Name Type Description Default
mosaico_data Union[Message, ForceTorque]

A Message wrapping a ForceTorque instance, or a raw ForceTorque.

required
typestore Typestore

The rosbags typestore for target type resolution.

required
ros_msg_type Optional[str]

Override for the output ROS type. Defaults to geometry_msgs/msg/Wrench if None.

None

Returns:

Name Type Description
MsgType MsgType

The constructed 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)
  • the ros_msg_type or default type are supported but translation is not implemented (NotImplementedError)

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.

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

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Returns:

Name Type Description
Message Message

The translated message containing a Polygon object.

Raises:

Type Description
Exception

Wraps any translation error with context (topic name, timestamp).

from_dict classmethod

from_dict(ros_data)

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 Point3d objects and construct a Polygon.

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

to_ros(mosaico_data, typestore, ros_msg_type=None)

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/Polygon
  • geometry_msgs/msg/PolygonStamped

Parameters:

Name Type Description Default
mosaico_data Union[Message, Polygon]

A Message wrapping a Polygon instance, or a raw Polygon.

required
typestore Typestore

The rosbags typestore for target type resolution.

required
ros_msg_type Optional[str]

Override for the output ROS type. Defaults to geometry_msgs/msg/Polygon if None.

None

Returns:

Name Type Description
MsgType MsgType

The constructed 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)
  • the ros_msg_type or default type are supported but translation is not implemented (NotImplementedError)

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.

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

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Returns:

Name Type Description
Message Message

The translated message containing a Inertia object.

Raises:

Type Description
Exception

Wraps any translation error with context (topic name, timestamp).

from_dict classmethod

from_dict(ros_data)

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 Inertia object.

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

to_ros(mosaico_data, typestore, ros_msg_type=None)

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/Inertia
  • geometry_msgs/msg/InertiaStamped

Parameters:

Name Type Description Default
mosaico_data Union[Message, Inertia]

A Message wrapping an Inertia instance, or a raw Inertia.

required
typestore Typestore

The rosbags typestore for target type resolution.

required
ros_msg_type Optional[str]

Override for the output ROS type. Defaults to geometry_msgs/msg/Inertia if None.

None

Returns:

Name Type Description
MsgType MsgType

The constructed 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)
  • the ros_msg_type or default type are supported but translation is not implemented (NotImplementedError)

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.