Add PcapngWriter

Author: dsseng

can/__init__.py

@@ -44,6 +44,7 @@
     "MessageSync",
     "ModifiableCyclicTaskABC",
     "Notifier",
+    "PcapngWriter",
     "Printer",
     "RedirectReader",
     "RestartableCyclicTaskABC",
@@ -111,6 +112,7 @@
     MessageSync,
     MF4Reader,
     MF4Writer,
+    PcapngWriter,
     Printer,
     SizedRotatingLogger,
     SqliteReader,

can/io/__init__.py

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

can/io/logger.py

@@ -37,6 +37,7 @@
     MessageWriter,
 )
 from .mf4 import MF4Writer
+from .pcapng import PcapngWriter
 from .printer import Printer
 from .sqlite import SqliteWriter
 from .trc import TRCWriter
@@ -50,6 +51,7 @@
     ".db": SqliteWriter,
     ".log": CanutilsLogWriter,
     ".mf4": MF4Writer,
+    ".pcapng": PcapngWriter,
     ".trc": TRCWriter,
     ".txt": Printer,
 }

can/io/pcapng.py

@@ -0,0 +1,110 @@
+"""
+Contains handling of pcapng logging files.
+
+pcapng file is a binary file format used for packet capture files.
+Spec: https://www.ietf.org/archive/id/draft-tuexen-opsawg-pcapng-03.html
+"""
+
+import logging
+from typing import Any, BinaryIO, Dict, Optional, Union
+
+from ..message import Message
+from ..socketcan_common import CAN_FRAME_HEADER_STRUCT_BE, build_can_frame
+from ..typechecking import Channel, StringPathLike
+from .generic import BinaryIOMessageWriter
+
+logger = logging.getLogger("can.io.pcapng")
+
+try:
+    import pcapng
+    from pcapng import blocks
+except ImportError:
+    pcapng = None
+
+
+# https://www.tcpdump.org/linktypes.html
+# https://www.tcpdump.org/linktypes/LINKTYPE_CAN_SOCKETCAN.html
+LINKTYPE_CAN_SOCKETCAN = 227
+
+
+class PcapngWriter(BinaryIOMessageWriter):
+    """
+    Logs CAN data to an pcapng file supported by Wireshark and other tools.
+    """
+
+    def __init__(
+        self,
+        file: Union[StringPathLike, BinaryIO],
+        append: bool = False,
+        tsresol: int = 9,
+        **kwargs: Any,
+    ) -> None:
+        """
+        :param file:
+            A path-like object or as file-like object to write to.
+            If this is a file-like object, is has to be opened in
+            binary write mode, not text write mode.
+
+        :param append:
+            If True, the file will be opened in append mode. Otherwise,
+            it will be opened in write mode. The default is False.
+
+        :param tsresol:
+            The time resolution of the timestamps in the pcapng file,
+            expressed as -log10(unit in seconds),
+            e.g. 9 for nanoseconds, 6 for microseconds.
+            The default is 9, which corresponds to nanoseconds.
+            .
+        """
+        if pcapng is None:
+            raise NotImplementedError(
+                "The python-pcapng package was not found. Install python-can with "
+                "the optional dependency [pcapng] to use the PcapngWriter."
+            )
+
+        mode = "wb+"
+        if append:
+            mode = "ab+"
+
+        # pcapng supports concatenation, and thus append
+        super().__init__(file, mode=mode)
+        self._header_block = blocks.SectionHeader(endianness=">")
+        self._writer = pcapng.FileWriter(self.file, self._header_block)
+        self._idbs: Dict[Channel, blocks.InterfaceDescription] = {}
+        self.tsresol = tsresol
+
+    def _resolve_idb(self, channel: Optional[Channel]) -> Any:
+        channel_name = str(channel)
+        if channel is None:
+            channel_name = "can0"
+
+        if channel_name not in self._idbs:
+            idb = blocks.InterfaceDescription(
+                section=self._header_block.section,
+                link_type=LINKTYPE_CAN_SOCKETCAN,
+                options={
+                    "if_name": channel_name,
+                    "if_tsresol": bytes([self.tsresol]),  # nanoseconds
+                },
+                endianness=">",  # big
+            )
+            self._header_block.register_interface(idb)
+            self._writer.write_block(idb)
+            self._idbs[channel_name] = idb
+
+        return self._idbs[channel_name]
+
+    def on_message_received(self, msg: Message) -> None:
+        idb: blocks.InterfaceDescription = self._resolve_idb(msg.channel)
+        timestamp_units = int(msg.timestamp * 10**self.tsresol)
+        self._writer.write_block(
+            blocks.EnhancedPacket(
+                self._header_block.section,
+                interface_id=idb.interface_id,
+                packet_data=build_can_frame(msg, structure=CAN_FRAME_HEADER_STRUCT_BE),
+                # timestamp (in tsresol units) = timestamp_high << 32 + timestamp_low
+                timestamp_high=timestamp_units >> 32,
+                timestamp_low=timestamp_units & 0xFFFFFFFF,
+                endianness=">",  # big
+            )
+        )

can/socketcan_common.py

@@ -87,19 +87,25 @@
 # The 32bit can id is directly followed by the 8bit data link count
 # The data field is aligned on an 8 byte boundary, hence we add padding
 # which aligns the data field to an 8 byte boundary.
+
+# host-endian for communication with kernel
 CAN_FRAME_HEADER_STRUCT = struct.Struct("=IBB1xB")
+# big-endian for pcapng
+CAN_FRAME_HEADER_STRUCT_BE = struct.Struct(">IBB1xB")
 
 
 log = logging.getLogger(__name__)
 
 
-def parse_can_frame(cf: bytes) -> Message:
+def parse_can_frame(
+    cf: bytes, structure: struct.Struct = CAN_FRAME_HEADER_STRUCT
+) -> Message:
     """Parse a CAN frame.
 
     :param cf: A CAN frame in socketcan format
     :return: A :class:`~can.Message` object with the parsed data
     """
-    can_id, can_dlc, flags, data = dissect_can_frame(cf)
+    can_id, can_dlc, flags, data = dissect_can_frame(cf, structure)
 
     # EXT, RTR, ERR flags -> boolean attributes
     #   /* special address description flags for the CAN_ID */
@@ -148,7 +154,9 @@ def _compose_arbitration_id(message: Message) -> int:
     return can_id
 
 
-def build_can_frame(msg: Message) -> bytes:
+def build_can_frame(
+    msg: Message, structure: struct.Struct = CAN_FRAME_HEADER_STRUCT
+) -> bytes:
     """CAN frame packing (see 'struct can_frame' in <linux/can.h>)
 
     :param msg: A :class:`~can.Message` object to convert to a CAN frame
@@ -178,7 +186,7 @@ def build_can_frame(msg: Message) -> bytes:
         data_len = msg.dlc
     else:
         data_len = min(i for i in can.util.CAN_FD_DLC if i >= len(msg.data))
-    header = CAN_FRAME_HEADER_STRUCT.pack(can_id, data_len, flags, msg.dlc)
+    header = structure.pack(can_id, data_len, flags, msg.dlc)
     return header + data
 
 
@@ -188,14 +196,16 @@ def is_frame_fd(frame: bytes) -> bool:
     return len(frame) == CANFD_MTU
 
 
-def dissect_can_frame(frame: bytes) -> Tuple[int, int, int, bytes]:
+def dissect_can_frame(
+    frame: bytes, structure: struct.Struct = CAN_FRAME_HEADER_STRUCT
+) -> Tuple[int, int, int, bytes]:
     """Dissect a CAN frame into its components.
 
     :param frame: A CAN frame in socketcan format
     :return: Tuple of (CAN ID, CAN DLC, flags, data)
     """
 
-    can_id, data_len, flags, len8_dlc = CAN_FRAME_HEADER_STRUCT.unpack_from(frame)
+    can_id, data_len, flags, len8_dlc = structure.unpack_from(frame)
 
     if data_len not in can.util.CAN_FD_DLC:
         data_len = min(i for i in can.util.CAN_FD_DLC if i >= data_len)

pyproject.toml

@@ -83,6 +83,7 @@ viewer = [
     "windows-curses; platform_system == 'Windows' and platform_python_implementation=='CPython'"
 ]
 mf4 = ["asammdf>=6.0.0"]
+pcapng = ["python-pcapng>=2.1.1"]
 
 [tool.setuptools.dynamic]
 readme = { file = "README.rst" }