sensor_msgs Adapters

mosaicolabs.ros_bridge.adapters.sensor_msgs

CameraInfoAdapter

Bases: ROSAdapterBase[CameraInfo]

Adapter for translating ROS CameraInfo messages to Mosaico CameraInfo.

Supported ROS Types:

• sensor_msgs/msg/CameraInfo

Example

ros_msg = ROSMessage(
    timestamp=17000,
    topic="/camera_info",
    msg_type="sensor_msgs/msg/CameraInfo",
    data = {
        "height": 480,
        "width": 640,
        "binning_x": 1,
        "binning_y": 1,
        "roi": {
            "x_offset": 0,
            "y_offset": 0,
            "height": 480,
            "width": 640,
            "do_rectify": False,
        },
        "distortion_model": "plumb_bob",
        "d": [0.0, 0.0, 0.0, 0.0, 0.0],
        "k": [1.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 1.0],
        "p": [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
        "r": [1.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 1.0],
    }
)
# Automatically resolves to a flat Mosaico CameraInfo with attached metadata
mosaico_camera_info = CameraInfoAdapter.translate(ros_msg)

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a CameraInfo object.

Raises:

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

from_dict classmethod

from_dict(ros_data)

Converts the raw dictionary data into the specific Mosaico type.

Example

ros_data = {
    "height": 480,
    "width": 640,
    "binning_x": 1,
    "binning_y": 1,
    "roi": {
        "x_offset": 0,
        "y_offset": 0,
        "height": 480,
        "width": 640,
        "do_rectify": False,
    },
    "distortion_model": "plumb_bob",
    "d": [0.0, 0.0, 0.0, 0.0, 0.0],
    "k": [1.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 1.0],
    "p": [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
    "r": [1.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 1.0],
}
# Automatically resolves to a flat Mosaico CameraInfo with attached metadata
mosaico_camera_info = CameraInfoAdapter.from_dict(ros_data)

Parameters:

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

Returns:

Name	Type	Description
CameraInfo	CameraInfo	The constructed Mosaico CameraInfo object.

Raises:

Type	Description
ValueError	If required keys are missing.

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico CameraInfo (or a Message wrapping one) into a sensor_msgs/msg/CameraInfo message.

Handles both ROS 1 (uppercase field names D, K, R, P) and ROS 2 (lowercase d, k, r, p) field conventions.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, CameraInfo]	A Message wrapping a CameraInfo instance, or a raw CameraInfo.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/CameraInfo is supported.	None

Returns:

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

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.

NavSatStatusAdapter

Bases: ROSAdapterBase[GPSStatus]

Adapter for translating ROS NavSatStatus messages to Mosaico GPSStatus.

Supported ROS Types:

• sensor_msgs/msg/NavSatStatus

Example

ros_msg = ROSMessage(
    timestamp=17000,
    topic="/gps_status",
    msg_type="sensor_msgs/msg/NavSatStatus",
    data = {
        "status": 0,
        "service": 1,
    }
)
# Automatically resolves to a flat Mosaico GPSStatus with attached metadata
mosaico_gps_status = NavSatStatusAdapter.translate(ros_msg)

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a GPSStatus object.

Raises:

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

from_dict classmethod

from_dict(ros_data)

Converts the raw dictionary data into the specific Mosaico type.

Example

ros_data = {
    "status": 0,
    "service": 1,
}
# Automatically resolves to a flat Mosaico GPSStatus with attached metadata
mosaico_gps_status = NavSatStatusAdapter.from_dict(ros_data)

Parameters:

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

Returns:

Name	Type	Description
GPSStatus	GPSStatus	The constructed Mosaico GPSStatus object.

Raises:

Type	Description
ValueError	If a required key is missing from ros_data.

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico GPSStatus (or a Message wrapping one) into a sensor_msgs/msg/NavSatStatus message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, GPSStatus]	A Message wrapping a GPSStatus instance, or a raw GPSStatus.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/NavSatStatus is supported.	None

Returns:

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

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.

GPSAdapter

Bases: ROSAdapterBase[GPS]

Adapter for translating ROS NavSatFix messages to Mosaico GPS.

Supported ROS Types:

• sensor_msgs/msg/NavSatFix

Example

ros_msg = ROSMessage(
    timestamp=17000,
    topic="/gps",
    msg_type="sensor_msgs/msg/NavSatFix",
    data = {
        "latitude": 45.5,
        "longitude": -122.5,
        "altitude": 100.0,
        "status": {
            "status": 0,
            "service": 1,
        },
    }
)
# Automatically resolves to a flat Mosaico GPS with attached metadata
mosaico_gps = GPSAdapter.translate(ros_msg)

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a GPS object.

Raises:

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

from_dict classmethod

from_dict(ros_data)

Converts the raw dictionary data into the specific Mosaico type.

Example

ros_data = {
    "latitude": 45.5,
    "longitude": -122.5,
    "altitude": 100.0,
    "status": {
        "status": 0,
        "service": 1,
    },
}
# Automatically resolves to a flat Mosaico GPS with attached metadata
mosaico_gps = GPSAdapter.from_dict(ros_data)

Parameters:

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

Returns:

Name	Type	Description
GPS	GPS	The constructed Mosaico GPS object.

Raises:

Type	Description
ValueError	If a required key is missing from ros_data.

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico GPS (or a Message wrapping one) into a sensor_msgs/msg/NavSatFix message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, GPS]	A Message wrapping a GPS instance, or a raw GPS.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/NavSatFix is supported.	None

Returns:

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

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.

IMUAdapter

Bases: ROSAdapterBase[IMU]

Adapter for translating ROS Imu messages to Mosaico IMU.

Supported ROS Types:

• sensor_msgs/msg/Imu

Example

ros_msg = ROSMessage(
    timestamp=17000,
    topic="/imu",
    msg_type="sensor_msgs/msg/Imu",
    data = {
        "linear_acceleration": {"x": 0.0, "y": 0.0, "z": 0.0},
        "angular_velocity": {"x": 0.0, "y": 0.0, "z": 0.0},
        "orientation": {"x": 0.0, "y": 0.0, "z": 0.0, "w": 1.0},
        "orientation_covariance": [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
        "linear_acceleration_covariance": [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
        "angular_velocity_covariance": [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
        "header": {"seq": 0, "stamp": {"sec": 0, "nanosec": 0}, "frame_id": "robot_link"},
    }
)
# Automatically resolves to a flat Mosaico IMU with attached metadata
mosaico_imu = IMUAdapter.translate(ros_msg)

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a IMU object.

Raises:

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

from_dict classmethod

from_dict(ros_data)

Converts the raw dictionary data into the specific Mosaico type.

Example

ros_data = {
    "linear_acceleration": {"x": 0.0, "y": 0.0, "z": 0.0},
    "angular_velocity": {"x": 0.0, "y": 0.0, "z": 0.0},
    "orientation": {"x": 0.0, "y": 0.0, "z": 0.0, "w": 1.0},
    "orientation_covariance": [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
    "linear_acceleration_covariance": [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
    "angular_velocity_covariance": [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
    "header": {"seq": 0, "stamp": {"sec": 0, "nanosec": 0}, "frame_id": "robot_link"},
}
# Automatically resolves to a flat Mosaico IMU with attached metadata
mosaico_imu = IMUAdapter.from_dict(ros_data)

Parameters:

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

Returns:

Name	Type	Description
IMU	IMU	The constructed Mosaico IMU object.

Raises:

Type	Description
ValueError	If a required key is missing from ros_data.

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico IMU (or a Message wrapping one) into a sensor_msgs/msg/Imu message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, IMU]	A Message wrapping an IMU instance, or a raw IMU.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/Imu is supported.	None

Returns:

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

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.

NMEASentenceAdapter

Bases: ROSAdapterBase[NMEASentence]

Adapter for translating ROS NMEASentence messages to Mosaico NMEASentence.

Supported ROS Types:

• nmea_msgs/msg/Sentence

Example

ros_msg = ROSMessage(
    timestamp=17000,
    topic="/nmea_sentence",
    msg_type="nmea_msgs/msg/Sentence",
    data = {
        "sentence": "GPGGA,123519,4807.038,N,01131.000,E,1,08,0.9,545.4,M,46.9,M,,*47",
    }
)
# Automatically resolves to a flat Mosaico NMEASentence with attached metadata
mosaico_nmea_sentence = NMEASentenceAdapter.translate(ros_msg)

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a NMEASentence object.

Raises:

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

from_dict classmethod

from_dict(ros_data)

Converts the raw dictionary data into the specific Mosaico type.

Example

ros_data = {
    "sentence": "GPGGA,123519,4807.038,N,01131.000,E,1,08,0.9,545.4,M,46.9,M,,*47",
}
# Automatically resolves to a flat Mosaico NMEASentence with attached metadata
mosaico_nmea_sentence = NMEASentenceAdapter.from_dict(ros_data)

Parameters:

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

Returns:

Name	Type	Description
NMEASentence	NMEASentence	The constructed Mosaico NMEASentence object.

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico NMEASentence (or a Message wrapping one) into a nmea_msgs/msg/Sentence message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, NMEASentence]	A Message wrapping a NMEASentence instance, or a raw NMEASentence.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only nmea_msgs/msg/Sentence is supported.	None

Returns:

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

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.

ImageAdapter

Bases: ROSAdapterBase[Image]

Adapter for translating ROS Image messages to Mosaico Image.

Supported ROS Types:

• sensor_msgs/msg/Image

Example

ros_msg = ROSMessage(
    timestamp=17000,
    topic="/image",
    msg_type="sensor_msgs/msg/Image",
    data = {
        "data": [...],
        "width": 1,
        "height": 1,
        "step": 4,
        "encoding": "bgr8",
    }
)
# Automatically resolves to a flat Mosaico Image with attached metadata
mosaico_image = ImageAdapter.translate(ros_msg)

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a Image object.

Raises:

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

from_dict classmethod

from_dict(ros_data, **kwargs)

Converts the raw dictionary data into the specific Mosaico type.

Example

ros_data = {
    "data": [...],
    "width": 1,
    "height": 1,
    "step": 4,
    "encoding": "bgr8",
}
# Automatically resolves to a flat Mosaico Image with attached metadata
mosaico_image = ImageAdapter.from_dict(ros_data)

Parameters:

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

Returns:

Name	Type	Description
Image	Image	The constructed Mosaico Image object.

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico Image (or a Message wrapping one) into a sensor_msgs/msg/Image message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, Image]	A Message wrapping an Image instance, or a raw Image.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/Image is supported.	None

Returns:

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

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.

CompressedImageAdapter

Bases: ROSAdapterBase[CompressedImage]

Adapter for translating ROS CompressedImage messages to Mosaico CompressedImage.

Supported ROS Types:

• sensor_msgs/msg/CompressedImage

Example

ros_msg = ROSMessage(
    timestamp=17000,
    topic="/compressed_image",
    msg_type="sensor_msgs/msg/CompressedImage",
    data = {
        "data": [...],
        "format": "jpeg",
    }
)
# Automatically resolves to a flat Mosaico CompressedImage with attached metadata
mosaico_compressed_image = CompressedImageAdapter.translate(ros_msg)

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a CompressedImage object.

Raises:

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

from_dict classmethod

from_dict(ros_data)

Converts the raw dictionary data into the specific Mosaico type.

Example

ros_data = {
    "data": [...],
    "format": "jpeg",
}
# Automatically resolves to a flat Mosaico CompressedImage with attached metadata
mosaico_compressed_image = CompressedImageAdapter.from_dict(ros_data)

Parameters:

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

Returns:

Name	Type	Description
CompressedImage	CompressedImage	The constructed Mosaico CompressedImage object.

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico CompressedImage (or a Message wrapping one) into a sensor_msgs/msg/CompressedImage message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, CompressedImage]	A Message wrapping a CompressedImage instance, or a raw CompressedImage.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/CompressedImage is supported.	None

Returns:

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

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.

ROIAdapter

Bases: ROSAdapterBase[ROI]

Adapter for translating ROS RegionOfInterest messages to Mosaico ROI.

Supported ROS Types:

• sensor_msgs/RegionOfInterest

Example

ros_msg = ROSMessage(
    timestamp=17000,
    topic="/roi",
    msg_type="sensor_msgs/msg/RegionOfInterest",
    data = {
        "height": 1,
        "width": 1,
        "x_offset": 0,
        "y_offset": 0,
    }
)
# Automatically resolves to a flat Mosaico ROI with attached metadata
mosaico_roi = ROIAdapter.translate(ros_msg)

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a ROI object.

Raises:

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

from_dict classmethod

from_dict(ros_data)

Converts the raw dictionary data into the specific Mosaico type.

Example

ros_data = {
    "height": 1,
    "width": 1,
    "x_offset": 0,
    "y_offset": 0,
}
# Automatically resolves to a flat Mosaico ROI with attached metadata
mosaico_roi = ROIAdapter.from_dict(ros_data)

Parameters:

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

Returns:

Name	Type	Description
ROI	ROI	The constructed Mosaico ROI object.

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico ROI (or a Message wrapping one) into a sensor_msgs/msg/RegionOfInterest message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, ROI]	A Message wrapping a ROI instance, or a raw ROI.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/RegionOfInterest is supported.	None

Returns:

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

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.

BatteryStateAdapter

Bases: ROSAdapterBase[BatteryState]

Adapter for translating ROS BatteryState messages to Mosaico BatteryState.

Supported ROS Types:

• sensor_msgs/msg/BatteryState

Example

ros_msg = ROSMessage(
    timestamp=17000,
    topic="/battery_state",
    msg_type="sensor_msgs/msg/BatteryState",
    data = {
        "voltage": 12.6,
        "capacity": 100,
        "cell_temperature": [25.0],
        "cell_voltage": [12.6],
        "location": "battery",
        "charge": 100,
        "current": 0,
        "design_capacity": 100,
        "location": "battery",
        "percentage": 100,
        "power_supply_health": "good",
        "power_supply_status": "charging",
        "power_supply_technology": "li-ion",
        "present": True,
        "serial_number": "1234567890",
        "temperature": 25,
    }
)
# Automatically resolves to a flat Mosaico BatteryState with attached metadata
mosaico_battery_state = BatteryStateAdapter.translate(ros_msg)

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a BatteryState object.

Raises:

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

from_dict classmethod

from_dict(ros_data)

Converts the raw dictionary data into the specific Mosaico type.

Example

ros_data = {
    "voltage": 12.6,
    "capacity": 100,
    "cell_temperature": [25.0],
    "cell_voltage": [12.6],
    "location": "battery",
    "charge": 100,
    "current": 0,
    "design_capacity": 100,
    "location": "battery",
    "percentage": 100,
    "power_supply_health": "good",
    "power_supply_status": "charging",
    "power_supply_technology": "li-ion",
    "present": True,
    "serial_number": "1234567890",
    "temperature": 25,
}
# Automatically resolves to a flat Mosaico BatteryState with attached metadata
mosaico_battery_state = BatteryStateAdapter.from_dict(ros_data)

Parameters:

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

Returns:

Name	Type	Description
BatteryState	BatteryState	The constructed Mosaico BatteryState object.

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico BatteryState (or a Message wrapping one) into a sensor_msgs/msg/BatteryState message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, BatteryState]	A Message wrapping a BatteryState instance, or a raw BatteryState.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/BatteryState is supported.	None

Returns:

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

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.

RobotJointAdapter

Bases: ROSAdapterBase[RobotJoint]

Adapter for translating ROS JointState messages to Mosaico RobotJoint.

Supported ROS Types:

• sensor_msgs/msg/JointState

Example

ros_msg = ROSMessage(
    timestamp=17000,
    topic="/joint_states",
    msg_type="sensor_msgs/msg/JointState",
    data={
        "header": {
            "stamp": {
                "sec": 17000,
                "nanosec": 0,
            },
            "frame_id": "",
        },
        "name": ["joint1", "joint2"],
        "position": [0.0, 0.0],
        "velocity": [0.0, 0.0],
        "effort": [0.0, 0.0],
    },
)
# Automatically resolves to a flat Mosaico RobotJoint with attached metadata
mosaico_robot_joint = RobotJointAdapter.translate(ros_msg)

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a RobotJoint object.

Raises:

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

from_dict classmethod

from_dict(ros_data)

Converts the raw dictionary data into the specific Mosaico type.

Example

ros_data={
    "header": {
        "stamp": {
            "sec": 17000,
            "nanosec": 0,
        },
        "frame_id": "",
    },
    "name": ["joint1", "joint2"],
    "position": [0.0, 0.0],
    "velocity": [0.0, 0.0],
    "effort": [0.0, 0.0],
}
# Automatically resolves to a flat Mosaico RobotJoint with attached metadata
mosaico_robot_joint = RobotJointAdapter.from_dict(ros_data)

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico RobotJoint (or a Message wrapping one) into a sensor_msgs/msg/JointState message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, RobotJoint]	A Message wrapping a RobotJoint instance, or a raw RobotJoint.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/JointState is supported.	None

Returns:

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

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.

PointCloudAdapterBase

Bases: ROSAdapterBase[PointCloudModel]

Base adapter for translating ROS PointCloud2 message to Mosaico specific ontology.

decode classmethod

decode(ros_data)

Deserialize the binary buffer of a ROS sensor_msgs/msg/PointCloud2 message into named field arrays.

Parameters:

Name	Type	Description	Default
ros_data	dict	Raw ROS message payload. Relevant keys: data, height, width, fields, is_bigendian, point_step.	required

Returns:

Type	Description
dict[str, list]	dict[str, list]: Dictionary mapping each field name to a list of decoded values. Returns empty lists for all fields if height * width == 0.

Raises:

Type	Description
ValueError	If a field exposes an unsupported datatype.

encode classmethod

encode(pcl_model)

Serializes a dictionary of named field arrays into the binary layout required by a sensor_msgs/msg/PointCloud2 message. This is the inverse operation of :meth:decode.

Parameters:

Name	Type	Description	Default
pcl_model	dict[str, list]	Mapping from field name to a list of scalar values, one per point. None-valued fields should be excluded before calling (e.g., via model_dump(exclude_none=True)).	required

Returns:

Name	Type	Description
dict	dict	A dictionary with all keys required to populate a PointCloud2 message: height, width, fields, is_bigendian, point_step, row_step, data, and is_dense.

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico PointCloudModel subtype (or a Message wrapping one) into a sensor_msgs/msg/PointCloud2 message.

Fields are serialized via :meth:encode. Only non-None model fields are included in the output point cloud.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, PointCloudModel]	A Message wrapping a PointCloudModel instance, or a raw PointCloudModel directly.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/PointCloud2 is supported.	None

Returns:

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

from_dict classmethod

from_dict(ros_data)

Convert a raw ROS PointCloud2 message dictionary into a typed Mosaico model.

Parameters:

Name	Type	Description	Default
ros_data	dict	Raw ROS message payload as a dictionary.	required

Returns:

Name	Type	Description
PointCloudModel	PointCloudModel	A fully populated instance of the target PointCloudModel subtype (e.g. Lidar, Radar, RGBDCamera, ...).

Raises:

Type	Description
ValueError	If any required message key is missing from ros_data, or if any required field is absent after decoding.

Note

This method is not meant to be overridden in most subclasses. Only PointCloudAdapter overrides it, as it operates at a different abstraction level, returning the raw PointCloud2 message instead of a decoded model.

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.

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message instance into a Mosaico Message.

Implementation should handle recursive unwrapping, unit conversion, and validation.

Parameters:

Name	Type	Description	Default
ros_msg	ROSMessage	The source container yielded by the ROSLoader.	required
**kwargs	Any	Contextual data such as calibration parameters or frame overrides.	{}

Returns:

Name	Type	Description
Message	Message	A Mosaico Message object containing the instantiated ontology data.

Raises:

Type	Description
Exception	If translation fails due to missing fields, type mismatches, or other errors.

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.

PointCloudAdapter

Bases: PointCloudAdapterBase[PointCloud2]

Adapter for translating ROS PointCloud2 messages to Mosaico PointCloud2.

Supported ROS Types:

• sensor_msgs/msg/PointCloud2

Example:

ros_msg = ROSMessage(
            timestamp=17000,
            topic="/point_cloud",
            msg_type="sensor_msgs/PointCloud2",
            data = {
                "height": 1,
                "width": 3,
                "fields": [
                    {"name": "x", "offset": 0,  "datatype": 7, "count": 1},
                    {"name": "y", "offset": 4,  "datatype": 7, "count": 1},
                    {"name": "z", "offset": 8,  "datatype": 7, "count": 1},
                ],
            "is_bigendian": False,
            "point_step": 12,
            "row_step": 36,
            "data": ...,
            "is_dense": True,
            }
        )
# Automatically resolves to a flat Mosaico PointCloud2 with attached metadata
mosaico_point_cloud = PointCloudAdapter.translate(ros_msg)

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a PointCloud2 object.

Raises:

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

from_dict classmethod

from_dict(ros_data)

Converts the raw dictionary data into the specific Mosaico type.

Example:

ros_data = {
    "height": 1,           # unorganized point cloud = 1 row
    "width": 3,            # 3 points
    "fields": [
        {
            "name": "x",
            "offset": 0,
            "datatype": 7,  # FLOAT32
            "count": 1
        },
        {
            "name": "y",
            "offset": 4,
            "datatype": 7,  # FLOAT32
            "count": 1
        },
        {
            "name": "z",
            "offset": 8,
            "datatype": 7,  # FLOAT32
            "count": 1
        },
    ],
    "is_bigendian": False,
    "point_step": 12,      # 3 fields * 4 bytes (float32) = 12 bytes per point
    "row_step": 36,        # point_step * width = 12 * 3 = 36 bytes per row
    "data": ...,
    "is_dense": True
    }

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico PointCloud2 (or a Message wrapping one) into a sensor_msgs/msg/PointCloud2 message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, PointCloud2]	A Message wrapping a PointCloud2 instance, or a raw PointCloud2.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/PointCloud2 is supported.	None

Returns:

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

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.

decode classmethod

decode(ros_data)

Deserialize the binary buffer of a ROS sensor_msgs/msg/PointCloud2 message into named field arrays.

Parameters:

Name	Type	Description	Default
ros_data	dict	Raw ROS message payload. Relevant keys: data, height, width, fields, is_bigendian, point_step.	required

Returns:

Type	Description
dict[str, list]	dict[str, list]: Dictionary mapping each field name to a list of decoded values. Returns empty lists for all fields if height * width == 0.

Raises:

Type	Description
ValueError	If a field exposes an unsupported datatype.

encode classmethod

encode(pcl_model)

Serializes a dictionary of named field arrays into the binary layout required by a sensor_msgs/msg/PointCloud2 message. This is the inverse operation of :meth:decode.

Parameters:

Name	Type	Description	Default
pcl_model	dict[str, list]	Mapping from field name to a list of scalar values, one per point. None-valued fields should be excluded before calling (e.g., via model_dump(exclude_none=True)).	required

Returns:

Name	Type	Description
dict	dict	A dictionary with all keys required to populate a PointCloud2 message: height, width, fields, is_bigendian, point_step, row_step, data, and is_dense.

LaserScannerAdapterBase

Bases: ROSAdapterBase[_LT]

Base adapter for translating ROS LaserScan and MultiEchoLaserScan messages to Mosaico LaserScan and MultiEchoLaserScan .

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a LaserScan or MultiEchoLaserScan object.

Raises:

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

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.

from_dict abstractmethod classmethod

from_dict(ros_data)

Maps the raw ROS dictionary to the Pydantic model.

This method performs field validation and reconstruction.

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.

to_ros abstractmethod classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico message or ontology object back into a native ROS message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, T]	A Message wrapper or a raw Serializable ontology instance.	required
typestore	Typestore	The rosbags typestore used to resolve and construct target ROS types.	required
ros_msg_type	Optional[str]	Override for the output ROS type string. If None, the adapter defaults to cls.get_default_ros_msg().	None

Returns:

Name	Type	Description
MsgType	MsgType	The constructed ROS message instance, or raises an error if: the ros_msg_type is unsupported by adapter (TypeError) the ros_msg_type or resolved_rosmsg_type are unsupported by typestore (TypeError) the ros_msg_type or resolved_rosmsg_type are supported but translation is not implemented (NotImplementedError)

ontology_data_type classmethod

ontology_data_type()

Returns the Ontology class type associated with this adapter.

LaserScanAdapter

Bases: LaserScannerAdapterBase[LaserScan]

Adapter for translating ROS LaserScan messages to Mosaico LaserScan.

Supported ROS Types:

• sensor_msgs/msg/LaserScan

from_dict classmethod

from_dict(ros_data)

Create a LaserScan instance from a ROS message dictionary.

Example:

ros_data = {
    "angle_min": -1.57,
    "angle_max":  1.57,
    "angle_increment": 0.01,
    "time_increment": 0.0,
    "scan_time": 0.1,
    "range_min": 0.2,
    "range_max": 10.0,
    "ranges": [1.0, 1.1, 1.2],
    "intensities": [100.0, 110.0, 120.0],
}
# Automatically resolves to a flat Mosaico LaserScan with attached data
mosaico_laser_scan = LaserScanAdapter.from_dict(ros_data)

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico LaserScan (or a Message wrapping one) into a sensor_msgs/msg/LaserScan message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, LaserScan]	A Message wrapping a LaserScan instance, or a raw LaserScan.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/LaserScan is supported.	None

Returns:

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

ros_msg_type abstractmethod classmethod

ros_msg_type()

Returns the specific ROS message type handled by this adapter.

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a LaserScan or MultiEchoLaserScan object.

Raises:

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

is_rosmsg_type_valid classmethod

is_rosmsg_type_valid(type_to_validate)

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

Parameters:

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

Returns:

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

unpack_mosaico_msg classmethod

unpack_mosaico_msg(mosaico_msg)

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

Handles two input cases:

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

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

Parameters:

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

Returns:

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

Raises:

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

schema_metadata classmethod

schema_metadata(typestore, ros_msg_type, ros_version)

Extract the ROS message specific schema metadata, if any.

Parameters:

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

Returns:

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

ontology_data_type classmethod

ontology_data_type()

Returns the Ontology class type associated with this adapter.

MultiEchoLaserScanAdapter

Bases: LaserScannerAdapterBase[MultiEchoLaserScan]

Adapter for translating ROS MultiEchoLaserScan messages to Mosaico MultiEchoLaserScan.

Supported ROS Types:

• sensor_msgs/msg/MultiEchoLaserScan

from_dict classmethod

from_dict(ros_data)

Create a MultiEchoLaserScan instance from a ROS message dictionary.

Example:

ros_data = {
    "angle_min": -1.57,
    "angle_max":  1.57,
    "angle_increment": 0.01,
    "time_increment": 0.0,
    "scan_time": 0.1,
    "range_min": 0.2,
    "range_max": 10.0,
    "ranges": [{"echoes": [1.0, 1.1, 1.2]}, {"echoes": [2.0, 2.1, 2.2]}, {"echoes": [3.0, 3.1, 3.2]}],
    "intensities": [{"echoes": [100.0, 110.0, 120.0]}, {"echoes": [200.0, 210.0, 220.0]}, {"echoes": [300.0, 310.0, 320.0]}],
}
# Automatically resolves to a flat Mosaico MultiEchoLaserScan with attached data
mosaico_laser_scan = MultiEchoLaserScanAdapter.from_dict(ros_data)

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico MultiEchoLaserScan (or a Message wrapping one) into a sensor_msgs/msg/MultiEchoLaserScan message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, MultiEchoLaserScan]	A Message wrapping a MultiEchoLaserScan instance, or a raw MultiEchoLaserScan.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/MultiEchoLaserScan is supported.	None

Returns:

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

ros_msg_type abstractmethod classmethod

ros_msg_type()

Returns the specific ROS message type handled by this adapter.

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a LaserScan or MultiEchoLaserScan object.

Raises:

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

is_rosmsg_type_valid classmethod

is_rosmsg_type_valid(type_to_validate)

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

Parameters:

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

Returns:

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

unpack_mosaico_msg classmethod

unpack_mosaico_msg(mosaico_msg)

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

Handles two input cases:

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

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

Parameters:

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

Returns:

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

Raises:

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

schema_metadata classmethod

schema_metadata(typestore, ros_msg_type, ros_version)

Extract the ROS message specific schema metadata, if any.

Parameters:

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

Returns:

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

ontology_data_type classmethod

ontology_data_type()

Returns the Ontology class type associated with this adapter.

MagneticFieldAdapter

Bases: ROSAdapterBase[Magnetometer]

Adapter for translating ROS MagneticField messages to Mosaico Magnetometer.

Supported ROS Types:

• sensor_msgs/msg/MagneticField

Example

ros_msg = ROSMessage(
    timestamp=17000,
    topic="/magnetic_field",
    msg_type="sensor_msgs/msg/MagneticField",
    data={
        "magnetic_field": {
            "x": 0.12,
            "y": -0.05,
            "z": 0.98,
        }
    }
)

mosaico_magnetometer = MagneticFieldAdapter.translate(ros_msg)

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a Magnetometer object.

Raises:

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

from_dict classmethod

from_dict(ros_data)

Converts the raw dictionary data into the specific Mosaico type.

Parameters:

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

Returns:

Name	Type	Description
Magnetometer	Magnetometer	The constructed Mosaico Magnetometer object.

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico Magnetometer (or a Message wrapping one) into a sensor_msgs/msg/MagneticField message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, Magnetometer]	A Message wrapping a Magnetometer instance, or a raw Magnetometer.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/MagneticField is supported.	None

Returns:

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

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.

JoyAdapter

Bases: ROSAdapterBase[Joy]

Adapter for translating ROS Joy messages to Mosaico Joy.

Supported ROS Types:

• sensor_msgs/msg/Joy

Example

ros_msg = ROSMessage(
    timestamp=17000,
    topic="/joy",
    msg_type="sensor_msgs/msg/Joy",
    data={
        # Axes and buttons are list-based fields
        "axes": [0.0, -1.0, 0.5],
        "buttons": [0, 1, 0, 1],
    }
)

mosaico_joy = JoyAdapter.translate(ros_msg)

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a Joy object.

Raises:

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

from_dict classmethod

from_dict(ros_data)

Converts the raw dictionary data into the specific Mosaico type.

Parameters:

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

Returns:

Name	Type	Description
Joy	Joy	The constructed Mosaico Joy object.

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico Joy (or a Message wrapping one) into a sensor_msgs/msg/Joy message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, Joy]	A Message wrapping a Joy instance, or a raw Joy.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/Joy is supported.	None

Returns:

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

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.

TemperatureAdapter

Bases: ROSAdapterBase[Temperature]

Adapter for translating ROS Temperature messages to Mosaico Temperature.

Supported ROS Types:

• sensor_msgs/msg/Temperature

Example

ros_msg = ROSMessage(
    timestamp=17000,
    topic="/temperature",
    msg_type="sensor_msgs/msg/Temperature",
    data={
        "temperature":  25.0,
        "variance": 3.0,
    }
)

mosaico_temperature = TemperatureAdapter.translate(ros_msg)

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a Temperature object.

Raises:

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

from_dict classmethod

from_dict(ros_data)

Converts the raw dictionary data into the specific Mosaico type.

Parameters:

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

Returns:

Name	Type	Description
Temperature	Temperature	The constructed Mosaico Temperature object.

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico Temperature (or a Message wrapping one) into a sensor_msgs/msg/Temperature message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, Temperature]	A Message wrapping a Temperature instance, or a raw Temperature.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/Temperature is supported.	None

Returns:

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

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.

PressureAdapter

Bases: ROSAdapterBase[Pressure]

Adapter for translating ROS FluidPressure messages to Mosaico Pressure.

Supported ROS Types:

• sensor_msgs/msg/FluidPressure

Example

ros_msg = ROSMessage(
    timestamp=17000,
    topic="/pressure",
    msg_type="sensor_msgs/msg/FluidPressure",
    data={
        "fluid_pressure":  25.0,
        "variance": 3.0,
    }
)

mosaico_pressure = PressureAdapter.translate(ros_msg)

translate classmethod

translate(ros_msg, **kwargs)

Translates a ROS message into a Mosaico Message.

Parameters:

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

Returns:

Name	Type	Description
Message	Message	The translated message containing a Pressure object.

Raises:

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

from_dict classmethod

from_dict(ros_data)

Converts the raw dictionary data into the specific Mosaico type.

Parameters:

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

Returns:

Name	Type	Description
Pressure	Pressure	The constructed Mosaico Pressure object.

to_ros classmethod

to_ros(mosaico_data, typestore, ros_msg_type=None)

Converts a Mosaico Pressure (or a Message wrapping one) into a sensor_msgs/msg/FluidPressure message.

Parameters:

Name	Type	Description	Default
mosaico_data	Union[Message, Pressure]	A Message wrapping a Pressure instance, or a raw Pressure.	required
typestore	Typestore	The rosbags typestore for target type resolution.	required
ros_msg_type	Optional[str]	Override for the output ROS type. Only sensor_msgs/msg/FluidPressure is supported.	None

Returns:

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

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.