diff --git a/dvc/cli.py b/dvc/cli.py index ba42d9f5b..9553e8cd3 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -113,6 +113,10 @@ def parse_args(argv=None): action='store_true', default=False, help='Lock data item - disable reproduction.') + run_parser.add_argument('-f', + '--file', + default='stage.dvc', + help='Specify name of the state file') run_parser.add_argument( 'command', nargs=argparse.REMAINDER, diff --git a/dvc/command/add.py b/dvc/command/add.py index 60c1b8951..cbfd1ec96 100644 --- a/dvc/command/add.py +++ b/dvc/command/add.py @@ -3,6 +3,7 @@ import os from dvc.command.common.base import CmdBase from dvc.logger import Logger from dvc.state_file import StateFile +from dvc.path.data_item import DataItem class CmdAdd(CmdBase): @@ -39,9 +40,9 @@ class CmdAdd(CmdBase): for data_item in targets: Logger.debug('Creating state file for {}'.format(data_item.data.relative)) - state_file = StateFile(data_item=data_item, + state_file = StateFile(fname=data_item.data.relative + StateFile.STATE_FILE_SUFFIX, cmd=None, - out=StateFile.parse_deps_state(self.settings, [data_item.data.dvc]), + out=StateFile.parse_deps_state(self.settings, [data_item.data.relative], currdir=data_item.data.dirname), out_git=[], deps=[], locked=True) diff --git a/dvc/command/checkout.py b/dvc/command/checkout.py index 9796c35fa..f163b934e 100644 --- a/dvc/command/checkout.py +++ b/dvc/command/checkout.py @@ -35,30 +35,6 @@ class CmdCheckout(CmdBase): Logger.info('Checkout \'{}\''.format(item.data.relative)) def run(self): - try: - states = self.git.state_files_for_previous_commit() - prev_items = self.settings.path_factory.data_items_from_states(states, existing=False) - except Exception as ex: - Logger.error(u'Unable to get data files from previous commit: {}'.format(ex)) - return 1 - - curr_items = self.settings.path_factory.all_existing_data_items() - self.checkout(curr_items) - - self.remove_files(list(set(prev_items) - set(curr_items))) + items = self.settings.path_factory.all_existing_data_items() + self.checkout(items) return 0 - - @staticmethod - def remove_files(removed_items_set): - for item in removed_items_set: - path = item.data.relative - if not os.path.exists(path): - Logger.debug(u'File \'{}\' does not exist'.format(path)) - else: - Logger.info(u'Remove \'{}\''.format(path)) - os.remove(path) - dir = os.path.dirname(path) - if not os.listdir(dir): - Logger.info(u'Remove directory \'{}\''.format(dir)) - os.removedirs(dir) - pass diff --git a/dvc/command/gc.py b/dvc/command/gc.py index ada0bac8e..7d138a553 100644 --- a/dvc/command/gc.py +++ b/dvc/command/gc.py @@ -1,49 +1,23 @@ import os -from dvc.command.common.traverse import Traverse +from dvc.command.common.base import CmdBase from dvc.logger import Logger -from dvc.path.data_item import DataItemError from dvc.config import ConfigI +from dvc.state_file import StateFile -class CmdGC(Traverse): +class CmdGC(CmdBase): def __init__(self, settings): - super(CmdGC, self).__init__(settings, "garbage collect", do_not_start_from_root=False) - self.clist = [] + super(CmdGC, self).__init__(settings) def run(self): - if not self._traverse(self.git.git_dir_abs): - Logger.error('Failed to collect used cache') - return 1 + clist = [str(x) for x in StateFile.find_all_cache_files(self.git)] for cache in os.listdir(ConfigI.CACHE_DIR): fname = os.path.join(ConfigI.CACHE_DIR, cache) - if fname in self.clist: + if os.path.basename(fname) in clist: continue os.remove(fname) - self._remove_cloud_cache(self.settings.path_factory.data_item('.empty', fname)) Logger.info('Cache \'{}\' was removed'.format(fname)) return 0 - - def process_file(self, target): - Logger.debug(u'[Cmd-GC] GC file {}.'.format(target)) - - try: - data_item = self.settings.path_factory.existing_data_item(target) - except DataItemError: - return - - self.clist.append(data_item.cache.relative) - Logger.debug(u'[Cmd-GC] GC data item {}. Success.'.format(data_item.data.relative)) - - def is_recursive(self): - return self.parsed_args.recursive - - @property - def no_git_actions(self): - return True - - @staticmethod - def not_committed_changes_warning(): - pass diff --git a/dvc/command/merge.py b/dvc/command/merge.py index 2a9566344..1d6e8436d 100644 --- a/dvc/command/merge.py +++ b/dvc/command/merge.py @@ -14,44 +14,35 @@ class CmdMerge(CmdBase): super(CmdMerge, self).__init__(settings) def print_info(self, targets): - for item in targets: + for fname in targets: Logger.info('Restored original data after merge:') - Logger.info(' {}'.format(item.data.relative)) - - def collect_data(self): - dlist = [] - flist = self.git.get_last_merge_changed_files() - for fname in flist: - if fname.endswith(DataItem.STATE_FILE_SUFFIX): - data = fname[:-len(DataItem.STATE_FILE_SUFFIX)] - dlist.append(data) - return dlist + Logger.info(' {}'.format(fname)) def collect_targets(self): targets = [] - flist = self.collect_data() - items = self.settings.path_factory.to_data_items(flist)[0] - for item in items: - command = StateFile.load(item) + for fname in self.git.get_last_merge_changed_files(): + if not StateFile._is_state_file(fname): + continue - if not command.cmd and command.locked: - targets.append(item) + state = StateFile.load(fname) + if not state.cmd and state.locked: + targets.append(fname) return targets def checkout_targets(self, targets): - data = [] - for item in targets: - prev_state = StateFile.loads(self.git.get_file_content_before_last_merge(item.state.relative)) - curr_state = StateFile.load(item) + items = [] + for fname in targets: + self.git.checkout_file_before_last_merge(fname) + state = StateFile.load(fname) + for out in state.out: + item = self.settings.path_factory.data_item(os.path.join(state.cwd, out)) + items.append(item) - curr_state.out[item.data.dvc] = prev_state.out[item.data.dvc] - curr_state.save() + CmdCheckout.checkout(items) - CmdCheckout.checkout([item]) - - msg = 'DVC merge files: {}'.format(' '.join(data)) + msg = 'DVC merge files: {}'.format(' '.join(targets)) self.commit_if_needed(msg) def run(self): diff --git a/dvc/command/repro.py b/dvc/command/repro.py index 2544664e7..0919f530a 100644 --- a/dvc/command/repro.py +++ b/dvc/command/repro.py @@ -88,7 +88,7 @@ class ReproChange(object): self._recursive = recursive self._force = force - self.command = StateFile.load(data_item) + self.command = StateFile.find(data_item) if not self.command.cmd and not self.command.locked: msg = 'Error: data item "{}" is not locked, but has no command for reproduction' @@ -105,15 +105,15 @@ class ReproChange(object): return self._cmd_obj def remove_output_files(self): - for output_dvc in self.command.out: - Logger.debug('Removing output file {} before reproduction.'.format(output_dvc)) - + for out in self.command.out: + path = os.path.join(self.command.cwd, out) + Logger.debug('Removing output file {} before reproduction.'.format(path)) try: - data_item = self.cmd_obj.settings.path_factory.data_item_from_dvc_path(output_dvc) + data_item = self.cmd_obj.settings.path_factory.data_item(path) os.remove(data_item.data.relative) except Exception as ex: msg = 'Data item {} cannot be removed before reproduction: {}' - Logger.debug(msg.format(output_dvc, ex)) + Logger.debug(msg.format(path, ex)) def reproduce_run(self): Logger.info('Reproducing run command for data item {}. Args: {}'.format( @@ -169,7 +169,8 @@ class ReproChange(object): def reproduce_deps(self, data_item_dvc, recursive): result = False - for path,md5 in self.command.deps.items(): + for name,md5 in self.command.deps.items(): + path = os.path.join(self.command.cwd, name) if self.reproduce_dep(path, md5, recursive): result = True diff --git a/dvc/command/run.py b/dvc/command/run.py index a1a53b59e..a9d4b9bed 100644 --- a/dvc/command/run.py +++ b/dvc/command/run.py @@ -19,44 +19,40 @@ class CmdRun(CmdBase): def run(self): cmd = ' '.join(self.parsed_args.command) - state = StateFile(data_item=None, - cmd=cmd, - out=self.parsed_args.out, - out_git=self.parsed_args.out_git, - deps=self.parsed_args.deps, - locked=self.parsed_args.lock) + + if os.path.isfile(self.parsed_args.file): + Logger.error("Stage file {} already exists".format(self.parsed_args.file)) + return 1 + + state = StateFile(fname=self.parsed_args.file, + cmd=cmd, + out=self.parsed_args.out, + out_git=self.parsed_args.out_git, + deps=self.parsed_args.deps, + locked=self.parsed_args.lock) self.run_command(self.settings, state) return self.commit_if_needed('DVC run: {}'.format(state.cmd)) @staticmethod def run_command(settings, state): - Executor.exec_cmd_only_success(state.cmd, shell=True) + Executor.exec_cmd_only_success(state.cmd, cwd=state.cwd, shell=True) - CmdRun.apply_to_files(state.out, state, CmdRun._create_cache_and_state_files, settings) - CmdRun.apply_to_files(state.out_git, state, CmdRun._create_state_file, settings) + CmdRun.move_output_to_cache(settings, state) + CmdRun.update_state_file(settings, state) @staticmethod - def apply_to_files(files, state, func, settings): - items = settings.path_factory.to_data_items(files)[0] - [func(i, state, settings) for i in items] + def update_state_file(settings, state): + Logger.debug('Update state file "{}"'.format(state.path)) + state.out = StateFile.parse_deps_state(settings, state.out) + state.out_git = StateFile.parse_deps_state(settings, state.out_git) + state.deps = StateFile.parse_deps_state(settings, state.deps) + state.save() @staticmethod - def _create_cache_and_state_files(data_item, state, settings): - Logger.debug('Move output file "{}" to cache dir "{}" and create a hardlink'.format( - data_item.data.relative, data_item.cache_dir_abs)) - data_item.move_data_to_cache() - return CmdRun._create_state_file(data_item, state, settings) - - @staticmethod - def _create_state_file(data_item, state, settings): - Logger.debug('Create state file "{}"'.format(data_item.state.relative)) - - state_file = StateFile(data_item=data_item, - cmd=state.cmd, - out=StateFile.parse_deps_state(settings, state.out), - out_git=StateFile.parse_deps_state(settings, state.out_git), - locked=state.locked, - deps=StateFile.parse_deps_state(settings, state.deps)) - state_file.save() - return state_file + def move_output_to_cache(settings, state): + items = settings.path_factory.to_data_items(state.out)[0] + for item in items: + Logger.debug('Move output file "{}" to cache dir "{}" and create a hardlink'.format( + item.data.relative, item.cache_dir_abs)) + item.move_data_to_cache() diff --git a/dvc/path/data_item.py b/dvc/path/data_item.py index 277c7b212..f9592919d 100644 --- a/dvc/path/data_item.py +++ b/dvc/path/data_item.py @@ -33,7 +33,6 @@ class NotInGitDirError(DataDirError): class DataItem(object): - STATE_FILE_SUFFIX = '.dvc' LOCAL_STATE_FILE_PREFIX = '.' LOCAL_STATE_FILE_SUFFIX = '.dvc_local_state' CACHE_FILE_SEP = '_' @@ -78,7 +77,7 @@ class DataItem(object): @cached_property def state(self): - return self._state('', self.STATE_FILE_SUFFIX) + return Path(StateFile.find(self).path, self._git) @cached_property def cache_dir_abs(self): @@ -99,7 +98,7 @@ class DataItem(object): if self._cache_file: file_name = os.path.relpath(os.path.realpath(self._cache_file), cache_dir) else: - file_name = StateFile.load(self).out[self.data.dvc] + file_name = str(StateFile.find_md5(self)) cache_file = os.path.join(cache_dir, file_name) return Path(cache_file, self._git) diff --git a/dvc/path/factory.py b/dvc/path/factory.py index 0793994cc..f16f01bb1 100644 --- a/dvc/path/factory.py +++ b/dvc/path/factory.py @@ -6,6 +6,7 @@ from dvc.path.data_item import DataItem, DataItemError, DataDirError from dvc.path.path import Path from dvc.path.stated_data_item import StatedDataItem from dvc.system import System +from dvc.state_file import StateFile class PathFactory(object): @@ -31,7 +32,8 @@ class PathFactory(object): return self.data_item(path.relative) def is_data_item(self, fname): - return os.path.isfile(fname + DataItem.STATE_FILE_SUFFIX) + item = DataItem(fname, self._git, self._config) + return StateFile.find(item) != None def existing_data_item(self, fname): if not self.is_data_item(fname): @@ -53,27 +55,5 @@ class PathFactory(object): return result, externally_created_files def all_existing_data_items(self, subdir='.', cache_exists=True): - states = [] - - for root, dirs, files in os.walk(os.path.join(self._git.git_dir_abs, subdir)): - for fname in files: - path = os.path.join(root, fname) - - if not fname.endswith(DataItem.STATE_FILE_SUFFIX): - continue - - states.append(path) - - data_items = self.data_items_from_states(states) - if cache_exists: - data_items = filter(lambda i: os.path.exists(i.state.relative), data_items) - - return data_items - - def data_items_from_states(self, states, existing=True): - return [self.data_item_from_dvc_path(self.state_path_to_dvc_path(s), existing) - for s in states] - - @staticmethod - def state_path_to_dvc_path(state): - return re.sub(DataItem.STATE_FILE_SUFFIX + '$', '', state) + files = StateFile.find_all_data_files(self._git, subdir) + return self.to_data_items(files)[0] diff --git a/dvc/state_file.py b/dvc/state_file.py index 395f623af..897ae6159 100644 --- a/dvc/state_file.py +++ b/dvc/state_file.py @@ -3,6 +3,7 @@ import yaml from dvc.exceptions import DvcException from dvc.data_cloud import file_md5 +from dvc.path.path import Path class StateFileError(DvcException): @@ -24,8 +25,7 @@ class StateFileBase(object): class StateFile(StateFileBase): - MAGIC = 'DVC-State' - VERSION = '0.1' + STATE_FILE_SUFFIX = '.dvc' PARAM_CMD = 'cmd' PARAM_DEPS = 'deps' @@ -34,14 +34,13 @@ class StateFile(StateFileBase): PARAM_OUT_GIT = 'out-git' def __init__(self, - data_item=None, + fname=None, cmd=None, out=None, out_git=None, deps=None, locked=None): super(StateFile, self).__init__() - self.data_item = data_item self.cmd = cmd self.out = out @@ -49,24 +48,30 @@ class StateFile(StateFileBase): self.deps = deps self.locked = locked + self.path = os.path.abspath(fname) if fname else None + self.cwd = os.path.dirname(self.path) if self.path else None + @staticmethod - def parse_deps_state(settings, deps): + def parse_deps_state(settings, deps, currdir=None): state = {} for dep in deps: if settings.path_factory.is_data_item(dep): item = settings.path_factory.data_item(dep) - state[item.data.dvc] = StateFile.load(item).out[item.data.dvc] + md5 = StateFile.find_md5(item) else: - state[dep] = file_md5(os.path.join(settings.git.git_dir_abs, dep))[0] + md5 = file_md5(os.path.join(settings.git.git_dir_abs, dep))[0] + + if currdir: + name = os.path.relpath(dep, currdir) + else: + name = dep + + state[name] = md5 return state - @property - def file(self): - return self.data_item.state.relative - @staticmethod - def loadd(data): - return StateFile(data_item=None, + def loadd(data, fname=None): + return StateFile(fname=fname, cmd=data.get(StateFile.PARAM_CMD, None), out=data.get(StateFile.PARAM_OUT, None), out_git=data.get(StateFile.PARAM_OUT_GIT, None), @@ -74,8 +79,86 @@ class StateFile(StateFileBase): locked=data.get(StateFile.PARAM_LOCKED, None)) @staticmethod - def load(data_item): - return StateFile._load(data_item.state.relative, StateFile) + def load(fname): + return StateFile._load(fname, StateFile, fname) + + @staticmethod + def _is_state_file(path): + return path.endswith(StateFile.STATE_FILE_SUFFIX) and os.path.isfile(path) + + @staticmethod + def _find_state(fname, dname): + name = os.path.relpath(fname, dname) + for entry in os.listdir(dname): + state_file = os.path.join(dname, entry) + if not StateFile._is_state_file(state_file): +# if not state_file.endswith(StateFile.STATE_FILE_SUFFIX) or not os.path.isfile(state_file): + continue + state = StateFile.load(os.path.join(dname, state_file)) + if name in state.out: + return state + return None + + @staticmethod + def _find(name, start, finish): + dname = start + fname = name + while True: + state = StateFile._find_state(fname, dname) + if state: + return state + + if dname == finish: + break + + dname = os.path.dirname(dname) + return None + + @staticmethod + def find(data_item): + return StateFile._find(data_item.data.abs, data_item.data.dirname, data_item._git.git_dir_abs) + + @staticmethod + def find_all_state_files(git, subdir='.'): + states = [] + for root, dirs, files in os.walk(os.path.join(git.git_dir_abs, subdir)): + for fname in files: + path = os.path.join(root, fname) + + if not StateFile._is_state_file(path): + continue + + states.append(path) + return states + + @staticmethod + def find_all_states(git, subdir='.'): + state_files = StateFile.find_all_state_files(git, subdir) + return [StateFile.load(state_file) for state_file in state_files] + + @staticmethod + def find_all_data_files(git, subdir='.'): + states = StateFile.find_all_states(git, subdir) + files = [] + for state in states: + for out in state.out: + files.append(os.path.join(state.cwd, out)) + return files + + @staticmethod + def find_all_cache_files(git, subdir='.'): + states = StateFile.find_all_states(git, subdir) + cache_files = [] + for state in states: + for out,md5 in state.out.items(): + cache_files.append(md5) + return cache_files + + @staticmethod + def find_md5(data_item): + state = StateFile.find(data_item) + name = os.path.relpath(data_item.data.abs, os.path.dirname(state.path)) + return state.out[name] @staticmethod def loads(content): @@ -91,11 +174,7 @@ class StateFile(StateFileBase): self.PARAM_LOCKED: self.locked } - file_dir = os.path.dirname(self.file) - if file_dir != '' and not os.path.isdir(file_dir): - os.makedirs(file_dir) - - self._save(self.file, res) + self._save(self.path, res) class LocalStateFile(StateFileBase): diff --git a/dvc/utils.py b/dvc/utils.py index 2a43660ec..0e5cb5294 100644 --- a/dvc/utils.py +++ b/dvc/utils.py @@ -79,7 +79,7 @@ def map_progress(func, targets, n_threads): try: ret = pool.map(func, targets) except Exception as exc: - Logger.error('Unexpected exception while processing targets: {}'.format(exc)) + Logger.error('Unexpected exception while processing targets: {}'.format(exc), exc_info=True) finally: progress.finish() diff --git a/functest/test-repro-code.sh b/functest/test-repro-code.sh index d8e592186..304efbe31 100755 --- a/functest/test-repro-code.sh +++ b/functest/test-repro-code.sh @@ -10,7 +10,7 @@ function test_generic() { dvc_create_repo dvc_info 'Copy foo into foo1' - dvc run -d code/code.sh -d data/foo -o data/foo1 bash code/code.sh data/foo data/foo1 + dvc run -f copy_foo_foo1.dvc -d code/code.sh -d data/foo -o data/foo1 bash code/code.sh data/foo data/foo1 dvc_info 'Modify code' echo " " >> code/code.sh @@ -52,9 +52,9 @@ function test_partial() { git commit -m "copy code" dvc_info "Create repro chain foo -> foo1 -> foo2 -> foo3" - dvc run -d code/code1.sh -d data/foo -o data/foo1 cp data/foo data/foo1 - dvc run -d code/code2.sh -d data/foo1 -o data/foo2 cp data/foo1 data/foo2 - dvc run -d code/code3.sh -d data/foo2 -o data/foo3 cp data/foo2 data/foo3 + dvc run -f copy_foo_foo1.dvc -d code/code1.sh -d data/foo -o data/foo1 cp data/foo data/foo1 + dvc run -f copy_foo1_foo2.dvc -d code/code2.sh -d data/foo1 -o data/foo2 cp data/foo1 data/foo2 + dvc run -f copy_foo2_foo3.dvc -d code/code3.sh -d data/foo2 -o data/foo3 cp data/foo2 data/foo3 dvc_info "Save original timestamps" FOO_TS=$(dvc_timestamp data/foo)