Refactor Keithley 2600 driver to use ParameterWithSetpoints for fast sweep

docs/changes/newsfragments/7851.improved_driver

@@ -0,0 +1,3 @@
+Refactored ``Keithley2600`` driver to use `ParameterWithSetpoints` instead
+of relying on deprecated `qcodes_loop` code. Contains breaking change since
+method `doFastSweep` has been replaced with `fastweep`.

src/qcodes/instrument_drivers/Keithley/_Keithley_2600.py

@@ -4,7 +4,7 @@
 import struct
 import warnings
 from enum import StrEnum
-from typing import TYPE_CHECKING, Any, Literal
+from typing import TYPE_CHECKING, Any, Literal, Self
 
 import numpy as np
 import numpy.typing as npt
@@ -17,7 +17,6 @@
     VisaInstrumentKWArgs,
 )
 from qcodes.parameters import (
-    ArrayParameter,
     Parameter,
     ParameterWithSetpoints,
     ParamRawDataType,
@@ -27,84 +26,131 @@
 if TYPE_CHECKING:
     from collections.abc import Sequence
 
-    from qcodes_loop.data.data_set import DataSet
     from typing_extensions import Unpack
 
 
 log = logging.getLogger(__name__)
 
 
-class LuaSweepParameter(ArrayParameter):
+class LuaSweepParameter(ParameterWithSetpoints[npt.NDArray, "Keithley2600Channel"]):
     """
     Parameter class to hold the data from a
     deployed Lua script sweep.
+
+    For more information on writing Lua scripts for the Keithley2600, please see
+    https://www.tek.com/en/documents/application-note/how-to-write-scripts-for-test-script-processing-(tsp)
     """
 
-    def __init__(self, name: str, instrument: Instrument, **kwargs: Any) -> None:
-        super().__init__(
-            name=name,
-            shape=(1,),
-            docstring="Holds a sweep",
-            instrument=instrument,
-            **kwargs,
-        )
+    def _set_mode(self, mode: Literal["IV", "VI", "VIfourprobe"]) -> None:
+        match mode:
+            case "IV":
+                self.unit = "A"
+                self.setpoint_names = ("Voltage",)
+                self.setpoint_units = ("V",)
+                self.label = "current"
+                self._short_name = "iv_sweep"
+            case "VI":
+                self.unit = "V"
+                self.setpoint_names = ("Current",)
+                self.setpoint_units = ("A",)
+                self.label = "voltage"
+                self._short_name = "vi_sweep"
+            case "VIfourprobe":
+                self.unit = "V"
+                self.setpoint_names = ("Current",)
+                self.setpoint_units = ("A",)
+                self.label = "voltage"
+                self._short_name = "vi_sweep_four_probe"
+
+    def _fast_sweep(self) -> npt.NDArray:
+        if self.instrument is None:
+            raise RuntimeError("No instrument attached to Parameter.")
 
-    def prepareSweep(self, start: float, stop: float, steps: int, mode: str) -> None:
-        """
-        Builds setpoints and labels
+        channel = self.instrument.channel
 
-        Args:
-            start: Starting point of the sweep
-            stop: Endpoint of the sweep
-            steps: No. of sweep steps
-            mode: Type of sweep, either 'IV' (voltage sweep),
-                'VI' (current sweep two probe setup) or
-                'VIfourprobe' (current sweep four probe setup)
+        # an extra visa query, a necessary precaution
+        # to avoid timing out when waiting for long
+        # measurements
+        nplc = self.instrument.nplc()
 
-        """
+        mode = self.instrument.fastsweep_mode()
+        start = self.instrument.fastsweep_start()
+        stop = self.instrument.fastsweep_stop()
+        steps = self.instrument.fastsweep_npts()
 
-        if mode not in ["IV", "VI", "VIfourprobe"]:
-            raise ValueError('mode must be either "VI", "IV" or "VIfourprobe"')
+        dV = (stop - start) / (steps - 1)
 
-        self.shape = (steps,)
+        match mode:
+            case "IV":
+                meas = "i"
+                source = "v"
+                func = "1"
+                sense_mode = "0"
+            case "VI":
+                meas = "v"
+                source = "i"
+                func = "0"
+                sense_mode = "0"
+            case "VIfourprobe":
+                meas = "v"
+                source = "i"
+                func = "0"
+                sense_mode = "1"
+            case _:
+                raise ValueError(f"Invalid fastsweep mode {mode}")
 
-        if mode == "IV":
-            self.unit = "A"
-            self.setpoint_names = ("Voltage",)
-            self.setpoint_units = ("V",)
-            self.label = "current"
-            self._short_name = "iv_sweep"
+        script = [
+            f"{channel}.measure.nplc = {nplc:.12f}",
+            f"{channel}.source.output = 1",
+            f"startX = {start:.12f}",
+            f"dX = {dV:.12f}",
+            f"{channel}.sense = {sense_mode}",
+            f"{channel}.source.output = 1",
+            f"{channel}.source.func = {func}",
+            f"{channel}.measure.count = 1",
+            f"{channel}.nvbuffer1.clear()",
+            f"{channel}.nvbuffer1.appendmode = 1",
+            f"for index = 1, {steps} do",
+            "  target = startX + (index-1)*dX",
+            f"  {channel}.source.level{source} = target",
+        ]
 
-        if mode == "VI":
-            self.unit = "V"
-            self.setpoint_names = ("Current",)
-            self.setpoint_units = ("A",)
-            self.label = "voltage"
-            self._short_name = "vi_sweep"
+        # Only add delay code to lua script if greater than 0
+        inter_delay = self.instrument.fastsweep_inter_delay.get_latest()
+        if inter_delay > 0:
+            script.append(f"  delay({inter_delay})")
+
+        script.extend(
+            [
+                f"  {channel}.measure.{meas}({channel}.nvbuffer1)",
+                "end",
+                "format.data = format.REAL32",
+                "format.byteorder = format.LITTLEENDIAN",
+                f"printbuffer(1, {steps}, {channel}.nvbuffer1.readings)",
+            ]
+        )
 
-        if mode == "VIfourprobe":
-            self.unit = "V"
-            self.setpoint_names = ("Current",)
-            self.setpoint_units = ("A",)
-            self.label = "voltage"
-            self._short_name = "vi_sweep_four_probe"
+        return self.instrument._execute_lua(script, steps)
 
-        self.setpoints = (tuple(np.linspace(start, stop, steps)),)
+    def get_raw(self) -> npt.NDArray:
+        data = self._fast_sweep()
 
-        self.start = start
-        self.stop = stop
-        self.steps = steps
-        self.mode = mode
+        return data
+
+
+class FastSweepSetpoints(Parameter[npt.NDArray, "Keithley2600Channel"]):
+    """
+    A simple :class:`Parameter` that holds all the setpoints for a fastsweep
+    """
 
     def get_raw(self) -> npt.NDArray:
-        if self.instrument is not None:
-            data = self.instrument._fast_sweep(
-                self.start, self.stop, self.steps, self.mode
-            )
-        else:
+        if self.instrument is None:
             raise RuntimeError("No instrument attached to Parameter.")
 
-        return data
+        npts = self.instrument.fastsweep_npts()
+        start = self.instrument.fastsweep_start()
+        stop = self.instrument.fastsweep_stop()
+        return np.linspace(start, stop, npts)
 
 
 class TimeTrace(ParameterWithSetpoints):
@@ -576,10 +622,64 @@ def __init__(self, parent: Instrument, name: str, channel: str) -> None:
         )
         """Current limit e.g. the maximum current allowed in voltage mode. If exceeded the voltage will be clipped."""
 
+        self.fastsweep_npts: Parameter[int, Self] = self.add_parameter(
+            "fastsweep_npts",
+            initial_value=20,
+            label="Number of fastweep points",
+            get_cmd=None,
+            set_cmd=None,
+        )
+        """Parameter fastweep_npts"""
+
+        self.fastsweep_start: Parameter[float, Self] = self.add_parameter(
+            "fastsweep_start", label="fastsweep start", get_cmd=None, set_cmd=None
+        )
+        """Starting value of fastsweep. Can be current or voltage."""
+
+        self.fastsweep_stop: Parameter[float, Self] = self.add_parameter(
+            "fastsweep_stop", label="fastsweep stop", get_cmd=None, set_cmd=None
+        )
+        """Stopping value of fastsweep. Can be current or voltage."""
+
+        self.fastsweep_inter_delay: Parameter[float, Self] = self.add_parameter(
+            name="fastsweep_inter_delay",
+            label="Fastsweep Inter Delay",
+            initial_value=0,
+            vals=vals.Numbers(min_value=0),
+            unit="s",
+            get_cmd=None,
+            set_cmd=None,
+        )
+        """Time in seconds to wait between setting a target value and taking a measurement."""
+
+        self.fastsweep_setpoints: FastSweepSetpoints = self.add_parameter(
+            name="fastsweep_setpoints",
+            label="Fastsweep setpoints",
+            snapshot_value=False,
+            vals=vals.Arrays(shape=(self.fastsweep_npts,)),
+            parameter_class=FastSweepSetpoints,
+        )
+        """Holds array of setpoints for doing a fastsweep. Can
+        be of units V or I depending on `Keithley2600Channel.fastsweep_mode`"""
+
         self.fastsweep: LuaSweepParameter = self.add_parameter(
-            "fastsweep", parameter_class=LuaSweepParameter
+            "fastsweep",
+            vals=vals.Arrays(shape=(self.fastsweep_npts,)),
+            setpoints=(self.fastsweep_setpoints,),
+            parameter_class=LuaSweepParameter,
+        )
+        """Performs buffered readout of desired sweep mode."""
+
+        self.fastsweep_mode: Parameter[Literal["IV", "VI", "VIfourprobe"], Self] = (
+            self.add_parameter(
+                "fastsweep_mode",
+                initial_value="IV",
+                get_cmd=None,
+                set_cmd=self.fastsweep._set_mode,
+                vals=vals.Enum("IV", "VI", "VIfourprobe"),
+            )
         )
-        """Parameter fastsweep"""
+        """Parameter fastsweep_mode"""
 
         self.timetrace_npts: Parameter = self.add_parameter(
             "timetrace_npts",
@@ -645,107 +745,6 @@ def reset(self) -> None:
         log.debug(f"Reset channel {self.channel}. Updating settings...")
         self.snapshot(update=True)
 
-    def doFastSweep(self, start: float, stop: float, steps: int, mode: str) -> DataSet:
-        """
-        Perform a fast sweep using a deployed lua script and
-        return a QCoDeS DataSet with the sweep.
-
-        Args:
-            start: starting sweep value (V or A)
-            stop: end sweep value (V or A)
-            steps: number of steps
-            mode: Type of sweep, either 'IV' (voltage sweep),
-                'VI' (current sweep two probe setup) or
-                'VIfourprobe' (current sweep four probe setup)
-
-        """
-        try:
-            # lazy import to avoid a geneal dependency on qcodes_loop
-            from qcodes_loop.measure import Measure
-        except ImportError as e:
-            raise ImportError(
-                "The doFastSweep method requires the "
-                "qcodes_loop package to be installed."
-            ) from e
-        # prepare setpoints, units, name
-        self.fastsweep.prepareSweep(start, stop, steps, mode)
-
-        data = Measure(self.fastsweep).run()
-
-        return data
-
-    def _fast_sweep(
-        self,
-        start: float,
-        stop: float,
-        steps: int,
-        mode: Literal["IV", "VI", "VIfourprobe"] = "IV",
-    ) -> npt.NDArray:
-        """
-        Perform a fast sweep using a deployed Lua script.
-        This is the engine that forms the script, uploads it,
-        runs it, collects the data, and casts the data correctly.
-
-        Args:
-            start: starting voltage
-            stop: end voltage
-            steps: number of steps
-            mode: Type of sweep, either 'IV' (voltage sweep),
-                'VI' (current sweep two probe setup) or
-                'VIfourprobe' (current sweep four probe setup)
-
-        """
-
-        channel = self.channel
-
-        # an extra visa query, a necessary precaution
-        # to avoid timing out when waiting for long
-        # measurements
-        nplc = self.nplc()
-
-        dV = (stop - start) / (steps - 1)
-
-        if mode == "IV":
-            meas = "i"
-            sour = "v"
-            func = "1"
-            sense_mode = "0"
-        elif mode == "VI":
-            meas = "v"
-            sour = "i"
-            func = "0"
-            sense_mode = "0"
-        elif mode == "VIfourprobe":
-            meas = "v"
-            sour = "i"
-            func = "0"
-            sense_mode = "1"
-        else:
-            raise ValueError(f"Invalid mode {mode}")
-
-        script = [
-            f"{channel}.measure.nplc = {nplc:.12f}",
-            f"{channel}.source.output = 1",
-            f"startX = {start:.12f}",
-            f"dX = {dV:.12f}",
-            f"{channel}.sense = {sense_mode}",
-            f"{channel}.source.output = 1",
-            f"{channel}.source.func = {func}",
-            f"{channel}.measure.count = 1",
-            f"{channel}.nvbuffer1.clear()",
-            f"{channel}.nvbuffer1.appendmode = 1",
-            f"for index = 1, {steps} do",
-            "  target = startX + (index-1)*dX",
-            f"  {channel}.source.level{sour} = target",
-            f"  {channel}.measure.{meas}({channel}.nvbuffer1)",
-            "end",
-            "format.data = format.REAL32",
-            "format.byteorder = format.LITTLEENDIAN",
-            f"printbuffer(1, {steps}, {channel}.nvbuffer1.readings)",
-        ]
-
-        return self._execute_lua(script, steps)
-
     def _execute_lua(self, _script: list[str], steps: int) -> npt.NDArray:
         """
         This is the function that sends the Lua script to be executed and