Repository: glotzerlab/signac-flow

Issue:
Refining conditions
There are a couple of things I find confusing about the existing conditions that should perhaps be discussed before a 1.0 release. First, should the `never` and `always` conditions be part of `pre` rather than `_condition`? Placing them as post-conditions just flips their meaning, so it's well-defined, but I'm not sure it conceptually makes sense to put them as post-conditions I think it's a worthwhile simplification. Note that the current behavior also allows nonsensical specifications such as tagging an operation with `pre.never` and `post.never`.

Second, even though neither of these take any arguments, I would still prefer to change them to require parenthesis for consistency. I never remember that the expected syntax for these is different than for other conditions. There are pros and cons to making this change, but in my opinion consistency is more important here, especially because I assume that these are among the least-used conditions so users will expect them to match the behavior of other conditions that they have already used. Is that consistent with what others have experienced?

Discussion:

Patch:
diff --git a/.bumpversion.cfg b/.bumpversion.cfg
index 3621bb1..eea7539 100644
--- a/.bumpversion.cfg
+++ b/.bumpversion.cfg
@@ -4,7 +4,7 @@ commit = True
 
 [bumpversion:file:setup.py]
 
-[bumpversion:file:flow/__init__.py]
+[bumpversion:file:flow/version.py]
 
 [bumpversion:file:doc/conf.py]
 
diff --git a/changelog.txt b/changelog.txt
index 1ad1935..b9e2696 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
 +++++++
@@ -20,6 +21,7 @@ Changed
 Removed
 +++++++
 - Drop support for Python version 2.7.
+- The "always" condition has been deprecated and will trigger a ``DeprecationWarning`` upon use until removed.
 
 
 Version 0.8
diff --git a/flow/__init__.py b/flow/__init__.py
index 231d328..970d07b 100644
--- a/flow/__init__.py
+++ b/flow/__init__.py
@@ -25,6 +25,7 @@ from .environment import get_environment
 from .template import init
 from .util.misc import redirect_log
 from .operations import with_job
+from .version import __version__
 
 # Import packaged environments unless disabled in config:
 from .util.config import get_config_value
@@ -32,8 +33,6 @@ if get_config_value('import_packaged_environments', default=True):
     from . import environments  # noqa:F401
 
 
-__version__ = '0.8.0'
-
 __all__ = [
     'environment',
     'scheduling',
@@ -50,5 +49,6 @@ __all__ = [
     'redirect_log',
     'get_environment',
     'with_job',
-    'testing'
+    'testing',
+    '__version__',
     ]
diff --git a/flow/legacy_templating.py b/flow/legacy_templating.py
index 413d61e..4807ee0 100644
--- a/flow/legacy_templating.py
+++ b/flow/legacy_templating.py
@@ -11,6 +11,7 @@ import warnings
 import functools
 import deprecation
 
+from .version import __version__
 
 NUM_NODES_WARNING = """Unable to determine the reqired number of nodes (nn) for this submission.
 Either provide this value directly with '--nn' or provide the number of processors
@@ -259,6 +260,7 @@ def deprecated_since_06(func):
 
     return deprecation.deprecated(
         deprecated_in=0.6, removed_in=0.8,
+        current_version=__version__,
         details="This function is part of the legacy templating system.")(func)
 
 
diff --git a/flow/project.py b/flow/project.py
index 115ea4f..dc2adc9 100644
--- a/flow/project.py
+++ b/flow/project.py
@@ -66,6 +66,7 @@ from .labels import classlabel
 from .labels import _is_label_func
 from . import legacy
 from .util import config as flow_config
+from .version import __version__
 
 
 logger = logging.getLogger(__name__)
@@ -93,41 +94,56 @@ 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
+    @deprecated(
+        deprecated_in="0.9", removed_in="1.0",
+        details="This condition decorator is obsolete.")
     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 +421,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 +479,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 +498,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 +526,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 +544,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 +719,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.
 
@@ -730,7 +849,9 @@ class FlowProject(signac.contrib.Project, metaclass=_FlowProjectClass):
             return x
 
     @classmethod
-    @deprecated(deprecated_in="0.8", removed_in="1.0")
+    @deprecated(
+        deprecated_in="0.8", removed_in="1.0",
+        current_version=__version__)
     def update_aliases(cls, aliases):
         "Update the ALIASES table for this class."
         cls.ALIASES.update(aliases)
@@ -2102,7 +2223,10 @@ class FlowProject(signac.contrib.Project, metaclass=_FlowProjectClass):
             help="Manually specify all labels that are required for the direct command "
                  "to be considered eligible for execution.")
 
-    @deprecated(deprecated_in="0.8", removed_in="1.0", details="Use export_job_statuses() instead.")
+    @deprecated(
+        deprecated_in="0.8", removed_in="1.0",
+        current_version=__version__,
+        details="Use export_job_statuses() instead.")
     def export_job_stati(self, collection, stati):
         "Export the job stati to a database collection."
         self.export_job_statuses(self, collection, stati)
@@ -2288,7 +2412,10 @@ class FlowProject(signac.contrib.Project, metaclass=_FlowProjectClass):
             raise KeyError("An operation with this identifier is already added.")
         self.operations[name] = FlowOperation(cmd=cmd, pre=pre, post=post, directives=kwargs)
 
-    @deprecated(deprecated_in="0.8", removed_in="1.0", details="Use labels() instead.")
+    @deprecated(
+        deprecated_in="0.8", removed_in="1.0",
+        current_version=__version__,
+        details="Use labels() instead.")
     def classify(self, job):
         """Generator function which yields labels for job.
 
@@ -2343,7 +2470,10 @@ class FlowProject(signac.contrib.Project, metaclass=_FlowProjectClass):
             for op in self._job_operations(job, True):
                 yield op
 
-    @deprecated(deprecated_in="0.8", removed_in="1.0", details="Use next_operations() instead.")
+    @deprecated(
+        deprecated_in="0.8", removed_in="1.0",
+        current_version=__version__,
+        details="Use next_operations() instead.")
     def next_operation(self, job):
         """Determine the next operation for this job.
 
@@ -2479,7 +2609,9 @@ class FlowProject(signac.contrib.Project, metaclass=_FlowProjectClass):
             return False
         return True
 
-    @deprecated(deprecated_in="0.8", removed_in="1.0")
+    @deprecated(
+        deprecated_in="0.8", removed_in="1.0",
+        current_version=__version__)
     def eligible_for_submission(self, job_operation):
         return self._eligible_for_submission(self, job_operation)
 
diff --git a/flow/util/progressbar.py b/flow/util/progressbar.py
index c718fb4..0885ca7 100644
--- a/flow/util/progressbar.py
+++ b/flow/util/progressbar.py
@@ -5,8 +5,10 @@
 import sys
 from deprecation import deprecated
 
+from .version import __version__
 
-@deprecated(deprecated_in="0.8", removed_in="1.0")
+
+@deprecated(deprecated_in="0.8", removed_in="1.0", current_version=__version__)
 def with_progressbar(iterable, total=None, width=120, desc='',
                      percentage=True, file=None):
     if file is None:
diff --git a/flow/version.py b/flow/version.py
new file mode 100644
index 0000000..592662b
--- /dev/null
+++ b/flow/version.py
@@ -0,0 +1,7 @@
+# Copyright (c) 2019 The Regents of the University of Michigan
+# All rights reserved.
+# This software is licensed under the BSD 3-Clause License.
+
+__all__ = ['__version__']
+
+__version__ = '0.8.0'


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..713d284 100644
--- a/tests/test_project.py
+++ b/tests/test_project.py
@@ -27,9 +27,11 @@ from flow.util.misc import add_path_to_environment_pythonpath
 from flow.util.misc import add_cwd_to_environment_pythonpath
 from flow.util.misc import switch_to_directory
 from flow import init
+from deprecation import fail_if_not_removed
 
 from define_test_project import TestProject
 from define_test_project import TestDynamicProject
+from define_dag_test_project import DagTestProject
 
 
 @contextmanager
@@ -513,6 +515,7 @@ class ProjectTest(BaseProjectTest):
     def test_instance(self):
         self.assertTrue(isinstance(self.project, FlowProject))
 
+    @fail_if_not_removed
     def test_labels(self):
         project = self.mock_project()
         for job in project:
@@ -860,6 +863,7 @@ class ExecutionProjectTest(BaseProjectTest):
             project.submit(bundle_size=0)
             self.assertEqual(len(list(MockScheduler.jobs())), 1)
 
+    @fail_if_not_removed
     def test_submit_status(self):
         MockScheduler.reset()
         project = self.mock_project()
@@ -914,6 +918,7 @@ class ExecutionProjectTest(BaseProjectTest):
                       'used by the template script, including: bad_directive\n',
                       stderr.getvalue())
 
+    @fail_if_not_removed
     def test_condition_evaluation(self):
         project = self.mock_project()
 
@@ -1071,5 +1076,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()