Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/changes/newsfragments/8363.underthehood
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
The duck typed scale and offset conversions in ``ParameterBase`` have been
factored out into dedicated module level helpers. The conversions assume the
data type is numeric and rely on catching ``TypeError``, which does not fit the
generic parameter data type. Giving them an explicit boundary lets the rest of
the class stay properly typed and removes twelve type checker suppressions.
There is no change in behaviour.
136 changes: 79 additions & 57 deletions src/qcodes/parameters/parameter_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -257,6 +257,62 @@ class ParameterBaseKWArgs(
"""


# The four helpers below convert between a parameter value and its raw
# counterpart. They are deliberately duck typed: they assume that the caller
# does not set ``scale``/``offset`` unless the data type is numeric, and either
# check for an iterable up front or fall back on catching ``TypeError``.
# Taking and returning ``Any`` keeps that boundary explicit, and keeps the
# generic ``ParameterDataTypeVar`` out of the arithmetic.


def _scale_raw_value(raw_value: Any, scale: float | Iterable[float]) -> Any:
"""Multiply a value by ``scale`` on the way to the instrument."""
if isinstance(scale, collections.abc.Iterable):
# Scale contains multiple elements, one for each value
return tuple(val * sub_scale for val, sub_scale in zip(raw_value, scale))
# Use single scale for all values
return raw_value * scale


def _offset_raw_value(raw_value: Any, offset: float | Iterable[float]) -> Any:
"""Add ``offset`` to a value on the way to the instrument."""
if isinstance(offset, collections.abc.Iterable):
# offset contains multiple elements, one for each value
return tuple(val + sub_offset for val, sub_offset in zip(raw_value, offset))
# Use single offset for all values
return raw_value + offset


def _unoffset_value(value: Any, offset: float | Iterable[float]) -> Any:
"""Subtract ``offset`` from a value coming back from the instrument."""
try:
return value - offset
except TypeError:
if isinstance(offset, collections.abc.Iterable):
# offset contains multiple elements, one for each value
return tuple(val - sub_offset for val, sub_offset in zip(value, offset))
elif isinstance(value, collections.abc.Iterable):
# Use single offset for all values
return tuple(val - offset for val in value)
else:
raise


def _unscale_value(value: Any, scale: float | Iterable[float]) -> Any:
"""Divide a value coming back from the instrument by ``scale``."""
try:
return value / scale
except TypeError:
if isinstance(scale, collections.abc.Iterable):
# Scale contains multiple elements, one for each value
return tuple(val / sub_scale for val, sub_scale in zip(value, scale))
elif isinstance(value, collections.abc.Iterable):
# Use single scale for all values
return tuple(val / scale for val in value)
else:
raise


class ParameterBase(
MetadatableWithName, Generic[ParameterDataTypeVar, InstrumentTypeVar_co]
):
Expand Down Expand Up @@ -363,9 +419,10 @@ def __init__(
self,
name: str,
*,
# mypy seems to be confused here. The bound and default for InstrumentTypeVar_co
# contains None but mypy will not allow None as a default as of v 1.19.0
instrument: InstrumentTypeVar_co = None, # type: ignore[assignment]
# The bound and default for InstrumentTypeVar_co contain None, but
# neither mypy (as of v1.19.0) nor ty accept None as the default for a
# parameter annotated with the type variable itself.
instrument: InstrumentTypeVar_co = None, # type: ignore[assignment] # ty: ignore[invalid-parameter-default]
snapshot_get: bool = True,
metadata: Mapping[Any, Any] | None = None,
step: float | None = None,
Expand Down Expand Up @@ -814,25 +871,11 @@ def _from_value_to_raw_value(self, value: ParameterDataTypeVar) -> ParamRawDataT
# transverse transformation in reverse order as compared to
# getter: apply scale first
if self.scale is not None:
if isinstance(self.scale, collections.abc.Iterable):
# Scale contains multiple elements, one for each value
raw_value = tuple(
val * scale for val, scale in zip(raw_value, self.scale)
)
else:
# Use single scale for all values
raw_value = raw_value * self.scale
raw_value = _scale_raw_value(raw_value, self.scale)

# apply offset next
if self.offset is not None:
if isinstance(self.offset, collections.abc.Iterable):
# offset contains multiple elements, one for each value
raw_value = tuple(
val + offset for val, offset in zip(raw_value, self.offset)
)
else:
# Use single offset for all values
raw_value = raw_value + self.offset
raw_value = _offset_raw_value(raw_value, self.offset)

# parser last
if self.set_parser is not None:
Expand All @@ -843,49 +886,23 @@ def _from_value_to_raw_value(self, value: ParameterDataTypeVar) -> ParamRawDataT
def _from_raw_value_to_value(
self, raw_value: ParamRawDataType
) -> ParameterDataTypeVar:
# ``value`` keeps the parameter's data type as its declared type; the
# offset and scale transformations below rely on duck typing and are
# therefore delegated to the helpers at the top of this module.
value: ParameterDataTypeVar

if self.get_parser is not None:
value = self.get_parser(raw_value)
else:
value = raw_value

# the code below is not very type safe but relies on duck typing / try except
# and assumes the user does not set scale/offset unless the datatype is numeric
# this should probably be rewritten but for now we ignore type errors
# apply offset first (native scale)

if self.offset is not None and value is not None:
# offset values
try:
value = value - self.offset # type: ignore[operator,assignment]
except TypeError:
if isinstance(self.offset, collections.abc.Iterable):
# offset contains multiple elements, one for each value
value = tuple( # type: ignore[assignment]
val - offset
for val, offset in zip(value, self.offset) # type: ignore[call-overload]
)
elif isinstance(value, collections.abc.Iterable):
# Use single offset for all values
value = tuple(val - self.offset for val in value) # type: ignore[assignment]
else:
raise
value = _unoffset_value(value, self.offset)

# scale second
if self.scale is not None and value is not None:
# Scale values
try:
value = value / self.scale # type: ignore[assignment,operator]
except TypeError:
if isinstance(self.scale, collections.abc.Iterable):
# Scale contains multiple elements, one for each value
value = tuple(val / scale for val, scale in zip(value, self.scale)) # type: ignore[call-overload,assignment]
elif isinstance(value, collections.abc.Iterable):
# Use single scale for all values
value = tuple(val / self.scale for val in value) # type: ignore[assignment]
else:
raise
value = _unscale_value(value, self.scale)

if self.inverse_val_mapping is not None:
if value in self.inverse_val_mapping:
Expand All @@ -896,7 +913,7 @@ def _from_raw_value_to_value(
except (ValueError, KeyError):
raise KeyError(f"'{value}' not in val_mapping")

return value # pyright: ignore[reportReturnType]
return value

def _wrap_get(
self, get_function: Callable[..., ParamRawDataType]
Expand Down Expand Up @@ -946,14 +963,16 @@ def set_wrapper(value: ParameterDataTypeVar, **kwargs: Any) -> None:
# In some cases intermediate sweep values must be used.
# Unless `self.step` is defined, get_sweep_values will return
# a list containing only `value`.
steps = self.get_ramp_values(value, step=self.step) # type: ignore[arg-type]
# The steps are deliberately untyped: ``get_ramp_values`` works
# in terms of numbers rather than the parameter's data type.
steps: Sequence[Any] = self.get_ramp_values(value, step=self.step) # type: ignore[arg-type]

for val_step in steps:
# even if the final value is valid we may be generating
# steps that are not so validate them too
self.validate(val_step) # type: ignore[arg-type]
self.validate(val_step)

raw_val_step = self._from_value_to_raw_value(val_step) # type: ignore[arg-type]
raw_val_step = self._from_value_to_raw_value(val_step)

# Check if delay between set operations is required
t_elapsed = time.perf_counter() - self._t_last_set
Expand All @@ -976,9 +995,9 @@ def set_wrapper(value: ParameterDataTypeVar, **kwargs: Any) -> None:
# Sleep until total time is larger than self.post_delay
time.sleep(self.post_delay - t_elapsed)

self.cache._update_with(value=val_step, raw_value=raw_val_step) # type: ignore[arg-type]
self.cache._update_with(value=val_step, raw_value=raw_val_step)

self._call_on_set_callback(val_step) # type: ignore[arg-type]
self._call_on_set_callback(val_step)

except Exception as e:
e.args = (*e.args, f"setting {self} to {value}")
Expand Down Expand Up @@ -1548,7 +1567,10 @@ def issubset(self, other: ParameterSet[P] | set) -> bool:
return all(item in other for item in self)

def issuperset(self, other: ParameterSet[P] | set) -> bool:
return all(item in self for item in other)
# ``item in self._dict`` rather than ``item in self`` because ty fails to
# resolve ``__contains__`` on ``Self`` when the class type parameter has
# a bound. ``__contains__`` delegates to ``_dict`` anyway.
return all(item in self._dict for item in other)

def update(self, other: Iterable[P]) -> None:
for item in other:
Expand Down
120 changes: 119 additions & 1 deletion tests/parameter/test_parameter_scale_offset.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,18 @@
from collections.abc import Iterable
from collections.abc import Callable, Iterable
from typing import Any

import hypothesis.strategies as hst
import numpy as np
import pytest
from hypothesis import event, given, settings

from qcodes.parameters import Parameter
from qcodes.parameters.parameter_base import (
_offset_raw_value,
_scale_raw_value,
_unoffset_value,
_unscale_value,
)


def test_scale_raw_value() -> None:
Expand Down Expand Up @@ -215,3 +223,113 @@ def test_setting_numpy_array_valued_param_if_scale_and_offset_are_not_none() ->
param(values)

assert isinstance(param.raw_value, np.ndarray)


def test_set_with_iterable_scale_and_offset() -> None:
"""Setting a sequence with per element scale and offset."""
param = Parameter(name="test_param", set_cmd=None, get_cmd=None)
param.scale = [2, 4]
param.offset = [1, 2]

param([10, 20])

# scale is applied first, offset second
assert param.raw_value == (21, 82)
# and reversed on the way back out
assert param.get() == (10, 20)


def test_set_with_iterable_scale_and_scalar_offset() -> None:
param = Parameter(name="test_param", set_cmd=None, get_cmd=None)
param.scale = [2, 4]
param.offset = np.array([1, 1])

param(np.array([10, 20]))

np.testing.assert_allclose(np.array(param.raw_value), [21, 81])
np.testing.assert_allclose(np.array(param.get()), [10, 20])


def test_set_numpy_array_with_scalar_scale_and_offset() -> None:
param = Parameter(name="test_param", set_cmd=None, get_cmd=None)
param.scale = 2
param.offset = 1

param(np.array([10, 20]))

np.testing.assert_allclose(param.raw_value, [21, 41])
np.testing.assert_allclose(param.get(), [10, 20])


def test_get_sequence_with_scalar_scale_and_offset() -> None:
"""A list valued raw value falls back on element wise arithmetic."""
param = Parameter(name="test_param", set_cmd=None, get_cmd=lambda: [10, 20])
param.scale = 2
param.offset = 4

assert param.get() == (3.0, 8.0)


def test_get_sequence_with_iterable_scale_and_offset() -> None:
param = Parameter(name="test_param", set_cmd=None, get_cmd=lambda: [10, 20])
param.scale = [2, 4]
param.offset = [4, 8]

assert param.get() == (3.0, 3.0)


@pytest.mark.parametrize("attribute", ["scale", "offset"])
def test_get_raises_for_non_numeric_value(attribute: str) -> None:
"""A non iterable value that cannot be scaled/offset re-raises TypeError."""
sentinel = object()
param: Parameter = Parameter(
name="test_param", set_cmd=None, get_cmd=lambda: sentinel
)
setattr(param, attribute, 2)

with pytest.raises(TypeError):
param.get()


def test_scale_raw_value_helper() -> None:
assert _scale_raw_value(10, 2) == 20
assert _scale_raw_value([10, 20], [2, 4]) == (20, 80)
np.testing.assert_allclose(_scale_raw_value(np.array([10, 20]), 2), [20, 40])


def test_offset_raw_value_helper() -> None:
assert _offset_raw_value(10, 2) == 12
assert _offset_raw_value([10, 20], [2, 4]) == (12, 24)
np.testing.assert_allclose(_offset_raw_value(np.array([10, 20]), 2), [12, 22])


def test_unoffset_value_helper() -> None:
assert _unoffset_value(10, 2) == 8
assert _unoffset_value([10, 20], [2, 4]) == (8, 16)
assert _unoffset_value([10, 20], 2) == (8, 18)
np.testing.assert_allclose(_unoffset_value(np.array([10, 20]), 2), [8, 18])

with pytest.raises(TypeError):
_unoffset_value(object(), 2)


def test_unscale_value_helper() -> None:
assert _unscale_value(10, 2) == 5
assert _unscale_value([10, 20], [2, 4]) == (5, 5)
assert _unscale_value([10, 20], 2) == (5, 10)
np.testing.assert_allclose(_unscale_value(np.array([10, 20]), 2), [5, 10])

with pytest.raises(TypeError):
_unscale_value(object(), 2)


@pytest.mark.parametrize(
"helper", [_scale_raw_value, _offset_raw_value, _unoffset_value, _unscale_value]
)
def test_helpers_do_not_mutate_their_input(
helper: Callable[[Any, Any], Any],
) -> None:
"""The helpers must not mutate the value they are handed."""
value = [10.0, 20.0]
helper(value, [2.0, 4.0])
assert value == [10.0, 20.0]
27 changes: 27 additions & 0 deletions tests/parameter/test_parameter_set.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,3 +86,30 @@ def test_parameter_set_operations(
intersection_set = ParameterSet([param1, param2]) & ParameterSet([param2, param3])
assert len(intersection_set) == 1
assert param2 in intersection_set


def test_parameter_set_issubset_and_issuperset(
manual_parameters: tuple[ManualParameter, ...],
) -> None:
param1, param2, param3 = manual_parameters
full_set = ParameterSet((param1, param2, param3))
subset = ParameterSet((param1, param2))
disjoint = ParameterSet((param3,))

assert subset.issubset(full_set)
assert full_set.issuperset(subset)
assert full_set.issuperset(full_set)
assert subset.issubset(subset)

assert not full_set.issubset(subset)
assert not subset.issuperset(full_set)
assert not subset.issuperset(disjoint)

# plain sets are accepted on both sides
assert full_set.issuperset({param1, param3})
assert subset.issubset({param1, param2, param3})

empty: ParameterSet[ManualParameter] = ParameterSet()
assert empty.issubset(full_set)
assert full_set.issuperset(empty)
assert not empty.issuperset(full_set)
Loading