Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from ..config import TaskConfig
from qcelemental.models import AtomicInput
class RDKitHarness(ProgramHarness):
_defaults = {
"name": "RDKit",
"scratch": False,
"thread_safe": True,
"thread_parallel": False,
"node_parallel": False,
"managed_memory": False,
}
class Config(ProgramHarness.Config):
pass
@staticmethod
def _process_molecule_rdkit(jmol):
from rdkit import Chem
# Handle errors
if abs(jmol.molecular_charge) > 1.0e-6:
raise InputError("RDKit does not currently support charged molecules.")
if not jmol.connectivity: # Check for empty list
raise InputError("RDKit requires molecules to have a connectivity graph.")
# Build out the base molecule
base_mol = Chem.Mol()
rw_mol = Chem.RWMol(base_mol)
class OpenMMHarness(ProgramHarness):
_CACHE = {}
_CACHE_MAX_SIZE = 10
_defaults = {
"name": "OpenMM",
"scratch": True,
"thread_safe": True, # true if we use separate `openmm.Context` objects per thread
"thread_parallel": True,
"node_parallel": False,
"managed_memory": True,
}
class Config(ProgramHarness.Config):
pass
def _get_off_forcefield(self, hashstring, offxml):
from openforcefield.typing.engines import smirnoff
key = hashlib.sha256(hashstring.encode()).hexdigest()
# get forcefield from cache, build new one if not present
off_forcefield = self._get_cache(key) if key in self._CACHE else smirnoff.ForceField(offxml)
# cache forcefield, no matter what
# handles updating time touched, dropping items if cache too large
self._cache_it(key, off_forcefield)
return off_forcefield
_dft_keywords_extra: Set[str] = _scf_keywords_extra.copy()
_hf_keywords_extra: Set[str] = _scf_keywords_extra.copy()
_xtb_keywords_extra: Set[str] = {}
# Energy commands that are currently supported and their available keywords
_energy_commands: Dict[str, Any] = {
"dft": _dft_keywords_extra,
"hf": _hf_keywords_extra,
"xtb": _xtb_keywords_extra,
}
# This map order converts entos ordering to CCA ordering
# Entos spherical basis ordering for each angular momentum. Follows reverse order of CCA.
_entos_to_cca_ao_order = {"spherical": get_ao_conversion(cca_ao_order_spherical(10), entos_ao_order_spherical(10))}
class Config(ProgramHarness.Config):
pass
def found(self, raise_error: bool = False) -> bool:
return which(
"entos", return_bool=True, raise_error=raise_error, raise_msg="Please install via https://www.entos.info/"
)
def get_version(self) -> str:
self.found(raise_error=True)
which_prog = which("entos")
if which_prog not in self.version_cache:
with popen([which_prog, "--version"]) as exc:
exc["proc"].wait(timeout=15)
self.version_cache[which_prog] = safe_version(exc["stdout"].split()[2])
-----
* Looks for basis set file ``../basis/GENBAS`` from ``xcfour`` executable. If this doesn't work, file an issue.
"""
_defaults = {
"name": "CFOUR",
"scratch": True,
"thread_safe": False,
"thread_parallel": True,
"node_parallel": False,
"managed_memory": True,
}
version_cache: Dict[str, str] = {}
class Config(ProgramHarness.Config):
pass
@staticmethod
def found(raise_error: bool = False) -> bool:
return which(
"xcfour", return_bool=True, raise_error=raise_error, raise_msg="Please install via http://cfour.de/"
)
def get_version(self) -> str:
self.found(raise_error=True)
which_prog = which("xcfour")
if which_prog not in self.version_cache:
success, output = execute([which_prog, "ZMAT"], {"ZMAT": "\nHe\n\n"})
if success:
class TorchANIHarness(ProgramHarness):
_CACHE = {}
_defaults = {
"name": "TorchANI",
"scratch": False,
"thread_safe": True,
"thread_parallel": False,
"node_parallel": False,
"managed_memory": False,
}
version_cache: Dict[str, str] = {}
class Config(ProgramHarness.Config):
pass
@staticmethod
def found(raise_error: bool = False) -> bool:
return which_import(
"torchani",
return_bool=True,
raise_error=raise_error,
raise_msg="Please install via `pip install torchani`.",
)
def get_version(self) -> str:
self.found(raise_error=True)
which_prog = which_import("torchani")
if which_prog not in self.version_cache:
* To use the TCE, specify ``AtomicInput.model.method`` as usual, then also include ``qc_module = True`` in ``AtomicInput.keywords``.
"""
_defaults = {
"name": "NWChem",
"scratch": True,
"thread_safe": False,
"thread_parallel": False,
"node_parallel": True,
"managed_memory": True,
}
# ATL: OpenMP only >=6.6 and only for Phi; potential for Mac using MKL and Intel compilers
version_cache: Dict[str, str] = {}
class Config(ProgramHarness.Config):
pass
@staticmethod
def found(raise_error: bool = False) -> bool:
"""Whether NWChem harness is ready for operation, with both the QC program and any particular dependencies found.
Parameters
----------
raise_error: bool
Passed on to control negative return between False and ModuleNotFoundError raised.
Returns
-------
bool
If both nwchem and its harness dependency networkx are found, returns True.
If raise_error is False and nwchem or networkx are missing, returns False.
pp = pprint.PrettyPrinter(width=120, compact=True, indent=1)
class DFTD3Harness(ProgramHarness):
_defaults = {
"name": "DFTD3",
"scratch": True,
"thread_safe": True,
"thread_parallel": False,
"node_parallel": False,
"managed_memory": False,
}
version_cache: Dict[str, str] = {}
class Config(ProgramHarness.Config):
pass
@staticmethod
def found(raise_error: bool = False) -> bool:
return which(
"dftd3",
return_bool=True,
raise_error=raise_error,
raise_msg="Please install via `conda install dftd3 -c psi4`.",
)
def get_version(self) -> str:
self.found(raise_error=True)
which_prog = which("dftd3")
if which_prog not in self.version_cache:
from .model import ProgramHarness
class MopacHarness(ProgramHarness):
_defaults = {
"name": "MOPAC",
"scratch": True, # Input/output file
"thread_safe": True,
"thread_parallel": True,
"node_parallel": False,
"managed_memory": True,
}
version_cache: Dict[str, str] = {}
class Config(ProgramHarness.Config):
pass
def __init__(self, **kwargs):
extras = { # All units taken from within MOPAC
"bohr_to_angstroms": 0.5291772083,
"hartree_to_ev": 27.2113834,
"ev_to_kcalmol": 23.060529,
}
extras["au_to_debye"] = 2.99792458e10 * 1.602176462e0 * 1e-10 * extras["bohr_to_angstroms"]
extras["hartree_to_kcalmol"] = extras["hartree_to_ev"] * extras["ev_to_kcalmol"]
kwargs["extras"] = extras
super().__init__(**kwargs)
@staticmethod
def found(raise_error: bool = False) -> bool:
from .model import ProgramHarness
class TeraChemHarness(ProgramHarness):
_defaults = {
"name": "TeraChem",
"scratch": True,
"thread_safe": False,
"thread_parallel": True,
"node_parallel": False,
"managed_memory": True,
}
version_cache: Dict[str, str] = {}
class Config(ProgramHarness.Config):
pass
@staticmethod
def found(raise_error: bool = False) -> bool:
return which(
"terachem",
return_bool=True,
raise_error=raise_error,
raise_msg="Please install via http://www.petachem.com/index.html",
)
def get_version(self) -> str:
self.found(raise_error=True)
which_prog = which("terachem")
if which_prog not in self.version_cache:
from .model import ProgramHarness
class Psi4Harness(ProgramHarness):
_defaults = {
"name": "Psi4",
"scratch": True,
"thread_safe": False,
"thread_parallel": True,
"node_parallel": False,
"managed_memory": True,
}
version_cache: Dict[str, str] = {}
class Config(ProgramHarness.Config):
pass
@staticmethod
def found(raise_error: bool = False) -> bool:
"""Whether Psi4 harness is ready for operation.
Parameters
----------
raise_error: bool
Passed on to control negative return between False and ModuleNotFoundError raised.
Returns
-------
bool
If psi4 (psithon or psiapi) is found, returns True.
If raise_error is False and psi4 is missing, returns False.