from __future__ import annotations
import collections.abc
import logging
import time
import warnings
from collections.abc import Iterator, MutableSet
from contextlib import contextmanager
from datetime import datetime
from functools import cached_property, wraps
from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypeVar, overload
import numpy as np
from qcodes.metadatable import Metadatable, MetadatableWithName
from qcodes.parameters import ParamSpecBase
from qcodes.utils import DelegateAttributes, full_class, qcodes_abstractmethod
from qcodes.validators import (
Arrays,
ComplexNumbers,
Enum,
Ints,
Numbers,
Strings,
Validator,
)
from ..utils.types import NumberType
from .cache import _Cache, _CacheProtocol
from .named_repr import named_repr
from .permissive_range import permissive_range
# for now the type the parameter may contain is not restricted at all
ParamDataType = Any
ParamRawDataType = Any
if TYPE_CHECKING:
from collections.abc import Callable, Generator, Iterable, Mapping, Sequence, Sized
from types import TracebackType
from qcodes.dataset.data_set_protocol import ValuesType
from qcodes.instrument import InstrumentBase
from qcodes.logger.instrument_logger import InstrumentLoggerAdapter
LOG = logging.getLogger(__name__)
class _SetParamContext:
"""
This class is returned by the ``set_to`` method of parameter
Example usage:
>>> v = dac.voltage()
>>> with dac.voltage.set_to(-1):
... # Do stuff with the DAC output set to -1 V.
...
>>> assert abs(dac.voltage() - v) <= tolerance
"""
def __init__(
self,
parameter: ParameterBase,
value: ParamDataType,
allow_changes: bool = False,
):
self._parameter: ParameterBase = parameter
self._value = value
self._allow_changes = allow_changes
self._original_value = None
self._original_settable: bool | None = None
def __enter__(self) -> None:
self._original_value = self._parameter.cache()
if self._original_value != self._value:
self._parameter.set(self._value)
if not self._allow_changes:
self._original_settable = self._parameter.settable
self._parameter._settable = False
def __exit__(
self,
typ: type[BaseException] | None,
value: BaseException | None,
traceback: TracebackType | None,
) -> None:
if not self._allow_changes:
assert self._original_settable is not None
self._parameter._settable = self._original_settable
if self._parameter.cache() != self._original_value:
try:
self._parameter.set(self._original_value)
except Exception:
# Likely an uninitialized Parameter
LOG.info(
"Encountered an exception setting the original value "
"when exiting set_to context of "
f"{self._parameter.full_name}",
exc_info=True,
)
[docs]
def invert_val_mapping(val_mapping: Mapping[Any, Any]) -> dict[Any, Any]:
"""Inverts the value mapping dictionary for allowed parameter values"""
return {v: k for k, v in val_mapping.items()}
[docs]
class ParameterBase(MetadatableWithName):
"""
Shared behavior for all parameters. Not intended to be used
directly, normally you should use ``Parameter``, ``ArrayParameter``,
``MultiParameter``, or ``CombinedParameter``.
Note that ``CombinedParameter`` is not yet a subclass of ``ParameterBase``
Args:
name: the local name of the parameter. Must be a valid
identifier, ie no spaces or special characters or starting with a
number. If this parameter is part of an Instrument or Station,
this should match how it will be referenced from that parent,
ie ``instrument.name`` or ``instrument.parameters[name]``
instrument: the instrument this parameter
belongs to, if any
snapshot_get: False prevents any update to the
parameter during a snapshot, even if the snapshot was called with
``update=True``, for example if it takes too long to update.
Default True.
snapshot_value: False prevents parameter value to be
stored in the snapshot. Useful if the value is large.
snapshot_exclude: True prevents parameter to be
included in the snapshot. Useful if there are many of the same
parameter which are clogging up the snapshot.
Default False
step: max increment of parameter value.
Larger changes are broken into multiple steps this size.
When combined with delays, this acts as a ramp.
scale: Scale to multiply value with before
performing set. the internally multiplied value is stored in
``cache.raw_value``. Can account for a voltage divider.
offset: Compensate for a parameter specific offset. (just as scale)
get value = raw value - offset.
set value = argument + offset.
If offset and scale are used in combination, when getting a value,
first an offset is added, then the scale is applied.
inter_delay: Minimum time (in seconds) between successive sets.
If the previous set was less than this, it will wait until the
condition is met. Can be set to 0 to go maximum speed with
no errors.
post_delay: time (in seconds) to wait after the *start* of each set,
whether part of a sweep or not. Can be set to 0 to go maximum
speed with no errors.
val_mapping: A bidirectional map data/readable values to instrument
codes, expressed as a dict: ``{data_val: instrument_code}``
For example, if the instrument uses '0' to mean 1V and '1' to mean
10V, set val_mapping={1: '0', 10: '1'} and on the user side you
only see 1 and 10, never the coded '0' and '1'
If vals is omitted, will also construct a matching Enum validator.
NOTE: only applies to get if get_cmd is a string, and to set if
set_cmd is a string.
You can use ``val_mapping`` with ``get_parser``, in which case
``get_parser`` acts on the return value from the instrument first,
then ``val_mapping`` is applied (in reverse).
get_parser: Function to transform the response from get to the final
output value. See also val_mapping
set_parser: Function to transform the input set value to an encoded
value sent to the instrument. See also val_mapping.
vals: a Validator object for this parameter
max_val_age: The max time (in seconds) to trust a saved value obtained
from ``cache.get`` (or ``get_latest``). If this parameter has not
been set or measured more recently than this, perform an
additional measurement.
metadata: extra information to include with the
JSON snapshot of the parameter
abstract: Specifies if this parameter is abstract or not. Default
is False. If the parameter is 'abstract', it *must* be overridden
by a non-abstract parameter before the instrument containing
this parameter can be instantiated. We override a parameter by
adding one with the same name and unit. An abstract parameter
can be added in a base class and overridden in a subclass.
bind_to_instrument: Should the parameter be registered as a delegate attribute
on the instrument passed via the instrument argument.
register_name: Specifies if the parameter should be registered in datasets
using a different name than the parameter's full_name
"""
global_on_set_callback: ClassVar[
Callable[[ParameterBase, ParamDataType], None] | None
] = None
def __init__(
self,
name: str,
instrument: InstrumentBase | None,
snapshot_get: bool = True,
metadata: Mapping[Any, Any] | None = None,
step: float | None = None,
scale: float | Iterable[float] | None = None,
offset: float | Iterable[float] | None = None,
inter_delay: float = 0,
post_delay: float = 0,
val_mapping: Mapping[Any, Any] | None = None,
get_parser: Callable[..., Any] | None = None,
set_parser: Callable[..., Any] | None = None,
snapshot_value: bool = True,
snapshot_exclude: bool = False,
max_val_age: float | None = None,
vals: Validator[Any] | None = None,
abstract: bool | None = False,
bind_to_instrument: bool = True,
register_name: str | None = None,
on_set_callback: Callable[[ParameterBase, ParamDataType], None] | None = None,
) -> None:
super().__init__(metadata)
if not str(name).isidentifier():
raise ValueError(
f"Parameter name must be a valid identifier "
f"got {name} which is not. Parameter names "
f"cannot start with a number and "
f"must not contain spaces or special characters"
)
self._short_name = str(name)
self._register_name = register_name
self._instrument = instrument
self._snapshot_get = snapshot_get
self._snapshot_value = snapshot_value
self.snapshot_exclude = snapshot_exclude
self.on_set_callback = on_set_callback
self._depends_on: ParameterSet = ParameterSet()
self._has_control_of: ParameterSet = ParameterSet()
self._is_controlled_by: ParameterSet = ParameterSet()
self._param_spec: ParamSpecBase | None = None
if not isinstance(vals, (Validator, type(None))):
raise TypeError("vals must be None or a Validator")
elif val_mapping is not None:
vals = Enum(*val_mapping.keys())
if vals is not None:
self._vals: list[Validator[Any]] = [vals]
else:
self._vals = []
self.step = step
self.scale = scale
self.offset = offset
self.inter_delay = inter_delay
self.post_delay = post_delay
self.val_mapping = val_mapping
if val_mapping is None:
self.inverse_val_mapping = None
else:
self.inverse_val_mapping = invert_val_mapping(val_mapping)
self.get_parser: Callable[..., Any] | None = get_parser
self.set_parser: Callable[..., Any] | None = set_parser
# ``_Cache`` stores "latest" value (and raw value) and timestamp
# when it was set or measured
self.cache: _CacheProtocol = _Cache(self, max_val_age=max_val_age)
# ``GetLatest`` is left from previous versions where it would
# implement a subset of features which ``_Cache`` has.
# It is left for now for backwards compatibility reasons and shall
# be deprecated and removed in the future versions.
self.get_latest: GetLatest
self.get_latest = GetLatest(self)
self.get: Callable[..., ParamDataType]
self._gettable = False
if self._implements_get_raw:
self.get = self._wrap_get(self.get_raw)
self._gettable = True
elif hasattr(self, "get"):
raise RuntimeError(
f"Overwriting get in a subclass of "
f"ParameterBase: "
f"{self.full_name} is not allowed."
)
self.set: Callable[..., None]
self._settable: bool = False
if self._implements_set_raw:
self.set = self._wrap_set(self.set_raw)
self._settable = True
elif hasattr(self, "set"):
raise RuntimeError(
f"Overwriting set in a subclass of "
f"ParameterBase: "
f"{self.full_name} is not allowed."
)
# subclasses should extend this list with extra attributes they
# want automatically included in the snapshot
self._meta_attrs = [
"name",
"instrument",
"step",
"scale",
"offset",
"inter_delay",
"post_delay",
"val_mapping",
"vals",
"validators",
]
# Specify time of last set operation, used when comparing to delay to
# check if additional waiting time is needed before next set
self._t_last_set = time.perf_counter()
# should we call validate when getting data. default to False
# intended to be changed in a subclass if you want the subclass
# to perform a validation on get
self._validate_on_get: bool = False
self._abstract = abstract
if instrument is not None and bind_to_instrument:
found_as_delegate = instrument.parameters.get(name, False)
# we allow properties since a pattern that has been seen in the wild
# is properties that are used to wrap parameters of the same name
# to define an interface for the instrument
is_property = isinstance(
getattr(instrument.__class__, name, None), property
)
found_as_attr = not is_property and hasattr(instrument, name)
if found_as_delegate or found_as_attr:
existing_parameter = instrument.parameters.get(name, None)
if existing_parameter is not None and not existing_parameter.abstract:
raise KeyError(
f"Duplicate parameter name {name} on instrument {instrument}"
)
if existing_parameter is None:
existing_attribute = getattr(instrument, name, None)
if isinstance(existing_attribute, ParameterBase):
raise KeyError(
f"Duplicate parameter name {name} on instrument {instrument}"
)
elif existing_attribute is not None:
warnings.warn(
f"Parameter {name} overrides an attribute of the same name on instrument {instrument} "
"This will be an error in the future.",
)
instrument.parameters[name] = self
@property
def _implements_get_raw(self) -> bool:
implements_get_raw = hasattr(self, "get_raw") and not getattr(
self.get_raw, "__qcodes_is_abstract_method__", False
)
return implements_get_raw
@property
def _implements_set_raw(self) -> bool:
implements_set_raw = hasattr(self, "set_raw") and not getattr(
self.set_raw, "__qcodes_is_abstract_method__", False
)
return implements_set_raw
def _get_logger(self) -> InstrumentLoggerAdapter | logging.Logger:
if self.root_instrument is not None:
mylogger: InstrumentLoggerAdapter | logging.Logger = (
self.root_instrument.log
)
else:
mylogger = LOG
return mylogger
def _build__doc__(self) -> str | None:
return self.__doc__
@property
def vals(self) -> Validator | None:
"""
The first validator of the parameter. None
if no validators are set for this parameter.
:getter: Returns the first validator or None if no validators.
:setter: Sets the first validator. Set to None to remove the first validator.
Raises:
RuntimeError: If removing the first validator when more than one validator is set.
"""
validators = self.validators
if len(validators):
return validators[0]
else:
return None
@vals.setter
def vals(self, vals: Validator | None) -> None:
if vals is not None and len(self._vals) > 0:
self._vals[0] = vals
elif vals is not None:
self._vals = [vals]
elif len(self._vals) == 1:
self._vals = []
elif len(self._vals) > 1:
raise RuntimeError(
"Cannot remove default validator from parameter with additional validators."
)
else:
# setting the validator to None but the parameter already doesn't have a validator
pass
self.__doc__ = self._build__doc__()
[docs]
def add_validator(self, vals: Validator) -> None:
"""Add a validator for the parameter. The parameter is validated against
all validators in reverse order of how they are added.
Args:
vals: Validator to add to the parameter.
"""
self._vals.append(vals)
self.__doc__ = self._build__doc__()
[docs]
def remove_validator(self) -> Validator | None:
"""
Remove the last validator added to the parameter and return it.
Returns None if there are no validators associated with the parameter.
Returns:
The last validator added to the parameter or None if there are no
validators associated with the parameter.
"""
if len(self._vals) > 0:
removed = self._vals.pop()
self.__doc__ = self._build__doc__()
return removed
else:
return None
@property
def validators(self) -> tuple[Validator, ...]:
"""
Tuple of all validators associated with the parameter.
:getter: All validators associated with the parameter.
"""
return tuple(self._vals)
@property
def raw_value(self) -> ParamRawDataType:
"""
Note that this property will be deprecated soon. Use
``cache.raw_value`` instead.
Represents the cached raw value of the parameter.
:getter: Returns the cached raw value of the parameter.
"""
return self.cache.raw_value
[docs]
@qcodes_abstractmethod
def get_raw(self) -> ParamRawDataType:
"""
``get_raw`` is called to perform the actual data acquisition from the
instrument. This method should either be overwritten to perform the
desired operation or alternatively for :class:`.Parameter` a
suitable method is automatically generated if ``get_cmd`` is supplied
to the parameter constructor. The method is automatically wrapped to
provide a ``get`` method on the parameter instance.
"""
raise NotImplementedError
[docs]
@qcodes_abstractmethod
def set_raw(self, value: ParamRawDataType) -> None:
"""
``set_raw`` is called to perform the actual setting of a parameter on
the instrument. This method should either be overwritten to perform the
desired operation or alternatively for :class:`.Parameter` a
suitable method is automatically generated if ``set_cmd`` is supplied
to the parameter constructor. The method is automatically wrapped to
provide a ``set`` method on the parameter instance.
"""
raise NotImplementedError
[docs]
def __str__(self) -> str:
"""Include the instrument name with the Parameter name if possible."""
inst_name = getattr(self._instrument, "name", "")
if inst_name:
return f"{inst_name}_{self.name}"
else:
return self.name
def __repr__(self) -> str:
return named_repr(self)
@overload
def __call__(self) -> ParamDataType:
pass
@overload
def __call__(self, value: ParamDataType, **kwargs: Any) -> None:
pass
def __call__(self, *args: Any, **kwargs: Any) -> ParamDataType | None:
if len(args) == 0 and len(kwargs) == 0:
if self.gettable:
return self.get()
else:
raise NotImplementedError(f"no get cmd found in Parameter {self.name}")
elif self.settable:
self.set(*args, **kwargs)
return None
else:
raise NotImplementedError(f"no set cmd found in Parameter {self.name}")
[docs]
def snapshot_base(
self,
update: bool | None = True,
params_to_skip_update: Sequence[str] | None = None,
) -> dict[Any, Any]:
"""
State of the parameter as a JSON-compatible dict (everything that
the custom JSON encoder class
:class:`.NumpyJSONEncoder` supports).
If the parameter has been initiated with ``snapshot_value=False``,
the snapshot will NOT include the ``value`` and ``raw_value`` of the
parameter.
Args:
update: If True, update the state by calling ``parameter.get()``
unless ``snapshot_get`` of the parameter is ``False``.
If ``update`` is ``None``, use the current value from the
``cache`` unless the cache is invalid. If ``False``, never call
``parameter.get()``.
params_to_skip_update: No effect but may be passed from superclass
Returns:
base snapshot
"""
if self.snapshot_exclude:
warnings.warn(
f"Parameter ({self.full_name}) is used in the snapshot while it "
f"should be excluded from the snapshot",
stacklevel=2,
)
state: dict[str, Any] = {"__class__": full_class(self), "full_name": str(self)}
if self.snapshot_value:
has_get = self.gettable
allowed_to_call_get_when_snapshotting = (
self._snapshot_get and update is not False
)
can_call_get_when_snapshotting = (
allowed_to_call_get_when_snapshotting and has_get
)
if can_call_get_when_snapshotting and update:
state["value"] = self.get()
else:
state["value"] = self.cache.get(
get_if_invalid=can_call_get_when_snapshotting
)
state["raw_value"] = self.cache.raw_value
state["ts"] = self.cache.timestamp
if isinstance(state["ts"], datetime):
dttime: datetime = state["ts"]
state["ts"] = dttime.strftime("%Y-%m-%d %H:%M:%S")
for attr in set(self._meta_attrs):
if attr == "instrument" and self._instrument:
state.update(
{
"instrument": full_class(self._instrument),
"instrument_name": self._instrument.name,
}
)
elif attr == "validators":
state["validators"] = [repr(validator) for validator in self.validators]
else:
val = getattr(self, attr, None)
if val is not None:
attr_strip = attr.lstrip("_") # strip leading underscores
if isinstance(val, Validator):
state[attr_strip] = repr(val)
elif isinstance(val, Metadatable):
state[attr_strip] = val.snapshot(update=update)
else:
state[attr_strip] = val
return state
@property
def snapshot_value(self) -> bool:
"""
If True the value of the parameter will be included in the snapshot.
"""
return self._snapshot_value
def _from_value_to_raw_value(self, value: ParamDataType) -> ParamRawDataType:
raw_value: ParamRawDataType
if self.val_mapping is not None:
# Convert set values using val_mapping dictionary
raw_value = self.val_mapping[value]
else:
raw_value = value
# 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
# 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
# parser last
if self.set_parser is not None:
raw_value = self.set_parser(raw_value)
return raw_value
def _from_raw_value_to_value(self, raw_value: ParamRawDataType) -> ParamDataType:
value: ParamDataType
if self.get_parser is not None:
value = self.get_parser(raw_value)
else:
value = raw_value
# apply offset first (native scale)
if self.offset is not None and value is not None:
# offset values
try:
value = value - self.offset
except TypeError:
if isinstance(self.offset, collections.abc.Iterable):
# offset contains multiple elements, one for each value
value = tuple(
val - offset for val, offset in zip(value, self.offset)
)
elif isinstance(value, collections.abc.Iterable):
# Use single offset for all values
value = tuple(val - self.offset for val in value)
else:
raise
# scale second
if self.scale is not None and value is not None:
# Scale values
try:
value = value / self.scale
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))
elif isinstance(value, collections.abc.Iterable):
# Use single scale for all values
value = tuple(val / self.scale for val in value)
else:
raise
if self.inverse_val_mapping is not None:
if value in self.inverse_val_mapping:
value = self.inverse_val_mapping[value]
else:
try:
value = self.inverse_val_mapping[int(value)]
except (ValueError, KeyError):
raise KeyError(f"'{value}' not in val_mapping")
return value
def _wrap_get(
self, get_function: Callable[..., ParamRawDataType]
) -> Callable[..., ParamDataType]:
@wraps(get_function)
def get_wrapper(*args: Any, **kwargs: Any) -> ParamDataType:
if not self.gettable:
raise TypeError("Trying to get a parameter that is not gettable.")
if self.abstract:
raise NotImplementedError(
f"Trying to get an abstract parameter: {self.full_name}"
)
try:
# There might be cases where a .get also has args/kwargs
raw_value = get_function(*args, **kwargs)
value = self._from_raw_value_to_value(raw_value)
if self._validate_on_get:
self.validate(value)
self.cache._update_with(value=value, raw_value=raw_value)
return value
except Exception as e:
e.args = (*e.args, f"getting {self}")
raise e
return get_wrapper
def _wrap_set(self, set_function: Callable[..., None]) -> Callable[..., None]:
@wraps(set_function)
def set_wrapper(value: ParamDataType, **kwargs: Any) -> None:
try:
if not self.settable:
raise TypeError("Trying to set a parameter that is not settable.")
if self.abstract:
raise NotImplementedError(
f"Trying to set an abstract parameter: {self.full_name}"
)
self.validate(value)
# 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)
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)
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
if t_elapsed < self.inter_delay:
# Sleep until time since last set is larger than
# self.inter_delay
time.sleep(self.inter_delay - t_elapsed)
# Start timer to measure execution time of set_function
t0 = time.perf_counter()
set_function(raw_val_step, **kwargs)
# Update last set time (used for calculating delays)
self._t_last_set = time.perf_counter()
# Check if any delay after setting is required
t_elapsed = self._t_last_set - t0
if t_elapsed < self.post_delay:
# 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)
self._call_on_set_callback(val_step)
except Exception as e:
e.args = (*e.args, f"setting {self} to {value}")
raise e
return set_wrapper
def _call_on_set_callback(self, value: ParamDataType) -> None:
try:
if self.on_set_callback is not None:
self.on_set_callback(self, value)
elif self.__class__.global_on_set_callback is not None:
self.__class__.global_on_set_callback(self, value)
except Exception as e:
LOG.warning(
f"Exception {e} in on set callback "
f"for {self.full_name} with value {value}",
exc_info=True,
)
[docs]
def get_ramp_values(
self, value: NumberType | Sized, step: NumberType | None = None
) -> Sequence[NumberType | Sized]:
"""
Return values to sweep from current value to target value.
This method can be overridden to have a custom sweep behaviour.
It can even be overridden by a generator.
Args:
value: target value
step: maximum step size
Returns:
List of stepped values, including target value.
"""
if step is None:
return [value]
else:
if isinstance(value, collections.abc.Sized) and len(value) > 1:
raise RuntimeError(
"Don't know how to step a parameter with more than one value"
)
if self.get_latest() is None:
self.get()
start_value = self.get_latest()
if not (
isinstance(start_value, NumberType) and isinstance(value, NumberType)
):
# parameter is numeric but either one of the endpoints
# is not or the starting point is unknown. The later
# can happen for a non gettable parameter in the initial set
# operation.
LOG.warning(
f"cannot sweep {self.name} from {start_value!r} "
f"to {value!r} - jumping."
)
return [value]
# drop the initial value, we're already there
return [*permissive_range(start_value, value, step)[1:], value]
@cached_property
def _validate_context(self) -> str:
# return string describing the context for a validator
if self._instrument:
context = (
(
getattr(self._instrument, "name", "")
or str(self._instrument.__class__)
)
+ "."
+ self.name
)
else:
context = self.name
return "Parameter: " + context
[docs]
def validate(self, value: ParamDataType) -> None:
"""
Validate the value supplied.
Args:
value: value to validate
Raises:
TypeError: If the value is of the wrong type.
ValueError: If the value is outside the bounds specified by the
validator.
"""
for validator in reversed(self._vals):
if validator is not None:
validator.validate(value, self._validate_context)
@property
def step(self) -> NumberType | None:
"""
Stepsize that this Parameter uses during set operation.
Stepsize must be a positive number or None.
If step is a positive number, this is the maximum value change
allowed in one hardware call, so a single set can result in many
calls to the hardware if the starting value is far from the target.
All but the final change will attempt to change by +/- step exactly.
If step is None stepping will not be used.
:getter: Returns the current stepsize.
:setter: Sets the value of the step.
Raises:
TypeError: if step is set to not numeric or None
ValueError: if step is set to negative
TypeError: if step is set to not integer or None for an
integer parameter
TypeError: if step is set to not a number on None
"""
return self._step
@step.setter
def step(self, step: NumberType | None) -> None:
if step is None:
self._step: NumberType | None = step
elif not all(getattr(vals, "is_numeric", True) for vals in self._vals):
raise TypeError("you can only step numeric parameters")
elif not isinstance(step, NumberType):
raise TypeError("step must be a number")
elif step == 0:
self._step = None
elif step <= 0:
raise ValueError("step must be positive")
elif any(isinstance(vals, Ints) for vals in self._vals) and not isinstance(
step, int
):
raise TypeError("step must be a positive int for an Ints parameter")
else:
self._step = step
@property
def post_delay(self) -> float:
"""
Delay time after *start* of set operation, for each set.
The actual time will not be shorter than this, but may be longer
if the underlying set call takes longer.
Typically used in conjunction with `step` to create an effective
ramp rate, but can also be used without a `step` to enforce a delay
*after* every set. One might think of post_delay as how long a set
operation is supposed to take. For example, there might be an
instrument that needs extra time after setting a parameter although
the command for setting the parameter returns quickly.
:getter: Returns the current post_delay.
:setter: Sets the value of the post_delay.
Raises:
TypeError: If delay is not int nor float
ValueError: If delay is negative
"""
return self._post_delay
@post_delay.setter
def post_delay(self, post_delay: float) -> None:
if not isinstance(post_delay, NumberType):
raise TypeError(f"post_delay ({post_delay}) must be a number")
if post_delay < 0:
raise ValueError(f"post_delay ({post_delay}) must not be negative")
self._post_delay = post_delay
@property
def inter_delay(self) -> float:
"""
Delay time between consecutive set operations.
The actual time will not be shorter than this, but may be longer
if the underlying set call takes longer.
Typically used in conjunction with `step` to create an effective
ramp rate, but can also be used without a `step` to enforce a delay
*between* sets.
:getter: Returns the current inter_delay.
:setter: Sets the value of the inter_delay.
Raises:
TypeError: If delay is not int nor float
ValueError: If delay is negative
"""
return self._inter_delay
@inter_delay.setter
def inter_delay(self, inter_delay: float) -> None:
if not isinstance(inter_delay, NumberType):
raise TypeError(f"inter_delay ({inter_delay}) must be a number")
if inter_delay < 0:
raise ValueError(f"inter_delay ({inter_delay}) must not be negative")
self._inter_delay = inter_delay
@property
def name(self) -> str:
"""Name of the parameter. This is identical to :meth:`short_name`."""
return self._short_name
@property
def short_name(self) -> str:
"""Short name of the parameter. This is without the name of the
instrument or submodule that the parameter may be bound to. For
full name refer to :meth:`full_name`."""
return self._short_name
@property
def full_name(self) -> str:
"""
Name of the parameter including the name of the instrument and
submodule that the parameter may be bound to. The names are separated
by underscores, like this: ``instrument_submodule_parameter``.
"""
return "_".join(self.name_parts)
@property
def register_name(self) -> str:
"""
Name that will be used to register this parameter in a dataset
By default, this returns ``full_name`` or the value of the
``register_name`` argument if it was passed at initialization.
"""
return self._register_name or self.full_name
@property
def instrument(self) -> InstrumentBase | None:
"""
Return the first instrument that this parameter is bound to.
E.g if this is bound to a channel it will return the channel
and not the instrument that the channel is bound too. Use
:meth:`root_instrument` to get the real instrument.
"""
return self._instrument
@property
def root_instrument(self) -> InstrumentBase | None:
"""
Return the fundamental instrument that this parameter belongs too.
E.g if the parameter is bound to a channel this will return the
fundamental instrument that that channel belongs to. Use
:meth:`instrument` to get the channel.
"""
if self._instrument is not None:
return self._instrument.root_instrument
else:
return None
[docs]
def set_to(
self, value: ParamDataType, allow_changes: bool = False
) -> _SetParamContext:
"""
Use a context manager to temporarily set a parameter to a value. By
default, the parameter value cannot be changed inside the context.
This may be overridden with ``allow_changes=True``.
Examples:
>>> from qcodes.parameters import Parameter
>>> p = Parameter("p", set_cmd=None, get_cmd=None)
>>> p.set(2)
>>> with p.set_to(3):
... print(f"p value in with block {p.get()}") # prints 3
... p.set(5) # raises an exception
>>> print(f"p value outside with block {p.get()}") # prints 2
>>> with p.set_to(3, allow_changes=True):
... p.set(5) # now this works
>>> print(f"value after second block: {p.get()}") # still prints 2
"""
context_manager = _SetParamContext(self, value, allow_changes=allow_changes)
return context_manager
[docs]
def restore_at_exit(self, allow_changes: bool = True) -> _SetParamContext:
"""
Use a context manager to restore the value of a parameter after a
``with`` block.
By default, the parameter value may be changed inside the block, but
this can be prevented with ``allow_changes=False``. This can be
useful, for example, for debugging a complex measurement that
unintentionally modifies a parameter.
Example:
>>> p = Parameter("p", set_cmd=None, get_cmd=None)
>>> p.set(2)
>>> with p.restore_at_exit():
... p.set(3)
... print(f"value inside with block: {p.get()}") # prints 3
>>> print(f"value after with block: {p.get()}") # prints 2
>>> with p.restore_at_exit(allow_changes=False):
... p.set(5) # raises an exception
"""
return self.set_to(self.cache(), allow_changes=allow_changes)
@property
def name_parts(self) -> list[str]:
"""
List of the parts that make up the full name of this parameter
"""
if self.instrument is not None:
name_parts = getattr(self.instrument, "name_parts", [])
if name_parts == []:
# add fallback for the case where someone has bound
# the parameter to something that is not an instrument
# but perhaps it has a name anyway?
name = getattr(self.instrument, "name", None)
if name is not None:
name_parts = [name]
else:
name_parts = []
name_parts.append(self.short_name)
return name_parts
@property
def gettable(self) -> bool:
"""
Is it allowed to call get on this parameter?
"""
return self._gettable
@property
def settable(self) -> bool:
"""
Is it allowed to call set on this parameter?
"""
return self._settable
@property
def underlying_instrument(self) -> InstrumentBase | None:
"""
Returns an instance of the underlying hardware instrument that this
parameter communicates with, per this parameter's implementation.
This is useful in the case where a parameter does not belongs to
an instrument instance that represents a real hardware instrument
but actually uses a real hardware instrument in its implementation
(e.g. via calls to one or more parameters of that real hardware
instrument). This is also useful when a parameter does belong to
an instrument instance but that instance does not represent the
real hardware instrument that the parameter interacts with: hence
``root_instrument`` of the parameter cannot be the
``hardware_instrument``, however ``underlying_instrument`` can be
implemented to return the ``hardware_instrument``.
By default it returns the ``root_instrument`` of the parameter.
"""
return self.root_instrument
@property
def abstract(self) -> bool | None:
return self._abstract
@property
def param_spec(self) -> ParamSpecBase:
if self._param_spec is None:
match self.vals:
case Arrays():
paramtype = "array"
case Strings():
paramtype = "text"
case ComplexNumbers():
paramtype = "complex"
case _:
paramtype = "numeric"
self._param_spec = ParamSpecBase(
name=self.register_name,
paramtype=paramtype,
label=None,
unit=None,
)
return self._param_spec
@property
def paramtype(self) -> str:
return self.param_spec.type
@paramtype.setter
def paramtype(self, paramtype: str) -> None:
self._set_paramtype(paramtype) # Indirected here, so subclasses can override
def _set_paramtype(self, paramtype: str) -> None:
paramtype = paramtype.lower()
if paramtype not in ["array", "text", "complex", "numeric"]:
raise ValueError(f"{paramtype} is not a valid paramtype")
if self.paramtype == paramtype:
return
new_vals: Validator
match paramtype:
case "array":
new_vals = Arrays()
case "text":
new_vals = Strings()
case "complex":
new_vals = ComplexNumbers()
case "numeric":
new_vals = Numbers()
case _:
raise NotImplementedError("This should not be possible")
if self.vals is None:
self.vals = new_vals
elif type(self.vals) is not type(new_vals):
logging.warning(
f"Tried to set a new paramtype {paramtype}, but this parameter already has paramtype {self.paramtype} which does not match"
)
self.param_spec.type = paramtype
@property
def depends_on(self) -> ParameterSet:
return self._depends_on
@property
def has_control_of(self) -> ParameterSet:
return self._has_control_of
@property
def is_controlled_by(self) -> ParameterSet:
# This is equivalent to the "inferred_from" relationship
return self._is_controlled_by
[docs]
def unpack_self(self, value: ValuesType) -> list[tuple[ParameterBase, ValuesType]]:
if isinstance(self.vals, Arrays):
if not isinstance(value, np.ndarray):
raise TypeError(
f"Expected data for Parameter with Array validator "
f"to be a numpy array but got: {type(value)}"
)
if self.vals.shape is not None and value.shape != self.vals.shape:
raise TypeError(
f"Expected data with shape {self.vals.shape}, "
f"but got {value.shape} for parameter: {self.full_name}"
)
return [(self, value)]
class GetLatest(DelegateAttributes):
"""
Wrapper for a class:`.Parameter` that just returns the last set or measured
value stored in the class:`.Parameter` itself. If get has never been called
on the parameter or the time since get was called is larger than
``max_val_age``, get will be called on the parameter. If the parameter
does not implement get, set should be called (or the initial_value set)
before calling get on this wrapper. It is an error to set
``max_val_age`` for a parameter that does not have a get function.
The functionality of this class is subsumed and improved in
parameter's cache that is accessible via ``.cache`` attribute of the
:class:`.Parameter`. Use of ``parameter.cache`` is recommended over use of
``parameter.get_latest``.
Examples:
>>> # Can be called:
>>> param.get_latest()
>>> # Or used as if it were a gettable-only parameter itself:
>>> Loop(...).each(param.get_latest)
Args:
parameter: Parameter to be wrapped.
"""
def __init__(self, parameter: ParameterBase):
self.parameter = parameter
delegate_attr_objects: ClassVar[list[str]] = ["parameter"]
omit_delegate_attrs: ClassVar[list[str]] = ["set"]
def get(self) -> ParamDataType:
"""
Return latest value if time since get was less than
`max_val_age`, otherwise perform `get()` and
return result. A `get()` will also be performed if the
parameter never has been captured.
It is recommended to use ``parameter.cache.get()`` instead.
"""
return self.parameter.cache.get()
def get_timestamp(self) -> datetime | None:
"""
Return the age of the latest parameter value.
It is recommended to use ``parameter.cache.timestamp`` instead.
"""
return self.cache.timestamp
def get_raw_value(self) -> ParamRawDataType | None:
"""
Return latest raw value of the parameter.
It is recommended to use ``parameter.cache.raw_value`` instead.
"""
return self.cache._raw_value
def __call__(self) -> ParamDataType:
"""
Same as ``get()``
It is recommended to use ``parameter.cache()`` instead.
"""
return self.cache()
P = TypeVar("P", bound=ParameterBase)
# Does not implement __hash__, not clear it needs to
[docs]
class ParameterSet(MutableSet, Generic[P]): # noqa: PLW1641
"""A set-like container that preserves the insertion order of its parameters.
This class implements the common set interface methods while maintaining
the order in which parameters were first added.
"""
def __init__(self, parameters: Sequence[P] | None = None) -> None:
self._dict: dict[P, None] = {}
if parameters is not None:
for item in parameters:
self.add(item)
[docs]
def add(self, value: P) -> None:
self._dict[value] = None
[docs]
def remove(self, value: P) -> None:
self._dict.pop(value)
[docs]
def discard(self, value: P) -> None:
if value in self._dict:
self._dict.pop(value)
[docs]
def clear(self) -> None:
self._dict.clear()
[docs]
def pop(self) -> ParameterBase:
if not self._dict:
raise KeyError("pop from an empty ParameterSet")
item = next(iter(self._dict))
self._dict.pop(item)
return item
[docs]
def union(self, other: ParameterSet[P]) -> ParameterSet[P]:
result = ParameterSet(list(self._dict.keys()))
for item in other:
result.add(item)
return result
[docs]
def intersection(self, other: ParameterSet[P]) -> ParameterSet[P]:
result: ParameterSet[P] = ParameterSet()
for item in self:
if item in other:
result.add(item)
return result
[docs]
def difference(self, other: ParameterSet[P]) -> ParameterSet[P]:
result: ParameterSet[P] = ParameterSet()
for item in self:
if item not in other:
result.add(item)
return result
[docs]
def issubset(self, other: ParameterSet[P] | set) -> bool:
return all(item in other for item in self)
[docs]
def issuperset(self, other: ParameterSet[P] | set) -> bool:
return all(item in self for item in other)
[docs]
def update(self, other: Iterable[P]) -> None:
for item in other:
self.add(item)
def __iter__(self) -> Iterator[P]:
return iter(self._dict)
def __contains__(self, item: object) -> bool:
return item in self._dict
def __len__(self) -> int:
return len(self._dict)
def __eq__(self, other: object) -> bool:
if isinstance(other, ParameterSet):
return set(self._dict) == set(other._dict)
return False
def __repr__(self) -> str:
if not self:
return f"{self.__class__.__name__}()"
return f"{self.__class__.__name__}({list(self._dict.keys())})"
def __or__(self, other: object) -> ParameterSet[P]:
if isinstance(other, ParameterSet):
return self.union(other)
raise NotImplementedError(
f"OR operation is not defined between ParameterSet and {type(other)}"
)
def __and__(self, other: object) -> ParameterSet[P]:
if isinstance(other, ParameterSet):
return self.intersection(other)
raise NotImplementedError(
f"AND operation is not defined between ParameterSet and {type(other)}"
)
def __sub__(self, other: object) -> ParameterSet[P]:
if isinstance(other, ParameterSet):
return self.difference(other)
raise NotImplementedError(
f"Difference operation is not defined between ParameterSet and {type(other)}"
)
def __le__(self, other: object) -> bool:
if isinstance(other, ParameterSet):
return self.issubset(other)
raise NotImplementedError(
f"<= operation is not defined between ParameterSet and {type(other)}"
)
def __ge__(self, other: object) -> bool:
if isinstance(other, ParameterSet):
return self.issuperset(other)
raise NotImplementedError(
f">+ operation is not defined between ParameterSet and {type(other)}"
)