Add usage of select instead of polling for Linux based platform (PCAN Interface) (#1410)

* - integrate handling of Linux based event using select for pcan interface

* - adapt pcan unit tests regarding PCAN_RECEIVE_EVENT GetValue call

* - add the case HAS_EVENTS = FALSE whne no special event basic mechanism is importes

* - force select mock for test_recv_no_message

* - just for testing : see what is going wrong with CI testing and my setup

* - correct patch on pcan test test_recv_no_message

- run black

* - remove debug log

* -just reformat test_pcan

* - move global variable IS_WINDOW and IS_LINUX from pcan to basic module

* - remove redundant comment in pcan _recv_internal

* refactor _recv_internal

Co-authored-by: zariiii9003 <52598363+zariiii9003@users.noreply.github.com>

Author: BKaDamien

can/interfaces/pcan/basic.py

@@ -21,9 +21,12 @@
 
 import logging
 
-if platform.system() == "Windows":
-    import winreg
+PLATFORM = platform.system()
+IS_WINDOWS = PLATFORM == "Windows"
+IS_LINUX = PLATFORM == "Linux"
 
+if IS_WINDOWS:
+    import winreg
 
 logger = logging.getLogger("can.pcan")
 

can/interfaces/pcan/pcan.py

@@ -1,12 +1,11 @@
 """
 Enable basic CAN over a PCAN USB device.
 """
-
 import logging
 import time
 from datetime import datetime
 import platform
-from typing import Optional, List
+from typing import Optional, List, Tuple
 
 from packaging import version
 
@@ -50,6 +49,8 @@
     PCAN_LISTEN_ONLY,
     PCAN_PARAMETER_OFF,
     TPCANHandle,
+    IS_LINUX,
+    IS_WINDOWS,
     PCAN_PCIBUS1,
     PCAN_USBBUS1,
     PCAN_PCCBUS1,
@@ -70,7 +71,6 @@
 
 MIN_PCAN_API_VERSION = version.parse("4.2.0")
 
-
 try:
     # use the "uptime" library if available
     import uptime
@@ -86,22 +86,27 @@
     )
     boottimeEpoch = 0
 
-try:
-    # Try builtin Python 3 Windows API
-    from _overlapped import CreateEvent
-    from _winapi import WaitForSingleObject, WAIT_OBJECT_0, INFINITE
+HAS_EVENTS = False
 
-    HAS_EVENTS = True
-except ImportError:
+if IS_WINDOWS:
     try:
-        # Try pywin32 package
-        from win32event import CreateEvent
-        from win32event import WaitForSingleObject, WAIT_OBJECT_0, INFINITE
+        # Try builtin Python 3 Windows API
+        from _overlapped import CreateEvent
+        from _winapi import WaitForSingleObject, WAIT_OBJECT_0, INFINITE
 
         HAS_EVENTS = True
     except ImportError:
-        # Use polling instead
-        HAS_EVENTS = False
+        pass
+
+elif IS_LINUX:
+    try:
+        import errno
+        import os
+        import select
+
+        HAS_EVENTS = True
+    except Exception:
+        pass
 
 
 class PcanBus(BusABC):
@@ -294,10 +299,16 @@ def __init__(
                 raise PcanCanInitializationError(self._get_formatted_error(result))
 
         if HAS_EVENTS:
-            self._recv_event = CreateEvent(None, 0, 0, None)
-            result = self.m_objPCANBasic.SetValue(
-                self.m_PcanHandle, PCAN_RECEIVE_EVENT, self._recv_event
-            )
+            if IS_WINDOWS:
+                self._recv_event = CreateEvent(None, 0, 0, None)
+                result = self.m_objPCANBasic.SetValue(
+                    self.m_PcanHandle, PCAN_RECEIVE_EVENT, self._recv_event
+                )
+            elif IS_LINUX:
+                result, self._recv_event = self.m_objPCANBasic.GetValue(
+                    self.m_PcanHandle, PCAN_RECEIVE_EVENT
+                )
+
             if result != PCAN_ERROR_OK:
                 raise PcanCanInitializationError(self._get_formatted_error(result))
 
@@ -441,84 +452,96 @@ def set_device_number(self, device_number):
             return False
         return True
 
-    def _recv_internal(self, timeout):
+    def _recv_internal(
+        self, timeout: Optional[float]
+    ) -> Tuple[Optional[Message], bool]:
+        end_time = time.time() + timeout if timeout is not None else None
 
-        if HAS_EVENTS:
-            # We will utilize events for the timeout handling
-            timeout_ms = int(timeout * 1000) if timeout is not None else INFINITE
-        elif timeout is not None:
-            # Calculate max time
-            end_time = time.perf_counter() + timeout
-
-        # log.debug("Trying to read a msg")
-
-        result = None
-        while result is None:
+        while True:
             if self.fd:
-                result = self.m_objPCANBasic.ReadFD(self.m_PcanHandle)
+                result, pcan_msg, pcan_timestamp = self.m_objPCANBasic.ReadFD(
+                    self.m_PcanHandle
+                )
             else:
-                result = self.m_objPCANBasic.Read(self.m_PcanHandle)
-            if result[0] == PCAN_ERROR_QRCVEMPTY:
-                if HAS_EVENTS:
-                    result = None
-                    val = WaitForSingleObject(self._recv_event, timeout_ms)
-                    if val != WAIT_OBJECT_0:
-                        return None, False
-                elif timeout is not None and time.perf_counter() >= end_time:
-                    return None, False
+                result, pcan_msg, pcan_timestamp = self.m_objPCANBasic.Read(
+                    self.m_PcanHandle
+                )
+
+            if result == PCAN_ERROR_OK:
+                # message received
+                break
+
+            if result == PCAN_ERROR_QRCVEMPTY:
+                # receive queue is empty, wait or return on timeout
+
+                if end_time is None:
+                    time_left: Optional[float] = None
+                    timed_out = False
                 else:
-                    result = None
+                    time_left = max(0.0, end_time - time.time())
+                    timed_out = time_left == 0.0
+
+                if timed_out:
+                    return None, False
+
+                if not HAS_EVENTS:
+                    # polling mode
                     time.sleep(0.001)
-            elif result[0] & (PCAN_ERROR_BUSLIGHT | PCAN_ERROR_BUSHEAVY):
-                log.warning(self._get_formatted_error(result[0]))
-                return None, False
-            elif result[0] != PCAN_ERROR_OK:
-                raise PcanCanOperationError(self._get_formatted_error(result[0]))
-
-        theMsg = result[1]
-        itsTimeStamp = result[2]
-
-        # log.debug("Received a message")
-
-        is_extended_id = (
-            theMsg.MSGTYPE & PCAN_MESSAGE_EXTENDED.value
-        ) == PCAN_MESSAGE_EXTENDED.value
-        is_remote_frame = (
-            theMsg.MSGTYPE & PCAN_MESSAGE_RTR.value
-        ) == PCAN_MESSAGE_RTR.value
-        is_fd = (theMsg.MSGTYPE & PCAN_MESSAGE_FD.value) == PCAN_MESSAGE_FD.value
-        bitrate_switch = (
-            theMsg.MSGTYPE & PCAN_MESSAGE_BRS.value
-        ) == PCAN_MESSAGE_BRS.value
-        error_state_indicator = (
-            theMsg.MSGTYPE & PCAN_MESSAGE_ESI.value
-        ) == PCAN_MESSAGE_ESI.value
-        is_error_frame = (
-            theMsg.MSGTYPE & PCAN_MESSAGE_ERRFRAME.value
-        ) == PCAN_MESSAGE_ERRFRAME.value
+                    continue
+
+                if IS_WINDOWS:
+                    # Windows with event
+                    if time_left is None:
+                        time_left_ms = INFINITE
+                    else:
+                        time_left_ms = int(time_left * 1000)
+                    _ret = WaitForSingleObject(self._recv_event, time_left_ms)
+                    if _ret == WAIT_OBJECT_0:
+                        continue
+
+                elif IS_LINUX:
+                    # Linux with event
+                    recv, _, _ = select.select([self._recv_event], [], [], time_left)
+                    if self._recv_event in recv:
+                        continue
+
+            elif result & (PCAN_ERROR_BUSLIGHT | PCAN_ERROR_BUSHEAVY):
+                log.warning(self._get_formatted_error(result))
+
+            else:
+                raise PcanCanOperationError(self._get_formatted_error(result))
+
+            return None, False
+
+        is_extended_id = bool(pcan_msg.MSGTYPE & PCAN_MESSAGE_EXTENDED.value)
+        is_remote_frame = bool(pcan_msg.MSGTYPE & PCAN_MESSAGE_RTR.value)
+        is_fd = bool(pcan_msg.MSGTYPE & PCAN_MESSAGE_FD.value)
+        bitrate_switch = bool(pcan_msg.MSGTYPE & PCAN_MESSAGE_BRS.value)
+        error_state_indicator = bool(pcan_msg.MSGTYPE & PCAN_MESSAGE_ESI.value)
+        is_error_frame = bool(pcan_msg.MSGTYPE & PCAN_MESSAGE_ERRFRAME.value)
 
         if self.fd:
-            dlc = dlc2len(theMsg.DLC)
-            timestamp = boottimeEpoch + (itsTimeStamp.value / (1000.0 * 1000.0))
+            dlc = dlc2len(pcan_msg.DLC)
+            timestamp = boottimeEpoch + (pcan_timestamp.value / (1000.0 * 1000.0))
         else:
-            dlc = theMsg.LEN
+            dlc = pcan_msg.LEN
             timestamp = boottimeEpoch + (
                 (
-                    itsTimeStamp.micros
-                    + 1000 * itsTimeStamp.millis
-                    + 0x100000000 * 1000 * itsTimeStamp.millis_overflow
+                    pcan_timestamp.micros
+                    + 1000 * pcan_timestamp.millis
+                    + 0x100000000 * 1000 * pcan_timestamp.millis_overflow
                 )
                 / (1000.0 * 1000.0)
             )
 
         rx_msg = Message(
             timestamp=timestamp,
-            arbitration_id=theMsg.ID,
+            arbitration_id=pcan_msg.ID,
             is_extended_id=is_extended_id,
             is_remote_frame=is_remote_frame,
             is_error_frame=is_error_frame,
             dlc=dlc,
-            data=theMsg.DATA[:dlc],
+            data=pcan_msg.DATA[:dlc],
             is_fd=is_fd,
             bitrate_switch=bitrate_switch,
             error_state_indicator=error_state_indicator,
@@ -597,6 +620,9 @@ def flash(self, flash):
 
     def shutdown(self):
         super().shutdown()
+        if HAS_EVENTS and IS_LINUX:
+            self.m_objPCANBasic.SetValue(self.m_PcanHandle, PCAN_RECEIVE_EVENT, 0)
+
         self.m_objPCANBasic.Uninitialize(self.m_PcanHandle)
 
     @property

test/test_pcan.py

@@ -6,7 +6,8 @@
 import platform
 import unittest
 from unittest import mock
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
+
 
 import pytest
 from parameterized import parameterized
@@ -30,7 +31,6 @@ def setUp(self) -> None:
         self.mock_pcan.SetValue = Mock(return_value=PCAN_ERROR_OK)
         self.mock_pcan.GetValue = self._mockGetValue
         self.PCAN_API_VERSION_SIM = "4.2"
-
         self.bus = None
 
     def tearDown(self) -> None:
@@ -45,6 +45,8 @@ def _mockGetValue(self, channel, parameter):
         """
         if parameter == PCAN_API_VERSION:
             return PCAN_ERROR_OK, self.PCAN_API_VERSION_SIM.encode("ascii")
+        elif parameter == PCAN_RECEIVE_EVENT:
+            return PCAN_ERROR_OK, int.from_bytes(PCAN_RECEIVE_EVENT, "big")
         raise NotImplementedError(
             f"No mock return value specified for parameter {parameter}"
         )
@@ -205,7 +207,8 @@ def test_recv_fd(self):
         self.assertEqual(recv_msg.timestamp, 0)
 
     @pytest.mark.timeout(3.0)
-    def test_recv_no_message(self):
+    @patch("select.select", return_value=([], [], []))
+    def test_recv_no_message(self, mock_select):
         self.mock_pcan.Read = Mock(return_value=(PCAN_ERROR_QRCVEMPTY, None, None))
         self.bus = can.Bus(interface="pcan")
         self.assertEqual(self.bus.recv(timeout=0.5), None)