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.Cluster(name: str, affiliation: str = '', alliance: str | None = None, partitions: Iterable[Partition] | None = None)[source]

Bases: NamedTuple

A computing cluster.

Create new instance of Cluster(name, affiliation, alliance, partitions)

affiliation: str

Alias for field number 1

alliance: str | None

Alias for field number 2

name: str

Alias for field number 0

partitions: Iterable[Partition] | None

Alias for field number 3

class autojob.hpc.Partition(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[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.SchedulerError(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

A scheduler error.

MEMORY_LIMIT = 'memory limit'
TIME_LIMIT = 'time limit'
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, cores_per_node: 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.

cores_per_node: 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(dir_name: str | Path) → SchedulerInputs[source]

Load scheduler inputs from Slurm file.

Parameters:

dir_name – 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_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'alias_generator': <function hyphenate>, 'populate_by_name': True}

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

model_fields: ClassVar[Dict[str, FieldInfo]] = {'account': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, alias='account', alias_priority=1), 'cores_per_node': FieldInfo(annotation=Union[int, NoneType], required=False, default=None, alias='ntasks-per-node', alias_priority=2, description='The number of cores per node'), 'job_name': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, alias='job-name', alias_priority=1), 'mail_type': FieldInfo(annotation=Union[list[str], NoneType], required=False, default=None, alias='mail-type', alias_priority=1, description='The list of conditions for which email notifications will be sent'), 'mail_user': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, alias='mail-user', alias_priority=1), 'mem': FieldInfo(annotation=Union[Annotated[int, WrapValidator, WrapSerializer], NoneType], required=False, default=None, alias='mem', alias_priority=1, description='The memory requested per node in MB.'), 'mem_per_cpu': FieldInfo(annotation=Union[Annotated[int, WrapValidator, WrapSerializer], NoneType], required=False, default=None, alias='mem-per-cpu', alias_priority=1, description='The memory requested per cpu in MB.'), 'nodes': FieldInfo(annotation=Union[int, NoneType], required=False, default=None, alias='nodes', alias_priority=1), 'partitions': FieldInfo(annotation=Union[list[Partition], NoneType], required=False, default=None, alias='partition', alias_priority=2, description='The partitions specified for the job'), 'time': FieldInfo(annotation=Union[Annotated[timedelta, WrapValidator, WrapSerializer], NoneType], required=False, default=None, alias='time', alias_priority=1, description='The time requested for a job.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

nodes: 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, error: SchedulerError | 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, MaxRSS: Annotated[int, WrapValidator(func=validate_memory, json_schema_input_type=PydanticUndefined), WrapSerializer(func=serialize_memory, return_type=PydanticUndefined, when_used=json)] | None = None, Partition: str | None = None, **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
error: SchedulerError | None
classmethod from_directory(dir_name: str | Path) → SchedulerOutputs[source]

Load scheduler outputs from Slurm file.

Parameters:

dir_name – The directory to load outputs from.

Returns:

A SchedulerOutputs dictionary.

idle_time: Time | None
job_id: int | None
max_rss: Memory | None
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

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

model_fields: ClassVar[Dict[str, FieldInfo]] = {'elapsed': FieldInfo(annotation=Union[Annotated[timedelta, WrapValidator, WrapSerializer], NoneType], required=False, default=None, description='The job wall time.'), 'error': FieldInfo(annotation=Union[SchedulerError, NoneType], required=False, default=None), 'idle_time': FieldInfo(annotation=Union[Annotated[timedelta, WrapValidator, WrapSerializer], NoneType], required=False, default=None, description='The time required for the job to start after submission.'), 'job_id': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'max_rss': FieldInfo(annotation=Union[Annotated[int, WrapValidator, WrapSerializer], NoneType], required=False, default=None, alias='MaxRSS', alias_priority=2, description='Maximum memory used by job.'), 'partition': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, alias='Partition', alias_priority=2, description='The partition on which the job ran.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

partition: str | None
time_validator() → SchedulerOutputs[source]

Calculate and set the time-related attributes.

autojob.hpc.convert(memory: float, from_units: str, to_units: str) → float[source]

Convert memory denominations in binary.

Units can be specified using uppercase, lowercase, one-, or two-letter abbreviations. E.g., ‘K’, ‘k’, ‘KB’, ‘kb’ are all interpreted as kilobytes.

Parameters:

  • memory (float) – memory to be converted.

  • from_units (str) – The units from which memory is to be converted.

  • to_units (str) – The units to which memory is to be converted.

Returns:

float – The memory in the desired units.

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, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[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.

Parameters:

  • 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

>>> from autojob.parametrizations import VariableReference
>>> context = {
...     "a": {
...         "b": 4,
...     }
... }
>>> ref = VariableReference(
...     set_path=["a"],
...     get_path=["a", "b"],
...     constant=4,
... )
>>> ref.evaluate(context)
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.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.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_file: DotenvType | None = PosixPath('.'), _env_file_encoding: str | None = None, _env_ignore_empty: bool | None = None, _env_nested_delimiter: str | 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 | None = None, _cli_ignore_unknown_args: bool | None = None, _secrets_dir: PathType | None = None, *, TASK_FILE: str = 'task.json', JOB_FILE: str = 'job.json', CALCULATION_FILE: str = 'calculation.json', RECORD_FILE: str = 'record.txt', STUDY_FILE: str = 'study.json', STUDY_GROUP_FILE: str = 'study_group.json', WORKFLOW_FILE: str = 'workflow.json', PARAMETRIZATION_FILE: str = 'parametrizations.json', DEFAULT_SCRIPT: str = 'run.sh', PYTHON_SCRIPT: str = 'run.py', SLURM_SCRIPT: str = 'vasp.sh', DEFAULT_TEMPLATE: str = 'default.sh.j2', PYTHON_TEMPLATE: str = 'run.py.j2', SLURM_TEMPLATE: str = 'run.sh.j2', JOB_STATS_FILE: str = 'job_stats.txt', LOG_FILE: Path | None = None, LOG_LEVEL: int = 10, INPUT_ATOMS: str = 'in.traj', OUTPUT_ATOMS: str = 'final.traj', TEMPLATE_DIR: Path | None = None, STRICT_MODE: bool = True)[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.

CALCULATION_FILE: str
DEFAULT_SCRIPT: str
DEFAULT_TEMPLATE: str
INPUT_ATOMS: str
JOB_FILE: str
JOB_STATS_FILE: str
LOG_FILE: Path | None
LOG_LEVEL: int
OUTPUT_ATOMS: str
PARAMETRIZATION_FILE: str
PYTHON_SCRIPT: str
PYTHON_TEMPLATE: str
RECORD_FILE: str
SLURM_SCRIPT: str
SLURM_TEMPLATE: str
STRICT_MODE: bool
STUDY_FILE: str
STUDY_GROUP_FILE: str
TASK_FILE: str
TEMPLATE_DIR: Path | None
WORKFLOW_FILE: str
configure_logging() → AutojobSettings[source]

Configure logging based on user settings.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

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_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_use_class_docs_for_groups': False, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': True, 'env_nested_delimiter': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': 'AUTOJOB_', 'extra': 'forbid', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_', 'settings_'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_file': None, 'yaml_file_encoding': None}

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

model_fields: ClassVar[Dict[str, FieldInfo]] = {'CALCULATION_FILE': FieldInfo(annotation=str, required=False, default='calculation.json', description='the default filename to use to store calculation metadata'), 'DEFAULT_SCRIPT': FieldInfo(annotation=str, required=False, default='run.sh', description='the default filename to use for the input script'), 'DEFAULT_TEMPLATE': FieldInfo(annotation=str, required=False, default='default.sh.j2', description='the filename of the default input script template to use'), 'INPUT_ATOMS': FieldInfo(annotation=str, required=False, default='in.traj', description='the default filename to use for the input Atoms file'), 'JOB_FILE': FieldInfo(annotation=str, required=False, default='job.json', description='the default filename to use to store job metadata'), 'JOB_STATS_FILE': FieldInfo(annotation=str, required=False, default='job_stats.txt', description='the default filename to use for the job stats file'), 'LOG_FILE': FieldInfo(annotation=Union[Path, NoneType], required=False, default=None, description='The filename for the log file. Note that this variable is mainly for storing state for application-like use. If you are using autojob as a library, you may be better served configuring handlers.'), 'LOG_LEVEL': FieldInfo(annotation=int, required=False, default=10, description='The default log level.'), 'OUTPUT_ATOMS': FieldInfo(annotation=str, required=False, default='final.traj', description='the default filename to use for the output Atoms file'), 'PARAMETRIZATION_FILE': FieldInfo(annotation=str, required=False, default='parametrizations.json', description='the default filename to use to store study parametrization data'), 'PYTHON_SCRIPT': FieldInfo(annotation=str, required=False, default='run.py', description='the default filename to use for the python script'), 'PYTHON_TEMPLATE': FieldInfo(annotation=str, required=False, default='run.py.j2', description='the filename of the default Python script template to use'), 'RECORD_FILE': FieldInfo(annotation=str, required=False, default='record.txt', description='the default filename to use to store the study record'), 'SLURM_SCRIPT': FieldInfo(annotation=str, required=False, default='vasp.sh', description='the default filename to use for the SLURM script'), 'SLURM_TEMPLATE': FieldInfo(annotation=str, required=False, default='run.sh.j2', description='the filename of the default SLURM script template to use'), 'STRICT_MODE': FieldInfo(annotation=bool, required=False, default=True, description='Sets the default behaviour of data retrieval functions and methods. If True, such functions will raise errors when they fail. Otherwise, failure will pass with a log message only. This may be useful if you are harvesting the results of incomplete tasks.'), 'STUDY_FILE': FieldInfo(annotation=str, required=False, default='study.json', description='the default filename to use to store study metadata'), 'STUDY_GROUP_FILE': FieldInfo(annotation=str, required=False, default='study_group.json', description='the default filename to use to store study group metadata'), 'TASK_FILE': FieldInfo(annotation=str, required=False, default='task.json', description='the default filename to use to store task metadata'), 'TEMPLATE_DIR': FieldInfo(annotation=Union[Path, NoneType], required=False, default=None, description='If not None, specifies the directory to use to load templates'), 'WORKFLOW_FILE': FieldInfo(annotation=str, required=False, default='workflow.json', description='the default filename to use to store study workflow data')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

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[Task] = [], date_created: datetime = None, study_id: str = None, study_group_id: str = None, Name: str = '', Notes: str = '', study_type: StudyType | None = None)[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.datetime
classmethod from_directory(dir_name: Path, *, strict_mode: bool = True, legacy_mode: bool = False) → 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.

  • legacy_mode – Whether or not use the legacy mode directory structure. Defaults to False.

Returns:

The Study contained in dir_name.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar = {'alias_generator': <function space_capitalize>, 'populate_by_name': True}

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

model_fields: ClassVar[Dict[str, FieldInfo]] = {'date_created': FieldInfo(annotation=datetime, required=False, default_factory=<lambda>, alias='Date Created', alias_priority=1), 'name': FieldInfo(annotation=str, required=False, default='', alias='Name', alias_priority=1), 'notes': FieldInfo(annotation=str, required=False, default='', alias='Notes', alias_priority=1), 'study_group_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, alias='Study Group ID', alias_priority=2), 'study_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, alias='Study ID', alias_priority=2), 'study_type': FieldInfo(annotation=Union[StudyType, NoneType], required=False, default=None, alias='Study Type', alias_priority=1), 'tasks': FieldInfo(annotation=list[Task], required=False, default=[], alias='Tasks', alias_priority=2, validation_alias=AliasChoices(choices=['Tasks', 'Calculations']))}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

name: str
notes: 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: str
study_id: str
study_type: legacy.StudyType | None
tasks: list[Task]
to_directory(dir_name: Path, *, legacy_mode: bool = False) → None[source]

Dump a study and its tasks to a directory.

Parameters:

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

  • legacy_mode – Whether or not use the legacy mode directory structure. Defaults to False.

autojob.study_group module

Create study groups.

class autojob.study_group.StudyGroup(*, date_created: datetime = datetime.datetime(2024, 11, 3, 3, 53, 17, 272796, tzinfo=datetime.timezone.utc), study_group_id: str = None, Studies: list[Study] = [], Name: str = '', Notes: 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 = True, legacy_mode: bool = False) → 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.

  • legacy_mode – Whether or not use the legacy mode directory structure. Defaults to False.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar = {'alias_generator': <function space_capitalize>, 'populate_by_name': True}

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

model_fields: ClassVar[Dict[str, FieldInfo]] = {'date_created': FieldInfo(annotation=datetime, required=False, default=datetime.datetime(2024, 11, 3, 3, 53, 17, 272796, tzinfo=datetime.timezone.utc), alias='Date Created', alias_priority=1), 'name': FieldInfo(annotation=str, required=False, default='', alias='Name', alias_priority=1), 'notes': FieldInfo(annotation=str, required=False, default='', alias='Notes', alias_priority=1), 'studies': FieldInfo(annotation=list[Study], required=False, default=[], alias='Studies', alias_priority=1), 'study_group_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>, alias='Study Group ID', alias_priority=2)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

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

Serialize the studies in the study group.

studies: list[Study]
study_group_id: str
to_directory(dir_name: Path, *, legacy_mode: bool = False) → None[source]

Create a directory for a study group.

Parameters:

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

  • legacy_mode – Whether or not use the legacy mode directory structure. Defaults to False.

autojob.task module

Represent and model the results of a task.

class autojob.task.Task(*, task_metadata: TaskMetadata, task_inputs: TaskInputs, task_outputs: TaskOutputs | None = None, **extra_data: Any)[source]

Bases: BaseModel

Represent the result of a 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.

create_new_task_tree(root: Path, create_legacy_dir: bool = False) → Path[source]

Create the directory and parent directories of a new task.

Parameters:

  • root – A Path representing the root directory from which to create the directory of the new task.

  • create_legacy_dir – Whether or not to create the legacy calculation directory. Note that calculation ID must be specified to create legacy directory.

Raises:

TypeError – The calculation ID is None and create_legacy_dir is True.

Returns:

A Path object representing the directory of the newly created task.

static create_shell(context: dict[str, Any] | None = None) → Task[source]

Recursively create a minimal Task, a shell, for writing inputs.

Parameters:

context – A dictionary mapping strings to Task attributes that will be used to populate the shell. Defaults to an empty dictionary.

Returns:

The minimal Task.

classmethod from_directory(dir_name: str | Path, *, strict_mode: bool = True, magic_mode: bool = False) → Self[source]

Generate a Task document from a completed task’s 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.

  • magic_mode – Whether to defer the final object creation. If True, the final object will be an instance of the class specified by the _build_class attribute of the TaskMetadata object created. Otherwise, a Task object will be returned. Defaults to False.

Returns:

A Task object.

Note

The class indicated by _build_class should be a subclass of the class from which this method is called as successive steps may expect attributes and methods of the calling class to be present.

classmethod load_magic(dir_name: str | Path, *, strict_mode: bool = True) → Self[source]

Magically load the contents of a directory as a Task subclass.

Parameters:

  • dir_name – The directory from which to load the task.

  • strict_mode – Whether to raise errors thrown due to missing outputs Defaults to True in which case errors will be thrown.

Raises:

RuntimeError – No build class specified in the task metadata. Only raised if strict_mode is True.

Returns:

The loaded Task.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

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

model_fields: ClassVar[Dict[str, FieldInfo]] = {'task_inputs': FieldInfo(annotation=TaskInputs, required=True, description='Task inputs'), 'task_metadata': FieldInfo(annotation=TaskMetadata, required=True), 'task_outputs': FieldInfo(annotation=Union[TaskOutputs, NoneType], required=False, default=None, description='Task outputs')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

patch_task(*, output_atoms: Atoms, converged: bool, error: SchedulerError | None, files_to_carry_over: list[str], strict_mode: bool = True) → None[source]

Patch Task attributes using Calculation values.

Note that this method modifies the Task in place. The following attributes are patched:

  • Task.task_outputs.atoms: replaced with output_atoms

  • Task.task_inputs.files_to_carryover: replaced with files_to_carry_over

  • Task.task_outputs.outcome: set according to converged and error

Parameters:

  • dir_name – The directory from which to source values.

  • output_atoms – An Atoms object representing the output geometry.

  • converged – Whether the Calculation is converged.

  • error – The hpc.SchedulerError from the calculation.

  • files_to_carry_over – The files to carry over from the previous calculation.

  • strict_mode – Whether to raise an error if no output atoms found. Defaults to True.

prepare_input_atoms() → None[source]

Copy the final magnetic moments to initial magnetic moments.

This function modifies atoms in place. Note that if atoms were obtained from a vasprun.xml via ase.io.read("vasprun.xml"), no magnetic moments will be read. In order to ensure continuity between runs, it is a good idea to retain the WAVECAR between runs.

task_inputs: TaskInputs
task_metadata: TaskMetadata
task_outputs: TaskOutputs | None
to_directory(dst: str | Path, *, legacy_mode: bool = False) → None[source]

Dump the results of a task to a directory.

Parameters:

  • dst – The directory in which to save the task results.

  • legacy_mode – Whether or not to use the legacy mode.

write_inputs(dir_name: str | Path, *, run_script_template: str = 'default.sh.j2') → list[Path][source]

Write the required inputs for a Task to a directory.

If input atoms are present, they will be written to dir_name with the filename found under the "filename" key in the Atoms.info dictionary if a value exists. Otherwise, the value of SETTINGS.INPUT_ATOMS will be used.

Parameters:

  • dir_name – The directory in which to write the inputs.

  • run_script_template – The template file to use. Defaults to SETTINGS.DEFAULT_TEMPLATE.

Returns:

A list of Path objects where each Path represents the filename of an input written to dir_name.

write_script(dst: Path, *, run_script_template: str = 'default.sh.j2', **kwargs) → Path[source]

Write the input script using the template given.

Parameters:

  • dst – The directory in which to write the input script.

  • run_script_template – The template file to use. Defaults to SETTINGS.DEFAULT_TEMPLATE.

  • **kwargs – additional keyword arguments to be used to render the script template.

Returns:

A Path representing the filename of the written input script.

class autojob.task.TaskInputs(*, atoms: Annotated[Atoms, AtomsAnnotation] | None = None, files_to_copy: list[str] = [], files_to_delete: list[str] = [], files_to_carry_over: list[str] = [], auto_restart: bool = True, **extra_data: Any)[source]

Bases: _TaskIODoc

The set of task-level inputs.

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.

auto_restart: bool
static check_auto_restart(stream: TextIO | list[str]) → bool[source]

Determines if auto-restart was enabled during job submission.

Parameters:

stream – A TextIO or list[str] containing the lines of the slurm job submission script.

Returns:

Whether or not auto-restart was enabled during job submission.

static extract_files_to_copy(stream: TextIO | list[str]) → list[str][source]

Parse the slurm submission script for the calculation files to copy.

This method will parse the files to copy when they are listed in legacy format (i.e., without assignment to a variable) or if they are assigned to a variable named AUTOJOB_FILES_TO_COPY.

Parameters:

stream – A TextIO or list of strings containing the text in the slurm submission script.

Returns:

A list of the strings passed as calculation files to copy.

static extract_files_to_delete(stream: TextIO | list[str]) → list[str][source]

Parse the slurm submission script for the deleted calculation files.

This method will parse the files to delete when they are listed in legacy format (i.e., without assignment to a variable) or if they are assigned to a variable named AUTOJOB_FILES_TO_DELETE.

Parameters:

stream – A TextIO or list of strings containing the text in the slurm submission script.

Returns:

A list of the strings passed as calculation files to delete.

files_to_carry_over: list[str]
files_to_copy: list[str]
files_to_delete: list[str]
classmethod from_directory(dir_name: str | Path) → TaskInputs[source]

Generate a TaskInputs document from a completed task’s directory.

Note that this method retrieves task inputs from the first .sh file in the dir_name returned by Path.rglob.

Parameters:

dir_name – The directory of a completed Task.

Returns:

A TaskInputs object.

static get_input_atoms(dir_name: str | Path) → Atoms[source]

Retrieve an Atoms object representing the input structure.

Note that the filename used to identify the structure file is saved to Atoms.info under the "structure" key.

Parameters:

dir_name – the directory containing the completed calculation

Returns:

An Atoms object.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

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

model_fields: ClassVar[Dict[str, FieldInfo]] = {'atoms': FieldInfo(annotation=Union[Annotated[Atoms, AtomsAnnotation], NoneType], required=False, default=None, description='Input or output ase.Atoms'), 'auto_restart': FieldInfo(annotation=bool, required=False, default=True, description='Whether or not to automatically restart this calculation with the same parameters if the task finishes unsuccessfully.'), 'files_to_carry_over': FieldInfo(annotation=list[str], required=False, default=[], description='The files to carry over from the completed task to the new job.'), 'files_to_copy': FieldInfo(annotation=list[str], required=False, default=[], description='The files to copy from the preceding task into the scratch directory of this task.'), 'files_to_delete': FieldInfo(annotation=list[str], required=False, default=[], description='The files to delete from the directory of the task after job completion.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class autojob.task.TaskMetadata(*, Name: str = '', Notes: list[str] = [], uri: str | None = None, study_group_id: Annotated[Annotated[UUID, UuidVersion(uuid_version=4)] | str | None, _PydanticGeneralMetadata(union_mode='left_to_right')] = None, study_id: Annotated[Annotated[UUID, UuidVersion(uuid_version=4)] | str | None, _PydanticGeneralMetadata(union_mode='left_to_right')] = None, workflow_step_id: Annotated[UUID, UuidVersion(uuid_version=4)] | None = None, task_id: Annotated[Annotated[UUID, UuidVersion(uuid_version=4)] | str, _PydanticGeneralMetadata(union_mode='left_to_right')] = UUID('17ff477e-9b92-45cc-8c08-28e4241e545f'), calculation_id: str | None = None, calculation_type: CalculationType | None = None, calculator_type: CalculatorType | None = None, study_type: StudyType | None = None, last_updated: datetime | None = None, **extra_data: Any)[source]

Bases: BaseModel

The metadata for a 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.

add_build_class() → TaskMetadata[source]

Add a build class to a constructed TaskMetadata object.

Note that this is for backwards-compatibility with Tasks created with calculation_type set and will be removed in future releases. _build_class can be set directly during instantiation.

calculation_id: str | None
calculation_type: CalculationType | None
calculator_type: CalculatorType | None
classmethod from_directory(dir_name: str | Path) → TaskMetadata[source]

Create a TaskMetadata document from a task directory.

label: str
last_updated: datetime | None
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'populate_by_name': True}

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

model_dump_legacy() → dict[str, Any][source]

Return a legacy calculation metadata dictionary.

model_fields: ClassVar[Dict[str, FieldInfo]] = {'calculation_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, alias='Calculation ID', alias_priority=2, description='The Calculation uuid (for backwards-compatibility)'), 'calculation_type': FieldInfo(annotation=Union[CalculationType, NoneType], required=False, default=None, alias='Calculation Type', alias_priority=2, description='The Calculation type (for backwards-compatibility)'), 'calculator_type': FieldInfo(annotation=Union[CalculatorType, NoneType], required=False, default=None, alias='Calculator Type', alias_priority=2, description='The Calculator type (for backwards-compatibility)'), 'label': FieldInfo(annotation=str, required=False, default='', alias='Name', alias_priority=2, description='A description of the job'), 'last_updated': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None, description='Timestamp for the most recent calculation for this task document'), 'study_group_id': FieldInfo(annotation=Union[Annotated[UUID, UuidVersion], str, NoneType], required=False, default=None, alias='Study Group ID', alias_priority=2, description='The study group uuid', metadata=[_PydanticGeneralMetadata(union_mode='left_to_right')]), 'study_id': FieldInfo(annotation=Union[Annotated[UUID, UuidVersion], str, NoneType], required=False, default=None, alias='Study ID', alias_priority=2, description='The study uuid', metadata=[_PydanticGeneralMetadata(union_mode='left_to_right')]), 'study_type': FieldInfo(annotation=Union[StudyType, NoneType], required=False, default=None, alias='Study Type', alias_priority=2, description='The study type (for backwards-compatibility)'), 'tags': FieldInfo(annotation=list[str], required=False, default=[], alias='Notes', alias_priority=2, title='tag', description='Metadata tagged to a given job'), 'task_id': FieldInfo(annotation=Union[Annotated[UUID, UuidVersion], str], required=False, default=UUID('17ff477e-9b92-45cc-8c08-28e4241e545f'), alias='Job ID', alias_priority=2, description='The task uuid.', metadata=[_PydanticGeneralMetadata(union_mode='left_to_right')]), 'uri': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='The uri for the directory containing this task'), 'workflow_step_id': FieldInfo(annotation=Union[Annotated[UUID, UuidVersion], NoneType], required=False, default=None, description='The workflow step uuid')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

static serialize_ids(v: Any, info: SerializationInfo) → str | None[source]

Serialize IDs.

static serialize_last_updated(v: Any, info: SerializationInfo) → str | None[source]

Serialize the last updated time.

static serialize_tags(v: Any, info: SerializationInfo) → str[source]

Serialize tags.

static serialize_types(v: Any, info: SerializationInfo) → str | None[source]

Serialize autojob types.

study_group_id: Annotated[UUID, UuidVersion(uuid_version=4)] | str | None
study_id: Annotated[UUID, UuidVersion(uuid_version=4)] | str | None
study_type: StudyType | None
tags: list[str]
task_id: Annotated[UUID, UuidVersion(uuid_version=4)] | str
uri: str | None
classmethod validate_ids(v: Any, handler: ValidatorFunctionWrapHandler, info: ValidationInfo) → str | Annotated[UUID, UuidVersion(uuid_version=4)][source]

Validate an ID.

classmethod validate_tags(v: Any, handler: ValidatorFunctionWrapHandler, info: ValidationInfo) → list[str][source]

Validate job notes/tags.

workflow_step_id: Annotated[UUID, UuidVersion(uuid_version=4)] | None
class autojob.task.TaskOutcome(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: StrEnum

The state of a task.

ERROR = 'error'
FAILED = 'failed'
IDLE = 'idle'
RUNNING = 'running'
SUCCESS = 'successful'
class autojob.task.TaskOutputs(*, atoms: Annotated[Atoms, AtomsAnnotation] | None = None, entry: ComputedEntry | None = None, outcome: TaskOutcome = TaskOutcome.IDLE)[source]

Bases: _TaskIODoc

The set of task-level outputs.

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.

entry: ComputedEntry | None
classmethod from_directory(dir_name: str | Path, *, strict_mode: bool = True) → TaskOutputs[source]

Generate a TaskOutputs document from a completed task’s directory.

Note that the atoms object may not be set if the task is incomplete. In such a case, one may need to use a task-specific get_output_atoms function (i.e., Calculation.get_output_atoms)

Parameters:

  • dir_name – The directory of a completed Task.

  • strict_mode – Whether or not to catch thrown errors. Errors will be thrown if strict_mode=True.

Returns:

A TaskOutputs object.

static get_output_atoms(dir_name: str | Path, *, strict_mode: bool = True) → Atoms | None[source]

Retrieve an Atoms object representing the output structure.

Parameters:

  • dir_name – The directory from which to retrieve the output structure.

  • strict_mode – Whether or not to raise an error if reading the output atoms file fails. Defaults to True.

Returns:

An Atoms object representing the output structure or None if no Atoms object can be retrieved.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

model_fields: ClassVar[Dict[str, FieldInfo]] = {'atoms': FieldInfo(annotation=Union[Annotated[Atoms, AtomsAnnotation], NoneType], required=False, default=None, description='Input or output ase.Atoms'), 'entry': FieldInfo(annotation=Union[ComputedEntry, NoneType], required=False, default=None, description='The ComputedEntry from the task'), 'outcome': FieldInfo(annotation=TaskOutcome, required=False, default=<TaskOutcome.IDLE: 'idle'>, description='The outcome of the task')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

outcome: TaskOutcome

autojob.workflow module

Create workflows from multiple step.

class autojob.workflow.Step(*, workflow_step_id: UUID, task_type: str, progression: Literal['independent', 'dependent'] = 'independent', parametrizations: Annotated[list[list[VariableReference[Any]]], 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_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

model_fields: ClassVar[Dict[str, FieldInfo]] = {'parametrizations': FieldInfo(annotation=list[list[VariableReference[Any]]], required=True, metadata=[MinLen(min_length=1)]), 'progression': FieldInfo(annotation=Literal['independent', 'dependent'], required=False, default='independent', description="How the step is connected to the tasks of the previous step. 'dependent' indicates that the given step cannot start until every task in the previous step has completed. 'independent' indicates the opposite."), 'task_type': FieldInfo(annotation=str, required=True), 'workflow_step_id': FieldInfo(annotation=UUID, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

parametrizations: list[list[VariableReference[Any]]]
progression: Literal['independent', 'dependent']
task_type: str
workflow_step_id: UUID
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.