feat: Add Channel Methods (#126)

Author: siliconlad

src/pybag/bag_reader.py

@@ -152,6 +152,17 @@ def get_topics(self) -> list[str]:
         """
         return list(set(c.topic for c in self._connections.values()))
 
+    def get_connections(self) -> list[ConnectionRecord]:
+        """Get all connection records in the bag file.
+
+        Connection records contain metadata about each topic including
+        the message type and message definition.
+
+        Returns:
+            List of ConnectionRecord objects.
+        """
+        return list(self._connections.values())
+
     def get_message_count(self, topic: str) -> int:
         """Get the number of messages for a topic.
 

src/pybag/cli/mcap_convert.py

@@ -161,10 +161,9 @@ def convert_bag_to_mcap(
             chunk_size=chunk_size,
             chunk_compression=chunk_compression,
         ) as writer:
-            # TODO: Make this nicer
             # Pre-register all channels with translated schemas
             # This ensures the schema is correct for the target format
-            for conn in reader._connections.values():
+            for conn in reader.get_connections():
                 conn_header = conn.connection_header
                 schema = translate_schema_ros1_to_ros2(
                     conn_header.type,
@@ -233,12 +232,10 @@ def convert_mcap_to_bag(
             compression=chunk_compression,
             chunk_size=chunk_size,
         ) as writer:
-            # TODO: Make this nicer
             # Pre-register all connections with translated schemas
             # This ensures the schema is correct for the target format
-            channels = reader._reader.get_channels()
-            for channel in channels.values():
-                schema_record = reader._reader.get_channel_schema(channel.id)
+            for channel in reader.get_channels():
+                schema_record = reader.get_schema(channel.topic)
                 if schema_record is None:
                     logger.warning(f"No schema found for channel {channel.topic}")
                     continue

src/pybag/mcap_reader.py

@@ -13,7 +13,12 @@
     BaseMcapRecordReader,
     McapRecordReaderFactory
 )
-from pybag.mcap.records import AttachmentRecord, MetadataRecord
+from pybag.mcap.records import (
+    AttachmentRecord,
+    ChannelRecord,
+    MetadataRecord,
+    SchemaRecord
+)
 
 logger = logging.getLogger(__name__)
 
@@ -71,6 +76,28 @@ def get_topics(self) -> list[str]:
         """Get all topics in the MCAP file."""
         return [c.topic for c in self._reader.get_channels().values()] # TODO: Use a set?
 
+    def get_channels(self) -> list[ChannelRecord]:
+        """Get all channels in the MCAP file.
+
+        Returns:
+            List of ChannelRecord objects.
+        """
+        return list(self._reader.get_channels().values())
+
+    def get_schema(self, topic: str) -> SchemaRecord | None:
+        """Get the schema for a particular topic.
+
+        Args:
+            topic: The topic name to get the schema for.
+
+        Returns:
+            SchemaRecord for the topic, or None if not found.
+        """
+        channel_id = self._reader.get_channel_id(topic)
+        if channel_id is None:
+            return None
+        return self._reader.get_channel_schema(channel_id)
+
     def get_message_count(self, topic: str) -> int:
         """Get the number of messages in a given topic."""
         channel_id = self._reader.get_channel_id(topic)