Chimie Théorique » scripts_chimie4psmn » DockOnSurf

with daemon.DaemonContext(working_directory=os.getcwd(), umask=0o002,

                          files_preserve=[logger.handlers[0].stream.fileno()]):

    logger.info(f'DockOnSurf started on {os.getcwd()}.')

    logger.info(f'To kill DockOnSurf execution type: `$ kill {os.getpid()}`.')

    logger.info(f"Using '{args.input}' as input.")

import os

import logging

def check_finished_calcs(run_type, code):

    @param run_type: The type of calculation to check.

    @param code: The code used for the specified job.

    normally.

    abnormally.

    """

    from glob import glob

    import ase.io

    from modules.utilities import tail, is_binary

            incar = Incar.from_file(inp_file)

            incar["SYSTEM"] = proj_name+"_"+run_type

    for c, conf in enumerate(atms_list):

        subdir = f'{run_type}/conf_{c}/'

        os.mkdir(subdir)

def run_calc(run_type, inp_vars, atms_list):

    @param run_type: Type of calculation. 'isolated', 'screening' or

    'refinement'

        submit_jobs(run_type, 'ccc_msub -r %s %s', inp_vars['subm_script'],

                    stat_cmd, stat_dict, inp_vars['max_jobs'],

                    inp_vars['project_name'])

    elif inp_vars['batch_q_sys'] == 'local':

        pass  # TODO implement local

    elif not inp_vars['batch_q_sys']:

"""Functions to cluster structures.

functions:

get_labels_affty: Clusters data in affinity matrix form by assigning labels to

get_labels_vector: Clusters data in vectorial form by assigning labels to

get_clusters: Groups data-points belonging to the same cluster into arrays of

get_exemplars_affty: Computes the exemplars for every cluster and returns a list

plot_clusters: Plots the clustered data casting a color to every cluster.

clustering: Directs the clustering process by calling the relevant functions.

"""

def get_rmsd(mol_list: list, remove_Hs="c"):

    @param mol_list: list of rdkit mol objects containing the conformers.

    @param remove_Hs: bool or str,

    @return: list of cluster labels. Every data point is assigned a number

    corresponding to the cluster it belongs to.

    """

    return []

    distances between points, RMSD Matrix, etc.) shape: [n_points, n_points].

    @param clusters: tuple of arrays. Every array contains the indices (relative

    to the affinity matrix) of the data points belonging to the same cluster.

    """

    from sklearn.cluster import AffinityPropagation

    clust_affty_mtcs = tuple(affty_mtx[np.ix_(clust, clust)]

                             for clust in clusters)

    exemplars = []

    for i, mtx in enumerate(clust_affty_mtcs):

        pref = -1e6 * np.max(np.abs(mtx))

        af = AffinityPropagation(affinity='precomputed', preference=pref,

import sys

import logging

import warnings

def log_exception(exc_type, exc_value, exc_tb):

    if issubclass(exc_type, KeyboardInterrupt):

        sys.__excepthook__(exc_type, exc_value, exc_tb)

        return

def log_warning(message, *args, **kwargs):

    logger = logging.getLogger('DockOnSurf')

    logger.warning(" ".join(f"{message}".split()))

    logger = logging.getLogger(label)

    logger.setLevel(logging.INFO)

"""Functions to deal with DockOnSurf input files.

str2lst: Converts a string of integers, and groups of them, to a list of lists.

check_expect_val: Checks whether the value of an option has an adequate value.

get_run_type: Gets 'run_type' value and checks that its value is acceptable.

get_code: Gets 'code' value and checks that its value is acceptable.

get_batch_q_sys: Gets 'batch_q_sys' value and checks that its value is

get_relaunch_err: Gets 'relaunch_err' value and checks that its value is

get_max_jobs: Gets 'max_jobs' value and checks that its value is acceptable.

get_special_atoms: Gets 'special_atoms' value and checks that its value is

get_isol_inp_file: Gets 'isol_inp_file' value and checks that its value is

get_num_conformers: Gets 'num_conformers' value and checks that its value is

get_pre_opt: Gets 'pre_opt' value and checks that its value is acceptable.

get_screen_inp_file: Gets 'screen_inp_file' value and checks that its value is

get_sites: Gets 'sites' value and checks that its value is acceptable.

get_pts_per_angle: Gets 'pts_per_angle' value and checks that its value is

get_coll_thrsld: Gets 'coll_thrsld' value and checks that its value is

get_refine_inp_file: Gets 'refine_inp_file' value and checks that its value is

get_energy_cutoff: Gets 'energy_cutoff' value and checks that its value is

"""

import os.path

import logging

dos_inp = ConfigParser(inline_comment_prefixes='#',

                       empty_lines_in_values=False)

new_answers = {'n': False, 'none': False, 'nay': False,

               'y': True, 'sí': True, 'aye': True, 'sure': True}

for answer, val in new_answers.items():

turn_true_answers = [answer for answer in dos_inp.BOOLEAN_STATES

                     if dos_inp.BOOLEAN_STATES[answer]]

no_sect_err = "Section '%s' not found on input file"

no_opt_err = "Option '%s' not found on section '%s'"

num_error = "'%s' value must be a %s"

def str2lst(cmplx_str, func=int):  # TODO: enable deeper level of nested lists

    # TODO Treat all-enclosing parenthesis as a list instead of list of lists.

    Keyword arguments:

    @param cmplx_str: str, string of integers (or floats) and groups of them

    return True

    if code == 'cp2k':

        from pycp2k import CP2K

        if not isinstance(inp_files, str):

                               "also enabled via the 'potcar_dir' keyword. The "

                               "POTCAR specified as input_file will be used "

                               "instead of the auto-generated one.")

        if not isinstance(inp_files, list) and all(isinstance(inp_file, str)

                                                   for inp_file in inp_files):

            err_msg = "'inp_files' should be a list of file names/paths"

def get_relaunch_err():

    relaunch_err_vals = ['geo_not_conv']

    relaunch_err = dos_inp.get('Global', 'relaunch_err',

                               fallback="False")

# Read input parameters

def read_input(in_file):

    from modules.formats import adapt_format

    err_msg = False

    try:

        dos_inp.read(in_file)

            logger.error(no_opt_err % (opt, 'Global'))

            raise NoOptionError(opt, 'Global')

    isolated, screening, refinement = get_run_type()

    inp_vars['isolated'] = isolated

    inp_vars['screening'] = screening

            logger.error(no_sect_err % 'Isolated')

            raise NoSectionError('Isolated')

        # Mandatory options

        iso_mand_opts = ['isol_inp_file', 'molec_file']

        for opt in iso_mand_opts:

            if not dos_inp.has_option('Isolated', opt):

            raise NoSectionError('Screening')

        # Mandatory options:

        # Checks whether the mandatory options are present.

        screen_mand_opts = ['screen_inp_file', 'surf_file', 'sites',

                            'molec_ctrs']

        for opt in screen_mand_opts:

        inp_vars['molec_ctrs'] = get_molec_ctrs()

        # Checks for PBC

        atms = adapt_format('ase', inp_vars['surf_file'],

                            inp_vars['special_atoms'])

        if inp_vars['code'] == 'vasp' and np.linalg.det(atms.cell) == 0.0 \

functions:

confs_to_mol_list: Converts the conformers inside a rdkit.Mol object to a list

adapt_format: Converts the coordinate files into a required library object type.

read_coords_cp2k: Reads the coordinates from a CP2K restart file and returns an

    ase.Atoms object.

read_energy_cp2k: Reads the CP2K out file and returns its final energy.

"""

import logging

    Depending on the library required to use and the file type, it converts the

    coordinate file into a library-workable object.

    @param requirement: str, the library for which the conversion should be

    @param coord_file: str, path to the coordinates file aiming to convert.

    @param spec_atms: List of tuples containing pairs of new/traditional

        chemical symbols.

    @return: an object the required library can work with.

                                        spec_atms)

            # Assign energy

            for fil in os.listdir(conf_path):

                    continue

                if conf_energy is not None:

                    ase_atms.info["energy"] = conf_energy

                    break

            logger.error(err_msg)

            raise NotImplementedError(err_msg)

    return atoms_list

functions:

remove_C_linked_Hs: Removes hydrogens bonded to a carbon atom from a molecule.

gen_confs: Generate a number of conformers in random orientations.

get_moments_of_inertia: Computes moments of inertia of the given conformers.

    new mol object and the energies of its conformers.

run_isolated: directs the execution of functions to achieve the goal

"""

def get_moments_of_inertia(mol: Chem.rdchem.Mol):

    @param mol: rdkit mol object of the relevant molecule.

    @return numpy array 2D: The inner array contains the moments of inertia for

    from modules.refinement import select_stable_confs

    logger.info('Carrying out procedures for the isolated molecule.')

    rd_mol = adapt_format('rdkit', inp_vars['molec_file'],

                          inp_vars['special_atoms'])

    confs = gen_confs(rd_mol, inp_vars['num_conformers'])

    if inp_vars['pre_opt']:

        confs, confs_ener = pre_opt_confs(confs, inp_vars['pre_opt'])

    else:

        confs_ener = pre_opt_confs(confs, max_iters=0)

    conf_list = confs_to_mol_list(confs)

    rmsd_mtx = get_rmsd(conf_list)

    confs_moi = get_moments_of_inertia(confs)

    exemplars = clustering(rmsd_mtx)

    mol_list = confs_to_mol_list(confs, exemplars)

    ase_atms_list = [rdkit_mol_to_ase_atoms(mol) for mol in mol_list]

                  "dockonsurf.inp"

        logger.error(err_msg)

        raise ValueError(err_msg)

    run_calc('isolated', inp_vars, ase_atms_list)

    logger.info("Finished the procedures for the isolated molecule section. ")

    if inp_vars["batch_q_sys"]:

def tail(f, lines=20):

    total_lines_wanted = lines

    block_size = 1024

def check_bak(file_name):

    """Checks if a file already exists and backs it up if so.

    @param file_name: file to be checked if exists

    """

    import os

def _human_key(key):

    import re

    parts = re.split('(\d*\.\d+|\d+)', key)

    return tuple((e.swapcase() if i % 2 == 0 else float(e))

def is_binary(file):

    try:

        with open(file, "r") as fh:

            fh.read(50)