autojob package

Tools for DFT job automation on massively parallel computing resources.

Subpackages

Submodules

autojob.hpc module

Interface with high performance computing resources.

class autojob.hpc.JobState(value)[source]

Bases: StrEnum

The state of a scheduler job.

The meanings of each state are defined in the sacct SLURM documentation (https://slurm.schedmd.com/sacct.html). The lone exception is “UNKNOWN”, which is applicable when the job accounting utility is not available.

BOOT_FAIL = 'BF'
CANCELLED = 'CA'
COMPLETED = 'CD'
DEADLINE = 'DL'
FAILED = 'F'
NODE_FAIL = 'NF'
OUT_OF_MEMORY = 'OOM'
PENDING = 'PD'
PREEMPTED = 'PR'
REQUEUED = 'RQ'
RESIZING = 'RS'
REVOKED = 'RV'
RUNNING = 'R'
SUSPENDED = 'S'
TIMEOUT = 'TO'
UNKNOWN = 'UNK'
class autojob.hpc.Partition(value)[source]

Bases: Enum

A partition on a computing cluster.

BIGMEM = 'bigmem'
CPU2013 = 'cpu2013'
CPU2017_BF05 = 'cpu2017-bf05'
CPU2019 = 'cpu2019'
CPU2019_BF05 = 'cpu2019-bf05'
CPU2021 = 'cpu2021'
CPU2021_BF05 = 'cpu2021-bf05'
CPU2021_BF24 = 'cpu2021-bf24'
CPU2022 = 'cpu2022'
CPU2022_BF24 = 'cpu2022-bf24'
CPU2023 = 'cpu2023'
GPU_V100 = 'gpu-v100'
LATTICE = 'lattice'
PARALLEL = 'parallel'
RAZI = 'razi'
SINGLE = 'single'
property cluster_name

The name of the cluster with underscores replaced with hyphens.

property cpus_per_node

The number of cores per node in the partition.

property gpu

Whether the partition is GPU-enabled.

property max_mem_per_node

The maximum memory per node on the partition.

property nodes

The number of nodes in the partition.

property time_limit

The maximum time limit for the partition.

class autojob.hpc.ScheduledMixin[source]

Bases: object

A mixin class for scheduled tasks.

scheduler_inputs: SchedulerInputs = FieldInfo(annotation=NoneType, required=False, default_factory=SchedulerInputs, description='The scheduler intputs')
scheduler_outputs: SchedulerOutputs | None = FieldInfo(annotation=NoneType, required=False, default=None, description='The scheduler statistics and outputs')
class autojob.hpc.SchedulerInputs(*, job_name: str | None = None, account: str | None = None, partition: list[Partition] | None = None, mem: Annotated[int, WrapValidator(func=validate_memory, json_schema_input_type=PydanticUndefined), WrapSerializer(func=serialize_memory, return_type=PydanticUndefined, when_used=json)] | None = None, mem_per_cpu: Annotated[int, WrapValidator(func=validate_memory, json_schema_input_type=PydanticUndefined), WrapSerializer(func=serialize_memory, return_type=PydanticUndefined, when_used=json)] | None = None, nodes: int | None = None, ntasks_per_node: int | None = None, cpus_per_task: int | None = None, time: Annotated[timedelta, WrapValidator(func=validate_time, json_schema_input_type=PydanticUndefined), WrapSerializer(func=serialize_time, return_type=PydanticUndefined, when_used=json)] | None = None, mail_user: str | None = None, mail_type: list[str] | None = None)[source]

Bases: BaseModel

The inputs for the scheduler.

Attributes are named for convenience/intuitiveness. Aliases are provided for consistency with SLURM. Special validation for parsing values as provided to SLURM options is performed by default.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

account: str | None
check_inputs() → list[str][source]

Check that the scheduler request is valid.

cpus_per_task: int | None
static extract_scheduler_inputs(stream: TextIO | list[str]) → dict[str, str][source]

Parse a slurm submission file to determine the slurm options.

Parameters:

stream – A TextIO containing the contents of the slurm submission file.

Returns:

A dictionary mapping slurm parameter names to parameter values.

Note that no validation/conversion is done to the field values. Conversion to valid (more useful) Python values can be performed using SchedulerInputs.model_validate.

classmethod from_directory(src: str | Path) → SchedulerInputs[source]

Load scheduler inputs from Slurm file.

Parameters:

src – The directory to load inputs from.

Returns:

A SchedulerInputs dictionary.

job_name: str | None
mail_type: list[str] | None
mail_user: str | None
mem: Memory | None
mem_per_cpu: Memory | None
model_config = {'alias_generator': <function hyphenate>, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

nodes: int | None
ntasks_per_node: int | None
partitions: list[Partition] | None
serialize_mail_type(v: Any, _: SerializerFunctionWrapHandler, info: FieldSerializationInfo) → list[str] | str | None[source]

Serialize the mail_type option value.

serialize_partitions(v: list[Partition], _: SerializerFunctionWrapHandler, info: FieldSerializationInfo) → list[Partition] | str | None[source]

Serialize the partitions in the scheduler request.

time: Time | None
static update_values(inputs: dict[str, Any], mods: dict[str, Any]) → None[source]

Safely updates scheduler parameters subject to SLURM conditions.

This method modifies inputs in place.

Parameters:

  • inputs – A dictionary containing SchedulerInputs fields.

  • mods – A dictionary containing new values with which to update inputs. Note that only one of mem and mem_per_cpu can be set.

classmethod validate_mail_type(v: Any, handler: ValidatorFunctionWrapHandler, _: ValidationInfo) → list[str] | None[source]

Validate the mail_type option value.

classmethod validate_partitions(v: Any, handler: ValidatorFunctionWrapHandler, _: ValidationInfo) → float | None[source]

Validate the partitions in the scheduler request.

class autojob.hpc.SchedulerOutputs(*, elapsed: Annotated[timedelta, WrapValidator(func=validate_time, json_schema_input_type=PydanticUndefined), WrapSerializer(func=serialize_time, return_type=PydanticUndefined, when_used=json)] | None = None, idle_time: Annotated[timedelta, WrapValidator(func=validate_time, json_schema_input_type=PydanticUndefined), WrapSerializer(func=serialize_time, return_type=PydanticUndefined, when_used=json)] | None = None, job_id: int | None = None, max_rss: Annotated[int, WrapValidator(func=validate_memory, json_schema_input_type=PydanticUndefined), WrapSerializer(func=serialize_memory, return_type=PydanticUndefined, when_used=json)] | None = None, nodes: list[str] = [], partition: str | None = None, state: JobState = JobState.UNKNOWN, **extra_data: Any)[source]

Bases: BaseModel

The outputs from a scheduler task.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

elapsed: Time | None
classmethod from_directory(src: str | Path, use_cache: bool = False) → SchedulerOutputs[source]

Load scheduler outputs from Slurm file.

Parameters:

  • src – The directory to load outputs from.

  • use_cache – Whether or not to use cached results. If True, then cached results will be ignored and overwritten. Otherwise, outputs will be read from an existing cache.

Returns:

A SchedulerOutputs dictionary.

idle_time: Time | None
job_id: int | None
max_rss: Memory | None
model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

nodes: list[str]
partition: str | None
state: JobState
classmethod validate_job_state(v: Any, handler: ValidatorFunctionWrapHandler, _: ValidationInfo) → JobState[source]

Validate the job state from the value or name of a JobState.

autojob.hpc.get_scheduler_command(scheduler: str) → str[source]

Return the command used to submit jobs to the scheduler.

Parameters:

scheduler – The name of the scheduler.

Raises:
autojob.hpc.serialize_memory(v: Any, _: FieldSerializationInfo) → str | None[source]

Serialize the memory request into a SLURM-compatible format.

autojob.hpc.serialize_time(v: Any, _: FieldSerializationInfo) → str | None[source]

Serialize the time request into a SLURM-compatible format.

autojob.hpc.validate_memory(v: Any, handler: ValidatorFunctionWrapHandler, _: ValidationInfo) → int | None[source]

Validate the memory request.

autojob.hpc.validate_time(v: Any, handler: ValidatorFunctionWrapHandler, _: ValidationInfo) → timedelta | None[source]

Validate the time request.

autojob.legacy module

Legacy classes.

Deprecated since version v.0.12.0b2.

class autojob.legacy.StudyType(value)[source]

Bases: Enum

A tyype of study.

ADSORPTION = 'adsorption'
DEFAULT = 'sensitivity'
MECHANISM = 'mechanism'
SENSITIVITY = 'sensitivity'
is_implemented() → bool[source]

Whether or not a StudyType is implemented.

autojob.parametrizations module

Represent a reference to a variable.

class autojob.parametrizations.VariableReference(*, set_path: Annotated[list[str], FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])], get_path: Annotated[list[str], FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])] | None = None, get_paths: Annotated[list[Annotated[list[str], FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])]], FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])] | None = None, constant: Any = None, composer: Callable | None = None)[source]

Bases: Generic[_T]

A reference to a variable.

Variables:

  • set_path – A list of strings indicating the path to the variable to be set.

  • get_path – A list of strings indicating the path to the variable to be obtained.

  • get_paths – A list of lists of strings each indicating a path to a variable to be obtained.

  • constant – A value to be used to set the variable.

  • composer – A function that takes in an AttributePath and AttributePaths and returns a value.

Example: Evaluate the value of a VariableReference
>>> from autojob.parametrizations import VariableReference
>>> context = {
...     "a": {
...         "b": 4,
...     }
... }
>>> ref = VariableReference(
...     set_path=["a"],
...     get_path=["a", "b"],
...     constant=4,
... )
>>> ref.evaluate(context)
4
Example: Set a value using a VariableReference
>>> from autojob.parametrizations import VariableReference
>>> context = {
...     "a": {
...         "b": 4,
...     }
... }
>>> ref = VariableReference(
...     set_path=["a"],
...     get_path=["a", "b"],
...     constant=4,
... )
>>> class Object(object):
...     pass
>>> target_object = Object()
>>> target_dict = {}
>>> ref.set_input_value(context, target_object)
>>> target_object.a
4
>>> ref.set_input_value(context, target_dict)
>>> target_dict["a"]
4

Instantiate a VariableReference.

Parameters:

  • set_path – An AttributePath indicating the variable to set.

  • get_path – An AttributePath indicating the source variable. Defaults to None.

  • get_paths – A list of AttributePath s, each of which will be combined to the source variable. Defaults to None.

  • constant – A constant value used to set the variable. Defaults to None.

  • composer – A function that accepts the value of the source variable(s) and returns a value to be used to set the variable. Defaults to None.

evaluate(context: _Referenceable) → _T[source]

Evaluate a variable reference in the given context.

Parameters:

context – A dictionary (or object) containing values to be used to evaluate the VariableReference.

Raises:
Returns:

The value.

set_input_value(context: dict[str, Any], shell: _Referenceable) → None[source]

Set the value of a key specified by the VariableReference.

This method modifies shell in place.

Parameters:

  • context – A dictionary containing values to be used to evaluate the VariableReference.

  • shell – A dictionary or object containing values to be set.

autojob.parametrizations.create_parametrization(previous: TaskBase, calc_mods: dict[str, Any] | None = None, sched_mods: dict[str, Any] | None = None, exclude_metadata: Iterable[str] | None = None) → list[VariableReference[Any]][source]

Create a parametrization from parameter mod and a previous task.

Parameters:

  • previous – A calculation.Calculation representing the previous calculation.

  • calc_mods – A dictionary containing modifications to calculator parameters. Defaults to an empty dictionary.

  • sched_mods – A dictionary containing modifications to scheduler inputs. Defaults to an empty dictionary.

  • exclude_metadata – A list of metadata fields to exclude from the parametrization.

Returns:

A list of VariableReference s that can be used to set the values of the new calculation.

Warning

When specifying sched_mods, be wary of setting mutually exclusive scheduler parameters (e.g, mem and mem_per_cpu or cores and cores_per_node). For example, if the mem parameter is set and one wants to set the mem_per_cpu parameter, set the mem key to Unset in sched_mods in addition to setting the mem_per_cpu key.

Note

When setting the input parameters for the restart task, this function will assume that any top-level task attribute suffixed with _inputs is an input. For example, when restarting a Calculation, attr:.Calculation.task_inputs, attr:.Calculation.calculation_inputs, and attr:.Calculation.scheduler_inputs will be carried over. When restarting a MolecularDynamics task, attr:.Calculation.md_inputs will be carried over in addition to the three aforementioned inputs.

autojob.parametrizations.getattrpath(obj: _Referenceable, path: Iterable[str]) → Any[source]

Access an attribute or dictionary value with an attribute path.

Parameters:

  • obj – A dictionary or object.

  • path – A iterable of strings indicating the sequence of attributes or dictionary keys pointing to the value to get.

Returns:

The attribute or dictionary value.

autojob.plugins module

This module loads and stores all registered plugins.

In particular, this module loads task, analysis, and harvest plugins.

autojob.plugins.discover_analysis_plugins() → None[source]

Discover analyses registered through entry points.

autojob.plugins.discover_harvester_plugins() → None[source]

Discover harvesters registered through entry points.

autojob.plugins.discover_task_plugins() → None[source]

Discover tasks registered through entry points.

autojob.plugins.get_analysis(analysis: str) → AnalysisBase[source]

Return an installed analysis.

Raises:

KeyError – The requested analysis is not installed.

autojob.plugins.get_harvester(harvester: str) → HarvesterBase[source]

Return an installed harvester.

Raises:

KeyError – The requested harvester is not installed.

autojob.plugins.get_task_class(task_class: str) → type[TaskBase][source]

Return an installed task class.

Raises:

KeyError – The requested task class is not installed.

autojob.plugins.register_plugin(name: str, plugin: Any, plugin_type: Literal['analysis', 'harvester', 'task_class']) → None[source]

Register a plugin.

Parameters:

  • name – The name used to store the plugin.

  • plugin – The plugin.

  • plugin_type – The type of plugin to be registered.

Raises:

autojob.settings module

Global settings and configuration for autojob.

class autojob.settings.AutojobSettings(_case_sensitive: bool | None = None, _nested_model_default_partial_update: bool | None = None, _env_prefix: str | None = None, _env_prefix_target: EnvPrefixTarget | None = None, _env_file: DotenvType | None = PosixPath('.'), _env_file_encoding: str | None = None, _env_ignore_empty: bool | None = None, _env_nested_delimiter: str | None = None, _env_nested_max_split: int | None = None, _env_parse_none_str: str | None = None, _env_parse_enums: bool | None = None, _cli_prog_name: str | None = None, _cli_parse_args: bool | list[str] | tuple[str, ...] | None = None, _cli_settings_source: CliSettingsSource[Any] | None = None, _cli_parse_none_str: str | None = None, _cli_hide_none_type: bool | None = None, _cli_avoid_json: bool | None = None, _cli_enforce_required: bool | None = None, _cli_use_class_docs_for_groups: bool | None = None, _cli_exit_on_error: bool | None = None, _cli_prefix: str | None = None, _cli_flag_prefix_char: str | None = None, _cli_implicit_flags: bool | Literal['dual', 'toggle'] | None = None, _cli_ignore_unknown_args: bool | None = None, _cli_kebab_case: bool | Literal['all', 'no_enums'] | None = None, _cli_shortcuts: Mapping[str, str | list[str]] | None = None, _secrets_dir: PathType | None = None, _build_sources: tuple[tuple[PydanticBaseSettingsSource, ...], dict[str, Any]] | None = None, *, INPUT_ATOMS_FILE: str = 'in.traj', INPUTS_FILE: str = 'inputs.json', DEFAULT_TASK_SCRIPT_FILE: str = 'run.sh', DEFAULT_CALCULATION_SCRIPT_FILE: str = 'run.py', OUTPUT_ATOMS_FILE: str = 'final.traj', SCHEDULER_STATS_FILE: str = 'job_stats.json', ARCHIVE_FILE: str = 'archive.json', TASK_METADATA_FILE: str = 'task.json', TASK_GROUP_METADATA_FILE: str = 'task_group.json', STUDY_METADATA_FILE: str = 'study.json', STUDY_GROUP_METADATA_FILE: str = 'study_group.json', RECORD_FILE: str = 'record.txt', WORKFLOW_FILE: str = 'workflow.json', PARAMETRIZATION_FILE: str = 'parametrizations.json', TEMPLATE_DIR: Path | None = None, TASK_SCRIPT_TEMPLATE: str = 'run.sh.j2', CALCULATION_SCRIPT_TEMPLATE: str = 'run.py.j2', SCHEDULER: str = 'slurm', LOG_FILE: Path | None = None, LOG_LEVEL: int = 10, STRICT_MODE: bool = True, LEGACY_MODE: bool = True, DEFAULT_TASK: str = 'task', VASP_KEEP_DOS: bool = False)[source]

Bases: BaseSettings

Settings for autojob.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

ARCHIVE_FILE: str
CALCULATION_SCRIPT_TEMPLATE: str
DEFAULT_CALCULATION_SCRIPT_FILE: str
DEFAULT_TASK: str
DEFAULT_TASK_SCRIPT_FILE: str
INPUTS_FILE: str
INPUT_ATOMS_FILE: str
LEGACY_MODE: bool
LOG_FILE: Path | None
LOG_LEVEL: int
OUTPUT_ATOMS_FILE: str
PARAMETRIZATION_FILE: str
RECORD_FILE: str
SCHEDULER: str
SCHEDULER_STATS_FILE: str
STRICT_MODE: bool
STUDY_GROUP_METADATA_FILE: str
STUDY_METADATA_FILE: str
TASK_GROUP_METADATA_FILE: str
TASK_METADATA_FILE: str
TASK_SCRIPT_TEMPLATE: str
TEMPLATE_DIR: Path | None
VASP_KEEP_DOS: bool
WORKFLOW_FILE: str
configure_logging() → AutojobSettings[source]

Configure logging based on user settings.

model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': True, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_kebab_case': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_shortcuts': None, 'cli_use_class_docs_for_groups': False, 'enable_decoding': True, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': True, 'env_nested_delimiter': None, 'env_nested_max_split': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': 'AUTOJOB_', 'env_prefix_target': 'variable', 'extra': 'forbid', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'pyproject_toml_table_header': ('tool', 'autojob'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_config_section': None, 'yaml_file': None, 'yaml_file_encoding': None}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod settings_customise_sources(settings_cls: type[BaseSettings], init_settings: PydanticBaseSettingsSource, env_settings: PydanticBaseSettingsSource, dotenv_settings: PydanticBaseSettingsSource, file_secret_settings: PydanticBaseSettingsSource) → tuple[PydanticBaseSettingsSource, ...][source]

Add a TOML configuration file to the settings sources.

Note that the name of the TOML file can be set via environment variables. Otherwise, the default filename is used.

classmethod validate_log_level(v: Any) → int[source]

Validate the global log level.

autojob.study module

Create studies.

class autojob.study.Study(*, tasks: list[TaskBase] = [], date_created: datetime = <factory>, study_id: UUID, ~pydantic.types.UuidVersion(uuid_version=4)] | str, _PydanticGeneralMetadata(union_mode='left_to_right')] = <factory>, study_group_id: UUID, ~pydantic.types.UuidVersion(uuid_version=4)] | str, _PydanticGeneralMetadata(union_mode='left_to_right')] = <factory>, name: str = '', notes: list[str] = [])[source]

Bases: BaseModel

A collection of tasks.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

date_created: datetime
classmethod from_directory(dir_name: Path, *, strict_mode: bool | None = None) → Study[source]

Recreate a study from a directory.

Parameters:

  • dir_name – The directory of a completed Task.

  • strict_mode – Whether or not to require all outputs. If True, errors will be thrown on missing outputs. Defaults to SETTINGS.STRICT_MODE.

Returns:

The Study contained in dir_name.

model_config: ClassVar = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str
notes: list[str]
serialize_date_created(v: datetime) → str[source]

Serialize the study creation date.

serialize_tasks(v: Any, _: SerializerFunctionWrapHandler, info: FieldSerializationInfo) → list[Task] | list[str][source]

Serialize the tasks in the study.

study_group_id: UUID4 | str
study_id: UUID4 | str
tasks: list[TaskBase]
to_directory(dest: Path, *, study_template: str | None = None, task_group_template: str | None = None, task_template: str | None = None) → Path[source]

Dump a study and its tasks to a directory.

Parameters:

  • dest – The directory in which to dump the Study.

  • study_template – A template string for naming study directories. Defaults to None in which case the study ID will be used to create the directory.

  • task_group_template – A template string for naming task group directories. Defaults to None in which case the task group ID will be used to create the directory.

  • task_template – A template string for naming task directories. Defaults to None in which case the task ID will be used to create the directory.

Returns:

The study group directory that was created.

autojob.study_group module

Create study groups.

class autojob.study_group.StudyGroup(*, date_created: datetime = <factory>, study_group_id: UUID, ~pydantic.types.UuidVersion(uuid_version=4)] | str, _PydanticGeneralMetadata(union_mode='left_to_right')] = <factory>, studies: list[Study] = [], name: str = '', notes: list[str] = [])[source]

Bases: BaseModel

A collection of studies.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

date_created: datetime
classmethod from_directory(dir_name: Path, *, strict_mode: bool | None = None) → Self[source]

Create a study group from a directory.

Parameters:

  • dir_name – The directory of a study group.

  • strict_mode – Whether or not to require all outputs. If True, errors will be thrown on missing outputs. Defaults to SETTINGS.STRICT_MODE.

model_config: ClassVar = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str
notes: list[str]
serialize_studies(v: Any, _: SerializerFunctionWrapHandler, info: FieldSerializationInfo) → list[Study] | list[str][source]

Serialize the studies in the study group.

studies: list[Study]
study_group_id: Annotated[UUID, UuidVersion(uuid_version=4)] | str
to_directory(src: Path, *, study_group_template: str | None = None, study_template: str | None = None, task_group_template: str | None = None, task_template: str | None = None) → Path[source]

Create a directory for a study group.

Parameters:

  • src – The directory in which to dump the StudyGroup.

  • study_group_template – A template string for naming study group directories. Defaults to None in which case the study group ID will be used to create the directory.

  • study_template – A template string for naming study directories. Defaults to None in which case the study ID will be used to create the directory.

  • task_group_template – A template string for naming task group directories. Defaults to None in which case the task group ID will be used to create the directory.

  • task_template – A template string for naming task directories. Defaults to None in which case the task ID will be used to create the directory.

Returns:

The study group directory that was created.

autojob.workflow module

Create workflows from multiple step.

class autojob.workflow.Step(*, workflow_step_id: ~uuid.UUID | None = None, task_class: str = <factory>, progression: ~typing.Literal['independent', 'dependent'] = 'independent', parametrizations: ~typing.Annotated[list[list[~autojob.parametrizations.VariableReference[~typing.Any]]], ~annotated_types.MinLen(min_length=1)])[source]

Bases: BaseModel

A step in a workflow.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parametrizations: list[list[VariableReference[Any]]]
progression: Literal['independent', 'dependent']
task_class: str
workflow_step_id: UUID | None
class autojob.workflow.Workflow(graph: dict[str, Iterable[str]])[source]

Bases: TopologicalSorter

The structure of a workflow.

Initialize a Workflow.

Parameters:

graph – A directed-acyclic graph representing the workflow.

classmethod from_directory(dir_name: Path) → Workflow[source]

Construct a Workflow from a directory.

get_next_steps(step_id: str, record: list[str] | None = None) → list[str][source]

Get all successors of a step.

Parameters:

  • step_id – A string representation of a workflow step ID

  • record – A list of strings where each string represents the workflow step ID of a completed task. If None, all successive steps will be returned. Defaults to None.

Returns:

A list of strings where each string represents a workflow step ID.

get_predecessors(step_id: str) → list[str][source]

Return the immediate ancestors of a workflow step.

Parameters:

step_id – A string representation of a workflow step ID.