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

Merge pull request #952 from efiop/master

Browse Source 
dvc: show filenames in `push/pull` output
This commit is contained in:
Ruslan Kuprieiev
2018-07-26 21:48:06 +03:00
committed by GitHub
parent c408fa7530 7e4f39d8eb
commit 627606d6f2
10 changed files with 176 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/cli.py
+5
View File
@@ -208,6 +208,11 @@ def parse_args(argv=None):
type=int,
default=8*cpu_count(),
help='Number of jobs to run simultaneously.')
parent_cache_parser.add_argument(
'--show-checksums',
action='store_true',
default=False,
help='Show checksums instead of file names.')
parent_cache_parser.add_argument(
'targets',
nargs='*',
dvc/command/data_sync.py
+3
View File
@@ -22,6 +22,7 @@ class CmdDataPull(CmdDataBase):
self.project.pull(target=target,
jobs=self.args.jobs,
remote=self.args.remote,
show_checksums=self.args.show_checksums,
all_branches=self.args.all_branches)
except Exception as exc:
self.project.logger.error('Failed to pull data from the cloud',
@@ -36,6 +37,7 @@ class CmdDataPush(CmdDataBase):
self.project.push(target=target,
jobs=self.args.jobs,
remote=self.args.remote,
show_checksums=self.args.show_checksums,
all_branches=self.args.all_branches)
except Exception as exc:
self.project.logger.error('Failed to push data to the cloud', exc)
@@ -49,6 +51,7 @@ class CmdDataFetch(CmdDataBase):
self.project.fetch(target=target,
jobs=self.args.jobs,
remote=self.args.remote,
show_checksums=self.args.show_checksums,
all_branches=self.args.all_branches)
except Exception as exc:
self.project.logger.error('Failed to fetch data from the cloud',
dvc/command/status.py
+6 -5
View File
@@ -24,11 +24,12 @@ class CmdDataStatus(CmdDataBase):
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,
remote=self.args.remote)
self._show(status, indent)
st = self.project.status(target=target,
jobs=self.args.jobs,
cloud=self.args.cloud,
show_checksums=self.args.show_checksums,
remote=self.args.remote)
self._show(st, indent)
except Exception as exc:
self.project.logger.error('Failed to obtain data status', exc)
return 1
dvc/data_cloud.py
+9 -6
View File
@@ -94,26 +94,29 @@ class DataCloud(object):
"with 'dvc config core.remote <name>' or use '-r <name>'."
raise ConfigError(msg)
def push(self, targets, jobs=1, remote=None):
def push(self, targets, jobs=1, remote=None, show_checksums=False):
"""
Push data items in a cloud-agnostic way.
"""
return self.project.cache.local.push(targets,
jobs=jobs,
remote=self._get_cloud(remote))
remote=self._get_cloud(remote),
show_checksums=show_checksums)
def pull(self, targets, jobs=1, remote=None):
def pull(self, targets, jobs=1, remote=None, show_checksums=False):
"""
Pull data items in a cloud-agnostic way.
"""
return self.project.cache.local.pull(targets,
jobs=jobs,
remote=self._get_cloud(remote))
remote=self._get_cloud(remote),
show_checksums=show_checksums)
def status(self, targets, jobs=1, remote=None):
def status(self, targets, jobs=1, remote=None, show_checksums=False):
"""
Check status of data items in a cloud-agnostic way.
"""
return self.project.cache.local.status(targets,
jobs=jobs,
remote=self._get_cloud(remote))
remote=self._get_cloud(remote),
show_checksums=show_checksums)
dvc/dependency/local.py
+1 -1
View File
@@ -74,6 +74,6 @@ class DependencyLOCAL(DependencyBase):
else:
path = self.path
info = self.info
info = self.info.copy()
info[self.PARAM_PATH] = path
return info
dvc/progress.py
+1 -8
View File
@@ -53,7 +53,6 @@ class Progress(object):
Make a progress bar out of info, which looks like:
(1/2): [########################################] 100% master.zip
"""
total_len = 100
bar_len = 30
if total is None:
@@ -70,13 +69,7 @@ class Progress(object):
n_sp = bar_len - n_sh
bar = "[" + '#'*n_sh + ' '*n_sp + "] "
name_len = total_len - len(num + bar + percent)
if len(target_name) > name_len:
name = target_name[:name_len]
else:
name = target_name
return num + bar + percent + name
return num + bar + percent + target_name
progress = Progress()
dvc/project.py
+48 -11
View File
@@ -326,7 +326,10 @@ class Project(object):
if not out.use_cache or not out.info:
continue
cache[out.path_info['scheme']] += [out.info]
info = out.dumpd()
info['branch'] = branch
cache[out.path_info['scheme']] += [info]
return cache
@@ -354,18 +357,39 @@ class Project(object):
if cloud:
self.cloud._get_cloud(remote).gc(clist)
def push(self, target=None, jobs=1, remote=None, all_branches=False):
def push(self,
target=None,
jobs=1,
remote=None,
all_branches=False,
show_checksums=False):
self.cloud.push(self._used_cache(target, all_branches)['local'],
jobs,
remote=remote)
remote=remote,
show_checksums=show_checksums)
def fetch(self, target=None, jobs=1, remote=None, all_branches=False):
def fetch(self,
target=None,
jobs=1,
remote=None,
all_branches=False,
show_checksums=False):
self.cloud.pull(self._used_cache(target, all_branches)['local'],
jobs,
remote=remote)
remote=remote,
show_checksums=show_checksums)
def pull(self, target=None, jobs=1, remote=None, all_branches=False):
self.fetch(target, jobs, remote=remote, all_branches=all_branches)
def pull(self,
target=None,
jobs=1,
remote=None,
all_branches=False,
show_checksums=False):
self.fetch(target,
jobs,
remote=remote,
all_branches=all_branches,
show_checksums=show_checksums)
self.checkout(target=target)
def _local_status(self, target=None):
@@ -386,13 +410,18 @@ class Project(object):
return status
def _cloud_status(self, target=None, jobs=1, remote=None):
def _cloud_status(self,
target=None,
jobs=1,
remote=None,
show_checksums=False):
import dvc.remote.base as cloud
status = {}
for md5, ret in self.cloud.status(self._used_cache(target)['local'],
jobs,
remote=remote):
remote=remote,
show_checksums=show_checksums):
if ret == cloud.STATUS_OK:
continue
@@ -405,9 +434,17 @@ class Project(object):
return status
def status(self, target=None, jobs=1, cloud=False, remote=None):
def status(self,
target=None,
jobs=1,
cloud=False,
remote=None,
show_checksums=False):
if cloud:
return self._cloud_status(target, jobs, remote=remote)
return self._cloud_status(target,
jobs,
remote=remote,
show_checksums=show_checksums)
return self._local_status(target)
def _read_metric_json(self, fd, json_path):
dvc/remote/local.py
+89 -30
View File
@@ -20,6 +20,7 @@ class RemoteLOCAL(RemoteBase):
scheme = ''
REGEX = r'^(?P<path>(/+|.:\\+).*)$'
PARAM_MD5 = 'md5'
PARAM_PATH = 'path'
PARAM_RELPATH = 'relpath'
MD5_DIR_SUFFIX = '.dir'
@@ -388,15 +389,48 @@ class RemoteLOCAL(RemoteBase):
for info in checksum_infos:
md5 = info[self.PARAM_MD5]
cache = self.get(md5)
if not self.is_dir_cache(info[self.PARAM_MD5]):
continue
if not os.path.exists(cache):
missing.append(info)
continue
collected.extend(self.load_dir_cache(md5))
for i in self.load_dir_cache(md5):
if info.get('branch'):
i['branch'] = info['branch']
i[self.PARAM_PATH] = posixpath.join(info[self.PARAM_PATH],
i[self.PARAM_RELPATH])
collected.append(i.copy())
collected.extend(checksum_infos)
return collected, missing
def _group(self, checksum_infos, show_checksums=False):
by_md5 = {}
for info in checksum_infos:
md5 = info[self.PARAM_MD5]
if show_checksums:
by_md5[md5] = md5
continue
name = info[self.PARAM_PATH]
branch = info.get('branch')
if branch:
name += '({})'.format(branch)
if md5 not in by_md5.keys():
by_md5[md5] = ''
else:
by_md5[md5] += ' '
by_md5[md5] += name
return list(by_md5.keys()), list(by_md5.values())
def gc(self, checksum_infos):
checksum_infos = self._collect(checksum_infos['local'])[0]
used_md5s = [info[self.PARAM_MD5] for info in checksum_infos]
@@ -406,93 +440,118 @@ class RemoteLOCAL(RemoteBase):
continue
remove(self.get(md5))
def status(self, checksum_infos, remote, jobs=1):
def status(self, checksum_infos, remote, jobs=1, show_checksums=False):
checksum_infos = self._collect(checksum_infos)[0]
md5s = [info[self.PARAM_MD5] for info in checksum_infos]
md5s, names = self._group(checksum_infos,
show_checksums=show_checksums)
path_infos = remote.md5s_to_path_infos(md5s)
remote_exists = remote.exists(path_infos)
local_exists = [not self.changed(md5) for md5 in md5s]
return [(md5, STATUS_MAP[l, r]) for md5, l, r in zip(md5s,
local_exists,
remote_exists)]
def _do_pull(self, checksum_infos, remote, jobs=1, no_progress_bar=False):
md5s = [info[self.PARAM_MD5] for info in checksum_infos]
return [(name, STATUS_MAP[l, r]) for name, l, r in zip(names,
local_exists,
remote_exists)]
def _do_pull(self,
checksum_infos,
remote,
jobs=1,
no_progress_bar=False,
show_checksums=False):
md5s = []
names = []
# NOTE: filter files that are not corrupted
md5s = list(filter(lambda md5: self.changed(md5), md5s))
for md5, name in zip(*self._group(checksum_infos,
show_checksums=show_checksums)):
if self.changed(md5):
md5s.append(md5)
names.append(name)
cache = [{'scheme': 'local', 'path': self.get(md5)} for md5 in md5s]
path_infos = remote.md5s_to_path_infos(md5s)
assert len(path_infos) == len(cache) == len(md5s)
assert len(path_infos) == len(cache) == len(md5s) == len(names)
chunks = list(zip(to_chunks(path_infos, jobs),
to_chunks(cache, jobs),
to_chunks(md5s, jobs)))
to_chunks(names, jobs)))
progress.set_n_total(len(md5s))
progress.set_n_total(len(names))
if len(chunks) == 0:
return
futures = []
with ThreadPoolExecutor(max_workers=len(chunks)) as executor:
for from_infos, to_infos, md5s in chunks:
for from_infos, to_infos, names in chunks:
res = executor.submit(remote.download,
from_infos,
to_infos,
names=md5s,
names=names,
no_progress_bar=no_progress_bar)
futures.append(res)
for f in futures:
f.result()
def pull(self, checksum_infos, remote, jobs=1):
def pull(self, checksum_infos, remote, jobs=1, show_checksums=False):
# NOTE: try fetching missing dir info
checksum_infos, missing = self._collect(checksum_infos)
if len(missing) > 0:
self._do_pull(missing, remote, jobs, no_progress_bar=True)
self._do_pull(missing,
remote,
jobs,
no_progress_bar=True,
show_checksums=show_checksums)
checksum_infos += self._collect(missing)[0]
self._do_pull(checksum_infos, remote, jobs)
self._do_pull(checksum_infos,
remote,
jobs,
show_checksums=show_checksums)
def push(self, checksum_infos, remote, jobs=1):
def push(self, checksum_infos, remote, jobs=1, show_checksums=False):
checksum_infos = self._collect(checksum_infos)[0]
md5s = [info[self.PARAM_MD5] for info in checksum_infos]
# NOTE: verifying that our cache is not corrupted
md5s = list(filter(lambda md5: not self.changed(md5), md5s))
def func(info):
return not self.changed(info[self.PARAM_MD5])
checksum_infos = list(filter(func, checksum_infos))
# NOTE: filter files that are already uploaded
path_infos = remote.md5s_to_path_infos(md5s)
lexists = remote.exists(path_infos)
md5s_exist = filter(lambda x: not x[1], list(zip(md5s, lexists)))
md5s = [md5 for md5, exists in md5s_exist]
md5s = [i[self.PARAM_MD5] for i in checksum_infos]
exists = remote.exists(remote.md5s_to_path_infos(md5s))
def func(entry):
return not entry[0]
assert len(exists) == len(checksum_infos)
infos_exist = list(filter(func, zip(exists, checksum_infos)))
checksum_infos = [i for e, i in infos_exist]
md5s, names = self._group(checksum_infos,
show_checksums=show_checksums)
cache = [{'scheme': 'local', 'path': self.get(md5)} for md5 in md5s]
path_infos = remote.md5s_to_path_infos(md5s)
assert len(path_infos) == len(cache) == len(md5s)
assert len(path_infos) == len(cache) == len(md5s) == len(names)
chunks = list(zip(to_chunks(path_infos, jobs),
to_chunks(cache, jobs),
to_chunks(md5s, jobs)))
to_chunks(names, jobs)))
progress.set_n_total(len(md5s))
progress.set_n_total(len(names))
if len(chunks) == 0:
return
futures = []
with ThreadPoolExecutor(max_workers=len(chunks)) as executor:
for to_infos, from_infos, md5s in chunks:
for to_infos, from_infos, names in chunks:
res = executor.submit(remote.upload,
from_infos,
to_infos,
names=md5s)
names=names)
futures.append(res)
for f in futures:
dvc/scm.py
-1
View File
@@ -53,7 +53,6 @@ class Base(object):
if not branches:
branches = self.list_branches() if all_branches else [saved]
for branch in branches:
print("OLOLO " + branch)
self.checkout(branch)
yield branch
self.checkout(saved)
tests/test_data_cloud.py
+14 -14
View File
@@ -206,20 +206,20 @@ class TestDataCloudBase(TestDvc):
stage = self.dvc.add(self.FOO)
cache = stage.outs[0].cache
info = stage.outs[0].info
info = stage.outs[0].dumpd()
md5 = info['md5']
stage_dir = self.dvc.add(self.DATA_DIR)
cache_dir = stage_dir.outs[0].cache
info_dir = stage_dir.outs[0].info
info_dir = stage_dir.outs[0].dumpd()
md5_dir = info_dir['md5']
# Check status
sleep()
status = self.cloud.status([info])
status = self.cloud.status([info], show_checksums=True)
self.assertEqual([(md5, STATUS_NEW)], status)
status_dir = self.cloud.status([info_dir])
status_dir = self.cloud.status([info_dir], show_checksums=True)
self.assertTrue((md5_dir, STATUS_NEW) in status_dir)
# Push and check status
@@ -231,20 +231,20 @@ class TestDataCloudBase(TestDvc):
self.assertTrue(os.path.isfile(cache_dir))
sleep()
status = self.cloud.status([info])
status = self.cloud.status([info], show_checksums=True)
self.assertEqual([(md5, STATUS_OK)], status)
status_dir = self.cloud.status([info_dir])
status_dir = self.cloud.status([info_dir], show_checksums=True)
self.assertTrue((md5_dir, STATUS_OK) in status_dir)
# Remove and check status
sleep()
shutil.rmtree(self.dvc.cache.local.cache_dir)
status = self.cloud.status([info])
status = self.cloud.status([info], show_checksums=True)
self.assertEqual([(md5, STATUS_DELETED)], status)
status_dir = self.cloud.status([info_dir])
status_dir = self.cloud.status([info_dir], show_checksums=True)
self.assertEqual([(md5_dir, STATUS_DELETED)], status_dir)
# Pull and check status
@@ -258,10 +258,10 @@ class TestDataCloudBase(TestDvc):
self.assertTrue(os.path.isfile(cache_dir))
sleep()
status = self.cloud.status([info])
status = self.cloud.status([info], show_checksums=True)
self.assertEqual([(md5, STATUS_OK)], status)
status_dir = self.cloud.status([info_dir])
status_dir = self.cloud.status([info_dir], show_checksums=True)
self.assertTrue((md5_dir, STATUS_OK) in status_dir)
def test(self):
@@ -366,7 +366,7 @@ class TestDataCloudCLIBase(TestDvc):
# FIXME check status output
sleep()
self.main(['status', '-c'] + args)
self.main(['status', '-c', '--show-checksums'] + args)
self.main(['push'] + args)
self.assertTrue(os.path.exists(cache))
@@ -374,12 +374,12 @@ class TestDataCloudCLIBase(TestDvc):
self.assertTrue(os.path.isfile(cache_dir))
sleep()
self.main(['status', '-c'] + args)
self.main(['status', '-c', '--show-checksums'] + args)
shutil.rmtree(self.dvc.cache.local.cache_dir)
sleep()
self.main(['status', '-c'] + args)
self.main(['status', '-c', '--show-checksums'] + args)
self.main(['fetch'] + args)
self.assertTrue(os.path.exists(cache))
@@ -398,7 +398,7 @@ class TestDataCloudCLIBase(TestDvc):
self.assertEqual(fd.read(), self.FOO_CONTENTS)
self.assertTrue(os.path.isfile(cache_dir))
self.main(['status', '-c'] + args)
self.main(['status', '-c', '--show-checksums'] + args)
def _should_test(self):
return True