Skip to content

Sweeping

Basic Example

A Sweep is created out of two main components, an iterable pointer and a variable number of actions. Both pointer and actions may generate records.

The most bare example would look like this:

>>> for data in Sweep(range(3)):
>>>     print(data)
{}
{}
{}

In this example, the range(3) iterable object is our pointer. This Sweep does not contain any actions or generate any records, but instead simply loops over the iterable pointer.

Recording Data

Concepts

Even though the pointer in the previous example does generate data, we cannot see it when we iterate through the sweep. To have pointers and actions generate visible data, we need to annotate them so that they generate records.

For a Sweep to know that either a pointer (an iterable object), or an action (a callable object) they need to be wrapped by an instance of DataSpec. DataSpec is a data class that holds information about the variable itself. Understanding the inner workings are not necessary to fully utilize Sweeps, however, it is good to know they exists and what information they hold.

Annotating a function or generator does not change what they do, it gets the values generated by them and places them inside a dictionary with the key for those values being the name of the variable we assign.

The important fields of a DataSpec are its name and depends_on fields. name, simply indicates the name of the variable, e.i. the key that the sweep will have for the value of this variable. depends_on indicates whether the variable is an independent variable (we control) or a dependent variable (the things we are trying to measure). If depends_on=None it means this variable is an independent variable. If depends_on=['x'], this variable is dependent on a separate variable with name x. If depends_on=[], the variable will be automatically assigned as a dependent of all other independents in the same Sweep.

Note

DataSpec, also contains two more fields: unit and types, these however, have no impact in the way the code behaves and are for adding extra metadata for the user.

While this might seem like a lot of information, its use is very intuitive and easy to use once you get used to it.

Implementation

To wrap functions we use the recording decorator on the function we want to annotate:

>>> @recording(DataSpec('x'), DataSpec('y', depends_on=['x'], type='array'))
>>> def measure_stuff(n, *args, **kwargs):
>>>     return n, np.random.normal(size=n)
>>>
>>> measure_stuff(1)
{'x': 1, 'y': array([0.70663348])}

In the example above we annotate the function measure_stuff() indicating that the first item it returns is x, an independent variable since it does not have a depends_on field, and the second item is y, a variable that depends on x.

We can annotate generators in the same way:

>>> @recording(DataSpec('a'))
>>> def make_sequence(n):
>>>     for i in range(n):
>>>         yield i
>>>
>>> for data in make_sequence(3):
>>>     print(data)
{'a': 0}
{'a': 1}
{'a': 2}

A nicer way of creating DataSpec instances is to use the functions independent and dependent. This function just makes the recording of data easier to read. independent does not let you indicate the depends_on field while dependent, has an empty list (indicating that it depends an all other independents) as a default.

>>> @recording(independent('x'), dependent('y', type='array'))
>>> def measure_stuff(n, *args, **kwargs):
>>>    return n, np.random.normal(size=n)
>>>
>>> measure_stuff(1)
{'x': 1, 'y': array([1.60113794])}

Note

You can also use the shorter versions of these functions:

Sometimes we don't want to annotate a function or generator itself, but instead we want to annotate at the moment of execution. For this we can use the function record_as() to annotate any function or generator on the fly:

>>> def get_some_data(n):
>>>     return np.random.normal(size=n)
>>>
>>> record_as(get_some_data, independent('random_var'))(3)
{'random_var': array([0.16099358, 0.74873271, 0.01160423])}
You can add multiple DataSpec with in a single record_as():

>>> for data in record_as(zip(np.linspace(0,1,3), np.arange(3)), indep('x'), dep('y')):
>>>     print(data)
{'x': 0.0, 'y': 0}
{'x': 0.2, 'y': 1}
{'x': 0.4, 'y': 2}
It will also make sure to add items for annotated records (by adding None items to any empty record) that do not have any values assigned to them:

>>> for data in record_as(np.linspace(0,1,3), indep('x'), dep('y')):
>>>     print(data)
{'x': 0.0, 'y': None}
{'x': 0.5, 'y': None}
{'x': 1.0, 'y': None}

And it will ignore any extra values that are not annotated:

>>> for data in record_as(zip(np.linspace(0,1,3), np.arange(3)), indep('x')):
>>>     print(data)
{'x': 0.0}
{'x': 0.5}
{'x': 1.0}

Construction of Sweeps

Now that we know how to annotate data so that it generates records, we can finally start creating Sweeps that creates some data. A Sweep is composed of two main parts: pointers and actions:

  • Pointers are iterables that the Sweep iterates through, these usually represent the independent variables of our experiments.
  • Actions are callables that get called after each iteration of our pointer and usually are in charge of performing anything that needs to happen at every iteration of the experiment. This can be either set up a instruments and usually includes measuring a dependent variable too.

Both pointers and actions can generate records if annotated correctly, but it is not a requirement.

Basic Sweeps

A basic annotated Sweep looks something like this:

def my_func():
    return 0

sweep = Sweep(
    record_as(range(3), independent('x')), # This is the pointer. We specify 'x' as an independent (we control it).
    record_as(my_func, dependent('y'))) # my_func is an action. We specify 'y' as a dependent.

Once the Sweep is created we can see the records it will produce by using the function method get_data_specs():

>>> sweep.get_data_specs()
(x, y(x))

Printing a Sweep will also display more information about, specifying the pointers, the actions taken afterwards and the records it will produce:

>>> print(sweep)
range(0, 3) as {x} >> my_func() as {y}
==> {x, y(x)}

Now to run the Sweep we just have to iterate through it:

>>> for data in sweep:
>>>     print(data)
{'x': 0, 'y': 0}
{'x': 1, 'y': 0}
{'x': 2, 'y': 0}

If you are trying to sweep over a single parameter, a more convenient syntax for creating Sweep is to utilize the sweep_parameter() function:

>>> sweep = sweep_parameter('x', range(3), record_as(my_func, 'y'))
>>> for data in sweep:
>>>     print(data)
{'x': 0, 'y': 0}
{'x': 1, 'y': 0}
{'x': 2, 'y': 0}

There is no restriction on how many parameters a pointer or an action can generate as long as each parameter is properly annotated.

>>> def my_func():
>>>     return 1, 2
>>>
>>> sweep = Sweep(
>>>     record_as(zip(range(3), ['a', 'b', 'c']), independent('number'), independent('string')), # a pointer with two parameters
>>>     record_as(my_func, 'one', 'two')) # an action with two parameters
>>>
>>> print(sweep.get_data_specs())
>>>
>>> for data in sweep:
>>>     print(data)
(number, string, one(number, string), two(number, string))
{'number': 0, 'string': 'a', 'one': 1, 'two': 2}
{'number': 1, 'string': 'b', 'one': 1, 'two': 2}
{'number': 2, 'string': 'c', 'one': 1, 'two': 2}

Specifying Options Before Executing a Sweep

Many actions we are using take optional parameters that we only want to specify just before executing the Sweep (but are constant throughout the Sweep).

If we don't want to resort to global variables we can do so by using the method set_options(). It accepts the names of any action function in that Sweep as keywords, and dictionaries containing keyword arguments to pass to those functions as values. Keywords specified in this way always override keywords that are passed around internally in the sweep (for more information see the Passing Parameters in a Sweep` section):

>>> def test_fun(a_property=False, *args, **kwargs):
>>>     print('inside test_fun:')
>>>     print(f"a_property: {a_property}")
>>>     print(f"other stuff:", args, kwargs)
>>>     print('----')
>>>     return 0
>>>
>>> sweep = sweep_parameter('value', range(3), record_as(test_fun, dependent('data')))
>>> sweep.set_options(test_fun=dict(a_property=True, another_property='Hello'))
>>>
>>> for data in sweep:
>>>     print("Data:", data)
>>>     print('----')
inside test_fun:
property: True
other stuff: () {'value': 0, 'another_property': 'Hello'}
----
Data: {'value': 0, 'data': 0}
----
inside test_fun:
property: True
other stuff: () {'value': 1, 'data': 0, 'another_property': 'Hello'}
----
Data: {'value': 1, 'data': 0}
----
inside test_fun:
property: True
other stuff: () {'value': 2, 'data': 0, 'another_property': 'Hello'}
----
Data: {'value': 2, 'data': 0}

A QCoDeS Parameter Sweep

If you are using QCoDeS to interact with hardware, it is very common to want to do a sweep over a QCoDeS parameter. In this minimal example we set a parameter (x) to a range of values, and get data from another parameter for each set value.

>>> def measure_stuff():
>>>     return np.random.normal()
>>>
>>> x = Parameter('x', set_cmd=lambda x: print(f'setting x to {x}'), initial_value=0) # QCoDeS Parameter
>>> data = Parameter('data', get_cmd=lambda: np.random.normal()) # QCoDeS Parameter
>>>
>>> for record in sweep_parameter(x, range(3), get_parameter(data)):
>>>     print(record)
setting x to 0
setting x to 0
{'x': 0, 'data': -0.4990053668503893}
setting x to 1
{'x': 1, 'data': -0.5132204673887943}
setting x to 2
{'x': 2, 'data': 1.8634243556469932}

Sweep Combinations

One of the most valuable features of Sweeps is their ability to be combined through the use of operators. This allows us to mix and match different aspects of an experiment without having to rewrite code. We can combine different Sweeps with each other or different annotated actions

Appending

The most basic combination of Sweeps is appending them. When appending two Sweeps, the resulting sweep will execute the first Sweep to completion followed by the second Sweep to completion. To append two Sweeps or actions we use the + symbol:

>>> def get_random_number():
>>>     return np.random.rand()
>>>
>>> Sweep.record_none = False # See note on what this does.
>>>
>>> sweep_1 = sweep_parameter('x', range(3), record_as(get_random_number, dependent('y')))
>>> sweep_2 = sweep_parameter('a', range(4), record_as(get_random_number, dependent('b')))
>>> my_sweep = sweep_1 + sweep_2
>>>
>>> for data in my_sweep:
>>>     print(data)
{'x': 0, 'y': 0.34404570192577155}
{'x': 1, 'y': 0.02104831292457654}
{'x': 2, 'y': 0.9006367857458307}
{'a': 0, 'b': 0.10539935409724577}
{'a': 1, 'b': 0.9368463758729733}
{'a': 2, 'b': 0.9550070757291859}
{'a': 3, 'b': 0.9812445448108895}

Note

Sweep.return_none controls whether we include data fields that have returned nothing during setting a pointer or executing an action. Setting it to true (the default) guarantees that each data spec of the sweep has an entry per sweep point, even if it is None. For more information see: Passing Parameters in a Sweep section.

Multiplying

By multiplying we refer to an inner product, i.e. the result is what you'd expect from zip two iterables. To multiply two Sweeps or actions we use the * symbol. A basic example is if we have a sweep and want to attach another action to each sweep point:

>>> my_sweep = (
>>>     sweep_parameter('x', range(3), record_as(get_random_number, dependent('data_1')))
>>>     * record_as(get_random_number, dependent('data_2'))
>>> )
>>>
>>> print(sweep.get_data_specs())
>>> print('----')
>>>
>>> for data in my_sweep:
>>>     print(data)
(x, data_1(x), data_2(x))
----
{'x': 0, 'data_1': 0.12599818360565485, 'data_2': 0.09261266841087679}
{'x': 1, 'data_1': 0.5665798938860637, 'data_2': 0.7493750740615404}
{'x': 2, 'data_1': 0.9035085438172156, 'data_2': 0.5419023528195611}

If you are combining two different Sweeps, then we get zip-like behavior while maintain the dependency structure separate:

>>> my_sweep = (
>>>     sweep_parameter('x', range(3), record_as(get_random_number, dependent('data_1')))
>>>     * sweep_parameter('y', range(5), record_as(get_random_number, dependent('data_2')))
>>> )
>>>
>>> print(sweep.get_data_specs())
>>> print('----')
>>>
>>> for data in my_sweep:
>>>     print(data)
(x, data_1(x), y, data_2(y))
----
{'x': 0, 'data_1': 0.3808452915069015, 'y': 0, 'data_2': 0.14309246334791337}
{'x': 1, 'data_1': 0.6094608905204076, 'y': 1, 'data_2': 0.3560530722571186}
{'x': 2, 'data_1': 0.15950240245080072, 'y': 2, 'data_2': 0.2477391943438858}

Nesting

Nesting two Sweeps runs the entire second Sweep for each point of the first Sweep. A basic example is if we have multiple Sweep parameters against each other and we want to perform a measurement at each point. To nest two Sweeps we use the @ symbol:

>>> def measure_something():
>>>     return np.random.rand()
>>>
>>> my_sweep = (
>>>     sweep_parameter('x', range(3))
>>>     @ sweep_parameter('y', np.linspace(0,1,3))
>>>     @ record_as(measure_something, 'my_data')
>>> )
>>>
>>> for data in my_sweep:
>>>     print(data)
{'x': 0, 'y': 0.0, 'my_data': 0.727404046865409}
{'x': 0, 'y': 0.5, 'my_data': 0.11112429412122715}
{'x': 0, 'y': 1.0, 'my_data': 0.09081900115421426}
{'x': 1, 'y': 0.0, 'my_data': 0.8160224024098803}
{'x': 1, 'y': 0.5, 'my_data': 0.1517092154216605}
{'x': 1, 'y': 1.0, 'my_data': 0.9253018251769569}
{'x': 2, 'y': 0.0, 'my_data': 0.881089486629102}
{'x': 2, 'y': 0.5, 'my_data': 0.3897577898200387}
{'x': 2, 'y': 1.0, 'my_data': 0.6895312744116066}

Nested sweeps can be as complex as needed, with as many actions as they need. An example of this can be executing measurements on each nested level:

>>> def measure_something():
>>>     return np.random.rand()
>>>
>>> sweep_1 = sweep_parameter('x', range(3), record_as(measure_something, 'a'))
>>> sweep_2 = sweep_parameter('y', range(2), record_as(measure_something, 'b'))
>>> my_sweep = sweep_1 @ sweep_2 @ record_as(get_random_number, 'more_data')
>>>
>>> for data in my_sweep:
>>>     print(data)
{'x': 0, 'a': 0.09522178419462424, 'y': 0, 'b': 0.1821505218348034, 'more_data': 0.13257002268089835}
{'x': 0, 'a': 0.09522178419462424, 'y': 1, 'b': 0.014940266372080457, 'more_data': 0.9460879863404558}
{'x': 1, 'a': 0.13994892182170526, 'y': 0, 'b': 0.4708657480125388, 'more_data': 0.12792337523097086}
{'x': 1, 'a': 0.13994892182170526, 'y': 1, 'b': 0.8209492135277935, 'more_data': 0.23270477191895111}
{'x': 2, 'a': 0.06159208933324678, 'y': 0, 'b': 0.651545802505077, 'more_data': 0.8944257582518365}
{'x': 2, 'a': 0.06159208933324678, 'y': 1, 'b': 0.9064557565446919, 'more_data': 0.8258102740474211}

Note

All operators symbols are just there for syntactic brevity. All three of them have corresponding functions attached to them:

Passing Parameters in a Sweep

Often times our measurement actions depend on the states of previous steps. Because of that, everything that is generated by pointers, actions or other Sweeps can be passed on subsequently executed elements.

Note

here are two different Sweep configuration related to passing arguments in Sweeps. For more information on them see the Configuring Sweeps.

Positional Arguments

When there are no record annotations, the values generated only by pointers are passed as positional arguments to all actions, but values generated by actions are not passed to other actions:

>>> def test(*args, **kwargs):
>>>     print('test:', args, kwargs)
>>>     return 101
>>>
>>> def test_2(*args, **kwargs):
>>>     print('test_2:', args, kwargs)
>>>     return 102
>>>
>>> for data in Sweep(range(3), test, test_2):
>>>     print(data)
test: (0,) {}
test_2: (0,) {}
{}
test: (1,) {}
test_2: (1,) {}
{}
test: (2,) {}
test_2: (2,) {}
{}

Because it would get too confusing otherwise, positional arguments only get passed when originating from a pointer to all actions in a single sweep. Meaning that if we combine two or more sweeps, positional arguments would only get to the actions of their respective Sweeps:

>>> for data in Sweep(range(3), test) * Sweep(zip(['x', 'y'], [True, False]), test):
>>>    print(data)
(0,) {}
('x', True) {}
{}
(1,) {}
('y', False) {}
{}
(2,) {}

As we can see the test function in the second sweep is only getting (x, True) or (y, False) but not any arguments from the first Sweep. It is also important to note that hte values generated by either test function are not being passed to any other object.

In previous examples, the functions we used were accepting the arguments because their signature included variation positional arguments (*args). The situation changes when this is not the case. Actions only receive arguments that they can accept:

>>> def test_3(x=10):
>>>     print(x)
>>>     return True
>>>
>>> for data in Sweep(zip([1,2], [3,4]), test_3):
>>>     pass
1
2

As we can see, test_3 only accepted the first argument.

Keyword Arguments

Passing keyword arguments is more flexible. Any record that gets produced is passed to all subsequent pointers or actions in the sweep that accept that keyword. This is true even across different sub-sweeps. If a pointer yields non-annotated values, these are still used as positional arguments, but only when accepted, and with higher priority given to keywords.

In the following example we can see this in action:

>>> def test(x, y, z=5):
>>>     print(f'my three arguments, x: {x}, y: {y}, z: {z}')
>>>     return x, y, z
>>>
>>> def print_all_args(*args, **kwargs):
>>>     print(f'arguments at the end of the line, args: {args}, kwargs: {kwargs}')
>>>
>>> sweep = sweep_parameter('x', range(3), record_as(test, dep('xx'), dep('yy'), dep('zz'))) * \
>>>         Sweep(range(3), print_all_args)
>>> for data in sweep:
>>>     pass
my three arguments, x: 0, y: None, z: 5
arguments at the end of the line, args:(0,), kwargs:{'x': 0, 'xx': 0, 'zz': 5}
my three arguments, x: 1, y: None, z: 5
arguments at the end of the line, args:(1,), kwargs:{'x': 1, 'xx': 1, 'zz': 5}
my three arguments, x: 2, y: None, z: 5
arguments at the end of the line, args:(2,), kwargs:{'x': 2, 'xx': 2, 'zz': 5}

In the example above we have two different sweeps. The pointer of the first one is producing records which is why we see its value in the test function for x. Since the first sweep is being multiplied to the second sweep we can see how all the records (both produced by the pointer and action) of the first sweep reach the second sweep as keyword arguments, and the non-annotated value of its own pointer reaches the action of the second sweep as a positional argument.

Warning

When creating records, it is very important that each record has a unique name. Having multiple variables create records with the same names, will make the passing of arguments behave in unpredictable ways.

A simple way of renaming conflicting arguments and records is to use the combination of lambda and record_as():

>>> sweep = (
>>>     Sweep(record_as(zip(range(3), range(10,13)), independent('x'), independent('y')), record_as(test, dependent('xx'), dependent('yy'), dependent('zz')))
>>>     @ record_as(lambda xx, yy, zz: test(xx, yy, zz), dependent('some'), dependent('different'), dependent('names'))
>>>     @ print_all_args
>>>     + print_all_args)
>>>
>>> print(sweep.get_data_specs())
>>>
>>> for data in sweep:
>>>     print("data:", data)
(x, y, xx(x, y), yy(x, y), zz(x, y), some(x, y), different(x, y), names(x, y))
my three arguments: 0 10 5
my three arguments: 0 10 5
arguments at the end of the line: () {'x': 0, 'y': 10, 'xx': 0, 'yy': 10, 'zz': 5, 'some': 0, 'different': 10, 'names': 5}
data: {'x': 0, 'y': 10, 'xx': 0, 'yy': 10, 'zz': 5, 'some': 0, 'different': 10, 'names': 5}
my three arguments: 1 11 5
my three arguments: 1 11 5
arguments at the end of the line: () {'x': 1, 'y': 11, 'xx': 1, 'yy': 11, 'zz': 5, 'some': 1, 'different': 11, 'names': 5}
data: {'x': 1, 'y': 11, 'xx': 1, 'yy': 11, 'zz': 5, 'some': 1, 'different': 11, 'names': 5}
my three arguments: 2 12 5
my three arguments: 2 12 5
arguments at the end of the line: () {'x': 2, 'y': 12, 'xx': 2, 'yy': 12, 'zz': 5, 'some': 2, 'different': 12, 'names': 5}
data: {'x': 2, 'y': 12, 'xx': 2, 'yy': 12, 'zz': 5, 'some': 2, 'different': 12, 'names': 5}
arguments at the end of the line: () {'x': 2, 'y': 12, 'xx': 2, 'yy': 12, 'zz': 5, 'some': 2, 'different': 12, 'names': 5}
data: {}

Configuring Sweeps

The class Sweep has three global parameters that are used to configure the behavior of it.

record_none

`Sweep.record_none , True by default, adds None to any action or pointer that didn't generate any record that iteration. This is useful if we want every variable we are storing to be composed of arrays of the same number of items:

>>> def get_random_number():
>>>     return np.random.rand()
>>>
>>> sweep_1 = sweep_parameter('x', range(3), record_as(get_random_number, dependent('y')))
>>> sweep_2 = sweep_parameter('a', range(4), record_as(get_random_number, dependent('b')))
>>> my_sweep = sweep_1 + sweep_2
>>>
>>> Sweep.record_none = False
>>> print(f'----record_none=False----')
>>> for data in my_sweep:
>>>     print(data)
>>>
>>> Sweep.record_none = True
>>> print(f'----record_none=True----')
>>> for data in my_sweep:
>>>     print(data)
----record_none=False----
{'x': 0, 'y': 0.804635124804199}
{'x': 1, 'y': 0.24410055642545125}
{'x': 2, 'y': 0.10828652013926787}
{'a': 0, 'b': 0.4303128288315823}
{'a': 1, 'b': 0.9498154942316515}
{'a': 2, 'b': 0.7150406031589893}
{'a': 3, 'b': 0.2012281139956017}
----record_none=True----
{'x': 0, 'y': 0.22753548379033073, 'a': None, 'b': None}
{'x': 1, 'y': 0.9024597689210428, 'a': None, 'b': None}
{'x': 2, 'y': 0.11393941613249503, 'a': None, 'b': None}
{'x': None, 'y': None, 'a': 0, 'b': 0.8678669225696442}
{'x': None, 'y': None, 'a': 1, 'b': 0.3537275760737344}
{'x': None, 'y': None, 'a': 2, 'b': 0.23555393946522196}
{'x': None, 'y': None, 'a': 3, 'b': 0.19388827122308672}

pass_on_returns

:class:Sweep.pass_on_returns , True by default, specifies if we want arguments to be passed between sweeps. When it is set to False no record will be passed either as positional arguments or as keyword arguments:

>>> sweep = (
>>>     sweep_parameter('y', range(3), record_as(test, dependent('xx'), dependent('yy'), dependent('zz')))
>>>     @ print_all_args)
>>>
>>> Sweep.pass_on_returns = False
>>> print(f'----pass_on_returns=False----')
>>> for data in sweep:
>>>     print("data:", data)
>>>
>>> Sweep.pass_on_returns = True
>>> print(f'----pass_on_returns=True----')
>>> for data in sweep:
>>>     print("data:", data)
----pass_on_returns=False----
my three arguments: None None 5
arguments at the end of the line: () {}
data: {'y': 0, 'xx': None, 'yy': None, 'zz': 5}
my three arguments: None None 5
arguments at the end of the line: () {}
data: {'y': 1, 'xx': None, 'yy': None, 'zz': 5}
my three arguments: None None 5
arguments at the end of the line: () {}
data: {'y': 2, 'xx': None, 'yy': None, 'zz': 5}
----pass_on_returns=True----
my three arguments: None 0 5
arguments at the end of the line: () {'zz': 5, 'y': 0, 'yy': 0}
data: {'y': 0, 'xx': None, 'yy': 0, 'zz': 5}
my three arguments: None 1 5
arguments at the end of the line: () {'zz': 5, 'y': 1, 'yy': 1}
data: {'y': 1, 'xx': None, 'yy': 1, 'zz': 5}
my three arguments: None 2 5
arguments at the end of the line: () {'zz': 5, 'y': 2, 'yy': 2}
data: {'y': 2, 'xx': None, 'yy': 2, 'zz': 5}

pass_on_none

Sweep.pass_on_none, False by default, specifies if variables that return None should be passed as arguments to other actions or Sweeps (Because None is typically indicating that function did not return anything as data even though a record was declared using recording or record_as():

>>> sweep = (
>>>     sweep_parameter('y', range(3), record_as(test, dependent('xx'), dependent('yy'), dependent('zz')))
>>>     @ print_all_args)
>>>
>>> Sweep.pass_on_none = False
>>> print(f'----pass_on_none=False----')
>>> for data in sweep:
>>>     print("data:", data)
>>>
>>> Sweep.pass_on_none = True
>>> print(f'----pass_on_returns=True----')
>>> for data in sweep:
>>>     print("data:", data)
----pass_on_none=False----
my three arguments: None 0 5
arguments at the end of the line: () {'y': 0, 'yy': 0, 'zz': 5}
data: {'y': 0, 'xx': None, 'yy': 0, 'zz': 5}
my three arguments: None 1 5
arguments at the end of the line: () {'y': 1, 'yy': 1, 'zz': 5}
data: {'y': 1, 'xx': None, 'yy': 1, 'zz': 5}
my three arguments: None 2 5
arguments at the end of the line: () {'y': 2, 'yy': 2, 'zz': 5}
data: {'y': 2, 'xx': None, 'yy': 2, 'zz': 5}
----pass_on_returns=True----
my three arguments: None 0 5
arguments at the end of the line: (None,) {'y': 0, 'yy': 0, 'zz': 5, 'xx': None}
data: {'y': 0, 'xx': None, 'yy': 0, 'zz': 5}
my three arguments: None 1 5
arguments at the end of the line: (None,) {'y': 1, 'yy': 1, 'zz': 5, 'xx': None}
data: {'y': 1, 'xx': None, 'yy': 1, 'zz': 5}
my three arguments: None 2 5
arguments at the end of the line: (None,) {'y': 2, 'yy': 2, 'zz': 5, 'xx': None}
data: {'y': 2, 'xx': None, 'yy': 2, 'zz': 5}

Running Sweeps

As seen in previous examples, the most basic way of running a Sweep is to just iterate through it. This is simple but does not do much else. If we only want to store the data generated by a Sweep in disk for later analysis we can use the function run_and_save_sweep():

>>> sweep = sweep_parameter('x', range(3), record_as(my_func, 'y'))
>>> run_and_save_sweep(sweep, './data', 'my_data')
Data location:  data/2022-12-05/2022-12-05T142539_fbfce3e4-my_data/data.ddh5
The measurement has finished successfully and all of the data has been saved.

Note

Because this guide has been ported from an older page plottr is not being used anymore, a fix will come.

run_and_save_sweep() automatically runs the Sweep indicated, stores the records generated by it in a :class:DataDict <plottr.data.datadict.DataDict> in a ddh5 file with time tag followed by a random sequence followed by the third argument, in the directory passed by the second argument. Internally the function utilizes the :class:DDH5Writer <plottr.data.datadict_storage.DDH5Writer> from plottr. For more information on how plottr handles data please see: :doc:../plottr/data.

Note

run_and_save_sweep() can save multiple objects to disk by accepting them as extra arguments. It is a good idea to read over its documentation if you want to be able to save things with it.

Sometimes we have an action that we want to run a single time, some kind of setup function or maybe a closing function (or any single action in between sweeps). If we also need this action to be a Sweep, the function once() will create a Sweep with no pointer that runs an action a single time:

>>> def startup_function():
>>>     print(f'starting an instrument')
>>>
>>> def closing_function():
>>>     print(f'closing an instrument')
>>>
>>> sweep = sweep_parameter('x', range(3), record_as(my_func, 'y'))
>>> starting_sweep = once(startup_function)
>>> closing_sweep = once(closing_function)
>>>
>>> for data in starting_sweep + sweep + closing_sweep:
>>>     print(data)
starting an instrument
{}
{'x': 0, 'y': 1}
{'x': 1, 'y': 1}
{'x': 2, 'y': 1}
closing an instrument
{}

Reference

Sweep Module

AsyncRecord

Base class decorator used to record asynchronous data from instrument. Use the decorator with create_background_sweep function to create Sweeps that collect asynchronous data from external devices running experiments independently of the measurement PC, e.i. the measuring happening is not being controlled by a Sweep but instead an external device (e.g. the OPX). Each instrument should have its own custom setup_wrapper (see setup_wrapper docstring for more info), and a custom collector. Auxiliary functions for the start_wrapper and collector should also be located in this class.

Parameters:

Name Type Description Default
specs

A list of the DataSpecs to record the data produced.

 ()
Source code in labcore/measurement/sweep.py 
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
class AsyncRecord:
    """
    Base class decorator used to record asynchronous data from instrument.
    Use the decorator with create_background_sweep function to create Sweeps that collect asynchronous data from
    external devices running experiments independently of the measurement PC,
    e.i. the measuring happening is not being controlled by a Sweep but instead an external device (e.g. the OPX).
    Each instrument should have its own custom setup_wrapper (see setup_wrapper docstring for more info),
    and a custom collector.
    Auxiliary functions for the start_wrapper and collector should also be located in this class.

    :param specs: A list of the DataSpecs to record the data produced.
    """

    wrapped_setup: Callable

    def __init__(self, *specs):
        self.specs = specs
        self.communicator = {}

    def __call__(self, fun) -> Callable:
        """
        When the decorator is called the experiment function gets wrapped so that it returns an Sweep object composed
        of 2 different Sweeps, the setup sweep and the collector Sweep.
        """

        def sweep(collector_options={}, **setup_kwargs) -> Sweep:
            """
            Returns a Sweep comprised of 2 different Sweeps: start_sweep and collector_sweep.
            start_sweep should perform any setup actions as well as starting the actual experiment. This sweep is only
            executed once. collector_sweep is iterated multiple time to collect all the data generated from the
            instrument.

            :param collector_kwargs: Any arguments that the collector needs.
            """
            start_sweep = once(self.wrap_setup(fun))
            collector_sweep = Sweep(as_pointer(self.collect, *self.specs).using(**collector_options))
            ret = start_sweep + collector_sweep
            ret.set_options(**{fun.__name__: setup_kwargs})
            return ret

        return sweep

    def wrap_setup(self, fun: Callable, *args: Any, **kwargs: Any) -> Callable:
        """
        Wraps the start function. setup_wrapper should consist of another function inside of it decorated with @wraps
        with fun as its argument.
        In this case the wrapped function is setup.
        Setup should accept the \*args and \**kwargs of fun. It should also place any returns from fun
        in the communicator. setup_wrapper needs to return the wrapped function (setup).

        :param fun: The measurement function. In the case of the OPX this would be the function that returns the QUA
                    code with any arguments that it might use.
        """
        self.wrapped_setup = partial(self.setup, fun, *args, **kwargs)
        update_wrapper(self.wrapped_setup, fun)
        return self.wrapped_setup

    def setup(self, fun, *args, **kwargs):
        return fun(*args, **kwargs)

    def collect(self, *args, **kwargs) -> Generator[Dict, None, None]:
        yield {}

__call__(fun)

When the decorator is called the experiment function gets wrapped so that it returns an Sweep object composed of 2 different Sweeps, the setup sweep and the collector Sweep.

Source code in labcore/measurement/sweep.py 
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
def __call__(self, fun) -> Callable:
    """
    When the decorator is called the experiment function gets wrapped so that it returns an Sweep object composed
    of 2 different Sweeps, the setup sweep and the collector Sweep.
    """

    def sweep(collector_options={}, **setup_kwargs) -> Sweep:
        """
        Returns a Sweep comprised of 2 different Sweeps: start_sweep and collector_sweep.
        start_sweep should perform any setup actions as well as starting the actual experiment. This sweep is only
        executed once. collector_sweep is iterated multiple time to collect all the data generated from the
        instrument.

        :param collector_kwargs: Any arguments that the collector needs.
        """
        start_sweep = once(self.wrap_setup(fun))
        collector_sweep = Sweep(as_pointer(self.collect, *self.specs).using(**collector_options))
        ret = start_sweep + collector_sweep
        ret.set_options(**{fun.__name__: setup_kwargs})
        return ret

    return sweep

wrap_setup(fun, *args, **kwargs)

Wraps the start function. setup_wrapper should consist of another function inside of it decorated with @wraps with fun as its argument. In this case the wrapped function is setup. Setup should accept the *args and **kwargs of fun. It should also place any returns from fun in the communicator. setup_wrapper needs to return the wrapped function (setup).

Parameters:

Name Type Description Default
fun Callable

The measurement function. In the case of the OPX this would be the function that returns the QUA code with any arguments that it might use.

 required
Source code in labcore/measurement/sweep.py 
587
588
589
590
591
592
593
594
595
596
597
598
599
600
def wrap_setup(self, fun: Callable, *args: Any, **kwargs: Any) -> Callable:
    """
    Wraps the start function. setup_wrapper should consist of another function inside of it decorated with @wraps
    with fun as its argument.
    In this case the wrapped function is setup.
    Setup should accept the \*args and \**kwargs of fun. It should also place any returns from fun
    in the communicator. setup_wrapper needs to return the wrapped function (setup).

    :param fun: The measurement function. In the case of the OPX this would be the function that returns the QUA
                code with any arguments that it might use.
    """
    self.wrapped_setup = partial(self.setup, fun, *args, **kwargs)
    update_wrapper(self.wrapped_setup, fun)
    return self.wrapped_setup

PointerFunction

Bases: FunctionToRecords

A class that allows using a generator function as a pointer.

Source code in labcore/measurement/sweep.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class PointerFunction(FunctionToRecords):
    """A class that allows using a generator function as a pointer."""

    def _iterator2records(self, *args, **kwargs):
        func_args, func_kwargs = map_input_to_signature(self.func_sig,
                                                        *args, **kwargs)
        ret = record_as(self.func(*func_args, **func_kwargs), *self.data_specs)
        return ret

    def __call__(self, *args, **kwargs):
        args = tuple(self._args + list(args))
        kwargs.update(self._kwargs)
        return self._iterator2records(*args, **kwargs)

    def __iter__(self):
        return iter(self._iterator2records(*self._args, **self._kwargs))

    def get_data_specs(self):
        return self.data_specs

    def using(self, *args, **kwargs) -> "PointerFunction":
        """Set the default positional and keyword arguments that will be
        used when the function is called.

        :returns: A copy of the object. This is to allow setting different
            defaults to multiple uses of the function.
        """
        ret = copy.copy(self)
        ret._args = list(args)
        ret._kwargs = kwargs
        return ret

using(*args, **kwargs)

Set the default positional and keyword arguments that will be used when the function is called.

Returns:

Type Description
PointerFunction

A copy of the object. This is to allow setting different defaults to multiple uses of the function.
Source code in labcore/measurement/sweep.py 
53
54
55
56
57
58
59
60
61
62
63
def using(self, *args, **kwargs) -> "PointerFunction":
    """Set the default positional and keyword arguments that will be
    used when the function is called.

    :returns: A copy of the object. This is to allow setting different
        defaults to multiple uses of the function.
    """
    ret = copy.copy(self)
    ret._args = list(args)
    ret._kwargs = kwargs
    return ret

Sweep

Base class for sweeps.

Can be iterated over; for each pointer value the associated actions are executed. Each iteration step produces a record, containing all values produced by pointer and actions that have been annotated as such. (see: :func:.record_as)

Parameters:

Name Type Description Default
pointer Optional[Iterable]

An iterable that defines the steps over which we iterate

 required
actions Callable

A variable number of functions. Each will be called for each iteration step, with the pointer value(s) as arguments, provided the function can accept it.

 ()
Source code in labcore/measurement/sweep.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
class Sweep:
    """Base class for sweeps.

    Can be iterated over; for each pointer value the associated actions are
    executed. Each iteration step produces a record, containing all values
    produced by pointer and actions that have been annotated as such.
    (see: :func:`.record_as`)

    :param pointer: An iterable that defines the steps over which we iterate
    :param actions: A variable number of functions. Each will be called for
        each iteration step, with the pointer value(s) as arguments, provided
        the function can accept it.
    """

    # TODO: (MAYBE?) should we check if there are conflicting record parameters?
    # TODO: some way of passing meta-info around (about the sweep state)
    #   probably nice to have some info on benchmarking, current indices, etc.
    # TODO: need a way to look through all subsweeps, for example to find all
    #   kw arguments of all actions recursively.
    # TODO: support qcodes parameters as action directly.

    # TODO: these flags should maybe be passed on to 'child' sweeps?
    record_none = True
    pass_on_returns = True
    pass_on_none = False

    # TODO: Add the rules.
    @staticmethod
    def update_option_dict(src: Dict[str, Any], target: Dict[str, Any], level: int) -> None:
        """Rules: work in progress :).
        """
        if not isinstance(src, dict) or not isinstance(target, dict):
            raise ValueError('inputs need to be dictionaries.')

        for k, v in src.items():
            if k in target:
                if isinstance(v, dict) and level > 0:
                    Sweep.update_option_dict(src[k], target[k], level=level-1)
            else:
                target[k] = v

    @staticmethod
    def propagate_sweep_options(sweep: "Sweep"):

        try:
            first = sweep.pointer.iterable.first
            Sweep.copy_sweep_options(sweep, first)
        except AttributeError:
            pass

        try:
            second = sweep.pointer.iterable.second
            Sweep.copy_sweep_options(sweep, second)
        except AttributeError:
            pass

    @staticmethod
    def copy_sweep_options(src: "Sweep", target: Optional["Sweep"]):
        if src is target:
            return

        Sweep.update_option_dict(src._action_kwargs, target._action_kwargs, level=2)
        Sweep.propagate_sweep_options(target)

    @staticmethod
    def link_sweep_properties(src: "Sweep", target: "Sweep") -> None:
        """Share state properties between sweeps."""
        for p in ['_state', '_pass_kwargs']:
            if hasattr(src, p):
                setattr(target, p, getattr(src, p))
                iterable = getattr(target.pointer, 'iterable', None)
                if iterable is not None and hasattr(iterable, 'first'):
                    first = getattr(iterable, 'first')
                    setattr(first, p, getattr(src, p))
                if iterable is not None and hasattr(iterable, 'second'):
                    second = getattr(iterable, 'second')
                    setattr(second, p, getattr(src, p))

        Sweep.copy_sweep_options(src, target)

    def __init__(self, pointer: Optional[Iterable], *actions: Callable):
        """Constructor of :class:`.Sweep`."""
        self._state = {}
        self._pass_kwargs = {}
        self._action_kwargs = {}

        if pointer is None:
            self.pointer = null_pointer
        elif isinstance(pointer, (collections.abc.Iterable, Sweep)):
            self.pointer = pointer
        else:
            raise TypeError('pointer needs to be iterable.')

        self.actions = []
        for a in actions:
            self.append_action(a)

    @property
    def state(self):
        return self._state

    @state.setter
    def state(self, value: Dict[str, Any]):
        for k, v in value.items():
            self._state[k] = v

    @property
    def pass_kwargs(self):
        return self._pass_kwargs

    @pass_kwargs.setter
    def pass_kwargs(self, value: Dict[str, Any]):
        for k, v in value.items():
            self._pass_kwargs[k] = v

    @property
    def action_kwargs(self):
        return self._action_kwargs

    @action_kwargs.setter
    def action_kwargs(self, value: Dict[str, Any]):
        for k, v in value.items():
            self._action_kwargs[k] = v

    def __iter__(self):
        return self.run()

    def __add__(self, other: Union[Callable, "Sweep"]) -> "Sweep":
        if isinstance(other, Sweep):
            sweep2 = other
        elif callable(other):
            sweep2 = Sweep(None, other)
        else:
            raise TypeError(f'can only combine with Sweep or callable, '
                            f'not {type(other)}')

        Sweep.link_sweep_properties(self, sweep2)
        return append_sweeps(self, sweep2)

    def __mul__(self, other: Union[Callable, "Sweep"]) -> "Sweep":
        if isinstance(other, Sweep):
            sweep2 = other
        elif callable(other):
            sweep2 = Sweep(self.pointer, other)
        else:
            raise TypeError(f'can only combine with Sweep or callable, '
                            f'not {type(other)}')

        Sweep.link_sweep_properties(self, sweep2)
        return zip_sweeps(self, sweep2)

    def __matmul__(self, other: Union[Callable, "Sweep"]) -> "Sweep":
        if isinstance(other, Sweep):
            sweep2 = other
        elif callable(other):
            sweep2 = Sweep(None, other)
        else:
            raise TypeError(f'can only combine with Sweep or callable, '
                            f'not {type(other)}')

        Sweep.link_sweep_properties(self, sweep2)
        return nest_sweeps(self, sweep2)

    def append_action(self, action: Callable):
        """Add an action to the sweep."""
        if callable(action):
            if produces_record(action):
                self.actions.append(action)
            else:
                self.actions.append(record_as(action))
        else:
            raise TypeError('action must be a callable.')

    def run(self) -> "SweepIterator":
        """Create the iterator for the sweep."""
        return SweepIterator(
            self,
            state=self.state,
            pass_kwargs=self.pass_kwargs,
            action_kwargs=self.action_kwargs)

    # FIXME: currently this only works for actions -- should be used also
    #   for pointer funcs?
    def set_options(self, **action_kwargs: Dict[str, Any]):
        """Configure the sweep actions.

        :param action_kwargs: Keyword arguments to pass to action functions
            format: {'<action_name>': {'key': 'value'}
            <action_name> is what action_function.__name__ returns.
        """
        for func, val in action_kwargs.items():
            self.action_kwargs[func] = val
            Sweep.propagate_sweep_options(self)

    def get_data_specs(self) -> Tuple[DataSpec, ...]:
        """Return the data specs of the sweep."""
        specs = []
        pointer_specs = []
        if produces_record(self.pointer):
            pointer_specs = self.pointer.get_data_specs()
            specs = combine_data_specs(*(list(specs) + list(pointer_specs)))

        for a in self.actions:
            if produces_record(a):
                action_specs = a.get_data_specs()
                pointer_independents = [ds.name for ds in pointer_specs
                                        if ds.depends_on is None]
                for aspec in action_specs:
                    aspec_ = aspec.copy()
                    if aspec_.depends_on is not None:
                        aspec_.depends_on = pointer_independents + aspec_.depends_on

                    specs = combine_data_specs(*(list(specs) + [aspec_]))

        return tuple(specs)

    def __repr__(self):
        ret = self.pointer.__repr__()
        for a in self.actions:
            ret += f" >> {a.__repr__()}"
        ret += f"\n==> {data_specs_label(*self.get_data_specs())}"
        return ret

__init__(pointer, *actions)

Constructor of :class:.Sweep.

Source code in labcore/measurement/sweep.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def __init__(self, pointer: Optional[Iterable], *actions: Callable):
    """Constructor of :class:`.Sweep`."""
    self._state = {}
    self._pass_kwargs = {}
    self._action_kwargs = {}

    if pointer is None:
        self.pointer = null_pointer
    elif isinstance(pointer, (collections.abc.Iterable, Sweep)):
        self.pointer = pointer
    else:
        raise TypeError('pointer needs to be iterable.')

    self.actions = []
    for a in actions:
        self.append_action(a)

append_action(action)

Add an action to the sweep.

Source code in labcore/measurement/sweep.py 
291
292
293
294
295
296
297
298
299
def append_action(self, action: Callable):
    """Add an action to the sweep."""
    if callable(action):
        if produces_record(action):
            self.actions.append(action)
        else:
            self.actions.append(record_as(action))
    else:
        raise TypeError('action must be a callable.')

get_data_specs()

Return the data specs of the sweep.

Source code in labcore/measurement/sweep.py 
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
def get_data_specs(self) -> Tuple[DataSpec, ...]:
    """Return the data specs of the sweep."""
    specs = []
    pointer_specs = []
    if produces_record(self.pointer):
        pointer_specs = self.pointer.get_data_specs()
        specs = combine_data_specs(*(list(specs) + list(pointer_specs)))

    for a in self.actions:
        if produces_record(a):
            action_specs = a.get_data_specs()
            pointer_independents = [ds.name for ds in pointer_specs
                                    if ds.depends_on is None]
            for aspec in action_specs:
                aspec_ = aspec.copy()
                if aspec_.depends_on is not None:
                    aspec_.depends_on = pointer_independents + aspec_.depends_on

                specs = combine_data_specs(*(list(specs) + [aspec_]))

    return tuple(specs)

Share state properties between sweeps.

Source code in labcore/measurement/sweep.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
@staticmethod
def link_sweep_properties(src: "Sweep", target: "Sweep") -> None:
    """Share state properties between sweeps."""
    for p in ['_state', '_pass_kwargs']:
        if hasattr(src, p):
            setattr(target, p, getattr(src, p))
            iterable = getattr(target.pointer, 'iterable', None)
            if iterable is not None and hasattr(iterable, 'first'):
                first = getattr(iterable, 'first')
                setattr(first, p, getattr(src, p))
            if iterable is not None and hasattr(iterable, 'second'):
                second = getattr(iterable, 'second')
                setattr(second, p, getattr(src, p))

    Sweep.copy_sweep_options(src, target)

run()

Create the iterator for the sweep.

Source code in labcore/measurement/sweep.py 
301
302
303
304
305
306
307
def run(self) -> "SweepIterator":
    """Create the iterator for the sweep."""
    return SweepIterator(
        self,
        state=self.state,
        pass_kwargs=self.pass_kwargs,
        action_kwargs=self.action_kwargs)

set_options(**action_kwargs)

Configure the sweep actions.

Parameters:

Name Type Description Default
action_kwargs Dict[str, Any]

Keyword arguments to pass to action functions format: {'': {'key': 'value'} is what action_function.name returns.

 {}
Source code in labcore/measurement/sweep.py 
311
312
313
314
315
316
317
318
319
320
def set_options(self, **action_kwargs: Dict[str, Any]):
    """Configure the sweep actions.

    :param action_kwargs: Keyword arguments to pass to action functions
        format: {'<action_name>': {'key': 'value'}
        <action_name> is what action_function.__name__ returns.
    """
    for func, val in action_kwargs.items():
        self.action_kwargs[func] = val
        Sweep.propagate_sweep_options(self)

update_option_dict(src, target, level) staticmethod

Rules: work in progress :).

Source code in labcore/measurement/sweep.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
@staticmethod
def update_option_dict(src: Dict[str, Any], target: Dict[str, Any], level: int) -> None:
    """Rules: work in progress :).
    """
    if not isinstance(src, dict) or not isinstance(target, dict):
        raise ValueError('inputs need to be dictionaries.')

    for k, v in src.items():
        if k in target:
            if isinstance(v, dict) and level > 0:
                Sweep.update_option_dict(src[k], target[k], level=level-1)
        else:
            target[k] = v

SweepIterator

Iterator for the :class:.Sweep class.

Manages the actual iteration of the pointer, and the execution of action functions. Manages and updates the state of the sweep.

Source code in labcore/measurement/sweep.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
class SweepIterator:
    """Iterator for the :class:`.Sweep` class.

    Manages the actual iteration of the pointer, and the execution of action
    functions. Manages and updates the state of the sweep.
    """

    def __init__(self, sweep: Sweep,
                 state: Dict[str, Any],
                 pass_kwargs=Dict[str, Any],
                 action_kwargs=Dict[str, Dict[str, Any]]):

        self.sweep = sweep
        self.state = state
        self.pass_kwargs = pass_kwargs
        self.action_kwargs = action_kwargs

        if isinstance(self.sweep.pointer, Sweep):
            self.pointer = iter(self.sweep.pointer)
        elif isinstance(self.sweep.pointer, collections.abc.Iterator):
            self.pointer = self.sweep.pointer
        elif isinstance(self.sweep.pointer, collections.abc.Iterable):
            self.pointer = iter(self.sweep.pointer)
        else:
            raise TypeError('pointer needs to be iterable.')

    def __next__(self):
        ret = {}
        next_point = next(self.pointer)
        if produces_record(self.sweep.pointer):
            ret.update(next_point)

        pass_args = []
        if self.sweep.pass_on_returns:
            if isinstance(next_point, (tuple, list)):
                if not self.sweep.pass_on_none:
                    pass_args = [r for r in next_point if r is not None]
                else:
                    pass_args = list(next_point)
            elif isinstance(next_point, dict):
                if not self.sweep.pass_on_none:
                    self.pass_kwargs.update({k: v for k, v in next_point.items()
                                             if v is not None})
                else:
                    self.pass_kwargs.update(next_point)
            else:
                if self.sweep.pass_on_none or next_point is not None:
                    pass_args.append(next_point)

        for a in self.sweep.actions:
            this_action_kwargs = {}
            if self.sweep.pass_on_returns:
                this_action_kwargs.update(self.pass_kwargs)
            this_action_kwargs.update(
                self.action_kwargs.get(a.__name__, {}))

            action_return = a(*pass_args, **this_action_kwargs)
            if produces_record(a):
                ret.update(action_return)

            # actions always return records, so no need to worry about args
            if not self.sweep.pass_on_none:
                self.pass_kwargs.update({k: v for k, v in action_return.items()
                                         if v is not None})
            else:
                self.pass_kwargs.update(action_return)

        if self.sweep.record_none is False:
            for k in list(ret.keys()):
                if ret[k] is None:
                    ret.pop(k)

        return ret

append_sweeps(first, second)

Append two sweeps.

Iteration over the combined sweep will first complete the first sweep, then the second sweep.

Source code in labcore/measurement/sweep.py 
427
428
429
430
431
432
433
434
435
436
437
438
439
440
def append_sweeps(first: Sweep, second: Sweep) -> Sweep:
    """Append two sweeps.

    Iteration over the combined sweep will first complete the first sweep, then
    the second sweep.
    """
    both = IteratorToRecords(
        AppendSweeps(first, second),
        *combine_data_specs(*(list(first.get_data_specs())
                            + list(second.get_data_specs())))
    )
    sweep = Sweep(both)
    Sweep.link_sweep_properties(first, sweep)
    return sweep

as_pointer(fun, *data_specs)

Convenient in-line creation of a pointer function.

Source code in labcore/measurement/sweep.py 
73
74
75
def as_pointer(fun: Callable, *data_specs: DataSpecCreationType) -> PointerFunction:
    """Convenient in-line creation of a pointer function."""
    return pointer(*data_specs)(fun)

nest_sweeps(outer, inner)

Nest two sweeps.

Iteration over the combined sweep will execute the full inner sweep for each iteration step of the outer sweep.

Source code in labcore/measurement/sweep.py 
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
def nest_sweeps(outer: Sweep, inner: Sweep) -> Sweep:
    """Nest two sweeps.

    Iteration over the combined sweep will execute the full inner sweep for each
    iteration step of the outer sweep.
    """
    outer_specs = outer.get_data_specs()
    outer_indeps = [s.name for s in outer_specs if s.depends_on is None]

    inner_specs = [s.copy() for s in inner.get_data_specs()]
    for s in inner_specs:
        if s.depends_on is not None:
            s.depends_on = outer_indeps + s.depends_on

    nested = IteratorToRecords(
        NestSweeps(outer, inner),
        *combine_data_specs(*(list(outer_specs) + inner_specs))
    )
    sweep = Sweep(nested)
    Sweep.link_sweep_properties(outer, sweep)
    return sweep

once(action)

Return a sweep that executes the action once.

Source code in labcore/measurement/sweep.py 
82
83
84
def once(action: Callable) -> "Sweep":
    """Return a sweep that executes the action once."""
    return Sweep(null_pointer, action)

pointer(*data_specs)

Create a decorator for functions that return pointer generators.

Source code in labcore/measurement/sweep.py 
66
67
68
69
70
def pointer(*data_specs: DataSpecCreationType) -> Callable:
    """Create a decorator for functions that return pointer generators."""
    def decorator(func: Callable) -> PointerFunction:
        return PointerFunction(func, *data_specs)
    return decorator

sweep_parameter(param, sweep_iterable, *actions)

Create a sweep over a parameter.

Parameters:

Name Type Description Default
param ParamSpecType

One of: * A string: Generates an independent, scalar data parameter. * A tuple or list: will be passed to the constructor of :class:.DataSpec; see :func:.make_data_spec. * A :class:.DataSpec instance. * A qcodes parameter. In this case the parameter's set method is called for each value during the iteration.

 required
sweep_iterable Iterable

An iterable that generates the values the parameter will be set to.

 required
actions Callable

An arbitrary number of action functions.

 ()
Source code in labcore/measurement/sweep.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def sweep_parameter(param: ParamSpecType, sweep_iterable: Iterable,
                    *actions: Callable) -> "Sweep":
    """Create a sweep over a parameter.

    :param param: One of:

        * A string: Generates an independent, scalar data parameter.
        * A tuple or list: will be passed to the constructor of :class:`.DataSpec`; see :func:`.make_data_spec`.
        * A :class:`.DataSpec` instance.
        * A qcodes parameter. In this case the parameter's ``set`` method is called for each value during the iteration.

    :param sweep_iterable: An iterable that generates the values the parameter
        will be set to.
    :param actions: An arbitrary number of action functions.
    """

    if isinstance(param, str):
        param_ds = independent(param)
    elif isinstance(param, (tuple, list)):
        param_ds = make_data_spec(*param)
    elif isinstance(param, DataSpec):
        param_ds = param
    elif QCODES_PRESENT and isinstance(param, QCParameter):
        param_ds = independent(param.name, unit=param.unit)

        def setfunc(*args, **kwargs):
            param.set(kwargs.get(param.name))

        actions = list(actions)
        actions.insert(0, setfunc)
    else:
        raise TypeError(f"Cannot make parameter from type {type(param)}")

    record_iterator = IteratorToRecords(sweep_iterable, param_ds)
    return Sweep(record_iterator, *actions)

zip_sweeps(first, second)

Zip two sweeps.

Iteration over the combined sweep will elementwise advance the two sweeps together.

Source code in labcore/measurement/sweep.py 
443
444
445
446
447
448
449
450
451
452
453
454
455
456
def zip_sweeps(first: Sweep, second: Sweep) -> Sweep:
    """Zip two sweeps.

    Iteration over the combined sweep will elementwise advance the two sweeps
    together.
    """
    both = IteratorToRecords(
        ZipSweeps(first, second),
        *combine_data_specs(*(list(first.get_data_specs())
                            + list(second.get_data_specs())))
    )
    sweep = Sweep(both)
    Sweep.link_sweep_properties(first, sweep)
    return sweep

Record Module

DataSpec dataclass

Specification for data parameters to be recorded.

Parameters:

Name Type Description Default
name str
required
depends_on None | List[str] | Tuple[str]
None
type str | DataType
'scalar'
unit str
''
Source code in labcore/measurement/record.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@dataclass
class DataSpec:
    """Specification for data parameters to be recorded."""
    #: name of the parameter
    name: str
    #: dependencies. if ``None``, it is independent.
    depends_on: Union[None, List[str], Tuple[str]] = None
    #: information about data format
    type: Union[str, DataType] = 'scalar'
    #: physical unit of the data
    unit: str = ''

    def __post_init__(self):
        if isinstance(self.type, str):
            self.type = DataType(self.type)

    def copy(self) -> "DataSpec":
        """return a deep copy of the DataSpec instance."""
        return copy.deepcopy(self)

    def __repr__(self) -> str:
        ret = self.name
        if self.depends_on is not None and len(self.depends_on) > 0:
            ret += f"({', '.join(list(self.depends_on))})"
        return ret

copy()

return a deep copy of the DataSpec instance.

Source code in labcore/measurement/record.py 
48
49
50
def copy(self) -> "DataSpec":
    """return a deep copy of the DataSpec instance."""
    return copy.deepcopy(self)

DataType

Bases: Enum

Valid options for data types used in :class:DataSpec

Source code in labcore/measurement/record.py 
24
25
26
27
28
29
class DataType(Enum):
    """Valid options for data types used in :class:`DataSpec`"""
    #: scalar (single-valued) data. typically numeric, but also bool, etc.
    scalar = 'scalar'
    #: multi-valued data. typically numpy-arrays.
    array = 'array'

FunctionToRecords

A wrapper that converts a function return to a record.

Source code in labcore/measurement/record.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
class FunctionToRecords:
    """A wrapper that converts a function return to a record."""

    def __init__(self, func, *data_specs):
        self.func = func
        self.func_sig = inspect.signature(self.func)
        self.data_specs = make_data_specs(*data_specs)
        update_wrapper(self, func)

        self._args: List[Any] = []
        self._kwargs: Dict[str, Any] = {}

    def get_data_specs(self):
        return self.data_specs

    def __call__(self, *args, **kwargs):
        args = tuple(self._args + list(args))
        kwargs.update(self._kwargs)
        func_args, func_kwargs = map_input_to_signature(self.func_sig,
                                                        *args, **kwargs)
        ret = self.func(*func_args, **func_kwargs)
        return _to_record(ret, self.get_data_specs())

    def __repr__(self):
        dnames = data_specs_label(*self.data_specs)
        ret = self.func.__name__ + str(self.func_sig)
        ret += f" as {dnames}"
        return ret

    def using(self, *args, **kwargs) -> "FunctionToRecords":
        """Set the default positional and keyword arguments that will be
        used when the function is called.

        :returns: a copy of the object. This is to allow setting different
            defaults to multiple uses of the function.
        """
        ret = copy.copy(self)
        ret._args = list(args)
        ret._kwargs = kwargs
        return ret

using(*args, **kwargs)

Set the default positional and keyword arguments that will be used when the function is called.

Returns:

Type Description
FunctionToRecords

a copy of the object. This is to allow setting different defaults to multiple uses of the function.
Source code in labcore/measurement/record.py 
267
268
269
270
271
272
273
274
275
276
277
def using(self, *args, **kwargs) -> "FunctionToRecords":
    """Set the default positional and keyword arguments that will be
    used when the function is called.

    :returns: a copy of the object. This is to allow setting different
        defaults to multiple uses of the function.
    """
    ret = copy.copy(self)
    ret._args = list(args)
    ret._kwargs = kwargs
    return ret

IteratorToRecords

A wrapper that converts the iteration values to records.

Source code in labcore/measurement/record.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
class IteratorToRecords:
    """A wrapper that converts the iteration values to records."""

    def __init__(self, iterable: Iterable,
                 *data_specs: DataSpecCreationType):
        self.iterable = iterable
        self.data_specs = make_data_specs(*data_specs)

    def get_data_specs(self):
        return self.data_specs

    def __iter__(self):
        for val in self.iterable:
            yield _to_record(val, self.data_specs)

    def __repr__(self):
        from .sweep import CombineSweeps

        ret = self.iterable.__repr__()
        if not isinstance(self.iterable, CombineSweeps):
            dnames = data_specs_label(*self.get_data_specs())
            ret += f" as {dnames}"

        return ret

combine_data_specs(*specs)

Create a tuple of DataSpecs from the inputs. Removes duplicates.

Source code in labcore/measurement/record.py 
121
122
123
124
125
126
127
128
129
130
def combine_data_specs(*specs: DataSpec) -> Tuple[DataSpec, ...]:
    """Create a tuple of DataSpecs from the inputs. Removes duplicates."""
    ret = []
    spec_names = []
    for s in specs:
        if s.name not in spec_names:
            ret.append(s)
            spec_names.append(s.name)

    return tuple(ret)

data_specs_label(*dspecs)

Create a readable label for multiple data specs.

Format:

Parameters:

Name Type Description Default
dspecs DataSpec

data specs as positional arguments.

 ()

Returns:

Type Description
str

label as string.
Source code in labcore/measurement/record.py 
73
74
75
76
77
78
79
80
81
82
def data_specs_label(*dspecs: DataSpec) -> str:
    """Create a readable label for multiple data specs.

    Format:
        {data_name_1 (dep_1, dep_2), data_name_2 (dep_3), etc.}

    :param dspecs: data specs as positional arguments.
    :return: label as string.
    """
    return r"{" + f"{', '.join([d.__repr__() for d in dspecs])}" + r"}"

dependent(name, depends_on=[], unit='', type='scalar')

Create a the spec for a dependent parameter. All arguments are forwarded to the :class:.DataSpec constructor. depends_on may not be set to None.

Source code in labcore/measurement/record.py 
143
144
145
146
147
148
149
150
def dependent(name: str, depends_on: List[str] = [], unit: str = "",
              type: str = 'scalar'):
    """Create a the spec for a dependent parameter.
    All arguments are forwarded to the :class:`.DataSpec` constructor.
    ``depends_on`` may not be set to ``None``."""
    if depends_on is None:
        raise TypeError("'depends_on' may not be None for a dependent.")
    return DataSpec(name, unit=unit, type=type, depends_on=depends_on)

independent(name, unit='', type='scalar')

Create a the spec for an independent parameter. All arguments are forwarded to the :class:.DataSpec constructor. depends_on is set to None.

Source code in labcore/measurement/record.py 
133
134
135
136
137
def independent(name: str, unit: str = '', type: str = 'scalar') -> DataSpec:
    """Create a the spec for an independent parameter.
    All arguments are forwarded to the :class:`.DataSpec` constructor.
    ``depends_on`` is set to ``None``."""
    return DataSpec(name, unit=unit, type=type, depends_on=None)

make_data_spec(value)

Instantiate a DataSpec object.

Parameters:

Name Type Description Default
value DataSpecCreationType

May be one of the following with the following behavior: - A string create a dependent with name given by the string - A tuple of values that can be used to pass to the constructor of :class:.DataSpec - A dictionary entries of which will be passed as keyword arguments to the constructor of :class:.DataSpec - A :class:.DataSpec instance

 required
Source code in labcore/measurement/record.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def make_data_spec(value: DataSpecCreationType) -> DataSpec:
    """Instantiate a DataSpec object.

    :param value:
        May be one of the following with the following behavior:

            - A string create a dependent with name given by the string
            - A tuple of values that can be used to pass to the constructor of :class:`.DataSpec`
            - A dictionary entries of which will be passed as keyword arguments to the constructor of :class:`.DataSpec`
            - A :class:`.DataSpec` instance

    """
    if isinstance(value, str):
        return dependent(value)
    elif isinstance(value, (tuple, list)):
        return DataSpec(*value)
    elif isinstance(value, dict):
        return DataSpec(**value)
    elif isinstance(value, DataSpec):
        return value
    else:
        raise TypeError(f"Cannot create DataSpec from {type(value)}")

make_data_specs(*specs)

Create a tuple of DataSpec instances.

Parameters:

Name Type Description Default
specs DataSpecCreationType

will be passed individually to :func:.make_data_spec

 ()
Source code in labcore/measurement/record.py 
109
110
111
112
113
114
115
116
117
118
def make_data_specs(*specs: DataSpecCreationType) -> Tuple[DataSpec, ...]:
    """Create a tuple of DataSpec instances.

    :param specs: will be passed individually to :func:`.make_data_spec`
    """
    ret = []
    for spec in specs:
        ret.append(make_data_spec(spec))
    ret = tuple(ret)
    return ret

produces_record(obj)

Check if obj is annotated to generate records.

Source code in labcore/measurement/record.py 
180
181
182
183
184
185
def produces_record(obj: Any) -> bool:
    """Check if `obj` is annotated to generate records."""
    if hasattr(obj, 'get_data_specs'):
        return True
    else:
        return False

record_as(obj, *specs)

Annotate produced data as records.

Parameters:

Name Type Description Default
obj Union[Callable, Iterable, Iterator]

a function that returns data or an iterable/iterator that produces data at each iteration step

 required
specs DataSpecCreationType

specs for the data produced (see :func:.make_data_specs)

 ()
Source code in labcore/measurement/record.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
def record_as(obj: Union[Callable, Iterable, Iterator],
              *specs: DataSpecCreationType):
    """Annotate produced data as records.

    :param obj: a function that returns data or an iterable/iterator that
        produces data at each iteration step
    :param specs: specs for the data produced (see :func:`.make_data_specs`)
    """
    specs = make_data_specs(*specs)
    if isinstance(obj, Callable):
        return recording(*specs)(obj)
    elif isinstance(obj, collections.abc.Iterable):
        return IteratorToRecords(obj, *specs)

recording(*data_specs)

Returns a decorator that allows adding data parameter specs to a function.

Source code in labcore/measurement/record.py 
156
157
158
159
160
161
162
def recording(*data_specs: DataSpecCreationType) -> Callable:
    """Returns a decorator that allows adding data parameter specs to a
    function.
    """
    def decorator(func):
        return FunctionToRecords(func, *make_data_specs(*data_specs))
    return decorator

Storage Module

plottr.data.datadict_storage

Provides file-storage tools for the DataDict class.

Description of the HDF5 storage format

We use a simple mapping from DataDict to the HDF5 file. Within the file, a single DataDict is stored in a (top-level) group of the file. The data fields are datasets within that group.

Global meta data of the DataDict are attributes of the group; field meta data are attributes of the dataset (incl., the unit and axes values). The meta data keys are given exactly like in the DataDict, i.e., incl the double underscore pre- and suffix.

run_and_save_sweep(sweep, data_dir, name, ignore_all_None_results=True, save_action_kwargs=False, add_timestamps=False, archive_files=None, return_data=False, safe_write_mode=False, **extra_saving_items)

Iterates through a sweep, saving the data coming through it into a file called at directory.

Parameters:

Name Type Description Default
sweep Sweep

Sweep object to iterate through.

 required
data_dir str

Directory of file location.

 required
name str

Name of the file.

 required
ignore_all_None_results bool

if True, don't save any records that contain a None. if False, only do not save records that are all-None.

 True
save_action_kwargs

If True, the action_kwargs of the sweep will be saved as a json file named after the first key of the kwargs dctionary followed by '_action_kwargs' in the same directory as the data.

 False
archive_files Optional[List[str]]

List of files to copy into a folder called 'archived_files' in the same directory that the data is saved. It should be a list of paths (str), regular expressions are supported. If a folder is passed, it will copy the entire folder and all of its subdirectories and files into the archived_files folder. If one of the arguments could not be found, a message will be printed and the measurement will be performed without the file being archived. An exception is raised if the type is invalid. e.g. archive_files=['.txt', 'calibration_files', '../test_file.py']. '.txt' will copy every txt file located in the working directory. 'calibration_files' will copy the entire folder called calibration_files from the working directory into the archived_files folder. '../test_file.py' will copy the script test_file.py from one directory above the working directory.

 None
extra_saving_items

Kwargs for extra objects that should be saved. If the kwarg is a dictionary, the function will try and save it as a JSON file. If the dictionary contains objects that are not JSON serializable it will be pickled. Any other kind of object will be pickled too. The files will have their keys as names.

 {}
safe_write_mode bool

Indicates if the data should be written in safe mode or not. Look into ddh5 writer for more info.

 False

Raises:

Type Description
TypeError

A Typerror is raised if the object passed for archive_files is not correct
Source code in labcore/measurement/storage.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def run_and_save_sweep(sweep: Sweep,
                       data_dir: str,
                       name: str,
                       ignore_all_None_results: bool = True,
                       save_action_kwargs: bool = False,
                       add_timestamps = False,
                       archive_files: Optional[List[str]] = None,
                       return_data: bool = False,
                       safe_write_mode: bool = False,
                       **extra_saving_items) -> Tuple[Union[str, Path], Optional[DataDict]]:
    """
    Iterates through a sweep, saving the data coming through it into a file called <name> at <data_dir> directory.

    :param sweep: Sweep object to iterate through.
    :param data_dir: Directory of file location.
    :param name: Name of the file.
    :param ignore_all_None_results: if ``True``, don't save any records that contain a ``None``.
        if ``False``, only do not save records that are all-``None``.
    :param  save_action_kwargs: If ``True``, the action_kwargs of the sweep will be saved as a json file named after
        the first key of the kwargs dctionary followed by '_action_kwargs' in the same directory as the data.
    :param archive_files: List of files to copy into a folder called 'archived_files' in
        the same directory that the data is saved. It should be a list of paths (str), regular expressions are supported.
        If a folder is passed, it will copy the entire folder and all of its subdirectories and files into the
        archived_files folder. If one of the arguments could not be found, a message will be printed and the measurement
        will be performed without the file being archived. An exception is raised if the type is invalid.

        e.g. archive_files=['*.txt', 'calibration_files', '../test_file.py'].  '*.txt' will copy every txt file
        located in the working directory. 'calibration_files' will copy the entire folder called calibration_files from
        the working directory into the archived_files folder. '../test_file.py' will copy the script test_file.py from
        one directory above the working directory.
    :param extra_saving_items: Kwargs for extra objects that should be saved. If the kwarg is a dictionary, the function
        will try and save it as a JSON file. If the dictionary contains objects that are not JSON serializable it will
        be pickled. Any other kind of object will be pickled too. The files will have their keys as names.
    :param safe_write_mode: Indicates if the data should be written in safe mode or not. Look into ddh5 writer for more
        info.

    :raises TypeError: A Typerror is raised if the object passed for archive_files is not correct
    """
    data_dict = _create_datadict_structure(sweep)

    # Creates a file even when it fails.
    with DDH5Writer(data_dict, data_dir, name=name, safe_write_mode=safe_write_mode) as writer:

        # Saving meta-data
        dir: Path = writer.filepath.parent
        if add_timestamps:
            t = time.localtime()
            time_stamp = time.strftime(TIMESTRFORMAT, t) + '_'

        for key, val in extra_saving_items.items():
            if callable(val):
                value = val()
            else:
                value = val

            if add_timestamps:
                pickle_path_file = os.path.join(dir, time_stamp + key + '.pickle')
                json_path_file = os.path.join(dir, time_stamp + key + '.json')
            else:
                pickle_path_file = os.path.join(dir, key + '.pickle')
                json_path_file = os.path.join(dir, key + '.json')

            if isinstance(value, dict):
                try:
                    _save_dictionary(value, json_path_file)
                except TypeError as error:
                    # Delete the file created by _save_dictionary. This file does not contain the complete dictionary.
                    if os.path.isfile(json_path_file):
                        os.remove(json_path_file)

                    logging.info(f'{key} has not been able to save to json: {error.args}.'
                                 f' The item will be pickled instead.')
                    _pickle_and_save(value, pickle_path_file)
            else:
                _pickle_and_save(value, pickle_path_file)

        # Save the kwargs
        if save_action_kwargs:
            if add_timestamps:
                json_path_file = os.path.join(dir, time_stamp + 'sweep_action_kwargs.json')
            else:
                json_path_file = os.path.join(dir, 'sweep_action_kwargs.json')
            _save_dictionary(sweep.action_kwargs, json_path_file)

        # Save archive_files
        if archive_files is not None:
            archive_files_dir = os.path.join(dir, 'archive_files')
            os.mkdir(archive_files_dir)
            if not isinstance(archive_files, list) and not isinstance(archive_files, tuple):
                if isinstance(archive_files, str):
                    archive_files = [archive_files]
                else:
                    raise TypeError(f'{type(archive_files)} is not a list.')
            for path in archive_files:
                if os.path.isdir(path):
                    folder_name = os.path.basename(path)
                    if folder_name == '':
                        folder_name = os.path.basename(os.path.dirname(path))

                    shutil.copytree(path, os.path.join(archive_files_dir, folder_name), dirs_exist_ok=True)
                elif os.path.isfile(path):
                    shutil.copy(path, archive_files_dir)
                else:
                    matches = glob.glob(path, recursive=True)
                    if len(matches) == 0:
                        logging.info(f'{path} could not be found. Measurement will continue without archiving {path}')
                    for file in matches:
                        shutil.copy(file, archive_files_dir)

        # Save data.
        try:
            for line in sweep:
                if not _check_none(line, all=ignore_all_None_results):
                    writer.add_data(**line)
        except KeyboardInterrupt:
            logger.warning('Sweep stopped by Keyboard interrupt. Data completed before interrupt should be saved.')
            ret = (dir, data_dict) if return_data else (dir, None)
            return ret

    logger.info('The measurement has finished successfully and all of the data has been saved.')
    ret = (dir, data_dict) if return_data else (dir, None)
    return ret