Implement _detect_available_configs for the Ixxat bus. (#1607)

* Add _detect_available_configs to ixxat bus

* Add typing and cover CI test failure

* Format code with black

* Format code with black

* re-order imports for ruff

* Update ixxat docs

* fix doctest

* Update test_interface_ixxat.py

* make ruff happy

---------

Co-authored-by: MattWoodhead <[email protected]>
Co-authored-by: zariiii9003 <[email protected]>

Author: MattWoodhead

can/interfaces/ixxat/canlib.py

@@ -1,4 +1,4 @@
-from typing import Callable, Optional, Sequence, Union
+from typing import Callable, List, Optional, Sequence, Union
 
 import can.interfaces.ixxat.canlib_vcinpl as vcinpl
 import can.interfaces.ixxat.canlib_vcinpl2 as vcinpl2
@@ -8,6 +8,7 @@
     CyclicSendTaskABC,
     Message,
 )
+from can.typechecking import AutoDetectedConfig
 
 
 class IXXATBus(BusABC):
@@ -170,3 +171,7 @@ def state(self) -> BusState:
         Return the current state of the hardware
         """
         return self.bus.state
+
+    @staticmethod
+    def _detect_available_configs() -> List[AutoDetectedConfig]:
+        return vcinpl._detect_available_configs()

can/interfaces/ixxat/canlib_vcinpl.py

@@ -14,7 +14,7 @@
 import logging
 import sys
 import warnings
-from typing import Callable, Optional, Sequence, Tuple, Union
+from typing import Callable, List, Optional, Sequence, Tuple, Union
 
 from can import (
     BusABC,
@@ -28,6 +28,7 @@
 from can.ctypesutil import HANDLE, PHANDLE, CLibrary
 from can.ctypesutil import HRESULT as ctypes_HRESULT
 from can.exceptions import CanInitializationError, CanInterfaceNotImplementedError
+from can.typechecking import AutoDetectedConfig
 from can.util import deprecated_args_alias
 
 from . import constants, structures
@@ -943,3 +944,55 @@ def get_ixxat_hwids():
     _canlib.vciEnumDeviceClose(device_handle)
 
     return hwids
+
+
+def _detect_available_configs() -> List[AutoDetectedConfig]:
+    config_list = []  # list in wich to store the resulting bus kwargs
+
+    # used to detect HWID
+    device_handle = HANDLE()
+    device_info = structures.VCIDEVICEINFO()
+
+    # used to attempt to open channels
+    channel_handle = HANDLE()
+    device_handle2 = HANDLE()
+
+    try:
+        _canlib.vciEnumDeviceOpen(ctypes.byref(device_handle))
+        while True:
+            try:
+                _canlib.vciEnumDeviceNext(device_handle, ctypes.byref(device_info))
+            except StopIteration:
+                break
+            else:
+                hwid = device_info.UniqueHardwareId.AsChar.decode("ascii")
+                _canlib.vciDeviceOpen(
+                    ctypes.byref(device_info.VciObjectId),
+                    ctypes.byref(device_handle2),
+                )
+                for channel in range(4):
+                    try:
+                        _canlib.canChannelOpen(
+                            device_handle2,
+                            channel,
+                            constants.FALSE,
+                            ctypes.byref(channel_handle),
+                        )
+                    except Exception:
+                        # Array outside of bounds error == accessing a channel not in the hardware
+                        break
+                    else:
+                        _canlib.canChannelClose(channel_handle)
+                        config_list.append(
+                            {
+                                "interface": "ixxat",
+                                "channel": channel,
+                                "unique_hardware_id": hwid,
+                            }
+                        )
+                _canlib.vciDeviceClose(device_handle2)
+        _canlib.vciEnumDeviceClose(device_handle)
+    except AttributeError:
+        pass  # _canlib is None in the CI tests -> return a blank list
+
+    return config_list

can/interfaces/ixxat/canlib_vcinpl2.py

@@ -509,17 +509,15 @@ def __init__(
             tseg1_abr is None or tseg2_abr is None or sjw_abr is None
         ):
             raise ValueError(
-                "To use bitrate {} (that has not predefined preset) is mandatory to use also parameters tseg1_abr, tseg2_abr and swj_abr".format(
-                    bitrate
-                )
+                f"To use bitrate {bitrate} (that has not predefined preset) is mandatory "
+                f"to use also parameters tseg1_abr, tseg2_abr and swj_abr"
             )
         if data_bitrate not in constants.CAN_DATABITRATE_PRESETS and (
             tseg1_dbr is None or tseg2_dbr is None or sjw_dbr is None
         ):
             raise ValueError(
-                "To use data_bitrate {} (that has not predefined preset) is mandatory to use also parameters tseg1_dbr, tseg2_dbr and swj_dbr".format(
-                    data_bitrate
-                )
+                f"To use data_bitrate {data_bitrate} (that has not predefined preset) is mandatory "
+                f"to use also parameters tseg1_dbr, tseg2_dbr and swj_dbr"
             )
 
         if rx_fifo_size <= 0:

can/interfaces/pcan/pcan.py

@@ -396,9 +396,7 @@ def bits(n):
             for b in bits(error):
                 stsReturn = self.m_objPCANBasic.GetErrorText(b, 0x9)
                 if stsReturn[0] != PCAN_ERROR_OK:
-                    text = "An error occurred. Error-code's text ({:X}h) couldn't be retrieved".format(
-                        error
-                    )
+                    text = f"An error occurred. Error-code's text ({error:X}h) couldn't be retrieved"
                 else:
                     text = stsReturn[1].decode("utf-8", errors="replace")
 

doc/interfaces/ixxat.rst

@@ -56,17 +56,42 @@ VCI documentation, section "Message filters" for more info.
 
 List available devices
 ----------------------
-In case you have connected multiple IXXAT devices, you have to select them by using their unique hardware id.
-To get a list of all connected IXXAT you can use the function ``get_ixxat_hwids()`` as demonstrated below:
+
+In case you have connected multiple IXXAT devices, you have to select them by using their unique hardware id. 
+The function :meth:`~can.detect_available_configs` can be used to generate a list of :class:`~can.BusABC` constructors
+(including the channel number and unique hardware ID number for the connected devices).
 
     .. testsetup:: ixxat
 
+        from unittest.mock import Mock
+        import can
+        assert hasattr(can, "detect_available_configs")
+        can.detect_available_configs = Mock(
+            "interface",
+            return_value=[{'interface': 'ixxat', 'channel': 0, 'unique_hardware_id': 'HW441489'}, {'interface': 'ixxat', 'channel': 0, 'unique_hardware_id': 'HW107422'}, {'interface': 'ixxat', 'channel': 1, 'unique_hardware_id': 'HW107422'}],
+        )
+
+    .. doctest:: ixxat
+
+        >>> import can 
+        >>> configs = can.detect_available_configs("ixxat")
+        >>> for config in configs:
+        ...     print(config)
+        {'interface': 'ixxat', 'channel': 0, 'unique_hardware_id': 'HW441489'}
+        {'interface': 'ixxat', 'channel': 0, 'unique_hardware_id': 'HW107422'}
+        {'interface': 'ixxat', 'channel': 1, 'unique_hardware_id': 'HW107422'}
+
+
+You may also get a list of all connected IXXAT devices using the function ``get_ixxat_hwids()`` as demonstrated below:
+
+    .. testsetup:: ixxat2
+
         from unittest.mock import Mock
         import can.interfaces.ixxat
         assert hasattr(can.interfaces.ixxat, "get_ixxat_hwids")
         can.interfaces.ixxat.get_ixxat_hwids = Mock(side_effect=lambda: ['HW441489', 'HW107422'])
 
-    .. doctest:: ixxat
+    .. doctest:: ixxat2
 
         >>> from can.interfaces.ixxat import get_ixxat_hwids
         >>> for hwid in get_ixxat_hwids():

test/test_interface_ixxat.py

@@ -51,6 +51,18 @@ def setUp(self):
             raise unittest.SkipTest("not available on this platform")
 
     def test_bus_creation(self):
+        try:
+            configs = can.detect_available_configs("ixxat")
+            if configs:
+                for interface_kwargs in configs:
+                    bus = can.Bus(**interface_kwargs)
+                    bus.shutdown()
+            else:
+                raise unittest.SkipTest("No adapters were detected")
+        except can.CanInterfaceNotImplementedError:
+            raise unittest.SkipTest("not available on this platform")
+
+    def test_bus_creation_incorrect_channel(self):
         # non-existent channel -> use arbitrary high value
         with self.assertRaises(can.CanInitializationError):
             can.Bus(interface="ixxat", channel=0xFFFF)