diff --git a/dvc/cli.py b/dvc/cli.py index 83f2a0458..8f9aea9eb 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -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,7 @@ 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 diff --git a/dvc/command/add.py b/dvc/command/add.py index 54ef60d6f..069c45d6b 100644 --- a/dvc/command/add.py +++ b/dvc/command/add.py @@ -3,6 +3,8 @@ import os from dvc.command.base import CmdBase, DvcLock from dvc.logger import Logger from dvc.state_file import StateFile +from dvc.command.run import CommandFile + class CmdAdd(CmdBase): def __init__(self, settings): @@ -38,12 +40,11 @@ class CmdAdd(CmdBase): 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, - argv=[], - input_items=[], - output_files=[], - lock=True) + command.dict, + []) state_file.save() Logger.debug('State file "{}" was created'.format(data_item.state.relative)) diff --git a/dvc/command/repro.py b/dvc/command/repro.py index 1da83b876..32e4e5c65 100644 --- a/dvc/command/repro.py +++ b/dvc/command/repro.py @@ -2,7 +2,7 @@ import os import copy from dvc.command.base import DvcLock -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 @@ -20,25 +20,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 = [] @@ -110,21 +91,21 @@ class ReproChange(object): self._force = force try: - self._state = StateFile.load(data_item, self.settings) + self.state = StateFile.load(data_item, self.settings) except Exception as ex: raise ReproError('Error: state file "{}" cannot be loaded: {}'. format(data_item.state.relative, ex)) - if not self.state.argv and not self.state.locked: - raise ReproError('Error: parameter {} is not defined in state file "{}"'. - format(StateFile.PARAM_ARGV, data_item.state.relative)) - if len(self.state.argv) < 1 and not self.state.locked: - 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: + raise ReproError('Error: parameter {} is not defined in "{}"'. + format(CommandFile.PARAM_CMD, self.command.fname if self.command.fname else data_item.state.relative)) 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) @@ -134,12 +115,8 @@ class ReproChange(object): def cmd_obj(self): return self._cmd_obj - @property - def state(self): - return self._state - def remove_output_files(self): - for output_dvc in self._state.output_files: + for output_dvc in self.command.out: Logger.debug('Removing output file {} before reproduction.'.format(output_dvc)) try: @@ -150,27 +127,11 @@ class ReproChange(object): Logger.error(msg.format(output_dvc, ex)) def reproduce_run(self): - cmd = CmdRun(self._settings) - cmd.set_git_action(True) - cmd.set_locker(False) - Logger.info('Reproducing run command for data item {}. Args: {}'.format( - self._data_item.data.relative, ' '.join(self.state.argv))) + self._data_item.data.relative, self.command.cmd)) - 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') + CmdRun.run_command(self.settings, self.command) + self._cmd_obj.commit_if_needed('DVC repro: {}'.format(self.command.cmd)) def reproduce_data_item(self): Logger.debug('Reproducing data item {}.'.format(self._data_item.data.dvc)) @@ -181,7 +142,7 @@ class ReproChange(object): 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 @@ -201,7 +162,7 @@ class ReproChange(object): def reproduce_deps(self, data_item_dvc, recursive): result = False - for dep in self._state.deps: + for dep in self.state.deps: path = dep[StateFile.PARAM_PATH] md5 = dep[StateFile.PARAM_MD5] diff --git a/dvc/command/run.py b/dvc/command/run.py index 9cbeebb36..c10b9fd09 100644 --- a/dvc/command/run.py +++ b/dvc/command/run.py @@ -9,7 +9,7 @@ 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 class RunError(DvcException): def __init__(self, msg): @@ -20,131 +20,74 @@ class CommandFile(object): MAGIC = 'DVC-Command-State' VERSION = '0.1' - PARAM_COMMAND = 'Command' - PARAM_CWD = 'Cwd' - PARAM_INPUT_FILES = 'InputFiles' - PARAM_OUTPUT_FILES = 'OutputFiles' - PARAM_CODE_DEPENDENCIES = 'CodeDependencies' - PARAM_SHELL = "Shell" - PARAM_DEPS = 'Deps' - PARAM_PATH = 'Path' + PARAM_CMD = 'cmd' + PARAM_OUT = 'out' + PARAM_DEPS = 'deps' + PARAM_LOCK = 'locked' - def __init__(self, cmd, cwd, deps): + def __init__(self, cmd, out, deps, locked, fname): self.cmd = cmd - self.cwd = cwd + 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 loads(data): + def loadd(data, fname=None): return CommandFile(data.get(CommandFile.PARAM_CMD, None), - data.get(CommandFile.PARAM_CWD, None), - data.get(CommandFile.PARAM_DEPS, [])) + data.get(CommandFile.PARAM_OUT, None), + data.get(CommandFile.PARAM_DEPS, None), + data.get(CommandFile.PARAM_LOCK, None), + fname) @staticmethod def load(fname): with open(fname, 'r') as fd: data = yaml.load(fd) - return CommandFile.loads(data) + return CommandFile.loadd(data, 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) + with DvcLock(self.is_locker, self.git): + try: + command = CommandFile.load(self.parsed_args.command) + #FIXME OLOLO don't forget to overwrite with cmdline args + except Exception as exc: + Logger.debug("Failed to load {}: {}".format(self.parsed_args.command[0], str(exc))) + command = CommandFile(self.parsed_args.command, 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() @@ -152,67 +95,10 @@ class CmdRun(CmdBase): Logger.debug('Create state file "{}"'.format(data_item.state.relative)) state_file = StateFile(data_item, - self.settings, - input_set, - 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) + 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/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 1f2cdd891..51bd2f830 100644 --- a/dvc/state_file.py +++ b/dvc/state_file.py @@ -2,6 +2,7 @@ import os import ntpath import sys import json +import yaml import re from dvc import utils @@ -34,63 +35,38 @@ class StateFile(object): PARAM_SHELL = "Shell" PARAM_TARGET_METRICS = 'TargetMetrics' TARGET_METRICS_SINGLE_METRIC = 'SingleMetric' - PARAM_DEPS = 'Deps' - PARAM_PATH = 'Path' - PARAM_MD5 = 'Md5' + + PARAM_COMMAND = 'command' + PARAM_DEPS = 'deps' + PARAM_PATH = 'path' + PARAM_MD5 = 'md5' def __init__(self, data_item, settings, - input_items, - output_files, - code_dependencies=[], - lock=False, - argv=sys.argv, - stdout=None, - stderr=None, - cwd=None, - shell=False, - target_metrics={}, - deps=None): + command, + deps, + target_metrics={}): self.data_item = data_item self.settings = settings - self.input_files = [x.data.dvc for x in input_items] - self.output_files = output_files - self.locked = lock - self.code_dependencies = code_dependencies - self.shell = shell - - self._argv = argv - - self.stdout = stdout - self.stderr = stderr - - if cwd: - self.cwd = cwd - else: - self.cwd = self.get_dvc_path() + self.command = command + self.deps = deps self.target_metrics = target_metrics - if deps: - self.deps = deps - else: - self.deps = self.parse_deps(settings.git, input_items, code_dependencies) - @staticmethod - def parse_deps(git, input_items, code_files): - deps = [] - - for item in input_items: - deps.append({StateFile.PARAM_PATH : item.data.dvc, - StateFile.PARAM_MD5: CacheStateFile.load(item).md5}) - - for code in code_files: - deps.append({StateFile.PARAM_PATH: code, - StateFile.PARAM_MD5: file_md5(os.path.join(git.git_dir_abs, code))[0]}) - - return deps + 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): @@ -125,120 +101,43 @@ class StateFile(object): except StateFileError: return None - @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(None, settings, - settings.path_factory.to_data_items(StateFile.decode_paths(json.get(StateFile.PARAM_INPUT_FILES, [])))[0], - 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_DEPS, [])) + json.get(StateFile.PARAM_COMMAND, None), + json.get(StateFile.PARAM_DEPS, []), + json.get(StateFile.PARAM_TARGET_METRICS, {})) @staticmethod def load(data_item, settings): with open(data_item.state.relative, 'r') as fd: - data = json.load(fd) + data = yaml.load(fd) return StateFile.load_json(data, settings) @staticmethod def loads(content, settings): - data = json.loads(content) + data = yaml.loads(content) return StateFile.load_json(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_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_DEPS: self.deps + 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) + yaml.dump(res, fd, default_flow_style=False) pass - def _argv_paths_normalization(self, argv): - result = [] - - from dvc.path.data_item import DataDirError - 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) - - return result - - 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') - - return os.path.relpath(pwd, self.settings.git.git_dir_abs) - class CacheStateFile(object): MAGIC = 'DVC-Cache-State'