Standalone .mcap dataloader withouth any ROS distribution.
(self, data_dir: str, topic: str, *_, **__)
| 26 | |
| 27 | class McapDataloader: |
| 28 | def __init__(self, data_dir: str, topic: str, *_, **__): |
| 29 | """Standalone .mcap dataloader withouth any ROS distribution.""" |
| 30 | # Conditional imports to avoid injecting dependencies for non mcap users |
| 31 | try: |
| 32 | from mcap.reader import make_reader |
| 33 | from mcap_ros2.reader import read_ros2_messages |
| 34 | except ImportError as e: |
| 35 | print("[ERROR] mcap plugins not installed: 'pip install mcap-ros2-support'") |
| 36 | exit(1) |
| 37 | |
| 38 | from kiss_icp.tools.point_cloud2 import read_point_cloud |
| 39 | |
| 40 | # we expect `data_dir` param to be a path to the .mcap file, so rename for clarity |
| 41 | assert os.path.isfile(data_dir), "mcap dataloader expects an existing MCAP file" |
| 42 | self.sequence_id = os.path.basename(data_dir).split(".")[0] |
| 43 | self.mcap_file = str(data_dir) |
| 44 | |
| 45 | self.make_reader = make_reader |
| 46 | self.read_ros2_messages = read_ros2_messages |
| 47 | self.read_point_cloud = read_point_cloud |
| 48 | |
| 49 | self.bag = self.make_reader(open(self.mcap_file, "rb")) |
| 50 | self.summary = self.bag.get_summary() |
| 51 | self.topic = self.check_topic(topic) |
| 52 | self.n_scans = self._get_n_scans() |
| 53 | self.msgs = self.read_ros2_messages(self.mcap_file, topics=[self.topic]) |
| 54 | self.timestamps = [] |
| 55 | self.use_global_visualizer = True |
| 56 | |
| 57 | def __del__(self): |
| 58 | if hasattr(self, "bag"): |
nothing calls this directly
no test coverage detected