diff --git a/dvc/git_wrapper.py b/dvc/git_wrapper.py index 1e77a48b9..754a317bc 100644 --- a/dvc/git_wrapper.py +++ b/dvc/git_wrapper.py @@ -4,7 +4,7 @@ from dvc.logger import Logger from dvc.config import Config from dvc.executor import Executor, ExecutorError from dvc.system import System -from dvc.workflow import Workflow +from dvc.workflow import Workflow, Commit class GitWrapperI(object): @@ -248,8 +248,8 @@ class GitWrapper(GitWrapperI): assert len(items) == 5, 'Git wrapper: git log format has {} items, 5 expected'.format(len(items)) hash, parent_hash, name, date, comment = items - wf.add_commit(hash, parent_hash, name, date, comment, - GitWrapper.is_target(hash, target)) + wf.add_commit(Commit(hash, parent_hash, name, date, comment, + GitWrapper.is_target(hash, target))) return wf except ExecutorError: diff --git a/dvc/workflow.py b/dvc/workflow.py index 2e18f6429..349fd34d7 100644 --- a/dvc/workflow.py +++ b/dvc/workflow.py @@ -9,76 +9,97 @@ class WorkflowError(DvcException): class Commit(object): - def __init__(self, hash, parent_hash, name, date, comment, is_target): + TEXT_LIMIT = 30 + DVC_REPRO_PREFIX = 'DVC repro' + COLLAPSED_TEXT = DVC_REPRO_PREFIX + '\n<< collapsed commits >>' + + def __init__(self, hash, parents, name, date, comment, is_target): self._hash = hash - self._parent_hash = parent_hash + self._parent_hashes = set(parents.split()) self._name = name self._date = date self._comment = comment self._is_target = is_target + self._is_collapsed = False @property def hash(self): return self._hash + @property + def parent_hashes(self): + return self._parent_hashes + + def add_parents(self, parent_hashes): + self._parent_hashes |= parent_hashes + + def remove_parent(self, hash): + self._parent_hashes.remove(hash) + + @property + def text(self): + if self._is_collapsed: + return self.COLLAPSED_TEXT + return self._comment[:self.TEXT_LIMIT] + '\n' + self.hash + + @property + def is_repro(self): + return self._comment.startswith(self.DVC_REPRO_PREFIX) + + def make_colapsed(self): + self._is_collapsed = True + class Workflow(object): - ''' - Options: - --all - all commits - //--no-repro - by default (no DVC commits) - --no-dvc - no DVC commits - - ''' - - COMMIT_TEXT_LIMIT = 20 - - def __init__(self, target, merges_map): + def __init__(self, target, merges_map, root='', no_repro_commits=True): self._target = target self._merges_map = merges_map self._commits = {} - self._root = {} + self._root = root + + self._edges = {} + self._back_edges = {} + + self._no_repro_commits = no_repro_commits pass - def add_commit(self, hash, parent_hash, name, date, comment, is_target): - # Extract params to the Commit class? - print 'add_commit %s' % (hash) - self._commits[hash] = Commit(hash, parent_hash, name, date, comment, is_target) - if not parent_hash: - self._root = hash + def add_commit(self, commit): + self._commits[commit.hash] = commit + + for p in commit.parent_hashes: + if p not in self._edges: + self._edges[p] = set() + self._edges[p].add(commit.hash) + + if commit.hash not in self._back_edges: + self._back_edges[commit.hash] = set() + for p in commit.parent_hashes: + self._back_edges[commit.hash].add(p) pass def build_graph(self): g = nx.DiGraph(name='DVC Workflow', directed=False) - if not self._root: - raise WorkflowError('cannot find root commit') + if self._no_repro_commits: + self.collaps_repro_commits() + + for hash in set(self._edges.keys() + self._back_edges.keys()): + commit = self._commits[hash] + g.add_node(hash, + attr_dict={ + 'label': commit.text, + 'color': 'red' if commit.hash == '85d03b2' else 'black', + 'weight': 8 if commit.hash == '411ea7c' else 18 + } + ) - # self._build_graph(self._commits[self._root], g) - visited = set() for commit in self._commits.values(): - if commit.hash not in visited: - g.add_node(commit.hash, - attr_dict={ - 'label': commit._comment[:Workflow.COMMIT_TEXT_LIMIT] + '\n' + commit.hash, - 'color': 'red' if commit.hash == '85d03b2' else 'black', - 'weight': 8 if commit.hash == '411ea7c' else 18 - } - ) - visited.add(commit.hash) - if commit._parent_hash: - g.add_edge(commit.hash, commit._parent_hash) - if commit.hash == '411ea7c': - g.add_edge(commit.hash, commit._parent_hash, attr_dict={ - 'color': 'green' - }) - g.add_edge(commit.hash, commit._parent_hash, attr_dict={ - 'color': 'red' - }) + for p in commit.parent_hashes: + g.add_edge(commit.hash, p) A = nx.nx_agraph.to_agraph(g) - fname = 'workflow' + fname = 'workflow.jpeg' A = A.to_undirected() A.write(fname + '.dot') A.draw(fname, format='jpeg', prog='dot') @@ -89,9 +110,39 @@ class Workflow(object): return next = self._commits[curr.hash] - print 'add node %s' % next.hash g.add_node(next.hash) - print 'add edge (%s, %s)' % (curr.hash, next.hash) g.add_edge(curr.hash, next.hash) self._build_graph(next, g) + + def collaps_repro_commits(self): + hashes_to_remove = [] + for commit in self._commits.values(): + if commit.is_repro: + parent_commit_hashes = self._back_edges[commit.hash] + child_commit_hashes = self._edges.get(commit.hash) + + # Remove if next commit is dvc-repro as well + if child_commit_hashes and all(self._commits[hash].is_repro for hash in child_commit_hashes): + for hash in child_commit_hashes: + self._commits[hash].add_parents(commit.parent_hashes) + self._commits[hash].remove_parent(commit.hash) + pass + + for hash in parent_commit_hashes: + self._edges[hash].remove(commit.hash) + self._edges[hash] |= child_commit_hashes + + for hash in child_commit_hashes: + self._back_edges[hash].remove(commit.hash) + self._back_edges[hash] |= parent_commit_hashes + + hashes_to_remove.append(commit.hash) + else: + commit.make_colapsed() + + for hash in hashes_to_remove: + del self._commits[hash] + del self._edges[hash] + del self._back_edges[hash] + pass diff --git a/functest/requirements.txt b/functest/requirements.txt index 22924891c..80a128f8e 100644 --- a/functest/requirements.txt +++ b/functest/requirements.txt @@ -2,3 +2,5 @@ dvc pandas sklearn scipy +configobj +networkx diff --git a/functest/tutorial_branches.sh b/functest/tutorial_branches.sh index 1a954760a..249e82a6a 100755 --- a/functest/tutorial_branches.sh +++ b/functest/tutorial_branches.sh @@ -6,6 +6,13 @@ set -e mkdir myrepo cd myrepo + +git init +echo "This is an empty readme" > README.md +git add README.md +git commit -m 'add readme' + +git checkout -b first_model mkdir code wget -nv -P code/ https://s3-us-west-2.amazonaws.com/dvc-share/so/code/featurization.py \ https://s3-us-west-2.amazonaws.com/dvc-share/so/code/evaluate.py \ @@ -13,15 +20,14 @@ wget -nv -P code/ https://s3-us-west-2.amazonaws.com/dvc-share/so/code/featuriza https://s3-us-west-2.amazonaws.com/dvc-share/so/code/split_train_test.py \ https://s3-us-west-2.amazonaws.com/dvc-share/so/code/xml_to_tsv.py \ https://s3-us-west-2.amazonaws.com/dvc-share/so/code/requirements.txt -pip install -U -r code/requirements.txt - -git init +#pip install -U -r code/requirements.txt +# Edit eval file format git add code/ git commit -m 'Download code' dvc init -dvc import https://s3-us-west-2.amazonaws.com/dvc-share/so/10K/Posts.xml.tgz data/ +dvc import https://s3-us-west-2.amazonaws.com/dvc-share/so/100K/Posts.xml.tgz data/ dvc run tar zxf data/Posts.xml.tgz -C data/ dvc run python code/xml_to_tsv.py data/Posts.xml data/Posts.tsv python @@ -34,17 +40,9 @@ dvc run python code/evaluate.py data/model.p data/matrix-test.p data/eval_auc.tx dvc target data/eval_auc.txt cat data/eval_auc.txt -# AUC: 0.349 +# AUC: 0.645320 -git checkout -b input_100K # <-- -dvc remove data/Posts.xml.tgz -dvc import https://s3-us-west-2.amazonaws.com/dvc-share/so/100K/Posts.xml.tgz data/ -dvc repro -cat data/eval_auc.txt -# 0.645320 - -git checkout master # <-- git checkout -b input_25K # <-- dvc remove data/Posts.xml.tgz dvc import https://s3-us-west-2.amazonaws.com/dvc-share/so/25K/Posts.xml.tgz data/ @@ -73,16 +71,21 @@ cat data/eval_auc.txt # 0.621002 git checkout 2195f1032 -b three_grams # <-- +dvc repro # not needed +cat data/eval_auc.txt +# 0.630682 + +git checkout first_model +git merge three_grams +dvc repro + + + vi code/featurization.py # 0.578447 - - - - - # 2. Reproduce: change input dataset dvc remove data/Posts.xml.tgz diff --git a/tests/test_workflow.py b/tests/test_workflow.py new file mode 100644 index 000000000..db9211f36 --- /dev/null +++ b/tests/test_workflow.py @@ -0,0 +1,66 @@ +from unittest import TestCase + +from dvc.workflow import Workflow, Commit + + +class TestWorkflow(TestCase): + def setUp(self): + self._commit4 = Commit('4', '3', 'name1', 'today', 'comment4', False) + self._commit3 = Commit('3', '2', 'name1', 'today', 'DVC repro-run ...', False) + self._commit2 = Commit('2', '1', 'name1', 'today', 'DVC repro-run ...', False) + self._commit1 = Commit('1', '', 'name1', 'today', 'comment1', False) + + def test_commits(self): + self.assertFalse(self._commit1.is_repro) + self.assertTrue(self._commit2.is_repro) + self.assertTrue(self._commit3.is_repro) + self.assertFalse(self._commit4.is_repro) + pass + + def test_workflow_basic(self): + wf = Workflow('', '') + wf.add_commit(self._commit1) + wf.add_commit(self._commit2) + wf.add_commit(self._commit3) + wf.add_commit(self._commit4) + + self.assertEqual(len(wf._commits), 4) + + self.assertEqual(wf._commits['1'].text, self._commit1._comment + '\n' + self._commit1.hash) + self.assertEqual(wf._commits['2'].text, self._commit2._comment + '\n' + self._commit2.hash) + self.assertEqual(wf._commits['3'].text, self._commit3._comment + '\n' + self._commit3.hash) + self.assertEqual(wf._commits['4'].text, self._commit4._comment + '\n' + self._commit4.hash) + pass + + def test_collapse(self): + wf = Workflow('', '') + wf.add_commit(self._commit1) + wf.add_commit(self._commit2) + wf.add_commit(self._commit3) + wf.add_commit(self._commit4) + + wf.collaps_repro_commits() + + self.assertEqual(len(wf._commits), 3) + self.assertEqual(wf._commits[self._commit1.hash].text, self._commit1._comment + '\n' + self._commit1.hash) + self.assertEqual(wf._commits[self._commit3.hash].text, Commit.COLLAPSED_TEXT) + self.assertTrue('2' not in wf._commits) + + self.assertFalse('2' in wf._edges) + self.assertFalse('2' in wf._back_edges) + pass + + def test_collapse_at_dead_end(self): + wf = Workflow('', '') + wf.add_commit(self._commit1) + wf.add_commit(self._commit2) + wf.add_commit(self._commit3) # Dead end which cannot be collapsed + + self.assertEqual(len(wf._commits), 3) + wf.collaps_repro_commits() + self.assertEqual(len(wf._commits), 2) + + self.assertEqual(wf._commits[self._commit1.hash].text, self._commit1._comment + '\n' + self._commit1.hash) + self.assertEqual(wf._commits[self._commit3.hash].text, Commit.COLLAPSED_TEXT) + self.assertTrue('2' not in wf._commits) + pass