except BlockSizeError from fsspec
MODIFY
pangeo_forge_recipes/storage.py
pangeo_forge_recipes/storage.py
@@ -24,26 +24,22 @@ def _get_url_size(fname):
return size
-@contextmanager
-def _fsspec_safe_open(fname: str, **kwargs) -> Iterator[OpenFileType]:
- # workaround for inconsistent behavior of fsspec.open
- # https://github.com/intake/filesystem_spec/issues/579
- with fsspec.open(fname, **kwargs) as fp:
- with fp as fp2:
- yield fp2
-
-
def _copy_btw_filesystems(input_opener, output_opener, BLOCK_SIZE=10_000_000):
with input_opener as source:
with output_opener as target:
while True:
try:
+ logger.debug("_copy_btw_filesystems reading data")
data = source.read(BLOCK_SIZE)
if not data:
break
+ logger.debug(f"_copy_btw_filesystems copying block of {len(data)} bytes")
target.write(data)
- except ValueError:
- raise BlockSizeError()
+ except BlockSizeError:
+ logger.error(
+ "Try re-instantiating recipe with fsspec_open_kwargs = {\"block_size\": 0}"
+ )
+ logger.debug("_copy_btw_filesystems done")
class AbstractTarget(ABC):
@@ -108,8 +104,12 @@ class FSSpecTarget(AbstractTarget):
@contextmanager
def open(self, path: str, **kwargs) -> Iterator[None]:
"""Open file with a context manager."""
- with self.fs.open(self._full_path(path), **kwargs) as f:
+ full_path = self._full_path(path)
+ logger.debug(f"entering fs.open context manager for {full_path}")
+ with self.fs.open(full_path, **kwargs) as f:
+ logger.debug(f"FSSpecTarget.open yielding {f}")
yield f
+ logger.debug("FSSpecTarget.open yielded")
def __post_init__(self):
if not self.fs.isdir(self.root_path):
@@ -117,7 +117,7 @@ class FSSpecTarget(AbstractTarget):
class FlatFSSpecTarget(FSSpecTarget):
- """A target that sanitizes all the path names so that everthing is stored
+ """A target that sanitizes all the path names so that everything is stored
in a single directory.
Designed to be used as a cache for inputs.
@@ -145,7 +145,7 @@ class CacheFSSpecTarget(FlatFSSpecTarget):
logger.info(f"File '{fname}' is already cached")
return
- input_opener = _fsspec_safe_open(fname, mode="rb", **open_kwargs)
+ input_opener = fsspec.open(fname, mode="rb", **open_kwargs)
target_opener = self.open(fname, mode="wb")
logger.info(f"Coping remote file '{fname}' to cache")
_copy_btw_filesystems(input_opener, target_opener)
@@ -189,7 +189,7 @@ def file_opener(
opener = cache.open(fname, mode="rb")
else:
logger.info(f"Opening '{fname}' directly.")
- opener = _fsspec_safe_open(fname, mode="rb", **open_kwargs)
+ opener = fsspec.open(fname, mode="rb", **open_kwargs)
if copy_to_local:
_, suffix = os.path.splitext(fname)
ntf = tempfile.NamedTemporaryFile(suffix=suffix)
@@ -200,8 +200,12 @@ def file_opener(
yield tmp_name
ntf.close() # cleans up the temporary file
else:
+ logger.debug(f"file_opener entering first context for {opener}")
with opener as fp:
+ logger.debug(f"file_opener entering second context for {fp}")
yield fp
+ logger.debug("file_opener yielded")
+ logger.debug("opener done")
def _slugify(value: str) -> str:
@@ -212,16 +216,3 @@ def _slugify(value: str) -> str:
value = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii")
value = re.sub(r"[^.\w\s-]+", "_", value.lower())
return re.sub(r"[-\s]+", "-", value).strip("-_")
\ No newline at end of file
-
-
-class StorageError(Exception):
- """Base class for exceptions in this module."""
-
- pass
-
-
-class BlockSizeError(StorageError):
- """Error for source file servers that require {"block_size": 0}"""
-
- def __str__(self):
- return 'Try re-instantiating your recipe with fsspec_open_kwargs = {"block_size": 0}'

except BlockSizeError from fsspec