combine _scan_gs_usb_devices() and _find_gs_usb_device() (@zariiii9003)

Author: BenGardiner

can/interfaces/gs_usb.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Any
 
 import usb
 from gs_usb.constants import CAN_EFF_FLAG, CAN_ERR_FLAG, CAN_MAX_DLC, CAN_RTR_FLAG
@@ -12,53 +13,49 @@
 logger = logging.getLogger(__name__)
 
 
-def _scan_gs_usb_devices() -> list[GsUsb]:
-    """Scan for gs_usb devices using auto-detected backend.
+def _find_gs_usb_devices(
+    bus: int | None = None, address: int | None = None
+) -> list[GsUsb]:
+    """Find gs_usb devices using auto-detected backend.
 
     Unlike :meth:`GsUsb.scan`, this does not force the ``libusb1`` backend,
     allowing ``pyusb`` to auto-detect the best available backend. This enables
     support for WinUSB on Windows in addition to libusbK.
+
+    :param bus: number of the bus that the device is connected to
+    :param address: address of the device on the bus it is connected to
+    :return: a list of found GsUsb devices
     """
+    kwargs = {}
+    if bus is not None:
+        kwargs["bus"] = bus
+    if address is not None:
+        kwargs["address"] = address
+
     return [
         GsUsb(dev)
         for dev in (
             usb.core.find(
                 find_all=True,
                 custom_match=GsUsb.is_gs_usb_device,
+                **kwargs,
             )
             or []
         )
     ]
 
 
-def _find_gs_usb_device(bus: int, address: int) -> GsUsb | None:
-    """Find a specific gs_usb device using auto-detected backend.
-
-    Unlike :meth:`GsUsb.find`, this does not force the ``libusb1`` backend,
-    allowing ``pyusb`` to auto-detect the best available backend. This enables
-    support for WinUSB on Windows in addition to libusbK.
-    """
-    dev = usb.core.find(
-        custom_match=GsUsb.is_gs_usb_device,
-        bus=bus,
-        address=address,
-    )
-    if dev:
-        return GsUsb(dev)
-    return None
-
-
 class GsUsbBus(can.BusABC):
     def __init__(
         self,
-        channel,
+        channel: can.typechecking.Channel,
         bitrate: int = 500_000,
-        index=None,
-        bus=None,
-        address=None,
-        can_filters=None,
-        **kwargs,
-    ):
+        index: int | None = None,
+        bus: int | None = None,
+        address: int | None = None,
+        can_filters: can.typechecking.CanFilters | None = None,
+        **kwargs: Any,
+    ) -> None:
         """
         :param channel: usb device name
         :param index: device number if using automatic scan, starting from 0.
@@ -74,24 +71,35 @@ def __init__(
             )
 
         if index is None and address is None and bus is None:
-            index = channel
+            _index: Any = channel
+        else:
+            _index = index
 
-        self._index = None
-        if index is not None:
-            devs = _scan_gs_usb_devices()
-            if len(devs) <= index:
+        self._index: int | None = None
+        if _index is not None:
+            if not isinstance(_index, int):
+                try:
+                    _index = int(_index)
+                except (ValueError, TypeError):
+                    raise CanInitializationError(
+                        f"index must be an integer, but got {type(_index).__name__} ({_index})"
+                    ) from None
+
+            devs = _find_gs_usb_devices()
+            if len(devs) <= _index:
                 raise CanInitializationError(
-                    f"Cannot find device {index}. Devices found: {len(devs)}"
+                    f"Cannot find device {_index}. Devices found: {len(devs)}"
                 )
-            gs_usb = devs[index]
-            self._index = index
+            gs_usb = devs[_index]
+            self._index = _index
         else:
-            gs_usb = _find_gs_usb_device(bus=bus, address=address)
-            if not gs_usb:
+            devs = _find_gs_usb_devices(bus=bus, address=address)
+            if not devs:
                 raise CanInitializationError(f"Cannot find device {channel}")
+            gs_usb = devs[0]
 
         self.gs_usb = gs_usb
-        self.channel_info = channel
+        self.channel_info = str(channel)
         self._can_protocol = can.CanProtocol.CAN_20
 
         bit_timing = can.BitTiming.from_sample_point(
@@ -116,7 +124,7 @@ def __init__(
             **kwargs,
         )
 
-    def send(self, msg: can.Message, timeout: float | None = None):
+    def send(self, msg: can.Message, timeout: float | None = None) -> None:
         """Transmit a message to the CAN bus.
 
         :param Message msg: A message object.
@@ -200,9 +208,10 @@ def shutdown(self):
 
         self.gs_usb.stop()
         if self._index is not None:
-            # Avoid errors on subsequent __init() by repeating the .scan() and .start() that would otherwise fail
-            # the next time the device is opened in __init__()
-            devs = _scan_gs_usb_devices()
+            # Avoid errors on subsequent __init() by repeating the .scan() and
+            # .start() that would otherwise fail the next time the device is
+            # opened in __init__()
+            devs = _find_gs_usb_devices()
             if self._index < len(devs):
                 gs_usb = devs[self._index]
                 try:

test/test_interface_gs_usb.py

@@ -6,18 +6,17 @@
 
 from can.interfaces.gs_usb import (
     GsUsbBus,
-    _find_gs_usb_device,
-    _scan_gs_usb_devices,
+    _find_gs_usb_devices,
 )
 
 
 @patch("can.interfaces.gs_usb.usb.core.find")
-def test_scan_does_not_force_backend(mock_find):
-    """Verify that _scan_gs_usb_devices does not pass a backend argument,
+def test_find_devices_does_not_force_backend(mock_find):
+    """Verify that _find_gs_usb_devices does not pass a backend argument,
     allowing pyusb to auto-detect the best available backend (WinUSB, libusbK, etc.)."""
     mock_find.return_value = []
 
-    _scan_gs_usb_devices()
+    _find_gs_usb_devices()
 
     mock_find.assert_called_once()
     call_kwargs = mock_find.call_args[1]
@@ -28,12 +27,11 @@ def test_scan_does_not_force_backend(mock_find):
 
 
 @patch("can.interfaces.gs_usb.usb.core.find")
-def test_find_does_not_force_backend(mock_find):
-    """Verify that _find_gs_usb_device does not pass a backend argument,
-    allowing pyusb to auto-detect the best available backend (WinUSB, libusbK, etc.)."""
-    mock_find.return_value = None
+def test_find_devices_with_args_does_not_force_backend(mock_find):
+    """Verify that _find_gs_usb_devices with bus/address does not pass a backend argument."""
+    mock_find.return_value = []
 
-    _find_gs_usb_device(bus=1, address=2)
+    _find_gs_usb_devices(bus=1, address=2)
 
     mock_find.assert_called_once()
     call_kwargs = mock_find.call_args[1]
@@ -42,49 +40,28 @@ def test_find_does_not_force_backend(mock_find):
     ), "backend should not be specified so pyusb can auto-detect"
     assert call_kwargs["bus"] == 1
     assert call_kwargs["address"] == 2
+    assert call_kwargs["find_all"] is True
 
 
 @patch("can.interfaces.gs_usb.usb.core.find")
-def test_scan_returns_gs_usb_devices(mock_find):
-    """Verify that _scan_gs_usb_devices wraps found USB devices in GsUsb objects."""
+def test_find_devices_returns_gs_usb_devices(mock_find):
+    """Verify that _find_gs_usb_devices wraps found USB devices in GsUsb objects."""
     mock_dev1 = MagicMock()
     mock_dev2 = MagicMock()
     mock_find.return_value = [mock_dev1, mock_dev2]
 
-    devices = _scan_gs_usb_devices()
+    devices = _find_gs_usb_devices()
 
     assert len(devices) == 2
     assert devices[0].gs_usb is mock_dev1
     assert devices[1].gs_usb is mock_dev2
 
 
 @patch("can.interfaces.gs_usb.usb.core.find")
-def test_find_returns_gs_usb_device(mock_find):
-    """Verify that _find_gs_usb_device wraps the found USB device in a GsUsb object."""
-    mock_dev = MagicMock()
-    mock_find.return_value = mock_dev
-
-    device = _find_gs_usb_device(bus=1, address=2)
-
-    assert device is not None
-    assert device.gs_usb is mock_dev
-
-
-@patch("can.interfaces.gs_usb.usb.core.find")
-def test_find_returns_none_when_no_device(mock_find):
-    """Verify that _find_gs_usb_device returns None when no device is found."""
-    mock_find.return_value = None
-
-    device = _find_gs_usb_device(bus=1, address=2)
-
-    assert device is None
-
-
-@patch("can.interfaces.gs_usb.usb.core.find")
-def test_scan_returns_empty_list_when_no_devices(mock_find):
-    """Verify that _scan_gs_usb_devices returns an empty list when no devices are found."""
+def test_find_devices_returns_empty_list_when_no_devices(mock_find):
+    """Verify that _find_gs_usb_devices returns an empty list when no devices are found."""
     mock_find.return_value = []
 
-    devices = _scan_gs_usb_devices()
+    devices = _find_gs_usb_devices()
 
     assert devices == []