Skip to content

hololinked.core.state_machine.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.

Source code in hololinked/hololinked/core/state_machine.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
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: Enum | str
    """initial state of the machine"""

    states = ClassSelector(
        default=None,
        allow_None=True,
        constant=True,
        class_=(EnumMeta, tuple, list),
        doc="list/enum of allowed states",
    )  # type: EnumMeta | tuple | list
    """list of allowed states"""

    on_enter = TypedDict(
        default=None,
        allow_None=True,
        key_type=str,
        doc="""callbacks to execute when a certain state is entered; 
            specified as map with state as keys and callbacks as list""",
    )  # type: dict[str, list[Callable]]
    """
    callbacks to execute when a certain state is entered; 
    specified as map with state as keys and callbacks as list
    """

    on_exit = TypedDict(
        default=None,
        allow_None=True,
        key_type=str,
        doc="""callbacks to execute when certain state is exited; 
            specified as map with state as keys and callbacks as list""",
    )  # type: dict[str, list[Callable]]
    """
    callbacks to execute when certain state is exited; 
    specified as map with state as keys and callbacks as list
    """

    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: dict[str, list[Callable | Property]]
    """the machine specification with state as key and objects as list"""

    push_state_change_event = Boolean(
        default=True,
        doc="if `True`, when the state changes, an event is pushed with the new state",
    )  # type: bool
    """if `True`, when the state changes, an event is pushed with the new state"""

    valid = Boolean(
        default=False,
        readonly=True,
        fget=lambda self: self._valid,
        doc="internally computed, `True` if states, initial_states and the machine is valid",
    )  # type: bool
    """internally computed, `True` if states, initial_states and the machine is valid"""

    def __init__(
        self,
        states: EnumMeta | list[str] | tuple[str],
        *,
        initial_state: StrEnum | str,
        push_state_change_event: bool = True,
        on_enter: dict[str, list[Callable] | Callable] = None,
        on_exit: dict[str, list[Callable] | Callable] = None,
        **machine: dict[str, Callable | Property],
    ) -> None:
        """
        Parameters
        ----------
        states: EnumMeta | List[str] | Tuple[str]
            enumeration of states
        initial_state: StrEnum | str
            initial state of machine
        push_state_change_event: bool, default `True`
            when the state changes, an event is pushed to clients with the new state as the payload
        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 and the list of functions or methods as values.
        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
            and the list of functions or methods as values.
        **machine:
            the state machine specification with state as key and list of methods or properties as values.

            `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
        self.logger = None

    def __set_name__(self, owner: ThingMeta, name: str) -> None:
        self.name = name
        self.owner = owner

    def validate(self, owner: Thing) -> None:
        """validate the state machine, whether the properties, actions and states are correctly specified"""

        # 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_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.logger = owner.logger.bind(component="state-machine", thing_id=owner.id)
        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: 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) -> 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}."
        )

    def contains_object(self, object: Property | Callable) -> bool:
        """
        Check if specified object is found in any of the state machine states.
        Supply unbound method for checking methods, as state machine is specified at class level
        when the methods are unbound.

        Parameters
        ----------
        object: Property | Callable
            The unbound method or property

        Returns
        -------
        bool
            `True` if the object is found in any of the states, `False` otherwise
        """
        for objects in self.machine.values():
            if object in objects:
                return True
        return False

Attributes

initial_state class-attribute instance-attribute 

initial_state = initial_state

initial state of the machine

machine class-attribute instance-attribute 

machine = machine

the machine specification with state as key and objects as list

on_enter class-attribute instance-attribute 

on_enter = on_enter

callbacks to execute when a certain state is entered; specified as map with state as keys and callbacks as list

on_exit class-attribute instance-attribute 

on_exit = on_exit

callbacks to execute when certain state is exited; specified as map with state as keys and callbacks as list

push_state_change_event class-attribute instance-attribute 

push_state_change_event = push_state_change_event

if True, when the state changes, an event is pushed with the new state

states class-attribute instance-attribute 

states = states

list of allowed states

valid class-attribute instance-attribute 

valid = Boolean(default=False, readonly=True, fget=lambda self: _valid, doc='internally computed, `True` if states, initial_states and the machine is valid')

internally computed, True if states, initial_states and the machine is valid

Functions

__init__ 

__init__(states: EnumMeta | list[str] | tuple[str], *, initial_state: StrEnum | str, push_state_change_event: bool = True, on_enter: dict[str, list[Callable] | Callable] = None, on_exit: dict[str, list[Callable] | Callable] = None, **machine: dict[str, Callable | Property]) -> None

Parameters:

Name Type Description Default

states

EnumMeta | list[str] | tuple[str]

enumeration of states

 required

initial_state

StrEnum | str

initial state of machine

 required

push_state_change_event

bool

when the state changes, an event is pushed to clients with the new state as the payload

 True

on_enter

dict[str, list[Callable] | Callable]

callbacks to be invoked when a certain state is entered. It is to be specified as a dictionary with the states being the keys and the list of functions or methods as values.

 None

on_exit

dict[str, list[Callable] | Callable]

callbacks to be invoked when a certain state is exited. It is to be specified as a dictionary with the states being the keys and the list of functions or methods as values.

 None

**machine

dict[str, Callable | Property]

the state machine specification with state as key and list of methods or properties as values.

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

 {}
Source code in hololinked/hololinked/core/state_machine.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def __init__(
    self,
    states: EnumMeta | list[str] | tuple[str],
    *,
    initial_state: StrEnum | str,
    push_state_change_event: bool = True,
    on_enter: dict[str, list[Callable] | Callable] = None,
    on_exit: dict[str, list[Callable] | Callable] = None,
    **machine: dict[str, Callable | Property],
) -> None:
    """
    Parameters
    ----------
    states: EnumMeta | List[str] | Tuple[str]
        enumeration of states
    initial_state: StrEnum | str
        initial state of machine
    push_state_change_event: bool, default `True`
        when the state changes, an event is pushed to clients with the new state as the payload
    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 and the list of functions or methods as values.
    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
        and the list of functions or methods as values.
    **machine:
        the state machine specification with state as key and list of methods or properties as values.

        `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
    self.logger = None

contains_object 

contains_object(object: Property | Callable) -> bool

Check if specified object is found in any of the state machine states. Supply unbound method for checking methods, as state machine is specified at class level when the methods are unbound.

Parameters:

Name Type Description Default

object

Property | Callable

The unbound method or property

 required

Returns:

Type Description
bool

True if the object is found in any of the states, False otherwise
Source code in hololinked/hololinked/core/state_machine.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
def contains_object(self, object: Property | Callable) -> bool:
    """
    Check if specified object is found in any of the state machine states.
    Supply unbound method for checking methods, as state machine is specified at class level
    when the methods are unbound.

    Parameters
    ----------
    object: Property | Callable
        The unbound method or property

    Returns
    -------
    bool
        `True` if the object is found in any of the states, `False` otherwise
    """
    for objects in self.machine.values():
        if object in objects:
            return True
    return False

validate 

validate(owner: Thing) -> None

validate the state machine, whether the properties, actions and states are correctly specified

Source code in hololinked/hololinked/core/state_machine.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def validate(self, owner: Thing) -> None:
    """validate the state machine, whether the properties, actions and states are correctly specified"""

    # 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_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.logger = owner.logger.bind(component="state-machine", thing_id=owner.id)
    self._valid = True