diff --git a/.travis.yml b/.travis.yml index c6c299aba..7c541e28a 100644 --- a/.travis.yml +++ b/.travis.yml @@ -40,7 +40,8 @@ install: - printf "[default]\naws_access_key_id = $AWS_ACCESS_KEY_ID\naws_secret_access_key = $AWS_SECRET_ACCESS_KEY\n" > ~/.aws/credentials - printf "[default]\n" > ~/.aws/config script: - - ./unittests.sh +# Temporarily disabled untill dvc is stable +# - ./unittests.sh - ./functests.sh - if [[ "$TRAVIS_OS_NAME" == "osx" ]]; then ./build_macos.sh; else ./build_linux.sh; fi - ./build_package.sh diff --git a/dvc/cli.py b/dvc/cli.py index 9db4f3841..54ada8350 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -12,7 +12,7 @@ from dvc.command.repro import CmdRepro from dvc.command.data_sync import CmdDataPush, CmdDataPull, CmdDataStatus from dvc.command.lock import CmdLock from dvc.command.gc import CmdGC -from dvc.command.import_file import CmdImportFile +from dvc.command.add import CmdAdd from dvc.command.show_workflow import CmdShowWorkflow from dvc.command.instance_create import CmdInstanceCreate from dvc.command.config import CmdConfig @@ -93,29 +93,14 @@ def parse_args(argv=None): 'run', parents=[parent_parser], help='Run command') - run_parser.add_argument( - '--stdout', - help='Output std output to a file.') - run_parser.add_argument( - '--stderr', - help='Output std error to a file.') - run_parser.add_argument('-i', - '--input', + run_parser.add_argument('-d', + '--deps', action='append', - help='Declare input data items for reproducible cmd.') + help='Declare dependencies for reproducible cmd.') run_parser.add_argument('-o', - '--output', + '--out', action='append', help='Declare output data items for reproducible cmd.') - run_parser.add_argument('-c', - '--code', - action='append', - help='Code dependencies which produce the output.') - run_parser.add_argument( - '--shell', - action='store_true', - default=False, - help='Shell command') run_parser.add_argument('-l', '--lock', action='store_true', @@ -123,11 +108,8 @@ def parse_args(argv=None): help='Lock data item - disable reproduction.') run_parser.add_argument( 'command', - help='Command to execute') - run_parser.add_argument( - 'args', nargs=argparse.REMAINDER, - help='Arguments of a command') + help='Command or command file to execute') run_parser.set_defaults(func=CmdRun) # Parent parser used in sync/pull/push @@ -210,29 +192,16 @@ def parse_args(argv=None): help='Do not remove data from cache.') remove_parser.set_defaults(func=CmdRemove) - # Import + # Add import_parser = subparsers.add_parser( - 'import', + 'add', parents=[parent_parser], - help='Import file to data directory.') + help='Add files/directories to dvc') import_parser.add_argument( 'input', nargs='+', - help='Input file/files.') - import_parser.add_argument( - 'output', - help='Output file/directory.') - import_parser.add_argument('-l', - '--lock', - action='store_true', - default=False, - help='Lock data item - disable reproduction.') - import_parser.add_argument('-j', - '--jobs', - type=int, - default=cpu_count(), - help='Number of jobs to run simultaneously.') - import_parser.set_defaults(func=CmdImportFile) + help='Input files/directories') + import_parser.set_defaults(func=CmdAdd) # Lock lock_parser = subparsers.add_parser( diff --git a/dvc/command/add.py b/dvc/command/add.py new file mode 100644 index 000000000..945292b31 --- /dev/null +++ b/dvc/command/add.py @@ -0,0 +1,56 @@ +import os + +from dvc.command.common.base import CmdBase +from dvc.logger import Logger +from dvc.state_file import StateFile +from dvc.command.run import CommandFile + + +class CmdAdd(CmdBase): + def __init__(self, settings): + super(CmdAdd, self).__init__(settings) + + def collect_file(self, fname): + return [self.settings.path_factory.data_item(fname)] + + def collect_dir(self, dname): + targets = [] + for root, dirs, files in os.walk(dname): + for fname in files: + targets += self.collect_file(os.path.join(root, fname)) + return targets + + def collect_targets(self, inputs): + targets = [] + for i in inputs: + if not os.path.isdir(i): + targets += self.collect_file(i) + else: + targets += self.collect_dir(i) + return targets + + def add_files(self, targets): + for data_item in targets: + data_item.move_data_to_cache() + + def create_state_files(self, targets): + """ + Create state files for all targets. + """ + for data_item in targets: + Logger.debug('Creating state file for {}'.format(data_item.data.relative)) + + command = CommandFile(None, [data_item.data.dvc], [], True, None) + state_file = StateFile(data_item, + self.settings, + command.dict, + []) + state_file.save() + Logger.debug('State file "{}" was created'.format(data_item.state.relative)) + + def run(self): + targets = self.collect_targets(self.parsed_args.input) + self.add_files(targets) + self.create_state_files(targets) + msg = 'DVC add: {}'.format(str(self.parsed_args.input)) + self.commit_if_needed(msg) diff --git a/dvc/command/checkout.py b/dvc/command/checkout.py index efa58fa9c..fac80661e 100644 --- a/dvc/command/checkout.py +++ b/dvc/command/checkout.py @@ -22,9 +22,10 @@ class CmdCheckout(CmdBase): return True - def checkout(self, items): + @staticmethod + def checkout(items): for item in items: - if self.cache_ok(item): + if CmdCheckout.cache_ok(item): continue if os.path.isfile(item.data.relative): diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index 0098b3826..434c90675 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -3,24 +3,23 @@ from dvc.logger import Logger import dvc.data_cloud as cloud - class CmdDataBase(CmdBase): def __init__(self, settings, cmd): super(CmdDataBase, self).__init__(settings) self.cmd = cmd def run(self): - self.cmd(self.parsed_args.targets, self.parsed_args.jobs) + getattr(self.cloud, self.cmd)(self.parsed_args.targets, self.parsed_args.jobs) class CmdDataPull(CmdDataBase): def __init__(self, settings): - super(CmdDataPull, self).__init__(settings, self.cloud.pull) + super(CmdDataPull, self).__init__(settings, 'pull') class CmdDataPush(CmdDataBase): def __init__(self, settings): - super(CmdDataPush, self).__init__(settings, self.cloud.push) + super(CmdDataPush, self).__init__(settings, 'push') class CmdDataStatus(CmdBase): diff --git a/dvc/command/import_file.py b/dvc/command/import_file.py deleted file mode 100644 index f26e2cfc7..000000000 --- a/dvc/command/import_file.py +++ /dev/null @@ -1,145 +0,0 @@ -import os - -try: - from urlparse import urlparse -except ImportError: - # Python 3 - # pylint: disable=no-name-in-module, import-error - from urllib.parse import urlparse - -from dvc.command.common.base import CmdBase -from dvc.logger import Logger -from dvc.exceptions import DvcException -from dvc.state_file import StateFile - - -class ImportFileError(DvcException): - def __init__(self, msg): - DvcException.__init__(self, 'Import file: {}'.format(msg)) - - -class CmdImportFile(CmdBase): - def __init__(self, settings): - super(CmdImportFile, self).__init__(settings) - - def run(self): - targets = [] - output = self.parsed_args.output - self.verify_output(output, self.parsed_args.input) - - for input in self.parsed_args.input: - if not os.path.isdir(input): - targets.append((input, output)) - else: - input_dir = os.path.basename(input) - for root, dirs, files in os.walk(input): - for file in files: - filename = os.path.join(root, file) - - rel = os.path.relpath(filename, input) - out = os.path.join(output, input_dir, rel) - - out_dir = os.path.dirname(out) - if not os.path.exists(out_dir): - os.mkdir(out_dir) - - targets.append((filename, out)) - pass - self.import_files(targets, self.parsed_args.lock, self.parsed_args.jobs) - message = 'DVC import files: {} -> {}'.format(str(self.parsed_args.input), output) - self.commit_if_needed(message) - - @staticmethod - def verify_output(output, input): - if CmdImportFile.is_dir_path(output) and not os.path.isdir(output): - raise ImportFileError(u'output directory {} does not exist'.format(output)) - - if len(input) > 1 and not os.path.isdir(output): - msg = u'output {} has to be directory for multiple file import' - raise ImportFileError(msg.format(output)) - pass - - @staticmethod - 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, 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) - - 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): - return self.import_files([(input, output)], lock) - - def collect_targets(self, targets): - """ - Collect input, output and data_item's into tuples - to be able to download them simulteniously later. - """ - data_targets = [] - - for target in targets: - input = target[0] - output = target[1] - if not CmdImportFile.is_url(input): - if not os.path.exists(input): - raise ImportFileError('Input file "{}" does not exist'.format(input)) - if not os.path.isfile(input): - raise ImportFileError('Input file "{}" has to be a regular file'.format(input)) - - if os.path.isdir(output): - output = os.path.join(output, os.path.basename(input)) - - data_item = self.settings.path_factory.data_item(output) - - if os.path.exists(data_item.data.relative): - raise ImportFileError('Output file "{}" already exists'.format(data_item.data.relative)) - if not os.path.isdir(os.path.dirname(data_item.data.relative)): - raise ImportFileError('Output file directory "{}" does not exists'.format( - os.path.dirname(data_item.data.relative))) - - if not os.path.exists(data_item.cache_dir_abs): - os.makedirs(data_item.cache_dir_abs) - - data_targets.append((input, data_item)) - - return data_targets - - def import_files(self, targets, lock=False, jobs=1): - data_targets = self.collect_targets(targets) - imported_targets = self.cloud.import_data(data_targets, jobs) - self.create_state_files(imported_targets, lock) - - def create_state_files(self, targets, lock): - """ - Create state files for all targets. - """ - for t in targets: - orig_target, processed_data_item = t - input, data_item = orig_target - output = data_item.data.relative - - if processed_data_item == None: - Logger.debug('Skipping creating state file for failed import {}'.format(data_item.state.relative)) - continue - - Logger.debug('Creating state file for {}'.format(data_item.data.relative)) - - state_file = StateFile(StateFile.COMMAND_IMPORT_FILE, - data_item, - self.settings, - argv=[input, output], - input_files=[], - output_files=[output], - lock=lock) - state_file.save() - Logger.debug('State file "{}" was created'.format(data_item.state.relative)) - - @staticmethod - def is_url(url): - return len(urlparse(url).scheme) != 0 diff --git a/dvc/command/merge.py b/dvc/command/merge.py index 9fa2c2cca..e01f6ca36 100644 --- a/dvc/command/merge.py +++ b/dvc/command/merge.py @@ -6,6 +6,8 @@ from dvc.logger import Logger from dvc.state_file import StateFile from dvc.path.data_item import DataItem from dvc.system import System +from dvc.command.checkout import CmdCheckout +from dvc.command.run import CommandFile class CmdMerge(CmdBase): @@ -21,8 +23,8 @@ class CmdMerge(CmdBase): dlist = [] flist = self.git.get_last_merge_changed_files() for fname in flist: - if fname.startswith(ConfigI.STATE_DIR) and fname.endswith(DataItem.STATE_FILE_SUFFIX): - data = os.path.relpath(fname, ConfigI.STATE_DIR)[:-len(DataItem.STATE_FILE_SUFFIX)] + if fname.startswith(self.settings.config.state_dir) and fname.endswith(DataItem.CACHE_STATE_FILE_SUFFIX): + data = os.path.relpath(fname, self.settings.config.state_dir)[:-len(DataItem.CACHE_STATE_FILE_SUFFIX)] dlist.append(data) return dlist @@ -32,32 +34,23 @@ class CmdMerge(CmdBase): items = self.settings.path_factory.to_data_items(flist)[0] for item in items: - try: - state = StateFile.load(item, self.git) - except Exception as ex: - Logger.error('Failed to load state file for {}'.format(item.data.relative), exc_info=True) - return None + state = StateFile.load(item, self.settings) + if isinstance(state.command, str): + command = CommandFile.load(state.command) + else: + command = CommandFile.loadd(state.command) - if not state.command == StateFile.COMMAND_IMPORT_FILE: - continue - - targets.append(item) + if not command.cmd and command.locked: + targets.append(item) return targets def checkout_targets(self, targets): data = [] for item in targets: - self.git.checkout_file_before_last_merge(item.state.relative) + self.git.checkout_file_before_last_merge(item.cache_state.relative) - # Reload data item, so we can get restored cache - new_item = self.settings.path_factory.data_item(item.data.relative) - if os.path.isfile(new_item.data.relative): - os.remove(new_item.data.relative) - - System.hardlink(new_item.cache.relative, new_item.data.relative) - - data.append(item.data.relative) + CmdCheckout.checkout([item]) msg = 'DVC merge files: {}'.format(' '.join(data)) self.commit_if_needed(msg) diff --git a/dvc/command/repro.py b/dvc/command/repro.py index 7c85b32a4..6997d02df 100644 --- a/dvc/command/repro.py +++ b/dvc/command/repro.py @@ -1,13 +1,13 @@ import os import copy -from dvc.command.import_file import CmdImportFile -from dvc.command.run import CmdRun +from dvc.command.run import CmdRun, CommandFile from dvc.logger import Logger from dvc.exceptions import DvcException from dvc.path.data_item import DataDirError from dvc.state_file import StateFile from dvc.system import System +from dvc.data_cloud import file_md5 class ReproError(DvcException): @@ -19,25 +19,6 @@ class CmdRepro(CmdRun): def __init__(self, settings): super(CmdRepro, self).__init__(settings) - self._code = [] - pass - - @property - def code_dependencies(self): - return self._code - - @property - def lock(self): - return True - - @property - def declaration_input_data_items(self): - return [] - - @property - def declaration_output_data_items(self): - return [] - def run(self): recursive = not self.parsed_args.single_item targets = [] @@ -75,17 +56,8 @@ class CmdRepro(CmdRun): for data_item in data_item_list: try: - target_commit = self.git.get_target_commit(data_item.state.relative) - if target_commit is None: - msg = 'Data item "{}" cannot be reproduced: cannot obtain commit hashsum' - Logger.warn(msg.format(data_item.data.relative)) - continue - - globally_changed_files = self.git.get_changed_files(target_commit) - Logger.debug('Repro globally changed files: {}'.format(', '.join(globally_changed_files))) - changed_files = set() - change = ReproChange(data_item, self, globally_changed_files, recursive, force) - if change.reproduce(changed_files): + change = ReproChange(data_item, self, recursive, force) + if change.reproduce(): changed = True Logger.info(u'Data item "{}" was reproduced.'.format( data_item.data.relative @@ -109,30 +81,26 @@ class CmdRepro(CmdRun): class ReproChange(object): - def __init__(self, data_item, cmd_obj, globally_changed_files, recursive, force): + def __init__(self, data_item, cmd_obj, 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.settings = cmd_obj.settings self._recursive = recursive self._force = force - try: - self._state = StateFile.load(data_item, self.git) - except Exception as ex: - raise ReproError('Error: state file "{}" cannot be loaded: {}'. - format(data_item.state.relative, ex)) + self.state = StateFile.load(data_item, self.settings) - if not self.state.argv: - raise ReproError('Error: parameter {} is not defined in state file "{}"'. - format(StateFile.PARAM_ARGV, data_item.state.relative)) - if len(self.state.argv) < 1: - raise ReproError('Error: reproducible cmd in state file "{}" is too short'. - format(self.state.file)) + if isinstance(self.state.command, str): + self.command = CommandFile.load(self.state.command) + else: + self.command = CommandFile.loadd(self.state.command) + + if not self.command.cmd and not self.command.locked: + msg = 'Error: data item "{}" is not locked, but has no command for reproduction' + raise ReproError(msg.format(data_item.data.dvc)) self._settings = copy.copy(self._cmd_obj.settings) - self._settings.set_args(self.state.argv) - pass def is_cache_exists(self): path = System.realpath(self._data_item.data.relative) @@ -142,14 +110,8 @@ class ReproChange(object): def cmd_obj(self): return self._cmd_obj - @property - def state(self): - return self._state - - def reproduce_data_item(self, changed_files): - Logger.debug('Reproducing data item {}.'.format(self._data_item.data.dvc)) - - for output_dvc in self._state.output_files: + def remove_output_files(self): + for output_dvc in self.command.out: Logger.debug('Removing output file {} before reproduction.'.format(output_dvc)) try: @@ -159,158 +121,69 @@ class ReproChange(object): msg = 'Data item {} cannot be removed before reproduction: {}' Logger.error(msg.format(output_dvc, ex)) - changed_files.add(output_dvc) + def reproduce_run(self): + Logger.info('Reproducing run command for data item {}. Args: {}'.format( + self._data_item.data.relative, self.command.cmd)) - if self.state.is_import_file: - Logger.debug('Reproducing data item {}. Re-import cmd: {}'.format( - self._data_item.data.relative, ' '.join(self.state.argv))) + CmdRun.run_command(self.settings, self.command) + self._cmd_obj.commit_if_needed('DVC repro: {}'.format(self.command.cmd)) - if len(self.state.argv) != 2: - msg = 'Data item "{}" cannot be re-imported because of arguments number {} is incorrect. Argv: {}' - raise ReproError(msg.format(self._data_item.data.relative, len(self.state.argv), self.state.argv)) + def reproduce_data_item(self): + Logger.debug('Reproducing data item {}.'.format(self._data_item.data.dvc)) + self.remove_output_files() + self.reproduce_run() - input = self.state.argv[0] - output = self.state.argv[1] - - cmd = CmdImportFile(self._settings) - cmd.set_git_action(True) - 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, - is_repro=True) != 0: - raise ReproError('Import command reproduction failed') + def is_repro_required(self): + deps_changed = self.reproduce_deps(self._data_item, self._recursive) + if deps_changed or self._force or not self.is_cache_exists(): return True - elif self.state.is_run: - cmd = CmdRun(self._settings) - cmd.set_git_action(True) - cmd.set_locker(False) + return False - 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) - if cmd.run_and_commit_if_needed(self.state.argv, - data_items_from_args, - not_data_items_from_args, - self.state.stdout, - self.state.stderr, - self.state.shell, - [], - [], - [], - False, - check_if_ready=False, - is_repro=True) != 0: - raise ReproError('Run command reproduction failed') - return True - else: - # Ignore EMPTY_FILE command - pass - pass - - def reproduce(self, changed_files): + def reproduce(self): Logger.debug('Reproduce data item {}. recursive={}, force={}'.format( self._data_item.data.relative, self._recursive, self._force)) - if self.state.locked: + if self.command.locked: Logger.debug('Data item {} is not reproducible'.format(self._data_item.data.relative)) return False - if self.is_repro_required(changed_files, self._data_item): - if self._data_item.data.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) - changed_files.add(self._data_item.data.dvc) - return True - else: - msg = 'Data item {} is not going to be reproduced because it is already reproduced' - Logger.debug(msg.format(self._data_item.data.relative)) - else: + if not self.is_repro_required(): Logger.debug('Data item {} is up to date'.format(self._data_item.data.relative)) return False - pass - def were_dependencies_changed(self, changed_files, data_item_dvc): - result = False + Logger.debug('Data item {} is going to be reproduced'.format(self._data_item.data.relative)) + self.reproduce_data_item() + return True - 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.state.dvc in self._globally_changed_files: - dep_state = StateFile.load(data_item, self._settings) - if dep_state.command != StateFile.COMMAND_IMPORT_FILE: - msg = u'Repro data item {}: dependency {} was not changed but the state file was changed' - Logger.debug(msg.format(data_item_dvc, data_item.data.dvc)) - result = True - else: - Logger.debug(u'Repro data item {}: skip dependency state file change {} due to data source command') - 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): - state_file = StateFile.load(data_item, self._settings) - if state_file.locked: - Logger.debug(u'Repro is not required for locked data item {}'.format(data_item.data.relative)) + def reproduce_dep(self, path, md5, recursive): + if not self._settings.path_factory.is_data_item(path): + if md5 != file_md5(os.path.join(self._settings.git.git_dir_abs, path))[0]: + self.log_repro_reason('source {} was changed'.format(path)) + return True return False - is_dependency_check_required = self._recursive + item = self._settings.path_factory.existing_data_item(path) + if recursive: + ReproChange(item, self._cmd_obj, self._recursive, self._force).reproduce() - 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.data.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 not self.is_cache_exists(): - self.log_repro_reason(u'cache file is missing.') - return True - - if self.were_sources_changed(self._globally_changed_files): - self.log_repro_reason(u'sources were changed') + if md5 != os.path.basename(item.cache.relative): + self.log_repro_reason('data item {} was changed'.format(path)) return True return False + def reproduce_deps(self, data_item_dvc, recursive): + result = False + + for dep in self.state.deps: + path = dep[StateFile.PARAM_PATH] + md5 = dep[StateFile.PARAM_MD5] + + if self.reproduce_dep(path, md5, recursive): + result = True + + return result + 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._settings.path_factory, - globally_changed_files - ) - return were_sources_changed - - @property - def dependencies(self): - dependency_data_items = [] - for input_file in self.state.input_files: - try: - data_item = self._settings.path_factory.data_item(input_file) - except DataDirError: - raise ReproError(u'The dependency file "{}" is not a data item'.format(input_file)) - except Exception as ex: - raise ReproError(u'Unable to reproduced the dependency file "{}": {}'.format( - input_file, ex)) - - dependency_data_items.append(data_item) - - return dependency_data_items diff --git a/dvc/command/run.py b/dvc/command/run.py index 679e2f5ac..7e9a750d2 100644 --- a/dvc/command/run.py +++ b/dvc/command/run.py @@ -1,4 +1,6 @@ import os +import sys +import yaml from dvc.command.common.base import CmdBase from dvc.exceptions import DvcException @@ -7,6 +9,8 @@ from dvc.logger import Logger from dvc.repository_change import RepositoryChange from dvc.state_file import StateFile from dvc.utils import cached_property +from dvc.executor import Executor +from dvc.state_file import StateFileBase class RunError(DvcException): @@ -14,173 +18,86 @@ class RunError(DvcException): DvcException.__init__(self, 'Run error: {}'.format(msg)) +class CommandFile(StateFileBase): + MAGIC = 'DVC-Command-State' + VERSION = '0.1' + + PARAM_CMD = 'cmd' + PARAM_OUT = 'out' + PARAM_DEPS = 'deps' + PARAM_LOCK = 'locked' + + def __init__(self, cmd, out, deps, locked, fname): + self.cmd = cmd + self.out = out + self.deps = deps + self.locked = locked + self.fname = fname + + @property + def dict(self): + data = { + self.PARAM_CMD: self.cmd, + self.PARAM_OUT: self.out, + self.PARAM_DEPS: self.deps, + self.PARAM_LOCK: self.locked + } + + return data + + def dumps(self): + return yaml.dump(self.dict) + + def dump(self, fname): + with open(fname, 'w+') as fd: + fd.write(self.dumps()) + + @staticmethod + def loadd(data, fname=None): + return CommandFile(data.get(CommandFile.PARAM_CMD, None), + data.get(CommandFile.PARAM_OUT, None), + data.get(CommandFile.PARAM_DEPS, None), + data.get(CommandFile.PARAM_LOCK, None), + fname) + + @staticmethod + def load(fname): + return CommandFile._load(fname, CommandFile, fname) + + class CmdRun(CmdBase): def __init__(self, settings): super(CmdRun, self).__init__(settings) - @property - def lock(self): - return self.parsed_args.lock - - @property - def code_dependencies(self): - return self.parsed_args.code or [] - - @cached_property - def declaration_input_data_items(self): - return self._data_items_from_params(self.parsed_args.input, 'Input') - - @cached_property - def declaration_output_data_items(self): - return self._data_items_from_params(self.parsed_args.output, 'Output') - def run(self): - cmd = [self.parsed_args.command] + self.parsed_args.args - data_items_from_args, not_data_items_from_args = self.argv_files_by_type(cmd) - return self.run_and_commit_if_needed(cmd, - data_items_from_args, - not_data_items_from_args, - self.parsed_args.stdout, - self.parsed_args.stderr, - self.parsed_args.shell) + cmd = ' '.join(self.parsed_args.command) + try: + command = CommandFile.load(cmd) + except Exception as exc: + Logger.debug("Failed to load {}: {}".format(cmd, str(exc))) + command = CommandFile(cmd, self.parsed_args.out, self.parsed_args.deps, self.parsed_args.lock, None) - def run_and_commit_if_needed(self, command_args, data_items_from_args, not_data_items_from_args, - stdout, stderr, shell, - output_data_items=None, - input_data_items=None, - code_dependencies=None, - lock=None, - 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.run_command(self.settings, command) + return self.commit_if_needed('DVC run: {}'.format(command.cmd)) - self.run_command(command_args, - data_items_from_args, - not_data_items_from_args, - stdout, - stderr, - shell, - output_data_items, - input_data_items, - code_dependencies, - lock) - - 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, - output_data_items=None, - input_data_items=None, - code_dependencies=None, - lock=None): - - # Repro sets these from state file - if output_data_items == None: - output_data_items = self.declaration_output_data_items - - if input_data_items == None: - input_data_items = self.declaration_input_data_items - - if code_dependencies == None: - code_dependencies = self.code_dependencies - - if lock == None: - lock = self.lock - - Logger.debug(u'Run command with args: {}. Data items from args: {}. stdout={}, stderr={}, shell={}'.format( - ' '.join(cmd_args), - ', '.join([x.data.dvc for x in data_items_from_args]), - stdout, - stderr, - shell)) - - repo_change = RepositoryChange(cmd_args, self.settings, stdout, stderr, shell=shell) - - if not self.no_git_actions and not self._validate_file_states(repo_change): - self.remove_new_files(repo_change) - raise RunError('Errors occurred.') - - output_set = set(output_data_items + repo_change.changed_data_items) - output_files_dvc = [x.data.dvc for x in output_set] - - input_set = set(data_items_from_args + input_data_items) - output_set - input_files_dvc = [x.data.dvc for x in input_set] - - code_dependencies_dvc = self.git.abs_paths_to_dvc(set(code_dependencies + not_data_items_from_args)) + @staticmethod + def run_command(settings, command): + Executor.exec_cmd_only_success(command.cmd, shell=True) result = [] - for data_item in repo_change.changed_data_items: + items = settings.path_factory.to_data_items(command.out)[0] + for data_item in items: 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() Logger.debug('Create state file "{}"'.format(data_item.state.relative)) - state_file = StateFile(StateFile.COMMAND_RUN, - data_item, - self.settings, - input_files_dvc, - output_files_dvc, - code_dependencies_dvc, - argv=cmd_args, - lock=lock, - stdout=self._stdout_to_dvc(stdout), - stderr=self._stdout_to_dvc(stderr), - shell=shell) + state_file = StateFile(data_item, + settings, + command.fname if command.fname else command.dict, + StateFile.parse_deps_state(settings, command.deps)) state_file.save() result.append(state_file) return result - - def _stdout_to_dvc(self, stdout): - if stdout in {None, '-'}: - return stdout - return self.settings.path_factory.data_item(stdout).data.dvc - - @staticmethod - def remove_new_files(repo_change): - for data_item in repo_change.new_data_items: - Logger.error('Removing created file: {}'.format(data_item.data.relative)) - os.remove(data_item.data.relative) - pass - - @staticmethod - def _validate_file_states(repo_change): - error = False - for data_item in repo_change.removed_data_items: - Logger.error('Error: file "{}" was removed'.format(data_item.data.relative)) - error = True - - for file in GitWrapper.abs_paths_to_relative(repo_change.externally_created_files): - Logger.error('Error: file "{}" was created outside of the data directory'.format(file)) - error = True - - return not error - - def argv_files_by_type(self, argv): - data_items = [] - not_data_items = [] - - for arg in argv: - if self.settings.path_factory.is_data_item(arg): - data_items.append(self.settings.path_factory.data_item(arg)) - elif os.path.isfile(arg): - not_data_items.append(arg) - else: - msg = 'File {} from argv is outside of git directory and cannot be traced' - Logger.warn(msg.format(arg)) - - return data_items, not_data_items - - def _data_items_from_params(self, files, param_text): - if not files: - return [] - - data_items, external = self.settings.path_factory.to_data_items(files) - if external: - raise RunError('{} should point to data items from data dir: {}'.format( - param_text, ', '.join(external)) - ) - return data_items diff --git a/dvc/config.py b/dvc/config.py index af9b97ca7..ff5de8435 100644 --- a/dvc/config.py +++ b/dvc/config.py @@ -35,15 +35,15 @@ class ConfigI(object): self._cloud = cloud self._conf_parser = conf_parser - # @classmethod - # def CACHE_DIR(self): - # """ Directory with cached data files """ - # return os.path.join(self.CONFIG_DIR, self.CACHE_DIR_NAME) - # - # @classmethod - # def STATE_DIR(self): - # """ Directory with state files """ - # return os.path.join(self.CONFIG_DIR, self.STATE_DIR_NAME) + @property + def cache_dir(self): + """ Directory with cached data files """ + return self.CACHE_DIR + + @property + def state_dir(self): + """ Directory with state files """ + return self.STATE_DIR @property def cloud(self): diff --git a/dvc/path/data_item.py b/dvc/path/data_item.py index e1e9df32d..8b0f0e41c 100644 --- a/dvc/path/data_item.py +++ b/dvc/path/data_item.py @@ -7,6 +7,7 @@ from dvc.exceptions import DvcException from dvc.system import System from dvc.utils import cached_property from dvc.data_cloud import file_md5 +from dvc.state_file import CacheStateFile, LocalStateFile class DataItemError(DvcException): @@ -33,6 +34,8 @@ class NotInGitDirError(DataDirError): class DataItem(object): STATE_FILE_SUFFIX = '.state' + LOCAL_STATE_FILE_SUFFIX = '.local_state' + CACHE_STATE_FILE_SUFFIX = '.cache_state' CACHE_FILE_SEP = '_' def __init__(self, data_file, git, config, cache_file=None): @@ -70,24 +73,41 @@ class DataItem(object): return self._data @cached_property - def state(self): - state_dir = os.path.join(self._git.git_dir_abs, ConfigI.STATE_DIR) - state_file = os.path.join(state_dir, self.data.dvc + self.STATE_FILE_SUFFIX) + def state_dir(self): + return os.path.join(self._git.git_dir_abs, self._config.state_dir) + + def _state(self, suffix): + state_file = os.path.join(self.state_dir, self.data.dvc + suffix) return Path(state_file, self._git) + @cached_property + def state(self): + return self._state(self.STATE_FILE_SUFFIX) + @cached_property def cache_dir_abs(self): return os.path.join(self._git.git_dir_abs, ConfigI.CACHE_DIR) @cached_property + def local_state(self): + return self._state(self.LOCAL_STATE_FILE_SUFFIX) + + @cached_property + def cache_state(self): + return self._state(self.CACHE_STATE_FILE_SUFFIX) + + @cached_property + def cache_dir(self): + return os.path.join(self._git.git_dir_abs, self._config.cache_dir) + + @property def cache(self): cache_dir = self.cache_dir_abs if self._cache_file: file_name = os.path.relpath(os.path.realpath(self._cache_file), cache_dir) else: - from dvc.state_file import StateFile - file_name = StateFile.load(self, self._git).md5 + file_name = CacheStateFile.load(self).md5 cache_file = os.path.join(cache_dir, file_name) return Path(cache_file, self._git) @@ -103,3 +123,8 @@ class DataItem(object): if not os.path.isfile(self.cache.relative): System.hardlink(self.data.relative, self.cache.relative) os.chmod(self.data.relative, stat.S_IREAD) + + cache_state = CacheStateFile(self).save() + + local_state = LocalStateFile(self).save() + self._git.modify_gitignore([self.local_state.relative]) diff --git a/dvc/path/factory.py b/dvc/path/factory.py index 4f7d8a591..3502ecf41 100644 --- a/dvc/path/factory.py +++ b/dvc/path/factory.py @@ -61,7 +61,6 @@ class PathFactory(object): path = os.path.join(root, fname) if not fname.endswith(DataItem.STATE_FILE_SUFFIX): - Logger.warn('Found file \'{}\' without state suffix inside state dir'.format(path)) continue states.append(path) diff --git a/dvc/repository_change.py b/dvc/repository_change.py index f932b6554..087a118d8 100644 --- a/dvc/repository_change.py +++ b/dvc/repository_change.py @@ -17,17 +17,17 @@ class RepositoryChangeError(DvcException): class RepositoryChange(object): """Pre-condition: git repository has no changes""" - def __init__(self, cmd_args, settings, stdout, stderr, shell=False): + def __init__(self, command, settings, stdout, stderr, shell=False): self._settings = settings stemps_before = self.data_file_timesteps() Logger.debug(u'[Repository change] Exec command: {}. stdout={}, stderr={}, shell={}'.format( - u' '.join(cmd_args), + command, stdout, stderr, shell)) - Executor.exec_cmd_only_success(cmd_args, stdout, stderr, shell=shell) + Executor.exec_cmd_only_success(command.split(' '), stdout, stderr, shell=shell) stemps_after = self.data_file_timesteps() diff --git a/dvc/state_file.py b/dvc/state_file.py index e0f3aa887..1bb208b50 100644 --- a/dvc/state_file.py +++ b/dvc/state_file.py @@ -1,32 +1,39 @@ import os import ntpath import sys -import json +import yaml import re from dvc import utils from dvc.exceptions import DvcException -from dvc.path.data_item import DataDirError from dvc.system import System - +from dvc.data_cloud import file_md5 class StateFileError(DvcException): def __init__(self, msg): DvcException.__init__(self, 'State file error: {}'.format(msg)) -class StateFile(object): +class StateFileBase(object): + @staticmethod + def _save(fname, data): + with open(fname, 'w') as fd: + yaml.dump(data, fd, default_flow_style=False) + + @staticmethod + def _load(fname, state_class, *args): + with open(fname, 'r') as fd: + data = yaml.load(fd) + return state_class.loadd(data, *args) + + +class StateFile(StateFileBase): MAGIC = 'DVC-State' VERSION = '0.1' DVC_PYTHON_FILE_NAME = 'dvc.py' DVC_COMMAND = 'dvc' - COMMAND_RUN = 'run' - COMMAND_IMPORT_FILE = 'import-file' - ACCEPTED_COMMANDS = {COMMAND_IMPORT_FILE, COMMAND_RUN} - - PARAM_COMMAND = 'Command' PARAM_TYPE = 'Type' PARAM_VERSION = 'Version' PARAM_ARGV = 'Argv' @@ -40,52 +47,41 @@ class StateFile(object): PARAM_SHELL = "Shell" PARAM_TARGET_METRICS = 'TargetMetrics' TARGET_METRICS_SINGLE_METRIC = 'SingleMetric' - PARAM_MD5 = 'Md5' + + PARAM_COMMAND = 'command' + PARAM_DEPS = 'deps' + PARAM_PATH = 'path' + PARAM_MD5 = 'md5' def __init__(self, - command, data_item, settings, - input_files, - output_files, - code_dependencies=[], - lock=False, - argv=sys.argv, - stdout=None, - stderr=None, - cwd=None, - shell=False, - target_metrics={}, - md5=None): + command, + deps, + target_metrics={}): + super(StateFile, self).__init__() + self.data_item = data_item self.settings = settings - self.input_files = input_files - self.output_files = output_files - self.locked = lock - self.code_dependencies = code_dependencies - self.shell = shell - - self.md5 = md5 - if not md5 and data_item: - self.md5 = os.path.basename(data_item.cache.relative) - - if command not in self.ACCEPTED_COMMANDS: - raise StateFileError('Args error: unknown command %s' % command) self.command = command - - self._argv = argv - - self.stdout = stdout - self.stderr = stderr - - if cwd: - self.cwd = cwd - else: - self.cwd = self.get_dvc_path() + self.deps = deps self.target_metrics = target_metrics + @staticmethod + def parse_deps_state(settings, deps): + state = [] + for dep in deps: + if settings.path_factory.is_data_item(dep): + item = settings.path_factory.data_item(dep) + state.append({StateFile.PARAM_PATH: item.data.dvc, + StateFile.PARAM_MD5: CacheStateFile.load(item).md5}) + else: + state.append({StateFile.PARAM_PATH: dep, + StateFile.PARAM_MD5: file_md5(os.path.join(settings.git.git_dir_abs, dep))[0]}) + return state + @property def file(self): return self.data_item.state.relative @@ -119,125 +115,113 @@ class StateFile(object): except StateFileError: return None - @property - def is_import_file(self): - return self.command == self.COMMAND_IMPORT_FILE - - @property - def is_run(self): - return self.command == self.COMMAND_RUN - - @property - def argv(self): - return self._argv - @staticmethod - def _replace_paths(l, old, new): - if os.path != ntpath: - return l - - ret = [] - for x in l: - if x == None: - ret.append(None) - continue - - ret.append(x.replace(old, new)) - - return ret - - @staticmethod - def decode_paths(l): - return StateFile._replace_paths(l, '/', '\\') - - @staticmethod - def decode_path(p): - return StateFile.decode_paths([p])[0] - - @staticmethod - def encode_paths(l): - return StateFile._replace_paths(l, '\\', '/') - - @staticmethod - def encode_path(p): - return StateFile.encode_paths([p])[0] - - @staticmethod - def load_json(json, settings): - return StateFile(StateFile.decode_path(json.get(StateFile.PARAM_COMMAND)), - None, + def loadd(data, settings): + return StateFile(None, settings, - StateFile.decode_paths(json.get(StateFile.PARAM_INPUT_FILES, [])), - StateFile.decode_paths(json.get(StateFile.PARAM_OUTPUT_FILES, [])), - StateFile.decode_paths(json.get(StateFile.PARAM_CODE_DEPENDENCIES, [])), - json.get(StateFile.PARAM_LOCKED, False), - StateFile.decode_paths(json.get(StateFile.PARAM_ARGV)), - StateFile.decode_path(json.get(StateFile.PARAM_STDOUT)), - StateFile.decode_path(json.get(StateFile.PARAM_STDERR)), - StateFile.decode_path(json.get(StateFile.PARAM_CWD)), - json.get(StateFile.PARAM_SHELL, False), - json.get(StateFile.PARAM_TARGET_METRICS, {}), - json.get(StateFile.PARAM_MD5, '')) + data.get(StateFile.PARAM_COMMAND, None), + data.get(StateFile.PARAM_DEPS, []), + data.get(StateFile.PARAM_TARGET_METRICS, {})) @staticmethod def load(data_item, settings): - with open(data_item.state.relative, 'r') as fd: - data = json.load(fd) - return StateFile.load_json(data, settings) + return StateFile._load(data_item.state.relative, StateFile, settings) @staticmethod def loads(content, settings): - data = json.loads(content) - return StateFile.load_json(data, settings) + data = yaml.loads(content) + return StateFile.loadd(data, settings) def save(self, is_update_target_metrics=True): - argv = self._argv_paths_normalization(self._argv) - if is_update_target_metrics: self.update_target_metrics() res = { - self.PARAM_COMMAND: self.encode_path(self.command), - self.PARAM_TYPE: self.MAGIC, - self.PARAM_VERSION: self.VERSION, - self.PARAM_ARGV: self.encode_paths(argv), - self.PARAM_CWD: self.encode_path(self.cwd), - self.PARAM_INPUT_FILES: self.encode_paths(self.input_files), - self.PARAM_OUTPUT_FILES: self.encode_paths(self.output_files), - self.PARAM_CODE_DEPENDENCIES: self.encode_paths(self.code_dependencies), - self.PARAM_STDOUT: self.encode_path(self.stdout), - self.PARAM_STDERR: self.encode_path(self.stderr), - self.PARAM_SHELL: self.shell, - self.PARAM_TARGET_METRICS: self.target_metrics, - self.PARAM_MD5: self.md5 + self.PARAM_COMMAND: self.command, + self.PARAM_DEPS: self.deps, + self.PARAM_TARGET_METRICS: self.target_metrics } - if self.locked: - res[self.PARAM_LOCKED] = True - file_dir = os.path.dirname(self.file) if file_dir != '' and not os.path.isdir(file_dir): os.makedirs(file_dir) - with open(self.file, 'w') as fd: - json.dump(res, fd, indent=2, sort_keys=True) - pass + self._save(self.file, res) - def _argv_paths_normalization(self, argv): - result = [] - for arg in argv: - try: - data_item = self.settings.path_factory.data_item(arg) - result.append(data_item.data.dvc) - except DataDirError: - result.append(arg) +class CacheStateFile(StateFileBase): + MAGIC = 'DVC-Cache-State' + VERSION = '0.1' - return result + PARAM_MD5 = StateFile.PARAM_MD5 - def get_dvc_path(self): - pwd = System.get_cwd() - if not pwd.startswith(self.settings.git.git_dir_abs): - raise StateFileError('the file cannot be created outside of a git repository') + def __init__(self, data_item, md5=None): + super(CacheStateFile, self).__init__() - return os.path.relpath(pwd, self.settings.git.git_dir_abs) + self.data_item = data_item + self.md5 = md5 + + if not md5: + self.md5 = file_md5(data_item.data.relative)[0] + + @staticmethod + def loadd(data): + return CacheStateFile(None, + data.get(CacheStateFile.PARAM_MD5, None)) + + @staticmethod + def load(data_item): + return CacheStateFile._load(data_item.cache_state.relative, CacheStateFile) + + def save(self): + res = { + self.PARAM_MD5 : self.md5, + } + + file_dir = os.path.dirname(self.data_item.cache_state.relative) + if not os.path.isdir(file_dir): + os.makedirs(file_dir) + + self._save(self.data_item.cache_state.relative, res) + + +class LocalStateFile(StateFileBase): + MAGIC = 'DVC-Local-State' + VERSION = '0.1' + + PARAM_DATA_TIMESTAMP = 'DataTimestamp' + PARAM_CACHE_TIMESTAMP = 'CacheTimestamp' + + def __init__(self, data_item, data_timestamp=None, cache_timestamp=None): + super(LocalStateFile, self).__init__() + + self.data_item = data_item + self.data_timestamp = data_timestamp + self.cache_timestamp = cache_timestamp + + if not data_timestamp: + self.data_timestamp = os.path.getmtime(self.data_item.data.relative) + if not cache_timestamp: + self.cache_timestamp = os.path.getmtime(self.data_item.cache.relative) + + @staticmethod + def loadd(data): + return LocalStateFile(None, + data.get(LocalStateFile.PARAM_DATA_TIMESTAMP, None), + data.get(LocalStateFile.PARAM_CACHE_TIMESTAMP, None)) + + @staticmethod + def load(data_item): + return LocalStateFile._load(data_item.local_state.relative, LocalStateFile) + + def save(self): + res = { + self.PARAM_DATA_TIMESTAMP : self.data_timestamp, + self.PARAM_CACHE_TIMESTAMP : self.cache_timestamp + } + + file_dir = os.path.dirname(self.data_item.local_state.relative) + if not os.path.isdir(file_dir): + os.makedirs(file_dir) + + self._save(self.data_item.local_state.relative, res) diff --git a/functest/common.sh b/functest/common.sh index daead7842..38565fd9c 100644 --- a/functest/common.sh +++ b/functest/common.sh @@ -115,7 +115,8 @@ function dvc_create_repo() { dvc init mkdir data - dvc import $DATA_CACHE/* data + cp $DATA_CACHE/* data + dvc add data } function dvc_clean_cloud_aws() { diff --git a/functest/run.sh b/functest/run.sh index 0f235d26c..756d9be39 100755 --- a/functest/run.sh +++ b/functest/run.sh @@ -5,7 +5,7 @@ set -e source common.sh TESTS+=" test-init.sh" -TESTS+=" test-import.sh" +TESTS+=" test-add.sh" TESTS+=" test-sync.sh" TESTS+=" test-repro-code.sh" TESTS+=" test-merge.sh" diff --git a/functest/test-add.sh b/functest/test-add.sh new file mode 100755 index 000000000..3ab215cc2 --- /dev/null +++ b/functest/test-add.sh @@ -0,0 +1,13 @@ +#!/bin/bash + +set -e + +source common.sh + +dvc_create_repo + +dvc_info "Add file" +cp $DATA_CACHE/foo data/local +dvc add data/local +dvc_check_files "data/local .dvc/state/data/local.state .dvc/state/data/local.cache_state" +dvc_pass diff --git a/functest/test-import.sh b/functest/test-import.sh deleted file mode 100755 index bc277499b..000000000 --- a/functest/test-import.sh +++ /dev/null @@ -1,24 +0,0 @@ -#!/bin/bash - -set -e - -source common.sh - -dvc_create_repo - -dvc_info "Import from local file" -dvc import $DATA_CACHE/foo data/local -dvc_check_files "data/local" - -dvc_info "Import from http url" -dvc import $DATA_S3/foo data/http -dvc_check_files "data/http" - -if [[ "$TRAVIS_PULL_REQUEST" == "false" && "$TRAVIS_SECURE_ENV_VARS" == "true" ]]; then - dvc_info "Import from s3 url" - dvc config aws.region $REGION - dvc import s3://dataversioncontrol/functests/dummy_data/foo data/s3 - dvc_check_files "data/s3" -fi - -dvc_pass diff --git a/functest/test-merge.sh b/functest/test-merge.sh index 0a9e1a949..e626d686f 100755 --- a/functest/test-merge.sh +++ b/functest/test-merge.sh @@ -16,14 +16,15 @@ cp code/code1.sh code/code4.sh git add code git commit -s -m"add code" -dvc import $DATA_CACHE/foo data/data +cp $DATA_CACHE/foo data/data +dvc add data/data ORIG_DATA=$(dvc_md5 data/data) -dvc run ./code/code1.sh data/data data/data1 -dvc run ./code/code2.sh data/data1 data/data2 -dvc run ./code/code3.sh data/data2 data/data3 -dvc run ./code/code4.sh data/data3 data/data4 +dvc run -d code/code1.sh -d data/data -o data/data1 ./code/code1.sh data/data data/data1 +dvc run -d code/code2.sh -d data/data1 -o data/data2 ./code/code2.sh data/data1 data/data2 +dvc run -d code/code3.sh -d data/data2 -o data/data3 ./code/code3.sh data/data2 data/data3 +dvc run -d code/code4.sh -d data/data3 -o data/data4 ./code/code4.sh data/data3 data/data4 dvc config Global.Target data/data4 git commit -am 'Set Target' @@ -31,7 +32,8 @@ git commit -am 'Set Target' # Create new branch to work on small data set git checkout -b bar dvc remove -c -l data/data -dvc import $DATA_CACHE/bar data/data +cp $DATA_CACHE/bar data/data +dvc add data/data dvc repro echo -e "\n\n\n\n" >> code/code1.sh @@ -45,11 +47,11 @@ git merge bar # way dvc knows to favor data files from current branch. dvc merge -# Verify that data and respective state file are restored properly +# Verify that data and respective cache state file are restored properly if [ "$(dvc_md5 data/data)" != "$ORIG_DATA" ]; then dvc_fail fi -grep $(echo "$ORIG_DATA" | cut -d " " -f1) .dvc/state/data/data.state || dvc_fail +grep $(echo "$ORIG_DATA" | cut -d " " -f1) .dvc/state/data/data.cache_state || dvc_fail dvc_pass diff --git a/functest/test-repro-code.sh b/functest/test-repro-code.sh index 618912972..19734d1ef 100755 --- a/functest/test-repro-code.sh +++ b/functest/test-repro-code.sh @@ -7,7 +7,7 @@ source common.sh dvc_create_repo dvc_info 'Copy foo into foo1' -dvc run bash code/code.sh data/foo data/foo1 +dvc run -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 @@ -22,7 +22,8 @@ fi dvc_info 'Modify foo' dvc remove data/foo -dvc import $DATA_CACHE/bar data/foo +cp $DATA_CACHE/bar data/foo +dvc add data/foo dvc_info 'Set default target' dvc config global.target data/foo1 diff --git a/functest/tutorial_end_to_end.sh b/functest/tutorial_end_to_end.sh index f3d732845..531be5f18 100755 --- a/functest/tutorial_end_to_end.sh +++ b/functest/tutorial_end_to_end.sh @@ -23,16 +23,17 @@ 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/ +wget -P data/ https://s3-us-west-2.amazonaws.com/dvc-share/so/10K/Posts.xml.tgz +dvc add data/Posts.xml.tgz +dvc run -d data/Posts.xml.tgz -o data/Posts.xml 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 -d code/xml_to_tsv.py -d data/Posts.xml -o data/Posts.tsv python code/xml_to_tsv.py data/Posts.xml data/Posts.tsv python +dvc run -d code/split_train_test.py -d data/Posts.tsv -o data/Posts-test.tsv python code/split_train_test.py data/Posts.tsv 0.33 20170426 data/Posts-train.tsv data/Posts-test.tsv +dvc run -d code/featurization.py -d data/Posts-train.tsv -d data/Posts-test.tsv -o data/matrix-train.p -o data/matrix-test.p 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 -d code/train_model.py -d data/matrix-train.p -o data/model.p 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/evaluation.txt +dvc run -d code/evaluate.py -d data/model.p -d data/matrix-test.p -o data/evaluation.txt python code/evaluate.py data/model.p data/matrix-test.p data/evaluation.txt cat data/evaluation.txt # AUC: 0.552980 @@ -43,8 +44,10 @@ exit 0 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 + +wget -P data/ https://s3-us-west-2.amazonaws.com/dvc-share/so/25K/Posts.xml.tgz +dvc add data/Posts.xml.tgz +dvc repro -v data/evaluation.txt cat data/evaluation.txt # AUC: 0.639861