Add protocol property to BusABC to determine active CAN Protocol (#1532)

* Implement is_fd property for BusABC and PCANBus

* Implement enum to represent CAN protocol

* Implement CANProtocol for VirtualBus

* Implement CANProtocol for UDPMulticastBus

* Implement CANProtocol for the CANalystIIBus

* Implement CANProtocol for the slcanBus

* Rename CANProtocol to CanProtocol

* Reimplement PcanBus.fd attribute as read-only property

The property is scheduled for removal in v5.0

* Reimplement UdpMulticastBus.is_fd attribute as read-only property

The property is superseded by BusABC.protocol and scheduled for removal
in version 5.0.

* Implement CanProtocol for robotellBus

* Implement CanProtocol for NicanBus

* Implement CanProtocol for IscanBus

* Implement CanProtocol for CantactBus

* Fix sphinx reference to CanProtocol

* Implement CanProtocol for GsUsbBus

* Implement CanProtocol for NiXNETcanBus

* Implement CanProtocol for EtasBus

* Implement CanProtocol for IXXATBus

* Implement CanProtocol for KvaserBus

* Implement CanProtocol for the SerialBus

* Implement CanProtocol for UcanBus

* Implement CanProtocol for VectorBus

* Implement CanProtocol for NeousysBus

* Implement CanProtocol for Usb2canBus

* Implement CanProtocol for NeoViBus

* Implement CanProtocol for SocketcanBus

* Permit passthrough of protocol field for SocketCanDaemonBus

* Implement CanProtocol for SeeedBus

* Remove CanProtocol attribute from BusABC constructor

The attribute is now set as class attribute with default value
and can be overridden in the subclass constructor.

* Apply suggestions from code review

Fix property access and enum comparison

Co-authored-by: zariiii9003 <[email protected]>

* Fix syntax error

* Fix more enum comparisons against BusABC.protocol

---------

Co-authored-by: zariiii9003 <[email protected]>

Author: lumagi

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: