Source code for storq.vasp.core

import numpy as np
import os
import storq.vasp.validate as validate
import warnings
from ase.calculators.calculator import equal, compare_atoms
from ase.io import read
from storq.tools import ascii_atoms
from storq.tools import read_configuration, read_siteconf, read_batchconf
from storq.vasp.state_engine import StateEngine

# Mixins - order matters!
from storq.vasp.readers import Readers
import storq.vasp.readers as readers
from storq.vasp.getters import Getters
from storq.vasp.helpers import Helpers
from storq.vasp.relaxer import Relaxer
from storq.vasp.resource_manager import ResourceManager
from storq.vasp.runner import Runner
from storq.vasp.setters import Setters
from storq.vasp.writers import Writers


# To extend the VASP class we inherit from mixins classes.
[docs]class Vasp( Getters, Helpers, Readers, ResourceManager, Runner, Setters, Writers, Relaxer ): """Class for doing VASP calculations. This is the VASP calculator class. While it does not strictly adhere to the ASE Calculator interface (and hence does not inherit from it) it should still be compatible with all other ASE modules which use calculators. Parameters ---------- label : str the directory where the calculation files will be and the calculation run. **kwargs Any VASP INCAR keyword can be used, e.g., ``encut=450``. Python types will be converted to a format appropriate for the INCAR file, e.g. True / False becomes .TRUE. / .FALSE. xc : str exchange-correlation functional to use; it is expanded from :class:`Vasp._xc_defaults` to the relevant VASP tags. kpts : list/dict In the simplest case this is just a list with 3 elements containing the **k**-point mesh divisions. More control can be achieved by passing a dict, which admits the following key-value pairs: - size (``list``) - spacing (``float``) - auto (``float``) - list (``list``) - lines (``list``): - gamma (``bool``): if False use Monkhorst-Pack else Gamma-centered - shift (``list``): shift of the **k**-point grid - intersections: number of intersections used in line `vasp_mode` - cartesian (`bool`) setups : list Special setups for the POTCARS. It is a list of the following items ``[atom_index, suffix]`` for example: ``[2, '_sv']`` ``[atom_symbol, suffix]`` for example ``['Zr', '_sv']`` If ``[atom_index, suffix]`` is used then only that atom index will have a POTCAR defined by ``'{}{}'.format(atoms[atom_index].symbol, suffix)`` If ``[atom_symbol, suffix]`` is used then atoms with that symbol (except any identified by ``[atom_index, suffix]`` will use a POTCAR defined by ``'{}{}'.format(atom_symbol, suffix)`` This syntax has changed from the old dictionary format. The reason for this is that this sorting must be deterministic. Getting keys in a dictionary is not deterministic. ldau_luj : dict This is a dictionary to set the DFT+U tags. For example, to put U=4 on the d-orbitals (L=2) of Cu, and nothing on the oxygen atoms in a calculation use:: ldau_luj={'Cu':{'L': 2, 'U':4.0, 'J':0.0}, 'O': {'L':-1, 'U':0.0, 'J':0.0}}, """ version = "1.0.0" name = "VASP" # These allow you to use simple strings for the xc kwarg and automatically # set the relevant INCAR tags. _xc_defaults = { "lda": {"pp": "LDA"}, # GGAs "gga": {"pp": "GGA"}, "pbe": {"pp": "PBE"}, "revpbe": {"pp": "LDA", "gga": "RE"}, "rpbe": {"pp": "LDA", "gga": "RP"}, "am05": {"pp": "LDA", "gga": "AM"}, "pbesol": {"pp": "PBE", "gga": "PS"}, # Meta-GGAs "tpss": {"pp": "PBE", "metagga": "TPSS"}, "revtpss": {"pp": "PBE", "metagga": "RTPSS"}, "m06l": {"pp": "PBE", "metagga": "M06L"}, # vdW-DFs "cx": {"pp": "PBE", "gga": "CX", "luse_vdw": True, "aggac": 0.0000}, "optpbe-vdw": {"pp": "LDA", "gga": "OR", "luse_vdw": True, "aggac": 0.0}, "optb88-vdw": { "pp": "LDA", "gga": "BO", "luse_vdw": True, "aggac": 0.0, "param1": 1.1 / 6.0, "param2": 0.22, }, "optb86b-vdw": { "pp": "LDA", "gga": "MK", "luse_vdw": True, "aggac": 0.0, "param1": 0.1234, "param2": 1.0, }, "vdw-df2": { "pp": "LDA", "gga": "ML", "luse_vdw": True, "aggac": 0.0, "zab_vdw": -1.8867, }, "beef-vdw": { "pp": "PBE", "gga": "BF", "luse_vdw": True, "zab_vdw": -1.8867, "lbeefens": True, }, # hybrids "pbe0": {"pp": "LDA", "gga": "PE", "lhfcalc": True}, "hse03": {"pp": "LDA", "gga": "PE", "lhfcalc": True, "hfscreen": 0.3}, "hse06": {"pp": "LDA", "gga": "PE", "lhfcalc": True, "hfscreen": 0.2}, "b3lyp": { "pp": "LDA", "gga": "B3", "lhfcalc": True, "aexx": 0.2, "aggax": 0.72, "aggac": 0.81, "aldac": 0.19, }, "hf": {"pp": "PBE", "lhfcalc": True, "aexx": 1.0, "aldac": 0.0, "aggac": 0.0}, } _default_parameters = dict( xc="PBE", pp="PBE", ismear=1, sigma=0.1, lwave=False, lcharg=False, kpts={ "size": [1, 1, 1], "gamma": False, "shift": [0.0, 0.0, 0.0], "surface": False, }, ) # VASP Calculator states EMPTY = 0 NEW = 1 QUEUED = 2 RUNNING = 3 FINISHED = 4 UNFINISHED = 5 EMPTYCONTCAR = 6 CONVERGED = 7 UNCONVERGED = 8 PARSED = 9 UNKNOWN = 100 siteconf = read_siteconf() def __init__(self, label, atoms=None, restart=True, **kwargs): # Initialization, note that the Atoms object will be set at a later time. self.atoms = None self.parameters = None self.state = None self.jobid = None self.results = {} self.conf = {} self.restart = restart if not hasattr(self, "name"): self.name = self.__class__.__name__.lower() self.set_label(label) # this sets the calculation directory self.db = os.path.join(self.directory, "storq.db") # Restart option and configuration files, if no restart is requested # any previously existing files are nuked. Configs are then read in # the order storq.db -> ~/.config/storq if self.restart is False: self.remove(self.directory, target="all") elif isinstance(self.restart, list): self.remove(self.directory, target="all", exceptions=restart) if not os.path.exists(self.db): self.conf.update(read_batchconf()) # Read any files that were already present in calculation dir. self.read() # Use default parameters if they were not read from file. # This only takes care of the normal kwargs and not the special dict kwargs, # which are handled further down. if self.parameters is None: self.parameters = self._default_parameters # Add default parameters if they aren't set otherwise. for key, val in self._default_parameters.items(): if key != "kpts": if key not in kwargs and key not in self.parameters: kwargs[key] = val else: if "kspacing" not in kwargs and "kspacing" not in self.parameters: if key not in kwargs and key not in self.parameters: kwargs[key] = val else: kwargs[key] = None # Next we update kwargs with the special kwarg # dictionaries. ispin, rwigs are special, and needs sorted # atoms. so we save it for later. if "ispin" in kwargs: ispin = kwargs["ispin"] del kwargs["ispin"] else: ispin = None if "rwigs" in kwargs: rwigs = kwargs["rwigs"] del kwargs["rwigs"] else: rwigs = None if "ldau_luj" in kwargs: ldau_luj = kwargs["ldau_luj"] del kwargs["ldau_luj"] else: ldau_luj = None # Now update the parameters. If there are any new kwargs here, it will # reset the calculator and cause a calculation to be run if needed. self.set(**kwargs) # Handle any Atoms object passed to the calculator. if atoms is not None: self.set_atoms(atoms) # These depend on having atoms already. if ispin is not None: self.set(**self.set_ispin_dict(ispin)) if rwigs is not None: self.set(**self.set_rwigs_dict(rwigs)) if ldau_luj is not None: self.set(**self.set_ldau_luj_dict(ldau_luj)) # Finally run vasp_validate functions if self.conf["vasp_validate"]: for key, val in self.parameters.items(): if key in validate.__dict__: if val is not None: f = validate.__dict__[key] f(self, val) else: print("WARNING: no validation for {}".format(key))
[docs] def sort_atoms(self, atoms): """Generates resorted list and list of ``POTCAR`` files to use. This sets the ``atoms`` attribute along with ``atoms_sorted``, which is the actual :class:`Atoms` object that is written to the ``POSCAR`` file. Parameters ---------- atoms : Atoms configuration to sort """ self.resort = None self.ppp_list = None self.symbol_count = None self.atoms = atoms # Now we sort the atoms and generate the list of POTCARS # We end up with ppp = [(index_or_symbol, potcar_file, count)] # and resort_indices setups = self.parameters.get("setups", []) pp = self.parameters["pp"] ppp = [] # [(index_or_symbol, potcar_file, count)] # indices of original atoms needed to make sorted atoms list sort_indices = [] # First the numeric index setups for setup in [x for x in setups if isinstance(x[0], int)]: ppp += [[setup[1], "potpaw_{}/{}/POTCAR".format(pp, setup[1]), 1]] sort_indices += [setup[0]] # now the rest of the setups. These are atom symbols for setup in [x for x in setups if not isinstance(x[0], int)]: symbol = setup[0] count = 0 for i, atom in enumerate(atoms): if atom.symbol == symbol and i not in sort_indices: count += 1 sort_indices += [i] ppp += [ [symbol, "potpaw_{}/{}{}/POTCAR".format(pp, symbol, setup[1]), count] ] # now the remaining atoms use default vasp_potentials # First get the chemical symbols that remain symbols = [] for atom in atoms or []: if atom.symbol not in symbols: symbols += [atom.symbol] for symbol in symbols: count = 0 for i, atom in enumerate(atoms): if atom.symbol == symbol and i not in sort_indices: sort_indices += [i] count += 1 if count > 0: ppp += [[symbol, "potpaw_{}/{}/POTCAR".format(pp, symbol), count]] assert len(sort_indices) == len(atoms), "Sorting error. sort_indices={}".format( sort_indices ) assert sum([x[2] for x in ppp]) == len(atoms) self.sort = sort_indices # This list is used to convert Vasp ordering back to the # user-defined order. self.resort = [ k[1] for k in sorted([[j, i] for i, j in enumerate(sort_indices)]) ] self.ppp_list = ppp self.atoms_sorted = atoms[sort_indices] self.symbol_count = [ (x[0] if isinstance(x[0], str) else atoms[x[0]].symbol, x[2]) for x in ppp
] def __str__(self): """Pretty VASP representation of a calculation.""" s = ["\n", "Vasp calculation directory:"] s += ["---------------------------"] s += [" [[{self.directory}]]"] atoms = self.atoms cell = atoms.get_cell() A, B, C = [i for i in cell] latt = list(map(np.linalg.norm, cell)) a, b, c = latt alpha = np.arccos(np.dot(B / b, C / c)) * 180 / np.pi beta = np.arccos(np.dot(A / a, C / c)) * 180 / np.pi gamma = np.arccos(np.dot(A / a, B / b)) * 180 / np.pi s += ["\nSystem:"] s += ["-----------------"] s += ascii_atoms(self.atoms) s += ["\nCell:"] s += [" {:^8}{:^8}{:^8}{:>12}".format("x", "y", "z", "|v|")] for i, v in enumerate(cell): s += [ " v{0}{2:>8.3f}{3:>8.3f}{4:>8.3f}" "{1:>12.3f} Ang".format(i, latt[i], *v) ] s += [ "\n alpha, beta, gamma (deg):" "{:>6.1f}{:>6.1f}{:>6.1f}".format(alpha, beta, gamma) ] volume = atoms.get_volume() s += [" Total volume:{:>25.3f} Ang^3".format(volume)] if self.state in [self.FINISHED, self.CONVERGED, self.UNCONVERGED]: self.read_results() stress = self.results.get("stress", None) if stress is not None: s += [ " Stress:{:>6}{:>7}{:>7}" "{:>7}{:>7}{:>7}".format("xx", "yy", "zz", "yz", "xz", "xy") ] s += [ "{:>15.3f}{:7.3f}{:7.3f}" "{:7.3f}{:7.3f}{:7.3f} GPa\n".format(*stress) ] s += ["\nAtoms:"] s += [ " {:<4}{:<8}{:<3}{:^10}{:^10}{:^10}" "{:>14}".format("ID", "tag", "sym", "x", "y", "z", "rmsF (eV/A)") ] from ase.constraints import FixAtoms, FixScaled constraints = [[None, None, None] for atom in atoms] for constraint in atoms.constraints: if isinstance(constraint, FixAtoms): for i in constraint.index: constraints[i] = [True, True, True] elif isinstance(constraint, FixScaled): constraints[constraint.a] = constraint.mask.tolist() forces = self.results.get("forces", None) if forces is not None: for i, atom in enumerate(atoms): rms_f = np.sum(forces[i] ** 2) ** 0.5 s += [ " {:<4}{:<8}{:3}{:9.3f}{}{:9.3f}{}{:9.3f}{}" "{:>10.2f}".format( i, atom.tag, atom.symbol, atom.x, "*" if constraints[i][0] is True else " ", atom.y, "*" if constraints[i][1] is True else " ", atom.z, "*" if constraints[i][1] is True else " ", rms_f, ) ] energy = self.results.get("energy", np.nan) s += ["\n Potential energy: {:.4f} eV".format(energy)] if self.state in [self.FINISHED, self.CONVERGED, self.UNCONVERGED]: if self.conf["vasp_convergence"] == "strict": s += [" Converged: {} ".format(self.check_convergence())] elif self.conf["vasp_convergence"] == "basic": s += [" Finished: {} ".format(self.check_outcar())] else: s += [" CONVERGENCE CHECKS ARE DISABLED"] s += ["\nINPUT Parameters:"] s += ["-----------------"] for key, value in self.parameters.items(): if type(value) != dict: s += [" {0:<10} = {1}".format(key, value)] elif key == "kpts": s += [" kpts = {{"] for kpts_key, kpts_val in value.items(): if kpts_val is not None: if kpts_key == "surface" and kpts_val is False: pass elif kpts_key == "shift" and np.isclose(np.sum(kpts_val), 0.0): pass else: s += [" {0:>17} : {1}".format(kpts_key, kpts_val)] s += [" }}"] s += ["\nPseudovasp_potentials used:"] s += ["----------------------"] for sym, ppp, date in self.get_pseudopotentials(): s += [" {}: {} {}".format(sym, ppp, date)] return "\n".join(s).format(self=self)
[docs] def set_label(self, label): """Sets working directory. Note that With regards to the ASE FileIOCalculator interface there is no prefix, only a working directory. Parameters ---------- label : string Absolute or relative path to the VASP calculation directory, i.e. where all input/output files related to VASP are put. """ if label is None: self.directory = os.path.abspath(".") self.prefix = None else: d = os.path.expanduser(label) d = os.path.abspath(d) self.directory, self.prefix = d, None if not os.path.isdir(self.directory): os.makedirs(self.directory) # Convenient attributes for file names for f in ["INCAR", "POSCAR", "CONTCAR", "POTCAR", "KPOINTS", "OUTCAR"]: fname = os.path.join(self.directory, f)
setattr(self, f.lower(), fname)
[docs] def check_changes(self, atoms): """Checks if any changes exist that require new calculations. Parameters ---------- atoms : Atoms configuration to compare with, typically read from disk """ system_changes = compare_atoms(self.atoms, atoms) # Ignore boundary conditions: if "pbc" in system_changes: system_changes.remove("pbc") # if dir is empty, there is nothing to read here. if self.state == Vasp.EMPTY: return system_changes # Check if the parameters have changed if not os.path.exists(self.db): file_params = {} file_params.update(self.read_incar()) file_params.update(self.read_potcar()) if os.path.exists(self.kpoints): file_params.update(self.read_kpoints()) xc_keys = sorted( Vasp._xc_defaults, key=lambda k: len(Vasp._xc_defaults[k]), reverse=True ) for ex in xc_keys: pd = {k: file_params.get(k, None) for k in Vasp._xc_defaults[ex]} if pd == Vasp._xc_defaults[ex]: file_params["xc"] = ex.lower() break # reconstruct ldau_luj if necessary if "ldauu" in file_params: ldaul = file_params["ldaul"] ldauj = file_params["ldauj"] ldauu = file_params["ldauu"] with open(self.potcar) as f: lines = f.readlines() # symbols are in the first line of each potcar symbols = [lines[0].split()[1]] for i, line in enumerate(lines): if "End of Dataset" in line and i != len(lines) - 1: symbols += [lines[i + 1].split()[1]] ldau_luj = {} for sym, l, j, u in zip(symbols, ldaul, ldauj, ldauu): ldau_luj[sym] = {"L": l, "U": u, "J": j} file_params["ldau_luj"] = ldau_luj if ( not {k: v for k, v in self.parameters.items() if v is not None} == file_params ): system_changes += ["params_on_file"]
return system_changes
[docs] def calculation_required(self): """Checks whether a calculation is needed. Uses the internal state of the calculator to judge whether a calculation should be run. The result will be influenced by the calculator settings, e.g., whether strict convergence checks are enforced or not. Returns ------- bool True if calculation needs to be run """ # if we have a brand new calculation or just # some input files we should run if self.state in [Vasp.NEW, Vasp.EMPTY]: return True # in we are in queue vasp_mode we check if we are waiting to run # or have an ongoing calculation if self.state == Vasp.QUEUED: print( "Batch job {1}: {0} is in the queue".format( os.path.basename(self.directory), self.jobid ) ) return False elif self.state == Vasp.RUNNING: print( "Batch job {1}: {0} is running".format( os.path.basename(self.directory), self.jobid ) ) return False # if something has changed we need to run system_changes = self.check_changes(self.read_atoms()) if system_changes: return True # check if we have already parsed everything if self.state == Vasp.PARSED: return False # first we perform convergence checks if asked for if self.conf["vasp_convergence"] == "strict": if self.state == Vasp.CONVERGED: return False else: # if the state is any other than converged if self.conf["vasp_restart_on"] == "convergence": return True else: # if we don't trigger a restarts just give a warning warnings.warn( "WARNING: {0}: Calculation is not converged.".format( os.path.basename(self.directory) ) ) return False elif self.conf["vasp_convergence"] == "basic": if self.state == Vasp.FINISHED: return False elif self.state == Vasp.UNFINISHED: if self.conf["vasp_restart_on"] == "basic": return True else: # if we do not care about convergence issue a warning warnings.warn( "WARNING: Vasp did not finish; directory: {}".format( os.path.basename(self.directory) ) ) return False else: # convergence checks were None
return False
[docs] def get_state(self): """Determines calculation state based on directory contents. This method is only responsible for determining the state and does not act upon it. Other methods that cause a transition to a known state should alter the state themselves. Returns ------- int An integer corresponding to the internal state of the calculator. """ # We do not check for KPOINTS here. That file may not exist if # the kspacing INCAR parameter is used. base_input = [ os.path.exists(os.path.join(self.directory, f)) for f in ["INCAR", "POSCAR", "POTCAR"] ] # Some input does not exist if False in base_input: return Vasp.EMPTY # Input files exist, but no jobid, and no output if ( np.array(base_input).all() and self.jobid is None and not os.path.exists(os.path.join(self.directory, "OUTCAR")) ): return Vasp.NEW # INPUT files exist, a jobid in the queue job_info = self.get_job_info() if job_info is not None: status = job_info.get("st", None) if not status: # certain versions of slurm use a different key status = job_info.get("state", None) if status == "PD": return Vasp.QUEUED elif status in ["R", "CG", "CD"]: return Vasp.RUNNING else: return Vasp.UNKNOWN else: # we are not in queue if self.conf["vasp_convergence"] in ["basic", "strict"]: if self.check_outcar(): if self.conf["vasp_convergence"] == "strict": if self.check_convergence(): return Vasp.CONVERGED else: return Vasp.UNCONVERGED else: # we just care about VASP exiting without errors return Vasp.FINISHED # if we get here there was an error or we hit batch_walltime else: return Vasp.UNFINISHED # check if contcar is empty if os.path.exists(self.contcar): with open(self.contcar) as f: if f.read() == "": return Vasp.EMPTYCONTCAR # if we haven' returned yet the state is unknown
return Vasp.UNKNOWN
[docs] def check_state(self, atoms, tol=1e-15): """ Required from interoperability with ASE DB """
return compare_atoms(self.atoms, atoms)
[docs] def get_atoms(self): """Gets a copy of the calculators Atoms object. Some ASE functions may rely on the calculator having this method. Returns ------- Atoms copy of the configuration associated with the calculator """
return self.atoms.copy()
[docs] def get_property(self, name, atoms=None, allow_calculation=False): """Gets a property from the self.results dict. Some ASE functions may relay on the calculator having this method. Parameters ---------- name : str name of property to get atoms : Atoms ignored but required by the ASE interface allow_calculation : bool ignored but required by the ASE interface Returns ------- object The object representing the desired result. """ if atoms is None: atoms = self.atoms if name == 'magmom' and 'magmom' not in self.results: return 0.0 if name == 'magmoms' and 'magmoms' not in self.results: return np.zeros(len(atoms)) if name not in self.results: return None else: result = self.results[name] if isinstance(result, np.ndarray): result = result.copy()
return result # Patched ASE-mandated getters
[docs] def get_potential_energy(self, atoms=None, force_consistent=False): """Gets the potential energy of the system. This method triggers a new calculation if necessary. Some ASE functions rely on the calculator having this method with this particular signature. Parameters ---------- atoms : Atoms Only used if atoms.get_potential_energy() is called force_consistent : bool Toggle parsing of free energy. Returns ------- float Energy or the free energy if force_consistent=True. """ with StateEngine(self): energy = self.get_property("energy") if force_consistent: if "free_energy" not in self.results: name = self.__class__.__name__ raise PropertyNotImplementedError( 'Force consistent/free energy ("free_energy") ' "not provided by {} calculator".format(name) ) return self.results["free_energy"] else:
return energy
[docs] def get_forces(self, atoms=None): """Gets the forces acting on the ions. This method triggers a new calculation if necessary. Some ASE functions rely on the calculator having this method with this particular signature. Parameters ---------- atoms : Atoms Not used but required by ASE. Returns ------- np.ndarray Array with forces acting on the ions. """ with StateEngine(self):
return self.get_property("forces", atoms)
[docs] def get_magnetic_moment(self, atoms=None): """Get the magnetic moment. This method triggers a new calculation if necessary. Some ASE functions rely on the calculator having this method with this particular signature. Parameters ---------- atoms : Atoms Not used but required by ASE. Returns ------- np.ndarray Magnetic moment """ with StateEngine(self):
return self.get_property("magmom", atoms)
[docs] def get_magnetic_moments(self, atoms=None): """Get the magnetic moments. This method triggers a new calculation if necessary. Some ASE functions rely on the calculator having this method with this particular signature. Parameters ---------- atoms : Atoms Not used but required by ASE. Returns ------- np.ndarray Magnetic moments """ with StateEngine(self):
return self.get_property("magmoms", atoms)
[docs] def get_stress(self, atoms=None): """Gets the stress. This method triggers a new calculation if necessary. Some ASE functions rely on the calculator having this method with this particular signature. Parameters ---------- atoms : Atoms Not used but required by ASE. Returns ------- np.ndarray The stress. """ with StateEngine(self):
return self.get_property("stress", atoms) @property def traj(self): """Gets a trajectory. This reads Atoms objects from ``vasprun.xml``. By default all images are returned. If index is an integer, returns that image. Technically, this is just a list of atoms with a SinglePointCalculator attached to them. This is usually only relevant if you have done a relaxation. If the calculation is an NEB, the images are returned. Returns ------- list Atoms objects representing the trajectory """ from ase.calculators.singlepoint import SinglePointCalculator as SPC LOA = [] for atoms in read(os.path.join(self.directory, "vasprun.xml"), ":"): catoms = atoms.copy() catoms = catoms[self.resort] catoms.set_calculator( SPC( catoms, energy=atoms.get_potential_energy(), forces=atoms.get_forces()[self.resort], ) ) LOA += [catoms] return LOA
[docs] def view(self, index=None): """Visualizes the calculation using the ASE gui. Parameters --------- index : int Index of a particular snapshot to view. Returns ------- ASE gui view of the trajectory. """ from ase.visualize import view if index is not None: return view(self.traj[index]) else:
return view(self.traj)
[docs] def describe_parameters(self, long=False): """Describes the parameters using docstrings from vasp_validate.""" for key in sorted(self.parameters.keys()): if key in validate.__dict__: f = validate.__dict__[key] d = f.__doc__ or "No description available." print("{} = {}:".format(key, self.parameters[key])) if long: print(" " + d) else: print(" " + d.split("\n")[0])
print("")
[docs] def run(self): """Convenience function for running the calculator."""
return self.get_potential_energy()
[docs] def stop_if(self, condition): """Stops the program based on the boolean evaluation of condition.""" if condition: import sys
sys.exit()
[docs] def configure(self, cores='ignore', nodes='ignore', walltime='ignore', account='ignore', options='ignore', mpi='ignore', **kwargs): """Changes the configuration of the VASP calculator. Only allows changes to the calculator and not to the site configuration of the cluster resource manager. Note that any changes made to the configuration will be written to a database and persist until changed. Parameters ---------- cores : int The number of cores to use in the calculation. nodes : int The number of nodes to use in the calculation. walltime : str The walltime of the calculation in the format '[d-]hh:mm:ss'. account : str The allocation on the supercomputing resource. options : str Space-separated string of extra options that will be passed to the resource manager. mpi : str Space-separated string of extra options that will be passed to the MPI executable or wrapper. **kwargs Any admissible storq settings. """ shorthands = {'cores': cores, 'nodes': nodes, 'walltime': walltime, 'account': account, 'options': options} for short, val in shorthands.items(): if val != 'ignore': key = "batch_{}".format(short) kwargs[key] = val if mpi != 'ignore': key = "mpi_options" kwargs[key] = mpi
self.conf.update(kwargs)
[docs] def check_convergence(self): """Checks whether a calculation has converged. Performs a strict convergence checks in the sense that electronic as well as ionic convergence is explicitly checked for by comparing their magnitude to EDIFF and EDIFFG. Returns ------- bool True if calculation is converged """ converged = readers.read_convergence(self.outcar) # it only makes sense to check for ionic relaxation # if we used ibrion 1 or 2 if ( self.parameters.get("ibrion", None) in [1, 2] and self.parameters.get("nsw", 0) > 1 ): return all(converged) else:
return converged[0]
[docs] def check_outcar(self): """Checks if VASP finished properly. If VASP itself is canceled (batch_walltime or explicitly by user) or encounters certain errors it won't output the typical Voluntary context switch count on the last line. Returns ------- bool True if VASP finished properly """
return readers.is_outcar_valid(self.outcar)
[docs] def ready(self): """Checks if the calculation is ready to be parsed. The outcome will be heavily dependent on the calculator configuration. For instance if the highest level of convergence check is enabled this will be enforced when checking for readiness. Returns ------- bool True if the calculation is ready """ if self.calculation_required() is True: return False elif self.state == Vasp.QUEUED: return False else: # handle the special case of cell relaxations and restarting if self.parameters.get("isif", None) in range(3, 7): if self.get_number_of_ionic_steps() == 1: # we're done return True else: # if we care about convergence we are not done if self.conf.get("vasp_restart_on", None) is "convergence": return False else: # we don't care and just warn the user warnings.warn( "WARNING: {}: Cell relaxation finished with more" " than one ionic step".format( os.path.basename(self.directory) ) ) return True else: # no cell relax so we're done
return True
[docs] def clone(self, newdir): self.copy_files(self.directory, newdir) self.set_label(newdir)
self.write_persistence_db()
[docs] def todict(self, skip_default=True): """. Parameters ---------- skip_default : bool Returns ------- ... """ defaults = self._default_parameters dct = {} for key, value in self.parameters.items(): if hasattr(value, "todict"): value = value.todict() if skip_default: default = defaults.get(key, "_no_default_") if default != "_no_default_" and equal(value, default): continue dct[key] = value
return dct