mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-02 07:18:25 +01:00
Ruslan Kuprieiev
+8
-26
@@ -5,12 +5,12 @@ import sys
|
||||
import argparse
|
||||
from multiprocessing import cpu_count
|
||||
|
||||
from dvc.command.fsck import CmdFsck
|
||||
from dvc.command.init import CmdInit
|
||||
from dvc.command.remove import CmdRemove
|
||||
from dvc.command.run import CmdRun
|
||||
from dvc.command.repro import CmdRepro
|
||||
from dvc.command.data_sync import CmdDataPush, CmdDataPull, CmdDataStatus, CmdDataFetch
|
||||
from dvc.command.data_sync import CmdDataPush, CmdDataPull, CmdDataFetch
|
||||
from dvc.command.status import CmdDataStatus
|
||||
from dvc.command.gc import CmdGC
|
||||
from dvc.command.add import CmdAdd
|
||||
from dvc.command.config import CmdConfig
|
||||
@@ -162,7 +162,12 @@ def parse_args(argv=None):
|
||||
status_parser = subparsers.add_parser(
|
||||
'status',
|
||||
parents=[parent_cache_parser],
|
||||
help='Show mismatches between local cache and cloud cache')
|
||||
help='Show the project status')
|
||||
status_parser.add_argument('-c',
|
||||
'--cloud',
|
||||
action='store_true',
|
||||
default=False,
|
||||
help='Show status of a local cache compared to a cloud')
|
||||
status_parser.set_defaults(func=CmdDataStatus)
|
||||
|
||||
# Repro
|
||||
@@ -222,27 +227,4 @@ def parse_args(argv=None):
|
||||
help='Option value')
|
||||
config_parser.set_defaults(func=CmdConfig)
|
||||
|
||||
# Fsck
|
||||
fsck_parser = subparsers.add_parser(
|
||||
'fsck',
|
||||
parents=[parent_parser],
|
||||
help='Data file consistency check')
|
||||
fsck_parser.add_argument(
|
||||
'targets',
|
||||
nargs='*',
|
||||
help='Data files to check')
|
||||
fsck_parser.add_argument(
|
||||
'-p',
|
||||
'--physical',
|
||||
action='store_true',
|
||||
default=False,
|
||||
help='Compute actual md5')
|
||||
fsck_parser.add_argument(
|
||||
'-a',
|
||||
'--all',
|
||||
action='store_true',
|
||||
default=False,
|
||||
help='Show all data files including correct ones')
|
||||
fsck_parser.set_defaults(func=CmdFsck)
|
||||
|
||||
return parser.parse_args(argv)
|
||||
|
||||
+1
-1
@@ -57,7 +57,7 @@ class DataCloudAWS(DataCloudBase):
|
||||
See notes http://docs.aws.amazon.com/general/latest/gr/rande.html#s3_region
|
||||
"""
|
||||
|
||||
region = self._cloud_settings.cloud_config['Region']
|
||||
region = self._cloud_settings.cloud_config.get('Region', None)
|
||||
if region is None or region == '':
|
||||
return 's3.amazonaws.com'
|
||||
if region == 'us-east-1':
|
||||
|
||||
@@ -48,31 +48,3 @@ class CmdDataFetch(CmdDataBase):
|
||||
self.project.logger.error('Failed to fetch data from the cloud', exc)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
class CmdDataStatus(CmdDataBase):
|
||||
def _show(self, status):
|
||||
for s in status:
|
||||
target, ret = s
|
||||
|
||||
if ret == cloud.STATUS_UNKNOWN or ret == cloud.STATUS_OK:
|
||||
continue
|
||||
|
||||
prefix_map = {
|
||||
cloud.STATUS_DELETED : 'deleted: ',
|
||||
cloud.STATUS_MODIFIED : 'modified:',
|
||||
cloud.STATUS_NEW : 'new file:',
|
||||
}
|
||||
|
||||
path = os.path.relpath(target, self.project.cache.cache_dir)
|
||||
|
||||
self.project.logger.info('\t{}\t{}'.format(prefix_map[ret], path))
|
||||
|
||||
def do_run(self, target=None):
|
||||
try:
|
||||
status = self.project.status(target=target, jobs=self.args.jobs)
|
||||
self._show(status)
|
||||
except Exception as exc:
|
||||
self.project.logger.error('Failed to obtain data status', exc)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
@@ -1,172 +0,0 @@
|
||||
import os
|
||||
|
||||
from dvc import stage
|
||||
from dvc.command.common.base import CmdBase
|
||||
from dvc.utils import file_md5
|
||||
|
||||
|
||||
class FsckFileDep(object):
|
||||
def __init__(self, dvc_file_name, dep):
|
||||
self.dvc_file_name = dvc_file_name
|
||||
self.md5 = dep.md5
|
||||
self.type_name = type(dep).__name__
|
||||
self.use_cache = dep.use_cache if hasattr(dep, 'use_cache') else False
|
||||
|
||||
@property
|
||||
def is_output(self):
|
||||
return self.type_name == stage.Output.__name__
|
||||
|
||||
|
||||
class FsckFile(object):
|
||||
ERR_STATUS_NO_CACHE_FILE = 'No cache file found'
|
||||
ERR_STATUS_CHECKSUM_MISMATCH = 'Checksum missmatch'
|
||||
|
||||
def __init__(self, dvc_path, full_path, md5, hardlink_md5, state, fsck_deps):
|
||||
self.dvc_path = dvc_path
|
||||
self.full_path = full_path
|
||||
self.md5 = md5
|
||||
self.hardlink_md5 = hardlink_md5
|
||||
self.state = state
|
||||
self.fsck_deps = fsck_deps
|
||||
self.mtime = os.path.getmtime(full_path) if os.path.exists(full_path) else None
|
||||
|
||||
def hardlink_msg(self):
|
||||
if not self.is_data_file():
|
||||
return 'Not data file - no cache needed'
|
||||
elif self.hardlink_md5:
|
||||
return self.hardlink_md5
|
||||
return 'none'
|
||||
|
||||
def local_state_checksum_msg(self):
|
||||
return u'{}'.format(self.state.md5 if self.state else 'none')
|
||||
|
||||
def state_mtime_msg(self):
|
||||
return self.state.mtime if self.state else 'none'
|
||||
|
||||
def mtime_msg(self):
|
||||
return os.path.getmtime(self.full_path) if os.path.exists(self.full_path) else 'none'
|
||||
|
||||
def is_data_file(self):
|
||||
return any([d.is_output and d.use_cache for d in self.fsck_deps])
|
||||
|
||||
def checksums_mismatch(self):
|
||||
md5_set = {self.md5, self.hardlink_md5}
|
||||
md5_deps = set([d.md5 for d in self.fsck_deps])
|
||||
md5_set = md5_set.union(md5_deps)
|
||||
if None in md5_set:
|
||||
md5_set.remove(None)
|
||||
return len(md5_set) != 1
|
||||
|
||||
@property
|
||||
def error_status(self):
|
||||
if self.checksums_mismatch():
|
||||
return self.ERR_STATUS_CHECKSUM_MISMATCH
|
||||
if self.is_data_file() and self.hardlink_md5 is None:
|
||||
return self.ERR_STATUS_NO_CACHE_FILE
|
||||
return None
|
||||
|
||||
def print_info(self):
|
||||
print(u'File {}:'.format(self.dvc_path))
|
||||
|
||||
if self.error_status:
|
||||
print(u' Error status: {}!!!'.format(self.error_status))
|
||||
|
||||
if self.md5:
|
||||
print(u' Actual checksum: {}'.format(self.md5))
|
||||
|
||||
print(u' Cache file name: {}'.format(self.hardlink_msg()))
|
||||
print(u' Local state checksum: {}'.format(self.local_state_checksum_msg()))
|
||||
print(u' Local state mtime: {}'.format(self.state_mtime_msg()))
|
||||
print(u' Actual mtime: {}'.format(self.mtime_msg()))
|
||||
|
||||
for fsck_deps in self.fsck_deps:
|
||||
print(u' DVC file: {}'.format(fsck_deps.dvc_file_name))
|
||||
print(u' Checksum: {}'.format(fsck_deps.md5))
|
||||
print(u' Type: {}'.format(fsck_deps.type_name))
|
||||
if fsck_deps.use_cache:
|
||||
print(u' Use cache: {}'.format(str(fsck_deps.use_cache).lower()))
|
||||
pass
|
||||
|
||||
|
||||
class Fsck(object):
|
||||
def __init__(self, project, targets=None, physical=False, all=False):
|
||||
self.project = project
|
||||
self.targets = targets or []
|
||||
self.physical = physical
|
||||
self.all = all
|
||||
|
||||
self.files_and_stages = self.directions_by_datafile(self.all_directions())
|
||||
self.caches = self.project.cache.find_cache(self.files_and_stages)
|
||||
|
||||
self.fsck_objs = self.dvc_files_to_fsck_objs()
|
||||
|
||||
@property
|
||||
def is_target_defined(self):
|
||||
return len(self.targets) > 0
|
||||
|
||||
def file_list(self, files_and_stages, targets):
|
||||
if targets:
|
||||
dvc_files = [os.path.relpath(os.path.abspath(f), self.project.root_dir)
|
||||
for f in targets]
|
||||
else:
|
||||
dvc_files = files_and_stages.keys()
|
||||
return dvc_files
|
||||
|
||||
def all_directions(self):
|
||||
result = []
|
||||
for stage in self.project.stages():
|
||||
for dep in stage.deps + stage.outs:
|
||||
result.append((stage, dep))
|
||||
return result
|
||||
|
||||
def create_file_fsck_state(self, dvc_path):
|
||||
full_path = os.path.join(self.project.root_dir, dvc_path)
|
||||
|
||||
if self.physical:
|
||||
md5 = file_md5(full_path)[0]
|
||||
else:
|
||||
md5 = None
|
||||
|
||||
hardlink_md5 = self.caches.get(dvc_path)
|
||||
state = self.project.state.get(dvc_path)
|
||||
fsck_deps = [FsckFileDep(stage.dvc_path, dep)
|
||||
for stage, dep in self.files_and_stages.get(dvc_path, [])]
|
||||
|
||||
return FsckFile(dvc_path, full_path, md5, hardlink_md5, state, fsck_deps)
|
||||
|
||||
def dvc_files_to_fsck_objs(self):
|
||||
if not self.is_target_defined or self.all:
|
||||
files = self.files_and_stages.keys()
|
||||
else:
|
||||
files = map(lambda t: os.path.relpath(os.path.abspath(t), self.project.root_dir),
|
||||
self.targets)
|
||||
|
||||
fsck_objs = map(lambda file: self.create_file_fsck_state(file), files)
|
||||
|
||||
if not self.is_target_defined and not self.all:
|
||||
fsck_objs = filter(lambda o: o.error_status is not None, fsck_objs)
|
||||
|
||||
return list(fsck_objs)
|
||||
|
||||
@staticmethod
|
||||
def directions_by_datafile(directions):
|
||||
result = {}
|
||||
|
||||
for stage, dep in directions:
|
||||
if dep.dvc_path not in result:
|
||||
result[dep.dvc_path] = []
|
||||
result[dep.dvc_path].append((stage, dep))
|
||||
|
||||
return result
|
||||
|
||||
|
||||
class CmdFsck(CmdBase):
|
||||
def run(self):
|
||||
return self.fsck(self.args.targets, self.args.physical, self.args.all)
|
||||
|
||||
def fsck(self, targets, physical, all):
|
||||
checker = Fsck(self.project, targets, physical, all)
|
||||
|
||||
for file_fsck in checker.fsck_objs:
|
||||
file_fsck.print_info()
|
||||
return 0
|
||||
@@ -0,0 +1,36 @@
|
||||
import os
|
||||
|
||||
import dvc.cloud.base as cloud
|
||||
|
||||
from dvc.command.data_sync import CmdDataBase
|
||||
|
||||
|
||||
class CmdDataStatus(CmdDataBase):
|
||||
STATUS_LEN = 10
|
||||
STATUS_INDENT = '\t'
|
||||
|
||||
def _normalize(self, s):
|
||||
s+=':'
|
||||
if len(s) >= self.STATUS_LEN:
|
||||
return s
|
||||
return s + (self.STATUS_LEN - len(s))*' '
|
||||
|
||||
def _show(self, status, indent=0):
|
||||
ind = indent * self.STATUS_INDENT
|
||||
|
||||
for key, value in status.items():
|
||||
if isinstance(value, dict):
|
||||
self.project.logger.info('{}{}'.format(ind, key))
|
||||
self._show(value, indent+1)
|
||||
else:
|
||||
self.project.logger.info('{}{}{}'.format(ind, self._normalize(value), key))
|
||||
|
||||
def do_run(self, target=None):
|
||||
indent = 1 if self.args.cloud else 0
|
||||
try:
|
||||
status = self.project.status(target=target, jobs=self.args.jobs, cloud=self.args.cloud)
|
||||
self._show(status, indent)
|
||||
except Exception as exc:
|
||||
self.project.logger.error('Failed to obtain data status', exc)
|
||||
return 1
|
||||
return 0
|
||||
@@ -75,6 +75,12 @@ class Dependency(object):
|
||||
def changed(self):
|
||||
return self._changed_md5()
|
||||
|
||||
def status(self):
|
||||
if self.changed():
|
||||
#FIXME better msgs
|
||||
return {self.rel_path: 'changed'}
|
||||
return {}
|
||||
|
||||
@staticmethod
|
||||
def is_dir_cache(cache):
|
||||
return cache.endswith(Output.MD5_DIR_SUFFIX)
|
||||
|
||||
+37
-2
@@ -2,6 +2,8 @@ import os
|
||||
import stat
|
||||
import networkx as nx
|
||||
|
||||
import dvc.cloud.base as cloud
|
||||
|
||||
from dvc.logger import Logger
|
||||
from dvc.exceptions import DvcException
|
||||
from dvc.stage import Stage, Output
|
||||
@@ -209,8 +211,41 @@ class Project(object):
|
||||
self.checkout()
|
||||
return ret
|
||||
|
||||
def status(self, target=None, jobs=1):
|
||||
return self.cloud.status(self._used_cache(target), jobs)
|
||||
def _local_status(self, target=None):
|
||||
status = {}
|
||||
|
||||
if target:
|
||||
stages = [Stage.load(self, target)]
|
||||
else:
|
||||
stages = self.stages()
|
||||
|
||||
for stage in self.stages():
|
||||
status.update(stage.status())
|
||||
|
||||
return status
|
||||
|
||||
def _cloud_status(self, target=None, jobs=1):
|
||||
status = {}
|
||||
for target, ret in self.cloud.status(self._used_cache(target), jobs):
|
||||
if ret == cloud.STATUS_UNKNOWN or ret == cloud.STATUS_OK:
|
||||
continue
|
||||
|
||||
prefix_map = {
|
||||
cloud.STATUS_DELETED: 'deleted',
|
||||
cloud.STATUS_MODIFIED: 'modified',
|
||||
cloud.STATUS_NEW: 'new',
|
||||
}
|
||||
|
||||
path = os.path.relpath(target, self.cache.cache_dir)
|
||||
|
||||
status[path] = prefix_map[ret]
|
||||
|
||||
return status
|
||||
|
||||
def status(self, target=None, jobs=1, cloud=False):
|
||||
if cloud:
|
||||
return self._cloud_status(target, jobs)
|
||||
return self._local_status(target)
|
||||
|
||||
def graph(self):
|
||||
G = nx.DiGraph()
|
||||
|
||||
@@ -183,3 +183,24 @@ class Stage(object):
|
||||
def checkout(self):
|
||||
for out in self.outs:
|
||||
out.checkout()
|
||||
|
||||
def _status(self, entries, name):
|
||||
ret = {}
|
||||
|
||||
for entry in entries:
|
||||
ret.update(entry.status())
|
||||
|
||||
if ret:
|
||||
return {name: ret}
|
||||
|
||||
return {}
|
||||
|
||||
def status(self):
|
||||
ret = {}
|
||||
ret.update(self._status(self.deps, 'deps'))
|
||||
ret.update(self._status(self.outs, 'outs'))
|
||||
|
||||
if ret:
|
||||
return {self.relpath: ret}
|
||||
|
||||
return {}
|
||||
|
||||
@@ -12,3 +12,4 @@ pyyaml>=3.12
|
||||
gitpython>=2.1.8
|
||||
ntfsutils>=0.1.3
|
||||
checksumdir>=1.1.4
|
||||
setuptools>=34.0.0
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
from PyInstaller.utils.hooks import copy_metadata
|
||||
datas = copy_metadata('google-api-core')
|
||||
@@ -17,6 +17,7 @@ install_requires = [
|
||||
"gitpython>=2.1.8",
|
||||
"ntfsutils>=0.1.3",
|
||||
"checksumdir>=1.1.4",
|
||||
"setuptools>=34.0.0",
|
||||
]
|
||||
|
||||
setup(
|
||||
|
||||
+2
-17
@@ -3,14 +3,14 @@ import os
|
||||
from dvc.cli import parse_args
|
||||
from dvc.command.init import CmdInit
|
||||
from dvc.command.run import CmdRun
|
||||
from dvc.command.data_sync import CmdDataPull, CmdDataPush, CmdDataStatus
|
||||
from dvc.command.data_sync import CmdDataPull, CmdDataPush
|
||||
from dvc.command.status import CmdDataStatus
|
||||
from dvc.command.repro import CmdRepro
|
||||
from dvc.command.add import CmdAdd
|
||||
from dvc.command.remove import CmdRemove
|
||||
from dvc.command.gc import CmdGC
|
||||
from dvc.command.config import CmdConfig
|
||||
from dvc.command.checkout import CmdCheckout
|
||||
from dvc.command.fsck import CmdFsck
|
||||
|
||||
from tests.basic_env import TestDvc
|
||||
|
||||
@@ -146,18 +146,3 @@ class TestCheckout(TestDvc):
|
||||
def test(self):
|
||||
args = parse_args(['checkout'])
|
||||
self.assertIsInstance(args.func(args), CmdCheckout)
|
||||
|
||||
|
||||
class TestFsck(TestDvc):
|
||||
def test(self):
|
||||
target1 = 'target1'
|
||||
target2 = 'target2'
|
||||
|
||||
args = parse_args(['fsck',
|
||||
'-p', '--physical',
|
||||
target1,
|
||||
target2])
|
||||
|
||||
self.assertIsInstance(args.func(args), CmdFsck)
|
||||
self.assertTrue(args.physical)
|
||||
self.assertEqual(args.targets, [target1, target2])
|
||||
|
||||
@@ -1,125 +0,0 @@
|
||||
import os
|
||||
|
||||
from dvc import cli
|
||||
from tests.basic_env import TestDvc
|
||||
from dvc.command.fsck import Fsck, FsckFile, CmdFsck
|
||||
|
||||
|
||||
class TestFsck(TestDvc):
|
||||
def no_file_corruptions_test(self):
|
||||
fsck_objs = Fsck(self.dvc).fsck_objs
|
||||
self.assertEqual(len(fsck_objs), 0)
|
||||
|
||||
def get_all_test(self):
|
||||
self.dvc.add(self.FOO)
|
||||
self.dvc.add(self.BAR)
|
||||
fsck_objs = Fsck(self.dvc, all=True).fsck_objs
|
||||
|
||||
self.assertEqual(len(fsck_objs), 2)
|
||||
for obj in fsck_objs:
|
||||
self.assertIsNone(obj.error_status)
|
||||
|
||||
def get_one_of_many_test(self):
|
||||
self.dvc.add(self.FOO)
|
||||
self.dvc.add(self.BAR)
|
||||
fsck_objs = Fsck(self.dvc, [self.FOO]).fsck_objs
|
||||
|
||||
self.assertEqual(len(fsck_objs), 1)
|
||||
self.assertEqual(fsck_objs[0].dvc_path, self.FOO)
|
||||
|
||||
def by_file_test(self):
|
||||
self.dvc.add(self.FOO)
|
||||
self.dvc.add(self.BAR)
|
||||
fsck_objs = Fsck(self.dvc, [self.FOO, self.BAR]).fsck_objs
|
||||
|
||||
self.assertEqual(len(fsck_objs), 2)
|
||||
for obj in fsck_objs:
|
||||
self.assertIsNone(obj.error_status)
|
||||
|
||||
def defined_target_with_mismatch_test(self):
|
||||
stage_foo = self.dvc.add(self.FOO)
|
||||
self.dvc.add(self.BAR)
|
||||
|
||||
for out in stage_foo.outs:
|
||||
if out.dvc_path == self.FOO:
|
||||
out.md5 = out.md5 + '1234fe'
|
||||
stage_foo.dump()
|
||||
|
||||
fsck_objs = Fsck(self.dvc, [self.FOO, self.BAR]).fsck_objs
|
||||
self.assertEqual(len(fsck_objs), 2)
|
||||
|
||||
def checksum_mismatch_test(self):
|
||||
stage_foo = self.dvc.add(self.FOO)
|
||||
self.dvc.add(self.BAR)
|
||||
|
||||
for out in stage_foo.outs:
|
||||
if out.dvc_path == self.FOO:
|
||||
out.md5 = out.md5 + '1234fe'
|
||||
stage_foo.dump()
|
||||
|
||||
fsck_objs = Fsck(self.dvc).fsck_objs
|
||||
self.assertEqual(len(fsck_objs), 1)
|
||||
self.assertEqual(fsck_objs[0].dvc_path, self.FOO)
|
||||
self.assertEqual(fsck_objs[0].error_status, FsckFile.ERR_STATUS_CHECKSUM_MISMATCH)
|
||||
|
||||
def no_cache_file_test(self):
|
||||
stage_foo = self.dvc.add(self.FOO)
|
||||
self.dvc.add(self.BAR)
|
||||
|
||||
md5 = ''
|
||||
for out in stage_foo.outs:
|
||||
if out.dvc_path == self.FOO:
|
||||
md5 = out.md5
|
||||
|
||||
foo_cache_file = os.path.join(self.dvc.cache.cache_dir, md5[0:2], md5[2:])
|
||||
self.assertTrue(os.path.exists(foo_cache_file))
|
||||
|
||||
os.remove(foo_cache_file)
|
||||
self.assertFalse(os.path.exists(foo_cache_file))
|
||||
|
||||
fsck_objs = Fsck(self.dvc).fsck_objs
|
||||
self.assertEqual(len(fsck_objs), 1)
|
||||
self.assertEqual(fsck_objs[0].dvc_path, self.FOO)
|
||||
self.assertEqual(fsck_objs[0].error_status, FsckFile.ERR_STATUS_NO_CACHE_FILE)
|
||||
|
||||
def physical_test(self):
|
||||
self.dvc.add(self.FOO)
|
||||
self.dvc.add(self.BAR)
|
||||
|
||||
os.remove(self.FOO)
|
||||
with open(self.FOO, 'w') as fd:
|
||||
fd.write('randome stuff s9dfj')
|
||||
|
||||
fsck_objs = Fsck(self.dvc, physical=True).fsck_objs
|
||||
|
||||
self.assertEqual(len(fsck_objs), 1)
|
||||
self.assertEqual(fsck_objs[0].dvc_path, self.FOO)
|
||||
|
||||
def by_timestemp_physical_test(self):
|
||||
self.dvc.add(self.FOO)
|
||||
self.dvc.add(self.BAR)
|
||||
|
||||
os.remove(self.FOO)
|
||||
with open(self.FOO, 'w') as fd:
|
||||
fd.write('randome stuff 39re2fwecsb nj')
|
||||
|
||||
fsck_objs = Fsck(self.dvc, physical=False).fsck_objs
|
||||
|
||||
self.assertEqual(len(fsck_objs), 1)
|
||||
self.assertEqual(fsck_objs[0].dvc_path, self.FOO)
|
||||
|
||||
def end_to_end_test(self):
|
||||
self.dvc.add(self.FOO)
|
||||
self.dvc.add(self.BAR)
|
||||
|
||||
os.remove(self.FOO)
|
||||
|
||||
no_data_file = 'baz'
|
||||
open(no_data_file, 'w').write('asd')
|
||||
|
||||
with open(self.FOO, 'w') as fd:
|
||||
fd.write('randome stuff s9dfj')
|
||||
|
||||
parser = cli.parse_args(['fsck', self.FOO, self.BAR, no_data_file])
|
||||
cmd = CmdFsck(args=parser)
|
||||
cmd.run()
|
||||
Reference in New Issue
Block a user