Skip to content

prep

Prepration for runners.

LOGGER module-attribute 

LOGGER = getLogger(__name__)

calc_cleanup 

calc_cleanup(
    atoms: Atoms | None,
    tmpdir: Path | str,
    job_results_dir: Path | str,
) -> None

Perform cleanup operations for a calculation, including gzipping files, copying files back to the original directory, and removing the tmpdir.

Parameters:

  • atoms (Atoms | None) –

    The Atoms object after the calculation. Must have a calculator attached. If None, no modifications to the calculator's directory will be made.

  • tmpdir (Path | str) –

    The path to the tmpdir, where the calculation will be run. It will be deleted after the calculation is complete.

  • job_results_dir (Path | str) –

    The path to the job_results_dir, where the files will ultimately be stored.

Returns:

  • None –
Source code in quacc/runners/prep.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def calc_cleanup(
    atoms: Atoms | None, tmpdir: Path | str, job_results_dir: Path | str
) -> None:
    """
    Perform cleanup operations for a calculation, including gzipping files, copying
    files back to the original directory, and removing the tmpdir.

    Parameters
    ----------
    atoms
        The Atoms object after the calculation. Must have a calculator
        attached. If None, no modifications to the calculator's directory will be made.
    tmpdir
        The path to the tmpdir, where the calculation will be run. It will be
        deleted after the calculation is complete.
    job_results_dir
        The path to the job_results_dir, where the files will ultimately be
        stored.

    Returns
    -------
    None
    """
    job_results_dir, tmpdir = Path(job_results_dir), Path(tmpdir)
    settings = get_settings()

    # Safety check
    if "tmp-" not in tmpdir.name:
        msg = f"{tmpdir} does not appear to be a tmpdir... exiting for safety!"
        raise ValueError(msg)

    # Update the calculator's directory
    if atoms is not None:
        atoms.calc.directory = job_results_dir

    # Gzip files in tmpdir
    if settings.GZIP_FILES:
        gzip_dir(tmpdir)

    # Move files from tmpdir to job_results_dir
    if settings.CREATE_UNIQUE_DIR:
        move(tmpdir, job_results_dir)
    else:
        for file_name in os.listdir(tmpdir):
            move(tmpdir / file_name, job_results_dir / file_name)
        rmtree(tmpdir)
    LOGGER.info(f"Calculation results stored at {job_results_dir}")

    # Remove symlink to tmpdir
    if os.name != "nt" and settings.SCRATCH_DIR:
        symlink_path = settings.RESULTS_DIR / f"symlink-{tmpdir.name}"
        symlink_path.unlink(missing_ok=True)

calc_setup 

calc_setup(
    atoms: Atoms | None,
    copy_files: (
        SourceDirectory
        | dict[SourceDirectory, Filenames]
        | None
    ) = None,
) -> tuple[Path, Path]

Perform staging operations for a calculation, including copying files to the scratch directory, setting the calculator's directory, decompressing files, and creating a symlink to the scratch directory.

Parameters:

  • atoms (Atoms | None) –

    The Atoms object to run the calculation on. Must have a calculator attached. If None, no modifications to the calculator's directory will be made.

  • copy_files (SourceDirectory | dict[SourceDirectory, Filenames] | None, default: None ) –

    Files to copy (and decompress) from source to the runtime directory.

Returns:

  • Path –

    The path to the unique tmpdir, where the calculation will be run. It will be deleted after the calculation is complete. By default, this will be located within QuaccSettings.SCRATCH_DIR, but if that is not set, it will be located within the QuaccSettings.RESULTS_DIR. For conenience, a symlink to this directory will be made in the QuaccSettings.RESULTS_DIR.

  • Path –

    The path to the results_dir, where the files will ultimately be stored. By defualt, this will be the QuaccSettings.RESULTS_DIR, but if QuaccSettings.CREATE_UNIQUE_DIR is set, it will be a unique directory within the QuaccSettings.RESULTS_DIR.

Source code in quacc/runners/prep.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def calc_setup(
    atoms: Atoms | None,
    copy_files: SourceDirectory | dict[SourceDirectory, Filenames] | None = None,
) -> tuple[Path, Path]:
    """
    Perform staging operations for a calculation, including copying files to the scratch
    directory, setting the calculator's directory, decompressing files, and creating a
    symlink to the scratch directory.

    Parameters
    ----------
    atoms
        The Atoms object to run the calculation on. Must have a calculator
        attached. If None, no modifications to the calculator's directory will be made.
    copy_files
        Files to copy (and decompress) from source to the runtime directory.

    Returns
    -------
    Path
        The path to the unique tmpdir, where the calculation will be run. It will be
        deleted after the calculation is complete. By default, this will be
        located within `QuaccSettings.SCRATCH_DIR`, but if that is not set, it will
        be located within the `QuaccSettings.RESULTS_DIR`. For conenience, a symlink
        to this directory will be made in the `QuaccSettings.RESULTS_DIR`.
    Path
        The path to the results_dir, where the files will ultimately be stored.
        By defualt, this will be the `QuaccSettings.RESULTS_DIR`, but if
        `QuaccSettings.CREATE_UNIQUE_DIR` is set, it will be a unique directory
        within the `QuaccSettings.RESULTS_DIR`.
    """
    # Create a tmpdir for the calculation
    settings = get_settings()
    tmpdir_base = settings.SCRATCH_DIR or settings.RESULTS_DIR
    tmpdir = make_unique_dir(base_path=tmpdir_base, prefix="tmp-quacc-")
    LOGGER.info(f"Calculation will run at {tmpdir}")

    # Set the calculator's directory
    if atoms is not None:
        atoms.calc.directory = tmpdir

    # Define the results directory
    job_results_dir = settings.RESULTS_DIR
    if settings.CREATE_UNIQUE_DIR:
        job_results_dir /= f"{tmpdir.name.split('tmp-')[-1]}"

    # Create a symlink to the tmpdir
    if os.name != "nt" and settings.SCRATCH_DIR:
        symlink_path = settings.RESULTS_DIR / f"symlink-{tmpdir.name}"
        symlink_path.symlink_to(tmpdir, target_is_directory=True)

    # Copy files to tmpdir and decompress them if needed
    if copy_files:
        if isinstance(copy_files, str | Path):
            copy_files = {copy_files: "*"}

        for source_directory, filenames in copy_files.items():
            if source_directory is not None:
                copy_decompress_files(source_directory, filenames, tmpdir)

    return tmpdir, job_results_dir

terminate 

terminate(tmpdir: Path | str, exception: Exception) -> None

Terminate a calculation and move files to a failed directory.

Parameters:

  • tmpdir (Path | str) –

    The path to the tmpdir, where the calculation was run.

  • exception (Exception) –

    The exception that caused the calculation to fail.

Returns:

  • None –

Raises:

  • JobFailure –

    The exception that caused the calculation to fail plus additional metadata.

Source code in quacc/runners/prep.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def terminate(tmpdir: Path | str, exception: Exception) -> None:
    """
    Terminate a calculation and move files to a failed directory.

    Parameters
    ----------
    tmpdir
        The path to the tmpdir, where the calculation was run.
    exception
        The exception that caused the calculation to fail.

    Returns
    -------
    None

    Raises
    -------
    JobFailure
        The exception that caused the calculation to fail plus additional
        metadata.
    """
    tmpdir = Path(tmpdir)
    settings = get_settings()
    job_failed_dir = tmpdir.with_name(tmpdir.name.replace("tmp-", "failed-"))
    tmpdir.rename(job_failed_dir)

    msg = f"Calculation failed! Files stored at {job_failed_dir}"
    LOGGER.info(msg)

    if os.name != "nt" and settings.SCRATCH_DIR:
        old_symlink_path = settings.RESULTS_DIR / f"symlink-{tmpdir.name}"
        symlink_path = settings.RESULTS_DIR / f"symlink-{job_failed_dir.name}"
        old_symlink_path.unlink(missing_ok=True)
        symlink_path.symlink_to(job_failed_dir, target_is_directory=True)

    raise JobFailure(job_failed_dir, message=msg, parent_error=exception) from exception