diff --git a/dvc/cli.py b/dvc/cli.py index 8b62fd473..f17c7e6c9 100644 --- a/dvc/cli.py +++ b/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): diff --git a/dvc/command/show_workflow.py b/dvc/command/show_workflow.py index 83ed1d163..37b6bdc2e 100644 --- a/dvc/command/show_workflow.py +++ b/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 diff --git a/dvc/graph/commit.py b/dvc/graph/commit.py index 79b6e4c66..754c3bad3 100644 --- a/dvc/graph/commit.py +++ b/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_TEXT = DVC_REPRO_PREFIX + ' <>' 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 diff --git a/dvc/graph/workflow.py b/dvc/graph/workflow.py index f33410b17..7f839b602 100644 --- a/dvc/graph/workflow.py +++ b/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: diff --git a/requirements.txt b/requirements.txt index 11f78e9b1..960334c00 100644 --- a/requirements.txt +++ b/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 diff --git a/tests/test_workflow.py b/tests/test_workflow.py index 45b77ea26..9e664aa0c 100644 --- a/tests/test_workflow.py +++ b/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 +