Refactor to use DirectUSB programming and SCPI commands instead of Windows-only DLL

Author: samantha-ho

src/qcodes/instrument_drivers/Minicircuits/_minicircuits_usb_spdt.py

@@ -1,6 +1,10 @@
 import os
 from typing import TYPE_CHECKING
 
+import usb1
+from libusb1 import libusb_error
+from usb1 import USBContext, USBDevice, USBDeviceHandle, USBError
+
 # QCoDeS imports
 from qcodes.instrument_drivers.Minicircuits.Base_SPDT import (
     MiniCircuitsSPDTBase,
@@ -12,37 +16,68 @@
 
     from qcodes.instrument import InstrumentBaseKWArgs
 
-try:
-    import clr  # pyright: ignore[reportMissingTypeStubs,reportMissingImports]
-except ImportError:
-    raise ImportError(
-        """Module clr not found. Please obtain it by
-                         installing QCoDeS with the
-                         minicircuits_usb_spdt extra, e.g. by running
-                         pip install qcodes[minicircuits_usb_spdt]"""
-    )
+MINICIRCUITS_VENDOR_ID = 0x20CE
+RF_SWITCH_PRODUCT_ID = 0x0022
+
+
+def open_switch_with_sn(serial_number: str | None) -> USBDeviceHandle | None:
+    usb_context = USBContext()
+    if serial_number is not None:
+        device_iterator = usb_context.getDeviceIterator(
+            skip_on_error=True,
+        )
+        try:
+            for device in device_iterator:
+                if (
+                    device.getVendorID() == MINICIRCUITS_VENDOR_ID
+                    and device.getProductID() == RF_SWITCH_PRODUCT_ID
+                ):
+                    handle = device.open()
+                    if get_serial_number(handle) == serial_number:
+                        return handle
+                    device.close()  # Unsure what missing arguments are needed or what they do
+        finally:
+            device_iterator.close()
+    else:  # If no SN is provided, we can use the built-in function to return the first Minicircuits switch
+        return usb_context.openByVendorIDAndProductID(
+            MINICIRCUITS_VENDOR_ID, RF_SWITCH_PRODUCT_ID
+        )
+    return None
+
+
+def get_serial_number(handle: USBDeviceHandle) -> str:
+    handle.resetDevice()
+    handle.claimInterface(0)
+    cmd = [
+        41,
+    ]
+    cmd_array = bytearray([0] * 64)
+    cmd_array[0 : len(cmd)] = cmd
+    handle.interruptWrite(endpoint=1, data=cmd_array, timeout=50)
+    response = handle.interruptRead(endpoint=1, length=64, timeout=1000)
+    resp_length = response.index(bytearray([0]))
+    trimmed_response = response[1:resp_length]
+    return trimmed_response.decode("ascii")
 
 
 class MiniCircuitsUsbSPDTSwitchChannel(
     MiniCircuitsSPDTSwitchChannelBase["MiniCircuitsUsbSPDT"]
 ):
     def _set_switch(self, switch: int) -> None:
-        self.parent.switch.Set_Switch(self.channel_letter, switch - 1)
+        self.parent._query_scpi(f"set{self.channel_letter}={switch - 1}")
 
     def _get_switch(self) -> int:
-        status = self.parent.switch.GetSwitchesStatus(self._parent.address)[1]
-        return int(f"{status:04b}"[-1 - self.channel_number]) + 1
+        all_ports_state = int(self.parent._query_scpi("SWPORT?"))
+        bitmask = 2**self.channel_number
+        return int((all_ports_state & bitmask) >= 1) + 1
 
 
 class MiniCircuitsUsbSPDT(MiniCircuitsSPDTBase):
     CHANNEL_CLASS = MiniCircuitsUsbSPDTSwitchChannel
-    PATH_TO_DRIVER = r"mcl_RF_Switch_Controller64"
-    PATH_TO_DRIVER_45 = r"mcl_RF_Switch_Controller_NET45"
 
     def __init__(
         self,
         name: str,
-        driver_path: str | None = None,
         serial_number: str | None = None,
         **kwargs: "Unpack[InstrumentBaseKWArgs]",
     ):
@@ -51,59 +86,42 @@ def __init__(
 
         Args:
             name: the name of the instrument
-            driver_path: path to the dll
             serial_number: the serial number of the device
                 (printed on the sticker on the back side, without s/n)
             kwargs: kwargs to be passed to Instrument class.
 
         """
-        # import .net exception so we can catch it below
-        # we keep this import local so that the module can be imported
-        # without a working .net install
-        clr.AddReference("System.IO")
-        from System.IO import (  # pyright: ignore[reportMissingImports]  # noqa: PLC0415
-            FileNotFoundException,
-        )
-
         super().__init__(name, **kwargs)
-        if os.name != "nt":
-            raise ImportError("""This driver only works in Windows.""")
-        try:
-            if driver_path is None:
-                try:
-                    clr.AddReference(self.PATH_TO_DRIVER)
-                except FileNotFoundError:
-                    clr.AddReference(self.PATH_TO_DRIVER_45)
-            else:
-                clr.AddReference(driver_path)
-
-        except (ImportError, FileNotFoundException):
-            raise ImportError(
-                """Load of mcl_RF_Switch_Controller64.dll or mcl_RF_Switch_Controller_NET45.dll
-                not possible. Make sure the dll file is not blocked by Windows.
-                To unblock right-click the dll to open properties and check the 'unblock' checkmark
-                in the bottom. Check that your python installation is 64bit."""
-            )
-        try:
-            import mcl_RF_Switch_Controller64 as mw_driver  # pyright: ignore[reportMissingImports]# noqa: PLC0415
-        except ImportError:
-            import mcl_RF_Switch_Controller_NET45 as mw_driver  # pyright: ignore[reportMissingImports]# noqa: PLC0415
-
-        self.switch = mw_driver.USB_RF_SwitchBox()
+        self._handle: USBDeviceHandle | None = open_switch_with_sn(serial_number)
 
-        if not self.switch.Connect(serial_number):
-            raise RuntimeError("Could not connect to device")
-        self.address = self.switch.Get_Address()
-        self.serial_number = self.switch.Read_SN("")[1]
-        self.connect_message()
         self.add_channels()
+        self.connect_message()
+
+    @property
+    def handle(self) -> USBDeviceHandle:
+        if self._handle is None:
+            raise Exception  # TODO Make better
+        return self._handle
+
+    def _query_scpi(self, command: str) -> str:
+        cmd_bytes = bytearray([42, 58])  # Interrupt code for Send SCPI Command
+        cmd_bytes.extend(bytearray(command, "ascii"))
+        cmd_bytes.extend(bytearray(64 - len(cmd_bytes)))
+
+        self.handle.interruptWrite(endpoint=1, data=cmd_bytes, timeout=50)
+
+        response = self.handle.interruptRead(endpoint=1, length=64, timeout=1000)
+        resp_length = response.index(bytearray([0]))
+        trimmed_response = response[1:resp_length]
+
+        return trimmed_response.decode("ascii")
 
     def get_idn(self) -> dict[str, str | None]:
         # the arguments in those functions is the serial number or none if
         # there is only one switch.
-        fw = self.switch.GetFirmware()
-        MN = self.switch.Read_ModelName("")[1]
-        SN = self.switch.Read_SN("")[1]
+        fw = self._query_scpi("FIRMWARE?")
+        MN = self._query_scpi("MN?")
+        SN = self._query_scpi("SN?")
 
         id_dict = {"firmware": fw, "model": MN, "serial": SN, "vendor": "Mini-Circuits"}
         return id_dict