Merge pull request #1188 from pkess/pcan_check_api_version

Implement check for minimum version of pcan library

Author: felixdivo

can/interfaces/pcan/pcan.py

@@ -9,6 +9,8 @@
 
 from typing import Optional
 
+from packaging import version
+
 from ...message import Message
 from ...bus import BusABC, BusState
 from ...util import len2dlc, dlc2len
@@ -19,13 +21,15 @@
     PCAN_BITRATES,
     PCAN_FD_PARAMETER_LIST,
     PCAN_CHANNEL_NAMES,
+    PCAN_NONEBUS,
     PCAN_BAUD_500K,
     PCAN_TYPE_ISA,
     PCANBasic,
     PCAN_ERROR_OK,
     PCAN_ALLOW_ERROR_FRAMES,
     PCAN_PARAMETER_ON,
     PCAN_RECEIVE_EVENT,
+    PCAN_API_VERSION,
     PCAN_DEVICE_NUMBER,
     PCAN_ERROR_QRCVEMPTY,
     PCAN_ERROR_BUSLIGHT,
@@ -58,6 +62,8 @@
 # Set up logging
 log = logging.getLogger("can.pcan")
 
+MIN_PCAN_API_VERSION = version.parse("4.2.0")
+
 
 try:
     # use the "uptime" library if available
@@ -100,7 +106,7 @@ def __init__(
         state=BusState.ACTIVE,
         bitrate=500000,
         *args,
-        **kwargs
+        **kwargs,
     ):
         """A PCAN USB interface to CAN.
 
@@ -206,6 +212,8 @@ def __init__(
         self.m_objPCANBasic = PCANBasic()
         self.m_PcanHandle = channel
 
+        self.check_api_version()
+
         if state is BusState.ACTIVE or state is BusState.PASSIVE:
             self.state = state
         else:
@@ -304,6 +312,21 @@ def bits(n):
 
         return complete_text
 
+    def get_api_version(self):
+        error, value = self.m_objPCANBasic.GetValue(PCAN_NONEBUS, PCAN_API_VERSION)
+        if error != PCAN_ERROR_OK:
+            raise CanInitializationError(f"Failed to read pcan basic api version")
+
+        return version.parse(value.decode("ascii"))
+
+    def check_api_version(self):
+        apv = self.get_api_version()
+        if apv < MIN_PCAN_API_VERSION:
+            log.warning(
+                f"Minimum version of pcan api is {MIN_PCAN_API_VERSION}."
+                f" Installed version is {apv}. Consider upgrade of pcan basic package"
+            )
+
     def status(self):
         """
         Query the PCAN bus status.

setup.py

@@ -90,6 +90,7 @@
         "typing_extensions>=3.10.0.0",
         'pywin32;platform_system=="Windows" and platform_python_implementation=="CPython"',
         'msgpack~=1.0.0;platform_system!="Windows"',
+        "packaging",
     ],
     extras_require=extras_require,
 )

test/test_pcan.py

@@ -12,6 +12,7 @@
 
 import can
 from can.bus import BusState
+from can.exceptions import CanInitializationError
 from can.interfaces.pcan.basic import *
 from can.interfaces.pcan import PcanBus, PcanError
 
@@ -26,6 +27,8 @@ def setUp(self) -> None:
         self.mock_pcan.Initialize.return_value = PCAN_ERROR_OK
         self.mock_pcan.InitializeFD = Mock(return_value=PCAN_ERROR_OK)
         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
 
@@ -34,6 +37,17 @@ def tearDown(self) -> None:
             self.bus.shutdown()
             self.bus = None
 
+    def _mockGetValue(self, channel, parameter):
+        """
+        This method is used as mock for GetValue method of PCANBasic object.
+        Only a subset of parameters are supported.
+        """
+        if parameter == PCAN_API_VERSION:
+            return PCAN_ERROR_OK, self.PCAN_API_VERSION_SIM.encode("ascii")
+        raise NotImplementedError(
+            f"No mock return value specified for parameter {parameter}"
+        )
+
     def test_bus_creation(self) -> None:
         self.bus = can.Bus(bustype="pcan")
         self.assertIsInstance(self.bus, PcanBus)
@@ -52,6 +66,24 @@ def test_bus_creation_fd(self) -> None:
         self.mock_pcan.Initialize.assert_not_called()
         self.mock_pcan.InitializeFD.assert_called_once()
 
+    def test_api_version_low(self) -> None:
+        self.PCAN_API_VERSION_SIM = "1.0"
+        with self.assertLogs("can.pcan", level="WARNING") as cm:
+            self.bus = can.Bus(bustype="pcan")
+            found_version_warning = False
+            for i in cm.output:
+                if "version" in i and "pcan" in i:
+                    found_version_warning = True
+            self.assertTrue(
+                found_version_warning,
+                f"No warning was logged for incompatible api version {cm.output}",
+            )
+
+    def test_api_version_read_fail(self) -> None:
+        self.mock_pcan.GetValue = Mock(return_value=(PCAN_ERROR_ILLOPERATION, None))
+        with self.assertRaises(CanInitializationError):
+            self.bus = can.Bus(bustype="pcan")
+
     @parameterized.expand(
         [
             ("no_error", PCAN_ERROR_OK, PCAN_ERROR_OK, "some ok text 1"),
@@ -108,8 +140,11 @@ def test_reset(self, name, status, expected_result) -> None:
     )
     def test_get_device_number(self, name, status, expected_result) -> None:
         with self.subTest(name):
-            self.mock_pcan.GetValue = Mock(return_value=(status, 1))
             self.bus = can.Bus(bustype="pcan", fd=True)
+            # Mock GetValue after creation of bus to use first mock of
+            # GetValue in constructor
+            self.mock_pcan.GetValue = Mock(return_value=(status, 1))
+
             self.assertEqual(self.bus.get_device_number(), expected_result)
             self.mock_pcan.GetValue.assert_called_once_with(
                 PCAN_USBBUS1, PCAN_DEVICE_NUMBER