VectorBus - add methods to handle non message events

can/interfaces/vector/canlib.py

@@ -396,6 +396,9 @@ def _recv_internal(self, timeout):
                             channel=channel,
                         )
                         return msg, self._is_filtered
+                    else:
+                        self.handle_canfd_event(event)
+
             else:
                 event_count.value = 1
                 try:
@@ -431,6 +434,8 @@ def _recv_internal(self, timeout):
                             channel=channel,
                         )
                         return msg, self._is_filtered
+                    else:
+                        self.handle_can_event(event)
 
             if end_time is not None and time.time() > end_time:
                 return None, self._is_filtered
@@ -447,6 +452,30 @@ def _recv_internal(self, timeout):
                 # Wait a short time until we try again
                 time.sleep(self.poll_interval)
 
+    def handle_can_event(self, event: xlclass.XLevent) -> None:
+        """Handle non-message CAN events.
+
+        Method is called by :meth:`~can.interfaces.vector.VectorBus._recv_internal`
+        when `event.tag` is not `XL_CAN_EV_TAG_RX_OK` or `XL_CAN_EV_TAG_TX_OK`.
+        Subclasses can implement this method.
+
+        :param event: XLevent that could have a `XL_CHIP_STATE`, `XL_TIMER` or `XL_SYNC_PULSE` tag.
+        :return: None
+        """
+        pass
+
+    def handle_canfd_event(self, event: xlclass.XLcanRxEvent) -> None:
+        """Handle non-message CAN FD events.
+
+        Method is called by :meth:`~can.interfaces.vector.VectorBus._recv_internal`
+        when `event.tag` is not `XL_RECEIVE_MSG`. Subclasses can implement this method.
+
+        :param event: `XLcanRxEvent` that could have a `XL_CAN_EV_TAG_RX_ERROR`, `XL_CAN_EV_TAG_TX_ERROR`
+            or `XL_CAN_EV_TAG_CHIP_STATE` tag.
+        :return: None
+        """
+        pass
+
     def send(self, msg, timeout=None):
         msg_id = msg.arbitration_id
 

can/interfaces/vector/xlclass.py

@@ -35,6 +35,14 @@ class s_xl_can_ev_error(ctypes.Structure):
     _fields_ = [("errorCode", ctypes.c_ubyte), ("reserved", ctypes.c_ubyte * 95)]
 
 
+class s_xl_chip_state(ctypes.Structure):
+    _fields_ = [
+        ("busStatus", ctypes.c_ubyte),
+        ("txErrorCounter", ctypes.c_ubyte),
+        ("rxErrorCounter", ctypes.c_ubyte),
+    ]
+
+
 class s_xl_can_ev_chip_state(ctypes.Structure):
     _fields_ = [
         ("busStatus", ctypes.c_ubyte),
@@ -55,7 +63,7 @@ class s_xl_can_ev_sync_pulse(ctypes.Structure):
 
 # BASIC bus message structure
 class s_xl_tag_data(ctypes.Union):
-    _fields_ = [("msg", s_xl_can_msg)]
+    _fields_ = [("msg", s_xl_can_msg), ("chipState", s_xl_chip_state)]
 
 
 # CAN FD messages

test/test_vector.py

@@ -55,12 +55,6 @@ def setUp(self) -> None:
         can.interfaces.vector.canlib.xldriver.xlClosePort = Mock(return_value=0)
         can.interfaces.vector.canlib.xldriver.xlCloseDriver = Mock()
 
-        # receiver functions
-        can.interfaces.vector.canlib.xldriver.xlReceive = Mock(side_effect=xlReceive)
-        can.interfaces.vector.canlib.xldriver.xlCanReceive = Mock(
-            side_effect=xlCanReceive
-        )
-
         # sender functions
         can.interfaces.vector.canlib.xldriver.xlCanTransmit = Mock(return_value=0)
         can.interfaces.vector.canlib.xldriver.xlCanTransmitEx = Mock(return_value=0)
@@ -173,16 +167,42 @@ def test_bus_creation_fd_bitrate_timings(self) -> None:
         self.assertEqual(canFdConf.tseg2Dbr, 15)
 
     def test_receive(self) -> None:
+        can.interfaces.vector.canlib.xldriver.xlReceive = Mock(side_effect=xlReceive)
         self.bus = can.Bus(channel=0, bustype="vector", _testing=True)
         self.bus.recv(timeout=0.05)
         can.interfaces.vector.canlib.xldriver.xlReceive.assert_called()
         can.interfaces.vector.canlib.xldriver.xlCanReceive.assert_not_called()
 
     def test_receive_fd(self) -> None:
+        can.interfaces.vector.canlib.xldriver.xlCanReceive = Mock(
+            side_effect=xlCanReceive
+        )
+        self.bus = can.Bus(channel=0, bustype="vector", fd=True, _testing=True)
+        self.bus.recv(timeout=0.05)
+        can.interfaces.vector.canlib.xldriver.xlReceive.assert_not_called()
+        can.interfaces.vector.canlib.xldriver.xlCanReceive.assert_called()
+
+    def test_receive_non_msg_event(self) -> None:
+        can.interfaces.vector.canlib.xldriver.xlReceive = Mock(
+            side_effect=xlReceive_chipstate
+        )
+        self.bus = can.Bus(channel=0, bustype="vector", _testing=True)
+        self.bus.handle_can_event = Mock()
+        self.bus.recv(timeout=0.05)
+        can.interfaces.vector.canlib.xldriver.xlReceive.assert_called()
+        can.interfaces.vector.canlib.xldriver.xlCanReceive.assert_not_called()
+        self.bus.handle_can_event.assert_called()
+
+    def test_receive_fd_non_msg_event(self) -> None:
+        can.interfaces.vector.canlib.xldriver.xlCanReceive = Mock(
+            side_effect=xlCanReceive_chipstate
+        )
         self.bus = can.Bus(channel=0, bustype="vector", fd=True, _testing=True)
+        self.bus.handle_canfd_event = Mock()
         self.bus.recv(timeout=0.05)
         can.interfaces.vector.canlib.xldriver.xlReceive.assert_not_called()
         can.interfaces.vector.canlib.xldriver.xlCanReceive.assert_called()
+        self.bus.handle_canfd_event.assert_called()
 
     def test_send(self) -> None:
         self.bus = can.Bus(channel=0, bustype="vector", _testing=True)
@@ -306,5 +326,31 @@ def xlCanReceive(
     return 0
 
 
+def xlReceive_chipstate(
+    port_handle: xlclass.XLportHandle,
+    event_count_p: ctypes.POINTER(ctypes.c_uint),
+    event: ctypes.POINTER(xlclass.XLevent),
+) -> int:
+    event.tag = xldefine.XL_EventTags.XL_CHIP_STATE.value
+    event.tagData.chipState.busStatus = 8
+    event.tagData.chipState.rxErrorCounter = 0
+    event.tagData.chipState.txErrorCounter = 0
+    event.timeStamp = 0
+    event.chanIndex = 2
+    return 0
+
+
+def xlCanReceive_chipstate(
+    port_handle: xlclass.XLportHandle, event: ctypes.POINTER(xlclass.XLcanRxEvent)
+) -> int:
+    event.tag = xldefine.XL_CANFD_RX_EventTags.XL_CAN_EV_TAG_CHIP_STATE.value
+    event.tagData.canChipState.busStatus = 8
+    event.tagData.canChipState.rxErrorCounter = 0
+    event.tagData.canChipState.txErrorCounter = 0
+    event.timeStamp = 0
+    event.chanIndex = 2
+    return 0
+
+
 if __name__ == "__main__":
     unittest.main()