First workflow graph

dvc/cli.py

@@ -403,6 +403,11 @@ def parse_args(argv=None):
                         'target',
                         nargs='*',
                         help='Target data')
+    show_parser.add_argument('-w',
+                             '--workflow',
+                             action='store_true',
+                             default=False,
+                             help='Show workflow')
     show_parser.set_defaults(func=CmdShow)
 
     if isinstance(argv, str):

dvc/command/import_file.py

@@ -66,13 +66,14 @@ class CmdImportFile(CmdBase):
     def is_dir_path(output):
         return len(output) > 0 and output[-1] == os.path.sep
 
-    def import_and_commit_if_needed(self, input, output, lock=False, check_if_ready=True):
+    def import_and_commit_if_needed(self, input, output, lock=False, check_if_ready=True, is_repro=False):
         if check_if_ready and not self.no_git_actions and not self.git.is_ready_to_go():
             return 1
 
         self.import_file(input, output, lock)
 
-        message = 'DVC import file: {} {}'.format(input, output)
+        cmd_name = 'repro-import' if is_repro else 'import'
+        message = 'DVC {} file: {} {}'.format(cmd_name, input, output)
         return self.commit_if_needed(message)
 
     def import_file(self, input, output, lock):

dvc/command/repro.py

@@ -180,7 +180,9 @@ class ReproChange(object):
             cmd.set_locker(False)
 
             Logger.info(u'Reproducing import command: {}'.format(output))
-            if cmd.import_and_commit_if_needed(input, output, lock=True, check_if_ready=False) != 0:
+            if cmd.import_and_commit_if_needed(input, output, lock=True,
+                                               check_if_ready=False,
+                                               is_repro=True) != 0:
                 raise ReproError('Import command reproduction failed')
             return True
         elif self.state.is_run:
@@ -202,7 +204,8 @@ class ReproChange(object):
                                             [],
                                             [],
                                             False,
-                                            check_if_ready=False) != 0:
+                                            check_if_ready=False,
+                                            is_repro=True) != 0:
                 raise ReproError('Run command reproduction failed')
             return True
         else:

dvc/command/run.py

@@ -54,7 +54,8 @@ class CmdRun(CmdBase):
                                  input_data_items=None,
                                  code_dependencies=None,
                                  lock=None,
-                                 check_if_ready=True):
+                                 check_if_ready=True,
+                                 is_repro=False):
         if check_if_ready and not self.no_git_actions and not self.git.is_ready_to_go():
             return 1
 
@@ -69,7 +70,8 @@ class CmdRun(CmdBase):
                          code_dependencies,
                          lock)
 
-        return self.commit_if_needed('DVC run: {}'.format(' '.join(command_args)))
+        cmd_name = 'repro-run' if is_repro else 'run'
+        return self.commit_if_needed('DVC {}: {}'.format(cmd_name, ' '.join(command_args)))
 
     def run_command(self, cmd_args, data_items_from_args, not_data_items_from_args,
                     stdout=None, stderr=None, shell=False,

dvc/command/show.py

@@ -47,6 +47,13 @@ class CmdShow(Traverse):
         return 0
 
     def run(self):
+
+        if self.parsed_args.workflow:
+            # Workflow
+            wf = self.git.get_all_commits('data/eval_auc.txt')
+            wf.build_graph()
+            return
+
         saved_targets = self.settings.parsed_args.target
         self.settings.parsed_args.target = [self.settings.config.data_dir]
  

dvc/git_wrapper.py

@@ -4,6 +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
 
 
 class GitWrapperI(object):
@@ -226,3 +227,47 @@ class GitWrapper(GitWrapperI):
                 code_files.append(code)
 
         return code_files, code_dirs
+
+    LOG_SEPARATOR = '|'
+    LOG_FORMAT = ['%h', '%p', '%an', '%ai', '%s']
+
+    @staticmethod
+    def get_all_commits(target):
+        # git log --all --abbrev=7 --pretty=format:"%h|%p|%an|%ai|%s"
+        try:
+            merges_map = GitWrapper.get_merges_map()
+
+            format_str = GitWrapper.LOG_SEPARATOR.join(GitWrapper.LOG_FORMAT)
+            git_cmd = ['git', 'log', '--all', '--abbrev={}'.format(GitWrapper.COMMIT_LEN),
+                       '--pretty=format:{}'.format(format_str)]
+            lines = Executor.exec_cmd_only_success(git_cmd).split('\n')
+
+            wf = Workflow(target, merges_map)
+            for line in lines:
+                items = line.split(GitWrapper.LOG_SEPARATOR, len(GitWrapper.LOG_FORMAT))
+                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))
+
+            return wf
+        except ExecutorError:
+            raise
+
+    @staticmethod
+    def is_target(hash, target):
+        # git show --pretty="" --name-only b6fa39b01
+        git_cmd = ['git', 'show', '--pretty=', '--name-only', hash]
+        files = set(Executor.exec_cmd_only_success(git_cmd).split('\n'))
+        return target in files
+
+    @staticmethod
+    def get_merges_map():
+        # git log --merges --all --abbrev=7
+        # {'a4b56f1': back_to_600_est}
+        git_cmd = ['git', 'log', '--all', '--merges',
+                   '--abbrev={}'.format(GitWrapper.COMMIT_LEN)]
+        # lines = map(str.strip, Executor.exec_cmd_only_success(git_cmd).split('\n'))
+        # lines.map
+        return {}

dvc/workflow.py

@@ -0,0 +1,97 @@
+import networkx as nx
+
+from dvc.exceptions import DvcException
+
+
+class WorkflowError(DvcException):
+    def __init__(self, msg):
+        super(WorkflowError, self).__init__('Workflow error: ' + msg)
+
+
+class Commit(object):
+    def __init__(self, hash, parent_hash, name, date, comment, is_target):
+        self._hash = hash
+        self._parent_hash = parent_hash
+        self._name = name
+        self._date = date
+        self._comment = comment
+        self._is_target = is_target
+
+    @property
+    def hash(self):
+        return self._hash
+
+
+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):
+        self._target = target
+        self._merges_map = merges_map
+
+        self._commits = {}
+        self._root = {}
+        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
+        pass
+
+    def build_graph(self):
+        g = nx.DiGraph(name='DVC Workflow', directed=False)
+
+        if not self._root:
+            raise WorkflowError('cannot find root commit')
+
+        # 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'
+                    })
+
+        A = nx.nx_agraph.to_agraph(g)
+        fname = 'workflow'
+        A = A.to_undirected()
+        A.write(fname + '.dot')
+        A.draw(fname, format='jpeg', prog='dot')
+        pass
+
+    def _build_graph(self, curr, g):
+        if '' not in self._commits or curr.hash not in self._commits:
+            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)

functest/tutorial_branches.sh

@@ -0,0 +1,102 @@
+#!/bin/bash
+
+set -e
+
+# 1. First ML model
+
+mkdir myrepo
+cd myrepo
+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 \
+        https://s3-us-west-2.amazonaws.com/dvc-share/so/code/train_model.py \
+        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
+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 run tar zxf data/Posts.xml.tgz -C data/
+
+dvc run python code/xml_to_tsv.py data/Posts.xml data/Posts.tsv python
+dvc run python code/split_train_test.py data/Posts.tsv 0.33 20170426 data/Posts-train.tsv data/Posts-test.tsv
+dvc run python code/featurization.py data/Posts-train.tsv data/Posts-test.tsv data/matrix-train.p data/matrix-test.p
+
+dvc run python code/train_model.py data/matrix-train.p 20170426 data/model.p
+
+dvc run python code/evaluate.py data/model.p data/matrix-test.p data/eval_auc.txt
+
+dvc target data/eval_auc.txt
+cat data/eval_auc.txt
+# AUC: 0.349
+
+
+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/
+dvc repro
+cat data/eval_auc.txt
+# 0.596182
+vi code/train_model.py  # estimators=500
+git commit -am 'estimators=500'
+dvc repro
+cat data/eval_auc.txt
+# 0.619262
+vi code/featurization.py
+git commit -am 'Add bigrams'
+dvc repro
+cat data/eval_auc.txt
+# 0.628989
+vi code/featurization.py
+git commit -am 'Add three-grams'
+dvc repro
+cat data/eval_auc.txt
+# 0.630682
+vi code/featurization.py
+git commit -am 'Add 4-grams'
+dvc repro
+cat data/eval_auc.txt
+# 0.621002
+
+git checkout 2195f1032 -b three_grams # <--
+vi code/featurization.py
+# 0.578447
+
+
+
+
+
+
+
+
+# 2. Reproduce: change input dataset
+
+dvc remove data/Posts.xml.tgz
+#dvc import https://s3-us-west-2.amazonaws.com/dvc-share/so/100K/Posts.xml.tgz data/
+dvc import https://s3-us-west-2.amazonaws.com/dvc-share/so/25K/Posts.xml.tgz data/
+dvc repro data/evaluation.txt
+cat data/evaluation.txt
+# AUC: 0.639861
+
+# 3. Share your research
+
+# Improve features
+echo " " >> code/featurization.py
+git add code/featurization.py
+git commit -m 'Include bigram'
+dvc repro data/evaluation.txt
+cat data/evaluation.txt