concatenator refactor

src/konnektor/network_planners/concatenator/_abstract_network_concatenator.py

@@ -3,10 +3,11 @@
 from typing import Iterable
 
 from gufe import AtomMapper
-from gufe import LigandNetwork, Component
+from gufe import LigandNetwork
 
-from .._networkx_implementations._abstract_network_generator import _AbstractNetworkGenerator
 from ..NetworkPlanner import NetworkPlanner
+from .._networkx_implementations._abstract_network_generator import \
+    _AbstractNetworkGenerator
 
 log = logging.getLogger(__name__)
 
@@ -19,8 +20,9 @@ def __init__(self, mapper: AtomMapper, scorer,
                  network_generator: _AbstractNetworkGenerator,
                  nprocesses: int = 1,
                  _initial_edge_lister=None):
-        """This class is an implementation for the LigandNetworkPlanner interface.
-        It defines the std. class for a Konnektor LigandNetworkPlanner
+        """This class is an implementation for the LigandNetworkPlanner
+        interface. It defines the std. class for a
+        Konnektor LigandNetworkPlanner
 
         Parameters
         ----------
@@ -31,10 +33,13 @@ def __init__(self, mapper: AtomMapper, scorer,
         scorer : AtomMappingScorer
             any callable which takes a AtomMapping and returns a float
         nprocesses: int, optional
-            number of processes that can be used for the network generation. (default: 1)
+            number of processes that can be used for the network generation.
+            (default: 1)
         _initial_edge_lister: LigandNetworkPlanner, optional
-            this LigandNetworkPlanner is used to give the initial set of edges. For standard usage, the Maximal NetworPlanner is used.
-            However in large scale approaches, it might be interesting to use the heuristicMaximalNetworkPlanner. (default: None)
+            this LigandNetworkPlanner is used to give the initial set of edges.
+             For standard usage, the Maximal NetworPlanner is used.
+            However in large scale approaches, it might be interesting to use
+             the heuristicMaximalNetworkPlanner. (default: None)
 
         """
 
@@ -49,14 +54,16 @@ def __init__(self, mapper: AtomMapper, scorer,
         # pass on the parallelization to the edge lister
         # edge lister performs usually the most expensive task!
         # So parallelization is most important here.
-        if self._initial_edge_lister is not None and hasattr(self._initial_edge_lister, "nprocesses"):
+        if self._initial_edge_lister is not None and hasattr(
+                self._initial_edge_lister, "nprocesses"):
             self.nprocesses = nprocesses
 
     def __call__(self, *args, **kwargs) -> LigandNetwork:
         return self.concatenate_networks(*args, **kwargs)
 
     @abc.abstractmethod
-    def concatenate_networks(self, ligand_networks: Iterable[LigandNetwork]) -> LigandNetwork:
+    def concatenate_networks(self, ligand_networks: Iterable[
+        LigandNetwork]) -> LigandNetwork:
         """
 
         Parameters
@@ -69,4 +76,4 @@ def concatenate_networks(self, ligand_networks: Iterable[LigandNetwork]) -> Liga
         LigandNetwork
             returns a concatenated LigandNetwork object, containing all networks.
 
-        """
+        """

src/konnektor/network_planners/concatenator/cyclic_concatenator.py

@@ -1,37 +1,54 @@
 import logging
 from typing import Iterable
 
-
 from gufe import AtomMapper, AtomMappingScorer, LigandNetwork
+
 from ._abstract_network_concatenator import NetworkConcatenator
+from .max_concatenator import MaxConcatenate
 from .._networkx_implementations import MstNetworkGenerator
 
 log = logging.getLogger(__name__)
 
 
 # Todo: check this algorithm again
 
-class MstConcatenate(NetworkConcatenator):
-    def __init__(self, mapper: AtomMapper, scorer: AtomMappingScorer, n_connecting_edges: int = 3,
-                 node_present_in_cycles: int = 2, cycle_sizes: Union[int, List[int]] = 3, nprocesses: int = 1):
+class CyclicConcatenate(NetworkConcatenator):
+    def __init__(self, mapper: AtomMapper, scorer: AtomMappingScorer,
+                 n_connecting_cycles: int = 2,
+                 cycle_sizes: Union[int, List[int]] = 3, nprocesses: int = 1,
+                 _initial_edge_lister: NetworkConcatenator = None):
         """
-        This concatenator is connnecting two Networks with a kruskal like approach up to the number of connecting edges.
+        This concatenator is connnecting two Networks with a kruskal like
+        approach up to the number of connecting edges.
 
         Parameters
         ----------
         mapper: AtomMapper
-            the atom mapper is required, to define the connection between two ligands.
+            the atom mapper is required, to define the connection
+            between two ligands.
         scorer: AtomMappingScorer
-            scoring function evaluating an atom mapping, and giving a score between [0,1].
-        n_connecting_edges: int, optional
-            number of connecting edges. (default: 3)
+            scoring function evaluating an atom mapping, and giving a
+            score between [0,1].
+        n_connecting_cycles: int, optional
+            build at least n cycles between th networks. (default: 2)
+        cycle_sizes: Union[int, List[int]], optional
+            build cycles of given size. (default:3)
         nprocesses: int
-            number of processes that can be used for the network generation. (default: 1)
+            number of processes that can be used for the network generation.
+            (default: 1)
         """
-        super().__init__(mapper=mapper, scorer=scorer, network_generator=MstNetworkGenerator(), nprocesses=nprocesses)
-        self.n_connecting_edges = n_connecting_edges
+        if _initial_edge_lister is None:
+            _initial_edge_lister = MaxConcatenate(mapper=mapper, scorer=scorer,
+                                                  nprocesses=nprocesses)
+
+        super().__init__(mapper=mapper, scorer=scorer,
+                         network_generator=MstNetworkGenerator(),
+                         nprocesses=nprocesses)
+        self.n_connecting_edges = n_connecting_cycles
+        self.cycle_sizes = cycle_sizes
 
-    def concatenate_networks(self, ligand_networks: Iterable[LigandNetwork]) -> LigandNetwork:
+    def concatenate_networks(self, ligand_networks: Iterable[
+        LigandNetwork]) -> LigandNetwork:
         """
 
         Parameters
@@ -44,9 +61,36 @@ def concatenate_networks(self, ligand_networks: Iterable[LigandNetwork]) -> Liga
         Returns
         -------
         LigandNetwork
-            returns a concatenated LigandNetwork object, containing all networks.
+            returns a concatenated LigandNetwork object,
+            containing all networks.
 
         """
         raise NotImplementedError()
-
         # Todo: implement.
+
+        selected_edges = []
+        selected_nodes = []
+        for ligandNetworkA, ligandNetworkB in itertools.combinations(
+                ligand_networks, 2):
+            # Generate fully connected Bipartite Graph
+            ligands = list(ligandNetworkA.nodes) + list(ligandNetworkB.nodes)
+            fully_connected_graph = self._initial_edge_lister(
+                [ligandNetworkA, ligandNetworkB])
+            bipartite_graph_mappings = list(fully_connected_graph.edges)
+
+            # TODO Cycle Selection
+
+            selected_edges.extend(selected_mappings)
+
+        # Constructed final Edges:
+        # Add all old network edges:
+        for network in ligand_networks:
+            selected_edges.extend(network.edges)
+            selected_nodes.extend(network.nodes)
+
+        concat_LigandNetwork = LigandNetwork(edges=selected_edges,
+                                             nodes=set(selected_nodes))
+
+        log.info("Total Concatenated Edges:  " + str(len(selected_edges)))
+
+        return concat_LigandNetwork

src/konnektor/network_planners/concatenator/max_concatenator.py

@@ -0,0 +1,91 @@
+import itertools
+import logging
+from typing import Iterable
+
+from gufe import AtomMapper, LigandNetwork
+
+from ._abstract_network_concatenator import NetworkConcatenator
+from ..generators._parallel_mapping_pattern import _parallel_map_scoring
+
+log = logging.getLogger(__name__)
+
+
+class MaxConcatenate(NetworkConcatenator):
+    def __init__(self, mapper: AtomMapper, scorer, nprocesses: int = 1,
+                 show_progress: bool = False):
+        """
+        This concatenator is connnecting two Networks with all possible
+         mappings. This is usually most useful for initial edge listing.
+
+        Parameters
+        ----------
+        mapper: AtomMapper
+            the atom mapper is required, to define the connection
+             between two ligands.
+        scorer: AtomMappingScorer
+            scoring function evaluating an atom mapping, and giving a
+            score between [0,1].
+        n_connecting_edges: int, optional
+            number of connecting edges. (default: 3)
+        nprocesses: int
+            number of processes that can be used for the network generation.
+            (default: 1)
+        show_progress: bool
+            show progress bar
+        """
+
+        super().__init__(mapper=mapper, scorer=scorer,
+                         network_generator=None,
+                         nprocesses=nprocesses)
+        self.progress = show_progress
+
+    def concatenate_networks(self, ligand_networks: Iterable[
+        LigandNetwork]) -> LigandNetwork:
+        """
+
+        Parameters
+        ----------
+        ligand_networks: Iterable[LigandNetwork]
+            an iterable of ligand networks, that shall be connected.
+
+        Returns
+        -------
+        LigandNetwork
+            returns a concatenated LigandNetwork object, containing all
+             networks and all possible edges, connecting them.
+
+        """
+
+        log.info("Number of edges in individual networks:\n " + str(
+            sum([len(s.edges) for s in ligand_networks])) +
+                 "/ " + str([len(s.edges) for s in ligand_networks]))
+        selected_edges = []
+        selected_nodes = []
+        for ligandNetworkA, ligandNetworkB in itertools.combinations(
+                ligand_networks, 2):
+            # Generate Full Bipartite Graph
+            nodesA = ligandNetworkA.nodes
+            nodesB = ligandNetworkB.nodes
+            pedges = itertools.product(nodesA, nodesB)
+
+            bipartite_graph_mappings = _parallel_map_scoring(
+                possible_edges=pedges,
+                scorer=self.scorer,
+                mapper=self.mapper, n_processes=self.nprocesses,
+                show_progress=self.progress)
+
+            # Add network connecting edges
+            selected_edges.extend(bipartite_graph_mappings)
+
+        # Constructed final Edges:
+        # Add all old network edges:
+        for network in ligand_networks:
+            selected_edges.extend(network.edges)
+            selected_nodes.extend(network.nodes)
+
+        concat_LigandNetwork = LigandNetwork(edges=selected_edges,
+                                             nodes=set(selected_nodes))
+
+        log.info("Total Concatenated Edges:  " + str(len(selected_edges)))
+
+        return concat_LigandNetwork