mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-02 07:18:25 +01:00
Collapse repro command in the workflow
This commit is contained in:
+3
-3
@@ -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:
|
||||
|
||||
+96
-45
@@ -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
|
||||
|
||||
@@ -2,3 +2,5 @@ dvc
|
||||
pandas
|
||||
sklearn
|
||||
scipy
|
||||
configobj
|
||||
networkx
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user