Repro: separate --recursive option from regular, one step repro. Close #54

dvc/command/repro.py

@@ -31,9 +31,11 @@ class CmdRepro(CmdRun):
     def define_args(self, parser):
         self.set_skip_git_actions(parser)
 
-        parser.add_argument('target', metavar='', help='Data item to reproduce', nargs='*')
+        parser.add_argument('target', metavar='', help='Data items to reproduce.', nargs='*')
         parser.add_argument('-f', '--force', action='store_true', default=False,
-                            help='Force reproduction')
+                            help='Reproduce even if dependencies were not changed')
+        parser.add_argument('-r', '--recursive', action='store_true', default=False,
+                            help='Recursively reproduce all changed dependencies')
         pass
 
     @property
@@ -56,14 +58,14 @@ class CmdRepro(CmdRun):
                 self.warning_dvc_is_busy()
                 return 1
         try:
-            return self.repro_target(self.parsed_args.target, self.parsed_args.force)
+            return self.repro_target(self.parsed_args.target, self.parsed_args.recursive, self.parsed_args.force)
         finally:
             if self.is_locker:
                 lock.release()
 
         pass
 
-    def repro_target(self, target, force):
+    def repro_target(self, target, recursive, force):
         if not self.skip_git_actions and not self.git.is_ready_to_go():
             return 1
 
@@ -73,11 +75,11 @@ class CmdRepro(CmdRun):
                          format(self.config.data_dir, ' '.join(external_files_names)))
             return 1
 
-        if self.repro_data_items(data_item_list, force):
+        if self.repro_data_items(data_item_list, recursive, force):
             return 0
         pass
 
-    def repro_data_items(self, data_item_list, force):
+    def repro_data_items(self, data_item_list, recursive, force):
         error = False
         changed = False
 
@@ -91,9 +93,8 @@ class CmdRepro(CmdRun):
 
                 globally_changed_files = self.git.get_changed_files(target_commit)
                 changed_files = set()
-
-                repro_change = ReproChange(data_item, self, globally_changed_files, force)
-                if repro_change.reproduce(changed_files):
+                change = ReproChange(data_item, self, globally_changed_files, recursive, force)
+                if change.reproduce(changed_files):
                     changed = True
                     Logger.info(u'Data item "{}" was reproduced.'.format(
                         data_item.data.relative
@@ -117,11 +118,12 @@ class CmdRepro(CmdRun):
 
 
 class ReproChange(object):
-    def __init__(self, data_item, cmd_obj, globally_changed_files, force):
+    def __init__(self, data_item, cmd_obj, globally_changed_files, recursive, force):
         self._data_item = data_item
         self.git = cmd_obj.git
         self._cmd_obj = cmd_obj
         self._globally_changed_files = globally_changed_files
+        self._recursive = recursive
         self._force = force
 
         if not System.islink(data_item.data.relative):
@@ -153,7 +155,7 @@ class ReproChange(object):
         return self._state
 
     def reproduce_data_item(self):
-        Logger.debug('Reproducing data item "{}".'.format(self._data_item.data.dvc))
+        Logger.debug('Reproducing data item {}.'.format(self._data_item.data.dvc))
 
         for output_dvc in self._state.output_files:
             Logger.debug('Removing output file {} before reproduction.'.format(output_dvc))
@@ -169,7 +171,7 @@ class ReproChange(object):
         settings.set_args(self.state.argv)
 
         if self.state.is_import_file:
-            Logger.debug('Reproducing data item "{}". Re-import cmd: {}'.format(
+            Logger.debug('Reproducing data item {}. Re-import cmd: {}'.format(
                 self._data_item.data.relative, ' '.join(self.state.argv)))
 
             if len(self.state.argv) != 2:
@@ -192,7 +194,7 @@ class ReproChange(object):
             cmd.set_git_action(True)
             cmd.set_locker(False)
 
-            Logger.info('Reproducing run command: {}. Args: {}'.format(
+            Logger.info('Reproducing run command for data item {}. Args: {}'.format(
                 self._data_item.data.relative, ' '.join(self.state.argv)))
 
             data_items_from_args, not_data_items_from_args = self.cmd_obj.argv_files_by_type(self.state.argv)
@@ -210,67 +212,70 @@ class ReproChange(object):
             pass
         pass
 
-    def reproduce(self, changed_files, force=False):
-        input_files_dependencies = self.dependencies
-        data_item_dvc = self._data_item.data.dvc
-        Logger.debug('Reproduce data item {} with dependencies, force={}: {}'.format(
-            data_item_dvc,
-            self._force,
-            ', '.join([x.data.dvc for x in input_files_dependencies])))
+    def reproduce(self, changed_files):
+        Logger.debug('Reproduce data item {}. recursive={}, force={}'.format(
+            self._data_item.data.relative, self._recursive, self._force))
 
-        if not self._force and not self.state.is_reproducible:
-            Logger.debug('Data item "{}" is not reproducible'.format(self._data_item.data.relative))
+        if not self.state.is_reproducible:
+            Logger.debug('Data item {} is not reproducible'.format(self._data_item.data.relative))
             return False
 
-        were_input_files_changed = False
-        for data_item in input_files_dependencies:
-            change = ReproChange(data_item, self._cmd_obj, self._globally_changed_files, force=False)
-            if change.reproduce(changed_files):
-                were_input_files_changed = True
-                Logger.debug('Repro: data dependency {} was changed for data item {}'.format(
-                    data_item.data.dvc,
-                    data_item_dvc
-                ))
+        data_item_dvc = self._data_item.data.dvc
+        if self.is_repro_required(changed_files, data_item_dvc):
+            if data_item_dvc not in changed_files:
+                Logger.debug('Data item {} is going to be reproduced'.format(self._data_item.data.relative))
+                self.reproduce_data_item()
+                changed_files.add(data_item_dvc)
+                return True
             else:
-                if data_item.data.dvc in self._globally_changed_files:
-                    msg = 'Repro: data dependency {} was not changed for data item {} but global checksum was changed'
-                    Logger.debug(msg.format(data_item.data.dvc, data_item_dvc))
-                    were_input_files_changed = True
-
-        if self.is_repro_required(changed_files, data_item_dvc, were_input_files_changed):
-            self.reproduce_data_item()
-            changed_files.add(data_item_dvc)
-            return True
+                msg = 'Data item {} is not going to be reproduced because it is already reproduced'
+                Logger.debug(msg.format(self._data_item.data.relative))
         else:
-            Logger.debug('Data item "{}" is up to date'.format(self._data_item.data.relative))
+            Logger.debug('Data item {} is up to date'.format(self._data_item.data.relative))
             return False
         pass
 
-    def is_repro_required(self, changed_files, data_item_dvc,
-                          were_input_files_changed):
-        is_repro_required = False
+    def were_dependencies_changed(self, changed_files, data_item_dvc):
+        result = False
+
+        for data_item in self.dependencies:
+            change = ReproChange(data_item, self._cmd_obj, self._globally_changed_files, self._recursive, self._force)
+            if change.reproduce(changed_files):
+                result = True
+                Logger.debug(u'Repro data item {}: dependency {} was changed'.format(
+                    data_item_dvc, data_item.data.dvc))
+            elif data_item.data.dvc in self._globally_changed_files:
+                msg = u'Repro data item {}: dependency {} was not changed but the data item global checksum was changed'
+                Logger.debug(msg.format(data_item_dvc, data_item.data.dvc))
+                result = True
+            else:
+                msg = u'Repro data item {}: dependency {} was not changed'
+                Logger.debug(msg.format(data_item_dvc, data_item.data.dvc))
+
+        return result
+
+    def is_repro_required(self, changed_files, data_item_dvc):
+        if self._recursive and self.were_dependencies_changed(changed_files, data_item_dvc):
+            self.log_repro_reason(u'input dependencies were changed')
+            return True
+
+        if self._force:
+            self.log_repro_reason(u'it was forced')
+            return True
 
         if self.is_cache_exists():
-            is_repro_required = True
             self.log_repro_reason(u'cache file is missing')
+            return True
 
-        if data_item_dvc not in changed_files:
-            is_repro_required = True
-            self.log_repro_reason(u'the data item was changed')
+        # if data_item_dvc not in changed_files:
+        #     self.log_repro_reason(u'the data item was changed')
+        #     return True
 
-        if not is_repro_required and were_input_files_changed:
-            is_repro_required = True
-            self.log_repro_reason(u'input dependencies were changed')
+        if self.were_sources_changed(self._globally_changed_files):
+            self.log_repro_reason(u'sources were changed')
+            return True
 
-        if not is_repro_required and self._force:
-            is_repro_required = True
-            self.log_repro_reason(u'it was forced')
-
-        if not is_repro_required and self.were_sources_changed(self._globally_changed_files):
-            is_repro_required = True
-            self.log_repro_reason(u'one of sources was changed')
-
-        return is_repro_required
+        return False
 
     def log_repro_reason(self, reason):
         msg = u'Repro is required for data item {} because of {}'