Base Adapter

mosaicolabs.ros_bridge.adapter_base ¶

RosSchemaMetadata ¶

RosSchemaMetadata(**fields)

Encapsulates Mosaico's reserved _ros_ topic-metadata namespace in a single place.

Every topic ingested by the ROS bridge carries ROS-specific bookkeeping (original msgtype, raw msgdef, extracted enums) plus bridge-internal fields (e.g. the source bag file) under one reserved key, so that:

The literal string "_ros_" exists in exactly one place (:attr:KEY), instead of being duplicated across adapters, loaders, and the injector.
Callers build up this namespace incrementally via :meth:update without ever touching the wrapping dict shape by hand, which is what previously caused bugs (e.g. accidentally overwriting the whole _ros_ block instead of merging into it).

Example

meta = RosSchemaMetadata(msgtype="sensor_msgs/msg/Imu").update(source_file="a.mcap")
topic_metadata = meta.merge_into(user_supplied_metadata)
# topic_metadata == {..., "_ros_": {"msgtype": "sensor_msgs/msg/Imu", "source_file": "a.mcap"}}

KEY `class-attribute` ¶

KEY = '_ros_'

The reserved metadata key. Adapters/loaders/the injector should reference this constant rather than the literal string, so the namespace can be renamed in one place.

update ¶

update(**fields)

Merges additional fields into this block, in place. Returns self for chaining.

Parameters:

Name	Type	Description	Default
`**fields`	`Any`	Additional fields to merge.	`{}`

Returns:

Name	Type	Description
`RosSchemaMetadata`	`'RosSchemaMetadata'`	The updated metadata instance.

to_dict ¶

to_dict()

Wraps the current fields under the reserved key, e.g. {"_ros_": {...}}.

Returns:

Name	Type	Description
`dict`	`dict`	A dictionary containing the `_ros_` block with the current fields.

merge_into ¶

merge_into(metadata)

Merges this block into an existing metadata dict's _ros_ namespace, creating it if absent. Mutates and returns metadata.

Parameters:

Name	Type	Description	Default
`metadata`	`dict`	The existing metadata dict to merge into.	required

Returns:

Name	Type	Description
`dict`	`dict`	The updated metadata dict with the `_ros_` block merged in.

extract `classmethod` ¶

extract(metadata)

Reads the _ros_ block out of a metadata dict (e.g. a topic's user_metadata), or {} if absent.

Parameters:

Name	Type	Description	Default
`metadata`	`Optional[dict]`	A metadata dict, typically `{"_ros_": {...}}` or `None`.	required

Returns:

Name	Type	Description
`dict`	`dict`	The extracted `_ros_` block, or an empty dict if not present.

from_dict `classmethod` ¶

from_dict(metadata)

Creates a RosSchemaMetadata from a plain metadata dict, e.g. the return value of ROSAdapterBase.schema_metadata(). Any keys outside the _ros_ namespace are ignored.

Parameters:

Name	Type	Description	Default
`metadata`	`Optional[dict]`	A metadata dict, typically `{"_ros_": {...}}` or `None`.	required

Returns:

Name	Type	Description
`RosSchemaMetadata`	`'RosSchemaMetadata'`	A new instance seeded with the extracted `_ros_` fields (empty if `metadata` is `None` or carries no `_ros_` block).

ROSAdapterBase ¶

Bases: ABC, Generic[T]

Abstract Base Class for converting ROS messages to Mosaico Ontology types.

The Adaptation Layer is the semantic core of the ROS Bridge. Rather than performing simple parsing, adapters actively translate raw ROS data into standardized, strongly-typed Mosaico Ontology objects.

Attributes:

Name	Type	Description
`ros_msgtype`	`str \| Tuple[str, ...]`	The ROS message type string (e.g., 'sensor_msgs/msg/Imu') or a tuple of supported types.
`is_default_adapter`	`str \| Tuple[str, ...]`	whether this adapter is the default one for mosaico_ontology_type
`__mosaico_ontology_type__`	`Type[T]`	The target Mosaico class (e.g., IMU).
`_REQUIRED_KEYS`	`Tuple[str, ...]`	Internal validation list for mandatory ROS message fields.

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.

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)

schema_metadata `classmethod` ¶

schema_metadata(typestore, ros_msg_type, ros_version)

Extract the ROS message specific schema metadata, if any.

Parameters:

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

Returns:

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

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

ontology_data_type `classmethod` ¶

ontology_data_type()

Returns the Ontology class type associated with this adapter.

Base Adapter

mosaicolabs.ros_bridge.adapter_base ¶

RosSchemaMetadata ¶

KEY class-attribute ¶

update ¶

to_dict ¶

merge_into ¶

extract classmethod ¶

from_dict classmethod ¶

ROSAdapterBase ¶

ros_msg_type abstractmethod classmethod ¶

translate classmethod ¶

from_dict abstractmethod classmethod ¶

is_rosmsg_type_valid classmethod ¶

unpack_mosaico_msg classmethod ¶

to_ros abstractmethod classmethod ¶

schema_metadata classmethod ¶

ontology_data_type classmethod ¶

KEY `class-attribute` ¶

extract `classmethod` ¶

from_dict `classmethod` ¶

ros_msg_type `abstractmethod` `classmethod` ¶

translate `classmethod` ¶

from_dict `abstractmethod` `classmethod` ¶

is_rosmsg_type_valid `classmethod` ¶

unpack_mosaico_msg `classmethod` ¶

to_ros `abstractmethod` `classmethod` ¶

schema_metadata `classmethod` ¶

ontology_data_type `classmethod` ¶