diff --git a/gufe/__init__.py b/gufe/__init__.py index 10e12d57..fc576f0c 100644 --- a/gufe/__init__.py +++ b/gufe/__init__.py @@ -16,11 +16,12 @@ from .chemicalsystem import ChemicalSystem -from .mapping import ( - ComponentMapping, # how individual Components relate - AtomMapping, AtomMapper, # more specific to atom based components - LigandAtomMapping, -) +from .setup.network_planning import (AtomMapping, AtomMapper, + AtomMappingScorer, + LigandAtomMapping, + LigandNetwork) + +from .network import AlchemicalNetwork from .settings import Settings @@ -36,7 +37,5 @@ from .transformations import Transformation, NonTransformation -from .network import AlchemicalNetwork -from .ligandnetwork import LigandNetwork __version__ = version("gufe") diff --git a/gufe/mapping/__init__.py b/gufe/mapping/__init__.py deleted file mode 100644 index bd6be51b..00000000 --- a/gufe/mapping/__init__.py +++ /dev/null @@ -1,7 +0,0 @@ -# This code is part of gufe and is licensed under the MIT license. -# For details, see https://github.com/OpenFreeEnergy/gufe -"""Defining the relationship between different components""" -from .componentmapping import ComponentMapping -from .atom_mapping import AtomMapping -from .atom_mapper import AtomMapper -from .ligandatommapping import LigandAtomMapping diff --git a/gufe/protocols/protocol.py b/gufe/protocols/protocol.py index bfda10c8..d9ac9067 100644 --- a/gufe/protocols/protocol.py +++ b/gufe/protocols/protocol.py @@ -13,7 +13,7 @@ from ..settings import Settings, SettingsBaseModel from ..tokenization import GufeTokenizable, GufeKey from ..chemicalsystem import ChemicalSystem -from ..mapping import ComponentMapping +from gufe.setup.network_planning import ComponentMapping from .protocoldag import ProtocolDAG, ProtocolDAGResult from .protocolunit import ProtocolUnit diff --git a/gufe/setup/__init__.py b/gufe/setup/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/gufe/setup/network_planning/__init__.py b/gufe/setup/network_planning/__init__.py new file mode 100644 index 00000000..16306905 --- /dev/null +++ b/gufe/setup/network_planning/__init__.py @@ -0,0 +1,18 @@ + +# Factory Classes +from .component_mapper import ComponentMapper +from .component_mapping_scorer import ComponentMappingScorer +from .network_planner import NetworkPlanner + +# Result Types +from .component_mapping import ComponentMapping +from .network_plan import NetworkPlan + + +# RBFE Protocol: +from .atom_mapping_based.atom_mapping import AtomMapping +from .atom_mapping_based.atom_mapper import AtomMapper +from .atom_mapping_based.atom_mapping_scorer import AtomMappingScorer + +from .atom_mapping_based.ligand_atom_mapping import LigandAtomMapping +from .atom_mapping_based.ligandnetwork import LigandNetwork \ No newline at end of file diff --git a/gufe/setup/network_planning/atom_mapping_based/__init__.py b/gufe/setup/network_planning/atom_mapping_based/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/gufe/mapping/atom_mapper.py b/gufe/setup/network_planning/atom_mapping_based/atom_mapper.py similarity index 85% rename from gufe/mapping/atom_mapper.py rename to gufe/setup/network_planning/atom_mapping_based/atom_mapper.py index 0fe1c9cd..093ae1eb 100644 --- a/gufe/mapping/atom_mapper.py +++ b/gufe/setup/network_planning/atom_mapping_based/atom_mapper.py @@ -4,11 +4,11 @@ from collections.abc import Iterator import gufe -from ..tokenization import GufeTokenizable +from ..component_mapper import ComponentMapper from .atom_mapping import AtomMapping -class AtomMapper(GufeTokenizable): +class AtomMapper(ComponentMapper): """A class for manufacturing mappings Implementations of this class can require an arbitrary and non-standardised @@ -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.") + diff --git a/gufe/mapping/atom_mapping.py b/gufe/setup/network_planning/atom_mapping_based/atom_mapping.py similarity index 76% rename from gufe/mapping/atom_mapping.py rename to gufe/setup/network_planning/atom_mapping_based/atom_mapping.py index 58c617f3..77edfe2f 100644 --- a/gufe/mapping/atom_mapping.py +++ b/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): @@ -15,16 +15,6 @@ class AtomMapping(ComponentMapping, abc.ABC): """A mapping between two different atom-based Components""" - @property - def componentA(self) -> gufe.Component: - """A copy of the first Component in the mapping""" - return self._componentA - - @property - def componentB(self) -> gufe.Component: - """A copy of the second Component in the mapping""" - return self._componentB - @property @abc.abstractmethod def componentA_to_componentB(self) -> Mapping[int, int]: @@ -37,22 +27,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.") + diff --git a/gufe/setup/network_planning/atom_mapping_based/atom_mapping_scorer.py b/gufe/setup/network_planning/atom_mapping_based/atom_mapping_scorer.py new file mode 100644 index 00000000..d344f5f3 --- /dev/null +++ b/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 ..component_mapping_scorer import ComponentMappingScorer +from .atom_mapping import AtomMapping + + +class AtomMappingScorer(ComponentMappingScorer): + """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.") diff --git a/gufe/mapping/ligandatommapping.py b/gufe/setup/network_planning/atom_mapping_based/ligand_atom_mapping.py similarity index 98% rename from gufe/mapping/ligandatommapping.py rename to gufe/setup/network_planning/atom_mapping_based/ligand_atom_mapping.py index 9a5b198b..aa39ed6a 100644 --- a/gufe/mapping/ligandatommapping.py +++ b/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.setup.network_planning import AtomMapping +from gufe.tokenization import JSON_HANDLER class LigandAtomMapping(AtomMapping): diff --git a/gufe/ligandnetwork.py b/gufe/setup/network_planning/atom_mapping_based/ligandnetwork.py similarity index 86% rename from gufe/ligandnetwork.py rename to gufe/setup/network_planning/atom_mapping_based/ligandnetwork.py index 177aa61e..0dd237a3 100644 --- a/gufe/ligandnetwork.py +++ b/gufe/setup/network_planning/atom_mapping_based/ligandnetwork.py @@ -9,11 +9,12 @@ import gufe from gufe import SmallMoleculeComponent -from .mapping import LigandAtomMapping -from .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 +49,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 +277,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) + # ) \ No newline at end of file diff --git a/gufe/setup/network_planning/component_mapper.py b/gufe/setup/network_planning/component_mapper.py new file mode 100644 index 00000000..86134bd0 --- /dev/null +++ b/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.") diff --git a/gufe/mapping/componentmapping.py b/gufe/setup/network_planning/component_mapping.py similarity index 64% rename from gufe/mapping/componentmapping.py rename to gufe/setup/network_planning/component_mapping.py index 92a098f3..bb3d0557 100644 --- a/gufe/mapping/componentmapping.py +++ b/gufe/setup/network_planning/component_mapping.py @@ -17,6 +17,17 @@ 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 + + @property + def componentA(self) -> gufe.Component: + """A copy of the first Component in the mapping""" + return self._componentA + + @property + def componentB(self) -> gufe.Component: + """A copy of the second Component in the mapping""" + return self._componentB \ No newline at end of file diff --git a/gufe/setup/network_planning/component_mapping_scorer.py b/gufe/setup/network_planning/component_mapping_scorer.py new file mode 100644 index 00000000..86d772c4 --- /dev/null +++ b/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.") diff --git a/gufe/setup/network_planning/network_plan.py b/gufe/setup/network_planning/network_plan.py new file mode 100644 index 00000000..118612c6 --- /dev/null +++ b/gufe/setup/network_planning/network_plan.py @@ -0,0 +1,190 @@ +# This code is part of gufe and is licensed under the MIT license. +# For details, see https://github.com/OpenFreeEnergy/gufe +from __future__ import annotations + +from itertools import chain +import json +import networkx as nx +from typing import FrozenSet, Iterable, Optional + +from gufe import Component +from gufe.tokenization import GufeTokenizable + +from .component_mapping import ComponentMapping + +class NetworkPlan(GufeTokenizable): + """A directed graph connecting many ligands according to their atom mapping + + Parameters + ---------- + edges : Iterable[ComponentMapping] + edges for this network + nodes : Iterable[SmallMoleculeComponent] + nodes for this network + """ + def __init__( + self, + edges: Iterable[ComponentMapping], + nodes: Optional[Iterable[Component]] = None + ): + if nodes is None: + nodes = [] + + self._edges = frozenset(edges) + edge_nodes = set(chain.from_iterable((e.componentA, e.componentB) + for e in edges)) + self._nodes = frozenset(edge_nodes) | frozenset(nodes) + self._graph = None + + @classmethod + def _defaults(cls): + return {} + + def _to_dict(self) -> dict: + return {'graphml': self.to_graphml()} + + @classmethod + 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[ComponentMapping]: + """A read-only view of the edges of the Network""" + return self._edges + + @property + def nodes(self) -> FrozenSet[Component]: + """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. + + This enables us to use easily use different serialization + approaches. + """ + # sorting ensures that we always preserve order in files, so two + # identical networks will show no changes if you diff their + # serialized versions + sorted_nodes = sorted(self.nodes, key=lambda m: (m.smiles, m.name)) + mol_to_label = {mol: f"mol{num}" + for num, mol in enumerate(sorted_nodes)} + + edge_data = sorted([ + ( + mol_to_label[edge.componentA], + mol_to_label[edge.componentB], + ) + for edge in self.edges + ]) + + # from here, we just build the graph + serializable_graph = nx.MultiDiGraph() + for mol, label in mol_to_label.items(): + serializable_graph.add_node(label, + moldict=json.dumps(mol.to_dict(), + sort_keys=True)) + + for molA, molB in edge_data: + serializable_graph.add_edge(molA, molB) + + return serializable_graph + + @classmethod + def _from_serializable_graph(cls, graph: nx.Graph): + """Create network from NetworkX graph with serializable attributes. + + This is the inverse of ``_serializable_graph``. + """ + label_to_mol = {node: Component.from_dict(json.loads(d)) + for node, d in graph.nodes(data='moldict')} + + edges = [ + ComponentMapping(componentA=label_to_mol[node1], + componentB=label_to_mol[node2]) + for node1, node2 in graph.edges(data='mapping') + ] + + return cls(edges=edges, nodes=label_to_mol.values()) + + def to_graphml(self) -> str: + """Return the GraphML string representing this Network + + This is the primary serialization mechanism for this class. + + Returns + ------- + str : + string representing this network in GraphML format + """ + return "\n".join(nx.generate_graphml(self._serializable_graph())) + + @classmethod + def from_graphml(cls, graphml_str: str) -> ComponentMapping: + """Create from a GraphML string. + + Parameters + ---------- + graphml_str : str + GraphML string representation of a :class:`.Network` + + Returns + ------- + LigandNetwork + new network from the GraphML + """ + return cls._from_serializable_graph(nx.parse_graphml(graphml_str)) + + def enlarge_graph(self, *, edges=None, nodes=None) -> ComponentMapping: + """ + Create a new network with the given edges and nodes added + + Parameters + ---------- + edges : Iterable[:class:`.LigandAtomMapping`] + edges to append to this network + nodes : Iterable[:class:`.SmallMoleculeComponent`] + nodes to append to this network + + Returns + ------- + LigandNetwork + a new network adding the given edges and nodes to this network + """ + if edges is None: + edges = set([]) + + if nodes is None: + nodes = set([]) + + return NetworkPlan(self.edges | set(edges), self.nodes | set(nodes)) + + 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) diff --git a/gufe/setup/network_planning/network_planner.py b/gufe/setup/network_planning/network_planner.py new file mode 100644 index 00000000..345eec32 --- /dev/null +++ b/gufe/setup/network_planning/network_planner.py @@ -0,0 +1,47 @@ +import abc +from typing import Iterable + +from ... import SmallMoleculeComponent + +from gufe.tokenization import GufeTokenizable + +from .network_plan import NetworkPlan +from .component_mapper import ComponentMapper +from .component_mapping_scorer import ComponentMappingScorer + +class NetworkPlanner(GufeTokenizable): + """A generic class for calculating :class:`.LigandNetworks`. + 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 __init__(self, mapper: ComponentMapper, scorer:ComponentMappingScorer, *args, **kwargs): + """ Generate a Ligand Network Planner. This class in general needs a mapper and a scorer. + Parameters + ---------- + mapper: AtomMapper + scorer: AtomMappingScorer + args + kwargs + """ + self.mapper = mapper + self.scorer = scorer + + + def __call__(self, *args, **kwargs)-> NetworkPlan: + return self.generate_ligand_network(*args, **kwargs) + + @abc.abstractmethod + def generate_ligand_network(self, ligands: Iterable[SmallMoleculeComponent])->NetworkPlan: + """Plan a Network which connects all ligands with minimal cost + Parameters + ---------- + ligands : Iterable[SmallMoleculeComponent] + the ligands to include in the Network + Returns + ------- + NetworkPlan + A Network, that connects all ligands with each other. + """ + raise NotImplementedError("This function was not implemented.") diff --git a/gufe/tests/test_ligand_network.py b/gufe/tests/test_ligand_network.py index a81a0888..ea25ab49 100644 --- a/gufe/tests/test_ligand_network.py +++ b/gufe/tests/test_ligand_network.py @@ -343,7 +343,7 @@ def test_to_rbfe_alchemical_network( assert compsA.get('protein') == compsB.get('protein') assert compsA.get('cofactor') == compsB.get('cofactor') - assert isinstance(edge.mapping, gufe.ComponentMapping) + assert isinstance(edge.mapping, gufe.setup.network_planning.ComponentMapping) assert edge.mapping in real_molecules_network.edges def test_to_rbfe_alchemical_network_autoname_false( diff --git a/gufe/tests/test_protocol.py b/gufe/tests/test_protocol.py index 3e4dc382..02a85d1d 100644 --- a/gufe/tests/test_protocol.py +++ b/gufe/tests/test_protocol.py @@ -13,7 +13,7 @@ import gufe from gufe.chemicalsystem import ChemicalSystem -from gufe.mapping import ComponentMapping +from gufe.setup.network_planning import ComponentMapping from gufe import settings from gufe.protocols import ( Protocol, diff --git a/gufe/transformations/transformation.py b/gufe/transformations/transformation.py index 58a898be..41f77bd5 100644 --- a/gufe/transformations/transformation.py +++ b/gufe/transformations/transformation.py @@ -10,7 +10,7 @@ from ..chemicalsystem import ChemicalSystem from ..protocols import Protocol, ProtocolDAG, ProtocolResult, ProtocolDAGResult -from ..mapping import ComponentMapping +from gufe.setup.network_planning import ComponentMapping class Transformation(GufeTokenizable):