Repository: glotzerlab/signac-flow

Issue:
Export workflow graph
**[Original report](https://bitbucket.org/glotzer/signac-flow/issue/41) by Carl Simon Adorf (Bitbucket: [csadorf](https://bitbucket.org/csadorf), GitHub: [csadorf](https://github.com/csadorf)).**

----------------------------------------

As per offline discussion, we determined that it should be possible to explicitly export the directed graph defined as `FlowProject`, *.e.g.*, as NetworkX graph structure.

In loose relation to that, we should think on how to submit a complete workflow instead of individual operations to a scheduler.


Discussion:
mikemhenry: > I had to make them text files for GitHub to allow the upload

What year is it github?! That annoys me so much. I can take a look at this tonight or tomorrow. 
vyasr: @mikemhenry sounds good. I recall that you reviewed an earlier PR on this topic so I figured you'd be a good second reviewer (I think @bdice has had some other PR reviews requested recently so I moved it to you).
vyasr: @mikemhenry I think I've addressed your question now. Feel free to merge if you think it's ready.

Patch:
diff --git a/changelog.txt b/changelog.txt
index 1ad1935..2181a51 100644
--- a/changelog.txt
+++ b/changelog.txt
@@ -11,7 +11,8 @@ next
 Added
 +++++
 
-- Add "fork" directive to enforce the execution of operations within a subprocess.
+- Add "fork" directive to enforce the execution of operations within a subprocess (#159).
+- Operation graph detection based on function comparison (#178).
 
 Changed
 +++++++
diff --git a/flow/project.py b/flow/project.py
index 115ea4f..b4ba91b 100644
--- a/flow/project.py
+++ b/flow/project.py
@@ -93,41 +93,53 @@ The available filters are:
 
 
 class _condition(object):
-
-    def __init__(self, condition):
+    # This counter should be incremented each time an "always" or "never"
+    # condition is created, and the value should be used as the tag for that
+    # condition to ensure that no pair of "always" and "never" conditions
+    # are found to be equal by the graph detection algorithm.
+    current_arbitrary_tag = 0
+
+    def __init__(self, condition, tag=None):
+        # Add tag to differentiate built-in conditions during graph detection.
+        if tag is None:
+            tag = condition.__code__.co_code
+        condition._flow_tag = tag
         self.condition = condition
 
     @classmethod
     def isfile(cls, filename):
         "True if the specified file exists for this job."
-        return cls(lambda job: job.isfile(filename))
+        return cls(lambda job: job.isfile(filename), 'isfile_' + filename)
 
     @classmethod
     def true(cls, key):
         """True if the specified key is present in the job document and
         evaluates to True."""
-        return cls(lambda job: job.document.get(key, False))
+        return cls(lambda job: job.document.get(key, False), 'true_' + key)
 
     @classmethod
     def false(cls, key):
         """True if the specified key is present in the job document and
         evaluates to False."""
-        return cls(lambda job: not job.document.get(key, False))
+        return cls(lambda job: not job.document.get(key, False), 'false_' + key)
 
     @classmethod
     def always(cls, func):
         "Returns True."
-        return cls(lambda _: True)(func)
+        cls.current_arbitrary_tag += 1
+        return cls(lambda _: True, str(cls.current_arbitrary_tag))(func)
 
     @classmethod
     def never(cls, func):
         "Returns False."
-        return cls(lambda _: False)(func)
+        cls.current_arbitrary_tag += 1
+        return cls(lambda _: False, str(cls.current_arbitrary_tag))(func)
 
     @classmethod
     def not_(cls, condition):
         "Returns ``not condition(job)`` for the provided condition function."
-        return cls(lambda job: not condition(job))
+        return cls(lambda job: not condition(job),
+                   'not_'.encode() + condition.__code__.co_code)
 
 
 def make_bundles(operations, size=None):
@@ -405,6 +417,19 @@ class FlowOperation(object):
             return self._cmd.format(job=job)
 
 
+def _create_all_metacondition(condition_dict, *other_funcs):
+    """Standard function for generating aggregate metaconditions that require
+    *all* provided conditions to be met. The resulting metacondition is
+    constructed with appropriate information for graph detection."""
+    condition_list = [c for f in other_funcs for c in condition_dict[f]]
+
+    def _flow_metacondition(job):
+        return all(c(job) for c in condition_list)
+
+    _flow_metacondition._composed_of = condition_list
+    return _flow_metacondition
+
+
 class _FlowProjectClass(type):
     """Metaclass for the FlowProject class."""
     def __new__(metacls, name, bases, namespace, **kwargs):
@@ -450,12 +475,17 @@ class _FlowProjectClass(type):
 
             The *hello*-operation would only execute if the 'hello' key in the job
             document does not evaluate to True.
+
+            An optional tag may be associated with the condition. These tags
+            are used by :meth:`~.detect_operation_graph` when comparing
+            conditions for equality. The tag defaults to the bytecode of the
+            function.
             """
 
             _parent_class = parent_class
 
-            def __init__(self, condition):
-                self.condition = condition
+            def __init__(self, condition, tag=None):
+                super(pre, self).__init__(condition, tag)
 
             def __call__(self, func):
                 self._parent_class._OPERATION_PRE_CONDITIONS[func].insert(0, self.condition)
@@ -464,20 +494,15 @@ class _FlowProjectClass(type):
             @classmethod
             def copy_from(cls, *other_funcs):
                 "True if and only if all pre conditions of other operation-function(s) are met."
-                def metacondition(job):
-                    return all(c(job)
-                               for other_func in other_funcs
-                               for c in cls._parent_class._collect_pre_conditions()[other_func])
-                return cls(metacondition)
+                return cls(_create_all_metacondition(cls._parent_class._collect_pre_conditions(),
+                                                     *other_funcs))
 
             @classmethod
             def after(cls, *other_funcs):
                 "True if and only if all post conditions of other operation-function(s) are met."
-                def metacondition(job):
-                    return all(c(job)
-                               for other_func in other_funcs
-                               for c in cls._parent_class._collect_post_conditions()[other_func])
-                return cls(metacondition)
+                return cls(_create_all_metacondition(cls._parent_class._collect_post_conditions(),
+                                                     *other_funcs))
+
         return pre
 
     @staticmethod
@@ -497,11 +522,16 @@ class _FlowProjectClass(type):
 
             The *bye*-operation would be considered complete and therefore no longer
             eligible for execution once the 'bye' key in the job document evaluates to True.
+
+            An optional tag may be associated with the condition. These tags
+            are used by :meth:`~.detect_operation_graph` when comparing
+            conditions for equality. The tag defaults to the bytecode of the
+            function.
             """
             _parent_class = parent_class
 
-            def __init__(self, condition):
-                self.condition = condition
+            def __init__(self, condition, tag=None):
+                super(post, self).__init__(condition, tag)
 
             def __call__(self, func):
                 self._parent_class._OPERATION_POST_CONDITIONS[func].insert(0, self.condition)
@@ -510,11 +540,9 @@ class _FlowProjectClass(type):
             @classmethod
             def copy_from(cls, *other_funcs):
                 "True if and only if all post conditions of other operation-function(s) are met."
-                def metacondition(job):
-                    return all(c(job)
-                               for other_func in other_funcs
-                               for c in cls._parent_class._collect_post_conditions()[other_func])
-                return cls(metacondition)
+                return cls(_create_all_metacondition(cls._parent_class._collect_post_conditions(),
+                                                     *other_funcs))
+
         return post
 
 
@@ -687,6 +715,93 @@ class FlowProject(signac.contrib.Project, metaclass=_FlowProjectClass):
 
         return label_func
 
+    def detect_operation_graph(self):
+        """Determine the directed acyclic graph defined by operation pre- and
+        post-conditions.
+
+        In general, executing a given operation registered with a FlowProject
+        just involves checking the operation's pre- and post-conditions to
+        determine eligibility. More generally, however, the pre- and
+        post-conditions define a directed acyclic graph that governs the
+        execution of all operations. Visualizing this graph can be useful for
+        finding logic errors in the specified conditions, and having this graph
+        computed also enables additional execution modes. For example, using
+        this graph it is possible to determine exactly what operations need to
+        be executed in order to make the operation eligible so that the task of
+        executing all necessary operations can be automated.
+
+        The graph is determined by iterating over all pairs of operations and
+        checking for equality of pre- and post-conditions. The algorithm builds
+        an adjacency matrix based on whether the pre-conditions for one
+        operation match the post-conditions for another. The comparison of
+        operations is conservative; by default, conditions must be composed of
+        identical code to be identified as equal (technically, they must be
+        bytecode equivalent i.e. `cond1.__code__.co_code ==
+        cond2.__code__.co_code`). Users can specify that conditions should be
+        treated as equal by providing tags to the operations.
+
+        Given a FlowProject subclass defined in a module `project.py`, the
+        output graph could be visualized using Matplotlib and NetworkX with the
+        following code:
+
+        .. code-block:: python
+
+            import numpy as np
+            import networkx as nx
+            from matplotlib import pyplot as plt
+
+            from project import Project
+
+            project = Project()
+            ops = project.operations.keys()
+            adj = np.asarray(project.detect_operation_graph())
+
+            plt.figure()
+            g = nx.DiGraph(adj)
+            pos = nx.spring_layout(g)
+            nx.draw(g, pos)
+            nx.draw_networkx_labels(
+                g, pos,
+                labels={key: name for (key, name) in
+                        zip(range(len(ops)), [o for o in ops])})
+
+            plt.show()
+        """
+
+        def to_callbacks(conditions):
+            """Get the actual callables associated with FlowConditions."""
+            return [condition._callback for condition in conditions]
+
+        def unpack_conditions(condition_functions):
+            """Identify any metaconditions in the list and reduce them to the
+            functions that they are composed of. The callbacks argument is used
+            in recursive calls to the function and appended to directly, but
+            only returned at the end."""
+            callbacks = set()
+            for cf in condition_functions:
+                if cf.__name__ == "_flow_metacondition":
+                    callbacks = callbacks.union(
+                        unpack_conditions(cf._composed_of))
+                else:
+                    callbacks.add(cf._flow_tag)
+
+            return callbacks
+
+        ops = list(self.operations.items())
+        mat = [[0 for _ in range(len(ops))] for _ in range(len(ops))]
+
+        for i, (name1, op1) in enumerate(ops):
+            for j, (name2, op2) in enumerate(ops[i:]):
+                postconds1 = unpack_conditions(to_callbacks(op1._postconds))
+                postconds2 = unpack_conditions(to_callbacks(op2._postconds))
+                prereqs1 = unpack_conditions(to_callbacks(op1._prereqs))
+                prereqs2 = unpack_conditions(to_callbacks(op2._prereqs))
+                if postconds1.intersection(prereqs2):
+                    mat[i][j+i] = 1
+                elif prereqs1.intersection(postconds2):
+                    mat[j+i][i] = 1
+        return mat
+
     def _register_class_labels(self):
         """This function registers all label functions, which are part of the class definition.
 


Test patch:
diff --git a/tests/define_dag_test_project.py b/tests/define_dag_test_project.py
new file mode 100644
index 0000000..6dd4ecb
--- /dev/null
+++ b/tests/define_dag_test_project.py
@@ -0,0 +1,71 @@
+from flow import FlowProject
+
+
+class DagTestProject(FlowProject):
+    pass
+
+
+def first_complete(job):
+    return job.doc.get('first') is not None
+
+
+@DagTestProject.operation
+@DagTestProject.post(first_complete)
+def first(job):
+    job.doc.first = True
+
+
+@DagTestProject.operation
+@DagTestProject.pre(first_complete)
+@DagTestProject.post.true('second')
+@DagTestProject.post.false('other_condition')
+def second(job):
+    job.doc.second = True
+
+
+@DagTestProject.operation
+@DagTestProject.pre.after(first)
+@DagTestProject.post.true('third')
+def third(job):
+    job.doc.third = True
+
+
+@DagTestProject.operation
+@DagTestProject.pre.true('second')
+@DagTestProject.pre.after(third)
+@DagTestProject.post.true('fourth')
+def fourth(job):
+    job.doc.fourth = True
+
+
+@DagTestProject.operation
+@DagTestProject.pre.copy_from(third)
+@DagTestProject.post.isfile('fifth.txt')
+def fifth(job):
+    import os
+
+    def touch(fname, times=None):
+        with open(fname, 'a'):
+            os.utime(fname, times)
+    with job:
+        touch('fifth.txt')
+
+
+@DagTestProject.operation
+@DagTestProject.pre.after(third)
+@DagTestProject.pre.isfile('fifth.txt')
+@DagTestProject.post.true('sixth')
+def sixth(job):
+    job.doc.sixth = True
+
+
+@DagTestProject.operation
+@DagTestProject.pre.after(third)
+@DagTestProject.pre.after(fourth)
+@DagTestProject.post.true('seventh')
+def seventh(job):
+    job.doc.seventh = True
+
+
+if __name__ == '__main__':
+    DagTestProject().main()
diff --git a/tests/test_project.py b/tests/test_project.py
index e88e6e2..4c0c9ca 100644
--- a/tests/test_project.py
+++ b/tests/test_project.py
@@ -30,6 +30,7 @@ from flow import init
 
 from define_test_project import TestProject
 from define_test_project import TestDynamicProject
+from define_dag_test_project import DagTestProject
 
 
 @contextmanager
@@ -1071,5 +1072,24 @@ class DynamicProjectMainInterfaceTest(ProjectMainInterfaceTest):
     project_class = TestDynamicProject
 
 
+class ProjectDagDetectionTest(BaseProjectTest):
+    """Tests of operation DAG detection."""
+    project_class = DagTestProject
+
+    def test_dag(self):
+        project = self.mock_project()
+        adj = project.detect_operation_graph()
+
+        adj_correct = [[0, 1, 1, 0, 1, 0, 0],
+                       [0, 0, 0, 1, 0, 0, 0],
+                       [0, 0, 0, 1, 0, 1, 1],
+                       [0, 0, 0, 0, 0, 0, 1],
+                       [0, 0, 0, 0, 0, 1, 0],
+                       [0, 0, 0, 0, 0, 0, 0],
+                       [0, 0, 0, 0, 0, 0, 0]]
+
+        self.assertEqual(adj, adj_correct)
+
+
 if __name__ == '__main__':
     unittest.main()