Fix cross-cutting issues: bool parsing, busy-wait polling, and misc

lakeshore/em_power_supply.py

@@ -228,14 +228,14 @@ def query(self, query_string, check_errors=True):
             error_response = response_list.pop()
             register = self.EMPowerSupplyStandardEventStatusRegister.from_integer(int(error_response))
             if register.command_error:
-                raise InstrumentException("Command Error: The instrument could not interpret the command due to a "
-                                          "syntax error, an unrecognized header, unrecognized terminator, or an "
-                                          "unsupported command.")
+                raise InstrumentException(f"Command Error for '{query_string}': The instrument could not interpret "
+                                          "the command due to a syntax error, an unrecognized header, unrecognized "
+                                          "terminator, or an unsupported command.")
             if register.execution_error:
-                raise InstrumentException("Execution Error: The instrument was instructed to do something not within "
-                                          "its capabilities.")
+                raise InstrumentException(f"Execution Error for '{query_string}': The instrument was instructed to "
+                                          "do something not within its capabilities.")
             if register.query_error:
-                raise InstrumentException("Query Error: The output queue is full.")
+                raise InstrumentException(f"Query Error for '{query_string}': The output queue is full.")
             response = ';'.join(response_list)
 
         return response

lakeshore/fast_hall_controller.py

@@ -1,7 +1,9 @@
 """Implements functionality unique to the Lake Shore M91 Fast Hall."""
 
 import json
+import time
 from .xip_instrument import XIPInstrument, RegisterBase, StatusByteRegister, StandardEventRegister
+from .generic_instrument import InstrumentException
 
 
 class FastHallOperationRegister(RegisterBase):
@@ -1017,11 +1019,13 @@ def get_resistivity_measurement_results(self):
 
         return measurement_results
 
-    def run_complete_contact_check_optimized(self, settings):
+    def run_complete_contact_check_optimized(self, settings, timeout=None):
         """Performs a contact check measurement and then returns the corresponding measurement results.
 
             Args:
                 settings(ContactCheckOptimizedParameters):
+                timeout (float):
+                    Optional timeout in seconds. If specified, raises InstrumentException if exceeded.
 
             Returns:
                 The measurement results as a dictionary.
@@ -1031,21 +1035,26 @@ def run_complete_contact_check_optimized(self, settings):
         self.start_contact_check_vdp_optimized(settings)
 
         # Loop until measurement has stopped running
+        start_time = time.time()
         while self.get_contact_check_running_status():
-            pass
+            if timeout is not None and (time.time() - start_time) > timeout:
+                raise InstrumentException(f"Measurement timed out after {timeout} seconds")
+            time.sleep(0.1)
 
         # Collect and return results
         results = self.get_contact_check_measurement_results()
         return results
 
-    def run_complete_contact_check_manual(self, settings, sample_type):
+    def run_complete_contact_check_manual(self, settings, sample_type, timeout=None):
         """Performs a manual contact check measurement and then returns the corresponding measurement results.
 
             Args:
                 settings (ContactCheckManualParameters):
                     Object with settings for FastHall link setup.
                 sample_type (str):
                     Indicates sample type. Options: "VDP" (Van der Pauw sample), or "HBAR" (Hall Bar sample).
+                timeout (float):
+                    Optional timeout in seconds. If specified, raises InstrumentException if exceeded.
 
             Returns:
                 The measurement results as a dictionary.
@@ -1062,19 +1071,24 @@ def run_complete_contact_check_manual(self, settings, sample_type):
                              'Sample type must be either "VDP" for a Van der Pauw sample or "HBAR for a Hall bar."')
 
         #  Loop until measurement has stopped running
+        start_time = time.time()
         while self.get_contact_check_running_status():
-            pass
+            if timeout is not None and (time.time() - start_time) > timeout:
+                raise InstrumentException(f"Measurement timed out after {timeout} seconds")
+            time.sleep(0.1)
 
         # Collect and return results
         results = self.get_contact_check_measurement_results()
         return results
 
-    def run_complete_fasthall_link(self, settings):
+    def run_complete_fasthall_link(self, settings, timeout=None):
         """Performs a FastHall Link measurement and then returns the corresponding measurement results.
 
             Args:
                 settings(FastHallLinkParameters):
                     Object with settings for FastHall link setup.
+                timeout (float):
+                    Optional timeout in seconds. If specified, raises InstrumentException if exceeded.
 
             Returns:
                 The measurement results as a dictionary.
@@ -1084,19 +1098,24 @@ def run_complete_fasthall_link(self, settings):
         self.start_fasthall_link_vdp(settings)
 
         # Loop until measurement has stopped running
+        start_time = time.time()
         while self.get_fasthall_running_status():
-            pass
+            if timeout is not None and (time.time() - start_time) > timeout:
+                raise InstrumentException(f"Measurement timed out after {timeout} seconds")
+            time.sleep(0.1)
 
         # Collect and return results
         results = self.get_fasthall_measurement_results()
         return results
 
-    def run_complete_fasthall_manual(self, settings):
+    def run_complete_fasthall_manual(self, settings, timeout=None):
         """Performs a manual FastHall measurement and then returns the corresponding measurement results.
 
             Args:
                 settings(FastHallManualParameters):
                     Object with settings for FastHall link setup.
+                timeout (float):
+                    Optional timeout in seconds. If specified, raises InstrumentException if exceeded.
             Returns:
                 The measurement results as a dictionary.
         """
@@ -1105,18 +1124,23 @@ def run_complete_fasthall_manual(self, settings):
         self.start_fasthall_vdp(settings)
 
         # Loop until measurement has stopped running
+        start_time = time.time()
         while self.get_fasthall_running_status():
-            pass
+            if timeout is not None and (time.time() - start_time) > timeout:
+                raise InstrumentException(f"Measurement timed out after {timeout} seconds")
+            time.sleep(0.1)
 
         # Collect and return results
         results = self.get_fasthall_measurement_results()
         return results
 
-    def run_complete_four_wire(self, settings):
+    def run_complete_four_wire(self, settings, timeout=None):
         """Performs a Four Wire measurement and then returns the corresponding measurement results.
 
             Args:
                 settings(FourWireParameters):
+                timeout (float):
+                    Optional timeout in seconds. If specified, raises InstrumentException if exceeded.
 
             Returns:
                 The measurement results as a dictionary.
@@ -1126,21 +1150,26 @@ def run_complete_four_wire(self, settings):
         self.start_four_wire(settings)
 
         # Loop until measurement has stopped running
+        start_time = time.time()
         while self.get_four_wire_running_status():
-            pass
+            if timeout is not None and (time.time() - start_time) > timeout:
+                raise InstrumentException(f"Measurement timed out after {timeout} seconds")
+            time.sleep(0.1)
 
         # Collect and return results
         results = self.get_four_wire_measurement_results()
         return results
 
-    def run_complete_dc_hall(self, settings, sample_type):
+    def run_complete_dc_hall(self, settings, sample_type, timeout=None):
         """Performs a DC Hall measurement and then returns the corresponding measurement results.
 
             Args:
                 settings(DCHallParameters):
                     Object with settings for FastHall link setup.
                 sample_type(str):
                     Indicates sample type. Options: "VDP" (Van der Pauw sample), or"HBAR" (Hall Bar sample).
+                timeout (float):
+                    Optional timeout in seconds. If specified, raises InstrumentException if exceeded.
 
             Returns:
                 The measurement results as a dictionary.
@@ -1156,19 +1185,25 @@ def run_complete_dc_hall(self, settings, sample_type):
                              'Sample type must be either "VDP" for a Van der Pauw sample or "HBAR for a Hall bar."')
 
         # Loop until measurement has stopped running or waiting
+        start_time = time.time()
         while self.get_dc_hall_running_status() or self.get_dc_hall_waiting_status():
             if self.get_dc_hall_waiting_status():
                 self.continue_dc_hall()
+            if timeout is not None and (time.time() - start_time) > timeout:
+                raise InstrumentException(f"Measurement timed out after {timeout} seconds")
+            time.sleep(0.1)
 
         # Collect and return results
         results = self.get_dc_hall_measurement_results()
         return results
 
-    def run_complete_resistivity_link(self, settings):
+    def run_complete_resistivity_link(self, settings, timeout=None):
         """Performs a resistivity link measurement and then returns the corresponding measurement results.
 
             Args:
                 settings(ResistivityLinkParameters):
+                timeout (float):
+                    Optional timeout in seconds. If specified, raises InstrumentException if exceeded.
 
             Returns:
                 The measurement results as a dictionary.
@@ -1178,21 +1213,26 @@ def run_complete_resistivity_link(self, settings):
         self.start_resistivity_link_vdp(settings)
 
         # Loop until measurement has stopped running
+        start_time = time.time()
         while self.get_resistivity_running_status():
-            pass
+            if timeout is not None and (time.time() - start_time) > timeout:
+                raise InstrumentException(f"Measurement timed out after {timeout} seconds")
+            time.sleep(0.1)
 
         # Collect and return results
         results = self.get_resistivity_measurement_results()
         return results
 
-    def run_complete_resistivity_manual(self, settings, sample_type):
+    def run_complete_resistivity_manual(self, settings, sample_type, timeout=None):
         """Performs a manual resistivity measurement and then returns the corresponding measurement results.
 
             Args:
                 settings(ResistivityManualParameters):
                     Object with settings for manual resistivity setup.
                 sample_type(str):
                     Indicates sample type. Options are: "VDP" (Van der Pauw sample), or "HBAR" (Hall Bar sample).
+                timeout (float):
+                    Optional timeout in seconds. If specified, raises InstrumentException if exceeded.
 
             Returns:
                 The measurement results as a dictionary.
@@ -1208,8 +1248,11 @@ def run_complete_resistivity_manual(self, settings, sample_type):
                              'Sample type must be either "VDP" for a Van der Pauw sample or "HBAR for a Hall bar."')
 
         # Loop until measurement has stopped running
+        start_time = time.time()
         while self.get_resistivity_running_status():
-            pass
+            if timeout is not None and (time.time() - start_time) > timeout:
+                raise InstrumentException(f"Measurement timed out after {timeout} seconds")
+            time.sleep(0.1)
 
         # Collect and return results
         results = self.get_resistivity_measurement_results()

lakeshore/generic_instrument.py

@@ -352,3 +352,11 @@ def _user_connection_query(self, query):
 
     def _get_identity(self):
         return self.query('*IDN?').split(',')
+
+    @staticmethod
+    def _validate_scpi_parameter(value, param_name):
+        """Validate a parameter value for SCPI safety."""
+        str_value = str(value)
+        if ';' in str_value or '\n' in str_value or '\r' in str_value:
+            raise ValueError(
+                f"Invalid characters in {param_name}: SCPI delimiters not allowed in parameter values")

lakeshore/model_240.py

@@ -145,7 +145,7 @@ def get_celsius_reading(self, channel):
                     Specifies channel (1-8).
 
         """
-        return self.query(f"CRDG? {channel}")
+        return float(self.query(f"CRDG? {channel}"))
 
     def set_factory_defaults(self):
         """Sets all configuration values to factory defaults and resets the instrument."""
@@ -169,7 +169,7 @@ def get_fahrenheit_reading(self, channel):
                     Specifies channel (1-8).
 
         """
-        return self.query(f"FRDG? {channel}")
+        return float(self.query(f"FRDG? {channel}"))
 
     def get_sensor_reading(self, input_channel):
         """Returns the sensor reading in the sensor's units.
@@ -286,6 +286,7 @@ def set_sensor_name(self, channel, name):
                     Specifies the name to associate with the sensor channel.
 
         """
+        self._validate_scpi_parameter(name, "name")
         self.command(f"INNAME {channel},{name}")
 
     def get_sensor_name(self, channel):
@@ -330,10 +331,10 @@ def get_input_parameter(self, channel):
         response = self.query(f"INTYPE? {channel}")
         data = response.split(",")
         input_parameter = Model240InputParameter(self.SensorTypes(int(data[0])),
-                                                 bool(data[1]),
-                                                 bool(data[3]),
+                                                 bool(int(data[1])),
+                                                 bool(int(data[3])),
                                                  self.Units(int(data[4])),
-                                                 bool(data[5]),
+                                                 bool(int(data[5])),
                                                  int(data[2]))
         return input_parameter
 
@@ -345,6 +346,7 @@ def set_modname(self, name):
                     Specifies the name or description to help identify the module.
 
         """
+        self._validate_scpi_parameter(name, "name")
         self.command(f"MODNAME {name}")
 
     def get_modname(self):
@@ -478,7 +480,7 @@ def get_sensor_units_channel_reading(self, channel):
                     Specifies which channel to query (1-8).
 
         """
-        return self.query(f"SRDG? {channel}")
+        return float(self.query(f"SRDG? {channel}"))
 
 
 __all__ = ['Model240', 'Model240CurveHeader', 'Model240InputParameter', 'Model240ProfiSlot']

lakeshore/model_372.py

@@ -971,7 +971,7 @@ def set_setpoint_ohms(self, output_channel, setpoint):
                     1: output 1 (warm up heater).
 
                 setpoint (float):
-                    Specifies the set-point the heater ramps to, in Kelvin.
+                    Specifies the set-point the heater ramps to, in Ohms.
 
         """
 
@@ -1009,9 +1009,9 @@ def get_setpoint_kelvin(self, output_channel):
         return float(self.query(f"SETP? {str(output_channel)}"))
 
     def get_setpoint_ohms(self, output_channel):
-        """Returns the set-point for the given output channel in kelvin.
+        """Returns the set-point for the given output channel in Ohms.
 
-            Changes the control input's preferred units to Kelvin as a result.
+            Changes the control input's preferred units to Ohms as a result.
 
             Args:
                 output_channel (int):

lakeshore/ssm_measure_module.py

@@ -1284,7 +1284,7 @@ def get_resistance_auto_range(self):
             bool: The state of resistance auto-range on the module.
         """
 
-        return bool(self.device.query(f"CALCulate:SENSe{self.module_number}:RESistance:RANGe:AUTO?"))
+        return bool(int(self.device.query(f"CALCulate:SENSe{self.module_number}:RESistance:RANGe:AUTO?")))
 
     def set_resistance_optimization_state(self, optimization_state):
         """Sets the state of resistance optimization on the module
@@ -1306,7 +1306,7 @@ def get_resistance_optimization_state(self):
             bool: The state of resistance optimization. True if optimizing for resistance, else False.
         """
 
-        return bool(self.device.query(f"CALCulate:SENSe{self.module_number}:RESistance:OPTimize?"))
+        return bool(int(self.device.query(f"CALCulate:SENSe{self.module_number}:RESistance:OPTimize?")))
 
     def set_resistance_observation_time_state(self, state):
         """Sets the state of the observation time on the module.

lakeshore/ssm_settings_profiles.py

@@ -107,7 +107,7 @@ def get_valid_for_restore(self, name):
         """
 
         response = self.device.query(f'PROFile:RESTore:VALid? "{name}"')
-        return bool(response)
+        return bool(int(response))
 
     def restore(self, name):
         """Restore a profile.

lakeshore/ssm_source_module.py

@@ -1089,7 +1089,7 @@ def set_disable_on_compliance(self, disable_on_compliance):
 
     def get_disable_on_compliance(self):
         """Returns the present state of disable on compliance."""
-        response = bool(self.device.query(f'SOURce{self.module_number}:DOCompliance?', check_errors=False))
+        response = bool(int(self.device.query(f'SOURce{self.module_number}:DOCompliance?', check_errors=False)))
         return response
 
     def set_current_output_limit_low(self, limit):
@@ -1483,10 +1483,14 @@ def _calculate_number_of_sweep_points(self, start_value, stop_value, sweep_spaci
         Returns:
             int: The number of sweep points.
         """
+        if start_value == stop_value:
+            return 1
         if sweep_spacing == 'LINEAR':
             step_size = 10 ** (math.floor(math.log10(abs(stop_value - start_value))) - 2)
             number_of_points = round(abs(stop_value - start_value) / step_size + 1)
         else:
+            if start_value == 0:
+                raise ValueError("Start value cannot be zero for logarithmic sweep")
             step_size = 10 ** (math.floor(math.log10(abs(stop_value / start_value))) - 2)
             number_of_points = round(abs(stop_value / start_value) / step_size + 1)
         return number_of_points