Base Adapter

mosaicolabs.bridges.mcap.adapter_base ¶

MCAPSchemaMetadata ¶

MCAPSchemaMetadata(**fields)

Bases: BaseSchemaMetadata

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

Every topic ingested by the MCAP bridge carries MCAP-specific bookkeeping (original channel_name, channel_encoding, schema_name, schema_encoding, raw schema_def) plus bridge-internal fields (e.g. the source mcap file) under one reserved key, so that:

The literal string "_mcap_" exists in exactly one place (KEY), instead of being duplicated across adapters, loaders, and the injector.
Callers build up this namespace incrementally via update without ever touching the wrapping dict shape by hand.

Example

# Every key in REQUIRED_KEYS must be provided, otherwise a ValueError is raised
meta = MCAPSchemaMetadata(
    channel_name="/imu",
    channel_encoding="protobuf",
    schema_name="sensor_msgs.Imu",
    schema_encoding="protobuf",
    schema_def="<stringified schema definition>",
).update(source_file="a.mcap")
topic_metadata = meta.merge_into(user_supplied_metadata)
# topic_metadata == {..., "_mcap_": {"channel_name": "/imu", ..., "source_file": "a.mcap"}}

KEY `class-attribute` ¶

KEY = '_mcap_'

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.

REQUIRED_KEYS `class-attribute` ¶

REQUIRED_KEYS = {
    "schema_name": str,
    "schema_encoding": str,
    "schema_def": str,
    "channel_name": str,
    "channel_encoding": str,
}

The keys that need to be present in order to create the MCAPSchemaMetadata.

get_channel_name ¶

get_channel_name()

Returns the channel name contained in the metadata.

get_channel_encoding ¶

get_channel_encoding()

Returns the channel encoding contained in the metadata.

get_schema_name ¶

get_schema_name()

Returns the schema name contained in the metadata.

get_schema_encoding ¶

get_schema_encoding()

Returns the schema encoding contained in the metadata.

get_schema_def ¶

get_schema_def()

Returns the schema def contained in the metadata.

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
`BaseSchemaMetadata`	`BaseSchemaMetadata`	The updated metadata instance.

to_dict ¶

to_dict()

Wraps the current fields under the reserved key.

Returns:

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

merge_into ¶

merge_into(metadata)

Merges this block into an existing metadata dict's KEY 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 `KEY` block merged in.

extract `classmethod` ¶

extract(metadata)

Reads the KEY block out of a metadata dict or {} if absent.

Parameters:

Name	Type	Description	Default
`metadata`	`Optional[dict]`	A metadata dict, typically `{KEY: {...}}` (e.g. `{"_ros_": {...}}` or `{"_mcap_": {...}}`), or `None`.	required

Returns:

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

from_dict `classmethod` ¶

from_dict(metadata)

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

Parameters:

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

Returns:

Name	Type	Description
`BaseSchemaMetadata`	`BaseSchemaMetadata`	A new instance seeded with the extracted `KEY` fields (empty if `metadata` is `None` or carries no `KEY` block, and `REQUIRED_KEYS` is empty).

Raises:

Type	Description
`ValueError`	If any of `REQUIRED_KEYS` is missing from the extracted fields.
`TypeError`	If any of `REQUIRED_KEYS` has an unexpected type.

McapReturnType `dataclass` ¶

McapReturnType(data_bytes, publish_time_ns)

Container returned by MCAPAdapterBase.to_native.

Attributes:

Name	Type	Description
`data_bytes`	`bytes`	The native MCAP message, serialized according to the adapter's `schema_encoding`.
`publish_time_ns`	`Optional[int]`	The message publish time (nanoseconds) returned by `to_mcap()`, or `None` if not available.

MCAPAdapterBase ¶

Bases: BridgeAdapterBase[OntologyT, McapReturnType], Generic[OntologyT, McapT]

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

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

Attributes:

Name	Type	Description
`schema_name`	`str`	The MCAP schema name this adapter handles (e.g. `"sensor_msgs/Imu"`).
`schema_encoding`	`str`	The MCAP schema encoding this adapter handles (e.g. `"protobuf"`).
`_REQUIRED_KEYS`	`Tuple[str, ...]`	Internal validation list for mandatory MCAP message fields.

translate `classmethod` ¶

translate(msg, **kwargs)

Translates a MCAP message instance into a Mosaico Message.

Parameters:

Name	Type	Description	Default
`msg`	`MCAPMessage`	The source container yielded by the MCAPLoader.	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(mcap_data, publish_time_ns)

Maps the raw MCAP dictionary to the Mosaico model.

This method performs field validation and reconstruction.

Parameters:

Name	Type	Description	Default
`mcap_data`	`dict`	The decoded MCAP message payload.	required
`publish_time_ns`	`int`	The message publish time (nanoseconds).	required

Returns:

Name	Type	Description
`OntologyT`	`OntologyT`	The constructed ontology instance.

Raises:

Type	Description
`NotImplementedError`	If the adapter does not override this method.

to_native `classmethod` ¶

to_native(mosaico_data, **kwargs)

Converts a Mosaico message or ontology object into a serialized native MCAP message.

Parameters:

Name	Type	Description	Default
`mosaico_data`	`Union[Message, OntologyT]`	A `Message` wrapper or a raw `Serializable` ontology instance to convert.	required
`**kwargs`	`Any`	Extra arguments forwarded to `to_mcap`.	`{}`

Returns:

Name	Type	Description
`McapReturnType`	`McapReturnType`	The native MCAP message serialized according to `schema_encoding` (protobuf wire bytes or UTF-8 JSON), plus the publish time returned by `to_mcap()`.

Raises:

Type	Description
`TypeError`	If `to_mcap()` returns a type not matching `schema_encoding`.
`NotImplementedError`	If `schema_encoding` is neither `"protobuf"` nor `"jsonschema"`, or if the adapter does not implement `to_mcap()`.

to_mcap `abstractmethod` `classmethod` ¶

to_mcap(mosaico_data)

Converts a Mosaico ontology object back into a native MCAP message.

Parameters:

Name	Type	Description	Default
`mosaico_data`	`OntologyT`	The ontology instance to convert.	required

Returns:

Type	Description
`Tuple[McapT, Optional[int]]`	Tuple[McapT, Optional[int]]: The native MCAP message (a protobuf `Message` or a plain `Dict`, depending on `schema_encoding`) and its publish time (nanoseconds), or `None` if not available.

Raises:

Type	Description
`NotImplementedError`	If the adapter does not override this method.

is_mcap_type_valid `classmethod` ¶

is_mcap_type_valid(
    encoding_to_validate, schema_name_to_validate
)

Checks whether a given MCAP message type (schema_name, schema_encoding) is handled by this adapter.

Parameters:

Name	Type	Description	Default
`encoding_to_validate`	`str`	The full MCAP encoding string to check (e.g., `"protobuf"`).	required
`schema_name_to_validate`	`str`	The full MCAP schema name string to check (e.g., `"foxglove.Imu"`).	required

Returns:

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

schema_metadata `classmethod` ¶

schema_metadata(channel_name, channel_encoding, schema_def)

Builds the MCAP-specific schema metadata for this adapter.

Parameters:

Name	Type	Description	Default
`channel_name`	`str`	The full name of the MCAP channel the adapter is associated to.	required
`channel_encoding`	`str`	The encoding of the MCAP channel the adapter is associated to.	required
`schema_def`	`str`	The string representation of the MCAP schema the adapter is associated to, as produced by the resolved `McapSchemaConverter.stringify_schema_def()`.	required

Returns:

Type	Description
`Optional[dict]`	Optional[dict]: The schema metadata, wrapped under the `MCAPSchemaMetadata.KEY` namespace.

ontology_data_type `classmethod` ¶

ontology_data_type()

Returns the Ontology class type associated with this adapter.

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.

In both cases the Header is extracted from the ontology (if supported), otherwise a 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.

compute_mcap_msg_type ¶

compute_mcap_msg_type(schema_name, schema_encoding)

Returns the specific combination of the schema name + schema encoding, defining the MCAP message type handled by this adapter.

Parameters:

Name	Type	Description	Default
`schema_name`	`str`	the schema name of the MCAP message.	required
`schema_encoding`	`str`	the schema encoding of the MCAP message.	required

Returns:

Name	Type	Description
`str`	`str`	The unique combination for passed schema_name and schema_encoding.

Base Adapter

mosaicolabs.bridges.mcap.adapter_base ¶

MCAPSchemaMetadata ¶

KEY class-attribute ¶

REQUIRED_KEYS class-attribute ¶

get_channel_name ¶

get_channel_encoding ¶

get_schema_name ¶

get_schema_encoding ¶

get_schema_def ¶

update ¶

to_dict ¶

merge_into ¶

extract classmethod ¶

from_dict classmethod ¶

McapReturnType dataclass ¶

MCAPAdapterBase ¶

translate classmethod ¶

from_dict abstractmethod classmethod ¶

to_native classmethod ¶

to_mcap abstractmethod classmethod ¶

is_mcap_type_valid classmethod ¶

schema_metadata classmethod ¶

ontology_data_type classmethod ¶

unpack_mosaico_msg classmethod ¶

compute_mcap_msg_type ¶

KEY `class-attribute` ¶

REQUIRED_KEYS `class-attribute` ¶

extract `classmethod` ¶

from_dict `classmethod` ¶

McapReturnType `dataclass` ¶

translate `classmethod` ¶

from_dict `abstractmethod` `classmethod` ¶

to_native `classmethod` ¶

to_mcap `abstractmethod` `classmethod` ¶

is_mcap_type_valid `classmethod` ¶

schema_metadata `classmethod` ¶

ontology_data_type `classmethod` ¶

unpack_mosaico_msg `classmethod` ¶