From a61c5a8c7f53bf647d6bec13e438f7df3670c2ef Mon Sep 17 00:00:00 2001 From: Dmitry Petrov Date: Fri, 28 Apr 2017 12:56:27 -0700 Subject: [PATCH 1/5] Repro data file without cache file. Close #53 --- dvc/command/repro.py | 74 +++++++++++++++++++++++++++++++------------- 1 file changed, 52 insertions(+), 22 deletions(-) diff --git a/dvc/command/repro.py b/dvc/command/repro.py index 4e1a58e87..cef5a9935 100644 --- a/dvc/command/repro.py +++ b/dvc/command/repro.py @@ -92,8 +92,8 @@ class CmdRepro(CmdRun): globally_changed_files = self.git.get_changed_files(target_commit) changed_files = set() - repro_change = ReproChange(data_item, self) - if repro_change.reproduce(changed_files, globally_changed_files, force): + repro_change = ReproChange(data_item, self, globally_changed_files, force) + if repro_change.reproduce(changed_files): changed = True Logger.info(u'Data item "{}" was reproduced.'.format( data_item.data.relative @@ -117,13 +117,15 @@ class CmdRepro(CmdRun): class ReproChange(object): - def __init__(self, data_item, cmd_obj): + def __init__(self, data_item, cmd_obj, globally_changed_files, force): self._data_item = data_item self.git = cmd_obj.git self._cmd_obj = cmd_obj + self._globally_changed_files = globally_changed_files + self._force = force if not System.islink(data_item.data.relative): - raise ReproError('data item {} is no symlink'.format(data_item.data.relative)) + raise ReproError('data item {} is not symlink'.format(data_item.data.relative)) try: self._state = StateFile.load(data_item.state.relative, self.git) @@ -131,8 +133,6 @@ class ReproChange(object): raise ReproError('Error: state file "{}" cannot be loaded: {}'. format(data_item.state.relative, ex)) - cmd_obj._code = self.state.code_dependencies - if not self.state.argv: raise ReproError('Error: parameter {} is not defined in state file "{}"'. format(StateFile.PARAM_ARGV, data_item.state.relative)) @@ -210,42 +210,34 @@ class ReproChange(object): pass pass - def reproduce(self, changed_files, globally_changed_files, force=False): + 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, - force, - ', '.join([x.data.dvc for x in input_files_dependencies]))) + self._force, + ', '.join([x.data.dvc for x in input_files_dependencies]))) - if not force and not self.state.is_reproducible: + if not self._force and 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) - if change.reproduce(changed_files, globally_changed_files, force=False): + 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 )) else: - if data_item.data.dvc in globally_changed_files: + 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 - were_sources_changed = self.git.were_files_changed( - self.state.code_dependencies, - self.cmd_obj.settings.path_factory, - globally_changed_files - ) - - is_change_needed = were_sources_changed or were_input_files_changed or force - # if not self.is_cache_exists() or (is_change_needed and data_item_dvc not in changed_files): - if is_change_needed and data_item_dvc not in changed_files: + 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 @@ -254,6 +246,44 @@ class ReproChange(object): return False pass + def is_repro_required(self, changed_files, data_item_dvc, + were_input_files_changed): + is_repro_required = False + + if self.is_cache_exists(): + is_repro_required = True + self.log_repro_reason(u'cache file is missing') + + if data_item_dvc not in changed_files: + is_repro_required = True + self.log_repro_reason(u'the data item was changed') + + if not is_repro_required and were_input_files_changed: + is_repro_required = True + self.log_repro_reason(u'input dependencies were changed') + + 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 + + def log_repro_reason(self, reason): + msg = u'Repro is required for data item {} because of {}' + Logger.debug(msg.format(self._data_item.data.relative, reason)) + + def were_sources_changed(self, globally_changed_files): + were_sources_changed = self.git.were_files_changed( + self.state.code_dependencies, + self.cmd_obj.settings.path_factory, + globally_changed_files + ) + return were_sources_changed + @property def dependencies(self): dependency_data_items = [] From 55111b082dc110bd8a41e36287e0b9f72a02aa9a Mon Sep 17 00:00:00 2001 From: Dmitry Petrov Date: Fri, 28 Apr 2017 14:12:45 -0700 Subject: [PATCH 2/5] Repro: separate --recursive option from regular, one step repro. Close #54 --- dvc/command/repro.py | 125 ++++++++++++++++++++++--------------------- 1 file changed, 65 insertions(+), 60 deletions(-) diff --git a/dvc/command/repro.py b/dvc/command/repro.py index cef5a9935..79b4b1bc8 100644 --- a/dvc/command/repro.py +++ b/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 {}' From 815f93c5f081877bcbf9e09829fc6ed5d101ad28 Mon Sep 17 00:00:00 2001 From: Dmitry Petrov Date: Fri, 28 Apr 2017 14:57:20 -0700 Subject: [PATCH 3/5] Repro fixes: Force dependency check if cache file is missing --- dvc/command/repro.py | 25 +++++++++++++++---------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/dvc/command/repro.py b/dvc/command/repro.py index 79b4b1bc8..84aefa32e 100644 --- a/dvc/command/repro.py +++ b/dvc/command/repro.py @@ -144,7 +144,8 @@ class ReproChange(object): pass def is_cache_exists(self): - return os.path.exists(self._data_item.cache.relative) + path = System.realpath(self._data_item.data.relative) + return os.path.exists(path) @property def cmd_obj(self): @@ -255,22 +256,26 @@ class ReproChange(object): 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 + is_dependency_check_required = self._recursive + + if not is_dependency_check_required and not self.is_cache_exists(): + is_dependency_check_required = True + Logger.info(u'Reproduction {}. Force dependency check since cache file is missing.'.format( + self._data_item.data.relative)) + + if is_dependency_check_required: + if 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(): - self.log_repro_reason(u'cache file is missing') + if not self.is_cache_exists(): + self.log_repro_reason(u'cache file is missing.') return True - # if data_item_dvc not in changed_files: - # self.log_repro_reason(u'the data item was changed') - # return True - if self.were_sources_changed(self._globally_changed_files): self.log_repro_reason(u'sources were changed') return True From 73980a1a4bb55ebfb6639beb11d02ee07d1d3f00 Mon Sep 17 00:00:00 2001 From: Dmitry Petrov Date: Fri, 28 Apr 2017 14:58:06 -0700 Subject: [PATCH 4/5] Sync: warning if file cannot be synced --- dvc/command/data_sync.py | 22 ++++++++++++---------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index 66ec4b7ac..8d093a954 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -66,21 +66,24 @@ class CmdDataSync(CmdBase): elif os.path.isdir(target): self.sync_dir(target) else: - # if data_item.data.dvc != empty_data_item.data.dvc: raise DataSyncError('File "{}" does not exit'.format(target)) finally: if self.is_locker: lock.release() def sync_dir(self, dir): - for f in os.listdir(dir): - fname = os.path.join(dir, f) - if os.path.isdir(fname): - self.sync_dir(fname) - elif System.islink(fname): - self.sync_symlink(self.settings.path_factory.data_item(fname)) - else: - raise DataSyncError('Unsupported file type "{}"'.format(fname)) + for file in os.listdir(dir): + try: + fname = os.path.join(dir, file) + if os.path.isdir(fname): + self.sync_dir(fname) + elif System.islink(fname): + self.sync_symlink(self.settings.path_factory.data_item(fname)) + else: + raise DataSyncError('Unsupported file type "{}"'.format(fname)) + except DataSyncError as ex: + Logger.warn(ex) + pass def sync_symlink(self, data_item): if os.path.isfile(data_item.resolved_cache.dvc): @@ -181,7 +184,6 @@ class CmdDataSync(CmdBase): blob.upload_from_filename(data_item.resolved_cache.relative) Logger.info('uploading %s completed' % data_item.resolved_cache.relative) - def sync_from_cloud(self, item): cloud = self.settings.config.cloud assert cloud in ['amazon', 'google'], 'unknown cloud %s' % cloud From a7f8989575d1ccf2ac08a2e1c5afce3e741d1d4b Mon Sep 17 00:00:00 2001 From: Dmitry Petrov Date: Fri, 28 Apr 2017 15:03:33 -0700 Subject: [PATCH 5/5] Unit tests --- tests/test_repro.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_repro.py b/tests/test_repro.py index 6bd78eaf4..2a363678a 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -86,7 +86,7 @@ class ReproChangedDependency(ReproBasicEnv): def test(self): self.recreate_file1() - self.settings._args = [self.file_name11] + self.settings._args = [self.file_name11, '-r'] CmdRepro(self.settings).run() self.assertEqual(open(self.file_name11).read(), 'Goodbye\n') @@ -107,7 +107,7 @@ class ReproChangedDeepDependency(ReproChangedDependency): def test(self): self.recreate_file1() - self.settings._args = [self.file_name_res] + self.settings._args = [self.file_name_res, '-r'] CmdRepro(self.settings).run() self.assertEqual(open(self.file_name_res).read().strip(), 'Goodbye\nBobby')