openstack/deb-python-taskflow
Code Issues Proposed changes

Merge "Replace internal fsm + table with automaton library"

Browse Source
This commit is contained in:
Jenkins
2015-07-15 17:24:48 +00:00
committed by Gerrit Code Review
parent 4a22dc3d3b 50f710eaee
commit 7896a183d3
8 changed files with 33 additions and 773 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
doc/source/types.rst
View File

@@ -6,11 +6,9 @@ Types
Even though these types **are** made for public consumption and usage
should be encouraged/easily possible it should be noted that these may be
moved out to new libraries at various points in the future (for example
the ``FSM`` code *may* move to its own oslo supported ``automaton`` library
at some point in the future [#f1]_). If you are using these
types **without** using the rest of this library it is **strongly**
encouraged that you be a vocal proponent of getting these made
moved out to new libraries at various points in the future. If you are
using these types **without** using the rest of this library it is
**strongly** encouraged that you be a vocal proponent of getting these made
into *isolated* libraries (as using these types in this manner is not
the expected and/or desired usage).
@@ -24,11 +22,6 @@ Failure
.. automodule:: taskflow.types.failure
FSM
===
.. automodule:: taskflow.types.fsm
Graph
=====
@@ -45,11 +38,6 @@ Sets
.. automodule:: taskflow.types.sets
Table
=====
.. automodule:: taskflow.types.table
Timing
======
@@ -60,5 +48,3 @@ Tree
.. automodule:: taskflow.types.tree
.. [#f1] See: https://review.openstack.org/#/c/141961 for a proposal to
do this.

3
requirements.txt
View File

@@ -37,6 +37,9 @@ monotonic>=0.1 # Apache-2.0
# Used for structured input validation
jsonschema!=2.5.0,<3.0.0,>=2.0.0
# For the state machine we run with
automaton>=0.2.0 # Apache-2.0
# For common utilities
oslo.utils>=1.9.0 # Apache-2.0
oslo.serialization>=1.4.0 # Apache-2.0

16
taskflow/engines/action_engine/runner.py
View File

@@ -14,10 +14,13 @@
# License for the specific language governing permissions and limitations
# under the License.
from automaton import machines
from automaton import runners
from taskflow import logging
from taskflow import states as st
from taskflow.types import failure
from taskflow.types import fsm
# Waiting state timeout (in seconds).
_WAITING_TIMEOUT = 60
@@ -236,7 +239,7 @@ class Runner(object):
watchers['on_exit'] = on_exit
watchers['on_enter'] = on_enter
m = fsm.FSM(_UNDEFINED)
m = machines.FiniteMachine()
m.add_state(_GAME_OVER, **watchers)
m.add_state(_UNDEFINED, **watchers)
m.add_state(st.ANALYZING, **watchers)
@@ -247,6 +250,7 @@ class Runner(object):
m.add_state(st.SUSPENDED, terminal=True, **watchers)
m.add_state(st.WAITING, **watchers)
m.add_state(st.FAILURE, terminal=True, **watchers)
m.default_start_state = _UNDEFINED
m.add_transition(_GAME_OVER, st.REVERTED, _REVERTED)
m.add_transition(_GAME_OVER, st.SUCCESS, _SUCCESS)
@@ -267,12 +271,14 @@ class Runner(object):
m.add_reaction(st.WAITING, _WAIT, wait)
m.freeze()
return (m, memory)
r = runners.FiniteRunner(m)
return (m, r, memory)
def run_iter(self, timeout=None):
"""Runs iteratively using a locally built state machine."""
machine, memory = self.build(timeout=timeout)
for (_prior_state, new_state) in machine.run_iter(_START):
machine, runner, memory = self.build(timeout=timeout)
for (_prior_state, new_state) in runner.run_iter(_START):
# NOTE(harlowja): skip over meta-states.
if new_state not in _META_STATES:
if new_state == st.FAILURE:

23
taskflow/tests/unit/action_engine/test_runner.py
View File

@@ -14,19 +14,18 @@
# License for the specific language governing permissions and limitations
# under the License.
from automaton import exceptions as excp
import six
from taskflow.engines.action_engine import compiler
from taskflow.engines.action_engine import executor
from taskflow.engines.action_engine import runner
from taskflow.engines.action_engine import runtime
from taskflow import exceptions as excp
from taskflow.patterns import linear_flow as lf
from taskflow import states as st
from taskflow import storage
from taskflow import test
from taskflow.tests import utils as test_utils
from taskflow.types import fsm
from taskflow.types import notifier
from taskflow.utils import persistence_utils as pu
@@ -185,9 +184,9 @@ class RunnerBuildTest(test.TestCase, _RunnerTestMixin):
flow.add(*tasks)
rt = self._make_runtime(flow, initial_state=st.RUNNING)
machine, memory = rt.runner.build()
machine, machine_runner, memory = rt.runner.build()
self.assertTrue(rt.runner.runnable())
self.assertRaises(fsm.NotInitialized, machine.process_event, 'poke')
self.assertRaises(excp.NotInitialized, machine.process_event, 'poke')
# Should now be pending...
self.assertEqual(st.PENDING, rt.storage.get_atom_state(tasks[0].name))
@@ -254,10 +253,10 @@ class RunnerBuildTest(test.TestCase, _RunnerTestMixin):
flow.add(*tasks)
rt = self._make_runtime(flow, initial_state=st.RUNNING)
machine, memory = rt.runner.build()
machine, machine_runner, memory = rt.runner.build()
self.assertTrue(rt.runner.runnable())
transitions = list(machine.run_iter('start'))
transitions = list(machine_runner.run_iter('start'))
self.assertEqual((runner._UNDEFINED, st.RESUMING), transitions[0])
self.assertEqual((runner._GAME_OVER, st.SUCCESS), transitions[-1])
self.assertEqual(st.SUCCESS, rt.storage.get_atom_state(tasks[0].name))
@@ -268,10 +267,10 @@ class RunnerBuildTest(test.TestCase, _RunnerTestMixin):
flow.add(*tasks)
rt = self._make_runtime(flow, initial_state=st.RUNNING)
machine, memory = rt.runner.build()
machine, machine_runner, memory = rt.runner.build()
self.assertTrue(rt.runner.runnable())
transitions = list(machine.run_iter('start'))
transitions = list(machine_runner.run_iter('start'))
self.assertEqual((runner._GAME_OVER, st.FAILURE), transitions[-1])
self.assertEqual(1, len(memory.failures))
@@ -281,10 +280,10 @@ class RunnerBuildTest(test.TestCase, _RunnerTestMixin):
flow.add(*tasks)
rt = self._make_runtime(flow, initial_state=st.RUNNING)
machine, memory = rt.runner.build()
machine, machine_runner, memory = rt.runner.build()
self.assertTrue(rt.runner.runnable())
transitions = list(machine.run_iter('start'))
transitions = list(machine_runner.run_iter('start'))
self.assertEqual((runner._GAME_OVER, st.REVERTED), transitions[-1])
self.assertEqual(st.REVERTED, rt.storage.get_atom_state(tasks[0].name))
@@ -295,8 +294,8 @@ class RunnerBuildTest(test.TestCase, _RunnerTestMixin):
flow.add(*tasks)
rt = self._make_runtime(flow, initial_state=st.RUNNING)
machine, memory = rt.runner.build()
transitions = list(machine.run_iter('start'))
machine, machine_runner, memory = rt.runner.build()
transitions = list(machine_runner.run_iter('start'))
occurrences = dict((t, transitions.count(t)) for t in transitions)
self.assertEqual(10, occurrences.get((st.SCHEDULING, st.WAITING)))

216
taskflow/tests/unit/test_types.py
View File

@@ -15,15 +15,11 @@
# under the License.
import networkx as nx
import six
from six.moves import cPickle as pickle
from taskflow import exceptions as excp
from taskflow import test
from taskflow.types import fsm
from taskflow.types import graph
from taskflow.types import sets
from taskflow.types import table
from taskflow.types import tree
@@ -251,218 +247,6 @@ class TreeTest(test.TestCase):
'horse', 'human', 'monkey'], things)
class TableTest(test.TestCase):
def test_create_valid_no_rows(self):
tbl = table.PleasantTable(['Name', 'City', 'State', 'Country'])
self.assertGreater(0, len(tbl.pformat()))
def test_create_valid_rows(self):
tbl = table.PleasantTable(['Name', 'City', 'State', 'Country'])
before_rows = tbl.pformat()
tbl.add_row(["Josh", "San Jose", "CA", "USA"])
after_rows = tbl.pformat()
self.assertGreater(len(before_rows), len(after_rows))
def test_create_invalid_columns(self):
self.assertRaises(ValueError, table.PleasantTable, [])
def test_create_invalid_rows(self):
tbl = table.PleasantTable(['Name', 'City', 'State', 'Country'])
self.assertRaises(ValueError, tbl.add_row, ['a', 'b'])
class FSMTest(test.TestCase):
def setUp(self):
super(FSMTest, self).setUp()
# NOTE(harlowja): this state machine will never stop if run() is used.
self.jumper = fsm.FSM("down")
self.jumper.add_state('up')
self.jumper.add_state('down')
self.jumper.add_transition('down', 'up', 'jump')
self.jumper.add_transition('up', 'down', 'fall')
self.jumper.add_reaction('up', 'jump', lambda *args: 'fall')
self.jumper.add_reaction('down', 'fall', lambda *args: 'jump')
def test_bad_start_state(self):
m = fsm.FSM('unknown')
self.assertRaises(excp.NotFound, m.run, 'unknown')
def test_contains(self):
m = fsm.FSM('unknown')
self.assertNotIn('unknown', m)
m.add_state('unknown')
self.assertIn('unknown', m)
def test_duplicate_state(self):
m = fsm.FSM('unknown')
m.add_state('unknown')
self.assertRaises(excp.Duplicate, m.add_state, 'unknown')
def test_duplicate_reaction(self):
self.assertRaises(
# Currently duplicate reactions are not allowed...
excp.Duplicate,
self.jumper.add_reaction, 'down', 'fall', lambda *args: 'skate')
def test_bad_transition(self):
m = fsm.FSM('unknown')
m.add_state('unknown')
m.add_state('fire')
self.assertRaises(excp.NotFound, m.add_transition,
'unknown', 'something', 'boom')
self.assertRaises(excp.NotFound, m.add_transition,
'something', 'unknown', 'boom')
def test_bad_reaction(self):
m = fsm.FSM('unknown')
m.add_state('unknown')
self.assertRaises(excp.NotFound, m.add_reaction, 'something', 'boom',
lambda *args: 'cough')
def test_run(self):
m = fsm.FSM('down')
m.add_state('down')
m.add_state('up')
m.add_state('broken', terminal=True)
m.add_transition('down', 'up', 'jump')
m.add_transition('up', 'broken', 'hit-wall')
m.add_reaction('up', 'jump', lambda *args: 'hit-wall')
self.assertEqual(['broken', 'down', 'up'], sorted(m.states))
self.assertEqual(2, m.events)
m.initialize()
self.assertEqual('down', m.current_state)
self.assertFalse(m.terminated)
m.run('jump')
self.assertTrue(m.terminated)
self.assertEqual('broken', m.current_state)
self.assertRaises(excp.InvalidState, m.run, 'jump', initialize=False)
def test_on_enter_on_exit(self):
enter_transitions = []
exit_transitions = []
def on_exit(state, event):
exit_transitions.append((state, event))
def on_enter(state, event):
enter_transitions.append((state, event))
m = fsm.FSM('start')
m.add_state('start', on_exit=on_exit)
m.add_state('down', on_enter=on_enter, on_exit=on_exit)
m.add_state('up', on_enter=on_enter, on_exit=on_exit)
m.add_transition('start', 'down', 'beat')
m.add_transition('down', 'up', 'jump')
m.add_transition('up', 'down', 'fall')
m.initialize()
m.process_event('beat')
m.process_event('jump')
m.process_event('fall')
self.assertEqual([('down', 'beat'),
('up', 'jump'), ('down', 'fall')], enter_transitions)
self.assertEqual(
[('start', 'beat'), ('down', 'jump'), ('up', 'fall')],
exit_transitions)
def test_run_iter(self):
up_downs = []
for (old_state, new_state) in self.jumper.run_iter('jump'):
up_downs.append((old_state, new_state))
if len(up_downs) >= 3:
break
self.assertEqual([('down', 'up'), ('up', 'down'), ('down', 'up')],
up_downs)
self.assertFalse(self.jumper.terminated)
self.assertEqual('up', self.jumper.current_state)
self.jumper.process_event('fall')
self.assertEqual('down', self.jumper.current_state)
def test_run_send(self):
up_downs = []
it = self.jumper.run_iter('jump')
while True:
up_downs.append(it.send(None))
if len(up_downs) >= 3:
it.close()
break
self.assertEqual('up', self.jumper.current_state)
self.assertFalse(self.jumper.terminated)
self.assertEqual([('down', 'up'), ('up', 'down'), ('down', 'up')],
up_downs)
self.assertRaises(StopIteration, six.next, it)
def test_run_send_fail(self):
up_downs = []
it = self.jumper.run_iter('jump')
up_downs.append(six.next(it))
self.assertRaises(excp.NotFound, it.send, 'fail')
it.close()
self.assertEqual([('down', 'up')], up_downs)
def test_not_initialized(self):
self.assertRaises(fsm.NotInitialized,
self.jumper.process_event, 'jump')
def test_copy_states(self):
c = fsm.FSM('down')
self.assertEqual(0, len(c.states))
d = c.copy()
c.add_state('up')
c.add_state('down')
self.assertEqual(2, len(c.states))
self.assertEqual(0, len(d.states))
def test_copy_reactions(self):
c = fsm.FSM('down')
d = c.copy()
c.add_state('down')
c.add_state('up')
c.add_reaction('down', 'jump', lambda *args: 'up')
c.add_transition('down', 'up', 'jump')
self.assertEqual(1, c.events)
self.assertEqual(0, d.events)
self.assertNotIn('down', d)
self.assertNotIn('up', d)
self.assertEqual([], list(d))
self.assertEqual([('down', 'jump', 'up')], list(c))
def test_copy_initialized(self):
j = self.jumper.copy()
self.assertIsNone(j.current_state)
for i, transition in enumerate(self.jumper.run_iter('jump')):
if i == 4:
break
self.assertIsNone(j.current_state)
self.assertIsNotNone(self.jumper.current_state)
def test_iter(self):
transitions = list(self.jumper)
self.assertEqual(2, len(transitions))
self.assertIn(('up', 'fall', 'down'), transitions)
self.assertIn(('down', 'jump', 'up'), transitions)
def test_freeze(self):
self.jumper.freeze()
self.assertRaises(fsm.FrozenMachine, self.jumper.add_state, 'test')
self.assertRaises(fsm.FrozenMachine,
self.jumper.add_transition, 'test', 'test', 'test')
self.assertRaises(fsm.FrozenMachine,
self.jumper.add_reaction,
'test', 'test', lambda *args: 'test')
def test_invalid_callbacks(self):
m = fsm.FSM('working')
m.add_state('working')
m.add_state('broken')
self.assertRaises(ValueError, m.add_state, 'b', on_enter=2)
self.assertRaises(ValueError, m.add_state, 'b', on_exit=2)
class OrderedSetTest(test.TestCase):
def test_pickleable(self):

381
taskflow/types/fsm.py
View File

@@ -1,381 +0,0 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import collections
import six
from taskflow import exceptions as excp
from taskflow.types import table
from taskflow.utils import misc
class _Jump(object):
"""A FSM transition tracks this data while jumping."""
def __init__(self, name, on_enter, on_exit):
self.name = name
self.on_enter = on_enter
self.on_exit = on_exit
class FrozenMachine(Exception):
"""Exception raised when a frozen machine is modified."""
def __init__(self):
super(FrozenMachine, self).__init__("Frozen machine can't be modified")
class NotInitialized(excp.TaskFlowException):
"""Error raised when an action is attempted on a not inited machine."""
class FSM(object):
"""A finite state machine.
This state machine can be used to automatically run a given set of
transitions and states in response to events (either from callbacks or from
generator/iterator send() values, see PEP 342). On each triggered event, a
on_enter and on_exit callback can also be provided which will be called to
perform some type of action on leaving a prior state and before entering a
new state.
NOTE(harlowja): reactions will *only* be called when the generator/iterator
from run_iter() does *not* send back a new event (they will always be
called if the run() method is used). This allows for two unique ways (these
ways can also be intermixed) to use this state machine when using
run_iter(); one where *external* events trigger the next state transition
and one where *internal* reaction callbacks trigger the next state
transition. The other way to use this state machine is to skip using run()
or run_iter() completely and use the process_event() method explicitly and
trigger the events via some *external* functionality.
"""
def __init__(self, start_state):
self._transitions = {}
self._states = collections.OrderedDict()
self._start_state = start_state
self._current = None
self.frozen = False
@property
def start_state(self):
return self._start_state
@property
def current_state(self):
"""Return the current state name.
:returns: current state name
:rtype: string
"""
if self._current is not None:
return self._current.name
return None
@property
def terminated(self):
"""Returns whether the state machine is in a terminal state.
:returns: whether the state machine is in
terminal state or not
:rtype: boolean
"""
if self._current is None:
return False
return self._states[self._current.name]['terminal']
@misc.disallow_when_frozen(FrozenMachine)
def add_state(self, state, terminal=False, on_enter=None, on_exit=None):
"""Adds a given state to the state machine.
:param on_enter: callback, if provided will be expected to take
two positional parameters, these being state being
entered and the second parameter is the event that is
being processed that caused the state transition
:param on_exit: callback, if provided will be expected to take
two positional parameters, these being state being
entered and the second parameter is the event that is
being processed that caused the state transition
:param state: state being entered or exited
:type state: string
"""
if state in self._states:
raise excp.Duplicate("State '%s' already defined" % state)
if on_enter is not None:
if not six.callable(on_enter):
raise ValueError("On enter callback must be callable")
if on_exit is not None:
if not six.callable(on_exit):
raise ValueError("On exit callback must be callable")
self._states[state] = {
'terminal': bool(terminal),
'reactions': {},
'on_enter': on_enter,
'on_exit': on_exit,
}
self._transitions[state] = collections.OrderedDict()
@misc.disallow_when_frozen(FrozenMachine)
def add_reaction(self, state, event, reaction, *args, **kwargs):
"""Adds a reaction that may get triggered by the given event & state.
:param state: the last stable state expressed
:type state: string
:param event: event that caused the transition
:param args: non-keyworded arguments
:type args: list
:param kwargs: key-value pair arguments
:type kwargs: dictionary
Reaction callbacks may (depending on how the state machine is ran) be
used after an event is processed (and a transition occurs) to cause
the machine to react to the newly arrived at stable state. The
expected result of a callback is expected to be a
new event that the callback wants the state machine to react to.
This new event may (depending on how the state machine is ran) get
processed (and this process typically repeats) until the state
machine reaches a terminal state.
"""
if state not in self._states:
raise excp.NotFound("Can not add a reaction to event '%s' for an"
" undefined state '%s'" % (event, state))
if not six.callable(reaction):
raise ValueError("Reaction callback must be callable")
if event not in self._states[state]['reactions']:
self._states[state]['reactions'][event] = (reaction, args, kwargs)
else:
raise excp.Duplicate("State '%s' reaction to event '%s'"
" already defined" % (state, event))
@misc.disallow_when_frozen(FrozenMachine)
def add_transition(self, start, end, event):
"""Adds an allowed transition from start -> end for the given event.
:param start: start of the transition
:param end: end of the transition
:param event: event that caused the transition
"""
if start not in self._states:
raise excp.NotFound("Can not add a transition on event '%s' that"
" starts in a undefined state '%s'" % (event,
start))
if end not in self._states:
raise excp.NotFound("Can not add a transition on event '%s' that"
" ends in a undefined state '%s'" % (event,
end))
self._transitions[start][event] = _Jump(end,
self._states[end]['on_enter'],
self._states[start]['on_exit'])
def process_event(self, event):
"""Trigger a state change in response to the provided event.
:param event: event to be processed to cause a potential transition
"""
current = self._current
if current is None:
raise NotInitialized("Can only process events after"
" being initialized (not before)")
if self._states[current.name]['terminal']:
raise excp.InvalidState("Can not transition from terminal"
" state '%s' on event '%s'"
% (current.name, event))
if event not in self._transitions[current.name]:
raise excp.NotFound("Can not transition from state '%s' on"
" event '%s' (no defined transition)"
% (current.name, event))
replacement = self._transitions[current.name][event]
if current.on_exit is not None:
current.on_exit(current.name, event)
if replacement.on_enter is not None:
replacement.on_enter(replacement.name, event)
self._current = replacement
return (
self._states[replacement.name]['reactions'].get(event),
self._states[replacement.name]['terminal'],
)
def initialize(self):
"""Sets up the state machine (sets current state to start state...)."""
if self._start_state not in self._states:
raise excp.NotFound("Can not start from a undefined"
" state '%s'" % (self._start_state))
if self._states[self._start_state]['terminal']:
raise excp.InvalidState("Can not start from a terminal"
" state '%s'" % (self._start_state))
# No on enter will be called, since we are priming the state machine
# and have not really transitioned from anything to get here, we will
# though allow 'on_exit' to be called on the event that causes this
# to be moved from...
self._current = _Jump(self._start_state, None,
self._states[self._start_state]['on_exit'])
def run(self, event, initialize=True):
"""Runs the state machine, using reactions only."""
for _transition in self.run_iter(event, initialize=initialize):
pass
def copy(self):
"""Copies the current state machine.
NOTE(harlowja): the copy will be left in an *uninitialized* state.
"""
c = FSM(self.start_state)
c.frozen = self.frozen
for state, data in six.iteritems(self._states):
copied_data = data.copy()
copied_data['reactions'] = copied_data['reactions'].copy()
c._states[state] = copied_data
for state, data in six.iteritems(self._transitions):
c._transitions[state] = data.copy()
return c
def run_iter(self, event, initialize=True):
"""Returns a iterator/generator that will run the state machine.
NOTE(harlowja): only one runner iterator/generator should be active for
a machine, if this is not observed then it is possible for
initialization and other local state to be corrupted and cause issues
when running...
"""
if initialize:
self.initialize()
while True:
old_state = self.current_state
reaction, terminal = self.process_event(event)
new_state = self.current_state
try:
sent_event = yield (old_state, new_state)
except GeneratorExit:
break
if terminal:
break
if reaction is None and sent_event is None:
raise excp.NotFound("Unable to progress since no reaction (or"
" sent event) has been made available in"
" new state '%s' (moved to from state '%s'"
" in response to event '%s')"
% (new_state, old_state, event))
elif sent_event is not None:
event = sent_event
else:
cb, args, kwargs = reaction
event = cb(old_state, new_state, event, *args, **kwargs)
def __contains__(self, state):
"""Returns if this state exists in the machines known states.
:param state: input state
:type state: string
:returns: checks whether the state exists in the machine
known states
:rtype: boolean
"""
return state in self._states
def freeze(self):
"""Freezes & stops addition of states, transitions, reactions..."""
self.frozen = True
@property
def states(self):
"""Returns the state names."""
return list(six.iterkeys(self._states))
@property
def events(self):
"""Returns how many events exist.
:returns: how many events exist
:rtype: number
"""
c = 0
for state in six.iterkeys(self._states):
c += len(self._transitions[state])
return c
def __iter__(self):
"""Iterates over (start, event, end) transition tuples."""
for state in six.iterkeys(self._states):
for event, target in six.iteritems(self._transitions[state]):
yield (state, event, target.name)
def pformat(self, sort=True):
"""Pretty formats the state + transition table into a string.
NOTE(harlowja): the sort parameter can be provided to sort the states
and transitions by sort order; with it being provided as false the rows
will be iterated in addition order instead.
**Example**::
>>> from taskflow.types import fsm
>>> f = fsm.FSM("sits")
>>> f.add_state("sits")
>>> f.add_state("barks")
>>> f.add_state("wags tail")
>>> f.add_transition("sits", "barks", "squirrel!")
>>> f.add_transition("barks", "wags tail", "gets petted")
>>> f.add_transition("wags tail", "sits", "gets petted")
>>> f.add_transition("wags tail", "barks", "squirrel!")
>>> print(f.pformat())
+-----------+-------------+-----------+----------+---------+
Start | Event | End | On Enter | On Exit
+-----------+-------------+-----------+----------+---------+
barks | gets petted | wags tail | |
sits[^] | squirrel! | barks | |
wags tail | gets petted | sits | |
wags tail | squirrel! | barks | |
+-----------+-------------+-----------+----------+---------+
"""
def orderedkeys(data):
if sort:
return sorted(six.iterkeys(data))
return list(six.iterkeys(data))
tbl = table.PleasantTable(["Start", "Event", "End",
"On Enter", "On Exit"])
for state in orderedkeys(self._states):
prefix_markings = []
if self.current_state == state:
prefix_markings.append("@")
postfix_markings = []
if self.start_state == state:
postfix_markings.append("^")
if self._states[state]['terminal']:
postfix_markings.append("$")
pretty_state = "%s%s" % ("".join(prefix_markings), state)
if postfix_markings:
pretty_state += "[%s]" % "".join(postfix_markings)
if self._transitions[state]:
for event in orderedkeys(self._transitions[state]):
target = self._transitions[state][event]
row = [pretty_state, event, target.name]
if target.on_enter is not None:
try:
row.append(target.on_enter.__name__)
except AttributeError:
row.append(target.on_enter)
else:
row.append('')
if target.on_exit is not None:
try:
row.append(target.on_exit.__name__)
except AttributeError:
row.append(target.on_exit)
else:
row.append('')
tbl.add_row(row)
else:
tbl.add_row([pretty_state, "", "", "", ""])
return tbl.pformat()

139
taskflow/types/table.py
View File

@@ -1,139 +0,0 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import itertools
import os
import six
class PleasantTable(object):
"""A tiny pretty printing table (like prettytable/tabulate but smaller).
Creates simply formatted tables (with no special sauce)::
>>> from taskflow.types import table
>>> tbl = table.PleasantTable(['Name', 'City', 'State', 'Country'])
>>> tbl.add_row(["Josh", "San Jose", "CA", "USA"])
>>> print(tbl.pformat())
+------+----------+-------+---------+
Name | City | State | Country
+------+----------+-------+---------+
Josh | San Jose | CA | USA
+------+----------+-------+---------+
"""
# Constants used when pretty formatting the table.
COLUMN_STARTING_CHAR = ' '
COLUMN_ENDING_CHAR = ''
COLUMN_SEPARATOR_CHAR = '|'
HEADER_FOOTER_JOINING_CHAR = '+'
HEADER_FOOTER_CHAR = '-'
LINE_SEP = os.linesep
@staticmethod
def _center_text(text, max_len, fill=' '):
return '{0:{fill}{align}{size}}'.format(text, fill=fill,
align="^", size=max_len)
@classmethod
def _size_selector(cls, possible_sizes):
"""Select the maximum size, utility function for adding borders.
The number two is used so that the edges of a column have spaces
around them (instead of being right next to a column separator).
:param possible_sizes: possible sizes available
:returns: maximum size
:rtype: number
"""
try:
return max(x + 2 for x in possible_sizes)
except ValueError:
return 0
def __init__(self, columns):
if len(columns) == 0:
raise ValueError("Column count must be greater than zero")
self._columns = [column.strip() for column in columns]
self._rows = []
def add_row(self, row):
if len(row) != len(self._columns):
raise ValueError("Row must have %s columns instead of"
" %s columns" % (len(self._columns), len(row)))
self._rows.append([six.text_type(column) for column in row])
def pformat(self):
# Figure out the maximum column sizes...
column_count = len(self._columns)
column_sizes = [0] * column_count
headers = []
for i, column in enumerate(self._columns):
possible_sizes_iter = itertools.chain(
[len(column)], (len(row[i]) for row in self._rows))
column_sizes[i] = self._size_selector(possible_sizes_iter)
headers.append(self._center_text(column, column_sizes[i]))
# Build the header and footer prefix/postfix.
header_footer_buf = six.StringIO()
header_footer_buf.write(self.HEADER_FOOTER_JOINING_CHAR)
for i, header in enumerate(headers):
header_footer_buf.write(self.HEADER_FOOTER_CHAR * len(header))
if i + 1 != column_count:
header_footer_buf.write(self.HEADER_FOOTER_JOINING_CHAR)
header_footer_buf.write(self.HEADER_FOOTER_JOINING_CHAR)
# Build the main header.
content_buf = six.StringIO()
content_buf.write(header_footer_buf.getvalue())
content_buf.write(self.LINE_SEP)
content_buf.write(self.COLUMN_STARTING_CHAR)
for i, header in enumerate(headers):
if i + 1 == column_count:
if self.COLUMN_ENDING_CHAR:
content_buf.write(headers[i])
content_buf.write(self.COLUMN_ENDING_CHAR)
else:
content_buf.write(headers[i].rstrip())
else:
content_buf.write(headers[i])
content_buf.write(self.COLUMN_SEPARATOR_CHAR)
content_buf.write(self.LINE_SEP)
content_buf.write(header_footer_buf.getvalue())
# Build the main content.
row_count = len(self._rows)
if row_count:
content_buf.write(self.LINE_SEP)
for i, row in enumerate(self._rows):
pieces = []
for j, column in enumerate(row):
pieces.append(self._center_text(column, column_sizes[j]))
if j + 1 != column_count:
pieces.append(self.COLUMN_SEPARATOR_CHAR)
blob = ''.join(pieces)
if self.COLUMN_ENDING_CHAR:
content_buf.write(self.COLUMN_STARTING_CHAR)
content_buf.write(blob)
content_buf.write(self.COLUMN_ENDING_CHAR)
else:
blob = blob.rstrip()
if blob:
content_buf.write(self.COLUMN_STARTING_CHAR)
content_buf.write(blob)
if i + 1 != row_count:
content_buf.write(self.LINE_SEP)
content_buf.write(self.LINE_SEP)
content_buf.write(header_footer_buf.getvalue())
return content_buf.getvalue()

8
tools/state_graph.py
View File

@@ -29,10 +29,11 @@ sys.path.insert(0, top_dir)
# $ pip install pydot2
import pydot
from automaton import machines
from taskflow.engines.action_engine import runner
from taskflow.engines.worker_based import protocol
from taskflow import states
from taskflow.types import fsm
# This is just needed to get at the runner builder object (we will not
@@ -52,7 +53,7 @@ def clean_event(name):
def make_machine(start_state, transitions):
machine = fsm.FSM(start_state)
machine = machines.FiniteMachine()
machine.add_state(start_state)
for (start_state, end_state) in transitions:
if start_state not in machine:
@@ -62,6 +63,7 @@ def make_machine(start_state, transitions):
# Make a fake event (not used anyway)...
event = "on_%s" % (end_state)
machine.add_transition(start_state, end_state, event.lower())
machine.default_start_state = start_state
return machine
@@ -192,7 +194,7 @@ def main():
start = pydot.Node("__start__", shape="point", width="0.1",
xlabel='start', fontcolor='green', **node_attrs)
g.add_node(start)
g.add_edge(pydot.Edge(start, nodes[source.start_state], style='dotted'))
g.add_edge(pydot.Edge(start, nodes[source.default_start_state], style='dotted'))
print("*" * len(graph_name))
print(graph_name)