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.parallel module#

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

Decorator used to capture the arguments of a function.

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

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

Decorator used to capture the arguments of a function.

class accelforge.util.fzs[source]#

Bases: frozenset[T], Generic[T]

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