mjolnir.measurements#
This module defines measurement routines.
Measurements are defined by sweeps.AbstractSweep and
measures.Measure, and are managed and executed using
handler.MeasurementHandler.
Examples
15 Minutes to Mjölnir#
This example reproduces the sweep examples from 15 minutes to QCoDeS.
We first import and set up the dummy instruments:
>>> import tempfile
>>> from pathlib import Path
>>> import numpy as np
>>> from qcodes import (initialise_or_create_database_at,
... load_or_create_experiment,
... Station, Instrument)
>>> from qcodes.dataset import plot_dataset
>>> from qcodes.instrument_drivers.mock_instruments import (
... DummyInstrument, DummyInstrumentWithMeasurement)
>>> from mjolnir.measurements import sweeps, measures
>>> from mjolnir.measurements.handler import DefaultMeasurementHandler
>>> initialise_or_create_database_at(Path(tempfile.gettempdir(), 'example.db'))
>>> exp = load_or_create_experiment('dummy_experiment', 'dummy_smaple')
>>> dac = DummyInstrument('dac', gates=['ch1', 'ch2'])
>>> dmm = DummyInstrumentWithMeasurement('dmm', setter_instr=dac)
>>> station = Station()
>>> station.add_component(dac)
'dac'
>>> station.add_component(dmm)
'dmm'
>>> station.add_component(Instrument('dummy_smaple'), update_snapshot=False)
'dummy_smaple'
The mjolnir.measurements.sweeps module provides three different classes
to define sweeps. The probably most common one is GridSweep,
which generates setpoints gridded in a linear, geometric, or logarithmic
sequence. To reproduce the first example from the QCoDeS doc, it is used as
follows:
>>> sweep = sweeps.GridSweep(dac.ch1, rng=(0, 25), num_points=10,
... spacing='lin', delay=0.01)
Next, the mjolnir.measurements.measures module defines the
Measure class, which is a compatiblity wrapper around a
gettable parameter:
>>> measure = measures.Measure(dmm.v1)
Finally, both are passed on to the measurement handler defined (or possibly
subclassed for customization) in mjolnir.measurements.handler:
>>> handler = DefaultMeasurementHandler(station, 'dummy_smaple')
>>> ds = handler.measure(sweep, measure)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> plot_dataset(ds)
...
A sweep with logarithmically spaced negative samples:
>>> sweep = sweeps.GridSweep(dac.ch1, rng=(-1e+1, -1e-1), num_points=11,
... spacing='geom', delay=0.1)
>>> ds = handler.measure(sweep, measure)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
Next, there is the CenteredLinearSweep class, which, as the
name suggests, performs a linear sweep first in one direction away from a
center point and then in the other:
>>> sweep = sweeps.CenteredLinearSweep(dac.ch1, rng=25, num_points=21,
... delay=0.01)
>>> ds = handler.measure(sweep, measure)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> plot_dataset(ds)
...
Finally, the most general type of sweep is given by ArraySweep,
which just has any number of setpoints. Additionally, multiple arrays can be
passed, in which case they are interpreted as ‘directions’ and swept over
consecutively, similar to CenteredLinearSweep. We can use it
to reproduce both examples above:
>>> sweep = sweeps.ArraySweep(dac.ch1, np.linspace(0, 25, 10), delay=0.01)
>>> ds = handler.measure(sweep, measure)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> plot_dataset(ds)
...
>>> arrays = (np.linspace(0, 25, 11), np.linspace(0, -25, 11)[1:])
>>> sweep = sweeps.ArraySweep(dac.ch1, arrays, delay=0.01)
>>> ds = handler.measure(sweep, measure)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> plot_dataset(ds)
...
Sweeps implement three binary operators: or, and, and matmul. The first nests sweeps (i.e., generates nd-loops), the second parallelizes them, and the last sequences them in time. To reproduce the doNd example from the QCoDeS doc, we use the following:
>>> sweep_list = (sweeps.GridSweep(dac.ch1, (-1, 1), 20, delay=0.01)
... | sweeps.GridSweep(dac.ch2, (-1, 1), 20, delay=0.01))
>>> measure_set = measures.Measure(dmm.v1) | measures.Measure(dmm.v2)
>>> ds = handler.measure(sweep_list, measure_set)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> plot_dataset(ds)
...
Note that live plotting can also be disabled per-measure (it has an overhead of tens of milliseconds):
>>> measure_set = measures.Measure(dmm.v1, live_plot=False) | dmm.v2
>>> ds = handler.measure(sweep_list, measure_set)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
We can also compose sweeps differently. This following example scans along the diagonal of the parameter space of the previous example.
>>> sweep = (sweeps.GridSweep(dac.ch1, (-1, 1), 20, delay=0.01)
... & sweeps.GridSweep(dac.ch2, (-1, 1), 20, delay=0.01))
>>> measure_set = measures.Measure(dmm.v1) | measures.Measure(dmm.v2)
>>> ds = handler.measure(sweep, measure_set)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> plot_dataset(ds)
...
The same can also be achieved by instantiating a GridSweep with multiple parameters. Here, keyword arguments are sequences of the same length as the parameters, or scalar, in which case they are broadcast (if it makes sense).
>>> sweep = sweeps.GridSweep([dac.ch1, dac.ch2], rng=[(-1, 1), (-1, 1)],
... num_points=20, delay=0.01)
>>> ds = handler.measure(sweep, measure_set)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> plot_dataset(ds)
...
The last composition operation is given by concatenation. This just
appends one sweep to the other, but only works for sweeps over the
same parameters. This allows for instance using
GridSweep s to construct a sweep with different
resolution in different regions:
>>> sweep = (sweeps.GridSweep(dac.ch1, (-1, 0), 20, delay=0.01)
... @ sweeps.GridSweep(dac.ch1, (0, 1), 30, delay=0.01))
>>> ds = handler.measure(sweep, measure_set)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> plot_dataset(ds)
...
However, it also works with different types of sweeps:
>>> sweep = (sweeps.CenteredLinearSweep(dac.ch1, 0.5, 20, delay=0.01)
... @ sweeps.GridSweep(dac.ch1, (3, 4), 30, delay=0.01))
>>> ds = handler.measure(sweep, measure_set)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> plot_dataset(ds)
...
To perform custom initialization tasks at the begin of a sweep, use the
init_actions argument. This is a (sequence of) callable(s) that is
run before the first setpoint is ramped to.
>>> sweep = (sweeps.ArraySweep(dac.ch1, [1, 2, 3],
... init_actions=lambda: print('INIT 1'))
... | sweeps.GridSweep(dac.ch2, (0, 2), 11,
... init_actions=lambda: print('INIT 2')))
>>> ds = handler.measure(sweep, measure, show_progress=False)
Starting experimental run with id: ...
INIT 1
INIT 2
INIT 2
INIT 2
MeasurementHandler.measure took ...
Setpoints given in sweeps can furthermore be transformed at runtime
using the transform argument. This can be used to transform the
coordinate of the inner loop conditioned on the state of an outer
loop, for example.
>>> sweep = (
... sweeps.GridSweep(dac.ch1, (-1, 1), 11)
... | sweeps.GridSweep(
... dac.ch2, (-1, 1), 21, transform=lambda x: x+dac.ch1()
... )
... )
>>> ds = handler.measure(sweep, measure_set)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> plot_dataset(ds)
...
As an explicit example, we could implement a hysteresis loop like this:
>>> from qcodes.parameters import ManualParameter
>>> direction = ManualParameter('direction', initial_value=1)
>>> sweep = (
... sweeps.ArraySweep(direction, [-1, +1])
... | sweeps.GridSweep(
... dac.ch1, (-1, 1), 11, transform=lambda x: x*direction(),
... delay=0.25
... )
... )
>>> ds = handler.measure(sweep, measure)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> # plot_dataset is a bit stupid here ...
>>> (da := ds.to_xarray_dataset().dmm_v1).plot()
...
Note that sweep setpoints are validated on construction:
>>> sweep = sweeps.ArraySweep(dac.ch1, ['asdf', 1.])
Traceback (most recent call last):
...
TypeError: np.str_('asdf') is not an int or float; Parameter: dac.ch1
Threaded Data Acquisition#
Using ThreadPoolParamsCaller, data can also be
acquired asynchronously by setting the threaded_acquisition flag in
measure(). For more details on the gotchas,
see the QCoDeS example notebook on threaded acquisition.
To demonstrate, we define parameters that take some time to acquire. Since data from a single instrument can only be gotten sequentially, we create a second one and make sure the delegates are registered with the instruments:
>>> import time
>>> from qcodes.parameters import DelegateParameter
>>> def slow_down(x):
... time.sleep(0.1)
... return x
>>> dmm2 = DummyInstrumentWithMeasurement('dmm2', setter_instr=dac)
>>> v1_slow = DelegateParameter('v1_slow', dmm.v1, get_parser=slow_down,
... instrument=dmm)
>>> v2_slow = DelegateParameter('v2_slow', dmm2.v2, get_parser=slow_down,
... instrument=dmm2)
Then we measure as usual, but set the flag for threaded acquisition.
>>> sweep = sweeps.GridSweep(dac.ch1, rng=(0, 25), num_points=26)
>>> measure_set = measures.Measure(v1_slow) | v2_slow
>>> ds = handler.measure(sweep, measure_set, live_plot=False,
... threaded_acquisition=True)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
For comparison, the same measurement using sequential acquisition:
>>> ds = handler.measure(sweep, measure_set, live_plot=False,
... threaded_acquisition=False)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
MultiParameters#
MultiParameters can be used for
heterogeneous data. These are also supported by Measure.
Using the QCoDeS example for MultiParameter, we measure I and Q like so:
>>> from qcodes.parameters import ManualParameter, MultiParameter
>>> class SingleIQPair(MultiParameter):
... def __init__(self, scale_param):
... # only name, names, and shapes are required
... # this version returns two scalars (shape = `()`)
... super().__init__(
... "single_iq",
... names=("I", "Q"),
... shapes=((), ()),
... labels=("In phase amplitude", "Quadrature amplitude"),
... units=("V", "V"),
... # including these setpoints is unnecessary here, but
... # if you have a parameter that returns a scalar alongside
... # an array you can represent the scalar as an empty sequence.
... setpoints=((), ()),
... docstring="param that returns two single values, I and Q",
... )
... self._scale_param = scale_param
... def get_raw(self):
... scale_val = self._scale_param()
... return scale_val, scale_val / 2
>>> scale = ManualParameter("scale", initial_value=2)
>>> iq = SingleIQPair(scale_param=scale)
>>> measure = measures.Measure(iq)
>>> ds = handler.measure(sweep, measure)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
This also works with array parameters:
>>> class IQArray(MultiParameter):
... def __init__(self, scale_param):
... # names, labels, and units are the same
... super().__init__(
... "iq_array",
... names=("I", "Q"),
... shapes=((5,), (5,)),
... labels=("In phase amplitude", "Quadrature amplitude"),
... units=("V", "V"),
... # note that EACH item needs a sequence of setpoint arrays
... # so a 1D item has its setpoints wrapped in a length-1 tuple
... setpoints=(((0, 1, 2, 3, 4),), ((0, 1, 2, 3, 4),)),
... docstring="param that returns two single values, I and Q",
... )
... self._scale_param = scale_param
... self._indices = np.array([0, 1, 2, 3, 4])
... def get_raw(self):
... scale_val = self._scale_param() * np.random.randn()
... return (self._indices * scale_val, self._indices * scale_val / 2)
>>> iq_array = IQArray(scale_param=scale)
>>> measure = measures.Measure(iq_array)
>>> ds = handler.measure(sweep, measure)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
Buffered / Batched parameters#
Let us now take a look at an example with a
ParameterWithSetpoints.
Here, we follow the QCoDeS example Simple Example of ParameterWithSetpoints
in a slimmed down version.
First, some setup code:
>>> from qcodes.instrument import Instrument
>>> from qcodes.parameters import Parameter, ParameterWithSetpoints
>>> from qcodes.validators import Arrays
>>> class DummySetpoints(Parameter):
... def __init__(self, start, stop, n_points, *args, **kwargs):
... super().__init__(*args, **kwargs)
... self._start = start
... self._stop = stop
... self._n_points = n_points
... def get_raw(self):
... return np.linspace(self._start(), self._stop(), self._n_points())
>>> class DummyArray(ParameterWithSetpoints):
... rng = np.random.default_rng()
... def get_raw(self):
... return self.rng.random(self.root_instrument.n_points())
>>> a = Instrument('foobar')
>>> a.f_start = Parameter('f_start', initial_value=0, set_cmd=None,
... instrument=a)
>>> a.f_stop = Parameter('f_stop', initial_value=500, set_cmd=None,
... instrument=a)
>>> a.n_points = Parameter('n_points', initial_value=501, set_cmd=None,
... get_parser=int, instrument=a)
>>> a.freq_axis = DummySetpoints(name='freq_axis', start=a.f_start,
... stop=a.f_stop, n_points=a.n_points,
... instrument=a, unit='Hz',
... vals=Arrays(shape=(a.n_points.get_latest,)))
>>> a.spectrum = DummyArray('spectrum', setpoints=(a.freq_axis,),
... instrument=a,
... vals=Arrays(shape=(a.n_points.get_latest,)))
>>> b = Parameter('external_param', set_cmd=None)
Finally, we can run a measurement:
>>> sweep = sweeps.GridSweep(b, (0, 10), 11, delay=0.1)
>>> measure = measures.Measure(a.spectrum)
>>> ds = handler.measure(sweep, measure)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> plot_dataset(ds)
...
Measure take an optional keyword argument that allows
saving processed data alongside the raw data. This could in our example
here for instance be used to save the integrated intensity of the
spectrum. To do this, we must define a
DelegateParameter that derives from
our spectrum parameter:
>>> from qcodes.parameters import DelegateParameter, ParamDataType
>>> from scipy.integrate import trapezoid
>>> def integrate_spectrum(x):
... return trapezoid(x, a.freq_axis.get_latest())
>>> class MyDelegateParameter(DelegateParameter):
... # Qcodes validates delegates of vector-valued parameters with
... # the parent validators. See GH437, GH6865
... def validate(self, value): ...
... @property
... def vals(self): ...
>>> integrated = MyDelegateParameter('integrated', a.spectrum,
... get_parser=integrate_spectrum,
... label='integrated spectrum')
Now we just pass our new parameter as the delegates argument to the
Measure constructor:
>>> measure = measures.Measure(a.spectrum, delegates=[integrated])
>>> ds = handler.measure(sweep, measure)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> plot_dataset(ds)
...
If a ParameterWithSetpoints has fewer
than eleven setpoints, it will be plotted as multiple lines in the same
plot:
>>> a.n_points(5)
>>> ds = handler.measure(sweep, measure)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
We can also do more complex sweeps and use
mjolnir.plotting.plot_nd() to display the results:
>>> sweep = sweeps.GridSweep([a.f_start, a.f_stop], [(0, 100), (500, 600)],
... 11, delay=0.1)
>>> ds = handler.measure(sweep, measure)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> from mjolnir.plotting import plot_nd
>>> plot_nd(ds, array_target='spectrum', horizontal_target='freq_axis',
... plot_slice=False)
...
… Or a 2d-loop producing 3d data:
>>> sweep_list = (sweeps.GridSweep(b, (0, 10), 11)
... | (sweeps.GridSweep(a.f_start, (0, 100), 11, delay=0.01))
... & sweeps.GridSweep(a.f_stop, (500, 600), 11, delay=0.01))
>>> ds = handler.measure(sweep_list, measure)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> plot_nd(ds, array_target='spectrum', horizontal_target='freq_axis',
... vertical_target='f_start', plot_slice=False)
...
Switching around the displayed dimensions:
>>> plot_nd(ds, array_target='spectrum', horizontal_target='f_start',
... vertical_target='f_stop', plot_slice=False)
...
It is not yet implemented to plot parallel-swept parameters on the same axis.
Multidimensional array parameters#
Parameters that return >1d arrays are also supported. Let us set up a spectrogram instrument:
>>> class DummySpectrogram(ParameterWithSetpoints):
... rng = np.random.default_rng()
... def get_raw(self):
... return self.rng.random((self.root_instrument.n_dt(),
... self.root_instrument.n_points()))
>>> a.dt = Parameter('dt', initial_value=0.1, set_cmd=None, instrument=a)
>>> a.n_dt = Parameter('n_dt', initial_value=5, set_cmd=None,
... get_parser=int, instrument=a)
>>> a.time_axis = DummySetpoints(name='time_axis', start=lambda: 0,
... stop=lambda: a.dt() * a.n_dt(),
... n_points=a.n_dt, instrument=a, unit='s',
... vals=Arrays(shape=(a.n_dt.get_latest,)))
>>> a.spectrogram = DummySpectrogram(
... 'spectrogram',
... setpoints=(a.time_axis, a.freq_axis,), instrument=a,
... vals=Arrays(shape=(a.n_dt.get_latest, a.n_points.get_latest,)))
Running a sweep on the external parameter will result in the following measurement:
>>> a.n_points(501)
>>> sweep = sweeps.GridSweep(b, (0, 10), 11, delay=0.1)
>>> measure = measures.Measure(a.spectrogram)
>>> ds = handler.measure(sweep, measure)
Starting experimental run with id: ...
MeasurementHandler.measure took ...
>>> plot_nd(ds, array_target='spectrogram', horizontal_target='freq_axis',
... vertical_target='time_axis')
...
Modules