cras_bag_tools package

Submodules

cras_bag_tools.fix_msg_defs module

cras_bag_tools.message_filter module

cras_bag_tools.message_filters module

cras_bag_tools.topic_set module

Efficient data structure to hold a static set of topics with super-fast is-in-set queries.

The querying is done using expressions like topic in set or topic not in set.

class cras_bag_tools.topic_set.TopicSet(items: Iterable[str] | None = None)

Bases: object

Efficient data structure to hold a static set of topics with super-fast is-in-set queries.

cras_bag_tools.tqdm_bag module

Bag file reader that shows progressbars when reading messages.

class cras_bag_tools.tqdm_bag.BagIterator(*args: Any, **kwargs: Any)

Bases: SequentialReader

Utility class for easier working with ROS2 BAG files.

read_messages(topics: list[str] | None = None, start_time: int | None = None, end_time: int | None = None) → Generator[tuple[str, bytes, int], Any, None]

Iterate through messages in the bag.

Parameters:

topics – If nonempty, this is the list of topics to filter by.
start_time – If set, this is the time of the first read message (in nanoseconds since epoch).
end_time – If set, this is the time of the last read message (in nanoseconds since epoch).

Returns:

Iterator over messages. Returns tuples (topic, serialized message byte content, receive stamp (nanos)).

class cras_bag_tools.tqdm_bag.TqdmBag(*args: Any, **kwargs: Any)

Bases: BagIterator

Extension of BagIterator which uses tqdm to report progress.

read_messages(topics: list[str] | None = None, start_time: int | None = None, end_time: int | None = None) → Generator[tuple[str, bytes, int], Any, None] | tqdm.tqdm