Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
for fl, contents in outfiles.items():
if contents is not None:
# LOG text += f'\n DFTD3 scratch file {fl} has been read.\n'
pass
# parse energy output (could go further and break into E6, E8, E10 and Cn coeff)
real = np.array(input_model.molecule.real)
full_nat = real.shape[0]
real_nat = np.sum(real)
for ln in stdout.splitlines():
if re.match(" Edisp /kcal,au", ln):
ene = Decimal(ln.split()[3])
elif re.match(r" E6\(ABC\) \" :", ln): # c. v3.2.0
raise ResourceError("Cannot process ATM results from DFTD3 prior to v3.2.1.")
elif re.match(r""" E6\(ABC\) /kcal,au:""", ln):
atm = Decimal(ln.split()[-1])
elif re.match(" normal termination of dftd3", ln):
break
else:
if not ((real_nat == 1) and (input_model.driver == "gradient")):
raise UnknownError(
f"Unsuccessful run. Check input, particularly geometry in [a0]. Model: {input_model.model}"
)
# parse gradient output
# * DFTD3 crashes on one-atom gradients. Avoid the error (above) and just force the correct result (below).
if outfiles["dftd3_gradient"] is not None:
srealgrad = outfiles["dftd3_gradient"].replace("D", "E")
realgrad = np.fromstring(srealgrad, count=3 * real_nat, sep=" ").reshape((-1, 3))
elif real_nat == 1:
if output_data["success"] is False:
error_message = output_data["error"]["error_message"]
error_type = output_data["error"]["error_type"]
else:
compute_success = True
else:
error_message = output["stderr"]
error_type = "execution_error"
# Dispatch errors, PSIO Errors are not recoverable for future runs
if compute_success is False:
if "PSIO Error" in error_message:
if "scratch directory" in error_message:
# Psi4 cannot access the folder or file
raise ResourceError(error_message)
else:
# Likely a random error, worth retrying
raise RandomError(error_message)
elif ("SIGSEV" in error_message) or ("SIGSEGV" in error_message) or ("segmentation fault" in error_message):
raise RandomError(error_message)
elif ("TypeError: set_global_option" in error_message) or (error_type == "ValidationError"):
raise InputError(error_message)
elif "RHF reference is only for singlets" in error_message:
raise InputError(error_message)
else:
raise UnknownError(error_message)
# Reset basis
output_data["model"]["basis"] = old_basis
# Move several pieces up a level
def compute(self, input_model: "AtomicInput", config: "TaskConfig") -> "AtomicResult":
from ..testing import is_program_new_enough
self.found(raise_error=True)
if not is_program_new_enough("mp2d", "1.1"):
raise ResourceError(f"MP2D version '{self.get_version()}' too old. Please update to at least '1.1'.")
job_inputs = self.build_input(input_model, config)
success, dexe = self.execute(job_inputs)
if success:
dexe["outfiles"]["stdout"] = dexe["stdout"]
dexe["outfiles"]["stderr"] = dexe["stderr"]
output_model = self.parse_output(dexe["outfiles"], input_model)
else:
output_model = input_model
output_model["error"] = {"error_type": "execution_error", "error_message": dexe["stderr"]}
return output_model
def compute(self, input_model: "AtomicInput", config: "TaskConfig") -> "AtomicResult":
"""
Runs Psi4 in API mode
"""
self.found(raise_error=True)
pversion = parse_version(self.get_version())
if pversion < parse_version("1.2"):
raise ResourceError("Psi4 version '{}' not understood.".format(self.get_version()))
# Location resolution order config.scratch_dir, $PSI_SCRATCH, /tmp
parent = config.scratch_directory
if parent is None:
parent = os.environ.get("PSI_SCRATCH", None)
error_type = None
error_message = None
compute_success = False
# Basis must not be None for HF3c
old_basis = input_model.model.basis
input_model.model.__dict__["basis"] = old_basis or ""
with temporary_directory(parent=parent, suffix="_psi_scratch") as tmpdir:
def compute(self, input_data: "AtomicInput", config: "TaskConfig") -> "AtomicResult":
"""
Runs TorchANI in FF typing
"""
# Check if existings and version
self.found(raise_error=True)
if parse_version(self.get_version()) < parse_version("0.9"):
raise ResourceError("QCEngine's TorchANI wrapper requires version 0.9 or greater.")
import torch
import torchani
import numpy as np
device = torch.device("cpu")
# Failure flag
ret_data = {"success": False}
# Build model
model = self.get_model(input_data.model.method)
if model is False:
raise InputError("TorchANI only accepts the ANI1x or ANI1ccx method.")
# Build species
``True`` Do raise error if program not found. ``False`` is handy for
the specialized case of calling non-execution methods (like parsing for testing)
on the returned ``Harness``.
"""
name = name.lower()
if name not in programs:
raise InputError(f"Program {name} is not registered to QCEngine.")
ret = programs[name]
if check:
try:
ret.found(raise_error=True)
except ModuleNotFoundError as err:
raise ResourceError(f"Program {name} is registered with QCEngine, but cannot be found.") from err
return ret
def get_procedure(name: str) -> "ProcedureHarness":
"""
Returns a procedures executor class
"""
name = name.lower()
if name not in procedures:
raise InputError(f"Procedure {name} is not registered to QCEngine.")
ret = procedures[name]
if not ret.found():
raise ResourceError(f"Procedure {name} is registered with QCEngine, but cannot be found.")
return ret