| 86 | return self.timestamps |
| 87 | |
| 88 | def check_topic(self, topic: str) -> str: |
| 89 | # Extract schema id from the .mcap file that encodes the PointCloud2 msg |
| 90 | schema_id = [ |
| 91 | schema.id |
| 92 | for schema in self.summary.schemas.values() |
| 93 | if schema.name == "sensor_msgs/msg/PointCloud2" |
| 94 | ][0] |
| 95 | |
| 96 | point_cloud_topics = [ |
| 97 | channel.topic |
| 98 | for channel in self.summary.channels.values() |
| 99 | if channel.schema_id == schema_id |
| 100 | ] |
| 101 | |
| 102 | def print_available_topics_and_exit(): |
| 103 | print(50 * "-") |
| 104 | for t in point_cloud_topics: |
| 105 | print(f"--topic {t}") |
| 106 | print(50 * "-") |
| 107 | sys.exit(1) |
| 108 | |
| 109 | if topic and topic in point_cloud_topics: |
| 110 | return topic |
| 111 | # when user specified the topic check that exists |
| 112 | if topic and topic not in point_cloud_topics: |
| 113 | print( |
| 114 | f'[ERROR] Dataset does not containg any msg with the topic name "{topic}". ' |
| 115 | "Please select one of the following topics with the --topic flag" |
| 116 | ) |
| 117 | print_available_topics_and_exit() |
| 118 | if len(point_cloud_topics) > 1: |
| 119 | print( |
| 120 | "[ERROR] Multiple sensor_msgs/msg/PointCloud2 topics available." |
| 121 | "Please select one of the following topics with the --topic flag" |
| 122 | ) |
| 123 | print_available_topics_and_exit() |
| 124 | |
| 125 | if len(point_cloud_topics) == 0: |
| 126 | print("[ERROR] Your dataset does not contain any sensor_msgs/msg/PointCloud2 topic") |
| 127 | if len(point_cloud_topics) == 1: |
| 128 | return point_cloud_topics[0] |