Workflow Utilities

ropt.workflow

Optimization workflow functionality.

find_backend_plugin

find_backend_plugin(method: str) -> str | None

Find an optimizer plugin for a given method.

The method argument can be specified in two ways:

1. Explicit Plugin: "plugin-name/method-name" checks if the specific plugin named plugin-name supports method-name.
2. Implicit Plugin: "method-name" searches through all discoverable plugins to see if any support method-name.

Parameters:

Name	Type	Description	Default
method	str	The method name.	required

Returns:

Type	Description
str | None	The name of the plugin that implements the optimizer method or None.

find_sampler_plugin

find_sampler_plugin(method: str) -> str | None

Find a sampler plugin for a given method.

The method argument can be specified in two ways:

1. Explicit Plugin: "plugin-name/method-name" checks if the specific plugin named plugin-name supports method-name.
2. Implicit Plugin: "method-name" searches through all discoverable plugins to see if any support method-name.

Parameters:

Name	Type	Description	Default
method	str	The method name.	required

Returns:

Type	Description
str | None	The name of the plugin that implements the sampler method or None.

validate_backend_options

validate_backend_options(
    method: str, options: dict[str, Any] | list[str]
) -> None

Validate the optimizer-specific options for a given method.

The method argument can be specified in two ways:

1. Explicit Plugin: "plugin-name/method-name" checks if the specific plugin named plugin-name supports method-name.
2. Implicit Plugin: "method-name" searches through all discoverable plugins to see if any support method-name.

Parameters:

Name	Type	Description	Default
method	str	The specific optimization method name.	required
options	dict[str, Any] | list[str]	The dictionary or a list of strings of options.	required

dispatch_tasks async

dispatch_tasks(
    functions: Sequence[Callable[[], None]]
    | Mapping[str, Callable[[], None]],
    server: Literal["async", "multiprocessing", "hpc"],
    *,
    report: Callable[[Any], None] | None = None,
    workers: int = 4,
    workdir: str = "./",
    cluster: str | None = None,
    queue: str | None = None,
    cores: int = 1,
) -> list[Any]

Dispatch a list of functions to run in parallel.

The dispatched functions will run either in threads, in a multiprocessing pool, or on a HPC server.

Parameters:

Name	Type	Description	Default
functions	Sequence[Callable[[], None]] | Mapping[str, Callable[[], None]]	The functions to run.	required
server	Literal['async', 'multiprocessing', 'hpc']	The type of server to run the functions.	required
report	Callable[[Any], None] | None	Optional report function.	None
workers	int	The number of workers to run in parallel.	4
workdir	str	Working directory used by the HPC server.	'./'
cluster	str | None	The name of the HPC cluster to use.	None
queue	str | None	Optional queue to use on the cluster.	None
cores	int	Optional number of cores per task.	1

Returns:

Type	Description
list[Any]	A list of function results.

Raises:

Type	Description
ValueError	If server has an invalid value.

Current working directory.

The functions to run cannot rely on the current directory to be set consistently. Assume that the current directory is unknown and use absolute paths to read or write files.

ALso, note that setting the current directory may not have the desired effect when using the thread server, since changing it in one thread affects all threads. In case of the multiprocessing and hpc servers, the current directory can be changed safely if needed.