wmaee.codes.lammps

 9def get_models() -> List[str]:
10    """
11    Get the list of interatomic potential models available in the LAMMPS potentials database.
12
13    Returns:
14        List[str]: A list of available potential models.
15    """
16    cfg = Config()
17    models = cfg.get('applications').get('lammps').get('potentials')
18    del models['root']
19    return list(models.keys())

23def get_potentials(model: str, species: Optional[List[str]] = None, return_df: Optional[bool] = True) -> Tuple[str, List[Dict[str, Union[str, List[str]]]]]:
24    """
25    Get LAMMPS potentials for a given model and optionally filtered by species.
26
27    Parameters:
28    - model (str): The name of the LAMMPS potential model.
29    - species (Optional[List[str]]): A list of element symbols to filter the 
30    potentials. If None, all potentials are returned.
31    - return_df (Optional[bool]): Whether to return potentials as pandas.DataFrame
32    or a simple dictionary. Default is True.
33
34    Returns:
35    Tuple[str, List[Dict[str, Union[str, List[str]]]]]: A tuple containing the 
36    potential root path and a list of potential dictionaries.
37        - The first element of the tuple is the potential root path.
38        - The second element is a list of potential dictionaries, where each 
39        dictionary represents a LAMMPS potential and has the following keys:
40            - 'elements' (List[str]): The list of element symbols associated 
41            with the potential.
42            - 'pot_file' (str): The filename of the potential file.
43    """
44    cfg = Config()
45    root = cfg.get('applications').get('lammps').get('potentials').get('root')
46    folder = cfg.get('applications').get('lammps').get('potentials').get(model)
47    
48    with open(join(root, folder, 'potentials.yaml')) as pot_file:
49        potentials = dict(yaml.safe_load(pot_file))
50    pots = []
51    if species != None:
52        for p in potentials['potentials']:
53            if len(set(p['elements']) & set(species)) > 0:
54                pots.append(p)
55    else:
56        # no species specified -> return all potentials
57        pots = potentials['potentials']
58    if return_df:
59        pots = pd.DataFrame(pots)
60    return potentials['pot_root'], pots

 7def run_lammps(script_name: Optional[str] = 'lammps.in',
 8               lmp_in: Optional[str] = None,
 9               command: Optional[str] = None,
10               args: Optional[Dict[str, Union[str, int]]] = None,
11               directory: Optional[str] = None,
12               log: Union[bool, str] = True) -> None:
13    """
14    Run LAMMPS (Large-scale Atomic/Molecular Massively Parallel Simulator) using the specified command,
15    arguments, working directory, and LAMMPS input script.
16
17    Parameters
18    ----------
19    script_name : str, optional
20        The name of the LAMMPS input script. Defaults to 'lammps.in'.
21    lmp_in : str, optional
22        The entire LAMMPS input script as a string. If provided, it will be written
23        into the file <script_name> before executing lammps. If this file exists, it will be
24        overwritten.
25    command : str, optional
26        The LAMMPS command. If not provided, it will be obtained from the
27        configuration file.
28    args : dict, optional
29        Additional arguments to be passed to the LAMMPS command.
30    directory : str, optional
31        The working directory for running LAMMPS.
32    log : bool or str, optional
33        If True, capture the output to the screen. If a string is provided,
34        capture the output to the specified log file. If False, run silently
35        without capturing output.
36
37    Returns
38    -------
39    None
40    """
41    
42    if command == None:
43        # get the lammps command from wmaee.conf.yaml
44        cfg = Config()
45        command = cfg.get('applications').get('lammps').get('runner').get('script')
46        args_template = cfg.get('applications').get('lammps').get('runner').get('args')
47        # replace defaults with whatever was provided
48        if args == None:
49            args = dict()
50        for a in args:
51            if a in args_template:
52                args_template[a] = args[a]
53        for a in args_template:
54            command = command.replace('{{ ' + str(a) + ' }}', str(args_template[a]))
55    command += ' < '+script_name
56    
57    logfile = False
58    if not log:
59        # silent
60        stdout = subprocess.DEVNULL
61        stderr = subprocess.DEVNULL
62    elif isinstance(log, str):
63        # capture to file
64        stdout = subprocess.PIPE
65        stderr = subprocess.STDOUT
66    else:
67        # don't capture -> screen
68        stdout = None
69        stderr = None
70    
71    if directory == None:
72        directory = '.'
73    with working_directory(directory):        
74        # write input script
75        if lmp_in != None:
76            with open(script_name, 'w') as infile:
77                infile.write(lmp_in)
78        
79        # run lammps
80        run = subprocess.run(command, shell=True, stdout = stdout,
81                             stderr = stderr, text = True)
82        if isinstance(log, str):
83            with open(log, 'w') as output:
84                output.write(run.stdout)

 7def parse_logfile(logfile: str, return_df: bool = True) -> Union[pd.DataFrame, List[Dict[str, List[Optional[float]]]]]:
 8    """
 9    Parse a LAMMPS logfile and extract thermo data.
10
11    Parameters
12    ----------
13    logfile : str
14        Path to the LAMMPS logfile.
15    return_df : bool, optional
16        If True, returns a pandas DataFrame; if False, returns a list of dictionaries,
17        by default True.
18
19    Returns
20    -------
21    Union[pd.DataFrame, List[Dict[str, List[Optional[float]]]]]
22        Parsed thermo data either as a pandas DataFrame or a list of dictionaries.
23    """
24    # Strings that indicate the start and stop of thermo data in the logfile
25    start_thermo_strings = ['Memory usage per processor', 'Per MPI rank memory allocation']
26    stop_thermo_strings = ['Loop time', 'ERROR']
27    
28    # List to store parsed thermo data
29    partial_logs = []
30
31    with open(logfile, 'r') as log:
32        contents = log.readlines()
33
34    keyword_flag = False
35    i = 0
36
37    while i < len(contents):
38        line = contents[i]
39
40        if keyword_flag:
41            tmpString = ''
42            # Check whether any of the thermo stop strings are in the present line
43            while not sum([string in line for string in stop_thermo_strings]) >= 1:
44                if "\n" in line:
45                    tmpString += line
46                i += 1
47                if i < len(contents):
48                    line = contents[i]
49                else:
50                    break
51
52            # Parse the thermo data using pandas
53            partialLogContents = pd.read_table(StringIO(tmpString), sep=r'\s+')
54
55            if return_df:
56                partial_logs.append(partialLogContents)
57            else:
58                # Convert DataFrame to a list of dictionaries if return_df is False
59                partial_logs.append(partialLogContents.to_dict(orient='list'))
60
61            keyword_flag = False
62
63        # Check whether the string matches any of the start string identifiers
64        if sum([line.startswith(string) for string in start_thermo_strings]) >= 1:
65            keyword_flag = True
66
67        i += 1
68
69    # Return either a single DataFrame or a list of dictionaries
70    if len(partial_logs) == 1:
71        return partial_logs[0]
72    return partial_logs

10def write_lammps_data(dumpfile: Union[str, IO], atoms: Atoms, specorder: Optional[List[str]] = None, 
11                      force_skew: bool = False, velocities: bool = False, units: str = 'metal', 
12                      atom_style: str = 'atomic', **kwargs):
13    """
14    Write atomic structure data to a LAMMPS data file. Method forwarded from ase.io.lammpsdata.write_lammps_data.
15
16    Parameters
17    ----------
18    dumpfile : Union[str, IO]
19        File to which the output will be written.
20    atoms : Atoms
21        Atoms to be written.
22    specorder : Optional[List[str]], optional
23        Chemical symbols in the order of LAMMPS atom types, by default None
24    force_skew : bool, optional
25        Force to write the cell as a triclinic box (see LAMMPS documentation),
26        by default False
27    units : str, optional
28        LAMMPS units (see LAMMPS documentation), by default 'metal'
29    atom_style : {'atomic', 'charge', 'full'}, optional
30        LAMMPS atom style (see LAMMPS documentation), by default 'atomic'
31    """
32    _write_lammps_data(dumpfile, atoms, specorder=specorder, force_skew=force_skew,
33                       velocities=velocities, units=units, atom_style=atom_style, **kwargs)

36def read_lammps_dump(dump_file: Union[str, IO], index: Any = slice(None), **kwargs) -> Union[Atoms, List[Atoms]]:
37    """
38    Read Atoms object(s) from a LAMMPS dump file. Method forwarded from ase.io.lammpsrun.read_lammps_dump.
39
40    Parameters
41    ----------
42    dump_file : Union[str, IO]
43        Name of the file to read from or a file descriptor.
44
45    index : Any, optional
46        Specifies which configuration to read. 
47        Default is slice(None), which returns all configurations.
48        
49        Examples
50        --------
51        index=0
52            First configuration
53        index=-2
54            Second to last
55        index=slice(None)
56            All configurations
57        index='-3:'
58            Three last configurations
59        index='::2'
60            Even configurations
61        index='1::2'
62            Odd configurations
63
64    Returns
65    -------
66    Union[Atoms, List[Atoms]]
67        An Atoms object or a list of Atoms objects depending on the number of configurations read.
68    """
69    return _read_lammps_dump(dump_file, index=index)