Add device enumeration to LEDD1B

cpdctrl/led_control_device/__init__.py

@@ -23,7 +23,8 @@ def list_devices() -> dict[str,list[str]]:
     }
     try:
         from .impl import thorlabs_ledd1b as th
-        devices[TYPENAME_LEDD1B] = ["Thorlabs LEDD1B"] #keithley2700.enumerate_devices()
+        # devices[TYPENAME_LEDD1B] = ["Thorlabs LEDD1B"] #keithley2700.enumerate_devices()
+        devices[TYPENAME_LEDD1B] = th.LEDD1B.enumerate_devices()
     except ImportError:
         pass
     try:
@@ -52,7 +53,10 @@ def connect_device(type_name: str, device_name: str) -> LedControlDevice:
     elif type_name == TYPENAME_LEDD1B:
         try:
             from .impl import thorlabs_ledd1b as th
-            return th.LEDD1B()
+            dev = th.LEDD1B.connect_device(device_name)
+            # make sure a potential software mismatch is being logged
+            dev.check_has_correct_software(no_error=True)
+            return dev
         except ImportError as e:
             raise ValueError(f"Arduino devices not available: {e}")
     elif type_name == TYPENAME_DC2200:

cpdctrl/led_control_device/base.py

@@ -6,7 +6,15 @@ Created on Tue Jan 21 16:26:13 2025
 """
 
 class LedControlDevice(ABC):
-    @abstractmethod 
+    @abstractmethod
+    def test_connection(self) -> None:
+        """
+        Verify that the device is still properly connected.
+        If not, raises ConnectionError
+        """
+        pass
+
+    @abstractmethod
     def on(self):
         """
         Set the led brightness to 100%

cpdctrl/led_control_device/impl/test.py

@@ -4,6 +4,9 @@ class TestLedControlDevice(LedControlDevice):
     def __init__(self):
         super().__init__()
 
+    def test_connection(self) -> None:
+        pass
+
     def on(self):
         pass
 

cpdctrl/led_control_device/impl/thorlabs_dc2200.py

@@ -36,6 +36,13 @@ class DC2200(LedControlDevice):
         self.off()
         self.instr.close()
 
+    def test_connection(self) -> None:
+        try:
+            res = self.instr.query('*IDN?')
+        except Exception as e:
+            log.error(f"Failed to query *IDN?:", e)
+            raise ConnectionError
+
     def on(self):
         self.instr.write(f'SOURCE1:CBRightness:BRIGhtness 100')
 

cpdctrl/led_control_device/impl/thorlabs_ledd1b.py

@@ -1,7 +1,12 @@
 import serial
+import serial.tools.list_ports
 
 from ..base import LedControlDevice
 
+import logging
+log = logging.getLogger(__name__)
+
+ARDUINO_SOFTWARE_VERSION_STRING = "Arduino Nano CPD 1"
 class LEDD1B(LedControlDevice):
     """
     Control a Thorlabs LEDD1B LED driver using an Arduino Nano.
@@ -10,27 +15,40 @@ class LEDD1B(LedControlDevice):
 
     Note: This currently has COM4 hardcoded
     """
-    def __init__(self, port="COM4"):    
-        self.arduino = serial.Serial(port=port, baudrate=9600, timeout=.1)
-        # self._check_arduino_software()
-        
+    def __init__(self, port="COM4"):
+        super().__init__()
+        self.arduino = serial.Serial(port=port, baudrate=9600, timeout=0.1)
+        # flush input
+        while self.arduino.read(1):
+            pass
+
     def __del__(self):
         self.off()
         self.arduino.close()
     
-    def _check_arduino_software(self):
+    def check_has_correct_software(self, no_error=False):
         """
         Run the identify command and raise an Exception
         if the Arduino does not reply with the expected output.
         """
         self._write('i')
         lines = self.read()
-        if len(lines) < 1 or not lines[-1].startswith(bytes("Arduino Nano CPD 1", "utf-8")):
-            print(lines)
-            raise Exception("Arduino did not return the expected output - does it have the correct software loaded?")
+        if len(lines) < 1 or not lines[-1].startswith(bytes(ARDUINO_SOFTWARE_VERSION_STRING, "utf-8")):
+            log.error(f"Arduino did not return the expected output - does it have the correct software loaded?\nExpected: '{ARDUINO_SOFTWARE_VERSION_STRING}'\nReceived:{lines}")
+            if not no_error:
+                raise ConnectionError("Arduino did not return the expected output - does it have the correct software loaded?")
+
+    def test_connection(self) -> None:
+        try:
+            self._write('i')
+            lines = self.read()
+        except Exception as e:
+            log.error(f"Failed to query 'i':", e)
+            raise ConnectionError
 
     def _write(self, val):
-        self.arduino.write(bytes(val, 'utf-8')) 
+        self.arduino.write(bytes(val, 'utf-8'))
+        self.arduino.flush()
         
     def read(self):
         data = self.arduino.readlines()
@@ -49,6 +67,23 @@ class LEDD1B(LedControlDevice):
     def __str__(self):
         return "Thorlabs LEDD1B"
 
+    @staticmethod
+    def enumerate_devices() -> list[str]:
+        devs = []
+        coms = serial.tools.list_ports.comports()
+        # Require a device with no product field and empty serial number
+        # Change these conditions to support different devices
+        for com in coms:
+            if com.product is not None: continue
+            if com.serial_number: continue
+            if com.manufacturer not in ["wch.cn"]: continue
+            devs.append(com.device)
+        return devs
+
+    @staticmethod
+    def connect_device(name: str):
+        return LEDD1B(port=name)
+
    
 if __name__ == '__main__':
     led = LEDD1B()

cpdctrl/led_script.py

@@ -306,8 +306,8 @@ class LedScript:
                     raise InvalidScriptUpdate(f"Cannot update row {idx+1} because with the new duration it would be skipped")
         self.script["dt"][idx] = dt
         self.script["led"][idx] = led
+        log.info(f"Updated row {idx+1} with values dt={dt}, led={led}")
         self._update_script_dtsums(idx)
-        log.critical("Updated!")
 
     def _update_script_dtsums(self, from_idx: int):
         for i in range(from_idx, self.script["dtsum"].shape[0]):

cpdctrl/power_switch_device/base.py

@@ -0,0 +1,17 @@
+from abc import ABC, abstractmethod
+
+class PowerSwitchDevice(ABC):
+    @abstractmethod
+    def status(self) -> bool:
+        pass
+    @abstractmethod   
+    def on(self):
+        pass
+
+    @abstractmethod
+    def off(self):
+        pass
+    
+    @abstractmethod
+    def __str__(self):
+        pass

cpdctrl/power_switch_device/impl/cleware_switch.py

@@ -0,0 +1,40 @@
+from ..base import PowerSwitchDevice
+import os
+from subprocess import Popen, PIPE
+
+class ClewareSwitch(PowerSwitchDevice):
+    def __init__(self, usbswitchcmd_exe_path: str):
+        super().__init__()
+        self.exe = usbswitchcmd_exe_path
+
+    def run_cmd(self, cmd) -> (int, str, str):
+        args = [self.exe, cmd]
+        p = Popen(args, stdout=PIPE, stderr=PIPE)
+        p.wait()
+        return p.returncode, p.stdout.read().decode("utf-8").strip("\n"), p.stderr.read().decode("utf-8").strip("\n")
+
+    def status(self) -> bool:
+        cmd = "-r"
+        ret, out, err = self.run_cmd(cmd)
+        print(f"status: ret: {ret}, out: '{out}', err: '{err}'")
+        if out == "1":
+            return True
+        elif out == "0":
+            return False
+        else:
+            raise RuntimeError(f"Failed to read USB switch status.\nCommand: '{cmd}'\nCode: {ret}, stdout: '{out}', stderr: '{err}'")
+
+    def on(self):
+        cmd = "1"
+        ret, out, err = self.run_cmd(cmd)
+        if ret != 1 or out or err:  # yes, it actually returns 1 when it turns on ._.
+            raise RuntimeError(f"Failed to turn on USB switch.\nCommand: '{cmd}'\nCode: {ret}, stdout: '{out}', stderr: '{err}'")
+
+    def off(self):
+        cmd = "0"
+        ret, out, err = self.run_cmd(cmd)
+        if ret != 0 or out or err:
+            raise RuntimeError(f"Failed to turn on USB switch.\nCommand: '{cmd}'\nCode: {ret}, stdout: '{out}', stderr: '{err}'")
+
+    def __str__(self):
+        return f"Cleware USB Switch"

cpdctrl/voltage_measurement_device/base.py

@@ -7,7 +7,14 @@ Created on Tue Jan 21 16:19:01 2025
 @author: Matthias Quintern
 """
 
-class VoltageMeasurementDevice(ABC):   
+class VoltageMeasurementDevice(ABC):
+    @abstractmethod
+    def test_connection(self) -> None:
+        """
+        Verify that the device is still properly connected.
+        If not, raises ConnectionError
+        """
+        pass
     # RUN COMMANDS ON THE DEVICE
     @abstractmethod
     def run(self, code, verbose=False):

cpdctrl/voltage_measurement_device/impl/keithley2700.py

@@ -7,11 +7,13 @@ from typing import Callable
 from ..base import VoltageMeasurementDevice
 from ...utility.visa import enumerate_devices
 
+import logging
+log = logging.getLogger(__name__)
+
 class Keithley2700(VoltageMeasurementDevice):
     """
     Wrapper class for the Keithley2700 SMU controlled via pyvisa
     """
-    
     def __init__(self, instr, check_front_switch=True):
         self.instr = instr
         if check_front_switch:
@@ -21,7 +23,8 @@ class Keithley2700(VoltageMeasurementDevice):
     def __del__(self):
         """Properly close the instrument connection"""
         self.instr.close()
-        
+
+
     def _check_front_input_selected(self):
         """
         Make sure the front switch selecting the inputs selects the FRONT inputs
@@ -39,8 +42,14 @@ class Keithley2700(VoltageMeasurementDevice):
         if switch != "1":
             raise Exception("The Keithley's INPUT switch must select the [F]ront inputs")
         return switch
-    
-        
+
+    def test_connection(self) -> None:
+        try:
+            res = self.query("*IDN?")
+        except Exception as e:
+            log.error(f"Failed to query *IDN?:", e)
+            raise ConnectionError
+
     def query(self, query):
         try:
             return self.instr.query(query).strip("\n")

cpdctrl/voltage_measurement_device/impl/test.py

@@ -10,6 +10,9 @@ class TestVoltageMeasurementDevice(VoltageMeasurementDevice):
         self.frequency = frequency
         self.t0 = now()
 
+    def test_connection(self) -> None:
+        pass
+
     # RUN COMMANDS ON THE DEVICE
     def run(self, code, verbose=False):
         pass