implement FilePattern.file_type
MODIFY
pangeo_forge_recipes/patterns.py
pangeo_forge_recipes/patterns.py
@@ -3,7 +3,7 @@ Filename / URL patterns.
"""
import inspect
from dataclasses import dataclass, field, replace
-from enum import Enum
+from enum import Enum, auto
from itertools import product
from typing import (
Any,
@@ -99,6 +99,27 @@ CombineDim = Union[MergeDim, ConcatDim]
FilePatternIndex = Index
+class AutoName(Enum):
+ # Recommended by official Python docs for auto naming:
+ # https://docs.python.org/3/library/enum.html#using-automatic-values
+ def _generate_next_value_(name, start, count, last_values):
+ return name
+
+
+class FileType(AutoName):
+ unknown = auto()
+ netcdf3 = auto()
+ netcdf4 = auto()
+ grib = auto()
+ opendap = auto()
+
+
+OPENER_MAP = {
+ FileType.netcdf3: dict(engine="scipy"),
+ FileType.netcdf4: dict(engine="h5netcdf"),
+}
+
+
class FilePattern:
"""Represents an n-dimensional matrix of individual files to be combined
through a combination of merge and concat operations. Each operation generates
@@ -114,8 +135,9 @@ class FilePattern:
May include ``block_size``, ``username``, ``password``, etc.
:param query_string_secrets: If provided, these key/value pairs are appended to
the query string of each ``file_pattern`` url at runtime.
- :param is_opendap: If True, assume all input fnames represent opendap endpoints.
- Cannot be used with caching.
+ :param file_type: The file format of the source files for this pattern. Must be one of
+ the options defined by ``pangeo_forge_recipes.patterns.FileType``.
+ Note: ``FileType.opendap`` cannot be used with caching.
"""
def __init__(
@@ -124,17 +146,17 @@ class FilePattern:
*combine_dims: CombineDim,
fsspec_open_kwargs: Optional[Dict[str, Any]] = None,
query_string_secrets: Optional[Dict[str, str]] = None,
- is_opendap: bool = False,
+ file_type: str = "netcdf4",
):
self.format_function = format_function
self.combine_dims = combine_dims
self.fsspec_open_kwargs = fsspec_open_kwargs if fsspec_open_kwargs else {}
self.query_string_secrets = query_string_secrets if query_string_secrets else {}
- self.is_opendap = is_opendap
- if self.fsspec_open_kwargs and self.is_opendap:
+ self.file_type = FileType(file_type)
+ if self.fsspec_open_kwargs and self.file_type == FileType.opendap:
raise ValueError(
"OPeNDAP inputs are not opened with `fsspec`. "
- "`is_opendap` must be `False` when passing `fsspec_open_kwargs`."
+ "When passing `fsspec_open_kwargs`, `file_type` cannot be `opendap`."
)
def __repr__(self):

MODIFY
pangeo_forge_recipes/recipes/xarray_zarr.py
pangeo_forge_recipes/recipes/xarray_zarr.py
@@ -21,7 +21,7 @@ import zarr
from ..chunk_grid import ChunkGrid
from ..executors.base import Pipeline, Stage
-from ..patterns import CombineOp, DimIndex, FilePattern, Index
+from ..patterns import OPENER_MAP, CombineOp, DimIndex, FilePattern, FileType, Index
from ..reference import create_hdf5_reference, unstrip_protocol
from ..storage import FSSpecTarget, MetadataTarget, file_opener
from ..utils import calc_subsets, fix_scalar_attr_encoding, lock_for_conflicts
@@ -142,7 +142,7 @@ def chunk_position(chunk_key: ChunkKey) -> int:
def cache_input(input_key: InputKey, *, config: XarrayZarrRecipe) -> None:
if config.cache_inputs:
- if config.file_pattern.is_opendap:
+ if config.file_pattern.file_type == FileType.opendap:
raise ValueError("Can't cache opendap inputs")
if config.storage_config.cache is None:
raise ValueError("input_cache is not set.")
@@ -167,7 +167,7 @@ def cache_input(input_key: InputKey, *, config: XarrayZarrRecipe) -> None:
logger.info(f"Metadata already ached for input '{input_key!s}'")
if config.open_input_with_fsspec_reference:
- if config.file_pattern.is_opendap:
+ if config.file_pattern.file_type == FileType.opendap:
raise ValueError("Can't make references for opendap inputs")
if config.storage_config.metadata is None:
raise ValueError("Can't make references; no metadata cache assigned")
@@ -248,7 +248,7 @@ def open_input(input_key: InputKey, *, config: XarrayZarrRecipe) -> xr.Dataset:
fname = config.file_pattern[input_key]
logger.info(f"Opening input with Xarray {input_key!s}: '{fname}'")
- if config.file_pattern.is_opendap:
+ if config.file_pattern.file_type == FileType.opendap:
if config.cache_inputs:
raise ValueError("Can't cache opendap inputs")
if config.copy_input_to_local_file:
@@ -286,19 +286,20 @@ def open_input(input_key: InputKey, *, config: XarrayZarrRecipe) -> xr.Dataset:
else:
cache = config.storage_config.cache if config.cache_inputs else None
+ bypass_open = True if config.file_pattern.file_type == FileType.opendap else False
with file_opener(
fname,
cache=cache,
copy_to_local=config.copy_input_to_local_file,
- bypass_open=config.file_pattern.is_opendap,
+ bypass_open=bypass_open,
secrets=config.file_pattern.query_string_secrets,
**config.file_pattern.fsspec_open_kwargs,
) as f:
with dask.config.set(scheduler="single-threaded"): # make sure we don't use a scheduler
kw = config.xarray_open_kwargs.copy()
if "engine" not in kw:
- kw["engine"] = "h5netcdf"
+ kw.update(OPENER_MAP[config.file_pattern.file_type])
logger.debug(f"about to enter xr.open_dataset context on {f}")
with xr.open_dataset(f, **kw) as ds:
logger.debug("successfully opened dataset")
@@ -743,7 +744,7 @@ class XarrayZarrRecipe(BaseRecipe, StorageMixin, FilePatternMixin):
)
self.nitems_per_input = self.file_pattern.nitems_per_input[self.concat_dim]
- if self.file_pattern.is_opendap:
+ if self.file_pattern.file_type == FileType.opendap:
if self.cache_inputs:
raise ValueError("Can't cache opendap inputs.")
else:

MODIFY
tests/recipe_tests/test_XarrayZarrRecipe.py
tests/recipe_tests/test_XarrayZarrRecipe.py
@@ -450,7 +450,7 @@ def calc_sequence_length_fixture(netcdf_local_file_pattern, tmp_metadata_target,
*rest,
fsspec_open_kwargs=netcdf_local_file_pattern.fsspec_open_kwargs,
query_string_secrets=netcdf_local_file_pattern.query_string_secrets,
- is_opendap=netcdf_local_file_pattern.is_opendap,
+ file_type=netcdf_local_file_pattern.file_type,
)
n_inputs = file_pattern.dims[concat_dim.name]

MODIFY
tests/test_patterns.py
tests/test_patterns.py
@@ -6,6 +6,7 @@ from pangeo_forge_recipes.patterns import (
CombineOp,
ConcatDim,
FilePattern,
+ FileType,
MergeDim,
pattern_from_file_sequence,
prune_pattern,
@@ -40,7 +41,7 @@ def concat_merge_pattern():
return make_concat_merge_pattern()
-@pytest.fixture(params=[dict(fsspec_open_kwargs={"block_size": "foo"}), dict(is_opendap=True)])
+@pytest.fixture(params=[dict(fsspec_open_kwargs={"block_size": "foo"}), dict(file_type="opendap")])
def concat_merge_pattern_with_kwargs(request):
return make_concat_merge_pattern(**request.param)
@@ -89,11 +90,11 @@ def test_file_pattern_concat_merge(runtime_secrets, pickle, concat_merge_pattern
if runtime_secrets:
if "fsspec_open_kwargs" in runtime_secrets.keys():
- if not fp.is_opendap:
+ if not fp.file_type == FileType.opendap:
fp.fsspec_open_kwargs.update(runtime_secrets["fsspec_open_kwargs"])
else:
pytest.skip(
- "`fsspec_open_kwargs` should never be used in combination with `is_opendap`. "
+ "`fsspec_open_kwargs` should never be used in combination with `opendap`."
" This is checked in `FilePattern.__init__` but not when updating attributes. "
"Proposed changes to secret handling will obviate the need for runtime updates"
" to attributes in favor of encryption. So for now, we'll just skip this."
@@ -128,20 +129,19 @@ def test_file_pattern_concat_merge(runtime_secrets, pickle, concat_merge_pattern
assert fp[key] == expected_fname
if "fsspec_open_kwargs" in kwargs.keys():
- assert fp.is_opendap is False
+ assert fp.file_type != FileType.opendap
if "fsspec_open_kwargs" in runtime_secrets.keys():
kwargs["fsspec_open_kwargs"].update(runtime_secrets["fsspec_open_kwargs"])
assert fp.fsspec_open_kwargs == kwargs["fsspec_open_kwargs"]
if "query_string_secrets" in runtime_secrets.keys():
assert fp.query_string_secrets == runtime_secrets["query_string_secrets"]
- if "is_opendap" in kwargs.keys():
- assert fp.is_opendap == kwargs["is_opendap"]
- assert fp.is_opendap is True
+ if kwargs.get("file_type", None) == "opendap":
+ assert fp.file_type == FileType.opendap
assert fp.fsspec_open_kwargs == {}
def test_incompatible_kwargs():
- kwargs = dict(fsspec_open_kwargs={"block_size": "foo"}, is_opendap=True)
+ kwargs = dict(fsspec_open_kwargs={"block_size": "foo"}, file_type="opendap")
with pytest.raises(ValueError):
make_concat_merge_pattern(**kwargs)
return
@@ -154,11 +154,11 @@ def test_prune(nkeep, concat_merge_pattern_with_kwargs, runtime_secrets):
if runtime_secrets:
if "fsspec_open_kwargs" in runtime_secrets.keys():
- if not fp.is_opendap:
+ if fp.file_type != FileType.opendap:
fp.fsspec_open_kwargs.update(runtime_secrets["fsspec_open_kwargs"])
else:
pytest.skip(
- "`fsspec_open_kwargs` should never be used in combination with `is_opendap`. "
+ "`fsspec_open_kwargs` should never be used in combination with `opendap`."
" This is checked in `FilePattern.__init__` but not when updating attributes. "
"Proposed changes to secret handling will obviate the need for runtime updates"
" to attributes in favor of encryption. So for now, we'll just skip this."

implement FilePattern.file_type