import datetime
import inspect
from abc import ABC, abstractmethod
from typing import Set, Sequence
from .action import Action, ActionGenerator
from ..base import Base, Property
[docs]class ActionableProperty(Property):
"""Property that is modified via an :class:`~.Action` with defined, non-equal start and end
times."""
def __init__(self, generator_cls,
cls=None, *, default=inspect.Parameter.empty,
doc=None, readonly=False):
super().__init__(cls=cls, default=default, doc=doc, readonly=readonly)
self.generator_cls = generator_cls
[docs]class Actionable(Base, ABC):
"""Base Actionable type.
Contains the core methods of an actionable sensor/platform type.
Notes
-----
An Actionable is required to have a `timestamp` attribute, in order to validate actions and
act. This is an abstract base class, and not intended for direct use. Attaining a timestamp is
left to the inheriting type.
"""
resolutions: dict = Property(
default=None,
doc="Mapping of each :class:`~.ActionableProperty` of the sensor and "
"corresponding resolutions at which the sensor is able to be tasked.")
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.scheduled_actions = dict() # dictionary of property - action pairs
@property
def _actionable_properties(self):
"""Dictionary of all name - property pairs where the property is an
:class:`~.ActionableProperty` (i.e. it is modified via action)."""
return {_name: _property for _name, _property in type(self).properties.items()
if isinstance(_property, ActionableProperty)}
def _default_action(self, name, property_, timestamp):
"""Returns the default action of the action generator associated with the property
(assumes the property is an :class:`~.ActionableProperty`."""
if self.resolutions and name in self.resolutions.keys():
generator = property_.generator_cls(owner=self,
attribute=name,
start_time=self.timestamp,
end_time=timestamp,
resolution=self.resolutions[name])
else:
generator = property_.generator_cls(owner=self,
attribute=name,
start_time=self.timestamp,
end_time=timestamp)
return generator.default_action
[docs] def actions(self, timestamp: datetime.datetime, start_timestamp: datetime.datetime = None
) -> Set[ActionGenerator]:
"""Method to return a set of action generators available up to a provided timestamp.
A generator is returned for each actionable property that the sensor has.
Parameters
----------
timestamp: datetime.datetime
Time of action finish.
start_timestamp: datetime.datetime, optional
Time of action start.
Returns
-------
: set of :class:`~.ActionGenerator`
Set of action generators, that describe the bounds of each action space.
"""
if not self.validate_timestamp():
self.timestamp = timestamp
if start_timestamp is None:
start_timestamp = self.timestamp
generators = set()
for name, property_ in self._actionable_properties.items():
if self.resolutions and name in self.resolutions.keys():
generators.add(property_.generator_cls(owner=self,
attribute=name,
start_time=start_timestamp,
end_time=timestamp,
resolution=self.resolutions[name]))
else:
generators.add(property_.generator_cls(owner=self,
attribute=name,
start_time=start_timestamp,
end_time=timestamp))
return generators
[docs] def add_actions(self, actions: Sequence[Action]) -> bool:
"""Add actions to the sensor
Parameters
----------
actions: sequence of :class:`~.Action`
Sequence of actions that will be executed in order
Returns
-------
bool
Return True if actions accepted. False if rejected.
Returns neither if timestamp is invalid.
Raises
------
NotImplementedError
If sensor cannot be tasked.
Notes
-----
Base class returns True
"""
if not self.validate_timestamp():
return
if any(action.end_time < self.timestamp for action in actions):
raise ValueError("Cannot schedule an action that ends before the current time.")
if len(actions) > len(self._actionable_properties):
raise ValueError("Cannot schedule more actions than there are actionable properties.")
for name in self._actionable_properties:
for action in actions:
if action.generator.attribute == name:
self.scheduled_actions[name] = action
break
return True
[docs] def act(self, timestamp: datetime.datetime):
"""Carry out actions up to a timestamp.
Parameters
----------
timestamp: datetime.datetime
Carry out actions up to this timestamp.
"""
if not self.validate_timestamp():
self.timestamp = timestamp
return
for name, property_ in self._actionable_properties.items():
value = getattr(self, name)
try:
action = self.scheduled_actions[name]
except KeyError:
action = self._default_action(name, property_, timestamp)
setattr(self, name, action.act(self.timestamp, timestamp, value))
else:
end_time = action.end_time
if end_time < timestamp:
# complete action, remove from schedule
# switch to default, and carry-out default until timestamp
interim_value = action.act(self.timestamp, end_time, value)
# remove scheduled action
self.scheduled_actions.pop(name)
action = self._default_action(name, property_, timestamp)
setattr(self, name, action.act(end_time, timestamp, interim_value))
elif end_time == timestamp:
# complete action and remove from schedule
setattr(self, name, action.act(self.timestamp, timestamp, value))
self.scheduled_actions.pop(name)
else:
# carry-out action to timestamp
setattr(self, name, action.act(self.timestamp, timestamp, value))
self.timestamp = timestamp
[docs] @abstractmethod
def validate_timestamp(self) -> bool:
"""Method to validate the timestamp of the actionable.
Returns
-------
bool
True if timestamp is valid, False otherwise.
"""
raise NotImplementedError