| 25 | |
| 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"): |
| 59 | del self.bag |
| 60 | |
| 61 | def __getitem__(self, idx): |
| 62 | msg = next(self.msgs).ros_msg |
| 63 | self.timestamps.append(self.stamp_to_sec(msg.header.stamp)) |
| 64 | return self.read_point_cloud(msg) |
| 65 | |
| 66 | def __len__(self): |
| 67 | return self.n_scans |
| 68 | |
| 69 | def _get_n_scans(self) -> int: |
| 70 | return sum( |
| 71 | count |
| 72 | for (id, count) in self.summary.statistics.channel_message_counts.items() |
| 73 | if self.summary.channels[id].topic == self.topic |
| 74 | ) |
| 75 | |
| 76 | def reset(self): |
| 77 | self.timestamps = [] |
| 78 | self.bag = self.make_reader(open(self.mcap_file, "rb")) |
| 79 | self.msgs = self.read_ros2_messages(self.mcap_file, topics=[self.topic]) |
| 80 | |
| 81 | @staticmethod |
| 82 | def stamp_to_sec(stamp): |
| 83 | return stamp.sec + float(stamp.nanosec) / 1e9 |
| 84 |
nothing calls this directly
no outgoing calls
no test coverage detected