"""
A collection of descriptors used to add meta data to
object attributes.
"""
import copy
import inspect
from collections.abc import Iterable, Mapping
from enum import Enum, IntEnum, IntFlag
from openff.units import unit
from openff.evaluator.attributes.typing import is_instance_of_type, is_supported_type
from openff.evaluator.utils.serialization import TypedBaseModel
class UndefinedAttribute:
"""A custom type used to differentiate between ``None`` values,
and an undeclared optional value."""
def __eq__(self, other):
return type(other) is UndefinedAttribute
def __ne__(self, other):
return not self.__eq__(other)
def __getstate__(self):
return {}
def __setstate__(self, state):
return
[docs]class PlaceholderValue:
"""A class to act as a place holder for an attribute whose value is
not known a priori, but will be set later by some specialised code.
This may include the input to a protocol which will be set by a
workflow as the output of an executed protocol.
"""
def __getstate__(self):
return {}
def __setstate__(self, state):
pass
UNDEFINED = UndefinedAttribute()
[docs]class AttributeClass(TypedBaseModel):
"""A base class for objects which require well defined
attributes with additional metadata.
"""
[docs] def validate(self, attribute_type=None):
"""Validate the values of the attributes. If `attribute_type`
is set, only attributes of that type will be validated.
Parameters
----------
attribute_type: type of Attribute, optional
The type of attribute to validate.
Raises
------
ValueError or AssertionError
"""
attribute_names = self.get_attributes()
for name in attribute_names:
attribute = getattr(self.__class__, name)
attribute_value = getattr(self, name)
if attribute_type is not None and type(attribute) is not attribute_type:
continue
if not attribute.optional and attribute_value == UNDEFINED:
raise ValueError(f"The required {name} attribute has not been set.")
iterable_values = []
if isinstance(attribute_value, AttributeClass):
attribute_value.validate(attribute_type)
elif isinstance(attribute_value, Mapping):
iterable_values = (
attribute_value[x]
for x in attribute_value
if isinstance(attribute_value[x], AttributeClass)
)
elif isinstance(attribute_value, Iterable) and not isinstance(
attribute_value, (unit.Quantity, unit.Measurement)
):
iterable_values = (
x for x in attribute_value if isinstance(x, AttributeClass)
)
for value in iterable_values:
value.validate()
[docs] @classmethod
def get_attributes(cls, attribute_type=None):
"""Returns all attributes of a specific `attribute_type`.
Parameters
----------
attribute_type: type of Attribute, optional
The type of attribute to search for.
Returns
-------
list of str
The names of the attributes of the specified
type.
"""
all_bases = [base_class for base_class in reversed(inspect.getmro(cls))]
attribute_names = []
for base_class in all_bases:
found_attributes = [
attribute_name
for attribute_name in base_class.__dict__
if isinstance(base_class.__dict__[attribute_name], Attribute)
]
if attribute_type is not None:
found_attributes = [
name
for name in found_attributes
if type(base_class.__dict__[name]) is attribute_type
]
attribute_names.extend(found_attributes)
return attribute_names
def _set_value(self, attribute_name, value):
"""Set the value of an attribute, by-passing
the read-only checks.
Parameters
----------
attribute_name: str
The name of the attribute.
value: Any
The value to set.
"""
attribute = getattr(self.__class__, attribute_name)
# This should handle type checking. We use the private
# set value method rather than setattr to handle read-only
# attributes.
# noinspection PyProtectedMember
attribute._set_value(self, value)
[docs] @classmethod
def parse_json(cls, string_contents):
return_object = super(AttributeClass, cls).parse_json(string_contents)
return_object.validate()
return return_object
def __getstate__(self):
attribute_names = self.get_attributes()
attributes = {}
for attribute_name in attribute_names:
attribute = getattr(self.__class__, attribute_name)
attribute_value = getattr(self, attribute_name)
if attribute.optional and attribute_value == UNDEFINED:
continue
attributes[attribute_name] = attribute_value
return attributes
def __setstate__(self, state):
attribute_names = self.get_attributes()
for name in attribute_names:
attribute = getattr(self.__class__, name)
if not attribute.optional and name not in state:
raise IndexError(
f"The {name} attribute was not present in " f"the state dictionary."
)
elif attribute.optional and name not in state:
state[name] = UNDEFINED
self._set_value(name, state[name])
[docs]class Attribute:
"""A custom descriptor used to add useful metadata to class
attributes.
This decorator expects the object to have a matching private field
in addition to the public attribute. For example if an object has
an attribute `substance`, the object must also have a `_substance`
field.
Notes
-----
The attribute class will automatically create this private
attribute on the object and populate it with the default value.
"""
[docs] def __init__(
self,
docstring,
type_hint,
default_value=UNDEFINED,
optional=False,
read_only=False,
):
"""Initializes a new Attribute object.
Parameters
----------
docstring: str
A docstring describing the attributes purpose. This will automatically
be decorated with additional information such as type hints, default
values, etc.
type_hint: type, typing.Union
The expected type of this attribute. This will be used to help the
workflow engine ensure that expected input types match corresponding
output values.
default_value: Any
The default value for this attribute.
optional: bool
Defines whether this is an optional input of a class. If true,
the `default_value` should be set to `UNDEFINED`.
read_only: bool
Defines whether this attribute is read-only.
"""
if not is_supported_type(type_hint):
raise ValueError(
f"The {type_hint} type is not supported by the "
f"workflow type hinting system."
)
if hasattr(type_hint, "__qualname__"):
docstring = f"{type_hint.__qualname__}: {docstring}"
elif hasattr(type_hint, "__name__"):
docstring = f"{type_hint.__name__}: {docstring}"
else:
docstring = f"{str(type_hint)}: {docstring}"
# Handle the default value.
self._default_value = default_value
if isinstance(
default_value, (int, float, str, unit.Quantity, unit.Measurement, Enum)
) or (
isinstance(default_value, (list, tuple, set, frozenset))
and len(default_value) <= 4
):
docstring = (
f"{docstring} The default value of this attribute "
f"is ``{str(default_value)}``."
)
elif default_value == UNDEFINED:
optional_string = "" if optional else " and must be set by the user."
docstring = (
f"{docstring} The default value of this attribute "
f"is not set{optional_string}."
)
self.optional = optional
self.read_only = read_only
if self.optional and self.read_only:
raise ValueError("An attribute cannot be both optional and read-only")
if optional is True:
docstring = f"{docstring} This attribute is *optional*."
if read_only:
docstring = f"{docstring} This attribute is *read-only*."
self.__doc__ = docstring
self.type_hint = type_hint
def _set_value(self, instance, value):
if (
isinstance(value, int)
and isinstance(self.type_hint, type)
and issubclass(self.type_hint, (IntFlag, IntEnum))
):
# This is necessary as the json library currently doesn't
# support custom serialization of IntFlag, IntEnum.
value = self.type_hint(value)
if (
isinstance(value, list)
and isinstance(self.type_hint, type)
and issubclass(self.type_hint, tuple)
):
# Automate the list -> tuple conversion to get
# around serialization issues.
value = self.type_hint(value)
if (
not is_instance_of_type(value, self.type_hint)
and not isinstance(value, PlaceholderValue)
and not value == UNDEFINED
):
raise ValueError(
f"The {self._private_attribute_name[1:]} attribute can only accept "
f"values of type {self.type_hint}"
)
setattr(instance, self._private_attribute_name, value)
def __set_name__(self, owner, name):
self._private_attribute_name = "_" + name
def __get__(self, instance, owner=None):
if instance is None:
# Handle the case where this is called on the class directly,
# rather than an instance.
return self
if not hasattr(instance, self._private_attribute_name):
# Make sure to only ever pass a copy of the default value to ensure
# mutable values such as lists don't get set by reference.
if not callable(self._default_value):
value = copy.deepcopy(self._default_value)
else:
value = copy.deepcopy(self._default_value())
setattr(
instance,
self._private_attribute_name,
value,
)
return getattr(instance, self._private_attribute_name)
def __set__(self, instance, value):
if self.read_only:
raise ValueError("This attribute is read-only.")
self._set_value(instance, value)