Merge changes

dvc/cli.py

@@ -389,11 +389,24 @@ def parse_args(argv=None):
     workflow_parser = show_subparsers.add_parser(
                         'workflow',
                         parents=[parent_parser],
-                        help='Show workflow image')
+                        help='Show workflow image. It collapses DVC repro commits if possible.')
     workflow_parser.add_argument(
                         'target',
                         nargs='?',
                         help='Target metric data file')
+    workflow_parser.add_argument(
+                        '-d',
+                        '--dvc-commits',
+                        action='store_true',
+                        default=False,
+                        help='Show DVC repro commits.')
+    workflow_parser.add_argument(
+                        '-a',
+                        '--all-commits',
+                        action='store_true',
+                        default=False,
+                        help='Show all commits')
+
     workflow_parser.set_defaults(func=CmdShowWorkflow)
 
     if isinstance(argv, str):

dvc/command/show_workflow.py

@@ -21,7 +21,7 @@ class CmdShowWorkflow(CmdBase):
                 Logger.debug(u'Set show workflow target as {}'.format(target))
 
         wf = self.git.get_all_commits(target, self.settings)
-        wf.build_graph()
+        wf.build_graph(self.settings.parsed_args.dvc_commits, self.settings.parsed_args.all_commits)
         return 0
 
     @property

dvc/graph/commit.py

@@ -1,7 +1,7 @@
 class Commit(object):
     TEXT_LIMIT = 30
     DVC_REPRO_PREFIX = 'DVC repro'
-    COLLAPSED_TEXT = DVC_REPRO_PREFIX + ' <<collapsed commits>>'
+    COLLAPSED_TEXT = DVC_REPRO_PREFIX + ' <<collapsed>>'
 
     def __init__(self, hash, parents, name, date, comment,
                  is_target=False,
@@ -16,6 +16,7 @@ class Commit(object):
         self._target_metric = target_metric
         self._target_metric_delta = None
         self._branch_tips = [] if branch_tips is None else branch_tips
+        self._collapsed_commits = []
 
         self._is_collapsed = False
 
@@ -33,38 +34,63 @@ class Commit(object):
     def remove_parent(self, hash):
         self._parent_hashes.remove(hash)
 
+    def add_collapsed_commit(self, commit):
+        self._collapsed_commits.append(commit)
+
     @property
     def text(self):
         branch_text = ''
         if self._branch_tips:
-            branch_text = 'Branch tips: {}\n'.format(', '.join(self._branch_tips))
+            branch_text = 'BRANCH TIPS: {}\n'.format(', '.join(self._branch_tips))
 
-        if self._is_collapsed:
-            return branch_text + self.COLLAPSED_TEXT + self._text_metrics_line()
+        return self._text_metrics_line() + branch_text + self._comments_text
 
-        return branch_text + self._text_comment() + self._text_metrics_line() + '\n' + self._text_hash()
+    @staticmethod
+    def _text_comment(commit):
+        if commit._is_collapsed:
+            text = Commit.COLLAPSED_TEXT
+        else:
+            text = Commit._limit_text(commit._comment)
+        return '[{}] {}'.format(commit.hash, text)
 
-    def _text_hash(self):
-        return 'Commit: ' + self.hash
+    @staticmethod
+    def _limit_text(comment):
+        if len(comment) < Commit.TEXT_LIMIT:
+            return comment
+        else:
+            return comment[:Commit.TEXT_LIMIT-3] + '...'
 
-    def _text_comment(self):
-        if len(self._comment) < self.TEXT_LIMIT:
-            return self._comment
-        return self._comment[:self.TEXT_LIMIT] + '...'
+    @property
+    def _comments_text(self):
+        res = []
+        if not self.is_repro:
+            res.append(self._text_comment(self))
+
+        for commit in self._collapsed_commits:
+            if not commit.is_repro:
+                text = self._text_comment(commit)
+                res.append(text)
+        if len(res) == 0:
+            return self._text_comment(self)
+
+        max_len = max(map(lambda x: len(x), res))
+        res_extended_len = map(lambda x: x.ljust(max_len), res)
+        return '\n'.join(res_extended_len)
 
     def _text_metrics_line(self):
         result = ''
         if self._is_target and self._target_metric:
-            result = '\nTarget: {}'.format(self._target_metric)
+            result = 'TARGET: {}'.format(self._target_metric)
             if self._target_metric_delta is not None:
                 result += ' ({:+f})'.format(self._target_metric_delta)
+            result += '\n'
         return result
 
     @property
     def is_repro(self):
         return self._comment.startswith(self.DVC_REPRO_PREFIX)
 
-    def make_colapsed(self):
+    def make_collapsed(self):
         self._is_collapsed = True
 
     @property

dvc/graph/workflow.py

@@ -9,8 +9,67 @@ class WorkflowError(DvcException):
         super(WorkflowError, self).__init__('Workflow error: ' + msg)
 
 
+class CommitCollapseStrategy(object):
+    def __init__(self, workflow):
+        self._workflow = workflow
+        pass
+
+    def is_change_needed(self, commit, child_commits):
+        raise NotImplementedError()
+
+    def is_remove_needed(self, commit):
+        raise NotImplementedError()
+
+    def make_collapsed(self, commit):
+        commit.make_collapsed()
+
+    def upstream(self, commit):
+        for child in self._workflow.child_commits(commit.hash):
+            self.upstream_to_child(commit, child)
+        pass
+
+    def upstream_to_child(self, commit, child):
+        if commit.has_target_metric:
+            if not child.has_target_metric:
+                child.set_target_metric(commit.target_metric)
+        if commit.branch_tips:
+            child.add_branch_tips(commit.branch_tips)
+
+
+class CollapseDvcReproCommitsStrategy(CommitCollapseStrategy):
+    def __init__(self, workflow):
+        super(CollapseDvcReproCommitsStrategy, self).__init__(workflow)
+
+    def is_change_needed(self, commit, hash):
+        return commit.is_repro
+
+    def is_remove_needed(self, commit):
+        child_commits = self._workflow.child_commits(commit.hash)
+        return child_commits and all(ch.is_repro for ch in child_commits)
+
+
+class CollapseNotMeticsCommitsStrategy(CommitCollapseStrategy):
+    def __init__(self, workflow):
+        super(CollapseNotMeticsCommitsStrategy, self).__init__(workflow)
+
+    def is_change_needed(self, commit, hash):
+        return not commit.has_target_metric
+
+    def is_remove_needed(self, commit):
+        child_commits = self._workflow.child_commits(commit.hash)
+        return len(child_commits) == 1 and not commit.has_target_metric
+
+    def upstream_to_child(self, commit, child):
+        super(CollapseNotMeticsCommitsStrategy, self).upstream_to_child(commit, child)
+        if not commit.is_repro:
+            child.add_collapsed_commit(commit)
+
+    def make_collapsed(self, commit):
+        pass
+
+
 class Workflow(object):
-    def __init__(self, target, merges_map, branches_map=None, no_repro_commits=True):
+    def __init__(self, target, merges_map, branches_map=None):
         self._target = target
         self._merges_map = merges_map
         self._branches_map = branches_map
@@ -20,8 +79,6 @@ class Workflow(object):
 
         self._edges = {}
         self._back_edges = {}
-
-        self._no_repro_commits = no_repro_commits
         pass
 
     def add_commit(self, commit):
@@ -39,14 +96,19 @@ class Workflow(object):
             self._root_hash = commit.hash
         pass
 
-    def build_graph(self):
+    def get_commit(self, hash):
+        return self._commits.get(hash)
+
+    def child_commits(self, hash):
+        if hash not in self._edges:
+            return []
+        return [self._commits[h] for h in self._edges[hash]]
+
+    def build_graph(self, show_dvc_commits, show_all_commits):
+        self.modify_workflow(show_all_commits, show_dvc_commits)
+
         g = nx.DiGraph(name='DVC Workflow', directed=False)
 
-        self.derive_target_metric_deltas()
-
-        if self._no_repro_commits:
-            self.collapse_repro_commits()
-
         for hash in set(self._edges.keys() + self._back_edges.keys()):
             commit = self._commits[hash]
 
@@ -68,6 +130,13 @@ class Workflow(object):
         A.draw(fname, format='jpeg', prog='dot')
         pass
 
+    def modify_workflow(self, show_all_commits, show_dvc_commits):
+        self.derive_target_metric_deltas()
+        if not show_all_commits:
+            self.collapse_commits(CollapseDvcReproCommitsStrategy(self))
+            if not show_dvc_commits:
+                self.collapse_commits(CollapseNotMeticsCommitsStrategy(self))
+
     @staticmethod
     def node_color(commit):
         if commit.target_metric_delta is not None:
@@ -87,25 +156,33 @@ class Workflow(object):
 
         self._build_graph(next, g)
 
-    def collapse_repro_commits(self):
+    def collapse_commits(self, strategy):
         hashes_to_remove = []
         for commit in self._commits.values():
-            if commit.is_repro:
-                if self._remove_or_collapse(commit):
+            if strategy.is_change_needed(commit, commit.hash):
+                if self._remove_or_collapse(commit, strategy):
                     hashes_to_remove.append(commit.hash)
 
+        self.remove_hashes(hashes_to_remove)
+
+    def remove_hashes(self, hashes_to_remove):
         for hash in hashes_to_remove:
             del self._commits[hash]
-            del self._edges[hash]
-            del self._back_edges[hash]
+            if hash in self._edges:
+                del self._edges[hash]
+            if hash in self._back_edges:
+                del self._back_edges[hash]
         pass
 
-    def _remove_or_collapse(self, commit):
-        parent_commit_hashes = self._back_edges[commit.hash]
+    def _remove_or_collapse(self, commit, strategy):
+        parent_commit_hashes = self._back_edges.get(commit.hash)
         child_commit_hashes = self._edges.get(commit.hash)
 
-        if child_commit_hashes and all(self._commits[hash].is_repro for hash in child_commit_hashes):
-            self._upstream_metrics_and_branch_tips(child_commit_hashes, commit)
+        if child_commit_hashes is None:
+            return False
+
+        if strategy.is_remove_needed(commit):
+            strategy.upstream(commit)
 
             for hash in child_commit_hashes:
                 self._commits[hash].add_parents(commit.parent_hashes)
@@ -116,24 +193,17 @@ class Workflow(object):
 
             return True
         else:
-            commit.make_colapsed()
+            strategy.make_collapsed(commit)
             return False
         pass
 
-    def _upstream_metrics_and_branch_tips(self, child_commit_hashes, commit):
-        for hash in child_commit_hashes:
-            if commit.has_target_metric:
-                if not self._commits[hash].has_target_metric:
-                    self._commits[hash].set_target_metric(commit.target_metric)
-            if commit.branch_tips:
-                self._commits[hash].add_branch_tips(commit.branch_tips)
-        pass
-
     @staticmethod
     def _redirect_edges(edges, child_commit_hashes, hash, commit_hashes):
-        for h in commit_hashes:
-            edges[h].remove(hash)
-            edges[h] |= child_commit_hashes
+        if commit_hashes:
+            for h in commit_hashes:
+                edges[h].remove(hash)
+                if child_commit_hashes:
+                    edges[h] |= child_commit_hashes
 
     def derive_target_metric_deltas(self):
         if not self._target:

requirements.txt

@@ -89,3 +89,4 @@ google-compute-engine==2.4.1
 configobj==5.0.6
 networkx==1.11
 gcs-oauth2-boto-plugin==1.14
+xmltodict==0.11.0

tests/test_workflow.py

@@ -1,6 +1,7 @@
 from unittest import TestCase
 
-from dvc.graph.workflow import Workflow
+from dvc.graph.workflow import Workflow,\
+    CollapseDvcReproCommitsStrategy, CollapseNotMeticsCommitsStrategy
 from dvc.graph.commit import Commit
 
 
@@ -11,6 +12,13 @@ class TestWorkflow(TestCase):
         self._commit2 = Commit('2', '1', 'name1', 'today', 'DVC repro-run ...')
         self._commit1 = Commit('1', '', 'name1', 'today', 'comment1')
 
+        self._showed_text1 = '[' + self._commit1.hash + '] ' + self._commit1._comment
+        self._showed_text2 = '[' + self._commit2.hash + '] ' + self._commit2._comment
+        self._showed_text3 = '[' + self._commit3.hash + '] ' + self._commit3._comment
+        self._showed_text4 = '[' + self._commit4.hash + '] ' + self._commit4._comment
+
+        self._showed_text3_collapsed = '[' + self._commit3.hash + '] ' + Commit.COLLAPSED_TEXT
+
     def commits_basic_test(self):
         self.assertFalse(self._commit1.is_repro)
         self.assertTrue(self._commit2.is_repro)
@@ -27,14 +35,10 @@ class TestWorkflow(TestCase):
 
         self.assertEqual(len(wf._commits), 4)
 
-        self.assertEqual(wf._commits['1'].text,
-                         self._commit1._comment + '\n' + self._commit1._text_hash())
-        self.assertEqual(wf._commits['2'].text,
-                         self._commit2._comment + '\n' + self._commit2._text_hash())
-        self.assertEqual(wf._commits['3'].text,
-                         self._commit3._comment + '\n' + self._commit3._text_hash())
-        self.assertEqual(wf._commits['4'].text,
-                         self._commit4._comment + '\n' + self._commit4._text_hash())
+        self.assertEqual(wf._commits['1'].text, self._showed_text1)
+        self.assertEqual(wf._commits['2'].text, self._showed_text2)
+        self.assertEqual(wf._commits['3'].text, self._showed_text3)
+        self.assertEqual(wf._commits['4'].text, self._showed_text4)
         pass
 
     def collapse_test(self):
@@ -44,12 +48,11 @@ class TestWorkflow(TestCase):
         wf.add_commit(self._commit3)
         wf.add_commit(self._commit4)
 
-        wf.collapse_repro_commits()
+        wf.collapse_commits(CollapseDvcReproCommitsStrategy(wf))
 
         self.assertEqual(len(wf._commits), 3)
-        self.assertEqual(wf._commits[self._commit1.hash].text,
-                         self._commit1._comment + '\n' + self._commit1._text_hash())
-        self.assertEqual(wf._commits[self._commit3.hash].text, Commit.COLLAPSED_TEXT)
+        self.assertEqual(wf._commits[self._commit1.hash].text, self._showed_text1)
+        self.assertEqual(wf._commits[self._commit3.hash].text, self._showed_text3_collapsed)
         self.assertTrue('2' not in wf._commits)
 
         self.assertFalse('2' in wf._edges)
@@ -63,30 +66,54 @@ class TestWorkflow(TestCase):
         wf.add_commit(self._commit3) # Dead end which cannot be collapsed
 
         self.assertEqual(len(wf._commits), 3)
-        wf.collapse_repro_commits()
+        wf.collapse_commits(CollapseDvcReproCommitsStrategy(wf))
         self.assertEqual(len(wf._commits), 2)
 
-        self.assertEqual(wf._commits[self._commit1.hash].text,
-                         self._commit1._comment + '\n' + self._commit1._text_hash())
-        self.assertEqual(wf._commits[self._commit3.hash].text, Commit.COLLAPSED_TEXT)
+        self.assertEqual(wf._commits[self._commit1.hash].text, self._showed_text1)
+        self.assertEqual(wf._commits[self._commit3.hash].text, self._showed_text3)
         self.assertTrue('2' not in wf._commits)
         pass
 
     def collapse_metric_commit_test(self):
         value = 0.812345
         branches = ['master', 'try_smth']
-        metric_commit3 = Commit('2', '1', 'name1', 'today', 'DVC repro-run ...',
+        metric_commit2 = Commit('2', '1', 'name1', 'today', 'DVC repro-run ...',
                                 True, value, branch_tips=branches)
 
         wf = Workflow('', '')
         wf.add_commit(self._commit1)
-        wf.add_commit(metric_commit3)
+        wf.add_commit(metric_commit2)
         wf.add_commit(self._commit3)
 
         self.assertEqual(len(wf._commits), 3)
-        wf.collapse_repro_commits()
+        wf.collapse_commits(CollapseDvcReproCommitsStrategy(wf))
         self.assertEqual(len(wf._commits), 2)
 
         self.assertEqual(wf._commits['3']._target_metric, value)
         self.assertEqual(wf._commits['3'].branch_tips, branches)
         pass
+
+    def collapse_all_commits_test(self):
+        value = 0.812345
+        branches = ['master', 'try_smth']
+        not_metric_commit2 = Commit('2', '1', 'name1', 'today', 'change smth 2...',
+                                    True, branch_tips=branches)
+        metric_commit3 = Commit('3', '2', 'name3', 'today3', 'hello3',
+                                True, value)
+
+        wf = Workflow('', '')
+        wf.add_commit(self._commit1)
+        wf.add_commit(not_metric_commit2)
+        wf.add_commit(metric_commit3)
+        wf.add_commit(self._commit4)
+
+        self.assertEqual(len(wf._commits), 4)
+        wf.collapse_commits(CollapseDvcReproCommitsStrategy(wf))
+        self.assertEqual(len(wf._commits), 4)
+        wf.collapse_commits(CollapseNotMeticsCommitsStrategy(wf))
+        self.assertEqual(len(wf._commits), 2)
+
+        self.assertEqual(wf._commits['3']._target_metric, value)
+        self.assertEqual(wf._commits['3'].branch_tips, branches)
+        pass
+