[WIP] Avoid dependency on dynamic attribute lookup in drivers

pyproject.toml

@@ -205,6 +205,24 @@ reportUnusedClass = "none"
 reportUnnecessaryCast = "none"  # mypy already checks this. If it fails for pyright its because mypy requires it
 reportUnnecessaryContains = "none"
 
+[tool.ty]
+
+[tool.ty.rules]
+
+# we catch these via pyright or mypy so ignore here
+# deprecated: we trigger deprecated on use of deprecated methods in other deprecated methods.
+# unresolved-imports: we have a lot of imports that are only available with various libraries for specific instrument drivers
+# unused-ignore-comment: mypy already checks for unused-ignores so it they are unused by ty its because mypy requires them
+unresolved-import = "ignore"
+deprecated = "ignore"
+unused-type-ignore-comment = "ignore"
+
+[[tool.ty.overrides]]
+include = ["src/qcodes/instrument_drivers/Harvard/Decadac.py"]
+
+[tool.ty.overrides.rules]
+unresolved-attribute = "ignore"
+
 [tool.pytest.ini_options]
 minversion = "7.2"
 testpaths = "tests"

src/qcodes/dataset/dond/do_nd.py

@@ -23,6 +23,7 @@
     _set_write_period,
     catch_interrupts,
 )
+from qcodes.dataset.experiment_container import Experiment
 from qcodes.dataset.measurements import Measurement
 from qcodes.dataset.threading import (
     SequentialParamsCaller,
@@ -42,7 +43,7 @@
         MultiAxesTupleListWithDataSet,
         ParamMeasT,
     )
-    from qcodes.dataset.experiment_container import Experiment
+
 
 LOG = logging.getLogger(__name__)
 SweepVarType = Any
@@ -400,8 +401,12 @@ def _get_experiments(
             experiments_internal: Sequence[Experiment | None] = [
                 experiments
             ] * n_experiments_required
-        else:
+        elif not isinstance(experiments, Experiment):
             experiments_internal = experiments
+        else:
+            raise TypeError(
+                f"Invalid type for experiments got {experiments} of type {type(experiments)}"
+            )
 
         if len(experiments_internal) != n_experiments_required:
             raise ValueError(

src/qcodes/extensions/infer.py

@@ -264,8 +264,10 @@ def get_parent_instruments_from_chain_of_type(
 
     param_chain = get_parameter_chain(parameter)
     return tuple(
+        # cast is required since mypy as of 1.19.1 cannot infer the type narrowing based
+        # on isinstance checks inside comprehensions
         [
-            cast("TInstrument", param.instrument)
+            cast("TInstrument", param.instrument)  # ty: ignore[redundant-cast]
             for param in param_chain
             if isinstance(param.instrument, instrument_type)
         ]

src/qcodes/instrument/channel.py

@@ -71,16 +71,16 @@
 
     # Pass any commands to read or write from the instrument up to the parent
     def write(self, cmd: str) -> None:
         return self._parent.write(cmd)
 
     def write_raw(self, cmd: str) -> None:
         return self._parent.write_raw(cmd)
 
     def ask(self, cmd: str) -> str:
         return self._parent.ask(cmd)
 
     def ask_raw(self, cmd: str) -> str:
         return self._parent.ask_raw(cmd)
 
     @property
     def parent(self) -> _TIB_co:
@@ -198,42 +198,42 @@
                 )
 
     @overload
-    def __getitem__(self, i: int) -> InstrumentModuleType: ...
+    def __getitem__(self, index: int) -> InstrumentModuleType: ...
 
     @overload
-    def __getitem__(self: Self, i: slice | tuple[int, ...]) -> Self: ...
+    def __getitem__(self: Self, index: slice | tuple[int, ...]) -> Self: ...
 
     def __getitem__(
-        self: Self, i: int | slice | tuple[int, ...]
+        self: Self, index: int | slice | tuple[int, ...]
     ) -> InstrumentModuleType | Self:
         """
         Return either a single channel, or a new :class:`ChannelTuple`
         containing only the specified channels
 
         Args:
-            i: Either a single channel index or a slice of channels
+            index: Either a single channel index or a slice of channels
               to get
 
         """
-        if isinstance(i, slice):
+        if isinstance(index, slice):
             return type(self)(
                 self._parent,
                 self._name,
                 self._chan_type,
-                self._channels[i],
+                self._channels[index],
                 multichan_paramclass=self._paramclass,
                 snapshotable=self._snapshotable,
             )
-        elif isinstance(i, tuple):
+        elif isinstance(index, tuple):
             return type(self)(
                 self._parent,
                 self._name,
                 self._chan_type,
-                [self._channels[j] for j in i],
+                [self._channels[j] for j in index],
                 multichan_paramclass=self._paramclass,
                 snapshotable=self._snapshotable,
             )
-        return self._channels[i]
+        return self._channels[index]
 
     def __iter__(self) -> Iterator[InstrumentModuleType]:
         return iter(self._channels)
@@ -244,8 +244,8 @@
     def __len__(self) -> int:
         return len(self._channels)
 
-    def __contains__(self, item: object) -> bool:
-        return item in self._channels
+    def __contains__(self, value: object) -> bool:
+        return value in self._channels
 
     def __repr__(self) -> str:
         return (
@@ -315,35 +315,31 @@
         name_parts.append(self.short_name)
         return name_parts
 
-    # the parameter obj should be called value but that would
-    # be an incompatible change
-    def index(  #  pyright: ignore[reportIncompatibleMethodOverride]
+    def index(
         self,
-        obj: InstrumentModuleType,
+        value: InstrumentModuleType,
         start: int = 0,
         stop: int = sys.maxsize,
     ) -> int:
         """
         Return the index of the given object
 
         Args:
-            obj: The object to find in the channel list.
+            value: The object to find in the channel list.
             start: Index to start searching from.
             stop: Index to stop searching at.
 
         """
-        return self._channels.index(obj, start, stop)
+        return self._channels.index(value, start, stop)
 
-    def count(  #  pyright: ignore[reportIncompatibleMethodOverride]
-        self, obj: InstrumentModuleType
-    ) -> int:
+    def count(self, value: InstrumentModuleType) -> int:
         """Returns number of instances of the given object in the list
 
         Args:
-            obj: The object to find in the ChannelTuple.
+            value: The object to find in the ChannelTuple.
 
         """
-        return self._channels.count(obj)
+        return self._channels.count(value)
 
     def get_channels_by_name(self: Self, *names: str) -> Self:
         """
@@ -717,15 +713,15 @@
             self._locked = False
 
     @overload
-    def __delitem__(self, key: int) -> None: ...
+    def __delitem__(self, index: int) -> None: ...
 
     @overload
-    def __delitem__(self, key: slice) -> None: ...
+    def __delitem__(self, index: slice) -> None: ...
 
-    def __delitem__(self, key: int | slice) -> None:
+    def __delitem__(self, index: int | slice) -> None:
         if self._locked:
             raise AttributeError("Cannot delete from a locked channel list")
-        self._channels.__delitem__(key)
+        self._channels.__delitem__(index)
         self._channel_mapping = {
             channel.short_name: channel for channel in self._channels
         }
@@ -759,27 +755,25 @@
             channel.short_name: channel for channel in self._channels
         }
 
-    def append(  #  pyright: ignore[reportIncompatibleMethodOverride]
-        self, obj: InstrumentModuleType
-    ) -> None:
+    def append(self, value: InstrumentModuleType) -> None:
         """
         Append a Channel to this list. Requires that the ChannelList is not
         locked and that the channel is of the same type as the ones in the list.
 
         Args:
-            obj: New channel to add to the list.
+            value: New channel to add to the list.
 
         """
         if self._locked:
             raise AttributeError("Cannot append to a locked channel list")
-        if not isinstance(obj, self._chan_type):
+        if not isinstance(value, self._chan_type):
             raise TypeError(
                 f"All items in a channel list must be of the same "
-                f"type. Adding {type(obj).__name__} to a "
+                f"type. Adding {type(value).__name__} to a "
                 f"list of {self._chan_type.__name__}."
             )
-        self._channel_mapping[obj.short_name] = obj
-        self._channels.append(obj)
+        self._channel_mapping[value.short_name] = value
+        self._channels.append(value)
 
     def clear(self) -> None:
         """
@@ -791,63 +785,59 @@
         self._channels.clear()
         self._channel_mapping.clear()
 
-    def remove(  #  pyright: ignore[reportIncompatibleMethodOverride]
-        self, obj: InstrumentModuleType
-    ) -> None:
+    def remove(self, value: InstrumentModuleType) -> None:
         """
-        Removes obj from ChannelList if not locked.
+        Removes value from ChannelList if not locked.
 
         Args:
-            obj: Channel to remove from the list.
+            value: Channel to remove from the list.
 
         """
         if self._locked:
             raise AttributeError("Cannot remove from a locked channel list")
         else:
-            self._channels.remove(obj)
-            self._channel_mapping.pop(obj.short_name)
+            self._channels.remove(value)
+            self._channel_mapping.pop(value.short_name)
 
-    def extend(  #  pyright: ignore[reportIncompatibleMethodOverride]
-        self, objects: Iterable[InstrumentModuleType]
-    ) -> None:
+    def extend(self, values: Iterable[InstrumentModuleType]) -> None:
         """
-        Insert an iterable of objects into the list of channels.
+        Insert an iterable of InstrumentModules into the list of channels.
 
         Args:
-            objects: A list of objects to add into the
+            values: A list of InstrumentModules to add into the
               :class:`ChannelList`.
 
         """
-        # objects may be a generator but we need to iterate over it twice
+        # values may be a generator but we need to iterate over it twice
         # below so copy it into a tuple just in case.
         if self._locked:
             raise AttributeError("Cannot extend a locked channel list")
-        objects_tuple = tuple(objects)
-        if not all(isinstance(obj, self._chan_type) for obj in objects_tuple):
+        values_tuple = tuple(values)
+        if not all(isinstance(value, self._chan_type) for value in values_tuple):
             raise TypeError("All items in a channel list must be of the same type.")
-        self._channels.extend(objects_tuple)
-        self._channel_mapping.update({obj.short_name: obj for obj in objects_tuple})
+        self._channels.extend(values_tuple)
+        self._channel_mapping.update(
+            {value.short_name: value for value in values_tuple}
+        )
 
-    def insert(  #  pyright: ignore[reportIncompatibleMethodOverride]
-        self, index: int, obj: InstrumentModuleType
-    ) -> None:
+    def insert(self, index: int, value: InstrumentModuleType) -> None:
         """
         Insert an object into the ChannelList at a specific index.
 
         Args:
             index: Index to insert object.
-            obj: Object of type chan_type to insert.
+            value: Object of type chan_type to insert.
 
         """
         if self._locked:
             raise AttributeError("Cannot insert into a locked channel list")
-        if not isinstance(obj, self._chan_type):
+        if not isinstance(value, self._chan_type):
             raise TypeError(
                 f"All items in a channel list must be of the same "
-                f"type. Adding {type(obj).__name__} to a list of {self._chan_type.__name__}."
+                f"type. Adding {type(value).__name__} to a list of {self._chan_type.__name__}."
             )
-        self._channels.insert(index, obj)
-        self._channel_mapping[obj.short_name] = obj
+        self._channels.insert(index, value)
+        self._channel_mapping[value.short_name] = value
 
     def get_validator(self) -> ChannelTupleValidator:
         """

src/qcodes/instrument/mockers/ami430.py

@@ -163,7 +163,7 @@ def _handle_messages(self, msg):
             if callable(handler):
                 # some of the callables in the dict does not take arguments.
                 # ignore that warning for now since this is mock code only
-                rval = handler(args)  # pyright: ignore[reportCallIssue]
+                rval = handler(args)  # pyright: ignore[reportCallIssue]  # ty: ignore[ too-many-positional-arguments]
             else:
                 rval = handler
 

src/qcodes/instrument_drivers/AlazarTech/dll_wrapper.py

@@ -62,22 +62,23 @@ def _mark_params_as_updated(*args: Any) -> None:
 def _check_error_code(
     return_code: int, func: Callable[..., Any], arguments: tuple[Any, ...]
 ) -> tuple[Any, ...]:
+    func_name: str = getattr(func, "__name__", "UnknownFunction")
     if return_code not in {API_SUCCESS, API_DMA_IN_PROGRESS}:
         argrepr = repr(arguments)
         if len(argrepr) > 100:
             argrepr = argrepr[:96] + "...]"
 
         logger.error(
             f"Alazar API returned code {return_code} from function "
-            f"{func.__name__} with args {argrepr}"
+            f"{func_name} with args {argrepr}"
         )
 
         if return_code not in ERROR_CODES:
             raise RuntimeError(
-                f"unknown error {return_code} from function {func.__name__} with args: {argrepr}"
+                f"unknown error {return_code} from function {func_name} with args: {argrepr}"
             )
         raise RuntimeError(
-            f"error {return_code}: {ERROR_CODES[ReturnCode(return_code)]} from function {func.__name__} with args: {argrepr}"
+            f"error {return_code}: {ERROR_CODES[ReturnCode(return_code)]} from function {func_name} with args: {argrepr}"
         )
 
     return arguments

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_2000.py

@@ -209,8 +209,6 @@ def __init__(
         )
         """Parameter amplitude"""
 
-        self.add_function("reset", call_cmd="*RST")
-
         if reset:
             self.reset()
 
@@ -220,6 +218,9 @@ def __init__(
 
         self.connect_message()
 
+    def reset(self) -> None:
+        self.write("*RST")
+
     def trigger(self) -> None:
         if not self.trigger_continuous():
             self.write("INIT")