Skip to content

Using State Machine

API Reference

The world of physical things is often constrained by certain physical conditions. One may need to constrain operations that are allowed when a Thing is said to be in a certain state. For example, a measurement device cannot modify a specific setting if a measurement is ongoing, as this could ruin the measurement. This could be considered as the device being in MEASURING state. Similarly, a device recovering from a fault condition (or FAULT state) requires a specific set of operations. To implement these contraints, a finite state machine may be used to prevent property writes or actions. Events are not supported.

Definition

A StateMachine is a class-level attribute which accepts a list of states and the allowed properties and actions in these states. One can custom define the set of state names:

Definition
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from hololinked.core import Thing, StateMachine, action, Property

class Picoscope(Thing):
    """A PC Oscilloscope from Picotech"""

    state_machine = StateMachine(
        states=["DISCONNECTED", "ON", "FAULT", "ALARM", "MEASURING"],
        initial_state="DISCONNECTED",
    )
    # state_machine is a reserved class attribute

Specify the machine conditions as keyword arguments to the state_machine with properties and actions in a list:

Specify Properties and Actions
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from hololinked.core import Thing, StateMachine, action, Property
from hololinked.core.properties import String


class Picoscope(Thing):
    """A PC Oscilloscope from Picotech"""

    @action()
    def connect(self): ...

    @action()
    def disconnect(self): ...

    @action()
    def start_acquisition(self): ...

    @action()
    def stop_acquisition(self): ...

    serial_number = String()

    state_machine = StateMachine(
        states=["DISCONNECTED", "ON", "FAULT", "ALARM", "MEASURING"],
        initial_state="DISCONNECTED",
        DISCONNECTED=[connect, serial_number],
        ON=[start_acquisition, disconnect],
        MEASURING=[stop_acquisition],
    )
    # state_machine is a reserved class attribute

Set the StateMachine state in properties or actions to indicate state changes using the set_state() method or the assignment operator on state_machine.current_state. Actions or python methods are the most common place to set the state:

set_state()
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
class Picoscope(Thing):
    """A PC Oscilloscope from Picotech"""

    def disconnect(self):
        """add disconnect logic here"""
        self.state_machine.current_state = "DISCONNECTED"
        # self.state_machine.set_state('DISCONNECTED', push_event=True, skip_callbacks=False)

    @action(state=[states.ON])
    def start_acquisition(self):
        """add start measurement logic here"""
        self.state_machine.set_state(states.MEASURING)

One can also specify the allowed state of a property or action directly on the corresponding objects:

Specify State Directly on Object
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
class Picoscope(Thing):
    """A PC Oscilloscope from Picotech"""

    @action(state=[states.MEASURING, states.FAULT, states.ALARM])
    def stop_acquisition(self):
        """add stop measurement logic here"""
        if self.state_machine.state == states.MEASURING:
            self.state_machine.set_state(states.ON)
        # else allow FAULT or ALARM state to persist to inform the user that something is wrong

    serial_number = String(
        state=[states.DISCONNECTED], doc="serial number of the device"
    )  # type: str

State Change Events

State machines also push state change event when the state changes. The state is also an observable property per definition in the base Thing class:

Definition
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
class Picoscope(Thing):
    """A PC Oscilloscope from Picotech"""

    state_machine = StateMachine(
        states=states,
        initial_state=states.DISCONNECTED,
        push_state_change_event=True,
        DISCONNECTED=[connect, serial_number],
        ON=[start_acquisition, disconnect],
        MEASURING=[stop_acquisition],
    )

One can suppress state change events by passing push_event=False when setting the state using the set_state() method:

suppress state change event
self.state_machine.set_state('STATE', push_event=False)

Clients can subscribe to these state change events like any other observable property:

subscription
1
2
3
4
def state_change_cb(event):
    print(f"State changed to {event.data}")

client.observe_property(name="state", callbacks=state_change_cb)

State Change Callbacks

One can also supply callbacks which are executed when entering and exiting certain states, irrespective of where or when the state change occured. The state name and the list of callbacks are supplied as a dictionary to the on_enter and on_exit arguments. These callbacks are executed after the state change is effected, and are mostly useful when there are state changes at multiple places which need to trigger the same side-effects.

enter and exit callbacks
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
class Picoscope(Thing):
    """A PC Oscilloscope from Picotech"""

    def cancel_polling(self):
        """add cancel polling logic here"""
        self._cancel_polling = True

    def polling_loop(self):
        """add polling logic here"""
        self._cancel_polling = False
        while not self._cancel_polling:
            ...

    state_machine = StateMachine(
        states=states,
        initial_state=states.DISCONNECTED,
        push_state_change_event=True,
        DISCONNECTED=[connect, serial_number],
        ON=[start_acquisition, disconnect],
        MEASURING=[stop_acquisition],
        on_enter=dict(DISCONNECTED=[cancel_polling]),
    )