Realize dev features

molpipeline/abstract_pipeline_elements/any2mol/string2mol.py

@@ -66,21 +66,14 @@ def pretransform_single(self, value: str) -> OptionalMol:
             Rdkit molecule if valid string representation, else None.
 
         """
-        if value is None:
-            return InvalidInstance(
-                self.uuid,
-                f"Invalid representation: {value}",
-                self.name,
-            )
-
         if not isinstance(value, str):
             return InvalidInstance(
                 self.uuid,
                 f"Not a string: {value}",
                 self.name,
             )
 
-        mol: RDKitMol = self.string_to_mol(value)
+        mol = self.string_to_mol(value)
 
         if not mol:
             return InvalidInstance(
@@ -92,7 +85,7 @@ def pretransform_single(self, value: str) -> OptionalMol:
         return mol
 
     @abc.abstractmethod
-    def string_to_mol(self, value: str) -> RDKitMol:
+    def string_to_mol(self, value: str) -> RDKitMol | None:
         """Transform string representation to molecule.
 
         Parameters
@@ -102,7 +95,7 @@ def string_to_mol(self, value: str) -> RDKitMol:
 
         Returns
         -------
-        RDKitMol
+        RDKitMol | None
             Rdkit molecule if valid representation, else None.
 
         """

molpipeline/abstract_pipeline_elements/mol2any/mol2bitvector.py

@@ -2,8 +2,15 @@
 
 import abc
 import copy
-from collections.abc import Iterable
-from typing import Any, Literal, Self, TypeAlias, get_args, overload
+from collections.abc import Iterable, Mapping, Sequence
+from typing import (
+    Any,
+    Literal,
+    Self,
+    TypeAlias,
+    get_args,
+    overload,
+)
 
 import numpy as np
 import numpy.typing as npt
@@ -19,7 +26,7 @@
 from molpipeline.abstract_pipeline_elements.core import MolToAnyPipelineElement
 from molpipeline.utils.matrices import sparse_from_index_value_dicts
 from molpipeline.utils.molpipeline_types import RDKitMol
-from molpipeline.utils.substructure_handling import CircularAtomEnvironment
+from molpipeline.utils.substructure_handling import AtomEnvironment
 
 # possible output types for a fingerprint:
 # - "sparse" is a sparse csr_matrix
@@ -74,7 +81,7 @@
 
 
 class MolToFingerprintPipelineElement(MolToAnyPipelineElement, abc.ABC):
-    """Abstract class for PipelineElements which transform molecules to fingerprints."""
+    """Abstract PipelineElement which transform molecules to fingerprints."""
 
     _n_bits: int
     _feature_names: list[str]
@@ -101,9 +108,9 @@ def __init__(
             and parameters.
         name: str
             Name of PipelineElement.
-        n_jobs:
+        n_jobs: int, default=1
             Number of jobs.
-        uuid: Optional[str]
+        uuid: str | None, optional
             Unique identifier.
 
         """
@@ -295,8 +302,42 @@ def pretransform_single(
 class MolToRDKitGenFPElement(MolToFingerprintPipelineElement, abc.ABC):
     """Abstract class for PipelineElements using the FingeprintGenerator64."""
 
+    @property
+    def n_bits(self) -> int:
+        """Get number of bits in (or size of) fingerprint."""
+        return self._n_bits
+
+    @n_bits.setter
+    def n_bits(self, value: int) -> None:
+        """Set number of bits in Morgan fingerprint.
+
+        Parameters
+        ----------
+        value: int
+            Number of bits in Morgan fingerprint.
+
+        Raises
+        ------
+        ValueError
+            If value is not a positive integer.
+
+        """
+        if not isinstance(value, int) or value < 1:
+            raise ValueError(
+                f"Number of bits has to be a integer > 0! (Received: {value})",
+            )
+        self._n_bits = value
+
+    @property
+    def output_type(self) -> str:
+        """Get output type."""
+        if self.counted:
+            return "integer"
+        return "binary"
+
     def __init__(
         self,
+        n_bits: int = 2048,
         counted: bool = False,
         return_as: FPReturnAsOption = "sparse",
         name: str = "MolToRDKitGenFin",
@@ -307,6 +348,8 @@ def __init__(
 
         Parameters
         ----------
+        n_bits: int, default=2048
+            Number of bits in fingerprint.
         counted: bool, default=False
             Whether to count the bits or not.
         return_as: FPReturnAsOption, default="sparse"
@@ -325,6 +368,7 @@ def __init__(
             n_jobs=n_jobs,
             uuid=uuid,
         )
+        self.n_bits = n_bits
         self.counted = counted
 
     @abc.abstractmethod
@@ -417,158 +461,11 @@ def set_params(self, **parameters: Any) -> Self:
         super().set_params(**parameter_dict_copy)
         return self
 
-
-class ABCMorganFingerprintPipelineElement(MolToRDKitGenFPElement, abc.ABC):
-    """Abstract Class for Morgan fingerprints."""
-
-    @property
-    def output_type(self) -> str:
-        """Get output type."""
-        if self.counted:
-            return "integer"
-        return "binary"
-
-    # pylint: disable=R0913
-    def __init__(
-        self,
-        radius: int = 2,
-        use_features: bool = False,
-        counted: bool = False,
-        return_as: FPReturnAsOption = "sparse",
-        name: str = "AbstractMorgan",
-        n_jobs: int = 1,
-        uuid: str | None = None,
-    ) -> None:
-        """Initialize abstract class.
-
-        Parameters
-        ----------
-        radius: int, default=2
-            Radius of fingerprint.
-        use_features: bool, default=False
-            Whether to represent atoms by element or category (donor, acceptor, etc.)
-        counted: bool, default=False
-            Whether to count the bits or not.
-        return_as: FPReturnAsOption, default="sparse"
-            Type of output.
-            When "sparse" the fingerprints will be returned as a scipy.sparse.csr_matrix
-            holding a sparse representation of the bit vectors.
-            With "dense" a numpy matrix will be returned.
-            With "rdkit" the fingerprints will be returned as a list of
-            RDKit's data structure, like ExplicitBitVect, IntSparseBitVect, etc.
-        name: str, default="AbstractMorgan"
-            Name of PipelineElement.
-        n_jobs: int, default=1
-            Number of jobs.
-        uuid: str | None, optional
-            Unique identifier.
-
-        Raises
-        ------
-        ValueError
-            If radius is not a positive integer.
-
-        """
-        # pylint: disable=R0801
-        super().__init__(
-            return_as=return_as,
-            counted=counted,
-            name=name,
-            n_jobs=n_jobs,
-            uuid=uuid,
-        )
-        self._use_features = use_features
-        if isinstance(radius, int) and radius >= 0:
-            self._radius = radius
-        else:
-            raise ValueError(
-                f"Number of bits has to be a positive integer! (Received: {radius})",
-            )
-
-    def get_params(self, deep: bool = True) -> dict[str, Any]:
-        """Get object parameters relevant for copying the class.
-
-        Parameters
-        ----------
-        deep: bool
-            If True get a deep copy of the parameters.
-
-        Returns
-        -------
-        dict[str, Any]
-            Dictionary of parameter names and values.
-
-        """
-        parameters = super().get_params(deep)
-        if deep:
-            parameters["radius"] = copy.copy(self.radius)
-            parameters["use_features"] = copy.copy(self.use_features)
-        else:
-            parameters["radius"] = self.radius
-            parameters["use_features"] = self.use_features
-
-        # remove fill_value from parameters
-        parameters.pop("fill_value", None)
-        return parameters
-
-    def set_params(self, **parameters: Any) -> Self:
-        """Set parameters.
-
-        Parameters
-        ----------
-        parameters: Any
-            Dictionary of parameter names and values.
-
-        Returns
-        -------
-        Self
-            PipelineElement with updated parameters.
-
-        """
-        parameter_copy = dict(parameters)
-        radius = parameter_copy.pop("radius", None)
-        use_features = parameter_copy.pop("use_features", None)
-
-        # explicitly check for None, since 0 is a valid value
-        if radius is not None:
-            self._radius = radius
-        # explicitly check for None, since False is a valid value
-        if use_features is not None:
-            self._use_features = bool(use_features)
-        super().set_params(**parameter_copy)
-        return self
-
-    @property
-    def radius(self) -> int:
-        """Get radius of Morgan fingerprint."""
-        return self._radius
-
-    @property
-    def use_features(self) -> bool:
-        """Get whether to encode atoms by features or not."""
-        return self._use_features
-
     @abc.abstractmethod
-    def _explain_rdmol(self, mol_obj: RDKitMol) -> dict[int, list[tuple[int, int]]]:
-        """Get central atom and radius of all features in molecule.
-
-        Parameters
-        ----------
-        mol_obj: RDKitMol
-            RDKit molecule to be encoded.
-
-        Returns
-        -------
-        dict[int, list[tuple[int, int]]]
-            Dictionary with mapping from bit to atom index and radius.
-
-        """
-        raise NotImplementedError
-
     def bit2atom_mapping(
         self,
         mol_obj: RDKitMol,
-    ) -> dict[int, list[CircularAtomEnvironment]]:
+    ) -> Mapping[int, Sequence[AtomEnvironment]]:
         """Obtain set of atoms for all features.
 
         Parameters
@@ -578,18 +475,9 @@ def bit2atom_mapping(
 
         Returns
         -------
-        dict[int, list[CircularAtomEnvironment]]
-            Dictionary with mapping from bit to encoded AtomEnvironments
+        Mapping[int, Sequence[AtomEnvironment]]
+            Dictionary with mapping from bit to encoded
+            AtomEnvironments
             (which contain atom indices).
 
         """
-        bit2atom_dict = self._explain_rdmol(mol_obj)
-        result_dict: dict[int, list[CircularAtomEnvironment]] = {}
-        # Iterating over all present bits and respective matches
-        for bit, matches in bit2atom_dict.items():  # type: int, list[tuple[int, int]]
-            result_dict[bit] = []
-            for central_atom, radius in matches:  # type: int, int
-                env = CircularAtomEnvironment.from_mol(mol_obj, central_atom, radius)
-                result_dict[bit].append(env)
-        # Transforming default dict to dict
-        return result_dict