gufe: Network Planning Stage - make this part ready for new methods.

gufe/__init__.py

@@ -16,12 +16,6 @@
 
 from .chemicalsystem import ChemicalSystem
 
-from .mapping import (
-    ComponentMapping,  # how individual Components relate
-    AtomMapping, AtomMapper,  # more specific to atom based components
-    LigandAtomMapping,
-)
-
 from .settings import Settings
 
 from .protocols import (
@@ -36,7 +30,7 @@
 
 from .transformations import Transformation, NonTransformation
 
-from .network import AlchemicalNetwork
-from .ligandnetwork import LigandNetwork
+from .alchemical_network import AlchemicalNetwork
+from gufe.setup.network_planning.atom_mapping_based.ligandnetwork import LigandNetwork
 
 __version__ = version("gufe")

gufe/protocols/protocol.py

@@ -10,10 +10,10 @@
 from openff.units import Quantity
 import warnings
 
-from ..settings import Settings, SettingsBaseModel
+from ..settings import Settings
 from ..tokenization import GufeTokenizable, GufeKey
 from ..chemicalsystem import ChemicalSystem
-from ..mapping import ComponentMapping
+from gufe.mapping import ComponentMapping
 
 from .protocoldag import ProtocolDAG, ProtocolDAGResult
 from .protocolunit import ProtocolUnit

gufe/setup/alchemical_network_planner.py

@@ -0,0 +1,16 @@
+# This code is part of OpenFE and is licensed under the MIT license.
+# For details, see https://github.com/OpenFreeEnergy/openfe
+
+import abc
+from typing import Iterable
+from .. import AlchemicalNetwork
+
+
+class AlchemicalNetworkPlanner(abc.ABC):
+    """
+    this abstract class defines the interface for the alchemical Network Planners.
+    """
+
+    @abc.abstractmethod
+    def __call__(self, *args, **kwargs) -> AlchemicalNetwork:
+        raise NotImplementedError()

gufe/setup/chemical_system_generator.py

@@ -0,0 +1,25 @@
+# This code is part of OpenFE and is licensed under the MIT license.
+# For details, see https://github.com/OpenFreeEnergy/openfe
+import abc
+from enum import Enum
+
+from typing import Iterable
+from gufe import ChemicalSystem
+
+# Todo: connect to protocols - use this for labels?
+
+class RFEComponentLabels(str, Enum):
+    PROTEIN = "protein"
+    LIGAND = "ligand"
+    SOLVENT = "solvent"
+    COFACTOR = "cofactor"
+
+
+class AbstractChemicalSystemGenerator(abc.ABC):
+    """
+    this abstract class defines the interface for the chemical system generators.
+    """
+
+    @abc.abstractmethod
+    def __call__(self, *args, **kwargs) -> Iterable[ChemicalSystem]:
+        raise NotImplementedError()

gufe/mapping/atom_mapper.py → gufe/setup/network_planning/atom_mapping_based/atom_mapper.py

@@ -4,7 +4,7 @@
 from collections.abc import Iterator
 import gufe
 
-from ..tokenization import GufeTokenizable
+from gufe.tokenization import GufeTokenizable
 from .atom_mapping import AtomMapping
 
 
@@ -27,4 +27,5 @@ def suggest_mappings(self,
         Suggests zero or more :class:`.AtomMapping` objects, which are possible
         atom mappings between two :class:`.Component` objects.
         """
-        ...
+        raise NotImplementedError("This function was not implemented.")
+

gufe/mapping/atom_mapping.py → gufe/setup/network_planning/atom_mapping_based/atom_mapping.py

@@ -6,7 +6,7 @@
 
 
 import gufe
-from .componentmapping import ComponentMapping
+from gufe.setup.network_planning.component_mapping import ComponentMapping
 
 
 class AtomMapping(ComponentMapping, abc.ABC):
@@ -37,22 +37,23 @@ def componentA_to_componentB(self) -> Mapping[int, int]:
         entity in the other component (e.g. the atom disappears), therefore
         resulting in a KeyError on query
         """
-        ...
+        raise NotImplementedError("This function was not implemented.")
 
     @property
     @abc.abstractmethod
     def componentB_to_componentA(self) -> Mapping[int, int]:
         """Similar to A to B, but reversed."""
-        ...
+        raise NotImplementedError("This function was not implemented.")
 
     @property
     @abc.abstractmethod
     def componentA_unique(self) -> Iterable[int]:
         """Indices of atoms in component A that aren't mappable to B"""
-        ...
+        raise NotImplementedError("This function was not implemented.")
 
     @property
     @abc.abstractmethod
     def componentB_unique(self) -> Iterable[int]:
         """Indices of atoms in component B that aren't mappable to A"""
-        ...
+        raise NotImplementedError("This function was not implemented.")
+

gufe/setup/network_planning/atom_mapping_based/atom_mapping_scorer.py

@@ -0,0 +1,43 @@
+# This code is part of kartograf and is licensed under the MIT license.
+# For details, see https://github.com/OpenFreeEnergy/gufe
+
+import abc
+from ....tokenization import GufeTokenizable
+
+from .atom_mapping import AtomMapping
+
+
+class AtomMappingScorer(GufeTokenizable):
+    """A generic class for scoring Atom mappings.
+    this class can be used for example to build graph algorithm based networks.
+
+    Implementations of this class can require an arbitrary and non-standardised
+    number of input arguments to create.
+
+    Implementations of this class provide the :meth:`.get_score` method
+
+    """
+
+    def __call__(self, mapping: AtomMapping) -> float:
+        return self.get_score(mapping)
+
+    @abc.abstractmethod
+    def get_score(self, mapping: AtomMapping) -> float:
+        """ calculate the score for an  :class:`.AtomMapping`
+            the scoring function returns a value between 0 and 1.
+            a value close to 1.0 indicates a small change - good score, a score close to zero indicates a large cost/change - bad score.
+
+        Parameters
+        ----------
+        mapping: AtomMapping
+            the mapping to be scored
+        args
+        kwargs
+
+        Returns
+        -------
+        float
+            a value between [0,1] where zero is a very bad score and one a very good one.
+
+        """
+        raise NotImplementedError("This function was not implemented.")

gufe/mapping/ligandatommapping.py → gufe/setup/network_planning/atom_mapping_based/ligand_atom_mapping.py

@@ -9,8 +9,8 @@
 
 from gufe.components import SmallMoleculeComponent
 from gufe.visualization.mapping_visualization import draw_mapping
-from . import AtomMapping
-from ..tokenization import JSON_HANDLER
+from gufe.mapping import AtomMapping
+from gufe.tokenization import JSON_HANDLER
 
 
 class LigandAtomMapping(AtomMapping):

gufe/ligandnetwork.py → gufe/setup/network_planning/atom_mapping_based/ligandnetwork.py

@@ -9,11 +9,13 @@
 import gufe
 
 from gufe import SmallMoleculeComponent
-from .mapping import LigandAtomMapping
-from .tokenization import GufeTokenizable
+from gufe.tokenization import GufeTokenizable
 
+from gufe.setup.network_planning.network_plan import NetworkPlan
+from gufe.setup.network_planning.atom_mapping_based.ligand_atom_mapping import LigandAtomMapping
 
-class LigandNetwork(GufeTokenizable):
+
+class LigandNetwork(NetworkPlan):
     """A directed graph connecting many ligands according to their atom mapping
 
     Parameters
@@ -48,37 +50,6 @@ def _to_dict(self) -> dict:
     def _from_dict(cls, dct: dict):
         return cls.from_graphml(dct['graphml'])
 
-    @property
-    def graph(self) -> nx.MultiDiGraph:
-        """NetworkX graph for this network
-
-        This graph will have :class:`.ChemicalSystem` objects as nodes and
-        :class:`.Transformation` objects as directed edges
-        """
-        if self._graph is None:
-            graph = nx.MultiDiGraph()
-            # set iterator order depends on PYTHONHASHSEED, sorting ensures
-            # reproducibility
-            for node in sorted(self._nodes):
-                graph.add_node(node)
-            for edge in sorted(self._edges):
-                graph.add_edge(edge.componentA, edge.componentB, object=edge,
-                               **edge.annotations)
-
-            self._graph = nx.freeze(graph)
-
-        return self._graph
-
-    @property
-    def edges(self) -> FrozenSet[LigandAtomMapping]:
-        """A read-only view of the edges of the Network"""
-        return self._edges
-
-    @property
-    def nodes(self) -> FrozenSet[SmallMoleculeComponent]:
-        """A read-only view of the nodes of the Network"""
-        return self._nodes
-
     def _serializable_graph(self) -> nx.Graph:
         """
         Create NetworkX graph with serializable attribute representations.
@@ -307,12 +278,4 @@ def to_rbfe_alchemical_network(
     #         protocol=protocol,
     #         autoname=autoname,
     #         autoname_prefix=autoname_prefix
-    #     )
-
-    def is_connected(self) -> bool:
-        """Are all ligands in the network (indirectly) connected to each other
-
-        A "False" value indicates that either some ligands have no edges or that
-        there are separate networks that do not link to each other.
-        """
-        return nx.is_weakly_connected(self.graph)
+    #     )

gufe/setup/network_planning/component_mapper.py

@@ -0,0 +1,30 @@
+# This code is part of gufe and is licensed under the MIT license.
+# For details, see https://github.com/OpenFreeEnergy/gufe
+import abc
+from collections.abc import Iterator
+import gufe
+
+from gufe.tokenization import GufeTokenizable
+from .component_mapping import ComponentMapping
+
+
+class ComponentMapper(GufeTokenizable):
+    """A class for manufacturing mappings
+
+    Implementations of this class can require an arbitrary and non-standardised
+    number of input arguments to create.
+
+    Implementations of this class provide the :meth:`.suggest_mappings` method
+    """
+
+    @abc.abstractmethod
+    def suggest_mappings(self,
+                         A: gufe.Component,
+                         B: gufe.Component
+                         ) -> Iterator[ComponentMapping]:
+        """Suggests possible mappings between two Components
+
+        Suggests zero or more :class:`.AtomMapping` objects, which are possible
+        atom mappings between two :class:`.Component` objects.
+        """
+        raise NotImplementedError("This function was not implemented.")

gufe/mapping/componentmapping.py → gufe/setup/network_planning/component_mapping.py

@@ -17,6 +17,7 @@ class ComponentMapping(GufeTokenizable, abc.ABC):
     def __init__(self, componentA: gufe.Component, componentB: gufe.Component):
         self._componentA = componentA
         self._componentB = componentB
+        # self.componentA_to_componentB # TODO: is that something we want here, thinking beyond AtomMappings?
 
     def __contains__(self, item: gufe.Component):
         return item == self._componentA or item == self._componentB

gufe/setup/network_planning/component_mapping_scorer.py

@@ -0,0 +1,36 @@
+# This code is part of kartograf and is licensed under the MIT license.
+# For details, see https://github.com/OpenFreeEnergy/gufe
+
+import abc
+from gufe.tokenization import GufeTokenizable
+
+from  .component_mapping import ComponentMapping
+
+class ComponentMappingScorer(GufeTokenizable):
+    """A generic class for scoring Atom mappings.
+    this class can be used for example to build graph algorithm based networks.
+    Implementations of this class can require an arbitrary and non-standardised
+    number of input arguments to create.
+    Implementations of this class provide the :meth:`.get_score` method
+    """
+
+    def __call__(self, mapping: ComponentMapping) -> float:
+        return self.get_score(mapping)
+
+    @abc.abstractmethod
+    def get_score(self, mapping: ComponentMapping) -> float:
+        """ calculate the score for an  :class:`.AtomMapping`
+            the scoring function returns a value between 0 and 1.
+            a value close to 1.0 indicates a small change, a score close to zero indicates a large cost/change.
+        Parameters
+        ----------
+        mapping: AtomMapping
+            the mapping to be scored
+        args
+        kwargs
+        Returns
+        -------
+        float
+            a value between [0,1] where zero is a very bad score and one a very good one.
+        """
+        raise NotImplementedError("This function was not implemented.")