mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-07 07:18:27 +01:00
Merge pull request #306 from efiop/master
Major misc changes(Cmdfile, repro, run, local_state etc)
This commit is contained in:
+2
-1
@@ -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
|
||||
|
||||
+11
-42
@@ -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(
|
||||
|
||||
@@ -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)
|
||||
@@ -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):
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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
|
||||
+13
-20
@@ -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)
|
||||
|
||||
+58
-185
@@ -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
|
||||
|
||||
+68
-151
@@ -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
|
||||
|
||||
+9
-9
@@ -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):
|
||||
|
||||
+30
-5
@@ -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])
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
+125
-141
@@ -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)
|
||||
|
||||
+2
-1
@@ -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() {
|
||||
|
||||
+1
-1
@@ -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"
|
||||
|
||||
Executable
+13
@@ -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
|
||||
@@ -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
|
||||
+10
-8
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user