diff --git a/dvc/cli.py b/dvc/cli.py index b025fb490..175e10bb1 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -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='*', diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index 5c0b2fac6..6d6f73a11 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -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', diff --git a/dvc/command/status.py b/dvc/command/status.py index 63b92e6a4..33b3ce3e5 100644 --- a/dvc/command/status.py +++ b/dvc/command/status.py @@ -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 diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index 1f296e5cc..ccf83de02 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -94,26 +94,29 @@ class DataCloud(object): "with 'dvc config core.remote ' or use '-r '." 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) diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py index 1d9a9a0c5..e8de9f68b 100644 --- a/dvc/dependency/local.py +++ b/dvc/dependency/local.py @@ -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 diff --git a/dvc/progress.py b/dvc/progress.py index b49c25045..6de01d021 100644 --- a/dvc/progress.py +++ b/dvc/progress.py @@ -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() diff --git a/dvc/project.py b/dvc/project.py index 26375d46b..97dcee0c1 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -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): diff --git a/dvc/remote/local.py b/dvc/remote/local.py index 7464ab9eb..b70e4ea80 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -20,6 +20,7 @@ class RemoteLOCAL(RemoteBase): scheme = '' REGEX = r'^(?P(/+|.:\\+).*)$' 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: diff --git a/dvc/scm.py b/dvc/scm.py index 19779b77b..6afc00bae 100644 --- a/dvc/scm.py +++ b/dvc/scm.py @@ -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) diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py index fecb09f81..1c38a20e8 100644 --- a/tests/test_data_cloud.py +++ b/tests/test_data_cloud.py @@ -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