Source code for mordred._base.calculator

from __future__ import print_function

import sys
import warnings
from types import ModuleType
from contextlib import contextmanager
from multiprocessing import cpu_count
from packaging.version import Version as StrictVersion

from .._util import Capture, DummyBar
from ..error import Error, Missing, MultipleFragments, DuplicatedDescriptorName
from .result import Result
from .context import Context
from .descriptor import Descriptor, MissingValueException, is_descriptor_class

from importlib.metadata import version as importlib_version

__version__ = importlib_version("mordredcommunity")

try:
    from tqdm import tqdm
    from .._util import NotebookWrapper
except ImportError:
    tqdm = NotebookWrapper = DummyBar


[docs] class Calculator(object): r"""descriptor calculator. Parameters: descs: see Calculator.register() method ignore_3D: see Calculator.register() method """ __slots__ = ( "_descriptors", "_name_dict", "_explicit_hydrogens", "_kekulizes", "_require_3D", "_cache", "_debug", "_progress_bar", "_config", ) def __setstate__(self, dict): ds = self._descriptors = dict.get("_descriptors", []) self._name_dict = {str(d): d for d in ds} self._explicit_hydrogens = dict.get("_explicit_hydrogens", {True, False}) self._kekulizes = dict.get("_kekulizes", {True, False}) self._require_3D = dict.get("_require_3D", False)
[docs] @classmethod def from_json(cls, obj): """Create Calculator from json descriptor objects. Parameters: obj(list or dict): descriptors to register Returns: Calculator: calculator """ calc = cls() calc.register_json(obj) return calc
[docs] def register_json(self, obj): """Register Descriptors from json descriptor objects. Parameters: obj(list or dict): descriptors to register """ if not isinstance(obj, list): obj = [obj] self.register(Descriptor.from_json(j) for j in obj)
[docs] def to_json(self): """Convert descriptors to json serializable data. Returns: list: descriptors """ return [d.to_json() for d in self.descriptors]
def __reduce_ex__(self, version): return ( self.__class__, (), { "_config": self._config, "_descriptors": self._descriptors, "_explicit_hydrogens": self._explicit_hydrogens, "_kekulizes": self._kekulizes, "_require_3D": self._require_3D, }, ) def __getitem__(self, key): return self._name_dict[key] def __init__(self, descs=None, version=None, ignore_3D=False, config=None): if descs is None: descs = [] if config is None: config = {} self._descriptors = [] self._name_dict = {} self._explicit_hydrogens = set() self._kekulizes = set() self._require_3D = False self._debug = False self._config = config self.register(descs, version=version, ignore_3D=ignore_3D)
[docs] def config(self, **configs): r"""Set global configuration.""" self._config.update(configs)
@property def descriptors(self): r"""All descriptors. you can get/set/delete descriptor. Returns: tuple[Descriptor]: registered descriptors """ return tuple(self._descriptors) @descriptors.setter def descriptors(self, descs): del self.descriptors self.register(descs) @descriptors.deleter def descriptors(self): self._descriptors = [] self._name_dict = {} self._explicit_hydrogens.clear() self._kekulizes.clear() self._require_3D = False def __len__(self): return len(self._descriptors) def _register_one(self, desc, check_only=False, ignore_3D=False): if not isinstance(desc, Descriptor): raise ValueError("{!r} is not descriptor".format(desc)) if ignore_3D and desc.require_3D: return self._explicit_hydrogens.add(bool(desc.explicit_hydrogens)) self._kekulizes.add(bool(desc.kekulize)) self._require_3D |= desc.require_3D for dep in (desc.dependencies() or {}).values(): if isinstance(dep, Descriptor): self._register_one(dep, check_only=True) if not check_only: sdesc = str(desc) old = self._name_dict.get(sdesc) if old is not None: raise DuplicatedDescriptorName(desc, old) self._name_dict[sdesc] = desc self._descriptors.append(desc)
[docs] def register(self, desc, version=None, ignore_3D=False): r"""Register descriptors. Descriptor-like: * Descriptor instance: self * Descriptor class: use Descriptor.preset() method * module: use Descriptor-likes in module * Iterable: use Descriptor-likes in Iterable Parameters: desc(Descriptor-like): descriptors to register version(str): version ignore_3D(bool): ignore 3D descriptors """ if version is None: version = __version__ version = StrictVersion(version) return self._register(desc, version, ignore_3D)
def _register(self, desc, version, ignore_3D): if not hasattr(desc, "__iter__"): if is_descriptor_class(desc): if desc.since > version: return for d in desc.preset(version=version): self._register_one(d, ignore_3D=ignore_3D) elif isinstance(desc, ModuleType): self._register( get_descriptors_in_module(desc), version=version, ignore_3D=ignore_3D, ) else: self._register_one(desc, ignore_3D=ignore_3D) else: for d in desc: self._register(d, version=version, ignore_3D=ignore_3D) def _calculate_one(self, cxt, desc, reset): if desc in self._cache: return self._cache[desc] if reset: cxt.reset() desc._context = cxt cxt.add_stack(desc) if desc.require_connected and desc._context.n_frags != 1: return False, Missing(MultipleFragments(), desc._context.get_stack()) args = {} for name, dep in (desc.dependencies() or {}).items(): if dep is None: args[name] = None else: ok, r = self._calculate_one(cxt, dep, False) if ok: args[name] = r else: return False, r ok = False try: r = desc.calculate(**args) if self._debug: self._check_rtype(desc, r) ok = True except MissingValueException as e: r = Missing(e.error, desc._context.get_stack()) except Exception as e: r = Error(e, desc._context.get_stack()) self._cache[desc] = ok, r return ok, r def _check_rtype(self, desc, result): if desc.rtype is None: return if isinstance(result, Error): return if not isinstance(result, desc.rtype): raise TypeError("{} not match {}".format(result, desc.rtype)) def _calculate(self, cxt): self._cache = {} for desc in self.descriptors: _, r = self._calculate_one(cxt, desc, True) yield r def __call__(self, mol, id=-1): r"""Calculate descriptors. :type mol: rdkit.Chem.Mol :param mol: molecular :type id: int :param id: conformer id :rtype: Result[scalar or Error] :returns: iterator of descriptor and value """ return self._wrap_result(mol, self._calculate(Context.from_calculator(self, mol, id))) def _wrap_result(self, mol, r): return Result(mol, r, self._descriptors) def _serial(self, mols, nmols, quiet, ipynb, id): with self._progress(quiet, nmols, ipynb) as bar: for m in mols: with Capture() as capture: r = self._wrap_result(m, self._calculate(Context.from_calculator(self, m, id))) for e in capture.result: e = e.rstrip() if not e: continue bar.write(e, file=capture.orig) yield r bar.update() @contextmanager def _progress(self, quiet, total, ipynb): args = {"dynamic_ncols": True, "leave": True, "total": total} if quiet: Bar = DummyBar elif ipynb: Bar = NotebookWrapper else: Bar = tqdm try: with Bar(**args) as self._progress_bar: yield self._progress_bar finally: if hasattr(self, "_progress_bar"): del self._progress_bar
[docs] def echo(self, s, file=sys.stdout, end="\n"): """Output message. Parameters: s(str): message to output file(file-like): output to end(str): end mark of message Return: None """ p = getattr(self, "_progress_bar", None) if p is not None: p.write(s, file=file, end="\n") return print(s, file=file, end="\n") # noqa: T003
[docs] def map(self, mols, nproc=None, nmols=None, quiet=False, ipynb=False, id=-1): r"""Calculate descriptors over mols. Parameters: mols(Iterable[rdkit.Mol]): moleculars nproc(int): number of process to use. default: multiprocessing.cpu_count() nmols(int): number of all mols to use in progress-bar. default: mols.__len__() quiet(bool): don't show progress bar. default: False ipynb(bool): use ipython notebook progress bar. default: False id(int): conformer id to use. default: -1. Returns: Iterator[Result[scalar]] """ if nproc is None: nproc = cpu_count() if hasattr(mols, "__len__"): nmols = len(mols) if nproc == 1: return self._serial(mols, nmols=nmols, quiet=quiet, ipynb=ipynb, id=id) else: return self._parallel(mols, nproc, nmols=nmols, quiet=quiet, ipynb=ipynb, id=id)
[docs] def pandas(self, mols, nproc=None, nmols=None, quiet=False, ipynb=False, id=-1): r"""Calculate descriptors over mols. Returns: pandas.DataFrame """ from .pandas_module import MordredDataFrame, Series if isinstance(mols, Series): index = mols.index else: index = None return MordredDataFrame( (list(r) for r in self.map(mols, nproc, nmols, quiet, ipynb, id)), columns=[str(d) for d in self.descriptors], index=index, )
[docs] def get_descriptors_in_module(mdl, submodule=True): r"""Get descriptors in module. Parameters: mdl(module): module to search submodule(bool): search recursively Returns: Iterator[Descriptor] """ __all__ = getattr(mdl, "__all__", None) if __all__ is None: __all__ = dir(mdl) all_values = (getattr(mdl, name) for name in __all__ if name[:1] != "_") if submodule: for v in all_values: if is_descriptor_class(v): yield v if isinstance(v, ModuleType): for v in get_descriptors_in_module(v, submodule=True): yield v else: for v in all_values: if is_descriptor_class(v): yield v