class StateMachine:
"""
A finite state machine to constrain property and action execution. Each `Thing` class can only have one state machine
instantiated in a reserved class-level attribute named `state_machine`. Other instantiations are not respected.
The `state` attribute defined as a `Thing`'s property reflects the current state of the state machine and
can be subscribed for state change events. When `state_machine` is accessed by a `Thing` instance,
a `BoundFSM` object is returned.
"""
initial_state = ClassSelector(default=None, allow_None=True, constant=True, class_=(Enum, str),
doc="initial state of the machine") # type: typing.Union[Enum, str]
states = ClassSelector(default=None, allow_None=True, constant=True, class_=(EnumMeta, tuple, list),
doc="list/enum of allowed states") # type: typing.Union[EnumMeta, tuple, list]
on_enter = TypedDict(default=None, allow_None=True, key_type=str,
doc="""callbacks to execute when a certain state is entered;
specfied as map with state as keys and callbacks as list""") # type: typing.Dict[str, typing.List[typing.Callable]]
on_exit = TypedDict(default=None, allow_None=True, key_type=str,
doc="""callbacks to execute when certain state is exited;
specfied as map with state as keys and callbacks as list""") # type: typing.Dict[str, typing.List[typing.Callable]]
machine = TypedDict(default=None, allow_None=True, item_type=(list, tuple), key_type=str, # i.e. its like JSON
doc="the machine specification with state as key and objects as list") # type: typing.Dict[str, typing.List[typing.Callable, Property]]
valid = Boolean(default=False, readonly=True, fget=lambda self: self._valid,
doc="internally computed, True if states, initial_states and the machine is valid")
def __init__(self,
states: EnumMeta | typing.List[str] | typing.Tuple[str], *,
initial_state: StrEnum | str,
push_state_change_event : bool = True,
on_enter: typing.Dict[str, typing.List[typing.Callable] | typing.Callable] = None,
on_exit: typing.Dict[str, typing.List[typing.Callable] | typing.Callable] = None,
**machine: typing.Dict[str, typing.Callable | Property]
) -> None:
"""
Parameters
----------
states: EnumMeta | List[str] | Tuple[str]
enumeration of states
initial_state: str
initial state of machine
push_state_change_event : bool, default True
when the state changes, an event is pushed with the new state
on_enter: Dict[str, Callable | Property]
callbacks to be invoked when a certain state is entered. It is to be specified
as a dictionary with the states being the keys
on_exit: Dict[str, Callable | Property]
callbacks to be invoked when a certain state is exited.
It is to be specified as a dictionary with the states being the keys
**machine:
state name: List[Callable, Property]
directly pass the state name as an argument along with the methods/properties which are allowed to execute
in that state
"""
self._valid = False#
self.name = None
self.on_enter = on_enter
self.on_exit = on_exit
# None cannot be passed in, but constant is necessary.
self.states = states
self.initial_state = initial_state
self.machine = machine
self.push_state_change_event = push_state_change_event
def __set_name__(self, owner: ThingMeta, name: str) -> None:
self.name = name
self.owner = owner
def validate(self, owner: Thing) -> None:
# cannot merge this with __set_name__ because descriptor objects are not ready at that time.
# reason - metaclass __init__ is called after __set_name__ of descriptors, therefore the new "proper" desriptor
# registries are available only after that. Until then only the inherited descriptor registries are available,
# which do not correctly account the subclass's objects.
if self.states is None and self.initial_state is None:
self._valid = False
return
elif self.initial_state not in self.states:
raise AttributeError(f"specified initial state {self.initial_state} not in Enum of states {self.states}.")
# owner._state_machine_state = self._get_machine_compliant_state(self.initial_state)
owner_properties = owner.properties.get_descriptors(recreate=True).values()
owner_methods = owner.actions.get_descriptors(recreate=True).values()
if isinstance(self.states, list):
with edit_constant(self.__class__.states): # type: ignore
self.states = tuple(self.states) # freeze the list of states
# first validate machine
for state, objects in self.machine.items():
if state in self:
for resource in objects:
if isinstance(resource, Action):
if resource not in owner_methods:
raise AttributeError("Given object {} for state machine does not belong to class {}".format(
resource, owner))
elif isinstance(resource, Property):
if resource not in owner_properties:
raise AttributeError("Given object {} for state machine does not belong to class {}".format(
resource, owner))
continue # for now
else:
raise AttributeError(f"Object {resource} was not made remotely accessible," +
" use state machine with properties and actions only.")
if resource.execution_info.state is None:
resource.execution_info.state = self._get_machine_compliant_state(state)
else:
resource.execution_info.state = resource._execution_info.state + (self._get_machine_compliant_state(state), )
else:
raise StateMachineError("Given state {} not in allowed states ({})".format(state, self.states.__members__))
# then the callbacks
if self.on_enter is None:
self.on_enter = {}
for state, objects in self.on_enter.items():
if isinstance(objects, list):
self.on_enter[state] = tuple(objects)
elif not isinstance(objects, (list, tuple)):
self.on_enter[state] = (objects, )
for obj in self.on_enter[state]: # type: ignore
if not isinstance(obj, (FunctionType, MethodType)):
raise TypeError(f"on_enter accept only methods. Given type {type(obj)}.")
if self.on_exit is None:
self.on_exit = {}
for state, objects in self.on_exit.items():
if isinstance(objects, list):
self.on_exit[state] = tuple(objects) # type: ignore
elif not isinstance(objects, (list, tuple)):
self.on_exit[state] = (objects, ) # type: ignore
for obj in self.on_exit[state]: # type: ignore
if not isinstance(obj, (FunctionType, MethodType)):
raise TypeError(f"on_enter accept only methods. Given type {type(obj)}.")
self._valid = True
def __get__(self, instance, owner) -> "BoundFSM":
if instance is None:
return self
return BoundFSM(instance, self)
def __set__(self, instance, value) -> None:
raise AttributeError("Cannot set state machine directly. It is a class level attribute and can be defined only once.")
def __contains__(self, state: typing.Union[str, StrEnum]):
if isinstance(self.states, EnumMeta) and state in self.states.__members__:
return True
elif isinstance(self.states, tuple) and state in self.states:
return True
return False
def _get_machine_compliant_state(self, state) -> typing.Union[StrEnum, str]:
"""
In case of not using StrEnum or iterable of str,
this maps the enum of state to the state name.
"""
if isinstance(state, str):
return state
if isinstance(state, Enum):
return state.name
raise TypeError(f"cannot comply state to a string: {state} which is of type {type(state)}. owner - {self.owner}.")