Source code for openff.evaluator.workflow.utils

"""A set of helper classes for manipulating and passing inputs between
buildings blocks in a property estimation workflow.

from openff.evaluator.attributes import PlaceholderValue
from openff.evaluator.utils import graph

[docs]class ReplicatorValue(PlaceholderValue): """A placeholder value which will be set by a protocol replicator with the specified id. """
[docs] def __init__(self, replicator_id=""): """Constructs a new ReplicatorValue object Parameters ---------- replicator_id: str The id of the replicator which will set this value. """ self.replicator_id = replicator_id
def __getstate__(self): return {"replicator_id": self.replicator_id} def __setstate__(self, state): self.replicator_id = state["replicator_id"]
[docs]class ProtocolPath(PlaceholderValue): """Represents a pointer to the output of another protocol.""" # The character which separates protocol ids. path_separator = "/" # The character which separates the property name from the path. property_separator = "." @property def property_name(self): """str: The property name pointed to by the path.""" return self._property_name @property def protocol_ids(self): """tuple of str: The ids of the protocols referenced by this object.""" return self._protocol_ids @property def start_protocol(self): """str: The leading protocol id of the path.""" return None if len(self._protocol_ids) == 0 else self._protocol_ids[0] @property def last_protocol(self): """str: The end protocol id of the path.""" return None if len(self._protocol_ids) == 0 else self._protocol_ids[-1] @property def protocol_path(self): """str: The full path referenced by this object excluding the property name.""" return self._protocol_path @property def full_path(self): """str: The full path referenced by this object.""" return self._full_path @property def is_global(self): return self.start_protocol == "global"
[docs] def __init__(self, property_name="", *protocol_ids): """Constructs a new ProtocolPath object. Parameters ---------- property_name: str The property name referenced by the path. protocol_ids: str An args list of protocol ids in the order in which they will appear in the path. """ if property_name is None: property_name = "" self._property_name = property_name self._protocol_ids = tuple(protocol_ids) self._protocol_path = None self._full_path = None self._update_string_paths()
def _update_string_paths(self): """Combines the property name and protocol ids into string representations and stores them on the object. """ self._protocol_path = "" if len(self._protocol_ids) > 0: self._protocol_path = ProtocolPath.path_separator.join(self._protocol_ids) property_name = "" if self._property_name is None else self._property_name self._full_path = ( f"{self._protocol_path}{ProtocolPath.property_separator}{property_name}" ) @classmethod def from_string(cls, existing_path_string: str): property_name, protocol_ids = ProtocolPath._to_components(existing_path_string) if any(x is None or len(x) == 0 for x in protocol_ids): raise ValueError("An invalid protocol id (either None or empty) was found.") return ProtocolPath(property_name, *protocol_ids) @staticmethod def _to_components(path_string): """Splits a protocol path string into the property name, and the individual protocol ids. Parameters ---------- path_string: str The protocol path to split. Returns ------- str, list of str A tuple of the property name, and a list of the protocol ids in the path. """ path_string = path_string.lstrip().rstrip() property_name_index = path_string.find(ProtocolPath.property_separator) if property_name_index < 0: raise ValueError( f"A protocol path must contain a {ProtocolPath.property_separator} " f"followed by the property name this path represents" ) property_name_index = path_string.find(ProtocolPath.property_separator) property_name = path_string[property_name_index + 1 :] protocol_id_path = path_string[:property_name_index] protocol_ids = protocol_id_path.split(ProtocolPath.path_separator) if len(protocol_id_path) == 0: protocol_ids = tuple() return property_name, protocol_ids
[docs] def prepend_protocol_id(self, id_to_prepend): """Prepend a new protocol id onto the front of the path. Parameters ---------- id_to_prepend: str The protocol id to prepend to the path """ if len(self._protocol_ids) > 0 and self._protocol_ids[0] == id_to_prepend: return self._protocol_ids = (id_to_prepend, *self._protocol_ids) self._update_string_paths()
[docs] def pop_next_in_path(self): """Pops and then returns the leading protocol id from the path. Returns ------- str: The previously leading protocol id. """ if len(self._protocol_ids) == 0: return None next_in_path = self._protocol_ids[0] self._protocol_ids = self._protocol_ids[1:] self._update_string_paths() return next_in_path
[docs] def append_uuid(self, uuid): """Appends a uuid to each of the protocol id's in the path Parameters ---------- uuid: str The uuid to append. """ if self.is_global: # Don't append uuids to global paths. return self._protocol_ids = tuple( graph.append_uuid(x, uuid) for x in self._protocol_ids ) self._update_string_paths()
[docs] def replace_protocol(self, old_id, new_id): """Redirect the input to point at a new protocol. The main use of this method is when merging multiple protocols into one. Parameters ---------- old_id : str The id of the protocol to replace. new_id : str The id of the new protocol to use. """ self._protocol_ids = tuple( new_id if x == old_id else x for x in self._protocol_ids ) self._update_string_paths()
[docs] def copy(self): """Returns a copy of this path.""" return ProtocolPath(self._property_name, *self._protocol_ids)
def __str__(self): return self._full_path def __repr__(self): return f"<ProtocolPath full_path={self._full_path}>" def __hash__(self): """Returns the hash key of this ProtocolPath.""" return hash(self._full_path) def __eq__(self, other): return type(self) == type(other) and self._full_path == other.full_path def __ne__(self, other): return not (self == other) def __getstate__(self): return {"full_path": self._full_path} def __setstate__(self, state): self._property_name, self._protocol_ids = ProtocolPath._to_components( state["full_path"] ) self._update_string_paths()