Merge pull request #7919 from jenshnielsen/driver_improvments

Avoid dynamic attribute lookup in several drivers

Author: jenshnielsen

src/qcodes/instrument_drivers/Galil/dmc_41x3.py

@@ -254,7 +254,7 @@ def clear_sequence(self, coord_sys: str) -> None:
 """
 
 
-class GalilDMC4133Motor(InstrumentChannel):
+class GalilDMC4133Motor(InstrumentChannel["GalilDMC4133Controller"]):
     """
     Class to control a single motor (independent of possible other motors)
     """
@@ -458,7 +458,7 @@ def wait_till_motor_motion_complete(self) -> None:
             while self.is_in_motion():
                 pass
         except KeyboardInterrupt:
-            self.root_instrument.abort()
+            self.parent.abort()
             self.off()
 
     def error_magnitude(self) -> float:

src/qcodes/instrument_drivers/Keithley/Keithley_s46.py

@@ -7,7 +7,6 @@
 from typing import TYPE_CHECKING, Any, ClassVar
 
 from qcodes.instrument import (
-    Instrument,
     VisaInstrument,
     VisaInstrumentKWArgs,
 )
@@ -56,7 +55,7 @@ def release(self, channel_number: int) -> None:
             self._locked_by = None
 
 
-class S46Parameter(Parameter):
+class S46Parameter(Parameter[ParamRawDataType, "KeithleyS46"]):
     """
     A parameter class for S46 channels. We do not use the QCoDeS
     InstrumentChannel class because our channel has one state parameter,
@@ -73,7 +72,7 @@ class S46Parameter(Parameter):
     def __init__(
         self,
         name: str,
-        instrument: Instrument | None,
+        instrument: "KeithleyS46",
         channel_number: int,
         lock: KeithleyS46RelayLock,
         **kwargs: Any,
@@ -95,7 +94,6 @@ def __init__(
                 ) from e
 
     def _get(self, get_cached: bool) -> str:
-        assert isinstance(self.instrument, KeithleyS46)
         closed_channels = self.instrument.closed_channels.get_latest()
 
         if not get_cached or closed_channels is None:

src/qcodes/instrument_drivers/QDev/QDac_channels.py

@@ -11,7 +11,7 @@
 from qcodes import validators as vals
 from qcodes.instrument import (
     ChannelList,
-    Instrument,
+    ChannelTuple,
     InstrumentBaseKWArgs,
     InstrumentChannel,
     VisaInstrument,
@@ -29,7 +29,7 @@
 log = logging.getLogger(__name__)
 
 
-class QDevQDacChannel(InstrumentChannel):
+class QDevQDacChannel(InstrumentChannel["QDevQDac"]):
     """
     A single output channel of the QDac.
 
@@ -40,7 +40,7 @@ class QDevQDacChannel(InstrumentChannel):
 
     def __init__(
         self,
-        parent: Instrument,
+        parent: "QDevQDac",
         name: str,
         channum: int,
         **kwargs: "Unpack[InstrumentBaseKWArgs]",
@@ -143,9 +143,20 @@ def snapshot_base(
         update: bool | None = False,
         params_to_skip_update: "Sequence[str] | None" = None,
     ) -> dict[Any, Any]:
-        update_currents = self._parent._update_currents and update
-        if update and not self._parent._get_status_performed:
-            self._parent._update_cache(readcurrents=update_currents)
+        # setting update not None will override parent setting
+        # otherwise we use parent setting
+        # parent._update | update | do update
+        # True           | True   | True
+        # True           | None   | True
+        # True           | False  | False
+        # False          | True   | True
+        # False          | None   | False
+        # False          | False  | False
+        update_currents = (
+            self.parent._update_currents and update is not False
+        ) or update is True
+        if update and not self.parent._get_status_performed:
+            self.parent._update_cache(readcurrents=update_currents)
         # call get_status rather than getting the status individually for
         # each parameter. This is only done if _get_status_performed is False
         # this is used to signal that the parent has already called it and
@@ -287,7 +298,10 @@ def __init__(
             channels.append(channel)
             # Should raise valueerror if name is invalid (silently fails now)
             self.add_submodule(f"ch{i:02}", channel)
-        self.add_submodule("channels", channels.to_channel_tuple())
+        self.channels: ChannelTuple[QDevQDacChannel] = self.add_submodule(
+            "channels", channels.to_channel_tuple()
+        )
+        """ChannelTuple containing all QDevQDacChannel instances"""
 
         for board in range(6):
             for sensor in range(3):

src/qcodes/instrument_drivers/oxford/MercuryiPS_VISA.py

@@ -76,14 +76,14 @@ def _temp_parser(response: str) -> float:
     return float(response.rsplit(":", maxsplit=1)[-1][:-1])
 
 
-class OxfordMercuryWorkerPS(InstrumentChannel):
+class OxfordMercuryWorkerPS(InstrumentChannel["OxfordMercuryiPS"]):
     """
     Class to hold a worker power supply for the Oxford MercuryiPS
     """
 
     def __init__(
         self,
-        parent: VisaInstrument,
+        parent: OxfordMercuryiPS,
         name: str,
         UID: str,
         **kwargs: Unpack[InstrumentBaseKWArgs],
@@ -108,7 +108,7 @@ def __init__(
 
         # The firmware update from 2.5 -> 2.6 changed the command
         # syntax slightly
-        if version.parse(self.root_instrument.firmware) >= version.parse("2.6"):
+        if version.parse(self.parent.firmware) >= version.parse("2.6"):
             self.psu_string = "SPSU"
         else:
             self.psu_string = "PSU"
@@ -348,10 +348,14 @@ def __init__(
         self.firmware = self.IDN()["firmware"]
 
         # TODO: Query instrument to ensure which PSUs are actually present
-        for grp in ["GRPX", "GRPY", "GRPZ"]:
-            psu_name = grp
-            psu = OxfordMercuryWorkerPS(self, psu_name, grp)
-            self.add_submodule(psu_name, psu)
+        GRPX = OxfordMercuryWorkerPS(self, "GRPX", "GRPX")
+        self.GRPX: OxfordMercuryWorkerPS = self.add_submodule("GRPX", GRPX)
+
+        GRPY = OxfordMercuryWorkerPS(self, "GRPY", "GRPY")
+        self.GRPY: OxfordMercuryWorkerPS = self.add_submodule("GRPY", GRPY)
+
+        GRPZ = OxfordMercuryWorkerPS(self, "GRPZ", "GRPZ")
+        self.GRPZ: OxfordMercuryWorkerPS = self.add_submodule("GRPZ", GRPZ)
 
         self._field_limits = field_limits if field_limits else lambda x, y, z: True
 

src/qcodes/instrument_drivers/stanford_research/SR86x.py

@@ -89,7 +89,7 @@ def get_raw(self) -> npt.NDArray:
         return self._capture_data
 
 
-class SR86xBuffer(InstrumentChannel):
+class SR86xBuffer(InstrumentChannel["SR86x"]):
     """
     Buffer module for the SR86x drivers.
 
@@ -529,7 +529,7 @@ def _get_raw_capture_data_block(
                 f"({size_of_currently_captured_data}kB)"
             )
 
-        values = self._parent.visa_handle.query_binary_values(
+        values = self.parent.visa_handle.query_binary_values(
             f"CAPTUREGET? {offset_in_kb}, {size_in_kb}",
             datatype="f",
             is_big_endian=False,
@@ -621,7 +621,7 @@ def capture_samples(self, sample_count: int) -> dict[str, npt.NDArray]:
         return self.get_capture_data(sample_count)
 
 
-class SR86xDataChannel(InstrumentChannel):
+class SR86xDataChannel(InstrumentChannel["SR86x"]):
     """
     Implements a data channel of SR86x lock-in amplifier. Parameters that are
     assigned to these channels get plotted on the display of the instrument.

src/qcodes/instrument_drivers/tektronix/AWG70000A.py

@@ -17,7 +17,6 @@
 from qcodes import validators as vals
 from qcodes.instrument import (
     ChannelList,
-    Instrument,
     InstrumentBaseKWArgs,
     InstrumentChannel,
     VisaInstrument,
@@ -157,14 +156,14 @@ def validate(self, value: float, context: str = "") -> None:
             validator.validate(value)
 
 
-class Tektronix70000AWGChannel(InstrumentChannel):
+class Tektronix70000AWGChannel(InstrumentChannel["TektronixAWG70000Base"]):
     """
     Class to hold a channel of the AWG.
     """
 
     def __init__(
         self,
-        parent: Instrument,
+        parent: TektronixAWG70000Base,
         name: str,
         channel: int,
         **kwargs: Unpack[InstrumentBaseKWArgs],
@@ -183,8 +182,8 @@ def __init__(
 
         self.channel = channel
 
-        num_channels = self.root_instrument.num_channels
-        self.model = self.root_instrument.model
+        num_channels = self.parent.num_channels
+        self.model = self.parent.model
 
         fg = "function generator"
 
@@ -264,7 +263,7 @@ def __init__(
             label=f"Channel {channel} {fg} signal path",
             set_cmd=f"FGEN:CHANnel{channel}:PATH {{}}",
             get_cmd=f"FGEN:CHANnel{channel}:PATH?",
-            val_mapping=_fg_path_val_map[self.root_instrument.model],
+            val_mapping=_fg_path_val_map[self.parent.model],
         )
         """Parameter fgen_signalpath"""
 
@@ -441,7 +440,7 @@ def _set_fgfreq(self, channel: int, frequency: float) -> None:
                 "Hz, minimum is 1 Hz"
             )
         else:
-            self.root_instrument.write(f"FGEN:CHANnel{channel}:FREQuency {frequency}")
+            self.parent.write(f"FGEN:CHANnel{channel}:FREQuency {frequency}")
 
     def setWaveform(self, name: str) -> None:
         """
@@ -451,10 +450,10 @@ def setWaveform(self, name: str) -> None:
             name: The name of the waveform
 
         """
-        if name not in self.root_instrument.waveformList:
+        if name not in self.parent.waveformList:
             raise ValueError("No such waveform in the waveform list")
 
-        self.root_instrument.write(f'SOURce{self.channel}:CASSet:WAVeform "{name}"')
+        self.parent.write(f'SOURce{self.channel}:CASSet:WAVeform "{name}"')
 
     def setSequenceTrack(self, seqname: str, tracknr: int) -> None:
         """
@@ -465,8 +464,7 @@ def setSequenceTrack(self, seqname: str, tracknr: int) -> None:
             tracknr: Which track to use (1 or 2)
 
         """
-
-        self.root_instrument.write(
+        self.parent.write(
             f'SOURCE{self.channel}:CASSet:SEQuence "{seqname}", {tracknr}'
         )
 
@@ -475,7 +473,7 @@ def clear_asset(self) -> None:
         Clear assigned assets on this channel
         """
 
-        self.root_instrument.write(f"SOURce{self.channel}:CASSet:CLEAR")
+        self.parent.write(f"SOURce{self.channel}:CASSet:CLEAR")
 
 
 AWGChannel = Tektronix70000AWGChannel