Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_dftd3__from_arrays__error(inp):
with pytest.raises(qcng.exceptions.InputError):
empirical_dispersion_resources.from_arrays(**inp)
return qcel.models.AtomicResult(
**{
**input_data.dict(),
**{
"properties": {"return_energy": grad_value},
"return_result": grad,
"success": True,
"extras": {"ncalls": self.ncalls},
"provenance": {"creator": "failure_engine", "ncores": config.ncores},
},
}
)
elif mode == "random_error":
raise qcng.exceptions.RandomError("Whoops!")
elif mode == "input_error":
raise qcng.exceptions.InputError("Whoops!")
else:
raise KeyError("Testing error, should not arrive here.")
def parse_logfile(self, outfiles: Dict[str, str]) -> AtomicResult:
"""
Parses a log file.
"""
warnings.warn(
"parse_logfile will result in precision loss for some fields due to trunctation in " "Q-Chem output files."
)
outtext = outfiles["dispatch.out"]
mobj = re.search(r"(?:User input\:|Running Job?)\s+\d+\s+of\s+(\d+)", outtext)
if mobj:
if int(mobj.group(1)) > 1:
raise InputError("Multi-job Q-Chem log files not supported.")
input_dict = {}
mobj = re.search(r"\n-{20,}\nUser input:\n-{20,}\n(.+)\n-{20,}", outtext, re.DOTALL)
if mobj:
inputtext = mobj.group(1)
rem_match = re.search(r"\$rem\s*\n([^\$]+)\n\s*\$end", inputtext, re.DOTALL | re.IGNORECASE)
if rem_match:
rem_text = rem_match.group(1)
lines = rem_text.split("\n")
keywords = {}
for line in lines:
s = re.sub(r"(^|[^\\])!.*", "", line).split()
if len(s) == 0:
continue
keywords[s[0].lower()] = s[1].lower()
def build_input(
self, input_model: "AtomicInput", config: "TaskConfig", template: Optional[str] = None
) -> Dict[str, Any]:
# strip engine hint
mtd = input_model.model.method
if mtd.startswith("d3-"):
mtd = mtd[3:]
if (input_model.driver.derivative_int() > 1) or (input_model.driver == "properties"):
raise InputError(f"""DFTD3 valid driver options are 'energy' and 'gradient', not {input_model.driver}""")
# temp until actual options object
input_model.extras["info"] = empirical_dispersion_resources.from_arrays(
name_hint=mtd,
level_hint=input_model.keywords.get("level_hint", None),
param_tweaks=input_model.keywords.get("params_tweaks", None),
dashcoeff_supplement=input_model.keywords.get("dashcoeff_supplement", None),
)
# this is what the dftd3 program needs, not what the job needs
# * form dftd3_parameters string that governs dispersion calc
# * form dftd3_geometry string that supplies geometry to dispersion calc
# * form command and arguments
# Need 'real' field later and that's only guaranteed for molrec
molrec = qcel.molparse.from_schema(input_model.molecule.dict())
def compute(self, input_model: AtomicInput, config: "TaskConfig") -> AtomicResult:
"""
Runs NWChem in executable mode
"""
self.found(raise_error=True)
job_inputs = self.build_input(input_model, config)
success, dexe = self.execute(job_inputs)
if "There is an error in the input file" in dexe["stdout"]:
raise InputError(dexe["stdout"])
if "not compiled" in dexe["stdout"]:
# recoverable with a different compilation with optional modules
raise InputError(dexe["stdout"])
if success:
dexe["outfiles"]["stdout"] = dexe["stdout"]
dexe["outfiles"]["stderr"] = dexe["stderr"]
return self.parse_output(dexe["outfiles"], input_model)
else:
raise UnknownError(dexe["stdout"])
def build_input(
self, input_model: "AtomicInput", config: "TaskConfig", template: Optional[str] = None
) -> Dict[str, Any]:
# Check some bounds on what cannot be parsed
if "ccsd" in input_model.model.method.lower() or "ccd" in input_model.model.method.lower():
raise InputError("Cannot handle CC* methods currently.")
# Build keywords
keywords = {k.upper(): v for k, v in input_model.keywords.items()}
keywords["INPUT_BOHR"] = "TRUE"
keywords["MEM_TOTAL"] = str(int(config.memory * 1024)) # In MB
if input_model.driver == "energy":
keywords["JOBTYPE"] = "sp"
elif input_model.driver == "gradient":
keywords["JOBTYPE"] = "force"
elif input_model.driver == "hessian":
keywords["JOBTYPE"] = "freq"
else:
raise InputError(f"Driver of type {input_model.driver} is not yet supported.")
if input_model.molecule.fix_com or input_model.molecule.fix_orientation:
def build_input(
self, input_model: "AtomicInput", config: "TaskConfig", template: Optional[str] = None
) -> Dict[str, Any]:
# strip engine hint
mtd = input_model.model.method
if mtd.startswith("mp2d-"):
mtd = mtd[5:]
if input_model.driver.derivative_int() > 1:
raise InputError(f"""MP2D valid driver options are 'energy' and 'gradient', not {input_model.driver}""")
# temp until actual options object
input_model.extras["info"] = empirical_dispersion_resources.from_arrays(
name_hint=mtd,
level_hint=input_model.keywords.get("level_hint", None),
param_tweaks=input_model.keywords.get("params_tweaks", None),
dashcoeff_supplement=input_model.keywords.get("dashcoeff_supplement", None),
)
# Need 'real' field later and that's only guaranteed for molrec
molrec = qcel.molparse.from_schema(input_model.molecule.dict())
xyz = qcel.molparse.to_string(molrec, dtype="xyz", units="Angstrom", ghost_format="")
infiles = {"mp2d_geometry": xyz}
# jobrec['molecule']['real'] = molrec['real']
# env = {
if "ccsd" in input_model.model.method.lower() or "ccd" in input_model.model.method.lower():
raise InputError("Cannot handle CC* methods currently.")
# Build keywords
keywords = {k.upper(): v for k, v in input_model.keywords.items()}
keywords["INPUT_BOHR"] = "TRUE"
keywords["MEM_TOTAL"] = str(int(config.memory * 1024)) # In MB
if input_model.driver == "energy":
keywords["JOBTYPE"] = "sp"
elif input_model.driver == "gradient":
keywords["JOBTYPE"] = "force"
elif input_model.driver == "hessian":
keywords["JOBTYPE"] = "freq"
else:
raise InputError(f"Driver of type {input_model.driver} is not yet supported.")
if input_model.molecule.fix_com or input_model.molecule.fix_orientation:
keywords["SYM_IGNORE"] = "TRUE"
keywords["METHOD"] = input_model.model.method
if input_model.model.basis:
keywords["BASIS"] = input_model.model.basis
# Begin the input file
input_file = []
input_file.append(
f"""$comment
Automatically generated Q-Chem input file by QCEngine
$end
"""
)
input_file = []
input_file.append(
f"""$comment
Automatically generated Q-Chem input file by QCEngine
$end
"""
)
# Add Molecule, TODO: Add to QCElemental
mol = input_model.molecule
input_file.append("$molecule")
input_file.append(f"""{int(mol.molecular_charge)} {mol.molecular_multiplicity}""")
for real, sym, geom in zip(mol.real, mol.symbols, mol.geometry):
if real is False:
raise InputError("Cannot handle ghost atoms yet.")
input_file.append(f"{sym} {geom[0]:14.8f} {geom[1]:14.8f} {geom[2]:14.8f}")
input_file.append("$end\n")
# Write out the keywords
input_file.append("$rem")
for k, v in keywords.items():
input_file.append(f"{k:20s} {v}")
input_file.append("$end\n")
ret = {
"infiles": {"dispatch.in": "\n".join(input_file)},
"commands": [which("qchem"), "-nt", str(config.ncores), "dispatch.in", "dispatch.out"],
"scratch_directory": config.scratch_directory,
}
def compute(self, input_data: "AtomicInput", config: "TaskConfig") -> "AtomicResult":
self.found(raise_error=True)
job_inputs = self.build_input(input_data, config)
success, dexe = self.execute(job_inputs)
if "INPUT HAS AT LEAST ONE SPELLING OR LOGIC MISTAKE" in dexe["stdout"]:
raise InputError(dexe["stdout"])
if success:
dexe["outfiles"]["stdout"] = dexe["stdout"]
dexe["outfiles"]["stderr"] = dexe["stderr"]
return self.parse_output(dexe["outfiles"], input_data)