accelforge.util package#

Submodules#

accelforge.util.exceptions module#

Public exceptions for AccelForge.

exception accelforge.util.exceptions.EvaluationError[source]#

Bases: Exception

Exception raised when parsing fails.

This exception is raised when there’s an error parsing specifications, architectures, workloads, or mappings.

Parameters:

  • *args – Standard exception arguments.

  • source_field (Any) – The field where the error occurred.

  • message (str) – Error message describing what went wrong.

  • **kwargs – Additional keyword arguments.

__init__(*args, source_field=None, message=None, **kwargs)[source]#
add_field(field)[source]#

Add a field to the error context. The output error message will include the field path as a period-separated string like “spec.arch.nodes.0.name”.

Parameters:

field (Any) – The field to add to the error context.

accelforge.util.indent module#

accelforge.util.indent.print(*args, sep=' ', end='\n', file=None, flush=False)[source]#
accelforge.util.indent.tqdm(*args, **kwargs)[source]#

accelforge.util.parallel module#

accelforge.util.parallel.delayed(function)[source]#

Decorator used to capture the arguments of a function.

Parameters:

function (callable) – The function to be decorated.

Returns:

A new function F such that calling F(*args, **kwargs) returns a tuple (function, args, kwargs), allowing the later execution of function(*args, **kwargs).

Return type:

callable

Notes

Be careful about the order in which decorators are applied, especially when using Memory.cache. For instance, Memory.cache(delayed(func)) will cache the outputs of delayed(func), that is, tuples of the form (func, args, kwargs). To cache the outputs of func itself, you must instead use delayed(Memory.cache(func)).

accelforge.util.parallel.get_n_parallel_jobs()[source]#

Returns the number of parallel jobs being used. If parallel processing is not enabled, returns 1.

Return type:

int

accelforge.util.parallel.is_using_parallel_processing()[source]#

Returns True if parallel processing is enabled.

Return type:

bool

accelforge.util.parallel.parallel(jobs, n_jobs=None, pbar=None, pbar_position=0, return_as=None)[source]#

Parallelizes a list of jobs.

Parameters:

  • jobs (list[tuple[Callable, tuple, dict]]) – The jobs to parallelize. The first element of each tuple is a function, the second is a tuple of arguments, and the third is a dictionary of keyword arguments.

  • n_jobs (int) – The number of jobs to run in parallel. If not provided, the number of parallel jobs is set to the number of CPU cores.

  • pbar (str) – A label for a progress bar. If not provided, no progress bar is shown.

  • pbar_position (int) – The position of the progress bar. If not provided, the progress bar is shown at the beginning of the output.

  • return_as (str) – The type of return value. If not provided, the return value is a list.

Returns:

The result of the parallelized jobs.

Return type:

Union[list[Any], Generator[Any, None, None], dict[Any, Any]]

accelforge.util.parallel.set_n_parallel_jobs(n_jobs, print_message=False)[source]#

Set the number of parallel jobs to use.

Parameters:

  • n_jobs (int) – The number of parallel jobs to use.

  • print_message (bool) – Whether to print a message when the number of parallel jobs is set.

Return type:

None

Module contents#

exception accelforge.util.EvaluationError[source]#

Bases: Exception

Exception raised when parsing fails.

This exception is raised when there’s an error parsing specifications, architectures, workloads, or mappings.

Parameters:

  • *args – Standard exception arguments.

  • source_field (Any) – The field where the error occurred.

  • message (str) – Error message describing what went wrong.

  • **kwargs – Additional keyword arguments.

__init__(*args, source_field=None, message=None, **kwargs)[source]#
add_field(field)[source]#

Add a field to the error context. The output error message will include the field path as a period-separated string like “spec.arch.nodes.0.name”.

Parameters:

field (Any) – The field to add to the error context.

class accelforge.util.LiteralString[source]#

Bases: str

A string literal that should not be evaluated.

accelforge.util.NUMPY_FLOAT_TYPE#

alias of float32

class accelforge.util.Permutation[source]#

Bases: object

A sequence on integers from the set of [0,N) representing a permutation of a sequence that has N elements.

__init__(permutation)[source]#
apply(sequence, include_remaining_unpermuted=True)[source]#

Apply permutation to sequence. If include_remaining_unpermuted and the sequence is longer than the permutation, then the remainder of the sequence is included unpermuted. Otherwise, the remainder of the sequence is omitted.

Return type:

Iterable[TypeVar(T)]

copy()[source]#
Return type:

Permutation

get_prefix(n)[source]#
Return type:

Permutation

accelforge.util.delayed(function)[source]#

Decorator used to capture the arguments of a function.

Parameters:

function (callable) – The function to be decorated.

Returns:

A new function F such that calling F(*args, **kwargs) returns a tuple (function, args, kwargs), allowing the later execution of function(*args, **kwargs).

Return type:

callable

Notes

Be careful about the order in which decorators are applied, especially when using Memory.cache. For instance, Memory.cache(delayed(func)) will cache the outputs of delayed(func), that is, tuples of the form (func, args, kwargs). To cache the outputs of func itself, you must instead use delayed(Memory.cache(func)).

class accelforge.util.fzs[source]#

Bases: frozenset[T], Generic[T]

static __new__(cls, *args, **kwargs)[source]#
accelforge.util.get_n_parallel_jobs()[source]#

Returns the number of parallel jobs being used. If parallel processing is not enabled, returns 1.

Return type:

int

accelforge.util.is_using_parallel_processing()[source]#

Returns True if parallel processing is enabled.

Return type:

bool

class accelforge.util.oset[source]#

Bases: set, Generic[T]

Set that iterates in sorted order for deterministic behavior.

copy()[source]#

Return a shallow copy of a set.

difference(*others)[source]#

Return the difference of two or more sets as a new set.

(i.e. all elements that are in this set but not the others.)

intersection(*others)[source]#

Return the intersection of two sets as a new set.

(i.e. all elements that are in both sets.)

pop()[source]#

Remove and return an arbitrary set element. Raises KeyError if the set is empty.

symmetric_difference(other)[source]#

Return the symmetric difference of two sets as a new set.

(i.e. all elements that are in exactly one of the sets.)

union(*others)[source]#

Return the union of sets as a new set.

(i.e. all elements that are in either set.)

accelforge.util.parallel(jobs, n_jobs=None, pbar=None, pbar_position=0, return_as=None)[source]#

Parallelizes a list of jobs.

Parameters:

  • jobs (list[tuple[Callable, tuple, dict]]) – The jobs to parallelize. The first element of each tuple is a function, the second is a tuple of arguments, and the third is a dictionary of keyword arguments.

  • n_jobs (int) – The number of jobs to run in parallel. If not provided, the number of parallel jobs is set to the number of CPU cores.

  • pbar (str) – A label for a progress bar. If not provided, no progress bar is shown.

  • pbar_position (int) – The position of the progress bar. If not provided, the progress bar is shown at the beginning of the output.

  • return_as (str) – The type of return value. If not provided, the return value is a list.

Returns:

The result of the parallelized jobs.

Return type:

Union[list[Any], Generator[Any, None, None], dict[Any, Any]]

accelforge.util.set_n_parallel_jobs(n_jobs, print_message=False)[source]#

Set the number of parallel jobs to use.

Parameters:

  • n_jobs (int) – The number of parallel jobs to use.

  • print_message (bool) – Whether to print a message when the number of parallel jobs is set.

Return type:

None

accelforge.util.tqdm(*args, **kwargs)[source]#