Improve graph generation with ROI

.github/workflows/pytest.yml

@@ -15,6 +15,12 @@ jobs:
     strategy:
       matrix:
         python-version: ["3.8", "3.9", "3.10", "3.11"]
+        extras: ["[diagrams]"]
+        include:
+          - python-version: "3.11"
+            extras: "[]"
+          - python-version: "3.11"
+            extras: "[diagrams,mypy]"
 
     steps:
     - uses: actions/checkout@v3
@@ -24,11 +30,18 @@ jobs:
         python-version: ${{ matrix.python-version }}
     - name: Install dependencies
       run: |
-        sudo apt-get install libgraphviz-dev graphviz
         python -m pip install --upgrade pip
-        python -m pip install -r requirements.txt -r requirements_diagrams.txt 
+        python -m pip install -r requirements.txt
         python -m pip install -U -r requirements_test.txt
-        python -c "print 'hello'" > /dev/null 2>&1 || pip install -r requirements_mypy.txt
+    - name: Install dependencies for diagrams
+      if: contains(matrix.extras, 'diagrams')
+      run: |
+        sudo apt-get install libgraphviz-dev graphviz
+        python -m pip install -r requirements.txt -r requirements_diagrams.txt 
+    - name: Install dependencies for mypy
+      if: contains(matrix.extras, 'mypy')
+      run: |
+          python -c "print 'hello'" > /dev/null 2>&1 || pip install -r requirements_mypy.txt
     - name: Test with pytest
       run: |
         coverage run --source=transitions -m pytest --doctest-modules tests/

Changelog.md

@@ -4,6 +4,10 @@
 
 - Bug #594: Fix may_<trigger> always returning false for internal transitions (thanks @a-schade)
 - PR #592: Pass investigated transition to `EventData` context in 'may' check (thanks @msclock)
+- PR #634: Improve the handling of diagrams when working with parallel states, especially when using the show_roi option (thanks @seanxlliu)
+- '_anchor' suffix has been removed for (py)graphviz cluster node anchors
+- local testing switched from [tox](https://github.com/tox-dev/tox) to [nox](https://github.com/wntrblm/nox)
+- PR #633: Remove surrounding whitespace from docstrings (thanks @artofhuman)
 
 ## 0.9.0 (September 2022)
 

MANIFEST.in

@@ -6,6 +6,7 @@ include LICENSE
 include MANIFEST
 include *.ini
 include conftest.py
+include noxfile.py
 
 recursive-include transitions *.pyi
 recursive-include examples *.ipynb

noxfile.py

@@ -0,0 +1,43 @@
+import nox
+
+
+python = ["2.7", "3.7", "3.8", "3.9", "3.10", "3.11"]
+nox.options.stop_on_first_error = True
+
+
+@nox.session(
+    python=python[-1]
+)
+def test_check_manifest(session):
+    session.install("check-manifest")
+    session.run("check-manifest")
+
+
+@nox.session(
+    python=python[-1]
+)
+def test_mypy(session):
+    session.install(".")
+    session.install("-rrequirements_test.txt")
+    session.install("-rrequirements_diagrams.txt")
+    session.install("-rrequirements_mypy.txt")
+    session.run("pytest", "-nauto", "--doctest-modules", "tests/")
+
+
+@nox.session(
+    python=python
+)
+def test(session):
+    session.install(".")
+    session.install("-rrequirements_test.txt")
+    session.install("-rrequirements_diagrams.txt")
+    session.run("pytest", "-nauto", "tests/")
+
+
+@nox.session(
+    python=python[-1]
+)
+def test_no_gv(session):
+    session.install(".")
+    session.install("-rrequirements_test.txt")
+    session.run("pytest", "-nauto", "tests/")

requirements_diagrams.txt

@@ -1 +1,2 @@
 pygraphviz
+graphviz

requirements_test.txt

@@ -4,5 +4,4 @@ pytest-runner
 pytest-xdist
 mock
 dill
-graphviz
 pycodestyle

tests/test_graphviz.py

@@ -1,31 +1,40 @@
-try:
-    from builtins import object
-except ImportError:
-    pass
-
 from .utils import Stuff, DummyModel
 from .test_core import TestTransitions, TYPE_CHECKING
 
 from transitions.extensions import (
     LockedGraphMachine, GraphMachine, HierarchicalGraphMachine, LockedHierarchicalGraphMachine
 )
-from transitions.extensions.nesting import NestedState
 from transitions.extensions.states import add_state_features, Timeout, Tags
 from unittest import skipIf
 import tempfile
 import os
 import re
+import sys
+from unittest import TestCase
 
 try:
     # Just to skip tests if graphviz not installed
     import graphviz as pgv  # @UnresolvedImport
 except ImportError:  # pragma: no cover
     pgv = None
 
-
 if TYPE_CHECKING:
     from typing import Type, List, Collection, Union
 
+edge_re = re.compile(r"^\s+(?P<src>\w+)\s*->\s*(?P<dst>\w+)\s*(?P<attr>\[.*\]?)\s*$")
+node_re = re.compile(r"^\s+(?P<node>\w+)\s+(?P<attr>\[.*\]?)\s*$")
+
+
+class TestDiagramsImport(TestCase):
+
+    use_pygraphviz = False
+    pgv = pgv
+
+    def test_import(self):
+        machine = GraphMachine(None, use_pygraphviz=self.use_pygraphviz)
+        if machine.graph_cls is None:
+            self.assertIsNone(pgv)
+
 
 @skipIf(pgv is None, 'Graph diagram test requires graphviz.')
 class TestDiagrams(TestTransitions):
@@ -38,16 +47,19 @@ def parse_dot(self, graph):
             dot = graph.string()
         else:
             dot = graph.source
-        nodes = []
+        nodes = set()
         edges = []
         for line in dot.split('\n'):
-            if '->' in line:
-                src, rest = line.split('->')
-                dst, attr = rest.split(None, 1)
-                nodes.append(src.strip().replace('"', ''))
-                nodes.append(dst)
-                edges.append(attr)
-        return dot, set(nodes), edges
+            match = edge_re.search(line)
+            if match:
+                nodes.add(match.group("src"))
+                nodes.add(match.group("dst"))
+                edges.append(match.group("attr"))
+            else:
+                match = node_re.search(line)
+                if match and match.group("node") not in ["node", "graph", "edge"]:
+                    nodes.add(match.group("node"))
+        return dot, nodes, edges
 
     def tearDown(self):
         pass
@@ -206,8 +218,8 @@ def test_roi(self):
         m.to_B()
         g3 = m.get_graph(show_roi=True)
         _, nodes, edges = self.parse_dot(g3)
-        self.assertEqual(len(edges), 3)
-        self.assertEqual(len(nodes), 4)
+        self.assertEqual(len(edges), 3)  # to_state_{A,C,F}
+        self.assertEqual(len(nodes), 5)  # B + A,C,F (edges) + E (previous)
 
     def test_state_tags(self):
 
@@ -283,6 +295,10 @@ class TestDiagramsLocked(TestDiagrams):
 
     machine_cls = LockedGraphMachine  # type: Type[LockedGraphMachine]
 
+    @skipIf(sys.version_info < (3, ), "Python 2.7 cannot retrieve __name__ from partials")
+    def test_function_callbacks_annotation(self):
+        super(TestDiagramsLocked, self).test_function_callbacks_annotation()
+
 
 @skipIf(pgv is None, 'NestedGraph diagram test requires graphviz')
 class TestDiagramsNested(TestDiagrams):
@@ -314,8 +330,7 @@ def test_diagram(self):
 
         self.assertEqual(len(edges), 8)
         # Test that graph properties match the Machine
-        self.assertEqual(set(m.states.keys()) - set(['C', 'C%s1' % NestedState.separator]),
-                         set(nodes) - set(['C_anchor', 'C%s1_anchor' % NestedState.separator]))
+        self.assertEqual(set(m.get_nested_state_names()), nodes)
         m.walk()
         m.run()
 
@@ -350,6 +365,50 @@ def is_fast(self, *args, **kwargs):
         self.assertEqual(len(edges), 2)
         self.assertEqual(len(nodes), 3)
 
+    def test_roi_parallel(self):
+        class Model:
+            @staticmethod
+            def is_fast(*args, **kwargs):
+                return True
+
+        self.states[0] = {"name": "A", "parallel": ["1", "2"]}
+
+        model = Model()
+        m = self.machine_cls(model, states=self.states, transitions=self.transitions, initial='A', title='A test',
+                             use_pygraphviz=self.use_pygraphviz, show_conditions=True)
+        g1 = model.get_graph(show_roi=True)
+        _, nodes, edges = self.parse_dot(g1)
+        self.assertEqual(len(edges), 2)  # reset and walk
+        print(nodes)
+        self.assertEqual(len(nodes), 4)
+        model.walk()
+        model.run()
+        model.sprint()
+        g2 = model.get_graph(show_roi=True)
+        dot, nodes, edges = self.parse_dot(g2)
+        self.assertEqual(len(edges), 2)
+        self.assertEqual(len(nodes), 3)
+
+    def test_roi_parallel_deeper(self):
+        states = ['A', 'B', 'C', 'D',
+                  {'name': 'P',
+                   'parallel': [
+                       '1',
+                       {'name': '2', 'parallel': [
+                           {'name': 'a'},
+                           {'name': 'b', 'parallel': [
+                               {'name': 'x', 'parallel': ['1', '2']}, 'y'
+                           ]}
+                       ]},
+                   ]}]
+        transitions = [["go", "A", "P"], ["reset", "*", "A"]]
+        m = self.machine_cls(states=states, transitions=transitions, initial='A', title='A test',
+                             use_pygraphviz=self.use_pygraphviz, show_conditions=True)
+        m.go()
+        _, nodes, edges = self.parse_dot(m.get_graph(show_roi=True))
+        self.assertEqual(len(edges), 2)
+        self.assertEqual(len(nodes), 10)
+
     def test_internal(self):
         states = ['A', 'B']
         transitions = [['go', 'A', 'B'],
@@ -359,6 +418,7 @@ def test_internal(self):
                              use_pygraphviz=self.use_pygraphviz)
 
         _, nodes, edges = self.parse_dot(m.get_graph())
+        print(nodes)
         self.assertEqual(len(nodes), 2)
         self.assertEqual(len([e for e in edges if '[internal]' in e]), 1)
 
@@ -441,3 +501,7 @@ class TestDiagramsLockedNested(TestDiagramsNested):
     def setUp(self):
         super(TestDiagramsLockedNested, self).setUp()
         self.machine_cls = LockedHierarchicalGraphMachine  # type: Type[LockedHierarchicalGraphMachine]
+
+    @skipIf(sys.version_info < (3, ), "Python 2.7 cannot retrieve __name__ from partials")
+    def test_function_callbacks_annotation(self):
+        super(TestDiagramsLockedNested, self).test_function_callbacks_annotation()

tests/test_pygraphviz.py

@@ -1,10 +1,5 @@
-try:
-    from builtins import object
-except ImportError:
-    pass
-
 from .utils import Stuff
-from .test_graphviz import TestDiagrams, TestDiagramsNested
+from .test_graphviz import TestDiagrams, TestDiagramsNested, TestDiagramsImport
 from transitions.extensions.states import add_state_features, Timeout, Tags
 from unittest import skipIf
 
@@ -15,6 +10,12 @@
     pgv = None
 
 
+class TestPygraphvizImport(TestDiagramsImport):
+
+    use_pygraphviz = True
+    pgv = pgv
+
+
 @skipIf(pgv is None, 'Graph diagram requires pygraphviz')
 class PygraphvizTest(TestDiagrams):
 

transitions/extensions/diagrams_base.py

@@ -8,7 +8,6 @@
 import copy
 import abc
 import logging
-
 import six
 
 _LOGGER = logging.getLogger(__name__)
@@ -107,6 +106,12 @@ def _get_global_name(self, path):
         else:
             return self.machine.get_global_name()
 
+    def _flatten(self, *lists):
+        return (e for a in lists for e in
+                (self._flatten(*a)
+                 if isinstance(a, (tuple, list))
+                 else (a.name if hasattr(a, 'name') else a,)))
+
     def _get_elements(self):
         states = []
         transitions = []
@@ -139,7 +144,7 @@ def _get_elements(self):
                         ini = ini.name if hasattr(ini, "name") else ini
                         tran = dict(
                             trigger="",
-                            source=self.machine.state_cls.separator.join(prefix + [state["name"]]) + "_anchor",
+                            source=self.machine.state_cls.separator.join(prefix + [state["name"]]),
                             dest=self.machine.state_cls.separator.join(
                                 prefix + [state["name"], ini]
                             ),