interface for changing bitrate

can/bus.py

@@ -314,6 +314,21 @@ def __iter__(self) -> Iterator[Message]:
             if msg is not None:
                 yield msg
 
+    @property
+    def bitrate(self) -> int:
+        """Bitrate of the underlying bus in bits/s"""
+        raise NotImplementedError("Device must implement bitrate")
+
+    @bitrate.setter
+    def bitrate(self, bitrate: int) -> None:
+        """Changes the bitrate of the underlying bus.
+
+        Override this method to enable runtime bitrate changes.
+
+        :param int bitrate: the new bitrate to set in bits/s.
+        """
+        raise NotImplementedError("Device not capable of runtime bitrate change")
+
     @property
     def filters(self) -> Optional[can.typechecking.CanFilters]:
         """

can/interfaces/socketcan/constants.py

@@ -43,11 +43,6 @@
 MSK_ARBID = 0x1FFFFFFF
 MSK_FLAGS = 0xE0000000
 
-PF_CAN = 29
-SOCK_RAW = 3
-SOCK_DGRAM = 2
-AF_CAN = PF_CAN
-
 SIOCGIFNAME = 0x8910
 SIOCGIFINDEX = 0x8933
 SIOCGSTAMP = 0x8906

can/interfaces/socketcan/socketcan.py

@@ -21,6 +21,29 @@
 log_tx = log.getChild("tx")
 log_rx = log.getChild("rx")
 
+can_config_lib = ctypes.util.find_library("socketcan123")
+if can_config_lib:
+
+    class BitrateTiming(ctypes.Structure):
+        _fields_ = [
+            ("bitrate", ctypes.c_uint32),
+            ("sample_point", ctypes.c_uint32),
+            ("tq", ctypes.c_uint32),
+            ("prop_seg", ctypes.c_uint32),
+            ("phase_seg1", ctypes.c_uint32),
+            ("phase_seg2", ctypes.c_uint32),
+            ("sjw", ctypes.c_uint32),
+            ("brp", ctypes.c_uint32),
+        ]
+
+    can_config = ctypes.cdll.LoadLibrary(can_config_lib)
+else:
+    can_config = None
+    can_config_error = (
+        "Dynamic bitrate changes not possible. Please install libsocketcan2."
+    )
+    log.error(can_config_error)
+
 try:
     import fcntl
 except ImportError:
@@ -267,7 +290,7 @@ def dissect_can_frame(frame: bytes) -> Tuple[int, int, int, bytes]:
 
 def create_bcm_socket(channel: str) -> socket.socket:
     """create a broadcast manager socket and connect to the given interface"""
-    s = socket.socket(PF_CAN, socket.SOCK_DGRAM, CAN_BCM)
+    s = socket.socket(socket.PF_CAN, socket.SOCK_DGRAM, socket.CAN_BCM)
     s.connect((channel,))
     return s
 
@@ -491,7 +514,7 @@ def create_socket() -> socket.socket:
     """Creates a raw CAN socket. The socket will
     be returned unbound to any interface.
     """
-    sock = socket.socket(PF_CAN, socket.SOCK_RAW, CAN_RAW)
+    sock = socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW)
 
     log.info("Created a socket")
 
@@ -854,6 +877,33 @@ def _detect_available_configs() -> List[can.typechecking.AutoDetectedConfig]:
             for channel in find_available_interfaces()
         ]
 
+    @property
+    def bitrate(self) -> int:
+        if can_config:
+            bitrate_timing = BitrateTiming()
+            error = can_config.can_get_bittiming(
+                self.channel.encode(), ctypes.pointer(bitrate_timing)
+            )
+            if error == -1:
+                log.error("likely need to run as sudo to be able to set the bitrate")
+                return None
+            return bitrate_timing.bitrate
+        else:
+            # TODO - better error?
+            raise NotImplementedError(can_config_error)
+
+    @bitrate.setter
+    def bitrate(self, bitrate: int) -> None:
+        if can_config:
+            can_config.can_do_stop(self.channel.encode())
+            error = can_config.can_set_bitrate(self.channel.encode(), bitrate)
+            can_config.can_do_start(self.channel.encode())
+            if error == -1:
+                log.error("likely need to run as sudo to be able to set the bitrate")
+        else:
+            # TODO - better error?
+            raise NotImplementedError(can_config_error)
+
 
 if __name__ == "__main__":
     # This example demonstrates how to use the internal methods of this module.

can/interfaces/vector/canlib.py

@@ -163,7 +163,7 @@ def __init__(
                     )
                 self.channels = channel_index
             else:
-                # Is there any better way to raise the error?
+                # TODO - Is there any better way to raise the error?
                 raise Exception(
                     "None of the configured channels could be found on the specified hardware."
                 )
@@ -233,6 +233,8 @@ def __init__(
         )
 
         if permission_mask.value == self.mask:
+            self._init_access = True
+            self._bitrate = bitrate
             if fd:
                 self.canFdConf = xlclass.XLcanFdConf()
                 if bitrate:
@@ -277,6 +279,7 @@ def __init__(
                     )
                     LOG.info("SetChannelBitrate: baudr.=%u", bitrate)
         else:
+            self._init_access = False
             LOG.info("No init access!")
 
         # Enable/disable TX receipts
@@ -515,6 +518,25 @@ def handle_canfd_event(self, event: xlclass.XLcanRxEvent) -> None:
     def send(self, msg: Message, timeout: typing.Optional[float] = None):
         self._send_sequence([msg])
 
+    @property
+    def bitrate(self) -> int:
+        return self._bitrate
+
+    @bitrate.setter
+    def bitrate(self, bitrate: int) -> None:
+        if self._init_access:
+            if self.fd:
+                self.canFdConf.arbitrationBitRate = bitrate
+                xldriver.xlCanFdSetConfiguration(
+                    self.port_handle, self.mask, self.canFdConf
+                )
+            else:
+                xldriver.xlCanSetChannelBitrate(self.port_handle, self.mask, bitrate)
+            self._bitrate = bitrate
+            LOG.info("SetChannelBitrate: baudr.=%u", bitrate)
+        else:
+            LOG.error("No init access!")
+
     def _send_sequence(self, msgs: typing.Sequence[Message]) -> int:
         """Send messages and return number of successful transmissions."""
         if self.fd: