from __future__ import annotations
import abc
import inspect
import os
import sys
import warnings
from collections.abc import Callable, Sequence, Collection
from pathlib import Path
from typing import Any, Literal, TypeVar
import numpy as np
import numpy.typing as npt
from qcodes.instrument import InstrumentBase, InstrumentChannel, InstrumentModule
from qcodes.parameters import DelegateParameter, ParamRawDataType, Parameter, ParameterBase
from qcodes.validators import Validator, validators as vals
try:
sys.path.append(str(Path(os.environ['TIMETAGGER_INSTALL_PATH'], 'driver', 'python')))
import TimeTagger as tt
except (KeyError, ImportError):
tt = None
_O = TypeVar('_O', bound=Callable[..., Any] | type[Any])
def _snake_to_camel(name: str) -> str:
humps = name.split('_')
return humps[0] + ''.join(hump.title() for hump in humps[1:])
[docs]
def refer_to_api_doc(api_obj: str = '') -> Callable[[_O], _O]:
"""Decorator factory to link a method to its TT API documentation."""
def decorator(obj: _O) -> _O:
if inspect.isclass(obj) and obj.__name__.endswith('Measurement'):
# A Measurement class
api_name = obj.__name__.removesuffix('Measurement')
typ = 'class'
elif inspect.isclass(obj) and obj.__name__.endswith('VirtualChannel'):
# A VirtualChannel class
api_name = obj.__name__.removesuffix('VirtualChannel')
typ = 'class'
elif inspect.isfunction(obj):
# A method, probably
api_name = '.'.join([api_obj, _snake_to_camel(obj.__name__)])
typ = 'meth'
else:
raise NotImplementedError(f'decorator not implemented for {obj}')
if obj.__doc__ is not None:
warnings.warn(f'Overwriting docstring of {obj}', RuntimeWarning, stacklevel=2)
obj.__doc__ = (f"Implements API object :{typ}:`TimeTagger:{api_name}` -- see the "
"documentation there.")
return obj
return decorator
[docs]
def cached_api_object(__func: Callable[..., Any] | None = None,
*, required_parameters: Collection[str] | None = None):
"""A custom descriptor for a cached API object with exception
handling, invalidation capability, and initialization checks."""
class CachedProperty:
def __init__(self, func: Callable[..., Any]):
self.func = func
self.required_parameters = [] if required_parameters is None else required_parameters
self.cache_name = f"__{func.__qualname__}_cached"
def __get__(self, instance, owner=None):
if instance is None:
return self
# Assert the required parameters have been initialized by checking
# that they pass the validator
not_initialized = set()
for param_name in self.required_parameters:
param: ParameterBase = getattr(instance, param_name)
try:
param.vals.validate(param.cache.get())
except AttributeError:
# No validator, cannot do anything
warnings.warn("All required parameters should have a validator.",
RuntimeWarning, stacklevel=2)
except Exception:
not_initialized.add(param_name)
if any(not_initialized):
raise RuntimeError('The following parameters need to be initialized first: '
+ ', '.join(not_initialized))
if hasattr(instance, self.cache_name):
value = getattr(instance, self.cache_name)
else:
value = self.func(instance)
setattr(instance, self.cache_name, value)
return value
def __set__(self, instance, value):
raise AttributeError('api property cannot be set directly.')
def __delete__(self, instance):
if hasattr(instance, self.cache_name):
delattr(instance, self.cache_name)
if __func is not None:
cached_property = CachedProperty(__func)
cached_property.__doc__ = __func.__doc__
return cached_property
else:
return CachedProperty
def _count_bins(start: float, stop: float, num: int) -> int:
bins = np.logspace(start + 12, stop + 12, int(num), dtype=np.int64)
return np.unique(bins).size
[docs]
class TypeValidator(vals.Validator[type]):
"""A validator for specific types."""
def __init__(self, cls: type):
self._valid_values = (cls,)
self.is_numeric = False
[docs]
def validate(self, value: type, context: str = "") -> None:
if not isinstance(value, self.valid_values):
raise TypeError(f'{value!r} is not of type {self.valid_values[0]}; {context}')
[docs]
class ArrayLikeValidator(vals.Arrays):
"""A validator for array_like objects.
Note that validation might be expensive since conversion to ndarray
takes place each time.
"""
def __repr__(self) -> str:
return super().__repr__().replace('Arrays', 'ArrayLike')
[docs]
def validate(self, value: npt.ArrayLike, context: str = "") -> None:
try:
array = np.array(value)
except ValueError as err:
raise ValueError(f'{value!r} is invalid: cannot convert to array; {context}') from err
super().validate(array, context)
[docs]
class DelegateParameterWithoutParentValidator(DelegateParameter):
"""A :class:`DelegateParameter` with a validator that does not
validate on the parent's validators."""
# Reverts behavior introduced in GH #6865 / 0000926
@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)
[docs]
class LogspaceStartValidator(vals.Validator[float]):
"""Validates the exp_start parameter of a :class:`HistogramLogBinsMeasurement`."""
def __init__(self, stop_param: ParameterBase, num_param: ParameterBase):
self.stop_param = stop_param
self.num_param = num_param
[docs]
def validate(self, value: float, context: str = "") -> None:
if (stop := self.stop_param.cache.get(False)) is None:
return
if (num := self.num_param.cache.get(False)) is None:
return
if _count_bins(value, stop, num) != int(num):
raise ValueError(f'{value!r} is invalid: results in redundant bin edges; {context}')
[docs]
class LogspaceStopValidator(vals.Validator[float]):
"""Validates the exp_stop parameter of a :class:`HistogramLogBinsMeasurement`."""
def __init__(self, start_param: ParameterBase, num_param: ParameterBase):
self.start_param = start_param
self.num_param = num_param
[docs]
def validate(self, value: float, context: str = "") -> None:
if (start := self.start_param.cache.get(False)) is None:
return
if (num := self.num_param.cache.get(False)) is None:
return
if _count_bins(start, value, num) != int(num):
raise ValueError(f'{value!r} is invalid: results in redundant bin edges; {context}')
[docs]
class LogspaceNumValidator(vals.Validator[int]):
"""Validates the n_bins parameter of a :class:`HistogramLogBinsMeasurement`."""
def __init__(self, start_param: ParameterBase, stop_param: ParameterBase):
self.start_param = start_param
self.stop_param = stop_param
[docs]
def validate(self, value: int, context: str = "") -> None:
if (start := self.start_param.cache.get(False)) is None:
return
if (stop := self.stop_param.cache.get(False)) is None:
return
if _count_bins(start, stop, value) != int(value):
raise ValueError(f'{value!r} is invalid: results in redundant bin edges; {context}')
[docs]
class ParameterWithSetSideEffect(Parameter):
"""A :class:`Parameter` allowing for side effects on set events.
Parameters
----------
set_side_effect :
A callable that is run on every set event. Receives the set
value as sole argument.
execute_before :
Run the side effect before or after setting the parameter.
"""
def __init__(self, name: str, set_side_effect: Callable[[Any], None],
set_cmd: Callable[[ParamRawDataType], Any] | None = None,
execute_before: bool = False, **kwargs: Any) -> None:
if set_cmd is False:
raise ValueError('ParameterWithSetSideEffect needs to be settable')
# Parameter does not allow overriding set_raw method
def set_raw(value: ParamRawDataType) -> ParamRawDataType:
if execute_before:
set_side_effect(value)
if set_cmd is not None:
set_cmd(value)
if not execute_before:
set_side_effect(value)
return value
super().__init__(name, set_cmd=set_raw, **kwargs)
[docs]
class MeasurementControlMixin(metaclass=abc.ABCMeta):
"""Mixin class forwarding common methods from the TimeTagger API."""
@property
@abc.abstractmethod
def api(self):
pass
[docs]
@refer_to_api_doc('IteratorBase')
def clear(self):
return self.api.clear()
[docs]
@refer_to_api_doc('IteratorBase')
def start(self):
return self.api.start()
[docs]
@refer_to_api_doc('IteratorBase')
def start_for(self, duration: int, clear: bool = True):
return self.api.startFor(duration, clear)
[docs]
@refer_to_api_doc('IteratorBase')
def stop(self):
return self.api.stop()
[docs]
@refer_to_api_doc('IteratorBase')
def is_running(self) -> bool:
return self.api.isRunning()
[docs]
@refer_to_api_doc('IteratorBase')
def wait_until_finished(self, timeout: int = -1):
return self.api.waitUntilFinished(timeout)