add FD support to slcan according to CANable 2.0 impementation

Author: spencer

can/interfaces/slcan.py

@@ -16,7 +16,7 @@
     CanOperationError,
     error_check,
 )
-from can.util import check_or_adjust_timing_clock, deprecated_args_alias
+from can.util import CAN_FD_DLC, check_or_adjust_timing_clock, deprecated_args_alias
 
 logger = logging.getLogger(__name__)
 
@@ -48,6 +48,10 @@ class slcanBus(BusABC):
         1000000: "S8",
         83300: "S9",
     }
+    _DATA_BITRATES = {
+        2000000: "Y2",
+        5000000: "Y5",
+    }
 
     _SLEEP_AFTER_SERIAL_OPEN = 2  # in seconds
 
@@ -86,7 +90,8 @@ def __init__(
             If this argument is set then it overrides the bitrate and btr arguments. The
             `f_clock` value of the timing instance must be set to 8_000_000 (8MHz)
             for standard CAN.
-            CAN FD and the :class:`~can.BitTimingFd` class are not supported.
+            CAN FD and the :class:`~can.BitTimingFd` class have partial support according to the non-standard
+            slcan protocol implementation in the CANABLE 2.0 firmware: currently only data rates of 2M and 5M.
         :param poll_interval:
             Poll interval in seconds when reading messages
         :param sleep_after_open:
@@ -143,9 +148,7 @@ def __init__(
                 timing = check_or_adjust_timing_clock(timing, valid_clocks=[8_000_000])
                 self.set_bitrate_reg(f"{timing.btr0:02X}{timing.btr1:02X}")
             elif isinstance(timing, BitTimingFd):
-                raise NotImplementedError(
-                    f"CAN FD is not supported by {self.__class__.__name__}."
-                )
+                self.set_bitrate(timing.nom_bitrate, timing.data_bitrate)
             else:
                 if bitrate is not None and btr is not None:
                     raise ValueError("Bitrate and btr mutually exclusive.")
@@ -157,10 +160,12 @@ def __init__(
 
         super().__init__(channel, **kwargs)
 
-    def set_bitrate(self, bitrate: int) -> None:
+    def set_bitrate(self, bitrate: int, dbitrate: int = 0) -> None:
         """
         :param bitrate:
             Bitrate in bit/s
+        :param dbitrate:
+            Data Bitrate in bit/s for FD frames
 
         :raise ValueError: if ``bitrate`` is not among the possible values
         """
@@ -169,9 +174,15 @@ def set_bitrate(self, bitrate: int) -> None:
         else:
             bitrates = ", ".join(str(k) for k in self._BITRATES.keys())
             raise ValueError(f"Invalid bitrate, choose one of {bitrates}.")
+        if dbitrate in self._DATA_BITRATES:
+            dbitrate_code = self._DATA_BITRATES[dbitrate]
+        else:
+            dbitrates = ", ".join(str(k) for k in self._DATA_BITRATES.keys())
+            raise ValueError(f"Invalid data bitrate, choose one of {dbitrates}.")
 
         self.close()
         self._write(bitrate_code)
+        self._write(dbitrate_code)
         self.open()
 
     def set_bitrate_reg(self, btr: str) -> None:
@@ -235,6 +246,8 @@ def _recv_internal(
         remote = False
         extended = False
         data = None
+        isFd = False
+        fdBrs = False
 
         if self._queue.qsize():
             string: Optional[str] = self._queue.get_nowait()
@@ -268,13 +281,43 @@ def _recv_internal(
             dlc = int(string[9])
             extended = True
             remote = True
+        elif string[0] == "d":
+            # FD standard frame
+            canId = int(string[1:4], 16)
+            dlc = int(string[4], 16)
+            isFd = True
+            data = bytearray.fromhex(string[5 : 5 + CAN_FD_DLC[dlc] * 2])
+        elif string[0] == "D":
+            # FD extended frame
+            canId = int(string[1:9], 16)
+            dlc = int(string[9], 16)
+            extended = True
+            isFd = True
+            data = bytearray.fromhex(string[10 : 10 + CAN_FD_DLC[dlc] * 2])
+        elif string[0] == "b":
+            # FD with bitrate switch
+            canId = int(string[1:4], 16)
+            dlc = int(string[4], 16)
+            isFd = True
+            fdBrs = True
+            data = bytearray.fromhex(string[5 : 5 + CAN_FD_DLC[dlc] * 2])
+        elif string[0] == "B":
+            # FD extended with bitrate switch
+            canId = int(string[1:9], 16)
+            dlc = int(string[9], 16)
+            extended = True
+            isFd = True
+            fdBrs = True
+            data = bytearray.fromhex(string[10 : 10 + CAN_FD_DLC[dlc] * 2])
 
         if canId is not None:
             msg = Message(
                 arbitration_id=canId,
                 is_extended_id=extended,
                 timestamp=time.time(),  # Better than nothing...
                 is_remote_frame=remote,
+                is_fd=isFd,
+                bitrate_switch=fdBrs,
                 dlc=dlc,
                 data=data,
             )
@@ -289,6 +332,19 @@ def send(self, msg: Message, timeout: Optional[float] = None) -> None:
                 sendStr = f"R{msg.arbitration_id:08X}{msg.dlc:d}"
             else:
                 sendStr = f"r{msg.arbitration_id:03X}{msg.dlc:d}"
+        elif msg.is_fd:
+            if msg.bitrate_switch:
+                if msg.is_extended_id:
+                    sendStr = f"B{msg.arbitration_id:08X}{msg.dlc:d}"
+                else:
+                    sendStr = f"b{msg.arbitration_id:03X}{msg.dlc:d}"
+                sendStr += msg.data.hex().upper()
+            else:
+                if msg.is_extended_id:
+                    sendStr = f"D{msg.arbitration_id:08X}{msg.dlc:d}"
+                else:
+                    sendStr = f"d{msg.arbitration_id:03X}{msg.dlc:d}"
+                sendStr += msg.data.hex().upper()
         else:
             if msg.is_extended_id:
                 sendStr = f"T{msg.arbitration_id:08X}{msg.dlc:d}"