Add PcapngReader

Author: dsseng

can/io/pcapng.py

@@ -8,12 +8,12 @@
 
 import logging
 
-from typing import Any, BinaryIO, Dict, Union
+from typing import Any, BinaryIO, Dict, Generator, Union
 
 from ..message import Message
 from ..typechecking import StringPathLike, Channel
-from .generic import BinaryIOMessageWriter
-from ..socketcan_common import build_can_frame, CAN_FRAME_HEADER_STRUCT_BE
+from .generic import BinaryIOMessageWriter, BinaryIOMessageReader
+from ..socketcan_common import build_can_frame, parse_can_frame, CAN_FRAME_HEADER_STRUCT_BE
 
 logger = logging.getLogger("can.io.pcapng")
 
@@ -106,3 +106,44 @@ def on_message_received(self, msg: Message) -> None:
             timestamp_low=timestamp_units & 0xFFFFFFFF,
             endianness=">" # big
         ))
+
+class PcapngReader(BinaryIOMessageReader):
+    """
+    Iterator of CAN messages from a Pcapng File.
+    """
+
+    file: BinaryIO
+
+    def __init__(
+        self,
+        file: Union[StringPathLike, BinaryIO],
+        **kwargs: Any,
+    ) -> None:
+        """
+        :param file: a path-like object or as file-like object to read from
+                     If this is a file-like object, is has to opened in binary
+                     read mode, not text read mode.
+        """
+        super().__init__(file, mode="rb")
+        self._scanner = pcapng.FileScanner(self.file)
+
+    def __iter__(self) -> Generator[Message, None, None]:
+        for block in self._scanner:
+            if isinstance(block, blocks.EnhancedPacket):
+                idn: blocks.InterfaceDescription = block.interface
+                # We only care about the CAN packets
+                if idn.link_type != LINKTYPE_CAN_SOCKETCAN:
+                    logger.debug("Skipping non-CAN packet, link type: %s", idn.link_type)
+                    continue
+
+                msg = parse_can_frame(block.packet_data, struct=CAN_FRAME_HEADER_STRUCT_BE)
+
+                timestamp64 = (block.timestamp_high << 32) + block.timestamp_low
+                msg.timestamp = timestamp64 * idn.timestamp_resolution
+
+                if 'if_name' in idn.options:
+                    msg.channel = idn.options['if_name']
+                else:
+                    msg.channel = block.interface_id
+
+                yield msg

can/io/player.py

@@ -29,6 +29,7 @@
 from .mf4 import MF4Reader
 from .sqlite import SqliteReader
 from .trc import TRCReader
+from .pcapng import PcapngReader
 
 #: A map of file suffixes to their corresponding
 #: :class:`can.io.generic.MessageReader` class
@@ -39,6 +40,7 @@
     ".db": SqliteReader,
     ".log": CanutilsLogReader,
     ".mf4": MF4Reader,
+    ".pcapng": PcapngReader,
     ".trc": TRCReader,
 }