Implement is_fd property for BusABC and PCANBus

Author: lumagi

can/bus.py

@@ -48,6 +48,7 @@ class BusABC(metaclass=ABCMeta):
     def __init__(
         self,
         channel: Any,
+        is_fd: bool = False,
         can_filters: Optional[can.typechecking.CanFilters] = None,
         **kwargs: object
     ):
@@ -59,6 +60,9 @@ def __init__(
         :param channel:
             The can interface identifier. Expected type is backend dependent.
 
+        :param is_fd:
+            Indicates that this bus supports CAN-FD.
+
         :param can_filters:
             See :meth:`~can.BusABC.set_filters` for details.
 
@@ -71,6 +75,7 @@ def __init__(
         :raises ~can.exceptions.CanInitializationError:
             If the bus cannot be initialized
         """
+        self._is_fd = is_fd
         self._periodic_tasks: List[_SelfRemovingCyclicTask] = []
         self.set_filters(can_filters)
 
@@ -442,6 +447,10 @@ def state(self, new_state: BusState) -> None:
         """
         raise NotImplementedError("Property is not implemented.")
 
+    @property
+    def is_fd(self) -> bool:
+        return self._is_fd
+
     @staticmethod
     def _detect_available_configs() -> List[can.typechecking.AutoDetectedConfig]:
         """Detect all configurations/channels that this interface could

can/interfaces/pcan/pcan.py

@@ -247,7 +247,7 @@ def __init__(
                 raise ValueError(err_msg)
 
         self.channel_info = str(channel)
-        self.fd = isinstance(timing, BitTimingFd) if timing else kwargs.get("fd", False)
+        is_fd = isinstance(timing, BitTimingFd) if timing else kwargs.get("fd", False)
 
         hwtype = PCAN_TYPE_ISA
         ioport = 0x02A0
@@ -271,7 +271,7 @@ def __init__(
             result = self.m_objPCANBasic.Initialize(
                 self.m_PcanHandle, pcan_bitrate, hwtype, ioport, interrupt
             )
-        elif self.fd:
+        elif is_fd:
             if isinstance(timing, BitTimingFd):
                 timing = check_or_adjust_timing_clock(
                     timing, sorted(VALID_PCAN_FD_CLOCKS, reverse=True)
@@ -338,7 +338,9 @@ def __init__(
             if result != PCAN_ERROR_OK:
                 raise PcanCanInitializationError(self._get_formatted_error(result))
 
-        super().__init__(channel=channel, state=state, bitrate=bitrate, **kwargs)
+        super().__init__(
+            channel=channel, is_fd=is_fd, state=state, bitrate=bitrate, **kwargs
+        )
 
     def _find_channel_by_dev_id(self, device_id):
         """
@@ -484,7 +486,7 @@ def _recv_internal(
         end_time = time.time() + timeout if timeout is not None else None
 
         while True:
-            if self.fd:
+            if self.is_fd:
                 result, pcan_msg, pcan_timestamp = self.m_objPCANBasic.ReadFD(
                     self.m_PcanHandle
                 )
@@ -546,7 +548,7 @@ def _recv_internal(
         error_state_indicator = bool(pcan_msg.MSGTYPE & PCAN_MESSAGE_ESI.value)
         is_error_frame = bool(pcan_msg.MSGTYPE & PCAN_MESSAGE_ERRFRAME.value)
 
-        if self.fd:
+        if self.is_fd:
             dlc = dlc2len(pcan_msg.DLC)
             timestamp = boottimeEpoch + (pcan_timestamp.value / (1000.0 * 1000.0))
         else:
@@ -592,7 +594,7 @@ def send(self, msg, timeout=None):
         if msg.error_state_indicator:
             msgType |= PCAN_MESSAGE_ESI.value
 
-        if self.fd:
+        if self.is_fd:
             # create a TPCANMsg message structure
             CANMsg = TPCANMsgFD()
 

test/test_pcan.py

@@ -53,8 +53,9 @@ def test_bus_creation(self) -> None:
         self.bus = can.Bus(interface="pcan")
 
         self.assertIsInstance(self.bus, PcanBus)
-        self.MockPCANBasic.assert_called_once()
+        self.assertFalse(self.bus.is_fd)
 
+        self.MockPCANBasic.assert_called_once()
         self.mock_pcan.Initialize.assert_called_once()
         self.mock_pcan.InitializeFD.assert_not_called()
 
@@ -80,6 +81,8 @@ def test_bus_creation_fd(self, clock_param: str, clock_val: int) -> None:
         )
 
         self.assertIsInstance(self.bus, PcanBus)
+        self.assertTrue(self.bus.is_fd)
+
         self.MockPCANBasic.assert_called_once()
         self.mock_pcan.Initialize.assert_not_called()
         self.mock_pcan.InitializeFD.assert_called_once()
@@ -452,10 +455,11 @@ def test_peak_fd_bus_constructor_regression(self):
 
     def test_constructor_bit_timing(self):
         timing = can.BitTiming.from_registers(f_clock=8_000_000, btr0=0x47, btr1=0x2F)
-        can.Bus(interface="pcan", channel="PCAN_USBBUS1", timing=timing)
+        bus = can.Bus(interface="pcan", channel="PCAN_USBBUS1", timing=timing)
 
         bitrate_arg = self.mock_pcan.Initialize.call_args[0][1]
         self.assertEqual(bitrate_arg.value, 0x472F)
+        self.assertFalse(bus.is_fd)
 
     def test_constructor_bit_timing_fd(self):
         timing = can.BitTimingFd(
@@ -469,7 +473,8 @@ def test_constructor_bit_timing_fd(self):
             data_tseg2=6,
             data_sjw=1,
         )
-        can.Bus(interface="pcan", channel="PCAN_USBBUS1", timing=timing)
+        bus = can.Bus(interface="pcan", channel="PCAN_USBBUS1", timing=timing)
+        self.assertTrue(bus.is_fd)
 
         bitrate_arg = self.mock_pcan.InitializeFD.call_args[0][-1]