refactor conditional file_pattern logic into helper func
MODIFY
tests/conftest.py
tests/conftest.py
@@ -94,6 +94,29 @@ def _split_up_files_by_variable_and_day(ds, day_param):
return all_dsets, all_fnames, fnames_by_variable
+def _make_file_pattern(netcdf_paths):
+ paths, items_per_file = netcdf_paths[:2]
+
+ if len(netcdf_paths) == 2:
+ file_pattern = pattern_from_file_sequence(
+ [str(path) for path in paths], "time", items_per_file
+ )
+ else:
+ _, path_format = netcdf_paths[2:]
+ time_index = list(range(len(paths) // 2))
+
+ def format_function(variable, time):
+ return path_format.format(variable=variable, time=time)
+
+ file_pattern = FilePattern(
+ format_function,
+ ConcatDim("time", time_index, items_per_file),
+ MergeDim("variable", ["foo", "bar"]),
+ )
+
+ return file_pattern
+
+
@pytest.fixture(scope="session", params=["D", "2D"])
def items_per_file(request):
return request.param
@@ -208,24 +231,7 @@ def tmp_metadata_target(tmpdir_factory):
def netCDFtoZarr_sequential_recipe(
daily_xarray_dataset, netcdf_paths, tmp_target, tmp_cache, tmp_metadata_target
):
- paths, items_per_file = netcdf_paths[:2]
-
- if len(netcdf_paths) == 2:
- file_pattern = pattern_from_file_sequence(
- [str(path) for path in paths], "time", items_per_file
- )
- else:
- _, path_format = netcdf_paths[2:]
- time_index = list(range(len(paths) // 2))
-
- def format_function(variable, time):
- return path_format.format(variable=variable, time=time)
-
- file_pattern = FilePattern(
- format_function,
- ConcatDim("time", time_index, items_per_file),
- MergeDim("variable", ["foo", "bar"]),
- )
+ file_pattern = _make_file_pattern(netcdf_paths)
kwargs = dict(
inputs_per_chunk=1,
@@ -240,25 +246,11 @@ def netCDFtoZarr_sequential_recipe(
def netCDFtoZarr_sequential_subset_recipe(
daily_xarray_dataset, netcdf_paths, tmp_target, tmp_cache, tmp_metadata_target
):
- paths, items_per_file = netcdf_paths[:2]
+ items_per_file = netcdf_paths[1]
if items_per_file != 2:
pytest.skip("This recipe only makes sense with items_per_file == 2.")
- if len(netcdf_paths) == 2:
- file_pattern = pattern_from_file_sequence(
- [str(path) for path in paths], "time", items_per_file
- )
- else:
- _, path_format = netcdf_paths[2:]
- time_index = list(range(len(paths) // 2))
- def format_function(variable, time):
- return path_format.format(variable=variable, time=time)
-
- file_pattern = FilePattern(
- format_function,
- ConcatDim("time", time_index, items_per_file),
- MergeDim("variable", ["foo", "bar"]),
- )
+ file_pattern = _make_file_pattern(netcdf_paths)
kwargs = dict(
subset_inputs={"time": 2},

refactor conditional file_pattern logic into helper func