Add option to disable bit rate switching for socket can

pycyphal/application/_transport_factory.py

@@ -283,6 +283,7 @@ def init(name: str, default: RelaxedValue) -> ValueProxy:
     iface_list = str(init("iface", "")).split()
     mtu = int(init("mtu", Natural16([64])))
     br_arb, br_data = init("bitrate", Natural32([1_000_000, 4_000_000])).ints
+    disable_brs = bool(init("disable_brs", br_arb == br_data))
 
     if iface_list:
         from pycyphal.transport.can import CANTransport
@@ -292,7 +293,7 @@ def init(name: str, default: RelaxedValue) -> ValueProxy:
             if iface.lower().startswith("socketcan:"):
                 from pycyphal.transport.can.media.socketcan import SocketCANMedia
 
-                media = SocketCANMedia(iface.split(":", 1)[-1], mtu=mtu)
+                media = SocketCANMedia(iface.split(":", 1)[-1], mtu=mtu, disable_brs=disable_brs)
             elif iface.lower().startswith("candump:"):
                 from pycyphal.transport.can.media.candump import CandumpMedia
 

pycyphal/transport/can/media/socketcan/_socketcan.py

@@ -41,7 +41,13 @@ class SocketCANMedia(Media):
     SocketCAN documentation: https://www.kernel.org/doc/Documentation/networking/can.txt
     """
 
-    def __init__(self, iface_name: str, mtu: int, loop: typing.Optional[asyncio.AbstractEventLoop] = None) -> None:
+    def __init__(
+        self,
+        iface_name: str,
+        mtu: int,
+        disable_brs: bool = False,
+        loop: typing.Optional[asyncio.AbstractEventLoop] = None,
+    ) -> None:
         """
         CAN Classic/FD is selected automatically based on the MTU. It is not possible to use CAN FD with MTU of 8 bytes.
 
@@ -50,6 +56,9 @@ def __init__(self, iface_name: str, mtu: int, loop: typing.Optional[asyncio.Abst
         :param mtu: The maximum data field size in bytes. CAN FD is used if this value > 8, Classic CAN otherwise.
             This value must belong to Media.VALID_MTU_SET.
 
+        :param disable_brs: When true, will disabele bitrate switching for CAN FD frames. Meaning that the data bitrate
+            will be the same as the nominal bitrate.
+
         :param loop: Deprecated.
         """
         # This can't be made a class attribute because these errnos are only available on GNU/Linux.
@@ -68,6 +77,8 @@ def __init__(self, iface_name: str, mtu: int, loop: typing.Optional[asyncio.Abst
         self._mtu = int(mtu)
         if self._mtu not in self.VALID_MTU_SET:
             raise ValueError(f"Invalid MTU: {self._mtu} not in {self.VALID_MTU_SET}")
+        self._disable_brs: bool = disable_brs
+
         if loop:
             warnings.warn("The loop argument is deprecated", DeprecationWarning)
 
@@ -260,7 +271,7 @@ def _read_frame(self, ts_mono_ns: int) -> typing.Tuple[Timestamp, Envelope]:
                 return timestamp, Envelope(out, loopback=loopback)
 
     def _compile_native_frame(self, source: DataFrame) -> bytes:
-        flags = _CANFD_BRS if self._is_fd else 0
+        flags = _CANFD_BRS if (self._is_fd and not self._disable_brs) else 0
         ident = source.identifier | (_CAN_EFF_FLAG if source.format == FrameFormat.EXTENDED else 0)
         header = _FRAME_HEADER_STRUCT.pack(ident, len(source.data), flags)
         out = header + source.data.ljust(self._native_frame_data_capacity, b"\x00")