pipefunc.sweep module

Provides the pipefunc.sweep module, for creating and managing parameter sweeps.

class pipefunc.sweep.Sweep(items, dims=None, exclude=None, constants=None, derivers=None)

Bases: object

Create a sweep of a pipeline.

Considering certain dimensions to be linked together (zipped) rather than forming a full Cartesian product. If ‘dims’ is not provided, a Cartesian product is formed.

Parameters:

• items (dict[str, Sequence[Any]]) – A dictionary where the key is the name of the dimension and the value is a sequence of values for that dimension.

• dims (list[str | tuple[str, ...]] | None) – A list of tuples. Each tuple contains names of dimensions that are linked together. If not provided, a Cartesian product is formed.

• exclude (Callable[[Mapping[str, Any]], bool] | None) – A function that takes a dictionary of dimension values and returns True if the combination should be excluded from the sweep.

• constants (Mapping[str, Any] | None) – A dictionary of constant values to be included in each combination.

• derivers (dict[str, Callable[[dict[str, Any]], Any]] | None) – A dictionary of functions to be applied to each dict. The dictionary keys are attribute names and the values are functions that take a dict as input and return a new attribute value. The keys might be a subset of the items keys, which means the values will be overwritten.

Returns:

• A list of dictionaries, each representing a specific combination

• of dimension values.

Examples

>>> items = {'a': [1, 2], 'b': [3, 4], 'c': [5, 6]}
>>> Sweep(items)
[{'a': 1, 'b': 3, 'c': 5}, {'a': 1, 'b': 3, 'c': 6}, {'a': 1, 'b': 4, 'c': 5}, {'a': 1, 'b': 4, 'c': 6},
 {'a': 2, 'b': 3, 'c': 5}, {'a': 2, 'b': 3, 'c': 6}, {'a': 2, 'b': 4, 'c': 5}, {'a': 2, 'b': 4, 'c': 6}]

Which is equivalent to the following:

>>> Sweep(items, dims=["a", "b", "c"]).list()
>>> Sweep(items, dims=[("a",), ("b",), ("c",)]).list()

Or zip together dimensions:

>>> Sweep(items, dims=[('a', 'b'), ('c',)]).list()
[{'a': 1, 'b': 3, 'c': 5}, {'a': 2, 'b': 4, 'c': 6}].list()

generate()

Generate the sweep combinations.

Returns the same combinations as the list method, but as a generator.

Yields:

A dictionary representing a specific combination of dimension values.

Return type:

Generator[dict[str, Any], None, None]

list()

Return the sweep as a list.

Return type:

list[dict[str, Any]]

filtered_sweep(keys)

Return the sweep as a list, but only include the specified keys in each dictionary, and remove duplicates.

Return type:

Sweep

combine(other)

Add another sweep to this MultiSweep.

Return type:

MultiSweep

product(*others)

Create a Cartesian product of this Sweep with other Sweeps.

Parameters:

*others (Sweep) – One or more Sweep objects to create a Cartesian product with.

Return type:

Sweep

Returns:

A new Sweep object representing the Cartesian product of the sweeps.

Examples

>>> sweep1 = Sweep({'a': [1, 2], 'b': [3, 4]})
>>> sweep2 = Sweep({'c': [5, 6]})
>>> sweep3 = sweep1.product(sweep2)
>>> sweep3.list()
[{'a': 1, 'b': 3, 'c': 5}, {'a': 1, 'b': 3, 'c': 6},
 {'a': 1, 'b': 4, 'c': 5}, {'a': 1, 'b': 4, 'c': 6},
 {'a': 2, 'b': 3, 'c': 5}, {'a': 2, 'b': 3, 'c': 6},
 {'a': 2, 'b': 4, 'c': 5}, {'a': 2, 'b': 4, 'c': 6}]

add_derivers(**derivers)

Add derivers to the sweep, which are functions that modify the sweep items.

Parameters:

derivers (Callable[[dict[str, Any]], Any]) – A dictionary of functions to be applied to each dict. The dictionary keys are attribute names and the values are functions that take a dict as input and return a new attribute value. The keys might be a subset of the items keys, which means the values will be overwritten.

Return type:

Sweep

Returns:

A new Sweep object with the added derivers.

Examples

>>> sweep = Sweep({'a': [1], 'b': [2, 3]})
>>> sweep = sweep.add_derivers(c=lambda x: x['a'] + x['b'])
>>> sweep.list()
[{'a': 1, 'b': 2, 'c': 3}, {'a': 1, 'b': 3, 'c': 4}]

class pipefunc.sweep.MultiSweep(*sweeps)

Bases: Sweep

A class to concatenate multiple sweeps into one.

Parameters:

*sweeps (Sweep) – Sweep objects to be concatenated together.

Return type:

A MultiSweep object containing the concatenated sweeps.

Examples

>>> sweep1 = Sweep({'a': [1, 2], 'b': [3, 4]})
>>> sweep2 = Sweep({'x': [5, 6], 'y': [7, 8]})
>>> multi_sweep = MultiSweep(sweep1, sweep2)

generate()

Generate the sweep combinations.

Returns the same combinations as the list method, but as a generator.

Yields:

A dictionary representing a specific combination of dimension values.

Return type:

Generator[dict[str, Any], None, None]

list()

Return the sweep as a list.

Return type:

list[dict[str, Any]]

filtered_sweep(keys)

Return a new MultiSweep, but only include the specified keys in each dictionary, and remove duplicates.

Return type:

MultiSweep

combine(other)

Add another sweep to this MultiSweep.

Return type:

MultiSweep

pipefunc.sweep.generate_sweep(items, dims=None, exclude=None, constants=None, derivers=None)

Create a sweep of a pipeline.

Considering certain dimensions to be linked together (zipped) rather than forming a full Cartesian product. If ‘dims’ is not provided, a Cartesian product is formed.

Parameters:

• items (dict[str, Sequence[Any]]) – A dictionary where the key is the name of the dimension and the value is a sequence of values for that dimension.

• dims (list[str | tuple[str, ...]] | None) – A list of tuples. Each tuple contains names of dimensions that are linked together. If not provided, a Cartesian product is formed.

• exclude (Callable[[Mapping[str, Any]], bool] | None) – A function that takes a dictionary of dimension values and returns True if the combination should be excluded from the sweep.

• constants (Mapping[str, Any] | None) – A dictionary with constant values that should be added to each combination.

• derivers (dict[str, Callable[[dict[str, Any]], Any]] | None) – A dictionary of functions to be applied to each dict. The dictionary keys are attribute names and the values are functions that take a dict as input and return a new attribute value. The keys might be a subset of the items keys, which means the values will be overwritten.

Return type:

list[dict[str, Any]]

Returns:

A list of dictionaries, each representing a specific combination of dimension values.

Examples

>>> items = {'a': [1, 2], 'b': [3, 4], 'c': [5, 6]}
>>> generate_sweep(items)
[{'a': 1, 'b': 3, 'c': 5}, {'a': 1, 'b': 3, 'c': 6}, {'a': 1, 'b': 4, 'c': 5}, {'a': 1, 'b': 4, 'c': 6},
 {'a': 2, 'b': 3, 'c': 5}, {'a': 2, 'b': 3, 'c': 6}, {'a': 2, 'b': 4, 'c': 5}, {'a': 2, 'b': 4, 'c': 6}]

Which is equivalent to the following:

>>> generate_sweep(items, dims=["a", "b", "c"])
>>> generate_sweep(items, dims=[("a",), ("b",), ("c",)])

Or zip together dimensions:

>>> generate_sweep(items, dims=[('a', 'b'), ('c',)])
[{'a': 1, 'b': 3, 'c': 5}, {'a': 2, 'b': 4, 'c': 6}]

pipefunc.sweep.count_sweep(output_name, sweep, pipeline, *, use_pandas=False)

Count the number of times each argument combination is used.

Useful for determining which functions to execute first or which functions to cache.

Parameters:

• output_name (str) – The name of the output to count the argument combinations for.

• sweep (list[dict[str, Any]] | Sweep) – A list of dictionaries, each representing a values for which the function is called. Or a Sweep object.

• pipeline (Pipeline) – The pipeline to count the argument combinations for.

• use_pandas (bool) – Whether to use pandas to create the counts. Note that this is slower than the default method for sweeps <≈ 1e6.

Return type:

dict[str | tuple[str, ...], dict[tuple[Any, ...], int]]

Returns:

A dictionary where the keys are the names of the arguments and the values are dictionaries where the keys are the argument combinations and the values are the number of times that combination is used.

pipefunc.sweep.set_cache_for_sweep(output_name, pipeline, sweep, min_executions=2, *, verbose=False)

Set the cache for a sweep of a pipeline.

Return type:

None