add BitTiming parameter to KvaserBus

can/interfaces/kvaser/canlib.py

@@ -10,11 +10,19 @@
 import time
 import logging
 import ctypes
-
-from can import BusABC
-from ...exceptions import CanError, CanInitializationError, CanOperationError
-from can import Message
-from can.util import time_perfcounter_correlation
+from typing import Optional, Union
+
+from can import (
+    BusABC,
+    BitTiming,
+    BitTimingFd,
+    Message,
+    CanError,
+    CanInitializationError,
+    CanOperationError,
+)
+from can.util import time_perfcounter_correlation, check_or_adjust_timing_clock
+from can.typechecking import CanFilters
 from . import constants as canstat
 from . import structures
 
@@ -199,6 +207,17 @@ def __check_bus_handle_validity(handle, function, arguments):
         errcheck=__check_status_initialization,
     )
 
+    canSetBusParamsC200 = __get_canlib_function(
+        "canSetBusParamsC200",
+        argtypes=[
+            c_canHandle,
+            ctypes.c_byte,
+            ctypes.c_byte,
+        ],
+        restype=canstat.c_canStatus,
+        errcheck=__check_status_initialization,
+    )
+
     canSetBusParamsFd = __get_canlib_function(
         "canSetBusParamsFd",
         argtypes=[
@@ -360,7 +379,13 @@ class KvaserBus(BusABC):
     The CAN Bus implemented for the Kvaser interface.
     """
 
-    def __init__(self, channel, can_filters=None, **kwargs):
+    def __init__(
+        self,
+        channel: int,
+        can_filters: Optional[CanFilters] = None,
+        timing: Optional[Union[BitTiming, BitTimingFd]] = None,
+        **kwargs,
+    ):
         """
         :param int channel:
             The Channel id to create this bus with.
@@ -370,6 +395,12 @@ def __init__(self, channel, can_filters=None, **kwargs):
 
         Backend Configuration
 
+        :param timing:
+            An instance of :class:`~can.BitTiming` or :class:`~can.BitTimingFd`
+            to specify the bit timing parameters for the Kvaser interface. If provided, it
+            takes precedence over the all other timing-related parameters.
+            Note that the `f_clock` property of the `timing` instance must be 16_000_000 (16MHz)
+            for standard CAN or 80_000_000 (80MHz) for CAN FD.
         :param int bitrate:
             Bitrate of channel in bit/s
         :param bool accept_virtual:
@@ -421,7 +452,7 @@ def __init__(self, channel, can_filters=None, **kwargs):
         single_handle = kwargs.get("single_handle", False)
         receive_own_messages = kwargs.get("receive_own_messages", False)
         accept_virtual = kwargs.get("accept_virtual", True)
-        fd = kwargs.get("fd", False)
+        fd = isinstance(timing, BitTimingFd) if timing else kwargs.get("fd", False)
         data_bitrate = kwargs.get("data_bitrate", None)
 
         try:
@@ -457,22 +488,43 @@ def __init__(self, channel, can_filters=None, **kwargs):
             ctypes.byref(ctypes.c_long(TIMESTAMP_RESOLUTION)),
             4,
         )
-
-        if fd:
-            if "tseg1" not in kwargs and bitrate in BITRATE_FD:
-                # Use predefined bitrate for arbitration
-                bitrate = BITRATE_FD[bitrate]
-            if data_bitrate in BITRATE_FD:
-                # Use predefined bitrate for data
-                data_bitrate = BITRATE_FD[data_bitrate]
-            elif not data_bitrate:
-                # Use same bitrate for arbitration and data phase
-                data_bitrate = bitrate
-            canSetBusParamsFd(self._read_handle, data_bitrate, tseg1, tseg2, sjw)
+        if isinstance(timing, BitTimingFd):
+            timing = check_or_adjust_timing_clock(timing, [80_000_000])
+            canSetBusParams(
+                self._read_handle,
+                timing.nom_bitrate,
+                timing.nom_tseg1,
+                timing.nom_tseg2,
+                timing.nom_sjw,
+                1,
+                0,
+            )
+            canSetBusParamsFd(
+                self._read_handle,
+                timing.data_bitrate,
+                timing.data_tseg1,
+                timing.data_tseg2,
+                timing.data_sjw,
+            )
+        elif isinstance(timing, BitTiming):
+            timing = check_or_adjust_timing_clock(timing, [16_000_000])
+            canSetBusParamsC200(self._read_handle, timing.btr0, timing.btr1)
         else:
-            if "tseg1" not in kwargs and bitrate in BITRATE_OBJS:
-                bitrate = BITRATE_OBJS[bitrate]
-        canSetBusParams(self._read_handle, bitrate, tseg1, tseg2, sjw, no_samp, 0)
+            if fd:
+                if "tseg1" not in kwargs and bitrate in BITRATE_FD:
+                    # Use predefined bitrate for arbitration
+                    bitrate = BITRATE_FD[bitrate]
+                if data_bitrate in BITRATE_FD:
+                    # Use predefined bitrate for data
+                    data_bitrate = BITRATE_FD[data_bitrate]
+                elif not data_bitrate:
+                    # Use same bitrate for arbitration and data phase
+                    data_bitrate = bitrate
+                canSetBusParamsFd(self._read_handle, data_bitrate, tseg1, tseg2, sjw)
+            else:
+                if "tseg1" not in kwargs and bitrate in BITRATE_OBJS:
+                    bitrate = BITRATE_OBJS[bitrate]
+            canSetBusParams(self._read_handle, bitrate, tseg1, tseg2, sjw, no_samp, 0)
 
         # By default, use local echo if single handle is used (see #160)
         local_echo = single_handle or receive_own_messages