rearrange package organization
RENAME
pangeo_forge/recipe.py
pangeo_forge/recipes/xarray_zarr.py
@@ -7,27 +7,26 @@ import logging
import os
import tempfile
import warnings
-from abc import ABC, abstractmethod
from contextlib import ExitStack, contextmanager
from dataclasses import dataclass, field
from itertools import product
-from typing import Callable, Dict, Hashable, Iterable, List, Optional, Sequence, Tuple
+from typing import Callable, Dict, List, Optional, Sequence, Tuple
import dask
import fsspec
import numpy as np
import xarray as xr
import zarr
-from rechunker.types import MultiStagePipeline, ParallelPipelines, Stage
-from .patterns import FilePattern
-from .storage import AbstractTarget, UninitializedTarget, UninitializedTargetError
-from .utils import (
+from ..patterns import FilePattern
+from ..storage import AbstractTarget, UninitializedTarget, UninitializedTargetError
+from ..utils import (
chunk_bounds_and_conflicts,
chunked_iterable,
fix_scalar_attr_encoding,
lock_for_conflicts,
)
+from .base import BaseRecipe
# use this filename to store global recipe metadata in the metadata_cache
# it will be written once (by prepare_target) and read many times (by store_chunk)
@@ -35,97 +34,6 @@ _GLOBAL_METADATA_KEY = "pangeo-forge-recipe-metadata.json"
logger = logging.getLogger(__name__)
-# How to manually execute a recipe: ###
-#
-# t = PangeoForgeTarget()
-# r = MyRecipe(target=t, **opts) # 1
-# # manual execution of recipe
-# for input_key in r.iter_inputs():
-# r.cache_input(input_key) # 4
-# r.prepare_target() # 3
-# for chunk_key in r.iter_chunks():
-# r.store_chunk(chunk_key) # 5
-# r.finalize_target() # 6
-
-
-# 1) Initialize the Recipe object
-# 2) Point the Recipe at its Target
-# 3) Initialize the recipe.
-# Check if the target exists; if not, create it.
-# 4) cache the inputs to proximate storage (OPTIONAL)
-# Some recipes won't need this (e.g. cloud to cloud)
-# If so, iter_inputs is just an empty iterator
-# 5) Load each chunk from the inputs and store it in the target
-# Might be coming from the cache or might be read directly.
-# 6)
-
-
-class BaseRecipe(ABC):
- """Base recipe class from which all other Recipes inherit.
- """
-
- @property
- @abstractmethod
- def prepare_target(self) -> Callable[[], None]:
- """Prepare the recipe for execution by initializing the target.
- Attribute that returns a callable function.
- """
- pass
-
- @abstractmethod
- def iter_inputs(self) -> Iterable[Hashable]:
- """Iterate over all inputs."""
- pass
-
- @property
- @abstractmethod
- def cache_input(self) -> Callable[[Hashable], None]:
- """Copy an input from its source location to the cache.
- Attribute that returns a callable function.
- """
- pass
-
- @abstractmethod
- def iter_chunks(self) -> Iterable[Hashable]:
- """Iterate over all target chunks."""
- pass
-
- @property
- @abstractmethod
- def store_chunk(self) -> Callable[[Hashable], None]:
- """Store a chunk of data in the target.
- Attribute that returns a callable function.
- """
- pass
-
- @property
- @abstractmethod
- def finalize_target(self) -> Callable[[], None]:
- """Final step to finish the recipe after data has been written.
- Attribute that returns a callable function.
- """
- pass
-
- def to_pipelines(self) -> ParallelPipelines:
- """Translate recipe to pipeline for execution.
- """
-
- pipeline = [] # type: MultiStagePipeline
- if getattr(self, "cache_inputs", False): # TODO: formalize this contract
- pipeline.append(Stage(self.cache_input, list(self.iter_inputs())))
- pipeline.append(Stage(self.prepare_target))
- pipeline.append(Stage(self.store_chunk, list(self.iter_chunks())))
- pipeline.append(Stage(self.finalize_target))
- pipelines = [] # type: ParallelPipelines
- pipelines.append(pipeline)
- return pipelines
-
- # https://stackoverflow.com/questions/59986413/achieving-multiple-inheritance-using-python-dataclasses
- def __post_init__(self):
- # just intercept the __post_init__ calls so they
- # aren't relayed to `object`
- pass
-
def _encode_key(key) -> str:
return "-".join([str(k) for k in key])

MODIFY
tests/conftest.py
tests/conftest.py
@@ -11,7 +11,7 @@ import pytest
import xarray as xr
from dask.distributed import Client, LocalCluster
-from pangeo_forge import recipe
+from pangeo_forge import recipes
from pangeo_forge.executors import (
DaskPipelineExecutor,
PrefectPipelineExecutor,
@@ -175,7 +175,7 @@ def netCDFtoZarr_sequential_recipe(daily_xarray_dataset, netcdf_local_paths, tmp
paths, items_per_file = netcdf_local_paths
file_pattern = pattern_from_file_sequence([str(path) for path in paths], "time", items_per_file)
kwargs = dict(inputs_per_chunk=1, target=tmp_target, input_cache=tmp_cache,)
- return recipe.XarrayZarrRecipe, file_pattern, kwargs, daily_xarray_dataset, tmp_target
+ return recipes.XarrayZarrRecipe, file_pattern, kwargs, daily_xarray_dataset, tmp_target
@pytest.fixture
@@ -194,7 +194,7 @@ def netCDFtoZarr_sequential_multi_variable_recipe(
MergeDim("variable", ["foo", "bar"]),
)
kwargs = dict(inputs_per_chunk=1, target=tmp_target, input_cache=tmp_cache,)
- return recipe.XarrayZarrRecipe, file_pattern, kwargs, daily_xarray_dataset, tmp_target
+ return recipes.XarrayZarrRecipe, file_pattern, kwargs, daily_xarray_dataset, tmp_target
@pytest.fixture(scope="session")

rearrange package organization