Merge "Provide a finite machine build() method"

This commit is contained in:
Jenkins 2015-09-16 12:00:25 +00:00 committed by Gerrit Code Review
commit 20c05120fd
3 changed files with 96 additions and 0 deletions

View File

@ -24,6 +24,31 @@ from automaton import _utils as utils
from automaton import exceptions as excp
class State(object):
"""Container that defines needed components of a single state.
Usage of this and the :meth:`~.FiniteMachine.build` make creating finite
state machines that much easier.
:ivar name: The name of the state.
:ivar is_terminal: Whether this state is terminal (or not).
:ivar next_states: Dictionary of 'event' -> 'next state name' (or none).
"""
def __init__(self, name, is_terminal=False, next_states=None):
self.name = name
self.is_terminal = bool(is_terminal)
self.next_states = next_states
def _convert_to_states(state_space):
# NOTE(harlowja): if provided dicts, convert them...
for state in state_space:
if isinstance(state, dict):
state = State(**state)
yield state
def _orderedkeys(data, sort=True):
if sort:
return sorted(six.iterkeys(data))
@ -105,6 +130,26 @@ class FiniteMachine(object):
" undefined state '%s'" % (state))
self._default_start_state = state
@classmethod
def build(cls, state_space):
"""Builds a machine from a state space listing.
Each element of this list must be an instance
of :py:class:`.State` or a ``dict`` with equivalent keys that
can be used to construct a :py:class:`.State` instance.
"""
state_space = list(_convert_to_states(state_space))
m = cls()
for state in state_space:
m.add_state(state.name, terminal=state.is_terminal)
for state in state_space:
if state.next_states:
for event, next_state in six.iteritems(state.next_states):
if isinstance(next_state, State):
next_state = next_state.name
m.add_transition(state.name, next_state, event)
return m
@property
def current_state(self):
"""The current state the machine is in (or none if not initialized)."""

View File

@ -49,6 +49,54 @@ class FSMTest(testcase.TestCase):
self.jumper.add_reaction('up', 'jump', lambda *args: 'fall')
self.jumper.add_reaction('down', 'fall', lambda *args: 'jump')
def test_build(self):
space = []
for a in 'abc':
space.append(machines.State(a))
m = machines.FiniteMachine.build(space)
for a in 'abc':
self.assertIn(a, m)
def test_build_transitions(self):
space = [
machines.State('down', is_terminal=False,
next_states={'jump': 'up'}),
machines.State('up', is_terminal=False,
next_states={'fall': 'down'}),
]
m = machines.FiniteMachine.build(space)
m.default_start_state = 'down'
expected = [('down', 'jump', 'up'), ('up', 'fall', 'down')]
self.assertEqual(expected, list(m))
def test_build_transitions_dct(self):
space = [
{
'name': 'down', 'is_terminal': False,
'next_states': {'jump': 'up'},
},
{
'name': 'up', 'is_terminal': False,
'next_states': {'fall': 'down'},
},
]
m = machines.FiniteMachine.build(space)
m.default_start_state = 'down'
expected = [('down', 'jump', 'up'), ('up', 'fall', 'down')]
self.assertEqual(expected, list(m))
def test_build_terminal(self):
space = [
machines.State('down', is_terminal=False,
next_states={'jump': 'fell_over'}),
machines.State('fell_over', is_terminal=True),
]
m = machines.FiniteMachine.build(space)
m.default_start_state = 'down'
m.initialize()
m.process_event('jump')
self.assertTrue(m.terminated)
def test_actionable(self):
self.jumper.initialize()
self.assertTrue(self.jumper.is_actionable_event('jump'))

View File

@ -6,6 +6,9 @@ API
Machines
--------
.. autoclass:: automaton.machines.State
:members:
.. autoclass:: automaton.machines.FiniteMachine
:members: