hat.stc - Python statechart library¶
This library provides basic implementation of hierarchical state machine engine. Statechart definition can be provided as structures defined by API or by SCXML definition. Additionally, Graphviz DOT graph can be generated based on state definition.
Statechart definitions¶
Prior to statechart execution, state with all transition definitions are required:
EventName = str
StateName = str
ActionName = str
ConditionName = str
class Transition(typing.NamedTuple):
event: EventName
target: typing.Optional[StateName]
actions: typing.List[ActionName] = []
conditions: typing.List[ConditionName] = []
internal: bool = False
class State(typing.NamedTuple):
name: StateName
children: typing.List['State'] = []
transitions: typing.List[Transition] = []
entries: typing.List[ActionName] = []
exits: typing.List[ActionName] = []
final: bool = False
State is defined by:
name
Unique state identifier.
children
Optional child states. If state has children, first child is considered as its initial state.
transitions
Possible transitions to other states.
entries
Actions executed when state is entered.
exists
Actions executed when state is exited.
final
Is state final.
Transition is defined by:
event
Event identifier. Occurrence of event with this exact identifier can trigger state transition.
target
Destination state identifier. If destination state is not defined, local transition is assumed - state is not changed and transition actions are triggered.
conditions
List of conditions. Transition is triggered only if all provided conditions are met.
internal
Internal transition modifier. Determines whether the source state is exited in transitions whose target state is a descendant of the source state.
Importing SCXML¶
State definitions can be created based on SCXML definitions:
def parse_scxml(scxml: typing.Union[typing.TextIO, pathlib.Path]
) -> typing.List[State]: ...
Notable differences between hat.stc and SCXML standard:
initial child state (in scxml and state tag) should be defined only by setting parent’s initial attribute
transitions without associated event name are not supported
parallel substates are not supported
history pseudo-state is not supported
data model is not supported
external communications is not supported
all actions and conditions are identified by name - arbitrary expressions or executable contents are not supported
transition event identifiers are used as exact event names without support for substring segmentation matching
Running statechart¶
Statechart instance is represented with instance of hat.stc.Statechart class:
Action = typing.Callable[[typing.Optional[Event]], None]
Condition = typing.Callable[[typing.Optional[Event]], bool]
class Event(typing.NamedTuple):
name: EventName
payload: typing.Any = None
class Statechart:
def __init__(self,
states: typing.Iterable[State],
actions: typing.Dict[str, Action],
conditions: typing.Dict[str, Condition] = {}): ...
@property
def state(self) -> typing.Optional[StateName]: ...
def register(self, event: Event): ...
async def run(self): ...
Each instance is initialized with state definitions (first state is considered
initial) and action and condition definitions. Statechart execution is
simulated by calling run coroutine. When this coroutine is called,
statechart will transition to initial state and wait for new event occurrences.
New events are registered with register method which accepts event instances
containing event name and optional event payload. All event registrations are
queued and processed sequentially. Coroutine run continues execution until
statechart transitions to final state. Once final state is reached, run
finishes execution. During statechart execution, actions and conditions
are called based on state changes and associated transitions provided during
initialization. Condition is considered met only if result of calling
condition function is True
.
Visualization¶
hat.stc provides function for generating Graphviz DOT graph definitions based on state definitions (which can be obtained by importing SCXML):
def create_dot_graph(states: typing.Iterable[State]) -> str: ...
Sphinx extension hat.sphinx.scxml can be used for generating statechart definition visualization.
Example¶
![digraph "stc" {
fontname = Helvetica
fontsize = 12
penwidth = 2.0
splines = true
ordering = out
compound = true
overlap = scale
nodesep = 0.3
ranksep = 0.1
node [
shape = plaintext
style = filled
fillcolor = transparent
fontname = Helvetica
fontsize = 12
penwidth = 2.0
]
edge [
fontname = Helvetica
fontsize = 12
]
state_initial [
shape = circle
style = filled
fillcolor = black
fixedsize = true
height = 0.15
label = ""
]
subgraph "cluster_state_0" {
label = <
<table cellborder="0" border="0">
<tr><td>on</td></tr>
<hr/>
<tr><td align="left">entry/ clear</td></tr>
<tr><td align="left">exit/ clear</td></tr>
</table>
>
style = rounded
penwidth = 2.0
state_0_initial [
shape = circle
style = filled
fillcolor = black
fixedsize = true
height = 0.15
label = ""
]
subgraph "cluster_state_0_0" {
label = <
<table cellborder="0" border="0">
<tr><td>operand1</td></tr>
</table>
>
style = rounded
penwidth = 2.0
state_0_0 [
shape=point
style=invis
margin=0
width=0
height=0
fixedsize=true
]
}
subgraph "cluster_state_0_1" {
label = <
<table cellborder="0" border="0">
<tr><td>opEntered</td></tr>
<hr/>
<tr><td align="left">entry/ setOp</td></tr>
<tr><td align="left">exit/ setOp</td></tr>
</table>
>
style = rounded
penwidth = 2.0
state_0_1 [
shape=point
style=invis
margin=0
width=0
height=0
fixedsize=true
]
}
subgraph "cluster_state_0_2" {
label = <
<table cellborder="0" border="0">
<tr><td>operand2</td></tr>
</table>
>
style = rounded
penwidth = 2.0
state_0_2 [
shape=point
style=invis
margin=0
width=0
height=0
fixedsize=true
]
}
subgraph "cluster_state_0_3" {
label = <
<table cellborder="0" border="0">
<tr><td>result</td></tr>
<hr/>
<tr><td align="left">entry/ calculate</td></tr>
<tr><td align="left">exit/ calculate</td></tr>
</table>
>
style = rounded
penwidth = 2.0
state_0_3 [
shape=point
style=invis
margin=0
width=0
height=0
fixedsize=true
]
}
state_0 [
shape=point
style=invis
margin=0
width=0
height=0
fixedsize=true
]
}
subgraph "cluster_state_1" {
label = <
<table cellborder="0" border="0">
<tr><td>off</td></tr>
</table>
>
style = rounded
penwidth = 2.0
state_1 [
shape=point
style=invis
margin=0
width=0
height=0
fixedsize=true
]
}
state_initial -> state_0 [
label = ""
lhead = "cluster_state_0"
ltail = ""
]
state_0 -> state_0 [
label = <
<table cellborder="0" border="0">
<tr><td>C</td></tr>
</table>
>
lhead = ""
ltail = ""
]
state_0 -> state_1 [
label = <
<table cellborder="0" border="0">
<tr><td>OFF</td></tr>
</table>
>
lhead = "cluster_state_1"
ltail = "cluster_state_0"
]
state_0_initial -> state_0_0 [
label = ""
lhead = "cluster_state_0_0"
ltail = ""
]
state_0_0 -> state_0_0 [
label = <
<table cellborder="0" border="0">
<tr><td>number</td></tr>
<hr/>
<tr><td>appendOperand1</td></tr>
</table>
>
lhead = ""
ltail = ""
]
state_0_0 -> state_0_1 [
label = <
<table cellborder="0" border="0">
<tr><td>operator</td></tr>
</table>
>
lhead = "cluster_state_0_1"
ltail = "cluster_state_0_0"
]
state_0_1 -> state_0_2 [
label = <
<table cellborder="0" border="0">
<tr><td>number</td></tr>
<hr/>
<tr><td>setOperand2</td></tr>
</table>
>
lhead = "cluster_state_0_2"
ltail = "cluster_state_0_1"
]
state_0_2 -> state_0_2 [
label = <
<table cellborder="0" border="0">
<tr><td>number</td></tr>
<hr/>
<tr><td>appendOperand2</td></tr>
</table>
>
lhead = ""
ltail = ""
]
state_0_2 -> state_0_3 [
label = <
<table cellborder="0" border="0">
<tr><td>equals</td></tr>
</table>
>
lhead = "cluster_state_0_3"
ltail = "cluster_state_0_2"
]
state_0_3 -> state_0_0 [
label = <
<table cellborder="0" border="0">
<tr><td>number</td></tr>
<hr/>
<tr><td>setOperand1</td></tr>
</table>
>
lhead = "cluster_state_0_0"
ltail = "cluster_state_0_3"
]
state_0_3 -> state_0_1 [
label = <
<table cellborder="0" border="0">
<tr><td>operator</td></tr>
<hr/>
<tr><td>resultAsOperand1</td></tr>
</table>
>
lhead = "cluster_state_0_1"
ltail = "cluster_state_0_3"
]
}](../../_images/graphviz-e57b59b4ff7f548c6a6bc3a1c25e227620aff425.png)
states = stc.parse_scxml(io.StringIO(r"""<?xml version="1.0" encoding="UTF-8"?>
<scxml xmlns="http://www.w3.org/2005/07/scxml" initial="on" version="1.0">
<state id="on" initial="operand1">
<onentry>clear</onentry>
<transition event="C" target="on"/>
<transition event="OFF" target="off"/>
<state id="operand1">
<transition event="number" target="operand1">appendOperand1</transition>
<transition event="operator" target="opEntered"/>
</state>
<state id="opEntered">
<onentry>setOperator</onentry>
<transition event="number" target="operand2">setOperand2</transition>
</state>
<state id="operand2">
<transition event="number" target="operand2">appendOperand2</transition>
<transition event="equals" target="result"/>
</state>
<state id="result">
<onentry>calculate</onentry>
<transition event="number" target="operand1">setOperand1</transition>
<transition event="operator" target="opEntered">resultAsOperand1</transition>
</state>
</state>
<final id="off"/>
</scxml>""")) # NOQA
class Calculator:
def __init__(self):
actions = {'clear': self._act_clear,
'setOperand1': self._act_setOperand1,
'appendOperand1': self._act_appendOperand1,
'setOperand2': self._act_setOperand2,
'appendOperand2': self._act_appendOperand2,
'resultAsOperand1': self._act_resultAsOperand1,
'setOperator': self._act_setOperator,
'calculate': self._act_calculate}
self._operand1 = None
self._operand2 = None
self._operator = None
self._result = None
self._machine = stc.Statechart(states, actions)
@property
def result(self):
return self._result
def push_number(self, number):
self._machine.register(stc.Event('number', number))
def push_operator(self, operator):
self._machine.register(stc.Event('operator', operator))
def push_equals(self):
self._machine.register(stc.Event('equals'))
def push_C(self):
self._machine.register(stc.Event('C'))
def push_OFF(self):
self._machine.register(stc.Event('OFF'))
async def run(self):
await self._machine.run()
def _act_clear(self, evt):
self._operand1 = 0
self._operand2 = 0
self._operator = None
self._result = 0
def _act_setOperand1(self, evt):
self._operand1 = evt.payload
def _act_appendOperand1(self, evt):
self._operand1 = self._operand1 * 10 + evt.payload
def _act_setOperand2(self, evt):
self._operand2 = evt.payload
def _act_appendOperand2(self, evt):
self._operand2 = self._operand2 * 10 + evt.payload
def _act_resultAsOperand1(self, evt):
self._operand1 = self._result
def _act_setOperator(self, evt):
self._operator = evt.payload
def _act_calculate(self, evt):
if self._operator == '+':
self._result = self._operand1 + self._operand2
elif self._operator == '-':
self._result = self._operand1 - self._operand2
elif self._operator == '*':
self._result = self._operand1 * self._operand2
elif self._operator == '/':
self._result = self._operand1 / self._operand2
else:
raise Exception('invalid operator')
calc = Calculator()
calc.push_number(1)
calc.push_number(2)
calc.push_number(3)
calc.push_operator('*')
calc.push_number(2)
calc.push_equals()
calc.push_OFF()
await calc.run()
assert calc.result == 246