MultiTraceReader

class osi_utilities.tracefile.readers.multi.MultiTraceReader(decoder_mode='precompiled', skip_incompatible_messages=True, log_incompatible_messages=True)

Bases: TraceReader

Reader for multi-channel OSI trace files (.mcap).

Supports multi-channel reading with topic-based filtering, schema-based message type detection, and metadata access.

Incompatible message handling:

• Controlled by set_skip_incompatible_messages(...).

• If enabled (default), incompatible messages are skipped.

• If disabled, incompatible messages are surfaced as ReadResult(status=ReadStatus.INCOMPATIBLE, ...).

Logging behavior for incompatible messages is controlled independently by set_log_incompatible_messages(...).

Parameters:

• decoder_mode (str)

• skip_incompatible_messages (bool)

• log_incompatible_messages (bool)

set_decoder_mode(decoder_mode)

Set protobuf decoding strategy.

Supported values: - precompiled: decode OSI schemas via precompiled osi3 classes - mcap-contained: decode via schema descriptor sets embedded in MCAP

Parameters:

decoder_mode (str)

Return type:

None

open(path)

Open an MCAP trace file.

Parameters:

path (Path) – Path to the .mcap file.

Returns:

True on success, False on failure.

Return type:

bool

set_topics(topics)

Filter reading to specific topics.

Parameters:

topics (list[str]) – List of topic names to read. If empty, reads all topics.

Return type:

None

set_topic_message_types(topic_message_types)

Set expected message types per topic.

Parameters:

topic_message_types (dict[str, MessageType | str | None]) – Mapping of topic name to expected message type. Entries with None remove expectations for that topic.

Return type:

None

set_log_incompatible_messages(log_messages)

Configure whether incompatible messages should be logged.

Parameters:

log_messages (bool) – If True (default), incompatible messages are logged. If False, incompatible messages are not logged.

Return type:

None

set_skip_incompatible_messages(skip)

Configure whether incompatible messages should be skipped.

Incompatible messages are non-matching encoding/schema/message-type cases. Actual decode/read errors are always surfaced as ReadStatus.ERROR and are never skipped by this setting.

Parameters:

skip (bool)

Return type:

None

read_message()

Read the next message from the MCAP file.

Returns:

• ReadStatus.OK for a valid decoded message

• ReadStatus.INCOMPATIBLE for incompatible messages when set_skip_incompatible_messages(False) is configured

• ReadStatus.ERROR for runtime read/decode failures

Return type:

None on EOF; otherwise a ReadResult with status

has_next()

Check if there are more messages to read.

Peeks ahead in the iterator to give an accurate answer.

Return type:

bool

close()

Close the MCAP file.

Return type:

None

get_available_topics()

Return list of available topic names in the file.

Return type:

list[str]

get_file_metadata()

Return file-level metadata entries.

Return type:

list[dict]

get_channel_specification(topic)

Return ChannelSpecification for a specific OSI-compatible topic.

Returns None when the topic is missing or not a supported OSI schema.

Parameters:

topic (str)

Return type:

ChannelSpecification | None

list_channel_specifications()

Return ChannelSpecification objects for all OSI-compatible topics.

Return type:

list[ChannelSpecification]

get_message_type_for_topic(topic)

Return the MessageType for a given topic based on its schema.

Parameters:

topic (str)

Return type:

MessageType | None

get_channel_metadata(topic)

Return metadata for a specific channel/topic.

Parameters:

topic (str)

Return type:

dict[str, str] | None