recipe tests pass
MODIFY
pangeo_forge/recipe/recipe.py
pangeo_forge/recipe/recipe.py
@@ -175,7 +175,7 @@ class XarrayConcatChunkOpener(XarrayInputOpener):
@dataclass
-class ZarrWriterMixin:
+class ZarrXarrayWriterMixin:
@property
def store_chunk(self) -> Callable:
@@ -313,7 +313,7 @@ class SequenceRecipe(DatasetRecipe):
# make sure the concat dim has a valid fill_value to avoid
# overruns when writing chunk
- #ds[self.sequence_dim].encoding = {'_FillValue': 0}
+ ds[self.sequence_dim].encoding = {'_FillValue': -1}
# actually not necessary if we use decode_times=False
self.initialize_target(ds)
@@ -327,7 +327,7 @@ class StandardSequentialRecipe(
SequenceRecipe,
InputCachingMixin,
XarrayConcatChunkOpener,
- ZarrWriterMixin,
+ ZarrXarrayWriterMixin,
ZarrConsolidatorMixin
):
pass

MODIFY
tests/fixtures.py
tests/fixtures.py
@@ -45,6 +45,8 @@ def daily_xarray_dataset():
def netcdf_local_paths(daily_xarray_dataset, tmpdir_factory, request):
"""Return a list of paths pointing to netcdf files."""
tmp_path = tmpdir_factory.mktemp("netcdf_data")
+ items_per_file = {"D": 1, "2D": 2}
+ daily_xarray_dataset.attrs['items_per_file'] = items_per_file[request.param]
gb = daily_xarray_dataset.resample(time=request.param)
_, datasets = zip(*gb)
fnames = [f"{n:03d}.nc" for n in range(len(datasets))]

MODIFY
tests/test_recipe.py
tests/test_recipe.py
import pytest
+import xarray as xr
+import zarr
from pangeo_forge.recipe import recipe
-from .fixtures import daily_xarray_dataset, netcdf_local_paths, tmp_target
+from .fixtures import daily_xarray_dataset, netcdf_local_paths, tmp_target, tmp_cache
dummy_fnames = ["a.nc", "b.nc", "c.nc"]
@pytest.mark.parametrize(
@@ -12,32 +14,43 @@ dummy_fnames = ["a.nc", "b.nc", "c.nc"]
(dummy_fnames, 2, [0, 1], [("a.nc", "b.nc",), ("c.nc",)])
]
)
-def test_file_sequence_recipe(file_urls, files_per_chunk, expected_keys, expected_filenames, tmp_target):
+def test_sequence_recipe(file_urls, files_per_chunk, expected_keys, expected_filenames, tmp_target):
- r = recipe.FileSequenceRecipe(
- file_urls=file_urls,
+ r = recipe.SequenceRecipe(
+ input_urls=file_urls,
sequence_dim="time",
- files_per_chunk=files_per_chunk,
- target=tmp_target
+ inputs_per_chunk=files_per_chunk,
+ target=tmp_target,
+ chunk_preprocess_funcs=[],
)
+ assert r.sequence_len() == len(file_urls)
+
chunk_keys = list(r.iter_chunks())
assert chunk_keys == expected_keys
for k, expected in zip(r.iter_chunks(), expected_filenames):
- fnames = r.filenames_for_chunk(k)
+ fnames = r.inputs_for_chunk(k)
assert fnames == expected
-def test_full_recipe(daily_xarray_dataset, netcdf_local_paths, tmp_target):
+def test_full_recipe(daily_xarray_dataset, netcdf_local_paths, tmp_target, tmp_cache):
- r = StandardSequentialRecipe(
- file_urls=netcdf_local_paths,
- sequence_dim='time',
+ r = recipe.StandardSequentialRecipe(
+ consolidate_zarr=True,
+ xarray_open_kwargs={},
+ xarray_concat_kwargs={},
+ require_cache=False,
+ input_cache=tmp_cache,
+ target=tmp_target,
+ chunk_preprocess_funcs=[],
+ input_urls=netcdf_local_paths,
+ sequence_dim="time",
+ inputs_per_chunk=1,
+ nitems_per_input=daily_xarray_dataset.attrs['items_per_file']
)
- r.set_target(tmp_dir)
- # manual execution of recipe
+ # this is the cannonical way to manually execute a recipe
r.prepare()
for input_key in r.iter_inputs():
r.cache_input(input_key)
@@ -45,5 +58,6 @@ def test_full_recipe(daily_xarray_dataset, netcdf_local_paths, tmp_target):
r.store_chunk(chunk_key)
r.finalize()
- ds_target = xr.open_dataset(tmp_dir)
- assert ds_target.identical(daily_xarray_dataset)
+ ds_target = xr.open_zarr(tmp_target.get_mapper(), consolidated=True).load()
+ ds_expected = daily_xarray_dataset.compute()
+ assert ds_target.identical(ds_expected)

recipe tests pass