pcan: keep a received frame read with a bus-load warning

can/interfaces/pcan/pcan.py

@@ -57,6 +57,7 @@
     PCAN_MESSAGE_FD,
     PCAN_MESSAGE_RTR,
     PCAN_MESSAGE_STANDARD,
+    PCAN_MESSAGE_STATUS,
     PCAN_NONEBUS,
     PCAN_PARAMETER_OFF,
     PCAN_PARAMETER_ON,
@@ -553,6 +554,12 @@ def _recv_internal(self, timeout: float | None) -> tuple[Message | None, bool]:
 
             elif result & (PCAN_ERROR_BUSLIGHT | PCAN_ERROR_BUSHEAVY):
                 log.warning(self._get_formatted_error(result))
+                if result & ~(PCAN_ERROR_BUSLIGHT | PCAN_ERROR_BUSHEAVY) == 0 and not (
+                    pcan_msg.MSGTYPE
+                    & (PCAN_MESSAGE_STATUS.value | PCAN_MESSAGE_ERRFRAME.value)
+                ):
+                    # the bus state came with a received frame, keep the frame
+                    break
 
             elif result == PCAN_ERROR_ILLDATA:
                 # When there is an invalid frame on CAN bus (in our case CAN FD), PCAN first reports result PCAN_ERROR_ILLDATA

doc/changelog.d/2130.fixed.rst

@@ -0,0 +1 @@
+Keep a received frame on the ``pcan`` interface when the read also reports a bus-load warning (``PCAN_ERROR_BUSLIGHT`` or ``PCAN_ERROR_BUSHEAVY``) instead of dropping it.

test/test_pcan.py

@@ -258,6 +258,45 @@ def test_recv_fd(self):
         self.assertEqual(recv_msg.timestamp, 0)
         self.assertEqual(recv_msg.channel, "PCAN_USBBUS1")
 
+    @parameterized.expand(
+        [
+            ("buslight", PCAN_ERROR_BUSLIGHT),
+            ("busheavy", PCAN_ERROR_BUSHEAVY),
+            ("both", PCAN_ERROR_BUSLIGHT | PCAN_ERROR_BUSHEAVY),
+        ]
+    )
+    def test_recv_bus_load_flags_keep_the_frame(self, name, result):
+        data = (ctypes.c_ubyte * 8)(*range(8))
+        msg = TPCANMsg(ID=0xC0FFEF, LEN=8, MSGTYPE=PCAN_MESSAGE_EXTENDED, DATA=data)
+        self.mock_pcan.Read = Mock(return_value=(result, msg, TPCANTimestamp()))
+        self.mock_pcan.GetErrorText = Mock(return_value=(PCAN_ERROR_OK, b"bus load"))
+        self.bus = can.Bus(interface="pcan")
+
+        recv_msg = self.bus.recv(timeout=0)
+        self.assertIsNotNone(recv_msg)
+        self.assertEqual(recv_msg.arbitration_id, msg.ID)
+        self.assertSequenceEqual(recv_msg.data, msg.DATA)
+        self.assertFalse(recv_msg.is_error_frame)
+
+    @parameterized.expand(
+        [
+            ("status", PCAN_ERROR_BUSHEAVY, PCAN_MESSAGE_STATUS.value),
+            ("error_frame", PCAN_ERROR_BUSHEAVY, PCAN_MESSAGE_ERRFRAME.value),
+            (
+                "other_error",
+                PCAN_ERROR_BUSHEAVY | PCAN_ERROR_QOVERRUN,
+                PCAN_MESSAGE_STANDARD.value,
+            ),
+        ]
+    )
+    def test_recv_bus_load_flags_drop_other_reads(self, name, result, msgtype):
+        msg = TPCANMsg(ID=0x123, LEN=0, MSGTYPE=msgtype)
+        self.mock_pcan.Read = Mock(return_value=(result, msg, TPCANTimestamp()))
+        self.mock_pcan.GetErrorText = Mock(return_value=(PCAN_ERROR_OK, b"bus load"))
+        self.bus = can.Bus(interface="pcan")
+
+        self.assertIsNone(self.bus.recv(timeout=0))
+
     @pytest.mark.timeout(3.0)
     @patch("select.select", return_value=([], [], []))
     def test_recv_no_message(self, mock_select):