Statistiques
| Branche: | Tag: | Révision :

dockonsurf / modules / calculation.py @ 234eefed

Historique | Voir | Annoter | Télécharger (13,08 ko)

1
import os
2
import logging
3

    
4
logger = logging.getLogger('DockOnSurf')
5

    
6

    
7
def check_finished_calcs(run_type, code):
8
    """Returns two lists of calculations finished normally and abnormally.
9

10
    @param run_type: The type of calculation to check.
11
    @param code: The code used for the specified job.
12
    @return finished_calcs: List of calculations that have finished normally.
13
    @return unfinished_calcs: List of calculations that have finished abnormally
14
    """
15
    from glob import glob
16
    from modules.utilities import tail
17

    
18
    finished_calcs = []
19
    unfinished_calcs = []
20
    for conf in os.listdir(run_type):
21
        if not os.path.isdir(f'{run_type}/{conf}') or 'conf_' not in conf:
22
            continue
23
        if code == 'cp2k':
24
            out_file_list = glob(f"{run_type}/{conf}/*.out")
25
            restart_file_list = glob(f"{run_type}/{conf}/*-1.restart")
26
            if len(out_file_list) == 0 or len(restart_file_list) == 0:
27
                unfinished_calcs.append(conf)  # TODO specify separetely out and
28
                # TODO restart
29
            elif len(out_file_list) > 1 or len(restart_file_list) > 1:
30
                warn_msg = f'There is more than one file matching the {code} ' \
31
                           f'pattern for finished calculation (*.out / ' \
32
                           f'*-1.restart) in {run_type}/{conf}: ' \
33
                           f'{out_file_list, restart_file_list}. ' \
34
                           f'Skipping directory.'
35
                logger.warning(warn_msg)
36
                unfinished_calcs.append(conf)
37
            else:
38
                with open(out_file_list[0], 'rb') as out_fh:
39
                    if "PROGRAM STOPPED IN" not in tail(out_fh):
40
                        unfinished_calcs.append(conf)
41
                    else:
42
                        finished_calcs.append(conf)
43
        elif code == 'vasp':
44
            out_file_list = glob(f"{run_type}/{conf}/OUTCAR")
45
            if len(out_file_list) == 0:
46
                unfinished_calcs.append(conf)
47
            elif len(out_file_list) > 1:
48
                warn_msg = f'There is more than one file matching the {code} ' \
49
                           f'pattern for finished calculation (*.out / ' \
50
                           f'*-1.restart) in {run_type}/{conf}: ' \
51
                           f'{out_file_list}. Skipping directory.'
52
                logger.warning(warn_msg)
53
                unfinished_calcs.append(conf)
54
            else:
55
                with open(f"{run_type}/{conf}/OUTCAR", 'rb') as out_fh:
56
                    if "General timing and accounting" not in tail(out_fh):
57
                        unfinished_calcs.append(conf)
58
                    else:
59
                        finished_calcs.append(conf)
60
        else:
61
            err_msg = f"Check not implemented for '{code}'."
62
            logger.error(err_msg)
63
            raise NotImplementedError(err_msg)
64
    return finished_calcs, unfinished_calcs
65

    
66

    
67
def prep_cp2k(inp_file: str, run_type: str, atms_list: list, proj_name: str):
68
    """Prepares the directories to run calculations with CP2K.
69

70
    @param inp_file: CP2K Input file to run the calculations with.
71
    @param run_type: Type of calculation. 'isolated', 'screening' or
72
        'refinement'
73
    @param atms_list: list of ase.Atoms objects to run the calculation of.
74
    @param proj_name: name of the project
75
    @return: None
76
    """
77
    from shutil import copy
78
    from pycp2k import CP2K
79
    from modules.utilities import check_bak
80
    if not isinstance(inp_file, str):
81
        err_msg = "'inp_file' must be a string with the path of the CP2K " \
82
                  "input file."
83
        logger.error(err_msg)
84
        raise ValueError(err_msg)
85
    cp2k = CP2K()
86
    cp2k.parse(inp_file)
87
    cp2k.CP2K_INPUT.GLOBAL.Project_name = proj_name+"_"+run_type
88
    force_eval = cp2k.CP2K_INPUT.FORCE_EVAL_list[0]
89
    if force_eval.SUBSYS.TOPOLOGY.Coord_file_name is None:
90
        logger.warning("'COORD_FILE_NAME' not specified on CP2K input. Using\n"
91
                       "'coord.xyz'. A new CP2K input file with "
92
                       "the 'COORD_FILE_NAME' variable is created.")
93
        force_eval.SUBSYS.TOPOLOGY.Coord_file_name = 'coord.xyz'
94
        check_bak(inp_file.split('/')[-1])
95
    cp2k.write_input_file(inp_file.split('/')[-1])
96

    
97
    coord_file = force_eval.SUBSYS.TOPOLOGY.Coord_file_name
98

    
99
    # Creating and setting up directories for every configuration.
100
    for i, conf in enumerate(atms_list):
101
        subdir = f'{run_type}/conf_{i}/'
102
        os.mkdir(subdir)
103
        copy(inp_file, subdir)
104
        conf.write(subdir + coord_file)
105

    
106

    
107
def prep_vasp(inp_files, run_type, atms_list, proj_name, cell):
108
    """Prepares the directories to run calculations with VASP.
109

110
    @param inp_files: VASP Input files to run the calculations with.
111
    @param run_type: Type of calculation. 'isolated', 'screening' or
112
        'refinement'
113
    @param atms_list: list of ase.Atoms objects to run the calculation of.
114
    @param proj_name: name of the project.
115
    @param cell: Cell for the Periodic Boundary Conditions.
116
    @return: None
117
    """
118
    from shutil import copy
119
    import os
120

    
121
    import numpy as np
122
    from pymatgen.io.vasp.inputs import Incar
123

    
124
    from modules.utilities import check_bak
125

    
126
    mand_files = ["INCAR", "KPOINTS", "POTCAR"]
127
    # Check that there are many specified files
128
    if not isinstance(inp_files, list) and all(isinstance(inp_file, str)
129
                                               for inp_file in inp_files):
130
        err_msg = "'inp_files' should be a list of file names/paths"
131
        logger.error(err_msg)
132
        ValueError(err_msg)
133
    # Check that all mandatory files are defined
134
    elif any(not any(mand_file in inp_file.split("/")[-1]
135
                     for inp_file in inp_files) for mand_file in mand_files):
136
        err_msg = f"At least one of the mandatory files {mand_files} was " \
137
                  "not specified."
138
        logger.error(err_msg)
139
        raise FileNotFoundError(err_msg)
140
    # Check that the defined files exist
141
    elif any(not os.path.isfile(inp_file) for inp_file in inp_files):
142
        err_msg = f"At least one of the mandatory files {mand_files} was " \
143
                  "not found."
144
        logger.error(err_msg)
145
        raise FileNotFoundError(err_msg)
146
    for i, inp_file in enumerate(inp_files):
147
        file_name = inp_file.split("/")[-1]
148
        if "INCAR" in file_name:
149
            incar = Incar.from_file(inp_file)
150
            incar["SYSTEM"] = proj_name+"_"+run_type
151
            check_bak("INCAR")
152
            incar.write_file("INCAR")
153
            inp_files[i] = "INCAR"
154
    for c, conf in enumerate(atms_list):
155
        subdir = f'{run_type}/conf_{c}/'
156
        os.mkdir(subdir)
157
        for inp_file in inp_files:
158
            file_name = inp_file.split("/")[-1]
159
            if file_name == "INCAR":
160
                copy("INCAR", subdir)
161
            else:
162
                copy(inp_file, subdir)
163
        if cell is not False and np.linalg.det(cell) != 0.0:
164
            conf.pbc = True
165
            conf.cell = cell
166
        elif np.linalg.det(conf.cell) == 0:
167
            err_msg = "Cell is not defined"
168
            logger.error(err_msg)
169
            raise ValueError(err_msg)
170
        conf.write(subdir+"POSCAR", format="vasp")
171

    
172

    
173
def get_jobs_status(job_ids, stat_cmd, stat_dict):
174
    """Returns a list of job status for a list of job ids.
175

176
    @param job_ids: list of all jobs to be checked their status.
177
    @param stat_cmd: Command to check job status.
178
    @param stat_dict: Dictionary with pairs of job status (r, p, f) and the
179
        pattern it matches in the output of the stat_cmd.
180
    @return: list of status for every job.
181
    """
182
    from subprocess import PIPE, Popen
183
    status_list = []
184
    for job in job_ids:
185
        stat_msg = Popen(stat_cmd % job, shell=True,
186
                         stdout=PIPE).communicate()[0].decode('utf-8').strip()
187
        if stat_dict['r'] == stat_msg:
188
            status_list.append('r')
189
        elif stat_dict['p'] == stat_msg:
190
            status_list.append('p')
191
        elif stat_dict['f'] == stat_msg:
192
            status_list.append('f')
193
        else:
194
            logger.warning(f'Unrecognized job {job} status: {stat_msg}')
195
    return status_list
196

    
197

    
198
def submit_jobs(run_type, sub_cmd, sub_script, stat_cmd, stat_dict, max_jobs,
199
                name):
200
    """Submits jobs to a custom queuing system with the provided script
201

202
    @param run_type: Type of calculation. 'isolated', 'screening', 'refinement'
203
    @param sub_cmd: Bash command used to submit jobs.
204
    @param sub_script: script for the job submission.
205
    @param stat_cmd: Bash command to check job status.
206
    @param stat_dict: Dictionary with pairs of job status: r, p, f (ie. running
207
        pending and finished) and the pattern it matches in the output of the
208
        stat_cmd.
209
    @param max_jobs: dict: Contains the maximum number of jobs to be both
210
        running, pending/queued and pending+running. When the relevant maximum
211
        is reached no jobs more are submitted.
212
    @param name: name of the project.
213
    """
214
    from shutil import copy
215
    from time import sleep
216
    from subprocess import PIPE, Popen
217
    subm_jobs = []
218
    init_dir = os.getcwd()
219
    for conf in os.listdir(run_type):
220
        i = conf.split('_')[1]
221
        while get_jobs_status(subm_jobs, stat_cmd, stat_dict).count("r") + \
222
                get_jobs_status(subm_jobs, stat_cmd, stat_dict).count("p") \
223
                >= max_jobs['rp']\
224
                or get_jobs_status(subm_jobs, stat_cmd, stat_dict).count("r") \
225
                >= max_jobs['r'] \
226
                or get_jobs_status(subm_jobs, stat_cmd, stat_dict).count("p") \
227
                >= max_jobs['p']:
228
            sleep(30)
229
        copy(sub_script, f"{run_type}/{conf}")
230
        os.chdir(f"{run_type}/{conf}")
231
        job_name = f'{name[:5]}{run_type[:3].capitalize()}{i}'
232
        sub_order = sub_cmd % (job_name, sub_script)
233
        subm_msg = Popen(sub_order, shell=True, stdout=PIPE).communicate()[0]
234
        job_id = None
235
        for word in subm_msg.decode("utf-8").split():
236
            try:
237
                job_id = int(word.replace('>', '').replace('<', ''))
238
                break
239
            except ValueError:
240
                continue
241
        subm_jobs.append(job_id)
242
        os.chdir(init_dir)
243

    
244
    logger.info('All jobs have been submitted, waiting for them to finish.')
245
    while not all([stat == 'f' for stat in
246
                   get_jobs_status(subm_jobs, stat_cmd, stat_dict)]):
247
        sleep(30)
248
    logger.info('All jobs have finished.')
249

    
250

    
251
def run_calc(run_type, inp_vars, atms_list):
252
    """Directs the calculation run according to the provided arguments.
253

254
    @param run_type: Type of calculation. 'isolated', 'screening' or
255
    'refinement'
256
    @param inp_vars: Calculation parameters from input file.
257
    @param atms_list: List of ase.Atoms objects containing the sets of atoms
258
    aimed to run the calculations of.
259
    """
260
    from modules.utilities import check_bak
261

    
262
    run_types = ['isolated', 'screening', 'refinement']
263
    if not isinstance(run_type, str) or run_type.lower() not in run_types:
264
        run_type_err = f"'run_type' must be one of the following: {run_types}"
265
        logger.error(run_type_err)
266
        raise ValueError(run_type_err)
267

    
268
    if inp_vars['batch_q_sys']:
269
        logger.info(f"Running {run_type} calculation with {inp_vars['code']} on"
270
                    f" {inp_vars['batch_q_sys']}.")
271
    else:
272
        logger.info(f"Doing a dry run of {run_type}.")
273
    check_bak(run_type)
274
    os.mkdir(run_type)
275

    
276
    # Prepare directories and files for relevant code.
277
    input_files = {'isolated': 'isol_inp_file', 'screening': 'screen_inp_file',
278
                   'refinement': 'refine_inp_file', }
279
    if inp_vars['code'] == 'cp2k':
280
        prep_cp2k(inp_vars[input_files[run_type]], run_type, atms_list,
281
                  inp_vars['project_name'])
282
    elif inp_vars['code'] == "vasp":
283
        prep_vasp(inp_vars[input_files[run_type]], run_type, atms_list,
284
                  inp_vars['project_name'], inp_vars['pbc_cell'])
285
    # elif: inp_vars['code'] == 'Other codes here'
286

    
287
    # Submit/run Jobs
288
    if inp_vars['batch_q_sys'] == 'sge':
289
        stat_cmd = "qstat | grep %s | awk '{print $5}'"
290
        stat_dict = {'r': 'r', 'p': 'qw', 'f': ''}
291
        submit_jobs(run_type, 'qsub -N %s %s', inp_vars['subm_script'],
292
                    stat_cmd, stat_dict, inp_vars['max_jobs'],
293
                    inp_vars['project_name'])
294
    elif inp_vars['batch_q_sys'] == 'lsf':
295
        stat_cmd = "bjobs -w | grep %s | awk '{print $3}'"
296
        stat_dict = {'r': 'RUN', 'p': 'PEND', 'f': ''}
297
        submit_jobs(run_type, 'bsub -J %s < %s', inp_vars['subm_script'],
298
                    stat_cmd, stat_dict, inp_vars['max_jobs'],
299
                    inp_vars['project_name'])
300
    elif inp_vars['batch_q_sys'] == 'irene':
301
        stat_cmd = "ccc_mstat | grep %s | awk '{print $10}' | cut -c1"
302
        stat_dict = {'r': 'R', 'p': 'P', 'f': ''}
303
        submit_jobs(run_type, 'ccc_msub -r %s %s', inp_vars['subm_script'],
304
                    stat_cmd, stat_dict, inp_vars['max_jobs'],
305
                    inp_vars['project_name'])
306

    
307
    elif inp_vars['batch_q_sys'] == 'local':
308
        pass  # TODO implement local
309
    elif not inp_vars['batch_q_sys']:
310
        pass
311
    else:
312
        err_msg = "Unknown value for 'batch_q_sys'."
313
        logger.error(err_msg)
314
        raise ValueError(err_msg)