diff --git a/can/interfaces/pcan/pcan.py b/can/interfaces/pcan/pcan.py index a11a7fa6c..601ebc9f1 100644 --- a/can/interfaces/pcan/pcan.py +++ b/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 diff --git a/doc/changelog.d/2130.fixed.rst b/doc/changelog.d/2130.fixed.rst new file mode 100644 index 000000000..a3a515a1f --- /dev/null +++ b/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. diff --git a/test/test_pcan.py b/test/test_pcan.py index a9c6ea922..3227515e9 100644 --- a/test/test_pcan.py +++ b/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):