Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-06 07:18:28 +01:00
Code Issues Packages Projects Releases Wiki Activity

repro: state: rework state format and repro internals

Browse Source 
Signed-off-by: Ruslan Kuprieiev <[email protected]>
This commit is contained in:
Ruslan Kuprieiev
2017-11-26 13:42:46 +02:00
parent 3083604a9c
commit 7620e14c82
6 changed files with 110 additions and 382 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/cli.py
+5 -24
View File
@@ -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
dvc/command/add.py
+5 -4
View File
@@ -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))
dvc/command/repro.py
+16 -55
View File
@@ -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]
dvc/command/run.py
+50 -164
View File
@@ -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
dvc/repository_change.py
+3 -3
View File
@@ -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()
dvc/state_file.py
+31 -132
View File
@@ -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'