Add PcapngReader

Author: dsseng

can/__init__.py

@@ -44,6 +44,7 @@
     "MessageSync",
     "ModifiableCyclicTaskABC",
     "Notifier",
+    "PcapngReader",
     "PcapngWriter",
     "Printer",
     "RedirectReader",
@@ -112,6 +113,7 @@
     MessageSync,
     MF4Reader,
     MF4Writer,
+    PcapngReader,
     PcapngWriter,
     Printer,
     SizedRotatingLogger,

can/io/__init__.py

@@ -20,6 +20,7 @@
     "MF4Reader",
     "MF4Writer",
     "MessageSync",
+    "PcapngReader",
     "PcapngWriter",
     "Printer",
     "SizedRotatingLogger",
@@ -53,7 +54,7 @@
 from .canutils import CanutilsLogReader, CanutilsLogWriter
 from .csv import CSVReader, CSVWriter
 from .mf4 import MF4Reader, MF4Writer
-from .pcapng import PcapngWriter
+from .pcapng import PcapngReader, PcapngWriter
 from .printer import Printer
 from .sqlite import SqliteReader, SqliteWriter
 from .trc import TRCFileVersion, TRCReader, TRCWriter

can/io/pcapng.py

@@ -6,12 +6,16 @@
 """
 
 import logging
-from typing import Any, BinaryIO, Dict, Optional, Union
+from typing import Any, BinaryIO, Dict, Generator, Optional, Union
 
 from ..message import Message
-from ..socketcan_common import CAN_FRAME_HEADER_STRUCT_BE, build_can_frame
+from ..socketcan_common import (
+    CAN_FRAME_HEADER_STRUCT_BE,
+    build_can_frame,
+    parse_can_frame,
+)
 from ..typechecking import Channel, StringPathLike
-from .generic import BinaryIOMessageWriter
+from .generic import BinaryIOMessageReader, BinaryIOMessageWriter
 
 logger = logging.getLogger("can.io.pcapng")
 
@@ -108,3 +112,56 @@ def on_message_received(self, msg: Message) -> None:
                 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.
+        """
+
+        if pcapng is None:
+            raise NotImplementedError(
+                "The python-pcapng package was not found. Install python-can with "
+                "the optional dependency [pcapng] to use the PcapngReader."
+            )
+
+        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, structure=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

@@ -27,6 +27,7 @@
 from .csv import CSVReader
 from .generic import BinaryIOMessageReader, MessageReader
 from .mf4 import MF4Reader
+from .pcapng import PcapngReader
 from .sqlite import SqliteReader
 from .trc import TRCReader
 
@@ -39,6 +40,7 @@
     ".db": SqliteReader,
     ".log": CanutilsLogReader,
     ".mf4": MF4Reader,
+    ".pcapng": PcapngReader,
     ".trc": TRCReader,
 }