DelegateParameterWithSetpoints

qcodes/instrument/parameter.py

@@ -44,6 +44,9 @@
     while allowing to specify label/unit/etc that is different from the
     source parameter.
 
+- :class:`.DelegateParameterWithSetpoints` is similar to :class:`.DelegateParameter`
+    and can be used for proxy-ing other  :class:`.ParameterWithSetpoints` parameters.
+
 - :class:`.ArrayParameter` is an older base class for array-valued parameters.
     For any new driver we strongly recommend using
     :class:`.ParameterWithSetpoints` which is both more flexible and
@@ -1681,6 +1684,128 @@ def snapshot_base(self, update: Optional[bool] = True,
         return snapshot
 
 
+class DelegateParameterWithSetpoints(ParameterWithSetpoints):
+    """
+    The :class:`.DelegateParameterWithSetpoints` wraps a given `source`
+    :class:`ParameterWithSetpoints`. Setting/getting it results in
+    a set/get of the source parameter with the provided arguments.
+
+    The reason for using a :class:`DelegateParameterWithSetpoints` instead
+    of the source parameter is to provide all the functionality of the
+    ParameterWithSetpoints class without overwriting properties of the source:
+    for example to set a different scaling factor and unit on the
+    :class:`.DelegateParameterWithSetpoints` or its setpoints without changing
+    those in the source parameter or the source parameter's setpoints.
+
+    Unlike :class:`DelegateParameter`, :class:`DelegateParameterWithSetpoints`
+    does not support changing the `source` :class:`ParameterWithSetpoints`.
+    :py:attr:`~gettable`, :py:attr:`~settable` and :py:attr:`snapshot_value`
+    properties automatically follow the source parameter.
+
+    :py:attr:`.unit` and :py:attr:`.label` can either be set when constructing
+    a :class:`DelegateParameterWithSetpoints` or inherited from the source
+    :class:`ParameterWithSetpoints`.
+
+    If new setpoints are not provided, then the setpoints of the `source`
+    :class:`ParameterWithSetpoints` will be used "as is".
+
+    Note:
+        DelegateParameterWithSetpoints only supports mappings between the
+        :class:`.DelegateParameterWithSetpoints` and
+        :class:`.ParameterWithSetpoints` that are invertible
+        (e.g. a bijection). It is therefor not allowed to create a
+        :class:`.DelegateParameterWithSetpoints` that performs non invertible
+        transforms in its ``get_raw`` method.
+
+        A DelegateParameterWithSetpoints is not registered on the instrument
+        by default. You should pass ``bind_to_instrument=True`` if you want this to
+        be the case.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        *,
+        source: ParameterWithSetpoints,
+        new_setpoints: Optional[Sequence[DelegateParameter]] = None,
+        **kwargs: Any,
+    ) -> None:
+        self._source = source
+
+        super().__init__(name=name, vals=self._source.vals, **kwargs)
+
+        self._gettable = self._source.gettable
+        self._settable = self._source.settable
+        self._snapshot_value = self._source._snapshot_value
+
+        self._delegate_setpoints: Optional[Sequence[DelegateParameter]]
+        if new_setpoints is not None:
+            self._delegate_setpoints = tuple(new_setpoints)
+            for source_setpoint, delegate_setpoint in zip(self._source.setpoints, self._delegate_setpoints):
+                delegate_setpoint.source = cast(Parameter, source_setpoint)
+        else:
+            self._delegate_setpoints = None
+
+        self._update_validators()
+
+    @property
+    def source(self) -> ParameterWithSetpoints:
+        """
+        The source parameter that this :class:`DelegateParameterWithSetpoints`
+        is bound to.
+
+        :getter: Returns the source parameter.
+        """
+        return self._source
+
+    def _update_validators(self) -> None:
+        self.vals = self._source.vals
+        if self._delegate_setpoints is not None:
+            for delegate_setpoint in self._delegate_setpoints:
+                assert delegate_setpoint.source is not None
+                delegate_setpoint.vals = delegate_setpoint.source.vals
+
+    @property
+    def setpoints(self) -> Sequence[_BaseParameter]:
+        self._update_validators()
+        setpoints = (
+            self._delegate_setpoints
+            if self._delegate_setpoints is not None
+            else self._source.setpoints
+        )
+        return setpoints
+
+    @setpoints.setter
+    def setpoints(self, setpoints: Sequence[_BaseParameter]) -> None:
+        if len(setpoints) > 0:
+            raise AttributeError(
+                f"Cannot set setpoints, since the delegate refers to "
+                f"the setpoints of {self._source}"
+            )
+
+    def validate_consistent_shape(self) -> None:
+        self._update_validators()
+        super().validate_consistent_shape()
+
+    def get_raw(self) -> Any:
+        return self._source.get()
+
+    def set_raw(self, value: Any) -> None:
+        self._source.set(value)
+
+    def snapshot_base(
+        self,
+        update: Optional[bool] = True,
+        params_to_skip_update: Optional[Sequence[str]] = None,
+    ) -> Dict[Any, Any]:
+        snapshot = super().snapshot_base(
+            update=update, params_to_skip_update=params_to_skip_update
+        )
+        source_parameter_snapshot = self._source.snapshot(update=update)
+        snapshot.update({"source_parameter": source_parameter_snapshot})
+        return snapshot
+
+
 class ArrayParameter(_BaseParameter):
     """
     A gettable parameter that returns an array of values.

qcodes/tests/parameter/test_delegate_parameter_with_setpoints.py

@@ -0,0 +1,208 @@
+import numpy as np
+from numpy.random import rand
+import pytest
+from qcodes.dataset import DataSetProtocol
+
+from qcodes.instrument.parameter import (
+    DelegateParameter,
+    ParameterWithSetpoints,
+    Parameter,
+    expand_setpoints_helper,
+)
+import qcodes.utils.validators as vals
+from qcodes.utils.dataset.doNd import dond
+
+from qcodes.instrument.parameter import DelegateParameterWithSetpoints
+
+
+@pytest.fixture(name="parameters")
+def _make_parameters():
+    n_points_1 = Parameter("n_points_1", set_cmd=None, vals=vals.Ints())
+    n_points_2 = Parameter("n_points_2", set_cmd=None, vals=vals.Ints())
+    n_points_3 = Parameter("n_points_3", set_cmd=None, vals=vals.Ints())
+
+    n_points_1.set(10)
+    n_points_2.set(20)
+    n_points_3.set(15)
+
+    setpoints_1 = Parameter(
+        "setpoints_1",
+        get_cmd=lambda: np.arange(n_points_1()),
+        vals=vals.Arrays(shape=(n_points_1,)),
+    )
+    setpoints_2 = Parameter(
+        "setpoints_2",
+        get_cmd=lambda: np.arange(n_points_2()),
+        vals=vals.Arrays(shape=(n_points_2,)),
+    )
+    setpoints_3 = Parameter(
+        "setpoints_3",
+        get_cmd=lambda: np.arange(n_points_3()),
+        vals=vals.Arrays(shape=(n_points_3,)),
+    )
+    yield (n_points_1, n_points_2, n_points_3, setpoints_1, setpoints_2, setpoints_3)
+
+
+def test_validation_shapes():
+    n_points_1 = Parameter("n_points_1", set_cmd=None, vals=vals.Ints())
+    n_points_2 = Parameter("n_points_2", set_cmd=None, vals=vals.Ints())
+
+    n_points_1.set(10)
+    n_points_2.set(20)
+
+    setpoints_1 = Parameter(
+        "setpoints_1",
+        get_cmd=lambda: rand(n_points_1()),
+        vals=vals.Arrays(shape=(n_points_1,)),
+    )
+    setpoints_2 = Parameter(
+        "setpoints_2",
+        get_cmd=lambda: rand(n_points_2()),
+        vals=vals.Arrays(shape=(n_points_2,)),
+    )
+
+    # 1D
+
+    param_with_setpoints_1 = ParameterWithSetpoints(
+        "param_1",
+        get_cmd=lambda: rand(n_points_1()),
+        setpoints=(setpoints_1,),
+        vals=vals.Arrays(shape=(n_points_1,)),
+    )
+
+    delegate_param_1 = DelegateParameterWithSetpoints(
+        "delegate_param_1",
+        source=param_with_setpoints_1,
+        new_setpoints=(DelegateParameter("delegate_stepoint_1", None),),
+    )
+
+    delegate_param_1.validate_consistent_shape()
+    delegate_param_1.validate(delegate_param_1.get())
+
+    # 2D
+
+    param_with_setpoints_2 = ParameterWithSetpoints(
+        "param_2",
+        get_cmd=lambda: rand(n_points_1(), n_points_2()),
+        setpoints=(setpoints_1, setpoints_2),
+        vals=vals.Arrays(shape=(n_points_1, n_points_2)),
+    )
+
+    delegate_param_2 = DelegateParameterWithSetpoints(
+        "delegate_param_2",
+        source=param_with_setpoints_2,
+        new_setpoints=(
+            DelegateParameter("delegate_setpoint_1", None),
+            DelegateParameter("delegate_setpoint_2", None),
+        ),
+    )
+
+    delegate_param_2.validate_consistent_shape()
+    delegate_param_2.validate(delegate_param_2.get())
+
+
+def test_expand_setpoints_1d(parameters):
+    """
+    Test that the setpoints expander helper function works correctly
+    """
+
+    (
+        n_points_1,
+        n_points_2,
+        n_points_3,
+        setpoints_1,
+        setpoints_2,
+        setpoints_3,
+    ) = parameters
+
+    param_with_setpoints_1 = ParameterWithSetpoints(
+        "param_1",
+        get_cmd=lambda: rand(n_points_1()),
+        setpoints=(setpoints_1,),
+        vals=vals.Arrays(shape=(n_points_1,)),
+    )
+
+    delegate_param_1 = DelegateParameterWithSetpoints(
+        "delegate_param_1",
+        source=param_with_setpoints_1,
+        new_setpoints=(DelegateParameter("delegate_setpoint_1", None),),
+    )
+
+    data = expand_setpoints_helper(delegate_param_1)
+
+    assert len(data) == 2
+    assert len(data[0][1]) == len(data[1][1])
+
+
+def test_expand_setpoints_2d(parameters):
+
+    (
+        n_points_1,
+        n_points_2,
+        n_points_3,
+        setpoints_1,
+        setpoints_2,
+        setpoints_3,
+    ) = parameters
+
+    param_with_setpoints_2 = ParameterWithSetpoints(
+        "param_2",
+        get_cmd=lambda: rand(n_points_1(), n_points_2()),
+        vals=vals.Arrays(shape=(n_points_1, n_points_2)),
+        setpoints=(setpoints_1, setpoints_2),
+    )
+
+    delegate_param_2 = DelegateParameterWithSetpoints(
+        "delegate_param_2",
+        source=param_with_setpoints_2,
+        new_setpoints=(
+            DelegateParameter("delegate_setpoint_1", None),
+            DelegateParameter("delegate_setpoint_2", None),
+        ),
+    )
+
+    data = expand_setpoints_helper(delegate_param_2)
+
+    assert len(data) == 3
+    assert data[0][1].shape == data[1][1].shape
+    assert data[0][1].shape == data[2][1].shape
+
+    sp1 = data[0][1]
+    sp2 = data[1][1]
+    # the first set of setpoints should be repeated along the second axis
+    for i in range(sp1.shape[1]):
+        np.testing.assert_array_equal(sp1[:, i], np.arange(sp1.shape[0]))
+    # the second set of setpoints should be repeated along the first axis
+    for i in range(sp2.shape[0]):
+        np.testing.assert_array_equal(sp2[i, :], np.arange(sp1.shape[1]))
+
+
+def test_delegate_parameter_with_setpoints_in_measurement(parameters, empty_experiment):
+    _ = empty_experiment
+    (
+        n_points_1,
+        n_points_2,
+        n_points_3,
+        setpoints_1,
+        setpoints_2,
+        setpoints_3,
+    ) = parameters
+
+    param_with_setpoints_1 = ParameterWithSetpoints(
+        "param_1",
+        get_cmd=lambda: rand(n_points_1()),
+        setpoints=(setpoints_1,),
+        vals=vals.Arrays(shape=(n_points_1,)),
+    )
+
+    delegate_param_1 = DelegateParameterWithSetpoints(
+        "delegate_param_1",
+        source=param_with_setpoints_1,
+        new_setpoints=(DelegateParameter("delegate_setpoint_1", None),),
+    )
+
+    ds, _, _ = dond(delegate_param_1)
+    assert isinstance(ds, DataSetProtocol)
+
+    all_param_names = set(ds.description.interdeps.names)
+    assert all_param_names == {delegate_param_1.name, "delegate_setpoint_1"}