vasp_custodian

Custodian handlers for VASP.

CustodianKwargs

Bases: TypedDict

Type hint for custodian_kwargs in quacc.calculators.vasp.vasp_custodian.run_custodian.

checkpoint instance-attribute

checkpoint: bool

gzipped_output instance-attribute

gzipped_output: bool

max_errors_per_job instance-attribute

max_errors_per_job: int | None

monitor_freq instance-attribute

monitor_freq: int

polling_time_step instance-attribute

polling_time_step: int

skip_over_errors instance-attribute

skip_over_errors: bool

terminate_func instance-attribute

terminate_func: Callable | None

terminate_on_nonzero_returncode instance-attribute

terminate_on_nonzero_returncode: bool

VaspJobKwargs

Bases: TypedDict

Type hint for vasp_job_kwargs in in quacc.calculators.vasp.vasp_custodian.run_custodian.

auto_continue instance-attribute

auto_continue: bool

auto_gamma instance-attribute

auto_gamma: bool

auto_npar instance-attribute

auto_npar: bool

backup instance-attribute

backup: bool

copy_magmom instance-attribute

copy_magmom: bool

final instance-attribute

final: bool

output_file instance-attribute

output_file: str

settings_override instance-attribute

settings_override: dict | None

stderr_file instance-attribute

stderr_file: str

suffix instance-attribute

suffix: str

run_custodian

run_custodian(
    vasp_parallel_cmd: str = _DEFAULT_SETTING,
    vasp_cmd: str = _DEFAULT_SETTING,
    vasp_gamma_cmd: str = _DEFAULT_SETTING,
    vasp_custodian_max_errors: int = _DEFAULT_SETTING,
    vasp_custodian_wall_time: float = _DEFAULT_SETTING,
    vtst_fixes: bool = _DEFAULT_SETTING,
    vasp_custodian_handlers: (
        list[str] | None
    ) = _DEFAULT_SETTING,
    vasp_custodian_validators: (
        list[str] | None
    ) = _DEFAULT_SETTING,
    scratch_dir: str | None = None,
    directory: str | Path | None = None,
    vasp_job_kwargs: VaspJobKwargs | None = None,
    custodian_kwargs: CustodianKwargs | None = None,
) -> list[list[dict]]

Function to run VASP Custodian.

Parameters:

• vasp_parallel_cmd (str, default: _DEFAULT_SETTING ) –

  VASP parallel command, e.g. "srun -N 2 --ntasks-per-node=24". Defaults to the $VASP_PARALLEL_CMD environment variable in settings.

• vasp_cmd (str, default: _DEFAULT_SETTING ) –

  VASP command. Defaults to "vasp_std" in settings.

• vasp_gamma_cmd (str, default: _DEFAULT_SETTING ) –

  VASP gamma command. Defaults to "vasp_gam" in settings.

• vasp_custodian_max_errors (int, default: _DEFAULT_SETTING ) –

  Maximum number of errors to allow before stopping the run. Defaults to 5 in settings.

• vasp_custodian_wall_time (float, default: _DEFAULT_SETTING ) –

  Maximum wall time to allow before creating a STOPCAR. Defaults to infinity in settings.

• vtst_fixes (bool, default: _DEFAULT_SETTING ) –

  Whether to apply VTST input swaps. Defaults to False in settings.

• vasp_custodian_handlers (list[str] | None, default: _DEFAULT_SETTING ) –

  List of handlers to use in Custodian. See settings for list.

• vasp_custodian_validators (list[str] | None, default: _DEFAULT_SETTING ) –

  List of validators to use in Custodian. See settings for list.

• scratch_dir (str | None, default: None ) –

  Scratch directory to use. Defaults to None.

• directory (str | Path | None, default: None ) –

  Directory to run the calculation in. Defaults to None.

• vasp_job_kwargs (VaspJobKwargs | None, default: None ) –

  Keyword arguments to pass to the Custodian VaspJob. Defaults to None.

• custodian_kwargs (CustodianKwargs | None, default: None ) –

  Any remaining keyword arguments to pass to Custodian. Defaults to None.

Returns:

• list[list[dict]] –

  List of errors from each Custodian job.

Source code in quacc/calculators/vasp/vasp_custodian.py

def run_custodian(
    vasp_parallel_cmd: str = _DEFAULT_SETTING,
    vasp_cmd: str = _DEFAULT_SETTING,
    vasp_gamma_cmd: str = _DEFAULT_SETTING,
    vasp_custodian_max_errors: int = _DEFAULT_SETTING,
    vasp_custodian_wall_time: float = _DEFAULT_SETTING,
    vtst_fixes: bool = _DEFAULT_SETTING,
    vasp_custodian_handlers: list[str] | None = _DEFAULT_SETTING,
    vasp_custodian_validators: list[str] | None = _DEFAULT_SETTING,
    scratch_dir: str | None = None,
    directory: str | Path | None = None,
    vasp_job_kwargs: VaspJobKwargs | None = None,
    custodian_kwargs: CustodianKwargs | None = None,
) -> list[list[dict]]:
    """
    Function to run VASP Custodian.

    Parameters
    ----------
    vasp_parallel_cmd
        VASP parallel command, e.g. "srun -N 2 --ntasks-per-node=24". Defaults
        to the $VASP_PARALLEL_CMD environment variable in settings.
    vasp_cmd
        VASP command. Defaults to "vasp_std" in settings.
    vasp_gamma_cmd
        VASP gamma command. Defaults to "vasp_gam" in settings.
    vasp_custodian_max_errors
        Maximum number of errors to allow before stopping the run. Defaults to 5
        in settings.
    vasp_custodian_wall_time
        Maximum wall time to allow before creating a STOPCAR. Defaults to
        infinity in settings.
    vtst_fixes
        Whether to apply VTST input swaps. Defaults to False in settings.
    vasp_custodian_handlers
        List of handlers to use in Custodian. See settings for list.
    vasp_custodian_validators
        List of validators to use in Custodian. See settings for list.
    scratch_dir
        Scratch directory to use. Defaults to None.
    directory
        Directory to run the calculation in. Defaults to None.
    vasp_job_kwargs
        Keyword arguments to pass to the Custodian VaspJob. Defaults to None.
    custodian_kwargs
        Any remaining keyword arguments to pass to Custodian. Defaults to None.

    Returns
    -------
    list[list[dict]]
        List of errors from each Custodian job.
    """
    # Adapted from atomate2.vasp.run.run_vasp

    from quacc import SETTINGS

    # Set defaults
    vasp_parallel_cmd = os.path.expandvars(
        SETTINGS.VASP_PARALLEL_CMD
        if vasp_parallel_cmd == _DEFAULT_SETTING
        else vasp_parallel_cmd
    )
    vasp_cmd = SETTINGS.VASP_CMD if vasp_cmd == _DEFAULT_SETTING else vasp_cmd
    vasp_gamma_cmd = (
        SETTINGS.VASP_GAMMA_CMD
        if vasp_gamma_cmd == _DEFAULT_SETTING
        else vasp_gamma_cmd
    )
    vasp_custodian_max_errors = (
        SETTINGS.VASP_CUSTODIAN_MAX_ERRORS
        if vasp_custodian_max_errors == _DEFAULT_SETTING
        else vasp_custodian_max_errors
    )
    vasp_custodian_wall_time = (
        SETTINGS.VASP_CUSTODIAN_WALL_TIME
        if vasp_custodian_wall_time == _DEFAULT_SETTING
        else vasp_custodian_wall_time
    )
    vtst_fixes = (
        SETTINGS.VASP_CUSTODIAN_VTST if vtst_fixes == _DEFAULT_SETTING else vtst_fixes
    )
    vasp_custodian_handlers = (
        SETTINGS.VASP_CUSTODIAN_HANDLERS
        if vasp_custodian_handlers == _DEFAULT_SETTING
        else vasp_custodian_handlers
    )
    vasp_custodian_validators = (
        SETTINGS.VASP_CUSTODIAN_VALIDATORS
        if vasp_custodian_validators == _DEFAULT_SETTING
        else vasp_custodian_validators
    )

    # Handlers for VASP
    handlers = []
    handlers_dict = {
        "VaspErrorHandler": VaspErrorHandler(vtst_fixes=vtst_fixes),
        "FrozenJobErrorHandler": FrozenJobErrorHandler(),
        "IncorrectSmearingHandler": IncorrectSmearingHandler(),
        "LargeSigmaHandler": LargeSigmaHandler(),
        "MeshSymmetryErrorHandler": MeshSymmetryErrorHandler(),
        "NonConvergingErrorHandler": NonConvergingErrorHandler(),
        "PositiveEnergyErrorHandler": PositiveEnergyErrorHandler(),
        "PotimErrorHandler": PotimErrorHandler(),
        "StdErrHandler": StdErrHandler(),
        "UnconvergedErrorHandler": UnconvergedErrorHandler(),
        "WalltimeHandler": WalltimeHandler(),
        "KspacingMetalHandler": KspacingMetalHandler(),
    }
    validators_dict = {
        "VaspFilesValidator": VaspFilesValidator(),
        "VasprunXMLValidator": VasprunXMLValidator(),
    }

    handlers = []
    for handler_flag in vasp_custodian_handlers:
        if handler_flag not in handlers_dict:
            msg = f"Unknown VASP error handler: {handler_flag}"
            raise ValueError(msg)
        handlers.append(handlers_dict[handler_flag])

    validators = []
    for validator_flag in vasp_custodian_validators:
        if validator_flag not in validators_dict:
            msg = f"Unknown VASP validator: {validator_flag}"
            raise ValueError(msg)
        validators.append(validators_dict[validator_flag])

    # Populate settings
    full_vasp_cmd = f"{vasp_parallel_cmd} {vasp_cmd}"
    full_vasp_gamma_cmd = f"{vasp_parallel_cmd} {vasp_gamma_cmd}"

    # Run VASP
    vasp_job_kwargs = {} if vasp_job_kwargs is None else vasp_job_kwargs
    custodian_kwargs = {} if custodian_kwargs is None else custodian_kwargs
    split_vasp_cmd = shlex.split(full_vasp_cmd)
    split_vasp_gamma_cmd = shlex.split(full_vasp_gamma_cmd)
    vasp_job_kwargs["gamma_vasp_cmd"] = split_vasp_gamma_cmd

    # Run with Custodian
    jobs = [VaspJob(split_vasp_cmd, **vasp_job_kwargs)]

    if vasp_custodian_wall_time:
        handlers = [
            *list(handlers),
            WalltimeHandler(wall_time=vasp_custodian_wall_time),
        ]

    c = Custodian(
        handlers,
        jobs,
        validators=validators,
        max_errors=vasp_custodian_max_errors,
        scratch_dir=scratch_dir,
        directory=directory,
        **custodian_kwargs,
    )

    return c.run()