From 184a3d72b6c2e026b873012d8a83e9707720bab5 Mon Sep 17 00:00:00 2001 From: Justin Fichtner Date: Tue, 17 Feb 2026 19:17:52 -0500 Subject: [PATCH 1/5] Fix bool("0") always returning True and boolean format issues bool() on a non-empty string like "0" always returns True. Wrap with int() first so "0" correctly becomes False. Also fix set_qualifier_latching_setting sending "True"/"False" instead of "1"/"0", and get_qualifier_latching_setting returning a raw string instead of bool. Affected locations: - model_240.py: get_input_parameter (3 fields) - ssm_source_module.py: get_disable_on_compliance - ssm_measure_module.py: get_resistance_auto_range, get_resistance_optimization_state - ssm_settings_profiles.py: get_valid_for_restore - teslameter.py: get/set_qualifier_latching_setting Co-Authored-By: Claude Opus 4.6 --- lakeshore/model_240.py | 6 +++--- lakeshore/ssm_measure_module.py | 4 ++-- lakeshore/ssm_settings_profiles.py | 2 +- lakeshore/ssm_source_module.py | 2 +- lakeshore/teslameter.py | 4 ++-- 5 files changed, 9 insertions(+), 9 deletions(-) diff --git a/lakeshore/model_240.py b/lakeshore/model_240.py index 5f0dbf4..763835b 100644 --- a/lakeshore/model_240.py +++ b/lakeshore/model_240.py @@ -330,10 +330,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 diff --git a/lakeshore/ssm_measure_module.py b/lakeshore/ssm_measure_module.py index a3dfe83..a927c86 100644 --- a/lakeshore/ssm_measure_module.py +++ b/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. diff --git a/lakeshore/ssm_settings_profiles.py b/lakeshore/ssm_settings_profiles.py index 505cb59..fe1587f 100644 --- a/lakeshore/ssm_settings_profiles.py +++ b/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. diff --git a/lakeshore/ssm_source_module.py b/lakeshore/ssm_source_module.py index 9b72915..76efec8 100644 --- a/lakeshore/ssm_source_module.py +++ b/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): diff --git a/lakeshore/teslameter.py b/lakeshore/teslameter.py index bb9c21e..2c38d26 100644 --- a/lakeshore/teslameter.py +++ b/lakeshore/teslameter.py @@ -637,7 +637,7 @@ def disable_qualifier_latching(self): @requires_firmware_version("1.6.2019092002") def get_qualifier_latching_setting(self): """Returns whether the qualifier latches.""" - return self.query("SENSE:QUALIFIER:LATCH?") + return bool(int(self.query("SENSE:QUALIFIER:LATCH?"))) @requires_firmware_version("1.6.2019092002") def set_qualifier_latching_setting(self, latching): @@ -647,7 +647,7 @@ def set_qualifier_latching_setting(self, latching): latching (bool): Determines whether the qualifier latches. """ - self.command(f"SENSE:QUALIFIER:LATCH {str(latching)}") + self.command(f"SENSE:QUALIFIER:LATCH {int(latching)}") @requires_firmware_version("1.6.2019092002") def reset_qualifier_latch(self): From 40d9e8c3d8c745b1fad909f9f8e6d6179459cd4c Mon Sep 17 00:00:00 2001 From: Justin Fichtner Date: Tue, 17 Feb 2026 19:20:05 -0500 Subject: [PATCH 2/5] Add sleep and timeout to busy-wait polling loops All run_complete_* methods in fast_hall_controller.py spin at max CPU speed with no sleep, and block forever if a measurement hangs. Add time.sleep(0.1) to each loop and an optional timeout parameter that raises InstrumentException when exceeded. Also add sleep to the Teslameter stream_buffered_data polling loop when no data is available. Co-Authored-By: Claude Opus 4.6 --- lakeshore/fast_hall_controller.py | 73 +++++++++++++++++++++++------- lakeshore/teslameter.py | 74 ++++++++++++++++--------------- 2 files changed, 97 insertions(+), 50 deletions(-) diff --git a/lakeshore/fast_hall_controller.py b/lakeshore/fast_hall_controller.py index 7f2dd7a..d51ab5f 100644 --- a/lakeshore/fast_hall_controller.py +++ b/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,14 +1035,17 @@ 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: @@ -1046,6 +1053,8 @@ def run_complete_contact_check_manual(self, settings, sample_type): 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,14 +1150,17 @@ 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: @@ -1141,6 +1168,8 @@ def run_complete_dc_hall(self, settings, sample_type): 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,14 +1213,17 @@ 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: @@ -1193,6 +1231,8 @@ def run_complete_resistivity_manual(self, settings, sample_type): 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() diff --git a/lakeshore/teslameter.py b/lakeshore/teslameter.py index 2c38d26..bb67109 100644 --- a/lakeshore/teslameter.py +++ b/lakeshore/teslameter.py @@ -1,5 +1,6 @@ """Implements functionality unique to the Lake Shore F41 and F71 Teslameters.""" +import time from collections import namedtuple from datetime import datetime @@ -137,41 +138,44 @@ def stream_buffered_data(self, length_of_time_in_seconds, sample_rate_in_ms): response = self.query('FETC:BUFF:DC?', check_errors=False).strip('"') # Ignore the response if it contains no data - if ';' in response: - # Split apart the response into single data points. - data_points = response.rstrip(';').split(';') - - for point in data_points: - # Split the data point along the delimiter. - point_data = point.split(',') - - # Convert the returned values from strings to appropriate types - for count, _ in enumerate(point_data): - if count == 0: - point_data[count] = iso8601.parse_date(point_data[count]) - elif count == len(point_data) - 1: - point_data[count] = int(point_data[count]) - else: - point_data[count] = float(point_data[count]) - - # If the instrument does not have a field control option, insert zero as the control set point. - if len(point_data) == 6: - input_state = point_data.pop() - point_data.append(0.0) - point_data.append(input_state) - - # Count how many samples have been collected and calculate the elapsed time. - number_of_samples += 1 - elapsed_time_in_seconds = sample_rate_in_ms * number_of_samples / 1000 - - # If we have exceeded the requested number of samples, end the stream. - if number_of_samples > total_number_of_samples: - break - - # Unpack the parsed point into a namedtuple and append it to the list - new_point = DataPoint(elapsed_time_in_seconds, *point_data) - - yield new_point + if ';' not in response: + time.sleep(0.1) + continue + + # Split apart the response into single data points. + data_points = response.rstrip(';').split(';') + + for point in data_points: + # Split the data point along the delimiter. + point_data = point.split(',') + + # Convert the returned values from strings to appropriate types + for count, _ in enumerate(point_data): + if count == 0: + point_data[count] = iso8601.parse_date(point_data[count]) + elif count == len(point_data) - 1: + point_data[count] = int(point_data[count]) + else: + point_data[count] = float(point_data[count]) + + # If the instrument does not have a field control option, insert zero as the control set point. + if len(point_data) == 6: + input_state = point_data.pop() + point_data.append(0.0) + point_data.append(input_state) + + # Count how many samples have been collected and calculate the elapsed time. + number_of_samples += 1 + elapsed_time_in_seconds = sample_rate_in_ms * number_of_samples / 1000 + + # If we have exceeded the requested number of samples, end the stream. + if number_of_samples > total_number_of_samples: + break + + # Unpack the parsed point into a namedtuple and append it to the list + new_point = DataPoint(elapsed_time_in_seconds, *point_data) + + yield new_point @requires_firmware_version('1.1.2018091003') def get_buffered_data_points(self, length_of_time_in_seconds, sample_rate_in_ms): From 22f27ae08b6f39c49ca599a181a1a23aacf43e74 Mon Sep 17 00:00:00 2001 From: Justin Fichtner Date: Tue, 17 Feb 2026 19:21:55 -0500 Subject: [PATCH 3/5] Fix SSM sweep calculation, add SCPI validation helper, and misc fixes - Guard _calculate_number_of_sweep_points against start == stop (previously caused math.log10(0) ValueError) and start_value == 0 in logarithmic mode (division by zero) - Add _validate_scpi_parameter static method to GenericInstrument for rejecting semicolons and newlines in user-supplied parameter values - Fix Model 372 docstring copy-paste errors: set_setpoint_ohms and get_setpoint_ohms incorrectly said "Kelvin" instead of "Ohms" - Make Model 240 get_celsius_reading, get_fahrenheit_reading, and get_sensor_units_channel_reading return float instead of raw string, consistent with get_kelvin_reading - Add command context to InstrumentException messages in EM power supply Co-Authored-By: Claude Opus 4.6 --- lakeshore/em_power_supply.py | 12 ++++++------ lakeshore/generic_instrument.py | 8 ++++++++ lakeshore/model_240.py | 6 +++--- lakeshore/model_372.py | 6 +++--- lakeshore/ssm_source_module.py | 4 ++++ tests/test_240.py | 6 +++--- 6 files changed, 27 insertions(+), 15 deletions(-) diff --git a/lakeshore/em_power_supply.py b/lakeshore/em_power_supply.py index 56e167f..dd278c1 100644 --- a/lakeshore/em_power_supply.py +++ b/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 diff --git a/lakeshore/generic_instrument.py b/lakeshore/generic_instrument.py index 7454f61..7911dcf 100644 --- a/lakeshore/generic_instrument.py +++ b/lakeshore/generic_instrument.py @@ -85,6 +85,14 @@ class GenericInstrument: vid_pid = [] logger = logging.getLogger(__name__) + @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") + def __init__(self, serial_number, com_port, baud_rate, data_bits, stop_bits, parity, flow_control, handshaking, timeout, ip_address, tcp_port, connection=None): # Initialize values common to all instruments diff --git a/lakeshore/model_240.py b/lakeshore/model_240.py index 763835b..b9755c7 100644 --- a/lakeshore/model_240.py +++ b/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. @@ -478,7 +478,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'] diff --git a/lakeshore/model_372.py b/lakeshore/model_372.py index e032a34..e46f803 100644 --- a/lakeshore/model_372.py +++ b/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): diff --git a/lakeshore/ssm_source_module.py b/lakeshore/ssm_source_module.py index 76efec8..6b2c260 100644 --- a/lakeshore/ssm_source_module.py +++ b/lakeshore/ssm_source_module.py @@ -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 diff --git a/tests/test_240.py b/tests/test_240.py index 4d1772b..df4c233 100644 --- a/tests/test_240.py +++ b/tests/test_240.py @@ -13,13 +13,13 @@ def test_get_kelvin_reading(self): def test_get_celsius_reading(self): self.fake_connection.setup_response('123') response = self.dut.get_celsius_reading("1") - self.assertEqual(response, '123') + self.assertEqual(response, 123.0) self.assertIn("CRDG? 1", self.fake_connection.get_outgoing_message()) def test_get_fahrenheit_reading(self): self.fake_connection.setup_response('123') response = self.dut.get_fahrenheit_reading("1") - self.assertEqual(response, '123') + self.assertEqual(response, 123.0) self.assertIn("FRDG? 1", self.fake_connection.get_outgoing_message()) @@ -74,7 +74,7 @@ def test_get_filter(self): def test_get_sensor_units_channel_reading(self): self.fake_connection.setup_response('12345') response = self.dut.get_sensor_units_channel_reading("1") - self.assertEqual(response, '12345') + self.assertEqual(response, 12345.0) self.assertIn("SRDG? 1", self.fake_connection.get_outgoing_message()) def test_get_channel_reading_status(self): From be4009f4d2a007ef350414adfd8bf58eaf02db88 Mon Sep 17 00:00:00 2001 From: Justin Fichtner Date: Tue, 17 Feb 2026 19:32:10 -0500 Subject: [PATCH 4/5] Add tests for review findings and wire up SCPI parameter validation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address all review findings with comprehensive test coverage: - Add test for bool "0" → False in Model 240 get_input_parameter - Add timeout tests for all 8 run_complete_* methods in FastHall - Add normal completion tests for run_complete_* methods - Add sweep edge case tests: start==stop, zero start in log mode - Add SCPI parameter validation tests and wire _validate_scpi_parameter into Model 240 set_sensor_name and set_modname - Add EM power supply error message context tests verifying command string appears in Command Error, Execution Error, and Query Error Co-Authored-By: Claude Opus 4.6 --- lakeshore/model_240.py | 2 + tests/test_240.py | 38 ++++++++++++++ tests/test_em_power_supply.py | 28 +++++++++++ tests/test_fast_hall.py | 95 +++++++++++++++++++++++++++++++++++ tests/test_ssm_system.py | 25 +++++++++ 5 files changed, 188 insertions(+) diff --git a/lakeshore/model_240.py b/lakeshore/model_240.py index b9755c7..e0ce4a4 100644 --- a/lakeshore/model_240.py +++ b/lakeshore/model_240.py @@ -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): @@ -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): diff --git a/tests/test_240.py b/tests/test_240.py index df4c233..ce9571a 100644 --- a/tests/test_240.py +++ b/tests/test_240.py @@ -93,6 +93,44 @@ def test_get_input_parameter(self): self.assertIn("INTYPE? 1", self.fake_connection.get_outgoing_message()) + def test_get_input_parameter_zeros_are_false(self): + self.fake_connection.setup_response('2,0,2,0,2,0') + response = self.dut.get_input_parameter(1) + + self.assertEqual(response.auto_range_enable, False) + self.assertEqual(response.current_reversal_enable, False) + self.assertEqual(response.input_enable, False) + + self.assertIn("INTYPE? 1", self.fake_connection.get_outgoing_message()) + + +class TestScpiParameterValidation(TestWithFakeModel240): + + def test_reject_semicolon_in_sensor_name(self): + with self.assertRaises(ValueError) as ctx: + self.dut.set_sensor_name(1, "evil;DFLT 99") + self.assertIn("SCPI delimiters", str(ctx.exception)) + + def test_reject_newline_in_sensor_name(self): + with self.assertRaises(ValueError) as ctx: + self.dut.set_sensor_name(1, "evil\nDFLT 99") + self.assertIn("SCPI delimiters", str(ctx.exception)) + + def test_reject_semicolon_in_modname(self): + with self.assertRaises(ValueError) as ctx: + self.dut.set_modname("evil;DFLT 99") + self.assertIn("SCPI delimiters", str(ctx.exception)) + + def test_valid_sensor_name_accepted(self): + self.fake_connection.setup_response('No error') + self.dut.set_sensor_name(1, "MySensor") + self.assertIn('INNAME 1,MySensor', self.fake_connection.get_outgoing_message()) + + def test_valid_modname_accepted(self): + self.fake_connection.setup_response('No error') + self.dut.set_modname("MyModule") + self.assertIn('MODNAME MyModule', self.fake_connection.get_outgoing_message()) + class TestBasicCommandMethods(TestWithFakeModel240): diff --git a/tests/test_em_power_supply.py b/tests/test_em_power_supply.py index 5509e3b..5153083 100644 --- a/tests/test_em_power_supply.py +++ b/tests/test_em_power_supply.py @@ -1,6 +1,34 @@ +from lakeshore.generic_instrument import InstrumentException from tests.utils import TestWithFakeEMPowerSupply +class TestErrorMessageContext(TestWithFakeEMPowerSupply): + + def test_command_error_includes_query_string(self): + # Standard Event Register bit 5 (0x20 = 32) = command error + self.fake_connection.setup_response("0; 32") + with self.assertRaises(InstrumentException) as ctx: + self.dut.get_current() + self.assertIn("SETI?", str(ctx.exception)) + self.assertIn("Command Error", str(ctx.exception)) + + def test_execution_error_includes_query_string(self): + # Standard Event Register bit 4 (0x10 = 16) = execution error + self.fake_connection.setup_response("0; 16") + with self.assertRaises(InstrumentException) as ctx: + self.dut.get_current() + self.assertIn("SETI?", str(ctx.exception)) + self.assertIn("Execution Error", str(ctx.exception)) + + def test_query_error_includes_query_string(self): + # Standard Event Register bit 2 (0x04 = 4) = query error + self.fake_connection.setup_response("0; 4") + with self.assertRaises(InstrumentException) as ctx: + self.dut.get_current() + self.assertIn("SETI?", str(ctx.exception)) + self.assertIn("Query Error", str(ctx.exception)) + + class TestCurrentSettings(TestWithFakeEMPowerSupply): def test_set_limits(self): diff --git a/tests/test_fast_hall.py b/tests/test_fast_hall.py index ad5274b..bf9efc6 100644 --- a/tests/test_fast_hall.py +++ b/tests/test_fast_hall.py @@ -1,4 +1,6 @@ +from unittest.mock import patch from tests.utils import TestWithFakeFastHall +from lakeshore.generic_instrument import InstrumentException from lakeshore.fast_hall_controller import ContactCheckManualParameters, ContactCheckOptimizedParameters, \ FastHallManualParameters, FastHallLinkParameters, FourWireParameters, DCHallParameters, \ ResistivityManualParameters, ResistivityLinkParameters @@ -338,3 +340,96 @@ def test_run_resistivity_hbar_non_default(self): self.dut.start_resistivity_hbar(parameters) self.assertIn('RESISTIVITY:HBAR:START CURRENT,0.01,0.02,0.03,4,5,0.006,0.007,800,9,0.001,11', self.fake_connection.get_outgoing_message()) + + +class TestRunCompleteTimeout(TestWithFakeFastHall): + def test_contact_check_optimized_timeout(self): + # Setup: start command succeeds, status returns running (True) + self.fake_connection.setup_response('No error') # start command + self.fake_connection.setup_response('1;No error') # running status = True + parameters = ContactCheckOptimizedParameters() + with self.assertRaises(InstrumentException) as ctx: + self.dut.run_complete_contact_check_optimized(parameters, timeout=0) + self.assertIn("timed out", str(ctx.exception)) + + def test_contact_check_manual_timeout(self): + self.fake_connection.setup_response('No error') + self.fake_connection.setup_response('1;No error') + parameters = ContactCheckManualParameters(excitation_type='VOLTAGE', excitation_start_value=-5, + excitation_end_value=5, compliance_limit=10e-3, number_of_points=25) + with self.assertRaises(InstrumentException) as ctx: + self.dut.run_complete_contact_check_manual(parameters, "VDP", timeout=0) + self.assertIn("timed out", str(ctx.exception)) + + def test_fasthall_link_timeout(self): + self.fake_connection.setup_response('No error') + self.fake_connection.setup_response('1;No error') + parameters = FastHallLinkParameters(user_defined_field=0.5) + with self.assertRaises(InstrumentException) as ctx: + self.dut.run_complete_fasthall_link(parameters, timeout=0) + self.assertIn("timed out", str(ctx.exception)) + + def test_fasthall_manual_timeout(self): + self.fake_connection.setup_response('No error') + self.fake_connection.setup_response('1;No error') + parameters = FastHallManualParameters(excitation_type='VOLTAGE', excitation_value=1, compliance_limit=10e-6, + user_defined_field=0.5) + with self.assertRaises(InstrumentException) as ctx: + self.dut.run_complete_fasthall_manual(parameters, timeout=0) + self.assertIn("timed out", str(ctx.exception)) + + def test_four_wire_timeout(self): + self.fake_connection.setup_response('No error') + self.fake_connection.setup_response('1;No error') + parameters = FourWireParameters(contact_point1=1, contact_point2=2, contact_point3=3, contact_point4=4, + excitation_type='CURRENT', excitation_value=10e-3, compliance_limit=1.5) + with self.assertRaises(InstrumentException) as ctx: + self.dut.run_complete_four_wire(parameters, timeout=0) + self.assertIn("timed out", str(ctx.exception)) + + def test_dc_hall_timeout(self): + self.fake_connection.setup_response('No error') + # dc_hall checks both running and waiting status + self.fake_connection.setup_response('1;No error') # running = True + self.fake_connection.setup_response('0;No error') # waiting = False + parameters = DCHallParameters(excitation_type='CURRENT', excitation_value=10e-3, compliance_limit=1.5, + averaging_samples=100, user_defined_field=0.5) + with self.assertRaises(InstrumentException) as ctx: + self.dut.run_complete_dc_hall(parameters, "VDP", timeout=0) + self.assertIn("timed out", str(ctx.exception)) + + def test_resistivity_link_timeout(self): + self.fake_connection.setup_response('No error') + self.fake_connection.setup_response('1;No error') + parameters = ResistivityLinkParameters() + with self.assertRaises(InstrumentException) as ctx: + self.dut.run_complete_resistivity_link(parameters, timeout=0) + self.assertIn("timed out", str(ctx.exception)) + + def test_resistivity_manual_timeout(self): + self.fake_connection.setup_response('No error') + self.fake_connection.setup_response('1;No error') + parameters = ResistivityManualParameters(excitation_type='CURRENT', excitation_value=10e-3, compliance_limit=5) + with self.assertRaises(InstrumentException) as ctx: + self.dut.run_complete_resistivity_manual(parameters, "VDP", timeout=0) + self.assertIn("timed out", str(ctx.exception)) + + +class TestRunCompleteNormal(TestWithFakeFastHall): + def test_contact_check_optimized_completes(self): + json_result = '{"Setup": {}, "OptimizationSetup": {}, "OptimizationDiagnostics": {}, "Result": "ok"}' + self.fake_connection.setup_response('No error') # start command + self.fake_connection.setup_response('0;No error') # running = False (done immediately) + self.fake_connection.setup_response(f'{json_result};No error') # results JSON + parameters = ContactCheckOptimizedParameters() + results = self.dut.run_complete_contact_check_optimized(parameters) + self.assertEqual(results, {"Result": "ok"}) + + def test_four_wire_completes(self): + self.fake_connection.setup_response('No error') # start command + self.fake_connection.setup_response('0;No error') # running = False + self.fake_connection.setup_response('{"Setup": {}, "Result": "ok"};No error') # results JSON + parameters = FourWireParameters(contact_point1=1, contact_point2=2, contact_point3=3, contact_point4=4, + excitation_type='CURRENT', excitation_value=10e-3, compliance_limit=1.5) + results = self.dut.run_complete_four_wire(parameters) + self.assertEqual(results, {"Result": "ok"}) diff --git a/tests/test_ssm_system.py b/tests/test_ssm_system.py index bafc605..c3fc8c0 100644 --- a/tests/test_ssm_system.py +++ b/tests/test_ssm_system.py @@ -1451,6 +1451,31 @@ def test_do_dc_sweep_step_and_measure(self): self.assertIn('TRACe:STARt 3', self.fake_connection.get_outgoing_message()) +class TestSweepPointCalculation(TestWithFakeSSMSSourceModule): + def test_start_equals_stop_returns_one(self): + result = self.dut_module._calculate_number_of_sweep_points(5.0, 5.0, 'LINEAR') + self.assertEqual(result, 1) + + def test_start_equals_stop_logarithmic_returns_one(self): + result = self.dut_module._calculate_number_of_sweep_points(5.0, 5.0, 'LOGARITHMIC') + self.assertEqual(result, 1) + + def test_zero_start_logarithmic_raises(self): + with self.assertRaises(ValueError) as ctx: + self.dut_module._calculate_number_of_sweep_points(0, 5.0, 'LOGARITHMIC') + self.assertIn("zero", str(ctx.exception).lower()) + + def test_linear_sweep_normal(self): + result = self.dut_module._calculate_number_of_sweep_points(0, 1.0, 'LINEAR') + self.assertIsInstance(result, int) + self.assertGreater(result, 1) + + def test_logarithmic_sweep_normal(self): + result = self.dut_module._calculate_number_of_sweep_points(1.0, 10.0, 'LOGARITHMIC') + self.assertIsInstance(result, int) + self.assertGreater(result, 1) + + class TestMeasureModule(TestWithFakeSSMSMeasureModule): def test_get_name(self): self.fake_connection.setup_response('Module_name;No error') From e4e6a27e2ab176093a82b85d3c51032d892356fb Mon Sep 17 00:00:00 2001 From: Justin Fichtner Date: Tue, 17 Feb 2026 21:16:12 -0500 Subject: [PATCH 5/5] Move _validate_scpi_parameter to end of class to avoid merge conflict Co-Authored-By: Claude Opus 4.6 --- lakeshore/generic_instrument.py | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/lakeshore/generic_instrument.py b/lakeshore/generic_instrument.py index 7911dcf..a1f455e 100644 --- a/lakeshore/generic_instrument.py +++ b/lakeshore/generic_instrument.py @@ -85,14 +85,6 @@ class GenericInstrument: vid_pid = [] logger = logging.getLogger(__name__) - @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") - def __init__(self, serial_number, com_port, baud_rate, data_bits, stop_bits, parity, flow_control, handshaking, timeout, ip_address, tcp_port, connection=None): # Initialize values common to all instruments @@ -360,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")