mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-11 07:18:28 +01:00
remote: progress callback for unpacked dir creation
This commit is contained in:
+6
-2
@@ -705,14 +705,18 @@ class RemoteBASE(object):
|
||||
def unprotect(path_info):
|
||||
pass
|
||||
|
||||
def _get_unpacked_dir_names(self, checksums):
|
||||
return set()
|
||||
|
||||
def extract_used_local_checksums(self, cinfos):
|
||||
from dvc.remote import RemoteLOCAL
|
||||
|
||||
used = {info[RemoteLOCAL.PARAM_CHECKSUM] for info in cinfos["local"]}
|
||||
return used
|
||||
unpacked = self._get_unpacked_dir_names(used)
|
||||
return used | unpacked
|
||||
|
||||
def _changed_unpacked_dir(self, checksum):
|
||||
return True
|
||||
|
||||
def _update_unpacked_dir(self, checksum):
|
||||
def _update_unpacked_dir(self, checksum, progress_callback=None):
|
||||
pass
|
||||
|
||||
@@ -35,7 +35,7 @@ from dvc.utils import (
|
||||
)
|
||||
from dvc.config import Config
|
||||
from dvc.exceptions import DvcException
|
||||
from dvc.progress import progress
|
||||
from dvc.progress import progress, ProgressCallback
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
|
||||
@@ -144,10 +144,11 @@ class RemoteLOCAL(RemoteBASE):
|
||||
return
|
||||
|
||||
if not link_type:
|
||||
self._default_link(from_info, to_info)
|
||||
link_types = self.cache_types
|
||||
else:
|
||||
link_method = self._get_link_method(link_type)
|
||||
self._link(from_info, to_info, link_method)
|
||||
link_types = [link_type]
|
||||
|
||||
self._try_links(from_info, to_info, link_types)
|
||||
|
||||
@classmethod
|
||||
def _get_link_method(cls, link_type):
|
||||
@@ -165,6 +166,7 @@ class RemoteLOCAL(RemoteBASE):
|
||||
)
|
||||
else:
|
||||
link_method(from_info.path, to_info.path)
|
||||
|
||||
if self.protected:
|
||||
self.protect(to_info)
|
||||
|
||||
@@ -178,18 +180,18 @@ class RemoteLOCAL(RemoteBASE):
|
||||
logger.debug(msg)
|
||||
|
||||
@slow_link_guard
|
||||
def _default_link(self, from_info, to_info):
|
||||
i = len(self.cache_types)
|
||||
def _try_links(self, from_info, to_info, link_types):
|
||||
i = len(link_types)
|
||||
while i > 0:
|
||||
link_method = self._get_link_method(self.cache_types[0])
|
||||
link_method = self._get_link_method(link_types[0])
|
||||
try:
|
||||
self._link(from_info, to_info, link_method)
|
||||
return
|
||||
|
||||
except DvcException as exc:
|
||||
msg = "Cache type '{}' is not supported: {}"
|
||||
logger.debug(msg.format(self.cache_types[0], str(exc)))
|
||||
del self.cache_types[0]
|
||||
logger.debug(msg.format(link_types[0], str(exc)))
|
||||
del link_types[0]
|
||||
i -= 1
|
||||
|
||||
raise DvcException("no possible cache types left to try out.")
|
||||
@@ -622,7 +624,7 @@ class RemoteLOCAL(RemoteBASE):
|
||||
try:
|
||||
dir_info = self.get_dir_cache(checksum)
|
||||
self._create_unpacked_dir(checksum, dir_info, unpacked_dir_info)
|
||||
except Exception:
|
||||
except DvcException:
|
||||
logger.warning(
|
||||
"Could not create '{}'".format(unpacked_dir_info.path)
|
||||
)
|
||||
@@ -630,8 +632,9 @@ class RemoteLOCAL(RemoteBASE):
|
||||
self.remove(unpacked_dir_info)
|
||||
|
||||
def _create_unpacked_dir(self, checksum, dir_info, unpacked_dir_info):
|
||||
progress_callback = ProgressCallback(len(dir_info))
|
||||
self.makedirs(unpacked_dir_info)
|
||||
for entry in dir_info:
|
||||
for index, entry in enumerate(dir_info):
|
||||
entry_cache_info = self.checksum_to_path_info(
|
||||
entry[self.PARAM_CHECKSUM]
|
||||
)
|
||||
@@ -642,6 +645,8 @@ class RemoteLOCAL(RemoteBASE):
|
||||
unpacked_dir_info.path, relpath
|
||||
)
|
||||
self.link(entry_cache_info, unpacked_entry_info, "hardlink")
|
||||
|
||||
progress_callback.update("Creating unpacked dir.")
|
||||
self.state.save(unpacked_dir_info, checksum)
|
||||
|
||||
def _changed_unpacked_dir(self, checksum):
|
||||
@@ -649,22 +654,9 @@ class RemoteLOCAL(RemoteBASE):
|
||||
|
||||
return not self.state.get(status_unpacked_dir_info)
|
||||
|
||||
def _save_dir(self, path_info, checksum):
|
||||
super(RemoteLOCAL, self)._save_dir(path_info, checksum)
|
||||
self._update_unpacked_dir(checksum)
|
||||
|
||||
def _checkout_dir(
|
||||
self, path_info, checksum, force, progress_callback=None
|
||||
):
|
||||
super(RemoteLOCAL, self)._checkout_dir(
|
||||
path_info, checksum, force, progress_callback
|
||||
)
|
||||
self._update_unpacked_dir(checksum)
|
||||
|
||||
def extract_used_local_checksums(self, cinfos):
|
||||
used = super(RemoteLOCAL, self).extract_used_local_checksums(cinfos)
|
||||
unpacked_dir = set()
|
||||
for checksum in used:
|
||||
if self.is_dir_checksum(checksum):
|
||||
unpacked_dir.add(self._append_unpacked_suffix(checksum))
|
||||
return used | unpacked_dir
|
||||
def _get_unpacked_dir_names(self, checksums):
|
||||
unpacked = set()
|
||||
for c in checksums:
|
||||
if self.is_dir_checksum(c):
|
||||
unpacked.add(self._append_unpacked_suffix(c))
|
||||
return unpacked
|
||||
|
||||
@@ -1,62 +1,10 @@
|
||||
import os
|
||||
import shutil
|
||||
|
||||
import mock
|
||||
|
||||
import dvc
|
||||
from dvc.main import main
|
||||
from dvc.remote.local import RemoteLOCAL
|
||||
from dvc.repo import Repo
|
||||
from dvc.utils import file_md5
|
||||
from dvc.utils.fs import get_inode
|
||||
from tests.utils import trees_equal
|
||||
|
||||
|
||||
def test_should_create_cache_status_check_dir_on_add(dvc_repo, repo_dir):
|
||||
|
||||
stages = dvc_repo.add(repo_dir.DATA_DIR)
|
||||
unpacked_dir = RemoteLOCAL._append_unpacked_suffix(
|
||||
stages[0].outs[0].cache_path
|
||||
)
|
||||
|
||||
assert os.path.exists(unpacked_dir)
|
||||
|
||||
trees_equal(repo_dir.DATA_DIR, unpacked_dir)
|
||||
data_cache_path = dvc_repo.cache.local.get(file_md5(repo_dir.DATA)[0])
|
||||
data_sub_cache_path = dvc_repo.cache.local.get(
|
||||
file_md5(repo_dir.DATA_SUB)[0]
|
||||
)
|
||||
assert get_inode(os.path.join(unpacked_dir, "data")) == get_inode(
|
||||
data_cache_path
|
||||
)
|
||||
assert get_inode(
|
||||
os.path.join(unpacked_dir, "data_sub_dir", "data_sub")
|
||||
) == get_inode(data_sub_cache_path)
|
||||
|
||||
|
||||
def test_should_create_cache_status_check_dir_on_checkout(dvc_repo, repo_dir):
|
||||
local_storage = repo_dir.mkdtemp()
|
||||
ret = main(["remote", "add", "-d", "storage", local_storage])
|
||||
assert ret == 0
|
||||
|
||||
dvc_repo = Repo(dvc_repo.dvc_dir)
|
||||
|
||||
stages = dvc_repo.add(repo_dir.DATA_DIR)
|
||||
|
||||
unpacked_dir = RemoteLOCAL._append_unpacked_suffix(
|
||||
stages[0].outs[0].cache_path
|
||||
)
|
||||
|
||||
assert dvc_repo.push() == 1
|
||||
|
||||
shutil.rmtree(dvc_repo.cache.local.cache_dir)
|
||||
shutil.rmtree(repo_dir.DATA_DIR)
|
||||
assert not os.path.exists(unpacked_dir)
|
||||
|
||||
dvc_repo.pull("{}.dvc".format(repo_dir.DATA_DIR))
|
||||
|
||||
assert os.path.exists(unpacked_dir)
|
||||
trees_equal(repo_dir.DATA_DIR, unpacked_dir)
|
||||
|
||||
|
||||
def test_should_not_fail_add_on_unpacked_dir_creation_exception(
|
||||
@@ -80,8 +28,6 @@ def test_should_create_unpacked_dir_on_status_check(dvc_repo, repo_dir):
|
||||
stages[0].outs[0].cache_path
|
||||
)
|
||||
|
||||
assert os.path.exists(unpacked_dir)
|
||||
shutil.rmtree(unpacked_dir)
|
||||
assert not os.path.exists(unpacked_dir)
|
||||
|
||||
assert dvc_repo.status("{}.dvc".format(repo_dir.DATA_DIR)) == {}
|
||||
|
||||
@@ -422,7 +422,11 @@ class TestCheckoutShouldHaveSelfClearingProgressBar(TestDvc):
|
||||
self.write_args = [w_c[1][0] for w_c in write_calls]
|
||||
|
||||
pattern = re.compile(".*\\[.{30}\\].*%.*")
|
||||
progress_bars = [arg for arg in self.write_args if pattern.match(arg)]
|
||||
progress_bars = [
|
||||
arg
|
||||
for arg in self.write_args
|
||||
if pattern.match(arg) and "unpacked" not in arg
|
||||
]
|
||||
|
||||
update_bars = progress_bars[:-1]
|
||||
finish_bar = progress_bars[-1]
|
||||
|
||||
Reference in New Issue
Block a user