4075885785

Build Type

push

github

Committed by Shyue Ping Ong

Commit Message

Merge branch 'master' of github.com:materialsproject/pymatgen

Run Details

96 of 96 new or added lines in 27 files covered. (100.0%)

81013 of 102710 relevant lines covered (78.88%)

0.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/pymatgen/cli/pmg_analyze.py

# Copyright (c) Pymatgen Development Team.
# Distributed under the terms of the MIT License.

"""
Implementation for `pmg analyze` CLI.
"""

from __future__ import annotations

import logging
import multiprocessing
import os
import re

from tabulate import tabulate

from pymatgen.apps.borg.hive import (
    SimpleVaspToComputedEntryDrone,
    VaspToComputedEntryDrone,
)
from pymatgen.apps.borg.queen import BorgQueen
from pymatgen.io.vasp import Outcar

__author__ = "Shyue Ping Ong"
__copyright__ = "Copyright 2012, The Materials Project"
__version__ = "4.0"
__maintainer__ = "Shyue Ping Ong"
__email__ = "ongsp@ucsd.edu"
__date__ = "Aug 13 2016"

SAVE_FILE = "vasp_data.gz"


def get_energies(rootdir, reanalyze, verbose, quick, sort, fmt):
    """
    Get energies of all vaspruns in directory (nested).
    Args:
        rootdir (str): Root directory.
        reanalyze (bool): Whether to ignore saved results and reanalyze
        verbose (bool): Verbose mode or not.
        quick (bool): Whether to perform a quick analysis (using OSZICAR instead
            of vasprun.xml
        sort (bool): Whether to sort the results in ascending order.
        fmt (str): tablefmt passed to tabulate.
    """
    if verbose:
        logformat = "%(relativeCreated)d msecs : %(message)s"
        logging.basicConfig(level=logging.INFO, format=logformat)

    if quick:
        drone = SimpleVaspToComputedEntryDrone(inc_structure=True)
    else:
        drone = VaspToComputedEntryDrone(inc_structure=True, data=["filename", "initial_structure"])

    ncpus = multiprocessing.cpu_count()
    logging.info(f"Detected {ncpus} cpus")
    queen = BorgQueen(drone, number_of_drones=ncpus)
    if os.path.exists(SAVE_FILE) and not reanalyze:
        msg = f"Using previously assimilated data from {SAVE_FILE}. Use -r to force re-analysis."
        queen.load_data(SAVE_FILE)
    else:
        if ncpus > 1:
            queen.parallel_assimilate(rootdir)
        else:
            queen.serial_assimilate(rootdir)
        msg = f"Analysis results saved to {SAVE_FILE} for faster subsequent loading."
        queen.save_data(SAVE_FILE)

    entries = queen.get_data()
    if sort == "energy_per_atom":
        entries = sorted(entries, key=lambda x: x.energy_per_atom)
    elif sort == "filename":
        entries = sorted(entries, key=lambda x: x.data["filename"])

    all_data = []
    for e in entries:
        if quick:
            delta_vol = "NA"
        else:
            delta_vol = e.structure.volume / e.data["initial_structure"].volume - 1
            delta_vol = f"{delta_vol * 100:.2f}"
        all_data.append(
            (
                e.data["filename"].replace("./", ""),
                re.sub(r"\s+", "", e.composition.formula),
                f"{e.energy:.5f}",
                f"{e.energy_per_atom:.5f}",
                delta_vol,
            )
        )
    if len(all_data) > 0:
        headers = ("Directory", "Formula", "Energy", "E/Atom", "% vol chg")
        print(tabulate(all_data, headers=headers, tablefmt=fmt))
        print()
        print(msg)
    else:
        print("No valid vasp run found.")
        os.unlink(SAVE_FILE)
    return 0


def get_magnetizations(dir: str, ion_list: list[int]):
    """
    Get magnetization info from OUTCARs.

    Args:
        mydir (str): Directory name
        ion_list (list[int]): List of ions to obtain magnetization information for.

    Returns:
        int: 0 if successful.
    """
    data = []
    max_row = 0
    for parent, _subdirs, files in os.walk(dir):
        for f in files:
            if re.match(r"OUTCAR*", f):
                try:
                    row = []
                    fullpath = os.path.join(parent, f)
                    outcar = Outcar(fullpath)
                    mags = outcar.magnetization
                    mags = [m["tot"] for m in mags]
                    all_ions = list(range(len(mags)))
                    row.append(fullpath.lstrip("./"))
                    if ion_list:
                        all_ions = ion_list
                    for ion in all_ions:
                        row.append(str(mags[ion]))
                    data.append(row)
                    if len(all_ions) > max_row:
                        max_row = len(all_ions)
                except Exception:
                    pass

    for d in data:
        if len(d) < max_row + 1:
            d.extend([""] * (max_row + 1 - len(d)))
    headers = ["Filename"]
    for i in range(max_row):
        headers.append(str(i))
    print(tabulate(data, headers))
    return 0


def analyze(args):
    """
    Master function controlling which analysis to call.

    Args:
        args (dict): args from argparse.
    """
    default_energies = not (args.get_energies or args.ion_list)

    if args.get_energies or default_energies:
        for d in args.directories:
            return get_energies(d, args.reanalyze, args.verbose, args.quick, args.sort, args.format)
    if args.ion_list:
        if args.ion_list[0] == "All":
            ion_list = None
        else:
            (start, end) = (int(i) for i in re.split(r"-", args.ion_list[0]))
            ion_list = list(range(start, end + 1))
        for d in args.directories:
            return get_magnetizations(d, ion_list)

    return -1

1	# Copyright (c) Pymatgen Development Team.
2	# Distributed under the terms of the MIT License.
3
4	"""
5	Implementation for `pmg analyze` CLI.
6	"""
7
8	from __future__ import annotations
9
10	import logging
11	import multiprocessing
12	import os
13	import re
14
15	from tabulate import tabulate
16
17	from pymatgen.apps.borg.hive import (
18	SimpleVaspToComputedEntryDrone,
19	VaspToComputedEntryDrone,
20	)
21	from pymatgen.apps.borg.queen import BorgQueen
22	from pymatgen.io.vasp import Outcar
23
24	__author__ = "Shyue Ping Ong"
25	__copyright__ = "Copyright 2012, The Materials Project"
26	__version__ = "4.0"
27	__maintainer__ = "Shyue Ping Ong"
28	__email__ = "ongsp@ucsd.edu"
29	__date__ = "Aug 13 2016"
30
31	SAVE_FILE = "vasp_data.gz"
32
33
34	def get_energies(rootdir, reanalyze, verbose, quick, sort, fmt):
35	"""
36	Get energies of all vaspruns in directory (nested).
37	Args:
38	rootdir (str): Root directory.
39	reanalyze (bool): Whether to ignore saved results and reanalyze
40	verbose (bool): Verbose mode or not.
41	quick (bool): Whether to perform a quick analysis (using OSZICAR instead
42	of vasprun.xml
43	sort (bool): Whether to sort the results in ascending order.
44	fmt (str): tablefmt passed to tabulate.
45	"""
46	if verbose:
47	logformat = "%(relativeCreated)d msecs : %(message)s"
48	logging.basicConfig(level=logging.INFO, format=logformat)
49
50	if quick:
51	drone = SimpleVaspToComputedEntryDrone(inc_structure=True)
52	else:
53	drone = VaspToComputedEntryDrone(inc_structure=True, data=["filename", "initial_structure"])
54
55	ncpus = multiprocessing.cpu_count()
56	logging.info(f"Detected {ncpus} cpus")
57	queen = BorgQueen(drone, number_of_drones=ncpus)
58	if os.path.exists(SAVE_FILE) and not reanalyze:
59	msg = f"Using previously assimilated data from {SAVE_FILE}. Use -r to force re-analysis."
60	queen.load_data(SAVE_FILE)
61	else:
62	if ncpus > 1:
63	queen.parallel_assimilate(rootdir)
64	else:
65	queen.serial_assimilate(rootdir)
66	msg = f"Analysis results saved to {SAVE_FILE} for faster subsequent loading."
67	queen.save_data(SAVE_FILE)
68
69	entries = queen.get_data()
70	if sort == "energy_per_atom":
71	entries = sorted(entries, key=lambda x: x.energy_per_atom)
72	elif sort == "filename":
73	entries = sorted(entries, key=lambda x: x.data["filename"])
74
75	all_data = []
76	for e in entries:
77	if quick:
78	delta_vol = "NA"
79	else:
80	delta_vol = e.structure.volume / e.data["initial_structure"].volume - 1
81	delta_vol = f"{delta_vol * 100:.2f}"
82	all_data.append(
83	(
84	e.data["filename"].replace("./", ""),
85	re.sub(r"\s+", "", e.composition.formula),
86	f"{e.energy:.5f}",
87	f"{e.energy_per_atom:.5f}",
88	delta_vol,
89	)
90	)
91	if len(all_data) > 0:
92	headers = ("Directory", "Formula", "Energy", "E/Atom", "% vol chg")
93	print(tabulate(all_data, headers=headers, tablefmt=fmt))
94	print()
95	print(msg)
96	else:
97	print("No valid vasp run found.")
98	os.unlink(SAVE_FILE)
99	return 0
100
101
102	def get_magnetizations(dir: str, ion_list: list[int]):
103	"""
104	Get magnetization info from OUTCARs.
105
106	Args:
107	mydir (str): Directory name
108	ion_list (list[int]): List of ions to obtain magnetization information for.
109
110	Returns:
111	int: 0 if successful.
112	"""
113	data = []
114	max_row = 0
115	for parent, _subdirs, files in os.walk(dir):
116	for f in files:
117	if re.match(r"OUTCAR*", f):
118	try:
119	row = []
120	fullpath = os.path.join(parent, f)
121	outcar = Outcar(fullpath)
122	mags = outcar.magnetization
123	mags = [m["tot"] for m in mags]
124	all_ions = list(range(len(mags)))
125	row.append(fullpath.lstrip("./"))
126	if ion_list:
127	all_ions = ion_list
128	for ion in all_ions:
129	row.append(str(mags[ion]))
130	data.append(row)
131	if len(all_ions) > max_row:
132	max_row = len(all_ions)
133	except Exception:
134	pass
135
136	for d in data:
137	if len(d) < max_row + 1:
138	d.extend([""] * (max_row + 1 - len(d)))
139	headers = ["Filename"]
140	for i in range(max_row):
141	headers.append(str(i))
142	print(tabulate(data, headers))
143	return 0
144
145
146	def analyze(args):
147	"""
148	Master function controlling which analysis to call.
149
150	Args:
151	args (dict): args from argparse.
152	"""
153	default_energies = not (args.get_energies or args.ion_list)
154
155	if args.get_energies or default_energies:
156	for d in args.directories:
157	return get_energies(d, args.reanalyze, args.verbose, args.quick, args.sort, args.format)
158	if args.ion_list:
159	if args.ion_list[0] == "All":
160	ion_list = None
161	else:
162	(start, end) = (int(i) for i in re.split(r"-", args.ion_list[0]))
163	ion_list = list(range(start, end + 1))
164	for d in args.directories:
165	return get_magnetizations(d, ion_list)
166
167	return -1

materialsproject / pymatgen / 4075885785

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous