Add connection checks

cpdctrl/led_control_device/base.py

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

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

@@ -2,6 +2,10 @@ import serial
 
 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.
@@ -12,7 +16,7 @@ class LEDD1B(LedControlDevice):
     """
     def __init__(self, port="COM4"):    
         self.arduino = serial.Serial(port=port, baudrate=9600, timeout=.1)
-        # self._check_arduino_software()
+        self._check_arduino_software()
         
     def __del__(self):
         self.off()
@@ -25,9 +29,16 @@ class LEDD1B(LedControlDevice):
         """
         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}")
+            raise ConnectionError("Arduino did not return the expected output - does it have the correct software loaded?")
+
+    def test_connection(self) -> None:
+        try:
+            self._check_arduino_software()
+        except Exception as e:
+            log.error(f"Failed to query *IDN?:", e)
+            raise ConnectionError
 
     def _write(self, val):
         self.arduino.write(bytes(val, 'utf-8')) 

cpdctrl/voltage_measurement_device/base.py

@@ -8,6 +8,13 @@ Created on Tue Jan 21 16:19:01 2025
 """
 
 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:
@@ -22,6 +24,7 @@ class Keithley2700(VoltageMeasurementDevice):
         """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
@@ -40,6 +43,12 @@ class Keithley2700(VoltageMeasurementDevice):
             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:

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