"""
A collection of descriptors used to mark-up elements in a workflow, such
as the inputs or outputs of workflow protocols.
"""
import abc
import copy
from enum import Enum
from propertyestimator import unit
from propertyestimator.utils.quantities import EstimatedQuantity
from propertyestimator.workflow.typing import is_instance_of_type, is_supported_type
from propertyestimator.workflow.utils import PlaceholderInput
class UndefinedAttribute:
"""A custom type used to differentiate between ``None`` values,
and an undeclared optional value."""
def __eq__(self, other):
return type(other) == UndefinedAttribute
def __ne__(self, other):
return not self.__eq__(other)
def __getstate__(self):
return {}
def __setstate__(self, state):
return
UNDEFINED = UndefinedAttribute()
class BaseMergeBehaviour(Enum):
"""A base class for enums which will describes how attributes should
be handled when attempting to merge similar protocols.
"""
pass
[docs]class MergeBehaviour(BaseMergeBehaviour):
"""A enum which describes how attributes should be handled when
attempting to merge similar protocols.
This enum may take values of
* ExactlyEqual: This attribute must be exactly equal between two protocols for
them to be able to merge.
"""
ExactlyEqual = 'ExactlyEqual'
[docs]class InequalityMergeBehaviour(BaseMergeBehaviour):
"""A enum which describes how attributes which can be compared
with inequalities should be merged.
This enum may take values of
* SmallestValue: When two protocols are merged, the smallest value of this
attribute from either protocol is retained.
* LargestValue: When two protocols are merged, the largest value of this
attribute from either protocol is retained.
"""
SmallestValue = 'SmallestValue'
LargestValue = 'LargestValue'
[docs]class BaseProtocolAttribute(abc.ABC):
"""A custom descriptor used to mark class attributes as being either
a required input, or provided output of a protocol.
Notes
-----
This decorator expects the protocol to have a matching private field
in addition to the public attribute. For example if a protocol has
an attribute `substance`, by default the protocol must also have a
`_substance` field.
"""
[docs] def __init__(self, docstring, type_hint):
"""Initializes a new BaseProtocolAttribute 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.
"""
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__'):
if type_hint.__qualname__ == 'build_quantity_class.<locals>.Quantity':
typed_docstring = f'Quantity: {docstring}'
elif type_hint.__qualname__ == 'build_quantity_class.<locals>.Unit':
typed_docstring = f'Unit: {docstring}'
else:
typed_docstring = f'{type_hint.__qualname__}: {docstring}'
elif hasattr(type_hint, '__name__'):
typed_docstring = f'{type_hint.__name__}: {docstring}'
else:
typed_docstring = f'{str(type_hint)}: {docstring}'
self.__doc__ = typed_docstring
self.type_hint = type_hint
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
try:
return getattr(instance, self._private_attribute_name)
except AttributeError:
return UNDEFINED
def __set__(self, instance, value):
if (not is_instance_of_type(value, self.type_hint) and
not isinstance(value, PlaceholderInput) 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)
class ProtocolInputAttribute(BaseProtocolAttribute):
"""A descriptor used to mark an attribute of a protocol as
an input to that protocol.
Examples
----------
To mark an attribute as an input:
>>> from propertyestimator.workflow.protocols import BaseProtocol
>>> from propertyestimator.workflow.decorators import protocol_input
>>>
>>> class MyProtocol(BaseProtocol):
>>>
>>> my_input = protocol_input(
>>> docstring='An input will be used.',
>>> type_hint=float,
>>> default_value=0.1
>>> )
"""
def __init__(self, docstring, type_hint, default_value, optional=False,
merge_behavior=MergeBehaviour.ExactlyEqual):
"""Initializes a new protocol_input object.
Parameters
----------
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` must be set to `UNDEFINED`.
merge_behavior: BaseMergeBehaviour
An enum describing how this input should be handled when considering
whether to, and actually merging two different protocols.
"""
docstring = f'**Protocol Input** - {docstring}'
if not isinstance(merge_behavior, BaseMergeBehaviour):
raise ValueError('The merge behaviour must inherit from `BaseMergeBehaviour`')
# Automatically extend the docstrings.
if (isinstance(default_value, (int, float, str, unit.Quantity, EstimatedQuantity, 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}.'
if (merge_behavior == InequalityMergeBehaviour.SmallestValue or
merge_behavior == InequalityMergeBehaviour.LargestValue):
merge_docstring = ''
if merge_behavior == InequalityMergeBehaviour.SmallestValue:
merge_docstring = 'When two protocols are merged, the smallest value of ' \
'this attribute from either protocol is retained.'
if merge_behavior == InequalityMergeBehaviour.SmallestValue:
merge_docstring = 'When two protocols are merged, the largest value of ' \
'this attribute from either protocol is retained.'
docstring = f'{docstring} {merge_docstring}'
if optional is True:
docstring = f'{docstring} This input is *optional*.'
super().__init__(docstring, type_hint)
self.optional = optional
self.merge_behavior = merge_behavior
self._default_value = default_value
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.
setattr(instance, self._private_attribute_name, copy.deepcopy(self._default_value))
return getattr(instance, self._private_attribute_name)
class ProtocolOutputAttribute(BaseProtocolAttribute):
"""A descriptor used to mark an attribute of a protocol as
an output of that protocol.
Examples
----------
To mark an attribute as an output:
>>> from propertyestimator.workflow.protocols import BaseProtocol
>>> from propertyestimator.workflow.decorators import protocol_output,
>>>
>>> class MyProtocol(BaseProtocol):
>>>
>>> my_output = protocol_output(
>>> docstring='An output that will be filled.',
>>> type_hint=float
>>> )
"""
def __init__(self, docstring, type_hint):
"""Initializes a new protocol_output object.
"""
docstring = f'**Protocol Output** - {docstring}'
super().__init__(docstring, type_hint)
protocol_input = ProtocolInputAttribute
protocol_output = ProtocolOutputAttribute