<commit_before>__all__ = ["calculate_mos", "create_point_folder",
           "split_file_geometries"]

# ================> Python Standard  and third-party <==========
from collections import (defaultdict, namedtuple)
from noodles import (gather, schedule)
from os.path import join

import fnmatch
import h5py
import logging
import os
import shutil

# ==================> Internal modules <==========
from nac.schedule.scheduleCp2k import prepare_job_cp2k
from nac.common import (
    Matrix, is_data_in_hdf5, read_cell_parameters_as_array, store_arrays_in_hdf5)
from qmflows.hdf5 import dump_to_hdf5
from qmflows.utils import chunksOf
from qmflows.warnings_qmflows import SCF_Convergence_Warning

# Tuple contanining file paths
JobFiles = namedtuple("JobFiles", ("get_xyz", "get_inp", "get_out", "get_MO"))

# Starting logger
logger = logging.getLogger(__name__)
# ==============================> Tasks <=====================================


def calculate_mos(config: dict) -> list:
    """
    Look for the MO in the HDF5 file if they do not exists calculate them by
    splitting the jobs in batches given by the ``restart_chunk`` variables.
    Only the first job is calculated from scratch while the rest of the
    batch uses as guess the wave function of the first calculation in
    the batch.

    The config dict contains:

    :param geometries: list of molecular geometries
    :param project_name: Name of the project used as root path for storing
    data in HDF5.
    :param path_hdf5: Path to the HDF5 file that contains the
    numerical results.
    :param folders: path to the directories containing the MO outputs
    :param settings_main: Settings for the job to run.
    :param calc_new_wf_guess_on_points: Calculate a new Wave function guess in
    each of the geometries indicated. By Default only an initial guess is
    computed.
    :param enumerate_from: Number from where to start enumerating the folders
    create for each point in the MD

    :returns: path to nodes in the HDF5 file to MO energies
              and MO coefficients.
    """
    # Read Cell parameters file
    general = config['cp2k_general_settings']
    file_cell_parameters = general["file_cell_parameters"]
    if file_cell_parameters is not None:
        array_cell_parameters = read_cell_parameters_as_array(file_cell_parameters)[1]

    # First calculation has no initial guess
    # calculate the rest of the jobs using the previous point as initial guess
    orbitals = []  # list to the nodes in the HDF5 containing the MOs
    energies = []
    guess_job = None
    for j, gs in enumerate(config.geometries):

        # number of the point with respect to all the trajectory
        k = j + config.enumerate_from

        # dictionary containing the information of the j-th job
        dict_input = defaultdict(lambda: None)
        dict_input["geometry"] = gs
        dict_input["k"] = k

        # Path where the MOs will be store in the HDF5
        root = join(config.project_name, 'point_{}'.format(k), config.package_name, 'mo')
        dict_input["node_MOs"] = [join(root, 'eigenvalues'), join(root, 'coefficients')]
        dict_input["node_energy"] = join(root, 'energy')

        # If the MOs are already store in the HDF5 format return the path
        # to them and skip the calculation
        if is_data_in_hdf5(config.path_hdf5, dict_input["node_MOs"]):
            logger.info("point_{} has already been calculated".format(k))
            orbitals.append(dict_input["node_MOs"])
        else:
            logger.info("point_{} has been scheduled".format(k))

            # Add cell parameters from file if given
            if file_cell_parameters is not None:
                adjust_cell_parameters(general, array_cell_parameters, j)
            # Path to I/O files
            dict_input["point_dir"] = config.folders[j]
            dict_input["job_files"] = create_file_names(dict_input["point_dir"], k)
            dict_input["job_name"] = 'point_{}'.format(k)

            # Compute the MOs and return a new guess
            promise_qm = compute_orbitals(config, dict_input, guess_job)

            # Check if the job finishes succesfully
            promise_qm = schedule_check(promise_qm, config, dict_input)

            # Store the computation
            orbitals.append(store_MOs(config, dict_input, promise_qm))
            energies.append(store_enery(config, dict_input, promise_qm))

            guess_job = promise_qm

    return gather(gather(*orbitals), gather(*energies))


@schedule
def store_MOs(config: dict, dict_input: dict, promise_qm: object) -> str:
    """
    Store the MOs in the HDF5
    """
    # Molecular Orbitals
    mos = promise_qm.orbitals

    # Store in the HDF5
    try:
        with h5py.File(config.path_hdf5, 'r+') as f5:
            dump_to_hdf5(
                mos, 'cp2k', f5,
                project_name=config.project_name, job_name=dict_input["job_name"])
    # Remove the ascii MO file
    finally:
        work_dir = promise_qm.archive['work_dir']
        path_MOs = fnmatch.filter(os.listdir(work_dir), 'mo_*MOLog')[0]
        os.remove(join(work_dir, path_MOs))

    return dict_input["node_MOs"]


@schedule
def store_enery(config: dict, dict_input: dict, promise_qm: object) -> str:
    """
    Store the total energy in the HDF5 file.
    """
    store_arrays_in_hdf5(
        config.path_hdf5, dict_input['node_energy'], promise_qm.energy)

    logger.info("Total energy point of point {} is: {}".format(
        dict_input['k'], promise_qm.energy))

    return dict_input["node_energy"]
<commit_msg>fixed typo<commit_after>__all__ = ["calculate_mos", "create_point_folder",
           "split_file_geometries"]

# ================> Python Standard  and third-party <==========
from collections import (defaultdict, namedtuple)
from noodles import (gather, schedule)
from os.path import join

import fnmatch
import h5py
import logging
import os
import shutil

# ==================> Internal modules <==========
from nac.schedule.scheduleCp2k import prepare_job_cp2k
from nac.common import (
    Matrix, is_data_in_hdf5, read_cell_parameters_as_array, store_arrays_in_hdf5)
from qmflows.hdf5 import dump_to_hdf5
from qmflows.utils import chunksOf
from qmflows.warnings_qmflows import SCF_Convergence_Warning

# Tuple contanining file paths
JobFiles = namedtuple("JobFiles", ("get_xyz", "get_inp", "get_out", "get_MO"))

# Starting logger
logger = logging.getLogger(__name__)
# ==============================> Tasks <=====================================


def calculate_mos(config: dict) -> list:
    """
    Look for the MO in the HDF5 file if they do not exists calculate them by
    splitting the jobs in batches given by the ``restart_chunk`` variables.
    Only the first job is calculated from scratch while the rest of the
    batch uses as guess the wave function of the first calculation in
    the batch.

    The config dict contains:

    :param geometries: list of molecular geometries
    :param project_name: Name of the project used as root path for storing
    data in HDF5.
    :param path_hdf5: Path to the HDF5 file that contains the
    numerical results.
    :param folders: path to the directories containing the MO outputs
    :param settings_main: Settings for the job to run.
    :param calc_new_wf_guess_on_points: Calculate a new Wave function guess in
    each of the geometries indicated. By Default only an initial guess is
    computed.
    :param enumerate_from: Number from where to start enumerating the folders
    create for each point in the MD

    :returns: path to nodes in the HDF5 file to MO energies
              and MO coefficients.
    """
    # Read Cell parameters file
    general = config['cp2k_general_settings']
    file_cell_parameters = general["file_cell_parameters"]
    if file_cell_parameters is not None:
        array_cell_parameters = read_cell_parameters_as_array(file_cell_parameters)[1]

    # First calculation has no initial guess
    # calculate the rest of the jobs using the previous point as initial guess
    orbitals = []  # list to the nodes in the HDF5 containing the MOs
    energies = []
    guess_job = None
    for j, gs in enumerate(config.geometries):

        # number of the point with respect to all the trajectory
        k = j + config.enumerate_from

        # dictionary containing the information of the j-th job
        dict_input = defaultdict(lambda: None)
        dict_input["geometry"] = gs
        dict_input["k"] = k

        # Path where the MOs will be store in the HDF5
        root = join(config.project_name, 'point_{}'.format(k), config.package_name, 'mo')
        dict_input["node_MOs"] = [join(root, 'eigenvalues'), join(root, 'coefficients')]
        dict_input["node_energy"] = join(root, 'energy')

        # If the MOs are already store in the HDF5 format return the path
        # to them and skip the calculation
        if is_data_in_hdf5(config.path_hdf5, dict_input["node_MOs"]):
            logger.info("point_{} has already been calculated".format(k))
            orbitals.append(dict_input["node_MOs"])
        else:
            logger.info("point_{} has been scheduled".format(k))

            # Add cell parameters from file if given
            if file_cell_parameters is not None:
                adjust_cell_parameters(general, array_cell_parameters, j)
            # Path to I/O files
            dict_input["point_dir"] = config.folders[j]
            dict_input["job_files"] = create_file_names(dict_input["point_dir"], k)
            dict_input["job_name"] = 'point_{}'.format(k)

            # Compute the MOs and return a new guess
            promise_qm = compute_orbitals(config, dict_input, guess_job)

            # Check if the job finishes succesfully
            promise_qm = schedule_check(promise_qm, config, dict_input)

            # Store the computation
            orbitals.append(store_MOs(config, dict_input, promise_qm))
            energies.append(store_enery(config, dict_input, promise_qm))

            guess_job = promise_qm

    return gather(gather(*orbitals), gather(*energies))


@schedule
def store_MOs(config: dict, dict_input: dict, promise_qm: object) -> str:
    """
    Store the MOs in the HDF5
    """
    # Molecular Orbitals
    mos = promise_qm.orbitals

    # Store in the HDF5
    try:
        with h5py.File(config.path_hdf5, 'r+') as f5:
            dump_to_hdf5(
                mos, 'cp2k', f5,
                project_name=config.project_name, job_name=dict_input["job_name"])
    # Remove the ascii MO file
    finally:
        work_dir = promise_qm.archive['work_dir']
        path_MOs = fnmatch.filter(os.listdir(work_dir), 'mo_*MOLog')[0]
        os.remove(join(work_dir, path_MOs))

    return dict_input["node_MOs"]


@schedule
def store_enery(config: dict, dict_input: dict, promise_qm: object) -> str:
    """
    Store the total energy in the HDF5 file.
    """
    store_arrays_in_hdf5(
        config.path_hdf5, dict_input['node_energy'], promise_qm.energy)

    logger.info("Total energy of point {} is: {}".format(
        dict_input['k'], promise_qm.energy))

    return dict_input["node_energy"]

{'fraction_non_alphanumeric': 0.10224729161070821, 'fraction_numerical': 0.005551078879040201, 'mean_word_length': 3.229458538432412, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2943370', 'n_tokens_mistral': 3380, 'n_tokens_neox': 3287, 'n_words': 1036}