Converter Modules

Internal converter modules for different message formats.

`tabular2mcap.converter.json`

JSON converter utilities for MCAP writing.

`JsonConverter`

Bases: ConverterBase

JSON format converter that wraps JSON-specific MCAP writer operations.

`writer` `property`

Get the underlying writer instance.

`init(writer=None)`

Initialize the JSON converter with a writer instance.

Parameters:

Name	Type	Description	Default
`writer`	`Writer \| None`	MCAP writer instance for JSON format	`None`

`get_schema_template(schema_name)`

Get the schema template for a given schema name.

Parameters:

Name	Type	Description	Default
`schema_name`	`str`	Name of the schema	required

Returns:

Type	Description
`str`	Schema template

`register_generic_schema(df, schema_name, exclude_keys=None)`

Register a generic JSON schema from a DataFrame.

Parameters:

Name	Type	Description	Default
`df`	`Any`	DataFrame to generate schema from	required
`schema_name`	`str`	Name for the schema	required
`exclude_keys`	`list[str] \| None`	Optional list of keys to exclude from schema	`None`

Returns:

Type	Description
`tuple[int, list]`	Tuple of (schema_id, schema_keys)

`register_schema(schema_name)`

Register a predefined schema by name.

Parameters:

Name	Type	Description	Default
`schema_name`	`str`	Name of the schema (e.g., "foxglove.LocationFix")	required

Returns:

Type	Description
`int`	Schema ID

`write_messages_from_iterator(iterator, topic_name, schema_id, data_length=None, unit='msg')`

Write messages to MCAP from an iterator.

Parameters:

Name	Type	Description	Default
`iterator`	`Iterable[tuple[int, ConvertedRow]]`	Iterator yielding (index, message) tuples	required
`topic_name`	`str`	Topic name for the messages	required
`schema_id`	`int \| None`	Schema ID for the messages	required
`data_length`	`int \| None`	Optional total length for progress tracking	`None`
`unit`	`str`	Unit label for progress tracking	`'msg'`

`json_schema_to_template(schema_json_str)`

Convert schema JSON to template dict with only properties.

`register_foxglove_schema(writer, schema_name)`

Register a Foxglove schema with the MCAP writer.

Parameters:

Name	Type	Description	Default
`writer`	`Writer`	MCAP writer instance	required
`schema_name`	`str`	Name of the Foxglove schema (e.g., "LocationFix")	required

Returns:

Type	Description
`int`	Schema ID for use in channel registration

`register_json_schema_from_columns(writer, schema_name, columns)`

Create and register a JSON schema from column names and their dtypes.

Parameters:

Name	Type	Description	Default
`writer`	`Writer`	MCAP writer instance	required
`schema_name`	`str`	Name of the schema (e.g., "LocationFix")	required
`columns`	`list[tuple[str, Any]]`	List of (key, dtype) tuples for schema generation	required

Returns:

Type	Description
`int`	Schema ID that was registered

`tabular2mcap.converter.ros2`

`Ros2Converter`

Bases: ConverterBase

ROS2 format converter that wraps ROS2-specific MCAP writer operations.

`writer` `property`

Get the underlying writer instance (accesses _writer attribute for ROS2).

`init(writer=None)`

Initialize the ROS2 converter with a writer instance.

Parameters:

Name	Type	Description	Default
`writer`	`Writer \| None`	MCAP ROS2 writer instance	`None`

`get_schema_template(schema_name)`

Get the schema template for a given ROS2 schema name.

Parameters:

Name	Type	Description	Default
`schema_name`	`str`	Name of the ROS2 schema (e.g., 'sensor_msgs/msg/NavSatFix')	required

Returns:

Type	Description
`str`	Schema template

`register_generic_schema(df, schema_name, exclude_keys=None)`

Register a generic ROS2 schema from a DataFrame.

Parameters:

Name	Type	Description	Default
`df`	`Any`	DataFrame to generate schema from	required
`schema_name`	`str`	Name for the schema	required
`exclude_keys`	`list[str] \| None`	Optional list of keys to exclude from schema	`None`

Returns:

Type	Description
`tuple[Any, dict[str, str]]`	Tuple of (schema, schema_keys)

`register_schema(schema_name)`

Register a predefined ROS2 schema by name.

Parameters:

Name	Type	Description	Default
`schema_name`	`str`	Name of the ROS2 message schema	required

Returns:

Type	Description
`Any`	Schema object

`sanitize_schema_name(schema_name)` `staticmethod`

Convert string to valid ROS2 schema name by removing invalid characters.

`write_messages_from_iterator(iterator, topic_name, schema_id, data_length=None, unit='msg')`

Write messages to MCAP from an iterator.

Parameters:

Name	Type	Description	Default
`iterator`	`Iterable[tuple[int, ConvertedRow]]`	Iterator yielding (index, message) tuples	required
`topic_name`	`str`	Topic name for the messages	required
`schema_id`	`int \| None`	Schema ID for the messages	required
`data_length`	`int \| None`	Optional total length for progress tracking	`None`
`unit`	`str`	Unit label for progress tracking	`'msg'`

`numpy_to_ros2_type(dtype, sample_data=None)`

Convert numpy dtype to ROS2 type.

`ros2_msg_to_template(msg_def_text, msg_type)`

Convert ROS2 message definition directly to template dict.

`sanitize_ros2_field_name(key)`

Convert string to valid ROS2 field name by removing invalid characters.

`tabular2mcap.converter.functions`

`ConverterFunction`

Bases: BaseModel

`convert_row(row)`

Convert a pandas row using the Jinja2 template.

`init_jinja2_template()`

Initialize Jinja2 template and environment after model creation.

`ConverterFunctionJinja2Environment`

Bases: Environment

Specialized Jinja2 environment for ConverterFunction that provides math functions, custom filters, and disables autoescape for JSON-like output.

`tabular2mcap.converter.others`

`LogConverter`

Converter for parsing log files into MCAP log messages.

Parses log files using regex patterns defined in a Jinja2 template format. Handles multi-line log entries by detecting log entry starts.

Attributes:

Name	Type	Description
`log_path`	`Path`	Path to the log file to parse
`name`	`str`	Name identifier for the log source (default: "Log")
`datetime_format`	`str`	Format of the timestamp in the log file (default: "%Y-%m-%d %H:%M:%S")
`zero_first_timestamp`	`bool`	If True, the first timestamp will be 0

`init(log_path, format_template, writer_format, zero_first_timestamp=False, name='Log', datetime_format='%Y-%m-%d %H:%M:%S')`

Initialize log converter.

Parameters:

Name	Type	Description	Default
`log_path`	`Path`	Path to the log file to parse	required
`format_template`	`str`	Jinja2 template string for regex pattern. Uses variables: levelname, message, asctime (default format: 'YYYY-MM-DD HH:MM:SS'), filename, lineno	required
`writer_format`	`WRITER_FORMATS`	Writer format to use for the log messages	required
`zero_first_timestamp`	`bool`	If True, the first timestamp will be 0	`False`
`name`	`str`	Name identifier for the log source	`'Log'`
`datetime_format`	`str`	Format of the timestamp in the log file (default: "%Y-%m-%d %H:%M:%S")	`'%Y-%m-%d %H:%M:%S'`

`log_iter()`

Generate log messages from log data.

Parses log file entries and yields ConvertedRow objects containing foxglove.Log format messages. Handles multi-line log entries.

Yields:

Type	Description
`Iterable[ConvertedRow]`	ConvertedRow with log message data in foxglove.Log format.
`Iterable[ConvertedRow]`	Each entry contains timestamp (epoch time), level, message, file, and line.

`compressed_image_message_iterator(video_frames, fps, format, frame_id, use_foxglove_format=True, writer_format='json')`

Generate compressed image messages from video frames.

Parameters:

Name	Type	Description	Default
`video_frames`	`list[ndarray]`	List of video frames as numpy arrays	required
`fps`	`float`	Frames per second	required
`format`	`str`	Image format (e.g., 'jpeg', 'png')	required
`frame_id`	`str`	Frame ID string	required
`use_foxglove_format`	`bool`	If True, use Foxglove message format; otherwise use ROS2 format	`True`
`writer_format`	`str`	MCAP writer format ('json', 'ros2', 'protobuf')	`'json'`

`compressed_video_message_iterator(video_frames, fps, format, frame_id, use_foxglove_format=True, writer_format='json')`

Generate compressed video messages from video frames.

Parameters:

Name	Type	Description	Default
`video_frames`	`list[ndarray]`	List of video frames as numpy arrays	required
`fps`	`float`	Frames per second	required
`format`	`str`	Video format (e.g., 'h264', 'h265')	required
`frame_id`	`str`	Frame ID string	required
`use_foxglove_format`	`bool`	Unused - CompressedVideo only has Foxglove schema (no ROS2 equivalent). Kept for API consistency with compressed_image_message_iterator.	`True`
`writer_format`	`str`	MCAP writer format ('json', 'ros2', 'protobuf')	`'json'`

`create_foxglove_compressed_image_data(frame_timestamp, frame_id, encoded_data, format)`

Create Foxglove CompressedImage message data structure (JSON format).

Parameters:

Name	Type	Description	Default
`frame_timestamp`	`float`	Timestamp in seconds	required
`frame_id`	`str`	Frame ID string	required
`encoded_data`	`bytes`	Encoded image data as bytes	required
`format`	`str`	Image format (e.g., 'jpeg', 'png')	required

Returns:

Type	Description
`dict`	Dictionary with Foxglove CompressedImage message structure

`create_foxglove_protobuf_compressed_image_data(frame_timestamp, frame_id, encoded_data, format)`

Create Foxglove CompressedImage message data structure (Protobuf format).

Parameters:

Name	Type	Description	Default
`frame_timestamp`	`float`	Timestamp in seconds	required
`frame_id`	`str`	Frame ID string	required
`encoded_data`	`bytes`	Encoded image data as bytes	required
`format`	`str`	Image format (e.g., 'jpeg', 'png')	required

Returns:

Type	Description
`dict`	Dictionary with Foxglove Protobuf CompressedImage message structure

`create_ros2_compressed_image_data(frame_timestamp, frame_id, encoded_data, format)`

Create ROS2 sensor_msgs/msg/CompressedImage message data structure.

Parameters:

Name	Type	Description	Default
`frame_timestamp`	`float`	Timestamp in seconds	required
`frame_id`	`str`	Frame ID string	required
`encoded_data`	`bytes`	Encoded image data as bytes	required
`format`	`str`	Image format (e.g., 'jpeg', 'png')	required

Returns:

Type	Description
`dict`	Dictionary with ROS2 CompressedImage message structure

`to_foxglove_log_level(level)`

Convert log level string to foxglove.Log level constant.

Parameters:

Name	Type	Description	Default
`level`	`str`	Log level string (DEBUG, INFO, WARNING, ERROR, FATAL)	required

Returns:

Type	Description
`int`	Integer level constant for foxglove.Log: - DEBUG: 1 - INFO: 2 - WARNING: 3 - ERROR: 4 - FATAL: 5 - Unknown: 0

`to_ros2_log_level(level)`

Convert log level string to ROS2 log level constant.

Parameters:

Name	Type	Description	Default
`level`	`str`	Log level string (DEBUG, INFO, WARNING, ERROR, FATAL)	required

Returns:

Type	Description
`int`	Integer level constant for ROS2 log

Converter Modules

tabular2mcap.converter.json

JsonConverter

writer property

__init__(writer=None)

get_schema_template(schema_name)

register_generic_schema(df, schema_name, exclude_keys=None)

register_schema(schema_name)

write_messages_from_iterator(iterator, topic_name, schema_id, data_length=None, unit='msg')

json_schema_to_template(schema_json_str)

register_foxglove_schema(writer, schema_name)

register_json_schema_from_columns(writer, schema_name, columns)

tabular2mcap.converter.ros2

Ros2Converter

writer property

__init__(writer=None)

get_schema_template(schema_name)

register_generic_schema(df, schema_name, exclude_keys=None)

register_schema(schema_name)

sanitize_schema_name(schema_name) staticmethod

write_messages_from_iterator(iterator, topic_name, schema_id, data_length=None, unit='msg')

numpy_to_ros2_type(dtype, sample_data=None)

ros2_msg_to_template(msg_def_text, msg_type)

sanitize_ros2_field_name(key)

tabular2mcap.converter.functions

ConverterFunction

convert_row(row)

init_jinja2_template()

ConverterFunctionJinja2Environment

tabular2mcap.converter.others

LogConverter

__init__(log_path, format_template, writer_format, zero_first_timestamp=False, name='Log', datetime_format='%Y-%m-%d %H:%M:%S')

log_iter()

compressed_image_message_iterator(video_frames, fps, format, frame_id, use_foxglove_format=True, writer_format='json')

compressed_video_message_iterator(video_frames, fps, format, frame_id, use_foxglove_format=True, writer_format='json')

create_foxglove_compressed_image_data(frame_timestamp, frame_id, encoded_data, format)

create_foxglove_protobuf_compressed_image_data(frame_timestamp, frame_id, encoded_data, format)

create_ros2_compressed_image_data(frame_timestamp, frame_id, encoded_data, format)

to_foxglove_log_level(level)

to_ros2_log_level(level)

`tabular2mcap.converter.json`

`JsonConverter`

`writer` `property`

`init(writer=None)`

`get_schema_template(schema_name)`

`register_generic_schema(df, schema_name, exclude_keys=None)`

`register_schema(schema_name)`

`write_messages_from_iterator(iterator, topic_name, schema_id, data_length=None, unit='msg')`

`json_schema_to_template(schema_json_str)`

`register_foxglove_schema(writer, schema_name)`

`register_json_schema_from_columns(writer, schema_name, columns)`

`tabular2mcap.converter.ros2`

`Ros2Converter`

`writer` `property`

`init(writer=None)`

`get_schema_template(schema_name)`

`register_generic_schema(df, schema_name, exclude_keys=None)`

`register_schema(schema_name)`

`sanitize_schema_name(schema_name)` `staticmethod`

`write_messages_from_iterator(iterator, topic_name, schema_id, data_length=None, unit='msg')`

`numpy_to_ros2_type(dtype, sample_data=None)`

`ros2_msg_to_template(msg_def_text, msg_type)`

`sanitize_ros2_field_name(key)`

`tabular2mcap.converter.functions`

`ConverterFunction`

`convert_row(row)`

`init_jinja2_template()`

`ConverterFunctionJinja2Environment`

`tabular2mcap.converter.others`

`LogConverter`

`init(log_path, format_template, writer_format, zero_first_timestamp=False, name='Log', datetime_format='%Y-%m-%d %H:%M:%S')`

`log_iter()`

`compressed_image_message_iterator(video_frames, fps, format, frame_id, use_foxglove_format=True, writer_format='json')`

`compressed_video_message_iterator(video_frames, fps, format, frame_id, use_foxglove_format=True, writer_format='json')`

`create_foxglove_compressed_image_data(frame_timestamp, frame_id, encoded_data, format)`

`create_foxglove_protobuf_compressed_image_data(frame_timestamp, frame_id, encoded_data, format)`

`create_ros2_compressed_image_data(frame_timestamp, frame_id, encoded_data, format)`

`to_foxglove_log_level(level)`

`to_ros2_log_level(level)`