Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from typing import TYPE_CHECKING
from qcelemental.models import AtomicResult, Provenance
from qcelemental.util import which_import
from ..exceptions import InputError
from ..units import ureg
from .model import ProgramHarness
if TYPE_CHECKING:
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 Chemimport re
from typing import Any, Dict, Optional
from qcelemental.models import AtomicResult, FailedOperation
from qcelemental.molparse.regex import DECIMAL, NUMBER
from qcelemental.util import parse_version, safe_version, which
import qcengine.util as uti
from ..exceptions import UnknownError
from ..util import popen
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:from qcengine.config import TaskConfig, get_config
from qcengine.exceptions import UnknownError
from ...exceptions import InputError
from ...util import create_mpi_invocation, execute
from ..model import ProgramHarness
from ..qcvar_identities_resources import build_atomicproperties, build_out
from .germinate import muster_modelchem
from .harvester import harvest
from .keywords import format_keywords
pp = pprint.PrettyPrinter(width=120, compact=True, indent=1)
logger = logging.getLogger(__name__)
class NWChemHarness(ProgramHarness):
"""
Notes
-----
* 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,
}pp = pprint.PrettyPrinter(width=120, compact=True, indent=1)
class MP2DHarness(ProgramHarness):
_defaults = {
"name": "MP2D",
"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(
"mp2d",
return_bool=True,
raise_error=raise_error,
raise_msg="Please install via `conda install mp2d -c psi4`",
)
def get_version(self) -> str:
self.found(raise_error=True)
which_prog = which("mp2d")
if which_prog not in self.version_cache:from typing import TYPE_CHECKING
if TYPE_CHECKING:
from ..config import TaskConfig
from qcelemental.models import AtomicInput
from qcelemental.models import AtomicResult, Provenance
from qcelemental.util import which_import
from ..exceptions import InputError
from ..util import capture_stdout
from .model import ProgramHarness
from .rdkit import RDKitHarness
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):
passfrom typing import Any, Dict, List, Optional, Tuple
import numpy as np
from qcelemental import constants
from qcelemental.models import AtomicInput, AtomicResult, Molecule
from qcelemental.molparse import regex
from qcelemental.util import parse_version, provenance_stamp, safe_version, which
from ..exceptions import InputError, UnknownError
from ..util import disk_files, execute, popen, temporary_directory
from .model import ProgramHarness
NUMBER = r"(?x:" + regex.NUMBER + ")"
class QChemHarness(ProgramHarness):
_defaults: Dict[str, Any] = {
"name": "QChem",
"scratch": True,
"thread_safe": False,
"thread_parallel": True,
"node_parallel": False,
"managed_memory": True,
}
version_cache: Dict[str, str] = {}
def found(self, raise_error: bool = False) -> bool:
return which(
"qchem",
return_bool=True,
raise_error=raise_error,
raise_msg="Please install by visiting the Q-Chem website. Check it's in your PATH with `which qchem'.",class XTBHarness(ProgramHarness):
"""Calculation harness for the extended tight binding (xtb) package."""
_defaults = {
"name": "xtb",
"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:
"""Check for the availability of the Python API of xtb"""
return which_import(
"xtb",
return_bool=True,
raise_error=raise_error,
raise_msg="Please install via `conda install xtb-python -c conda-forge`.",
)
def get_version(self) -> str:
"""Return the currently used version of xtb-python"""
self.found(raise_error=True)