Add protocol property to BusABC to determine active CAN Protocol

can/__init__.py

@@ -25,6 +25,7 @@
     "CanInitializationError",
     "CanInterfaceNotImplementedError",
     "CanOperationError",
+    "CanProtocol",
     "CanTimeoutError",
     "CanutilsLogReader",
     "CanutilsLogWriter",
@@ -88,7 +89,7 @@
     ModifiableCyclicTaskABC,
     RestartableCyclicTaskABC,
 )
-from .bus import BusABC, BusState
+from .bus import BusABC, BusState, CanProtocol
 from .exceptions import (
     CanError,
     CanInitializationError,

can/bus.py

@@ -26,6 +26,14 @@ class BusState(Enum):
     ERROR = auto()
 
 
+class CanProtocol(Enum):
+    """The CAN protocol type supported by a :class:`can.BusABC` instance"""
+
+    CAN_20 = auto()
+    CAN_FD = auto()
+    CAN_XL = auto()
+
+
 class BusABC(metaclass=ABCMeta):
     """The CAN Bus Abstract Base Class that serves as the basis
     for all concrete interfaces.
@@ -44,6 +52,7 @@ class BusABC(metaclass=ABCMeta):
     RECV_LOGGING_LEVEL = 9
 
     _is_shutdown: bool = False
+    _can_protocol: CanProtocol = CanProtocol.CAN_20
 
     @abstractmethod
     def __init__(
@@ -459,6 +468,15 @@ def state(self, new_state: BusState) -> None:
         """
         raise NotImplementedError("Property is not implemented.")
 
+    @property
+    def protocol(self) -> CanProtocol:
+        """Return the CAN protocol used by this bus instance.
+
+        This value is set at initialization time and does not change
+        during the lifetime of a bus instance.
+        """
+        return self._can_protocol
+
     @staticmethod
     def _detect_available_configs() -> List[can.typechecking.AutoDetectedConfig]:
         """Detect all configurations/channels that this interface could

can/interfaces/canalystii.py

@@ -6,7 +6,7 @@
 
 import canalystii as driver
 
-from can import BitTiming, BitTimingFd, BusABC, Message
+from can import BitTiming, BitTimingFd, BusABC, CanProtocol, Message
 from can.exceptions import CanTimeoutError
 from can.typechecking import CanFilters
 from can.util import check_or_adjust_timing_clock, deprecated_args_alias
@@ -54,8 +54,11 @@ def __init__(
             raise ValueError("Either bitrate or timing argument is required")
 
         # Do this after the error handling
-        super().__init__(channel=channel, can_filters=can_filters, **kwargs)
-
+        super().__init__(
+            channel=channel,
+            can_filters=can_filters,
+            **kwargs,
+        )
         if isinstance(channel, str):
             # Assume comma separated string of channels
             self.channels = [int(ch.strip()) for ch in channel.split(",")]
@@ -64,11 +67,11 @@ def __init__(
         else:  # Sequence[int]
             self.channels = list(channel)
 
-        self.rx_queue: Deque[Tuple[int, driver.Message]] = deque(maxlen=rx_queue_size)
-
         self.channel_info = f"CANalyst-II: device {device}, channels {self.channels}"
-
+        self.rx_queue: Deque[Tuple[int, driver.Message]] = deque(maxlen=rx_queue_size)
         self.device = driver.CanalystDevice(device_index=device)
+        self._can_protocol = CanProtocol.CAN_20
+
         for single_channel in self.channels:
             if isinstance(timing, BitTiming):
                 timing = check_or_adjust_timing_clock(timing, valid_clocks=[8_000_000])

can/interfaces/cantact.py

@@ -7,7 +7,7 @@
 from typing import Any, Optional, Union
 from unittest.mock import Mock
 
-from can import BitTiming, BitTimingFd, BusABC, Message
+from can import BitTiming, BitTimingFd, BusABC, CanProtocol, Message
 
 from ..exceptions import (
     CanInitializationError,
@@ -87,6 +87,7 @@ def __init__(
 
         self.channel = int(channel)
         self.channel_info = f"CANtact: ch:{channel}"
+        self._can_protocol = CanProtocol.CAN_20
 
         # Configure the interface
         with error_check("Cannot setup the cantact.Interface", CanInitializationError):
@@ -114,7 +115,10 @@ def __init__(
             self.interface.start()
 
         super().__init__(
-            channel=channel, bitrate=bitrate, poll_interval=poll_interval, **kwargs
+            channel=channel,
+            bitrate=bitrate,
+            poll_interval=poll_interval,
+            **kwargs,
         )
 
     def _recv_internal(self, timeout):

can/interfaces/etas/__init__.py

@@ -1,4 +1,3 @@
-import ctypes
 import time
 from typing import Dict, List, Optional, Tuple
 
@@ -17,9 +16,12 @@ def __init__(
         bitrate: int = 1000000,
         fd: bool = True,
         data_bitrate: int = 2000000,
-        **kwargs: object,
+        **kwargs: Dict[str, any],
     ):
+        super().__init__(channel=channel, **kwargs)
+
         self.receive_own_messages = receive_own_messages
+        self._can_protocol = can.CanProtocol.CAN_FD if fd else can.CanProtocol.CAN_20
 
         nodeRange = CSI_NodeRange(CSI_NODE_MIN, CSI_NODE_MAX)
         self.tree = ctypes.POINTER(CSI_Tree)()
@@ -297,7 +299,7 @@ def _detect_available_configs() -> List[can.typechecking.AutoDetectedConfig]:
         tree = ctypes.POINTER(CSI_Tree)()
         CSI_CreateProtocolTree(ctypes.c_char_p(b""), nodeRange, ctypes.byref(tree))
 
-        nodes: Dict[str, str] = []
+        nodes: List[Dict[str, str]] = []
 
         def _findNodes(tree, prefix):
             uri = f"{prefix}/{tree.contents.item.uriName.decode()}"

can/interfaces/gs_usb.py

@@ -52,11 +52,16 @@ def __init__(
 
         self.gs_usb = gs_usb
         self.channel_info = channel
+        self._can_protocol = can.CanProtocol.CAN_20
 
         self.gs_usb.set_bitrate(bitrate)
         self.gs_usb.start()
 
-        super().__init__(channel=channel, can_filters=can_filters, **kwargs)
+        super().__init__(
+            channel=channel,
+            can_filters=can_filters,
+            **kwargs,
+        )
 
     def send(self, msg: can.Message, timeout: Optional[float] = None):
         """Transmit a message to the CAN bus.

can/interfaces/ics_neovi/neovi_bus.py

@@ -16,7 +16,7 @@
 from threading import Event
 from warnings import warn
 
-from can import BusABC, Message
+from can import BusABC, CanProtocol, Message
 
 from ...exceptions import (
     CanError,
@@ -169,7 +169,11 @@ def __init__(self, channel, can_filters=None, **kwargs):
         if ics is None:
             raise ImportError("Please install python-ics")
 
-        super().__init__(channel=channel, can_filters=can_filters, **kwargs)
+        super().__init__(
+            channel=channel,
+            can_filters=can_filters,
+            **kwargs,
+        )
 
         logger.info(f"CAN Filters: {can_filters}")
         logger.info(f"Got configuration of: {kwargs}")
@@ -190,6 +194,9 @@ def __init__(self, channel, can_filters=None, **kwargs):
         serial = kwargs.get("serial")
         self.dev = self._find_device(type_filter, serial)
 
+        is_fd = kwargs.get("fd", False)
+        self._can_protocol = CanProtocol.CAN_FD if is_fd else CanProtocol.CAN_20
+
         with open_lock:
             ics.open_device(self.dev)
 
@@ -198,7 +205,7 @@ def __init__(self, channel, can_filters=None, **kwargs):
                 for channel in self.channels:
                     ics.set_bit_rate(self.dev, kwargs.get("bitrate"), channel)
 
-            if kwargs.get("fd", False):
+            if is_fd:
                 if "data_bitrate" in kwargs:
                     for channel in self.channels:
                         ics.set_fd_bit_rate(

can/interfaces/iscan.py

@@ -13,6 +13,7 @@
     CanInitializationError,
     CanInterfaceNotImplementedError,
     CanOperationError,
+    CanProtocol,
     Message,
 )
 
@@ -99,6 +100,7 @@ def __init__(
 
         self.channel = ctypes.c_ubyte(int(channel))
         self.channel_info = f"IS-CAN: {self.channel}"
+        self._can_protocol = CanProtocol.CAN_20
 
         if bitrate not in self.BAUDRATES:
             raise ValueError(f"Invalid bitrate, choose one of {set(self.BAUDRATES)}")
@@ -107,7 +109,10 @@ def __init__(
         iscan.isCAN_DeviceInitEx(self.channel, self.BAUDRATES[bitrate])
 
         super().__init__(
-            channel=channel, bitrate=bitrate, poll_interval=poll_interval, **kwargs
+            channel=channel,
+            bitrate=bitrate,
+            poll_interval=poll_interval,
+            **kwargs,
         )
 
     def _recv_internal(

can/interfaces/ixxat/canlib.py

@@ -131,6 +131,9 @@ def __init__(
                 **kwargs
             )
 
+        super().__init__(channel=channel, **kwargs)
+        self._can_protocol = self.bus.protocol
+
     def flush_tx_buffer(self):
         """Flushes the transmit buffer on the IXXAT"""
         return self.bus.flush_tx_buffer()

can/interfaces/ixxat/canlib_vcinpl.py

@@ -15,7 +15,7 @@
 import sys
 from typing import Callable, Optional, Tuple
 
-from can import BusABC, Message
+from can import BusABC, CanProtocol, Message
 from can.broadcastmanager import (
     LimitedDurationCyclicSendTaskABC,
     RestartableCyclicTaskABC,
@@ -490,6 +490,7 @@ def __init__(
         self._channel_capabilities = structures.CANCAPABILITIES()
         self._message = structures.CANMSG()
         self._payload = (ctypes.c_byte * 8)()
+        self._can_protocol = CanProtocol.CAN_20
 
         # Search for supplied device
         if unique_hardware_id is None:

can/interfaces/ixxat/canlib_vcinpl2.py

@@ -15,16 +15,15 @@
 import sys
 from typing import Callable, Optional, Tuple
 
-import can.util
-from can import BusABC, Message
+from can import BusABC, CanProtocol, Message
 from can.broadcastmanager import (
     LimitedDurationCyclicSendTaskABC,
     RestartableCyclicTaskABC,
 )
 from can.ctypesutil import HANDLE, PHANDLE, CLibrary
 from can.ctypesutil import HRESULT as ctypes_HRESULT
 from can.exceptions import CanInitializationError, CanInterfaceNotImplementedError
-from can.util import deprecated_args_alias
+from can.util import deprecated_args_alias, dlc2len, len2dlc
 
 from . import constants, structures
 from .exceptions import *
@@ -536,6 +535,7 @@ def __init__(
         self._channel_capabilities = structures.CANCAPABILITIES2()
         self._message = structures.CANMSG2()
         self._payload = (ctypes.c_byte * 64)()
+        self._can_protocol = CanProtocol.CAN_FD
 
         # Search for supplied device
         if unique_hardware_id is None:
@@ -865,7 +865,7 @@ def _recv_internal(self, timeout):
             # Timed out / can message type is not DATA
             return None, True
 
-        data_len = can.util.dlc2len(self._message.uMsgInfo.Bits.dlc)
+        data_len = dlc2len(self._message.uMsgInfo.Bits.dlc)
         # The _message.dwTime is a 32bit tick value and will overrun,
         # so expect to see the value restarting from 0
         rx_msg = Message(
@@ -915,7 +915,7 @@ def send(self, msg: Message, timeout: Optional[float] = None) -> None:
         message.uMsgInfo.Bits.edl = 1 if msg.is_fd else 0
         message.dwMsgId = msg.arbitration_id
         if msg.dlc:  # this dlc means number of bytes of payload
-            message.uMsgInfo.Bits.dlc = can.util.len2dlc(msg.dlc)
+            message.uMsgInfo.Bits.dlc = len2dlc(msg.dlc)
             data_len_dif = msg.dlc - len(msg.data)
             data = msg.data + bytearray(
                 [0] * data_len_dif

can/interfaces/kvaser/canlib.py

@@ -11,7 +11,7 @@
 import sys
 import time
 
-from can import BusABC, Message
+from can import BusABC, CanProtocol, Message
 from can.util import time_perfcounter_correlation
 
 from ...exceptions import CanError, CanInitializationError, CanOperationError
@@ -428,11 +428,12 @@ def __init__(self, channel, can_filters=None, **kwargs):
             channel = int(channel)
         except ValueError:
             raise ValueError("channel must be an integer")
-        self.channel = channel
 
-        log.debug("Initialising bus instance")
+        self.channel = channel
         self.single_handle = single_handle
+        self._can_protocol = CanProtocol.CAN_FD if fd else CanProtocol.CAN_20
 
+        log.debug("Initialising bus instance")
         num_channels = ctypes.c_int(0)
         canGetNumberOfChannels(ctypes.byref(num_channels))
         num_channels = int(num_channels.value)
@@ -520,7 +521,11 @@ def __init__(self, channel, can_filters=None, **kwargs):
             self._timestamp_offset = time.time() - (timer.value * TIMESTAMP_FACTOR)
 
         self._is_filtered = False
-        super().__init__(channel=channel, can_filters=can_filters, **kwargs)
+        super().__init__(
+            channel=channel,
+            can_filters=can_filters,
+            **kwargs,
+        )
 
     def _apply_filters(self, filters):
         if filters and len(filters) == 1:

can/interfaces/neousys/neousys.py

@@ -34,12 +34,13 @@
 except ImportError:
     from ctypes import CDLL
 
-from can import BusABC, Message
-
-from ...exceptions import (
+from can import (
+    BusABC,
     CanInitializationError,
     CanInterfaceNotImplementedError,
     CanOperationError,
+    CanProtocol,
+    Message,
 )
 
 logger = logging.getLogger(__name__)
@@ -150,6 +151,7 @@ def __init__(self, channel, device=0, bitrate=500000, **kwargs):
         self.channel = channel
         self.device = device
         self.channel_info = f"Neousys Can: device {self.device}, channel {self.channel}"
+        self._can_protocol = CanProtocol.CAN_20
 
         self.queue = queue.Queue()
 

can/interfaces/nican.py

@@ -19,13 +19,14 @@
 from typing import Optional, Tuple, Type
 
 import can.typechecking
-from can import BusABC, Message
-
-from ..exceptions import (
+from can import (
+    BusABC,
     CanError,
     CanInitializationError,
     CanInterfaceNotImplementedError,
     CanOperationError,
+    CanProtocol,
+    Message,
 )
 
 logger = logging.getLogger(__name__)
@@ -219,6 +220,7 @@ def __init__(
 
         self.channel = channel
         self.channel_info = f"NI-CAN: {channel}"
+        self._can_protocol = CanProtocol.CAN_20
         channel_bytes = channel.encode("ascii")
 
         config = [(NC_ATTR_START_ON_OPEN, True), (NC_ATTR_LOG_COMM_ERRS, log_errors)]