Merge pull request #11 from nazavode/dev

Automaton formatting for transition table and state-transition graph

automaton/__init__.py

@@ -21,12 +21,15 @@
 import re
 import collections
 
-from .automaton import (
+from .automaton import (  # noqa: F401
     Event,
     Automaton,
     AutomatonError,
     DefinitionError,
     InvalidTransitionError,
+    tabulate,
+    plantuml,
+    transition_table,
 )
 
 
@@ -35,18 +38,6 @@
 __license__ = 'Apache License Version 2.0'
 __version__ = '1.0.0'
 
-__all__ = (
-    # Version
-    "VERSION_INFO",
-    "VERSION",
-    # Automaton API
-    "Event",
-    "Automaton",
-    # Exceptions
-    "AutomatonError",
-    "DefinitionError",
-    "InvalidTransitionError",
-)
 
 ###################################################
 VERSION = __version__

automaton/automaton.py

@@ -21,6 +21,7 @@
 from itertools import chain, product, filterfalse
 from collections import namedtuple, Iterable
 
+import tabulate as tabulator
 import networkx as nx
 
 
@@ -30,6 +31,9 @@
     "AutomatonError",
     "DefinitionError",
     "InvalidTransitionError",
+    "transition_table",
+    "plantuml",
+    "tabulate",
 )
 
 
@@ -226,7 +230,7 @@ def __new__(mcs, class_name, class_bases, class_dict):
                         len(components), ", ".join("{}".format(c.nodes()) for c in components))
                 )
             # 4. Save
-            cls.__graph__ = graph
+            cls.__graph__ = nx.freeze(graph)
         return cls
 
 
@@ -355,6 +359,11 @@ def _get_cut(cls, *states, inbound=True):
             If set, the inbound events will be returned,
             outbound otherwise. Defaults to `True`.
 
+        Raises
+        ------
+        KeyError
+            When an unknown state is found while iterating.
+
         Yields
         ------
         any
@@ -401,6 +410,11 @@ def in_events(cls, *states):
         states : tuple(any)
             The states subset.
 
+        Raises
+        ------
+        KeyError
+            When an unknown state is found while iterating.
+
         Yields
         ------
         any
@@ -413,6 +427,11 @@ def out_events(cls, *states):
         """ Retrieves all the outbound events leaving the
         specified states with no duplicates.
 
+        Raises
+        ------
+        KeyError
+            When an unknown state is found while iterating.
+
         Parameters
         ----------
         states : tuple(any)
@@ -435,3 +454,147 @@ def get_default_initial_state(cls):
             The automaton default initial state.
         """
         return cls.__default_initial_state__
+
+    def __str__(self):
+        return '<{}@{}>'.format(self.__class__.__name__, self.state)
+
+
+#########################################################################
+# Rendering stuff, everything beyond this point is formatting for humans.
+
+def plantuml(automaton, traversal=None):
+    """ Render an automaton's state-transition graph as a
+    `PlantUML state diagram <http://plantuml.com/state-diagram>`_.
+
+    A simple example will be formatted as follows:
+
+        >>> class TrafficLight(Automaton):
+        ...     go = Event('red', 'green')
+        ...     slowdown = Event('green', 'yellow')
+        ...     stop = Event('yellow', 'red')
+        >>> print( plantuml(TrafficLight) )  # doctest: +SKIP
+
+        @startuml
+            green --> yellow : slowdown
+            yellow --> red : stop
+            red --> green : go
+        @enduml
+
+    Parameters
+    ----------
+    automaton : `~automaton.Automaton`
+        The automaton to be rendered. It can be both
+        a class and an instance.
+    traversal : callable(graph), optional
+        An optional callable used to sort the events.
+        It has the same meaning as the ``traversal``
+        parameter of `automaton.transition_table`.
+
+    Returns
+    -------
+    str
+        Returns the formatted state graph.
+    """
+    graph = automaton.__graph__
+    source_nodes = filter(lambda n: not graph.in_edges(n), graph.nodes())
+    sink_nodes = filter(lambda n: not graph.out_edges(n), graph.nodes())
+    sources = [('[*]', node) for node in source_nodes]
+    sinks = [(node, '[*]') for node in sink_nodes]
+    table = transition_table(automaton, traversal=traversal)
+    return """@startuml
+{}
+{}
+{}
+@enduml""".format('\n'.join(['    {} --> {}'.format(*row) for row in sources]),
+                  '\n'.join(['    {} --> {} : {}'.format(*row) for row in table]),
+                  '\n'.join(['    {} --> {}'.format(*row) for row in sinks]))
+
+
+def transition_table(automaton, traversal=None):
+    """ Build the adjacency table of the given graph.
+
+    Parameters
+    ----------
+    automaton : `~automaton.Automaton`
+        The automaton to be rendered. It can be both
+        a class and an instance.
+    traversal : callable(graph), optional
+        An optional callable used to yield the
+        edges of the graph representation of the
+        automaton. It must accept a `networkx.MultiDiGraph`
+        as the first positional argument
+        and yield one edge at a time as a tuple in the
+        form ``(source_state, destination_state)``. The default
+        traversal sorts the states in ascending order by
+        inbound grade (number of incoming events).
+
+    Yields
+    ------
+    (source, dest, event)
+        Yields one row at a time as a tuple containing
+        the source and destination node of the edge and
+        the name of the event associated with the edge.
+    """
+    graph = automaton.__graph__
+    if not traversal:
+        traversal = lambda G: sorted(G.edges(), key=lambda e: len(G.in_edges(e[0])))
+    # Retrieve event names since networkx traversal
+    # functions lack data retrieval
+    events = nx.get_edge_attributes(graph, 'event')  # -> (source, dest, key==0): event
+    # Build raw data table to be rendered
+    for source, dest in traversal(graph):
+        yield (source, dest, events[source, dest, 0])
+
+
+def tabulate(automaton, header=None, tablefmt=None, traversal=None):
+    """ Render an automaton's transition table as in
+    text format.
+
+    The transition table has three columns: source node,
+    destination node and the name of the event.
+
+    A simple example will be formatted as follows:
+
+        >>> class TrafficLight(Automaton):
+        ...     go = Event('red', 'green')
+        ...     slowdown = Event('green', 'yellow')
+        ...     stop = Event('yellow', 'red')
+        >>> tabulate(TrafficLight) # doctest: +SKIP
+
+        ========  ======  ========
+        Source    Dest    Event
+        ========  ======  ========
+        green     yellow  slowdown
+        yellow    red     stop
+        red       green   go
+        ========  ======  ========
+
+    Parameters
+    ----------
+    automaton : `~automaton.Automaton`
+        The automaton to be rendered. It can be both
+        a class and an instance.
+    header : list[str, str, str]
+        An optional list of fields to be used as table
+        headers. Defaults to a predefined header.
+    tablefmt str, optional
+        Specifies the output format for the table.
+        All formats supported by
+        `tabulate <https://pypi.python.org/pypi/tabulate>`_
+        package are supported (e.g.: ``rst`` for
+        reStructuredText, ``pipe`` for Markdown).
+        Defaults to ``rst``.
+    traversal : callable(graph), optional
+        An optional callable used to sort the events.
+        It has the same meaning as the ``traversal``
+        parameter of `automaton.transition_table`.
+
+    Returns
+    -------
+    str
+        Returns the formatted transition table.
+    """
+    header = header or ['Source', 'Dest', 'Event']
+    tablefmt = tablefmt or 'rst'
+    table = transition_table(automaton=automaton, traversal=traversal)
+    return tabulator.tabulate(table, header, tablefmt=tablefmt)

flake8.ini

@@ -1,5 +1,5 @@
 [flake8]
-ignore = W391,F841
+ignore = W391,F841,E731
 max-line-length = 120
 # exclude = tests/*
-# max-complexity = 10
+# max-complexity = 10

requirements/develop.txt

@@ -16,3 +16,5 @@ tox==2.5.0
 virtualenv==15.1.0
 wrapt==1.10.8
 bumpversion==0.5.3
+pytest-sugar==0.8.0
+termcolor==1.1.0

requirements/install.txt

@@ -1,2 +1,3 @@
 decorator==4.0.11
 networkx==1.11
+tabulate==0.7.7

setup.py

@@ -35,7 +35,7 @@
     author_email="federico.ficarelli@gmail.com",
     url="https://github.com/nazavode/automaton",
     packages=['automaton'],
-    install_requires=['networkx'],
+    install_requires=['networkx', 'tabulate'],
     classifiers=[
         'Development Status :: 5 - Production/Stable',
         'Intended Audience :: Developers',

tests/test_automaton.py

@@ -33,6 +33,13 @@ class TrafficLight(Automaton):
     return TrafficLight
 
 
+def test_str(traffic_light):
+    auto = traffic_light()
+    assert str(auto) == '<TrafficLight@red>'
+    auto.go()
+    assert str(auto) == '<TrafficLight@green>'
+
+
 def test_definition():
     class Simple(Automaton):
         __default_initial_state__ = 'state_a'
@@ -396,3 +403,69 @@ class Star(Automaton):
 
     with pytest.raises(KeyError):
         set(Star.out_events('unknown1', 'unknown2', 'center'))  # Unknown state
+
+
+def test_event_edges():
+    event = Event('a', 'b')
+    event.bind('testevent')
+    assert list(event.edges()) == [('a', 'b')]
+    assert list(event.edges(data=True)) == [('a', 'b', {'event': 'testevent'})]
+    #
+    event = Event(('a', 'b', 'c', 'd'), 'x')
+    event.bind('testevent')
+    assert list(event.edges()) == [('a', 'x'), ('b', 'x'), ('c', 'x'), ('d', 'x')]
+    assert list(event.edges(data=True)) == [('a', 'x', {'event': 'testevent'}),
+        ('b', 'x', {'event': 'testevent'}), ('c', 'x', {'event': 'testevent'}),
+        ('d', 'x', {'event': 'testevent'})]
+    #
+    class Sink(Automaton):
+        event1 = Event('state_a', 'state_b')
+        event2 = Event(('state_a', 'state_b', 'state_c', 'state_d'), 'sink1')
+        event3 = Event(('state_a', 'state_b', 'state_c', 'state_d', 'sink1'), 'sink2')
+        event4 = Event('sink2', 'state_a')
+    assert list(Sink.event2.edges()) == \
+        [('state_a', 'sink1'), ('state_b', 'sink1'), ('state_c', 'sink1'), ('state_d', 'sink1')]
+
+
+@pytest.fixture(params=[None, lambda G: G.edges(data=False)])
+def traversal(request):
+    return request.param
+
+
+@pytest.mark.parametrize('header', [None, [], [1, 2, 3], ['a', 'b', 'c']])
+@pytest.mark.parametrize('tablefmt', [None, '', 'rst', 'pipe'])
+def test_tabulate(header, tablefmt, traversal):
+
+    class Sink(Automaton):
+        event1 = Event('state_a', 'state_b')
+        event2 = Event(('state_a', 'state_b', 'state_c', 'state_d'), 'sink1')
+        event3 = Event(('state_a', 'state_b', 'state_c', 'state_d'), 'sink2')
+        event4 = Event('sink2', 'state_a')
+
+    assert tabulate(Sink, header=header, tablefmt=tablefmt, traversal=traversal)
+
+
+def test_plantuml(traversal):
+
+    class Sink(Automaton):
+        event1 = Event('state_a', 'state_b')
+        event2 = Event(('state_a', 'state_b', 'state_c', 'state_d'), 'sink1')
+        event3 = Event(('state_a', 'state_b', 'state_c', 'state_d'), 'sink2')
+        event4 = Event('sink2', 'state_a')
+
+    assert plantuml(Sink, traversal=traversal)
+
+
+def test_transition_table(traversal):
+
+    class Sink(Automaton):
+        event1 = Event('state_a', 'state_b')
+        event2 = Event(('state_a', 'state_b', 'state_c', 'state_d'), 'sink1')
+        event3 = Event(('state_a', 'state_b', 'state_c', 'state_d'), 'sink2')
+        event4 = Event('sink2', 'state_a')
+
+    table = list(transition_table(Sink, traversal=traversal))
+    assert table
+    assert len(table) == 10
+    for row in table:
+        assert len(row) == 3