autojob.bases package¶

This subpackage defines core base classes for autojob.

Submodules¶

autojob.bases.analysis_base module¶

This module defines the AnalysisBase protocol.

class autojob.bases.analysis_base.AnalysisBase(*args, **kwargs)[source]¶

Bases: Protocol

A protocol defining functions that perform a post-calculation analysis.

autojob.bases.harvester_base module¶

This module defines the HarvesterBase protocol.

class autojob.bases.harvester_base.HarvesterBase(*args, **kwargs)[source]¶

Bases: Protocol

A protocol for functions that harvest post-calculation analysis results.

autojob.bases.task_base module¶

Represent and model the results of a task.

class autojob.bases.task_base.InputWriter(*args, **kwargs)[source]¶

Bases: Protocol

A protocol for objects that can write their inputs to a directory.

write_inputs(dest: str | Path, **kwargs) → list[Path][source]¶

Write inputs of the class to the directory for execution.

This method should also perform any modifications necesary to prepare inputs. For example, such a modification may be copying magnetic moments to initial magnetic moments.

Parameters:

dest – The directory to which the inputs will be written.
kwargs – Additional keyword arguments.

Returns:

A list of input files written.

class autojob.bases.task_base.PathLoadable(*args, **kwargs)[source]¶

Bases: Protocol

A protocol for objects that can be loaded from a directory.

classmethod from_directory(src: str | Path, **kwargs) → Self[source]¶

Load an instance of the class from a directory.

Parameters:

src – The directory from which to load the object.
kwargs – Additional keyword arguments.

Returns:

An instance of the class or a subclass.

class autojob.bases.task_base.SetTaskClassMixin[source]¶

Bases: object

A mixin that sets task_class for the TaskMetadataBase of a task.

set_task_class() → Self[source]¶: Sets the task_class for the TaskMetadataBase of a task.

class autojob.bases.task_base.TaskBase(*, task_metadata: TaskMetadataBase, task_inputs: TaskInputsBase, task_outputs: TaskOutputsBase | None, **extra_data: Any)[source]¶

Bases: BaseModel, ABC

An abstract base class for tasks.

Note that by definition, concrete implementations of the class adhere to the PathLoadable and InputWriter protocols.

Concrete implementations of this class must be defined such that they can be instantiated without arguments.

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.

abstractmethod classmethod from_directory(src: str | Path, **kwargs) → Self[source]¶

Load a task from a directory.

Parameters:

src – The directory from which to load the task.
kwargs – Additional keyword arguments:
- strict_mode – Whether or not to fail on any error. Defaults to SETTINGS.STRICT_MODE.
- magic_mode – Whether or not to instantiate subclasses. If True, the task returned must be an instance determined by metadata in the directory. Defaults to False.

Returns:

A task whose type is that of the calling class or one of its subclasses.

model_config = {'extra': 'allow'}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

task_inputs: TaskInputsBase¶

task_metadata: TaskMetadataBase¶

task_outputs: TaskOutputsBase | None¶

abstractmethod write_inputs(dest: str | Path, **kwargs) → list[Path][source]¶

Write a task to the directory for execution.

Concrete implementations must at least write an inputs JSON, a metadata file, and the input Atoms (if non-None).

Parameters:

dest – The directory in which to write inputs.
kwargs – Additional keyword arguments.

class autojob.bases.task_base.TaskInputsBase(*, atoms: Annotated[list[Annotated[Atoms, AtomsAnnotation]] | Annotated[Atoms, AtomsAnnotation] | None, _PydanticGeneralMetadata(union_mode='left_to_right')] = None, files_to_copy: list[str] = [], files_to_delete: list[str] = [], files_to_carry_over: list[str] = [], auto_restart: bool = True, atoms_filename: str = 'in.traj', task_script: str = 'run.sh', task_script_template: str = 'run.sh.j2', **extra_data: Any)[source]¶

Bases: _TaskIODoc

The set of task-level inputs.

Note that TaskInputsBase instances are PathLoadable.

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.

atoms_filename: str¶

auto_restart: bool¶

files_to_carry_over: list[str]¶

files_to_copy: list[str]¶

files_to_delete: list[str]¶

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

task_script: str¶

task_script_template: str¶

class autojob.bases.task_base.TaskMetadataBase(*, label: str = '', tags: list[str] = [], uri: str | None = None, study_group_id: UUID, ~pydantic.types.UuidVersion(uuid_version=4)] | str | None, _PydanticGeneralMetadata(union_mode='left_to_right')] = <factory>, study_id: UUID, ~pydantic.types.UuidVersion(uuid_version=4)] | str | None, _PydanticGeneralMetadata(union_mode='left_to_right')] = <factory>, workflow_step_id: UUID, ~pydantic.types.UuidVersion(uuid_version=4)] | None = None, task_id: UUID, ~pydantic.types.UuidVersion(uuid_version=4)] | str, _PydanticGeneralMetadata(union_mode='left_to_right')] = <factory>, task_group_id: UUID, ~pydantic.types.UuidVersion(uuid_version=4)] | str | None, _PydanticGeneralMetadata(union_mode='left_to_right')] = <factory>, date_created: datetime = <factory>, last_updated: datetime = <factory>, task_class: str | None = None, **extra_data: Any)[source]¶

Bases: BaseModel, ABC

The metadata for a task.

Note that TaskMetadataBase instances are PathLoadable.

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¶

abstractmethod classmethod from_directory(src: str | Path) → Self[source]¶: Load a TaskMetadataBase from a directory.

label: str¶

last_updated: datetime¶

model_config = {'extra': 'allow', '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].

study_group_id: Annotated[UUID, UuidVersion(uuid_version=4)] | str | None¶

study_id: Annotated[UUID, UuidVersion(uuid_version=4)] | str | None¶

tags: list[str]¶

task_class: str | None¶

task_group_id: Annotated[UUID, UuidVersion(uuid_version=4)] | str | None¶

task_id: Annotated[UUID, UuidVersion(uuid_version=4)] | str¶

uri: str | None¶

workflow_step_id: Annotated[UUID, UuidVersion(uuid_version=4)] | None¶

class autojob.bases.task_base.TaskOutcome(value)[source]¶

Bases: StrEnum

The outcome of a task.

The meanings of each outcome are as follows:

SUCCESS: The task completed successfully. FAILED: The task completed with errors. IDLE: The task is yet to run. RUNNING: The task is running. UNKNOWN: The outcome of the task is unknown.

FAILED = 'failed'¶

IDLE = 'idle'¶

RUNNING = 'running'¶

SUCCESS = 'success'¶

UNKNOWN = 'unknown'¶

class autojob.bases.task_base.TaskOutputsBase(*, atoms: Annotated[list[Annotated[Atoms, AtomsAnnotation]] | Annotated[Atoms, AtomsAnnotation] | None, _PydanticGeneralMetadata(union_mode='left_to_right')] = None, entry: ComputedEntry | None = None, outcome: TaskOutcome = TaskOutcome.UNKNOWN)[source]¶

Bases: _TaskIODoc

The set of task-level outputs.

Note that TaskOutputsBase instances are PathLoadable.

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¶

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

outcome: TaskOutcome¶