Remove some contexts and simplify storage code
MODIFY
ci/py3.8.yml
ci/py3.8.yml
@@ -10,7 +10,7 @@ dependencies:
- codecov
- dask
- distributed
- # - fsspec>=2021.04.0
+ - fsspec>=2021.04.0
- h5netcdf
- h5py
- hdf5
@@ -33,6 +33,7 @@ dependencies:
- toolz
- xarray>=0.18.0
- zarr>=2.6.0
+ - pytest-repeat
+ - pip
- pip:
- pytest-timeout
- - git+https://github.com/martindurant/filesystem_spec.git@log_and_cache

MODIFY
pangeo_forge_recipes/recipes/xarray_zarr.py
pangeo_forge_recipes/recipes/xarray_zarr.py
@@ -353,7 +353,10 @@ class XarrayZarrRecipe(BaseRecipe):
with file_opener(fname, cache=cache, copy_to_local=self.copy_input_to_local_file) as f:
with dask.config.set(scheduler="single-threaded"): # make sure we don't use a scheduler
logger.debug(f"about to call xr.open_dataset on {f}")
- ds = xr.open_dataset(f, **self.xarray_open_kwargs)
+ kw = self.xarray_open_kwargs.copy()
+ if "engine" not in kw:
+ kw['engine'] = "h5netcdf"
+ ds = xr.open_dataset(f, **kw)
logger.debug("successfully opened dataset")
ds = fix_scalar_attr_encoding(ds)

MODIFY
pangeo_forge_recipes/storage.py
pangeo_forge_recipes/storage.py
import hashlib
+import io
import json
import logging
import os
@@ -24,23 +25,6 @@ def _get_url_size(fname):
return size
-@contextmanager
-def _fsspec_safe_open(fname: str, **kwargs) -> Iterator[OpenFileType]:
- fs, _, paths = fsspec.get_fs_token_paths(fname, mode="rb")
- path = paths[0]
- logger.debug(f"_fsspec_safe_open opening {path} with fs {fs}")
- with fs.open(path, mode="rb") as fp:
- logger.debug(f"_fsspec_safe_open yielding {fp}")
- yield fp
- logger.debug("_fsspec_safe_open yielded")
- # workaround for inconsistent behavior of fsspec.open
- # https://github.com/intake/filesystem_spec/issues/579
- # with fsspec.open(fname, **kwargs) as fp:
- # yield 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:
@@ -51,6 +35,7 @@ def _copy_btw_filesystems(input_opener, output_opener, BLOCK_SIZE=10_000_000):
break
logger.debug(f"_copy_btw_filesystems copying block of {len(data)} bytes")
target.write(data)
+ logger.debug(f"_copy_btw_filesystems done")
class AbstractTarget(ABC):
@@ -128,7 +113,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.
@@ -156,7 +141,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)
@@ -200,7 +185,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)
@@ -214,10 +199,9 @@ def file_opener(
logger.debug(f"file_opener entering first context for {opener}")
with opener as fp:
logger.debug(f"file_opener entering second context for {fp}")
- with fp as fp2: # type: ignore
- logger.debug(f"file_opener yielding {fp2}")
- yield fp2
+ yield fp
logger.debug("file_opener yielded")
+ logger.debug("opener done")
def _slugify(value: str) -> str:

MODIFY
tests/test_fixtures.py
tests/test_fixtures.py
@@ -7,7 +7,8 @@ from pangeo_forge_recipes.utils import fix_scalar_attr_encoding
def test_fixture_local_files(daily_xarray_dataset, netcdf_local_paths):
paths, items_per_file = netcdf_local_paths
paths = [str(path) for path in paths]
- ds = xr.open_mfdataset(paths, combine="nested", concat_dim="time")
+ ds = xr.open_mfdataset(paths, combine="nested", concat_dim="time",
+ engine="h5netcdf")
assert ds.identical(daily_xarray_dataset)
@@ -15,13 +16,15 @@ def test_fixture_local_files(daily_xarray_dataset, netcdf_local_paths):
def test_fixture_local_files_by_variable(daily_xarray_dataset, netcdf_local_paths_by_variable):
paths, items_per_file, fnames_by_variable, path_format = netcdf_local_paths_by_variable
paths = [str(path) for path in paths]
- ds = xr.open_mfdataset(paths, combine="by_coords", concat_dim="time")
+ ds = xr.open_mfdataset(paths, combine="by_coords", concat_dim="time",
+ engine="h5netcdf")
assert ds.identical(daily_xarray_dataset)
def test_fixture_http_files(daily_xarray_dataset, netcdf_http_paths):
urls, items_per_file = netcdf_http_paths
open_files = [fsspec.open(url).open() for url in urls]
- ds = xr.open_mfdataset(open_files, combine="nested", concat_dim="time").load()
+ ds = xr.open_mfdataset(open_files, combine="nested", concat_dim="time",
+ engine="h5netcdf").load()
ds = fix_scalar_attr_encoding(ds)
assert ds.identical(daily_xarray_dataset)

MODIFY
tests/test_recipes.py
tests/test_recipes.py
@@ -45,12 +45,6 @@ def test_recipe_caching_copying(
):
"""The basic recipe test. Use this as a template for other tests."""
- if not copy_input_to_local_file:
- pytest.skip(
- "Opening xarray datasets without this option causes hanging. "
- "See https://github.com/pangeo-forge/pangeo-forge-recipes/pull/146."
- )
-
RecipeClass, file_pattern, kwargs, ds_expected, target = netCDFtoZarr_sequential_recipe
if not cache_inputs:
kwargs.pop("input_cache") # make sure recipe doesn't require input_cache

MODIFY
tests/test_storage.py
tests/test_storage.py
@@ -42,20 +42,13 @@ def test_metadata_target(tmp_metadata_target):
@pytest.mark.parametrize(
"file_paths", [lazy_fixture("netcdf_local_paths"), lazy_fixture("netcdf_http_paths")]
)
-@pytest.mark.parametrize("copy_to_local", [False, True])
+@pytest.mark.parametrize("copy_to_local", [False])
@pytest.mark.parametrize("use_cache, cache_first", [(False, False), (True, False), (True, True)])
-@pytest.mark.parametrize("use_dask", [True, False])
-@pytest.mark.parametrize("use_xarray", [True, False])
+@pytest.mark.parametrize("use_dask", [False])
+@pytest.mark.parametrize("use_xarray", [True])
def test_file_opener(
file_paths, tmp_cache, copy_to_local, use_cache, cache_first, dask_cluster, use_dask, use_xarray
):
-
- if use_dask and not copy_to_local:
- pytest.skip(
- "Opening files without copy_to_local causes hanging with dask. "
- "See https://github.com/pangeo-forge/pangeo-forge-recipes/pull/146."
- )
-
all_paths, _ = file_paths
path = str(all_paths[0])
cache = tmp_cache if use_cache else None
@@ -80,19 +73,19 @@ def test_file_opener(
assert isinstance(fp, str)
with open(fp, mode="rb") as fp2:
if use_xarray:
- ds = xr.open_dataset(fp2)
+ ds = xr.open_dataset(fp2, engine="h5netcdf")
ds.load()
else:
_ = fp2.read()
else:
if use_xarray:
- ds = xr.open_dataset(fp)
+ ds = xr.open_dataset(fp, engine="h5netcdf")
ds.load()
else:
_ = fp.read()
if use_dask:
- client = Client(dask_cluster)
+ with Client(dask_cluster) as client:
to_actual_test_delayed = delayed(do_actual_test)()
to_actual_test_delayed.compute()
client.close()

Remove some contexts and simplify storage code